diff --git a/.gitignore b/.gitignore index 66a86d9..f46064e 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,14 @@ tmp .vscode coverage.txt .build +pihole-guard.db* +inventory.ndjson + +# local design notes and session artifacts — not for publication +CHAT_SUMMARY_*.md +CLAUDE.md +OVERVIEW.md +SETUP.md +SNI_DETECTION_DESIGN.md +zeek-integration-guide.md +IMPLEMENTATION_NOTES.md diff --git a/.golangci.yml b/.golangci.yml new file mode 100644 index 0000000..667a405 --- /dev/null +++ b/.golangci.yml @@ -0,0 +1,10 @@ +version: "2" +linters: + enable: + - staticcheck + - errcheck + - revive + +issues: + max-issues-per-linter: 0 + max-same-issues: 0 diff --git a/internal/deghost/deghost.go b/internal/deghost/deghost.go index f7e6788..6ec2d6b 100644 --- a/internal/deghost/deghost.go +++ b/internal/deghost/deghost.go @@ -119,3 +119,79 @@ func ShouldKill(report *IPReport) bool { return report.Security.IsAbuser || report.Security.IsAttacker || report.Security.IsThreat } + +// DomainReport matches the API payload for domain reputation checks. +type DomainReport struct { + Status string `json:"status"` + HasMX bool `json:"has_mx"` + Disposable bool `json:"disposable"` + Spam bool `json:"spam"` + PublicDomain bool `json:"public_domain"` + RelayDomain bool `json:"relay_domain"` + Blacklisted bool `json:"blacklisted"` + DomainAgeInDays int `json:"domain_age_in_days"` +} + +// CheckDomain fetches a reputation report for a single domain. +// Unlike CheckIP's 403 handling (nil report for private/reserved IPs), this endpoint +// returns a real JSON body on HTTP 200, 400, and 403 — all are decoded into a DomainReport. +// Only 499/504/500/other status codes are treated as hard errors. +func (c *Client) CheckDomain(ctx context.Context, domain string) (*DomainReport, error) { + if c == nil { + return nil, errors.New("nil deghost client") + } + if c.httpClient == nil { + return nil, errors.New("nil deghost http client") + } + + baseURL := strings.TrimSpace(c.baseURL) + if baseURL == "" { + return nil, errors.New("deghost base URL is required") + } + + domain = strings.ToLower(strings.TrimSpace(domain)) + if domain == "" { + return nil, errors.New("domain is required") + } + + endpoint, err := url.JoinPath(baseURL, "domain", domain) + if err != nil { + return nil, fmt.Errorf("build endpoint: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer func() { + _ = resp.Body.Close() + }() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusBadRequest && resp.StatusCode != http.StatusForbidden { + return nil, fmt.Errorf("deghost returned status %d %s", resp.StatusCode, http.StatusText(resp.StatusCode)) + } + + var report DomainReport + if err := json.NewDecoder(resp.Body).Decode(&report); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + + return &report, nil +} + +// ShouldBlockDomain reports whether the domain report matches the current block policy. +// It triggers only on explicit negative signals (status "not_allowed" or blacklisted), +// ignoring softer signals like disposable, spam, or domain age which are too noisy +// to act on alone — consistent with how ShouldKill only fires on explicit threat fields. +func ShouldBlockDomain(report *DomainReport) bool { + if report == nil { + return false + } + + return report.Status == "not_allowed" || report.Blacklisted +} diff --git a/internal/deghost/deghost_test.go b/internal/deghost/deghost_test.go new file mode 100644 index 0000000..7f2153f --- /dev/null +++ b/internal/deghost/deghost_test.go @@ -0,0 +1,284 @@ +package deghost + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" +) + +func TestCheckDomainOK(t *testing.T) { + t.Parallel() + + want := DomainReport{ + Status: "allowed", + HasMX: true, + Disposable: false, + Spam: false, + PublicDomain: true, + RelayDomain: false, + Blacklisted: false, + DomainAgeInDays: 11117, + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Errorf("method = %s, want GET", r.Method) + } + if r.URL.Path != "/domain/example.com" { + t.Errorf("path = %s, want /domain/example.com", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(want) + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + got, err := client.CheckDomain(context.Background(), "example.com") + if err != nil { + t.Fatalf("CheckDomain() error = %v", err) + } + if got == nil { + t.Fatal("CheckDomain() = nil, want report") + } + if got.Status != want.Status { + t.Errorf("Status = %q, want %q", got.Status, want.Status) + } + if got.HasMX != want.HasMX { + t.Errorf("HasMX = %v, want %v", got.HasMX, want.HasMX) + } + if got.DomainAgeInDays != want.DomainAgeInDays { + t.Errorf("DomainAgeInDays = %d, want %d", got.DomainAgeInDays, want.DomainAgeInDays) + } +} + +func TestCheckDomainStatusForbiddenDecodesBody(t *testing.T) { + t.Parallel() + + want := DomainReport{ + Status: "not_allowed", + HasMX: false, + Disposable: false, + Spam: false, + PublicDomain: false, + RelayDomain: false, + Blacklisted: false, + DomainAgeInDays: 0, + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusForbidden) + _ = json.NewEncoder(w).Encode(want) + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + got, err := client.CheckDomain(context.Background(), "blocked.example.com") + if err != nil { + t.Fatalf("CheckDomain() on 403 error = %v", err) + } + if got == nil { + t.Fatal("CheckDomain() on 403 = nil, want decoded report (not nil like IP endpoint)") + } + if got.Status != "not_allowed" { + t.Errorf("Status = %q, want %q", got.Status, "not_allowed") + } +} + +func TestCheckDomainStatusBadRequestDecodesBody(t *testing.T) { + t.Parallel() + + want := DomainReport{ + Status: "not_allowed", + HasMX: false, + Blacklisted: true, + DomainAgeInDays: 0, + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(want) + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + got, err := client.CheckDomain(context.Background(), "bad.example.com") + if err != nil { + t.Fatalf("CheckDomain() on 400 error = %v", err) + } + if got == nil { + t.Fatal("CheckDomain() on 400 = nil, want decoded report") + } + if !got.Blacklisted { + t.Error("Blacklisted = false, want true") + } +} + +func TestCheckDomainStatusInternalServerError(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + got, err := client.CheckDomain(context.Background(), "example.com") + if err == nil { + t.Fatalf("CheckDomain() on 500 = %v, want error", got) + } +} + +func TestCheckDomainStatusGatewayTimeout(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusGatewayTimeout) + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + got, err := client.CheckDomain(context.Background(), "example.com") + if err == nil { + t.Fatalf("CheckDomain() on 504 = %v, want error", got) + } +} + +func TestCheckDomainStatusClientClosedRequest(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(499) + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + got, err := client.CheckDomain(context.Background(), "example.com") + if err == nil { + t.Fatalf("CheckDomain() on 499 = %v, want error", got) + } +} + +func TestCheckDomainMalformedJSON(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if _, err := fmt.Fprint(w, `{"status": "allowed" bad json`); err != nil { + t.Errorf("fmt.Fprint: %v", err) + } + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + got, err := client.CheckDomain(context.Background(), "example.com") + if err == nil { + t.Fatalf("CheckDomain() on bad JSON = %v, want error", got) + } +} + +func TestCheckDomainNilClient(t *testing.T) { + t.Parallel() + + var c *Client + got, err := c.CheckDomain(context.Background(), "example.com") + if err == nil { + t.Fatalf("CheckDomain() on nil client = %v, want error", got) + } +} + +func TestCheckDomainEmptyBaseURL(t *testing.T) { + t.Parallel() + + client := &Client{baseURL: "", httpClient: &http.Client{}} + got, err := client.CheckDomain(context.Background(), "example.com") + if err == nil { + t.Fatalf("CheckDomain() on empty baseURL = %v, want error", got) + } +} + +func TestCheckDomainEmptyDomain(t *testing.T) { + t.Parallel() + + client := NewClient("https://example.com", 5) + got, err := client.CheckDomain(context.Background(), " ") + if err == nil { + t.Fatalf("CheckDomain() on empty domain = %v, want error", got) + } +} + +func TestCheckDomainNormalized(t *testing.T) { + t.Parallel() + + var gotPath string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(DomainReport{Status: "allowed"}) + })) + defer srv.Close() + + client := NewClient(srv.URL, srv.Client().Timeout) + _, err := client.CheckDomain(context.Background(), " Example.COM ") + if err != nil { + t.Fatalf("CheckDomain() error = %v", err) + } + want := "/domain/example.com" + if gotPath != want { + t.Errorf("request path = %q, want %q (domain should be trimmed and lowercased)", gotPath, want) + } +} + +func TestShouldBlockDomainNilReport(t *testing.T) { + t.Parallel() + if ShouldBlockDomain(nil) { + t.Error("ShouldBlockDomain(nil) = true, want false") + } +} + +func TestShouldBlockDomainNotAllowed(t *testing.T) { + t.Parallel() + + report := &DomainReport{Status: "not_allowed"} + if !ShouldBlockDomain(report) { + t.Error("ShouldBlockDomain(not_allowed) = false, want true") + } +} + +func TestShouldBlockDomainBlacklisted(t *testing.T) { + t.Parallel() + + report := &DomainReport{Status: "allowed", Blacklisted: true} + if !ShouldBlockDomain(report) { + t.Error("ShouldBlockDomain(blacklisted) = false, want true") + } +} + +func TestShouldBlockDomainBothFalse(t *testing.T) { + t.Parallel() + + report := &DomainReport{Status: "allowed", Blacklisted: false} + if ShouldBlockDomain(report) { + t.Error("ShouldBlockDomain(allowed, not blacklisted) = true, want false") + } +} + +func TestShouldBlockDomainDisposableAloneDoesNotBlock(t *testing.T) { + t.Parallel() + + report := &DomainReport{Status: "allowed", Disposable: true} + if ShouldBlockDomain(report) { + t.Error("ShouldBlockDomain(disposable alone) = true, want false") + } +} + +func TestShouldBlockDomainSpamAloneDoesNotBlock(t *testing.T) { + t.Parallel() + + report := &DomainReport{Status: "allowed", Spam: true} + if ShouldBlockDomain(report) { + t.Error("ShouldBlockDomain(spam alone) = true, want false") + } +} diff --git a/internal/enforcer/enforcer.go b/internal/enforcer/enforcer.go new file mode 100644 index 0000000..ae77a25 --- /dev/null +++ b/internal/enforcer/enforcer.go @@ -0,0 +1,114 @@ +// Package enforcer terminates processes that hold blocked connections. +// +// STOPGAP: this is deliberately coarser than intended. The goal is to terminate a single +// connection; killing the owning process takes down every other connection it holds too. +// It exists because somo offers no non-interactive per-connection kill. Replace with +// `ss -K dst dport ` (requires CONFIG_INET_DIAG_DESTROY) when that work is done. +package enforcer + +import ( + "fmt" + "log/slog" + "os" + "strconv" + "strings" + "syscall" + "time" +) + +// protectedPrograms is the default set of programs that must never be killed. +// All keys MUST be lowercase: isProtected lowercases the input before lookup. +var protectedPrograms = map[string]bool{ + "systemd": true, + "init": true, + "sshd": true, + "networkmanager": true, + "systemd-resolved": true, + "systemd-networkd": true, + "dbus-daemon": true, + "containerd": true, + "dockerd": true, +} + +// SetProtectedPrograms replaces the default protected-program set. +// Keys are normalized to lowercase on entry because isProtected lowercases +// its input, so stored keys must match for the lookup to work. +func SetProtectedPrograms(programs map[string]bool) { + normalized := make(map[string]bool, len(programs)) + for k, v := range programs { + normalized[strings.ToLower(k)] = v + } + protectedPrograms = normalized +} + +// isProtected reports whether the program name (case-insensitive) is in the protected set. +func isProtected(name string) bool { + lower := strings.ToLower(name) + return protectedPrograms[lower] +} + +// KillProcess terminates the process holding a connection. +// +// Guardrails: +// - Refuses PID 1 (init). +// - Refuses hexwall's own PID. +// - Refuses any PID <= 0. +// - Refuses protected programs (case-insensitive match). +// - Sends SIGTERM first, waits a grace period, then SIGKILL if still alive. +func KillProcess(pid int, program string, remoteAddr string) error { + if pid <= 0 { + return fmt.Errorf("invalid PID %d", pid) + } + + if pid == 1 { + return fmt.Errorf("refusing to signal init (PID 1)") + } + + if pid == os.Getpid() { + return fmt.Errorf("refusing to signal own process (PID %d)", pid) + } + + if isProtected(program) { + return fmt.Errorf("refusing to kill protected program %q (PID %d)", program, pid) + } + + slog.Warn("killing connection", "pid", pid, "program", program, "address", remoteAddr) + + // SIGTERM first — allows graceful shutdown. + if err := syscall.Kill(pid, syscall.SIGTERM); err != nil { + return fmt.Errorf("send SIGTERM to PID %d: %w", pid, err) + } + + // Grace period: wait up to 2 seconds for the process to exit. + time.Sleep(2 * time.Second) + + // Check whether the process still exists. + if err := syscall.Kill(pid, 0); err != nil { + // Process is gone — SIGTERM worked. + slog.Info("process terminated by SIGTERM", "pid", pid, "program", program) + return nil + } + + // Process still alive — escalate to SIGKILL. + slog.Warn("process survived SIGTERM, sending SIGKILL", "pid", pid, "program", program) + if err := syscall.Kill(pid, syscall.SIGKILL); err != nil { + return fmt.Errorf("send SIGKILL to PID %d: %w", pid, err) + } + + return nil +} + +// ParsePID converts a string PID to int, returning an error for invalid input. +func ParsePID(s string) (int, error) { + s = strings.TrimSpace(s) + if s == "" || s == "-" { + return 0, fmt.Errorf("empty or dash PID %q", s) + } + + pid, err := strconv.Atoi(s) + if err != nil { + return 0, fmt.Errorf("invalid PID %q: %w", s, err) + } + + return pid, nil +} diff --git a/internal/enforcer/enforcer_test.go b/internal/enforcer/enforcer_test.go new file mode 100644 index 0000000..16fe2bd --- /dev/null +++ b/internal/enforcer/enforcer_test.go @@ -0,0 +1,150 @@ +package enforcer + +import ( + "os" + "strings" + "testing" +) + +func TestKillProcessRefusesPIDZero(t *testing.T) { + t.Parallel() + + err := KillProcess(0, "curl", "1.2.3.4:443") + if err == nil { + t.Fatal("KillProcess(0) = nil, want error") + } +} + +func TestKillProcessRefusesNegativePID(t *testing.T) { + t.Parallel() + + err := KillProcess(-1, "curl", "1.2.3.4:443") + if err == nil { + t.Fatal("KillProcess(-1) = nil, want error") + } +} + +func TestKillProcessRefusesPIDOne(t *testing.T) { + t.Parallel() + + err := KillProcess(1, "systemd", "1.2.3.4:443") + if err == nil { + t.Fatal("KillProcess(1) = nil, want error for init") + } +} + +func TestKillProcessRefusesOwnPID(t *testing.T) { + t.Parallel() + + err := KillProcess(os.Getpid(), "test", "1.2.3.4:443") + if err == nil { + t.Fatal("KillProcess(own PID) = nil, want error") + } +} + +func TestKillProcessRefusesProtectedPrograms(t *testing.T) { + t.Parallel() + + protected := []string{ + "systemd", "init", "sshd", "NetworkManager", + "systemd-resolved", "systemd-networkd", "dbus-daemon", + "containerd", "dockerd", + } + + for _, prog := range protected { + // Use a non-init, non-own PID that we know exists won't be matched + // by the other guards. PID 2 is usually kthreadd on Linux. + err := KillProcess(2, prog, "1.2.3.4:443") + if err == nil { + t.Errorf("KillProcess(2, %q) = nil, want error for protected program", prog) + continue + } + if !strings.Contains(err.Error(), "protected") { + t.Errorf("KillProcess(2, %q) error = %q, want it to mention 'protected'", prog, err) + } + } +} + +func TestKillProcessCaseInsensitiveProtectedMatch(t *testing.T) { + t.Parallel() + + cases := []string{"SSHD", "SystemD", "NETWORKMANAGER", "containerd", "DockerD"} + for _, prog := range cases { + err := KillProcess(2, prog, "1.2.3.4:443") + if err == nil { + t.Errorf("KillProcess(2, %q) = nil, want error for case-insensitive protected match", prog) + continue + } + if !strings.Contains(err.Error(), "protected") { + t.Errorf("KillProcess(2, %q) error = %q, want it to mention 'protected'", prog, err) + } + } +} + +func TestParsePIDValid(t *testing.T) { + t.Parallel() + + tests := []struct { + input string + want int + }{ + {"1234", 1234}, + {" 5678 ", 5678}, + {"0", 0}, + {"1", 1}, + } + + for _, tt := range tests { + got, err := ParsePID(tt.input) + if err != nil { + t.Errorf("ParsePID(%q) error = %v", tt.input, err) + } + if got != tt.want { + t.Errorf("ParsePID(%q) = %d, want %d", tt.input, got, tt.want) + } + } +} + +func TestParsePIDInvalid(t *testing.T) { + t.Parallel() + + tests := []string{"", "-", "abc", "12.34", "12ab"} + for _, input := range tests { + _, err := ParsePID(input) + if err == nil { + t.Errorf("ParsePID(%q) = nil, want error", input) + } + } +} + +func TestIsProtected(t *testing.T) { + t.Parallel() + + if !isProtected("sshd") { + t.Error("isProtected(\"sshd\") = false, want true") + } + if !isProtected("SSHD") { + t.Error("isProtected(\"SSHD\") = false, want true (case-insensitive)") + } + if !isProtected("SystemD") { + t.Error("isProtected(\"SystemD\") = false, want true (case-insensitive)") + } + if isProtected("curl") { + t.Error("isProtected(\"curl\") = true, want false") + } + if isProtected("") { + t.Error("isProtected(\"\") = true, want false") + } +} + +func TestIsProtectedMixedCaseAllEntries(t *testing.T) { + t.Parallel() + + for name := range protectedPrograms { + // Construct a mixed-case variant: uppercase first char + lowercase rest. + mixed := strings.ToUpper(name[:1]) + name[1:] + if !isProtected(mixed) { + t.Errorf("isProtected(%q) = false, want true for mixed-case spelling of protected program", mixed) + } + } +} diff --git a/internal/monitor/monitor.go b/internal/monitor/monitor.go index 74265e9..daa7ecd 100644 --- a/internal/monitor/monitor.go +++ b/internal/monitor/monitor.go @@ -12,8 +12,11 @@ import ( "github.com/hexbytedev/hexwall/internal/allowlist" "github.com/hexbytedev/hexwall/internal/deghost" + "github.com/hexbytedev/hexwall/internal/enforcer" + "github.com/hexbytedev/hexwall/internal/pihole" "github.com/hexbytedev/hexwall/internal/somo" "github.com/hexbytedev/hexwall/internal/store" + "github.com/hexbytedev/hexwall/internal/zeek" ) const ( @@ -23,6 +26,18 @@ const ( ModeEnforce = "enforce" ) +// scanSummary tracks per-scan outcome counts for the summary line. +type scanSummary struct { + connections int + allowlisted int + withSNI int + withoutSNI int + trusted int + blocked int + unknown int + errored int +} + func normalizeMode(mode string) string { switch strings.TrimSpace(strings.ToLower(mode)) { case ModeEnforce: @@ -58,32 +73,63 @@ func remoteIP(address string) (net.IP, error) { return ip, nil } -func logScanConnection(debug bool, ip, program, status string) { - if !debug { +// HandleZeekEvent evaluates a Zeek notice event against the current policy and persists it. +func HandleZeekEvent(checker *pihole.Checker, hexwallStore *store.Store, mode string, event zeek.Event) { + if checker == nil || hexwallStore == nil { + return + } + if event.SNI == "" { return } - slog.Info("scan connection", "ip", ip, "program", program, "status", status) + selectedMode := normalizeMode(mode) + blocked, err := checker.IsBlockedByPolicy(event.SNI) + if err != nil { + slog.Warn("zeek policy lookup failed", "sni", event.SNI, "err", err) + return + } + + confidence := "medium" + actionTaken := "logged" + if blocked { + confidence = "high" + slog.Warn("zeek bypass alert", "src_ip", event.SrcIP, "dst_ip", event.DstIP, "dst_port", event.DstPort, "sni", event.SNI, "mode", selectedMode) + } else { + slog.Info("zeek notice was not policy-blocked", "src_ip", event.SrcIP, "dst_ip", event.DstIP, "dst_port", event.DstPort, "sni", event.SNI) + } + + if err := hexwallStore.LogZeekAlert(event.SrcIP, event.DstIP, event.DstPort, event.SNI, blocked, confidence, actionTaken); err != nil { + slog.Error("failed to persist zeek alert", "src_ip", event.SrcIP, "sni", event.SNI, "err", err) + } + + if selectedMode == ModeEnforce && blocked { + slog.Info("enforce mode: zeek alert recorded without direct connection kill", "src_ip", event.SrcIP, "sni", event.SNI) + } } -// RunScan inspects established connections and applies the selected trust and kill policy. -func RunScan(ctx context.Context, hexwallStore *store.Store, deghostClient *deghost.Client, mode string, debug bool) { +// RunScan inspects established connections and applies the decision ladder. +// +// The ladder is ordered from cheapest/most-authoritative to most-expensive/least-authoritative: +// 1. Static CIDR allowlist (loopback, RFC1918, cloud metadata) +// 2. Zeek SNI presence — if no SNI, fall back to IP-only evaluation (step 7) +// 3. Pi-hole explicit allowlist (user intent, highest authority) +// 4. Pi-hole explicit denylist / gravity (user intent) +// 5. Inferred trust from DNS history (hexwall's own cache) +// 6. Unknown domain → deghost escalation if enabled; no action if disabled +// 7. IP-only fallback → deghost escalation if enabled; no action if disabled +// +// Safety property: with deghost disabled, only rung 4 (Pi-hole blocklist match) can kill. +func RunScan(ctx context.Context, checker *pihole.Checker, hexwallStore *store.Store, deghostClient *deghost.Client, deghostIP bool, deghostDomain bool, zeekClient *zeek.Client, mode string, debug bool) { if hexwallStore == nil { slog.Error("scan aborted: nil hexwall store") return } - if deghostClient == nil { - slog.Error("scan aborted: nil deghost client") - return - } selectedMode := normalizeMode(mode) if selectedMode != mode { slog.Warn("invalid scan mode; defaulting to watch", "mode", mode, "fallback", selectedMode) } - fmt.Printf("[%s] Scanning connections (%s mode)...\n", time.Now().Format("15:04:05"), selectedMode) - connections, err := somo.GetEstablishedConnections() if err != nil { slog.Error("error fetching connections", "err", err) @@ -95,108 +141,301 @@ func RunScan(ctx context.Context, hexwallStore *store.Store, deghostClient *degh return } + var summary scanSummary + summary.connections = len(connections) + for _, conn := range connections { ip, err := remoteIP(conn.RAddress) if err != nil { slog.Warn("invalid IP address", "address", conn.RAddress, "err", err) + summary.errored++ continue } ipStr := ip.String() - // Trust allowlisted IPs immediately. + // Rung 1: static CIDR allowlist. if allowlist.Contains(ip) { - logScanConnection(debug, ipStr, conn.Program, "allowed") + summary.allowlisted++ + summary.trusted++ continue } - allowed, err := hexwallStore.IsAllowed(ipStr) - if err != nil { - slog.Error("store lookup failed", "address", conn.RAddress, "err", err) - continue + // Rung 2: did Zeek see an SNI for this connection? + var sniDomain string + sniFound := false + if zeekClient != nil { + if domain, ok := zeekClient.Lookup(conn.LPort, ipStr, conn.RPort); ok { + sniDomain = domain + sniFound = true + summary.withSNI++ + } else { + summary.withoutSNI++ + } + } else { + summary.withoutSNI++ } - if allowed { - logScanConnection(debug, ipStr, conn.Program, "allowed") - // Keep long-running connections trusted after their Pi-hole refresh window expires. - if err := hexwallStore.UpdateEstablished(ipStr); err != nil { - slog.Error("failed to update established", "address", conn.RAddress, "err", err) - } + if !sniFound { + // Rung 7: IP-only fallback — no SNI means non-TLS, pre-existing connection, or ECH. + action := evaluateIPOnly(ctx, hexwallStore, deghostClient, deghostIP, ipStr, conn, selectedMode, debug) + applyOutcome(&summary, action) continue } - cachedFraudCheck, err := hexwallStore.GetRecentFraudCheck(ipStr) + domain := strings.ToLower(strings.TrimSpace(sniDomain)) + + // Rung 3: Pi-hole explicit allowlist — user intent outranks everything inferred. + allowed, err := checker.IsAllowedByPolicy(domain) if err != nil { - slog.Error("fraud cache lookup failed", "ip", ipStr, "program", conn.Program, "err", err) + // Fail-open: if we cannot determine whether a domain is explicitly allowed, + // we must not act on a block from rung 4. Rung 4 is the only rung that can + // kill a connection, and killing a legitimate connection on the basis of + // incomplete policy data is worse than missing one block. + slog.Warn("domain allowlist lookup failed, skipping blocklist check", "domain", domain, "err", err) + } else if allowed { + if logErr := hexwallStore.LogSNIObservation(ipStr, domain, conn.LPort, store.OutcomePolicyAllowed); logErr != nil { + slog.Debug("failed to log sni observation", "ip", ipStr, "err", logErr) + } + summary.trusted++ continue } - if cachedFraudCheck != nil { - if !cachedFraudCheck.ShouldKill { - slog.Info("unrecognized but clean ip", "ip", ipStr, "program", conn.Program, "reason", "cached-fraud-check") - logScanConnection(debug, ipStr, conn.Program, "unrecognized-clean") + // Rung 4: Pi-hole explicit denylist / gravity — only rung that can kill. + if err == nil { + if blocked, err := checker.IsBlockedByPolicy(domain); err != nil { + slog.Warn("domain denylist lookup failed", "domain", domain, "err", err) + } else if blocked { + if logErr := hexwallStore.LogSNIObservation(ipStr, domain, conn.LPort, store.OutcomePolicyBlocked); logErr != nil { + slog.Debug("failed to log sni observation", "ip", ipStr, "err", logErr) + } + slog.Warn("policy-blocked domain", "domain", domain, "address", conn.RAddress, "pid", conn.PID, "program", conn.Program) + applyBlockedOutcome(hexwallStore, selectedMode, ipStr, conn) + summary.blocked++ continue } + } - logScanConnection(debug, ipStr, conn.Program, "vulnerable") - slog.Warn("vulnerable connection detected", "address", conn.RAddress, "pid", conn.PID, "program", conn.Program, "reason", "cached-fraud-check") - if selectedMode == ModeWatch { - slog.Warn("watch mode: would kill connection", "address", conn.RAddress, "pid", conn.PID, "program", conn.Program) - continue + // Rung 5: inferred trust from DNS history — scoped to this IP. + known := false + knownDomains, err := hexwallStore.DomainsForIP(ipStr) + if err != nil { + slog.Debug("sni domain check failed", "domain", domain, "ip", ipStr, "err", err) + } else { + for _, knownDomain := range knownDomains { + if pihole.DomainMatches(domain, knownDomain) { + known = true + break + } } + } + outcome := store.OutcomeUnknown + if known { + outcome = store.OutcomeIPDomainMatch + } + if logErr := hexwallStore.LogSNIObservation(ipStr, domain, conn.LPort, outcome); logErr != nil { + slog.Debug("failed to log sni observation", "ip", ipStr, "err", logErr) + } - if err := hexwallStore.LogKill(ipStr, conn.PID, conn.Program); err != nil { - slog.Error("failed to log kill", "address", conn.RAddress, "err", err) - } - if err := somo.KillConnection(conn.PID); err != nil { - slog.Error("failed to kill connection", "address", conn.RAddress, "pid", conn.PID, "err", err) - } else { - slog.Info("killed connection", "address", conn.RAddress) + if known { + summary.trusted++ + continue + } + + // Rung 6: unknown domain — deghost escalation if enabled. + if !deghostDomain { + if debug { + slog.Info("unknown domain, deghost disabled; no action", "domain", domain, "ip", ipStr, "program", conn.Program) } + summary.unknown++ continue } - report, err := deghostClient.CheckIP(ctx, ipStr) - if err != nil { - slog.Error("deghost check failed", "ip", ipStr, "program", conn.Program, "err", err) + shouldBlock, reason := false, "" + + if cached, err := hexwallStore.GetRecentDomainCheck(domain); err != nil { + slog.Error("domain cache lookup failed", "domain", domain, "err", err) + summary.errored++ continue + } else if cached != nil { + shouldBlock = cached.ShouldBlock + reason = "cached-domain-check" + } + + if !shouldBlock && reason == "" { + report, err := deghostClient.CheckDomain(ctx, domain) + if err != nil { + slog.Error("deghost domain check failed", "domain", domain, "program", conn.Program, "err", err) + summary.errored++ + continue + } + + shouldBlock = deghost.ShouldBlockDomain(report) + reason = "live-domain-check" + if err := hexwallStore.UpsertDomainCheck(domain, shouldBlock); err != nil { + slog.Error("failed to cache domain check", "domain", domain, "program", conn.Program, "err", err) + } } - if report == nil { - if err := hexwallStore.UpsertFraudCheck(ipStr, false); err != nil { - slog.Error("failed to cache fraud check", "ip", ipStr, "program", conn.Program, "err", err) + if !shouldBlock { + if debug { + slog.Info("unknown but clean domain", "domain", domain, "ip", ipStr, "program", conn.Program, "reason", reason) } - slog.Info("unrecognized but clean ip", "ip", ipStr, "program", conn.Program, "reason", "403/private-or-reserved") - logScanConnection(debug, ipStr, conn.Program, "unrecognized-clean") + summary.unknown++ continue } - shouldKill := deghost.ShouldKill(report) - if err := hexwallStore.UpsertFraudCheck(ipStr, shouldKill); err != nil { - slog.Error("failed to cache fraud check", "ip", ipStr, "program", conn.Program, "err", err) + slog.Warn("deghost-blocked domain", "domain", domain, "address", conn.RAddress, "pid", conn.PID, "program", conn.Program, "reason", reason) + applyBlockedOutcome(hexwallStore, selectedMode, ipStr, conn) + summary.blocked++ + } + + fmt.Printf("[%s] Scan complete: %d connections, %d trusted, %d blocked, %d unknown (allowlisted=%d sni=%d no-sni=%d errored=%d)\n", + time.Now().Format("15:04:05"), summary.connections, summary.trusted, summary.blocked, summary.unknown, summary.allowlisted, summary.withSNI, summary.withoutSNI, summary.errored) +} + +// evaluateIPOnly handles the IP-only fallback path (rung 7) when no SNI is available. +// It returns "trusted", "blocked", "unknown", or "error". +// Every exit path records an IP observation before returning. +func evaluateIPOnly(ctx context.Context, hexwallStore *store.Store, deghostClient *deghost.Client, deghostIP bool, ipStr string, conn somo.Connection, mode string, debug bool) string { + allowed, err := hexwallStore.IsAllowed(ipStr) + if err != nil { + slog.Error("store lookup failed", "address", conn.RAddress, "err", err) + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPUnknown); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) } + return "error" + } - if !shouldKill { - slog.Info("unrecognized but clean ip", "ip", ipStr, "program", conn.Program) - logScanConnection(debug, ipStr, conn.Program, "unrecognized-clean") - continue + if allowed { + if err := hexwallStore.UpdateEstablished(ipStr); err != nil { + slog.Error("failed to update established", "address", conn.RAddress, "err", err) } + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPTrusted); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + return "trusted" + } - logScanConnection(debug, ipStr, conn.Program, "vulnerable") + if !deghostIP { + if debug { + slog.Info("unknown IP, deghost disabled; no action", "ip", ipStr, "program", conn.Program) + } + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPUnknown); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + return "unknown" + } - slog.Warn("vulnerable connection detected", "address", conn.RAddress, "pid", conn.PID, "program", conn.Program) - if selectedMode == ModeWatch { - slog.Warn("watch mode: would kill connection", "address", conn.RAddress, "pid", conn.PID, "program", conn.Program) - continue + cachedFraudCheck, err := hexwallStore.GetRecentFraudCheck(ipStr) + if err != nil { + slog.Error("fraud cache lookup failed", "ip", ipStr, "program", conn.Program, "err", err) + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPUnknown); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) } + return "error" + } - if err := hexwallStore.LogKill(ipStr, conn.PID, conn.Program); err != nil { - slog.Error("failed to log kill", "address", conn.RAddress, "err", err) + if cachedFraudCheck != nil { + if !cachedFraudCheck.ShouldKill { + if debug { + slog.Info("unknown but clean IP (cached)", "ip", ipStr, "program", conn.Program) + } + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPUnknown); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + return "unknown" } - if err := somo.KillConnection(conn.PID); err != nil { - slog.Error("failed to kill connection", "address", conn.RAddress, "pid", conn.PID, "err", err) - } else { - slog.Info("killed connection", "address", conn.RAddress) + + slog.Warn("deghost-blocked IP (cached)", "address", conn.RAddress, "pid", conn.PID, "program", conn.Program) + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPBlocked); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + applyBlockedOutcome(hexwallStore, mode, ipStr, conn) + return "blocked" + } + + report, err := deghostClient.CheckIP(ctx, ipStr) + if err != nil { + slog.Error("deghost check failed", "ip", ipStr, "program", conn.Program, "err", err) + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPUnknown); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + return "error" + } + + if report == nil { + if err := hexwallStore.UpsertFraudCheck(ipStr, false); err != nil { + slog.Error("failed to cache fraud check", "ip", ipStr, "program", conn.Program, "err", err) + } + if debug { + slog.Info("unknown but clean IP (private/reserved)", "ip", ipStr, "program", conn.Program) } + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPReserved); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + return "unknown" + } + + shouldKill := deghost.ShouldKill(report) + if err := hexwallStore.UpsertFraudCheck(ipStr, shouldKill); err != nil { + slog.Error("failed to cache fraud check", "ip", ipStr, "program", conn.Program, "err", err) + } + + if !shouldKill { + if debug { + slog.Info("unknown but clean IP", "ip", ipStr, "program", conn.Program) + } + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPUnknown); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + return "unknown" + } + + slog.Warn("deghost-blocked IP", "address", conn.RAddress, "pid", conn.PID, "program", conn.Program) + if logErr := hexwallStore.LogIPObservation(ipStr, conn.Program, store.OutcomeIPBlocked); logErr != nil { + slog.Debug("failed to log ip observation", "ip", ipStr, "err", logErr) + } + applyBlockedOutcome(hexwallStore, mode, ipStr, conn) + return "blocked" +} + +// applyBlockedOutcome enforces the kill/watch policy for a blocked connection. +func applyBlockedOutcome(hexwallStore *store.Store, mode, ipStr string, conn somo.Connection) { + selectedMode := normalizeMode(mode) + if selectedMode == ModeWatch { + slog.Warn("watch mode: would kill connection", "address", conn.RAddress, "pid", conn.PID, "program", conn.Program) + return + } + + if err := hexwallStore.LogKill(ipStr, conn.PID, conn.Program); err != nil { + slog.Error("failed to log kill", "address", conn.RAddress, "err", err) + } + + // Connection-level termination via `ss -K` (requires CONFIG_INET_DIAG_DESTROY) + // is deferred work. KillProcess terminates the owning process, which is coarser. + pidInt, err := enforcer.ParsePID(conn.PID) + if err != nil { + slog.Warn("invalid PID, skipping kill", "pid", conn.PID, "address", conn.RAddress, "err", err) + return + } + if err := enforcer.KillProcess(pidInt, conn.Program, conn.RAddress); err != nil { + slog.Warn("failed to kill process", "address", conn.RAddress, "pid", conn.PID, "err", err) + } else { + slog.Info("killed connection", "address", conn.RAddress) + } +} + +// applyOutcome increments the appropriate summary counter for a given outcome string. +func applyOutcome(summary *scanSummary, outcome string) { + switch outcome { + case "trusted": + summary.trusted++ + case "blocked": + summary.blocked++ + case "unknown": + summary.unknown++ + case "error": + summary.errored++ } } diff --git a/internal/pihole/cache.go b/internal/pihole/cache.go index 12df0ff..5979c4b 100644 --- a/internal/pihole/cache.go +++ b/internal/pihole/cache.go @@ -53,13 +53,14 @@ func (c *IPCache) Refresh(ctx context.Context) { sort.Strings(domains) - type result struct { - domain string - ips []string - } - results := make(chan result, maxConcurrentLookups) sem := make(chan struct{}, maxConcurrentLookups) + // Workers write straight into resolved under mu. Using a shared map instead of a + // results channel keeps a worker from ever blocking on a consumer that has not + // started yet, which would otherwise pin its slot in sem and stall the spawn loop. + var mu sync.Mutex + resolved := make(map[string][]string, len(domains)) + var wg sync.WaitGroup spawnLoop: for _, domain := range domains { @@ -82,26 +83,24 @@ spawnLoop: // DNS failure is normal for blocked domains, expired records, and similar cases. return } - results <- result{domain: d, ips: ips} + + mu.Lock() + resolved[d] = ips + mu.Unlock() }(domain) } - go func() { - wg.Wait() - close(results) - }() - - resolved := make(map[string][]string, len(domains)) - for r := range results { - resolved[r.domain] = r.ips - } + wg.Wait() if ctx.Err() != nil { return } - // The store keeps one domain per IP, so keep the first domain in a stable order. - uniqueIPs := make(map[string]string, len(resolved)) + // A shared CDN edge IP serves many unrelated domains, so keep the full set per IP. + // domains is sorted, so each IP's slice stays in a stable order and its first entry + // is the one recorded as the representative domain in allowed_ips. + ipDomains := make(map[string][]string, len(resolved)) + seen := make(map[string]struct{}, len(resolved)) for _, domain := range domains { ips, ok := resolved[domain] if !ok { @@ -109,23 +108,34 @@ spawnLoop: } for _, ip := range ips { - if _, exists := uniqueIPs[ip]; exists { + key := ip + "\x00" + domain + if _, exists := seen[key]; exists { continue } - uniqueIPs[ip] = domain + seen[key] = struct{}{} + ipDomains[ip] = append(ipDomains[ip], domain) } } var totalIPs int - for ip, domain := range uniqueIPs { - if err := c.store.UpsertAllowedIP(ip, domain); err != nil { - slog.Error("cache refresh: failed to upsert IP", "ip", ip, "domain", domain, "err", err) + var totalPairs int + for ip, ipDomainSet := range ipDomains { + if err := c.store.UpsertAllowedIP(ip, ipDomainSet[0]); err != nil { + slog.Error("cache refresh: failed to upsert IP", "ip", ip, "domain", ipDomainSet[0], "err", err) } else { totalIPs++ } + + for _, domain := range ipDomainSet { + if err := c.store.UpsertAllowedIPDomain(ip, domain); err != nil { + slog.Error("cache refresh: failed to upsert IP domain", "ip", ip, "domain", domain, "err", err) + } else { + totalPairs++ + } + } } - slog.Info("cache refreshed", "domains", len(domains), "ips", totalIPs) + slog.Info("cache refreshed", "domains", len(domains), "ips", totalIPs, "pairs", totalPairs) } // RunRefresh calls Refresh on the given interval until ctx is cancelled. diff --git a/internal/pihole/domain_match_test.go b/internal/pihole/domain_match_test.go new file mode 100644 index 0000000..902e216 --- /dev/null +++ b/internal/pihole/domain_match_test.go @@ -0,0 +1,42 @@ +package pihole + +import ( + "testing" +) + +func TestDomainMatches(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + observed string + known string + want bool + }{ + {name: "exact match", observed: "github.com", known: "github.com", want: true}, + {name: "subdomain observed", observed: "www.github.com", known: "github.com", want: true}, + {name: "subdomain known", observed: "github.com", known: "www.github.com", want: true}, + {name: "deeper subdomain", observed: "a.b.github.com", known: "github.com", want: true}, + {name: "deeper subdomain reversed", observed: "github.com", known: "a.b.github.com", want: true}, + {name: "lookalike suffix is not a match", observed: "github.com.evil.com", known: "github.com", want: false}, + {name: "lookalike suffix reversed is not a match", observed: "github.com", known: "github.com.evil.com", want: false}, + {name: "partial label is not a match", observed: "notgithub.com", known: "github.com", want: false}, + {name: "unrelated domains", observed: "example.org", known: "github.com", want: false}, + {name: "empty observed", observed: "", known: "github.com", want: false}, + {name: "empty known", observed: "github.com", known: "", want: false}, + {name: "both empty", observed: "", known: "", want: false}, + {name: "whitespace only", observed: " ", known: "github.com", want: false}, + {name: "case and whitespace normalized", observed: " GitHub.COM ", known: "github.com", want: true}, + {name: "case and whitespace normalized subdomain", observed: " WWW.GitHub.com ", known: " GITHUB.com ", want: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + if got := DomainMatches(tt.observed, tt.known); got != tt.want { + t.Fatalf("DomainMatches(%q, %q) = %v, want %v", tt.observed, tt.known, got, tt.want) + } + }) + } +} diff --git a/internal/pihole/pihole.go b/internal/pihole/pihole.go index 028e834..d5fd0c2 100644 --- a/internal/pihole/pihole.go +++ b/internal/pihole/pihole.go @@ -6,8 +6,10 @@ import ( "database/sql" "errors" "fmt" + "log/slog" "net/url" "os" + "path/filepath" "strings" "time" @@ -17,12 +19,14 @@ import ( // Config holds the Pi-hole database configuration. type Config struct { - DBPath string + DBPath string + GravityDBPath string // optional; derived from DBPath when empty } -// Checker reads Pi-hole query history from the FTL database. +// Checker reads Pi-hole query history from the FTL database and policy lists from gravity. type Checker struct { - db *sql.DB + db *sql.DB + gravityDB *sql.DB } const domainLookback = time.Hour @@ -67,14 +71,64 @@ func NewChecker(config *Config) (*Checker, error) { return nil, fmt.Errorf("failed to connect to pihole-db: %w", err) } + // Open gravity.db for policy list lookups. + gravityPath := config.GravityDBPath + if gravityPath == "" { + gravityPath = filepath.Join(filepath.Dir(config.DBPath), "gravity.db") + } + + var gravityDB *sql.DB + if err := openGravityDB(gravityPath, &gravityDB); err != nil { + slog.Warn("pi-hole gravity database unavailable; policy checks disabled", + "path", gravityPath, "err", err) + } + return &Checker{ - db: db, + db: db, + gravityDB: gravityDB, }, nil } -// Close closes the Checker database connection. +// openGravityDB opens the Pi-hole gravity database in read-only mode. +// On success it sets *out; on failure it returns an error but does not touch *out. +func openGravityDB(path string, out **sql.DB) error { + file, err := os.Open(path) + if err != nil { + return fmt.Errorf("cannot access %s: %w", path, err) + } + if err := file.Close(); err != nil { + return fmt.Errorf("close access check for %s: %w", path, err) + } + + dsn := "file:" + (&url.URL{Path: path, RawQuery: "mode=ro"}).String() + db, err := sql.Open("sqlite", dsn) + if err != nil { + return fmt.Errorf("failed to open gravity-db: %w", err) + } + + if err := db.Ping(); err != nil { + _ = db.Close() + return fmt.Errorf("failed to connect to gravity-db: %w", err) + } + + *out = db + return nil +} + +// Close closes both database connections. func (c *Checker) Close() error { - return c.db.Close() + if c == nil { + return nil + } + var errFTL error + var errGravity error + if c.db != nil { + errFTL = c.db.Close() + } + if c.gravityDB != nil { + errGravity = c.gravityDB.Close() + } + return errors.Join(errFTL, errGravity) } // IsDomainKnown reports whether domain appeared in Pi-hole query history within the last hour. @@ -106,6 +160,92 @@ func (c *Checker) IsDomainKnown(domain string) (bool, error) { return true, nil } +// IsAllowedByPolicy reports whether the domain currently appears in Pi-hole's allowlist. +func (c *Checker) IsAllowedByPolicy(domain string) (bool, error) { + domain = normalizeDomain(domain) + if domain == "" { + return false, nil + } + if c == nil || c.gravityDB == nil { + return false, nil + } + + var found int + err := c.gravityDB.QueryRow(` + SELECT 1 FROM vw_allowlist + WHERE domain = ? COLLATE NOCASE + LIMIT 1 + `, domain).Scan(&found) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("allowlist lookup failed: %w", err) + } + + return true, nil +} + +// IsBlockedByPolicy reports whether the domain currently matches a Pi-hole denylist or gravity entry. +func (c *Checker) IsBlockedByPolicy(domain string) (bool, error) { + domain = normalizeDomain(domain) + if domain == "" { + return false, nil + } + if c == nil || c.gravityDB == nil { + return false, nil + } + + var found int + err := c.gravityDB.QueryRow(` + SELECT 1 FROM vw_denylist + WHERE domain = ? COLLATE NOCASE + LIMIT 1 + `, domain).Scan(&found) + if err == nil { + return true, nil + } + if !errors.Is(err, sql.ErrNoRows) { + return false, fmt.Errorf("denylist lookup failed: %w", err) + } + + err = c.gravityDB.QueryRow(` + SELECT 1 FROM vw_gravity + WHERE domain = ? COLLATE NOCASE + LIMIT 1 + `, domain).Scan(&found) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("gravity lookup failed: %w", err) + } + + return true, nil +} + +// PolicyCounts returns the number of entries in each Pi-hole policy view. +// The returned map is keyed by view name. Returns nil if gravity DB is unavailable. +func (c *Checker) PolicyCounts() (map[string]int64, error) { + if c == nil || c.gravityDB == nil { + return nil, nil + } + + views := []string{"vw_allowlist", "vw_denylist", "vw_gravity", "vw_regex_allowlist", "vw_regex_denylist"} + counts := make(map[string]int64, len(views)) + + for _, view := range views { + var count int64 + err := c.gravityDB.QueryRow(fmt.Sprintf(`SELECT COUNT(*) FROM %s`, view)).Scan(&count) + if err != nil { + return nil, fmt.Errorf("count %s: %w", view, err) + } + counts[view] = count + } + + return counts, nil +} + // DomainsSeenSince returns the distinct domains Pi-hole recorded since the given Unix timestamp. func (c *Checker) DomainsSeenSince(since int64) ([]string, error) { rows, err := c.db.Query(` @@ -143,6 +283,34 @@ func (c *Checker) DomainsSeenSince(since int64) ([]string, error) { return domains, nil } +// DomainMatches reports whether an observed SNI domain corresponds to a known domain, +// treating a subdomain as a match for its parent in either direction. +// +// The direction is intentionally symmetric: Zeek reports the SNI exactly as sent on the wire +// (www.github.com) while the store holds whatever Pi-hole logged and the cache resolved +// (github.com), and either can be the more specific name. +// +// The match requires a label boundary, so github.com.evil.com does not match github.com: +// it does not end in ".github.com". A plain suffix test would have accepted it. +// +// A registrable-domain (eTLD+1) comparison via golang.org/x/net/publicsuffix would handle more +// edge cases -- notably unrelated names under a shared public suffix -- but it adds a third-party +// dependency for a marginal gain, so the label-boundary rule is the deliberate v1 tradeoff. +func DomainMatches(observed, known string) bool { + observed = normalizeDomain(observed) + known = normalizeDomain(known) + + if observed == "" || known == "" { + return false + } + + if observed == known { + return true + } + + return strings.HasSuffix(observed, "."+known) || strings.HasSuffix(known, "."+observed) +} + func normalizeDomain(domain string) string { return strings.ToLower(strings.TrimSpace(domain)) } diff --git a/internal/pihole/policy_test.go b/internal/pihole/policy_test.go new file mode 100644 index 0000000..fb722b9 --- /dev/null +++ b/internal/pihole/policy_test.go @@ -0,0 +1,531 @@ +package pihole + +import ( + "database/sql" + "net/url" + "path/filepath" + "testing" + + _ "modernc.org/sqlite" +) + +func TestNormalizeDomain(t *testing.T) { + if got := normalizeDomain(" Example.COM "); got != "example.com" { + t.Fatalf("normalizeDomain() = %q, want %q", got, "example.com") + } +} + +// openTestGravityDB creates a temporary SQLite database with Pi-hole-like views +// and returns the opened DB along with its path. The caller must close it. +func openTestGravityDB(t *testing.T) (*sql.DB, string) { + t.Helper() + dir := t.TempDir() + dbPath := filepath.Join(dir, "gravity.db") + + db, err := sql.Open("sqlite", "file:"+(&url.URL{Path: dbPath, RawQuery: "mode=rwc"}).String()) + if err != nil { + t.Fatalf("open test gravity db: %v", err) + } + + // Create backing tables and views matching Pi-hole v6 shapes. + schema := ` + CREATE TABLE adlist (id INTEGER PRIMARY KEY, address TEXT, enabled INTEGER, date_added INTEGER, date_modified INTEGER, comment TEXT, type INTEGER, number INTEGER); + CREATE TABLE domainlist (id INTEGER PRIMARY KEY, type INTEGER, domain TEXT, enabled INTEGER, date_added INTEGER, date_modified INTEGER, comment TEXT, groups TEXT); + CREATE TABLE gravity (id INTEGER PRIMARY KEY, domain TEXT, adlist_id INTEGER); + CREATE INDEX idx_gravity ON gravity (domain, adlist_id); + + CREATE VIEW vw_allowlist AS SELECT domain FROM domainlist WHERE type = 0 AND enabled = 1 AND TRIM(domain) <> ''; + CREATE VIEW vw_denylist AS SELECT domain FROM domainlist WHERE type = 1 AND enabled = 1 AND TRIM(domain) <> ''; + CREATE VIEW vw_gravity AS SELECT domain FROM gravity; + CREATE VIEW vw_regex_allowlist AS SELECT domain FROM domainlist WHERE type = 2 AND enabled = 1 AND TRIM(domain) <> ''; + CREATE VIEW vw_regex_denylist AS SELECT domain FROM domainlist WHERE type = 3 AND enabled = 1 AND TRIM(domain) <> ''; + ` + if _, err := db.Exec(schema); err != nil { + _ = db.Close() + t.Fatalf("create schema: %v", err) + } + + return db, dbPath +} + +// insertDomain inserts a domain into the domainlist table with the given type. +// type 0 = allowlist, type 1 = denylist, type 2 = regex allowlist, type 3 = regex denylist. +func insertDomain(t *testing.T, db *sql.DB, domain string, listType int) { + t.Helper() + _, err := db.Exec(`INSERT INTO domainlist (type, domain, enabled) VALUES (?, ?, 1)`, listType, domain) + if err != nil { + t.Fatalf("insert domain %q type %d: %v", domain, listType, err) + } +} + +// insertGravity inserts a domain into the gravity table. +func insertGravity(t *testing.T, db *sql.DB, domain string) { + t.Helper() + _, err := db.Exec(`INSERT INTO gravity (domain, adlist_id) VALUES (?, 1)`, domain) + if err != nil { + t.Fatalf("insert gravity domain %q: %v", domain, err) + } +} + +func newTestChecker(t *testing.T, gravityPath string) *Checker { + t.Helper() + ftlPath := filepath.Join(t.TempDir(), "pihole-FTL.db") + db, err := sql.Open("sqlite", "file:"+(&url.URL{Path: ftlPath, RawQuery: "mode=rwc"}).String()) + if err != nil { + t.Fatalf("open test FTL db: %v", err) + } + if _, err := db.Exec(`CREATE TABLE queries (id INTEGER PRIMARY KEY, timestamp INTEGER, type INTEGER, domain TEXT, status INTEGER, ip TEXT)`); err != nil { + _ = db.Close() + t.Fatalf("create FTL schema: %v", err) + } + + var gravityDB *sql.DB + if gravityPath != "" { + gravityDB, err = sql.Open("sqlite", "file:"+(&url.URL{Path: gravityPath, RawQuery: "mode=ro"}).String()) + if err != nil { + _ = db.Close() + t.Fatalf("open gravity db: %v", err) + } + } + + return &Checker{db: db, gravityDB: gravityDB} +} + +func TestIsBlockedByPolicy_Denylist(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "blocked.example.com", 1) // type 1 = denylist + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + blocked, err := checker.IsBlockedByPolicy("blocked.example.com") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if !blocked { + t.Fatal("IsBlockedByPolicy() = false, want true for denylist domain") + } +} + +func TestIsBlockedByPolicy_Gravity(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertGravity(t, gravDB, "malware.example.net") + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + blocked, err := checker.IsBlockedByPolicy("malware.example.net") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if !blocked { + t.Fatal("IsBlockedByPolicy() = false, want true for gravity domain") + } +} + +func TestIsBlockedByPolicy_NoMatch(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertGravity(t, gravDB, "other.example.com") + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + blocked, err := checker.IsBlockedByPolicy("safe.example.org") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if blocked { + t.Fatal("IsBlockedByPolicy() = true, want false for unknown domain") + } +} + +func TestIsAllowedByPolicy_Allowlist(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "trusted.example.com", 0) // type 0 = allowlist + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + allowed, err := checker.IsAllowedByPolicy("trusted.example.com") + if err != nil { + t.Fatalf("IsAllowedByPolicy() error = %v", err) + } + if !allowed { + t.Fatal("IsAllowedByPolicy() = false, want true for allowlist domain") + } +} + +func TestIsAllowedByPolicy_NoMatch(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "trusted.example.com", 0) + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + allowed, err := checker.IsAllowedByPolicy("unknown.example.org") + if err != nil { + t.Fatalf("IsAllowedByPolicy() error = %v", err) + } + if allowed { + t.Fatal("IsAllowedByPolicy() = true, want false for unknown domain") + } +} + +func TestPolicyLookup_CaseNormalization(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "CaseSensitive.COM", 1) + insertGravity(t, gravDB, "GravityDomain.ORG") + insertDomain(t, gravDB, "AllowCase.COM", 0) + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + // Denylist: case-insensitive match + blocked, err := checker.IsBlockedByPolicy(" casesensitive.com ") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if !blocked { + t.Fatal("IsBlockedByPolicy() = false, want true for case-insensitive denylist match") + } + + // Gravity: case-insensitive match + blocked, err = checker.IsBlockedByPolicy("GRAVITYDOMAIN.ORG") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if !blocked { + t.Fatal("IsBlockedByPolicy() = false, want true for case-insensitive gravity match") + } + + // Allowlist: case-insensitive match + allowed, err := checker.IsAllowedByPolicy("allowcase.com") + if err != nil { + t.Fatalf("IsAllowedByPolicy() error = %v", err) + } + if !allowed { + t.Fatal("IsAllowedByPolicy() = false, want true for case-insensitive allowlist match") + } +} + +func TestPolicyLookup_EmptyDomain(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "blocked.example.com", 1) + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + blocked, err := checker.IsBlockedByPolicy("") + if err != nil { + t.Fatalf("IsBlockedByPolicy(\"\") error = %v", err) + } + if blocked { + t.Fatal("IsBlockedByPolicy(\"\") = true, want false") + } + + allowed, err := checker.IsAllowedByPolicy("") + if err != nil { + t.Fatalf("IsAllowedByPolicy(\"\") error = %v", err) + } + if allowed { + t.Fatal("IsAllowedByPolicy(\"\") = true, want false") + } +} + +func TestPolicyLookup_NilGravityDB(t *testing.T) { + t.Parallel() + + // Create a checker with nil gravityDB (simulates gravity.db unavailable). + checker := &Checker{ + db: nil, // not needed for these methods + gravityDB: nil, + } + + blocked, err := checker.IsBlockedByPolicy("anything.example.com") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v, want nil", err) + } + if blocked { + t.Fatal("IsBlockedByPolicy() = true, want false for nil gravity DB") + } + + allowed, err := checker.IsAllowedByPolicy("anything.example.com") + if err != nil { + t.Fatalf("IsAllowedByPolicy() error = %v, want nil", err) + } + if allowed { + t.Fatal("IsAllowedByPolicy() = true, want false for nil gravity DB") + } +} + +func TestPolicyCounts(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "allow1.com", 0) + insertDomain(t, gravDB, "allow2.com", 0) + insertDomain(t, gravDB, "deny1.com", 1) + insertGravity(t, gravDB, "g1.com") + insertGravity(t, gravDB, "g2.com") + insertGravity(t, gravDB, "g3.com") + insertDomain(t, gravDB, "regex_allow", 2) + insertDomain(t, gravDB, "regex_deny1", 3) + insertDomain(t, gravDB, "regex_deny2", 3) + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + counts, err := checker.PolicyCounts() + if err != nil { + t.Fatalf("PolicyCounts() error = %v", err) + } + if counts == nil { + t.Fatal("PolicyCounts() = nil, want non-nil map") + } + + want := map[string]int64{ + "vw_allowlist": 2, + "vw_denylist": 1, + "vw_gravity": 3, + "vw_regex_allowlist": 1, + "vw_regex_denylist": 2, + } + for view, wantCount := range want { + if got := counts[view]; got != wantCount { + t.Errorf("PolicyCounts()[%q] = %d, want %d", view, got, wantCount) + } + } +} + +func TestPolicyCounts_NilGravityDB(t *testing.T) { + t.Parallel() + checker := &Checker{gravityDB: nil} + + counts, err := checker.PolicyCounts() + if err != nil { + t.Fatalf("PolicyCounts() error = %v, want nil", err) + } + if counts != nil { + t.Fatalf("PolicyCounts() = %v, want nil for nil gravity DB", counts) + } +} + +func TestGravityDBPath_Derived(t *testing.T) { + t.Parallel() + + // Create the FTL db in a known directory. + dir := t.TempDir() + ftlPath := filepath.Join(dir, "pihole-FTL.db") + db, err := sql.Open("sqlite", "file:"+(&url.URL{Path: ftlPath, RawQuery: "mode=rwc"}).String()) + if err != nil { + t.Fatalf("open FTL db: %v", err) + } + if _, err := db.Exec(`CREATE TABLE queries (id INTEGER PRIMARY KEY)`); err != nil { + _ = db.Close() + t.Fatalf("init FTL db: %v", err) + } + _ = db.Close() + + // Create gravity.db next to it. + gravPath := filepath.Join(dir, "gravity.db") + gravDB, err := sql.Open("sqlite", "file:"+(&url.URL{Path: gravPath, RawQuery: "mode=rwc"}).String()) + if err != nil { + t.Fatalf("open gravity db: %v", err) + } + // Create minimal schema so gravity opens cleanly. + if _, err := gravDB.Exec(`CREATE TABLE adlist (id INTEGER PRIMARY KEY); CREATE VIEW vw_gravity AS SELECT 1 WHERE 0;`); err != nil { + _ = gravDB.Close() + t.Fatalf("create gravity schema: %v", err) + } + _ = gravDB.Close() + + // NewChecker should auto-discover gravity.db next to pihole-FTL.db. + checker, err := NewChecker(&Config{DBPath: ftlPath}) + if err != nil { + t.Fatalf("NewChecker() error = %v", err) + } + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + if checker.gravityDB == nil { + t.Fatal("gravityDB is nil, want it to be auto-discovered") + } +} + +func TestGravityDBPath_Missing(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + ftlPath := filepath.Join(dir, "pihole-FTL.db") + db, err := sql.Open("sqlite", "file:"+(&url.URL{Path: ftlPath, RawQuery: "mode=rwc"}).String()) + if err != nil { + t.Fatalf("open FTL db: %v", err) + } + if _, err := db.Exec(`CREATE TABLE queries (id INTEGER PRIMARY KEY)`); err != nil { + _ = db.Close() + t.Fatalf("init FTL db: %v", err) + } + _ = db.Close() + + // No gravity.db in dir — NewChecker should succeed with nil gravityDB. + checker, err := NewChecker(&Config{DBPath: ftlPath}) + if err != nil { + t.Fatalf("NewChecker() error = %v, want nil when gravity.db is missing", err) + } + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + if checker.gravityDB != nil { + t.Fatal("gravityDB is non-nil, want nil when gravity.db is missing") + } + + // Policy methods should gracefully return false. + blocked, err := checker.IsBlockedByPolicy("anything.com") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if blocked { + t.Fatal("IsBlockedByPolicy() = true, want false when gravity unavailable") + } +} + +func TestCloseJoinsErrors(t *testing.T) { + t.Parallel() + + // Close on a checker with both handles should not panic. + gravDB, gravPath := openTestGravityDB(t) + _ = gravDB.Close() // close it first so Close() returns an error + + checker := newTestChecker(t, gravPath) + // Force gravity DB to nil after construction to test the nil path. + _ = checker.gravityDB.Close() + checker.gravityDB = nil + + err := checker.Close() + if err != nil { + // We expect an error from closing the already-closed FTL db or nil gravity. + // Just verify it doesn't panic. + t.Logf("Close() returned (expected for double-close): %v", err) + } +} + +func TestIsBlockedByPolicy_DenylistBeatsGravity(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "both.example.com", 1) // denylist + insertGravity(t, gravDB, "both.example.com") // also in gravity + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + blocked, err := checker.IsBlockedByPolicy("both.example.com") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if !blocked { + t.Fatal("IsBlockedByPolicy() = false, want true when domain is in both denylist and gravity") + } +} + +func TestIsAllowedByPolicy_TakesPrecedenceOverGravity(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertDomain(t, gravDB, "special.example.com", 0) // allowlist + insertGravity(t, gravDB, "special.example.com") // also in gravity + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + // Allowlist match should be independent of gravity presence. + allowed, err := checker.IsAllowedByPolicy("special.example.com") + if err != nil { + t.Fatalf("IsAllowedByPolicy() error = %v", err) + } + if !allowed { + t.Fatal("IsAllowedByPolicy() = false, want true for allowlisted domain") + } +} + +func TestIsBlockedByPolicy_WhitespaceNormalization(t *testing.T) { + t.Parallel() + gravDB, gravPath := openTestGravityDB(t) + insertGravity(t, gravDB, "trim.example.com") + _ = gravDB.Close() + + checker := newTestChecker(t, gravPath) + defer func() { + if err := checker.Close(); err != nil { + t.Errorf("checker.Close(): %v", err) + } + }() + + blocked, err := checker.IsBlockedByPolicy(" Trim.Example.COM ") + if err != nil { + t.Fatalf("IsBlockedByPolicy() error = %v", err) + } + if !blocked { + t.Fatal("IsBlockedByPolicy() = false, want true with whitespace around domain") + } +} diff --git a/internal/somo/somo.go b/internal/somo/somo.go index 79d9a73..b3bbb94 100644 --- a/internal/somo/somo.go +++ b/internal/somo/somo.go @@ -46,28 +46,6 @@ func GetEstablishedConnections() ([]Connection, error) { return result, nil } -// KillConnection kills a process by PID via somo. -func KillConnection(pid string) error { - pid = strings.TrimSpace(pid) - if pid == "" || pid == "-" { - return fmt.Errorf("invalid PID %q", pid) - } - - for _, r := range pid { - if r < '0' || r > '9' { - return fmt.Errorf("invalid PID %q", pid) - } - } - - cmd := exec.Command("somo", "-k", "-p", pid) - output, err := cmd.CombinedOutput() - if err != nil { - return commandError(fmt.Sprintf("failed to kill connection with PID %s", pid), err, output) - } - - return nil -} - func commandError(message string, err error, output []byte) error { trimmed := strings.TrimSpace(string(output)) if trimmed == "" { diff --git a/internal/store/store.go b/internal/store/store.go index 5b03123..658777a 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -7,7 +7,9 @@ import ( "database/sql" "errors" "fmt" + "log/slog" "net/url" + "strings" "time" // Register the pure-Go SQLite driver used for the local hexwall database. @@ -19,6 +21,12 @@ const ( establishedTrustWindow = time.Minute fraudCheckCacheWindow = 6 * time.Hour sqliteBusyTimeout = 5 * time.Second + + // Prune windows: each table is deleted on a different schedule. + pruneSNIWindow = 7 * 24 * time.Hour // SNI observations: operational state data; 7 days is enough for hand-inspection. + pruneKilledAlertsWindow = 90 * 24 * time.Hour // Audit tables (killed_connections, zeek_alerts): rare events, forensically valuable. + // fraud_checks and domain_checks are bounded by upsert-per-key and fraudCheckCacheWindow (6h); + // they never grow beyond one row per IP/domain and are self-cleaning on read, so pruning is unnecessary. ) const schema = ` @@ -30,6 +38,14 @@ CREATE TABLE IF NOT EXISTS allowed_ips ( last_established INTEGER ); +CREATE TABLE IF NOT EXISTS allowed_ip_domains ( + ip TEXT NOT NULL, + domain TEXT NOT NULL, + first_seen INTEGER NOT NULL, + last_refreshed INTEGER NOT NULL, + PRIMARY KEY (ip, domain) +); + CREATE TABLE IF NOT EXISTS killed_connections ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT NOT NULL, @@ -43,6 +59,53 @@ CREATE TABLE IF NOT EXISTS fraud_checks ( should_kill INTEGER NOT NULL, checked_at INTEGER NOT NULL ); + +CREATE TABLE IF NOT EXISTS sni_observations ( + ip TEXT NOT NULL, + domain TEXT NOT NULL, + local_port TEXT NOT NULL, + outcome TEXT NOT NULL, + first_seen INTEGER NOT NULL, + last_seen INTEGER NOT NULL, + times_seen INTEGER NOT NULL, + PRIMARY KEY (ip, domain, local_port) +); + +CREATE TABLE IF NOT EXISTS zeek_alerts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + src_ip TEXT NOT NULL, + dst_ip TEXT NOT NULL, + dst_port TEXT NOT NULL, + sni TEXT NOT NULL, + blocked INTEGER NOT NULL, + confidence TEXT NOT NULL, + detected_at INTEGER NOT NULL, + action_taken TEXT +); + +CREATE TABLE IF NOT EXISTS domain_checks ( + domain TEXT PRIMARY KEY, + should_block INTEGER NOT NULL, + checked_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS ip_observations ( + ip TEXT NOT NULL, + program TEXT NOT NULL, + outcome TEXT NOT NULL, + first_seen INTEGER NOT NULL, + last_seen INTEGER NOT NULL, + times_seen INTEGER NOT NULL, + PRIMARY KEY (ip, program) +); +` + +// backfillAllowedIPDomains seeds the per-IP domain set from the older IP-level trust table. +// It runs on every startup: INSERT OR IGNORE makes it idempotent, and without it an upgraded +// install would treat every Zeek-observed SNI as a mismatch until the first cache refresh. +const backfillAllowedIPDomains = ` +INSERT OR IGNORE INTO allowed_ip_domains (ip, domain, first_seen, last_refreshed) +SELECT ip, domain, first_approved, last_refreshed FROM allowed_ips WHERE TRIM(domain) <> ''; ` // FraudCheckCacheEntry stores the cached kill decision for a prior fraud API lookup. @@ -51,12 +114,141 @@ type FraudCheckCacheEntry struct { CheckedAt int64 } +// DomainCheckCacheEntry stores the cached block decision for a prior domain reputation lookup. +type DomainCheckCacheEntry struct { + ShouldBlock bool + CheckedAt int64 +} + +// SNIOutcome records which rung of the decision ladder classified a connection. +type SNIOutcome string + +const ( + // OutcomePolicyAllowed indicates the connection was allowed by Pi-hole policy. + OutcomePolicyAllowed SNIOutcome = "policy-allowed" + // OutcomePolicyBlocked indicates the connection was blocked by Pi-hole policy. + OutcomePolicyBlocked SNIOutcome = "policy-blocked" + // OutcomeIPDomainMatch indicates the SNI domain matched a known domain for the IP. + OutcomeIPDomainMatch SNIOutcome = "ip-domain-match" + // OutcomeUnknown indicates the connection could not be classified. + OutcomeUnknown SNIOutcome = "unknown" +) + +// Valid reports whether the outcome is one of the known values. +func (o SNIOutcome) Valid() bool { + switch o { + case OutcomePolicyAllowed, OutcomePolicyBlocked, OutcomeIPDomainMatch, OutcomeUnknown: + return true + } + return false +} + +// ParseSNIOutcome converts a string read from the database into a typed SNIOutcome. +// Rows written by older builds may contain unrecognized values; those parse to OutcomeUnknown with false. +func ParseSNIOutcome(s string) (SNIOutcome, bool) { + o := SNIOutcome(s) + if o.Valid() { + return o, true + } + return OutcomeUnknown, false +} + +// IPOutcome records how a connection was classified on the IP-only fallback path. +type IPOutcome string + +const ( + // OutcomeIPTrusted indicates the IP is present and fresh in allowed_ips. + OutcomeIPTrusted IPOutcome = "ip-trusted" + // OutcomeIPReserved indicates deghost returned 403: private or reserved address. + OutcomeIPReserved IPOutcome = "private-reserved" + // OutcomeIPBlocked indicates deghost verdict said kill. + OutcomeIPBlocked IPOutcome = "deghost-blocked" + // OutcomeIPUnknown indicates no verdict available or check disabled. + OutcomeIPUnknown IPOutcome = "unknown" +) + +// Valid reports whether the outcome is one of the known values. +func (o IPOutcome) Valid() bool { + switch o { + case OutcomeIPTrusted, OutcomeIPReserved, OutcomeIPBlocked, OutcomeIPUnknown: + return true + } + return false +} + +// ParseIPOutcome converts a string read from the database into a typed IPOutcome. +// Rows written by older builds may contain unrecognized values; those parse to OutcomeIPUnknown with false. +func ParseIPOutcome(s string) (IPOutcome, bool) { + o := IPOutcome(s) + if o.Valid() { + return o, true + } + return OutcomeIPUnknown, false +} + +// ZeekAlertEntry stores a Zeek-derived alert that was persisted for audit. +type ZeekAlertEntry struct { + ID int64 + SrcIP string + DstIP string + DstPort string + SNI string + Blocked bool + Confidence string + DetectedAt int64 + ActionTaken string +} + // Store wraps the local hexwall database. type Store struct { readWrite *sql.DB readOnly *sql.DB } +// migrateSNITable detects the old event-log schema (identified by a "known" column) +// and drops it so the new state-table schema is applied by the CREATE TABLE IF NOT EXISTS. +// This is safe specifically because nothing in the codebase reads this table; a future reader +// must not assume this reasoning generalises to other tables. +func migrateSNITable(db *sql.DB) error { + cols, err := db.Query(`PRAGMA table_info(sni_observations)`) + if err != nil { + return fmt.Errorf("query table info: %w", err) + } + defer func() { + if err := cols.Close(); err != nil { + slog.Debug("failed to close columns", "err", err) + } + }() + + hasOldSchema := false + for cols.Next() { + var cid int + var name, ctype string + var notNull int + var dfltValue sql.NullString + var pk int + if err := cols.Scan(&cid, &name, &ctype, ¬Null, &dfltValue, &pk); err != nil { + return fmt.Errorf("scan table info: %w", err) + } + if name == "known" { + hasOldSchema = true + } + } + if err := cols.Err(); err != nil { + return fmt.Errorf("iterate table info: %w", err) + } + + if !hasOldSchema { + return nil + } + + slog.Warn("dropping old sni_observations table (write-only audit data; no readers)", "reason", "schema migration to state-table format") + if _, err := db.Exec(`DROP TABLE sni_observations`); err != nil { + return fmt.Errorf("drop old sni_observations: %w", err) + } + return nil +} + // NewStore opens or creates the hexwall database at dbPath and applies the schema. func NewStore(dbPath string) (*Store, error) { readWrite, err := sql.Open("sqlite", sqliteDSN(dbPath, "rwc")) @@ -74,6 +266,26 @@ func NewStore(dbPath string) (*Store, error) { return nil, fmt.Errorf("failed to apply schema: %w", err) } + // Migration: detect and replace old sni_observations event-log schema with the new state table. + // The old table used an autoincrement id + known INTEGER, which is structurally incompatible + // with the new (ip, domain, local_port) primary key. This is safe to drop because nothing in + // the codebase reads this table — it was only ever written to and consulted by hand. + if err := migrateSNITable(readWrite); err != nil { + _ = readWrite.Close() + return nil, fmt.Errorf("failed to migrate sni_observations: %w", err) + } + + // Re-apply schema after migration so that any table dropped by migrateSNITable is recreated. + if _, err := readWrite.Exec(schema); err != nil { + _ = readWrite.Close() + return nil, fmt.Errorf("failed to re-apply schema after migration: %w", err) + } + + if _, err := readWrite.Exec(backfillAllowedIPDomains); err != nil { + _ = readWrite.Close() + return nil, fmt.Errorf("failed to backfill allowed ip domains: %w", err) + } + readOnly, err := sql.Open("sqlite", sqliteDSN(dbPath, "ro")) if err != nil { _ = readWrite.Close() @@ -93,7 +305,10 @@ func sqliteDSN(dbPath, mode string) string { query := url.Values{} query.Set("mode", mode) - return "file:" + (&url.URL{Path: dbPath, RawQuery: query.Encode()}).String() + return "file:" + (&url.URL{ + Path: dbPath, + RawQuery: query.Encode(), + }).String() } func configureConnection(db *sql.DB, readOnly bool) error { @@ -159,6 +374,64 @@ func (s *Store) UpsertAllowedIP(ip, domain string) error { return nil } +// UpsertAllowedIPDomain inserts or refreshes one domain in the set of domains known for an IP. +// Unlike allowed_ips, which keeps a single representative domain per IP, this table records +// every domain observed for the IP so an SNI can be matched against that IP specifically. +// On conflict, it updates last_refreshed while preserving first_seen. +func (s *Store) UpsertAllowedIPDomain(ip, domain string) error { + domain = strings.ToLower(strings.TrimSpace(domain)) + now := time.Now().Unix() + + _, err := s.readWrite.Exec(` + INSERT INTO allowed_ip_domains (ip, domain, first_seen, last_refreshed) + VALUES (?, ?, ?, ?) + ON CONFLICT(ip, domain) DO UPDATE SET + last_refreshed = excluded.last_refreshed + `, ip, domain, now, now) + + if err != nil { + return fmt.Errorf("upsert allowed ip domain %s/%s: %w", ip, domain, err) + } + + return nil +} + +// DomainsForIP returns every domain on file for the IP that was refreshed within the trust window. +// It returns an empty slice when the IP has no fresh domains on record. +func (s *Store) DomainsForIP(ip string) ([]string, error) { + cutoff := time.Now().Add(-refreshTrustWindow).Unix() + + rows, err := s.readOnly.Query(` + SELECT domain + FROM allowed_ip_domains + WHERE ip = ? + AND last_refreshed >= ? + ORDER BY domain + `, ip, cutoff) + if err != nil { + return nil, fmt.Errorf("domains for ip %s: %w", ip, err) + } + defer func() { + _ = rows.Close() + }() + + domains := []string{} + for rows.Next() { + var domain string + if err := rows.Scan(&domain); err != nil { + return nil, fmt.Errorf("scan domain for ip %s: %w", ip, err) + } + + domains = append(domains, domain) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate domains for ip %s: %w", ip, err) + } + + return domains, nil +} + // UpdateEstablished stamps the current time as last_established for an IP. // The monitor calls it when somo confirms the connection is still active. func (s *Store) UpdateEstablished(ip string) error { @@ -254,6 +527,59 @@ func (s *Store) UpsertFraudCheck(ip string, shouldKill bool) error { return nil } +// GetRecentDomainCheck returns the cached domain-check decision when it was recorded within the cache window. +func (s *Store) GetRecentDomainCheck(domain string) (*DomainCheckCacheEntry, error) { + domain = strings.ToLower(strings.TrimSpace(domain)) + cutoff := time.Now().Add(-fraudCheckCacheWindow).Unix() + + var shouldBlock int + var checkedAt int64 + err := s.readOnly.QueryRow(` + SELECT should_block, checked_at + FROM domain_checks + WHERE domain = ? + AND checked_at >= ? + LIMIT 1 + `, domain, cutoff).Scan(&shouldBlock, &checkedAt) + + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + + if err != nil { + return nil, fmt.Errorf("get recent domain check for %s: %w", domain, err) + } + + return &DomainCheckCacheEntry{ + ShouldBlock: shouldBlock != 0, + CheckedAt: checkedAt, + }, nil +} + +// UpsertDomainCheck stores the current domain-check decision. +func (s *Store) UpsertDomainCheck(domain string, shouldBlock bool) error { + domain = strings.ToLower(strings.TrimSpace(domain)) + shouldBlockInt := 0 + + if shouldBlock { + shouldBlockInt = 1 + } + + _, err := s.readWrite.Exec(` + INSERT INTO domain_checks (domain, should_block, checked_at) + VALUES (?, ?, ?) + ON CONFLICT(domain) DO UPDATE SET + should_block = excluded.should_block, + checked_at = excluded.checked_at + `, domain, shouldBlockInt, time.Now().Unix()) + + if err != nil { + return fmt.Errorf("upsert domain check %s: %w", domain, err) + } + + return nil +} + // LogKill records a killed connection in the audit log. func (s *Store) LogKill(ip, pid, program string) error { _, err := s.readWrite.Exec(` @@ -267,3 +593,161 @@ func (s *Store) LogKill(ip, pid, program string) error { return nil } + +// LogZeekAlert persists a Zeek-derived event for later review. +func (s *Store) LogZeekAlert(srcIP, dstIP, dstPort, sni string, blocked bool, confidence, actionTaken string) error { + blockedInt := 0 + if blocked { + blockedInt = 1 + } + + _, err := s.readWrite.Exec(` + INSERT INTO zeek_alerts (src_ip, dst_ip, dst_port, sni, blocked, confidence, detected_at, action_taken) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, srcIP, dstIP, dstPort, sni, blockedInt, confidence, time.Now().Unix(), actionTaken) + if err != nil { + return fmt.Errorf("log zeek alert %s: %w", srcIP, err) + } + + return nil +} + +// RecentZeekAlert returns the most recent Zeek alert for the given source IP and SNI. +func (s *Store) RecentZeekAlert(srcIP, sni string) (*ZeekAlertEntry, error) { + cutoff := time.Now().Add(-time.Hour).Unix() + + var id int64 + var dstIP string + var dstPort string + var blockedInt int + var confidence string + var detectedAt int64 + var actionTaken string + + err := s.readOnly.QueryRow(` + SELECT id, dst_ip, dst_port, blocked, confidence, detected_at, action_taken + FROM zeek_alerts + WHERE src_ip = ? + AND sni = ? + AND detected_at >= ? + ORDER BY id DESC + LIMIT 1 + `, srcIP, sni, cutoff).Scan(&id, &dstIP, &dstPort, &blockedInt, &confidence, &detectedAt, &actionTaken) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("get recent zeek alert for %s: %w", srcIP, err) + } + + return &ZeekAlertEntry{ + ID: id, + SrcIP: srcIP, + DstIP: dstIP, + DstPort: dstPort, + SNI: sni, + Blocked: blockedInt != 0, + Confidence: confidence, + DetectedAt: detectedAt, + ActionTaken: actionTaken, + }, nil +} + +// LogSNIObservation records an SNI domain decision as a state-table upsert. +// On first sight it inserts the row; on subsequent observations it overwrites the outcome, +// updates last_seen, and increments times_seen while preserving first_seen. +func (s *Store) LogSNIObservation(ip, domain, localPort string, outcome SNIOutcome) error { + domain = strings.ToLower(strings.TrimSpace(domain)) + now := time.Now().Unix() + + _, err := s.readWrite.Exec(` + INSERT INTO sni_observations (ip, domain, local_port, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, ?, 1) + ON CONFLICT(ip, domain, local_port) DO UPDATE SET + outcome = excluded.outcome, + last_seen = excluded.last_seen, + times_seen = sni_observations.times_seen + 1 + `, ip, domain, localPort, string(outcome), now, now) + + if err != nil { + return fmt.Errorf("log sni observation %s: %w", ip, err) + } + + return nil +} + +// LogIPObservation records an IP-only fallback decision as a state-table upsert. +// On first sight it inserts the row; on subsequent observations it overwrites the outcome, +// updates last_seen, and increments times_seen while preserving first_seen. +func (s *Store) LogIPObservation(ip, program string, outcome IPOutcome) error { + now := time.Now().Unix() + + _, err := s.readWrite.Exec(` + INSERT INTO ip_observations (ip, program, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, 1) + ON CONFLICT(ip, program) DO UPDATE SET + outcome = excluded.outcome, + last_seen = excluded.last_seen, + times_seen = ip_observations.times_seen + 1 + `, ip, program, string(outcome), now, now) + + if err != nil { + return fmt.Errorf("log ip observation %s: %w", ip, err) + } + + return nil +} + +// PruneResult reports how many rows were removed per table in a single Prune call. +type PruneResult struct { + AllowedIPDomains int64 + SNIObservations int64 + IPObservations int64 + KilledConnections int64 + ZeekAlerts int64 +} + +// Prune removes stale rows from all pruneable tables. +// It returns the number of rows deleted per table so the caller can log a summary. +func (s *Store) Prune() (*PruneResult, error) { + now := time.Now().Unix() + result := &PruneResult{} + + // allowed_ip_domains: rows older than refreshTrustWindow are already excluded by + // DomainsForIP and can never grant trust again. Safe to remove. + r, err := s.readWrite.Exec(`DELETE FROM allowed_ip_domains WHERE last_refreshed < ?`, now-int64(refreshTrustWindow.Seconds())) + if err != nil { + return nil, fmt.Errorf("prune allowed_ip_domains: %w", err) + } + result.AllowedIPDomains, _ = r.RowsAffected() + + // sni_observations: operational state data with a 7-day window. + r, err = s.readWrite.Exec(`DELETE FROM sni_observations WHERE last_seen < ?`, now-int64(pruneSNIWindow.Seconds())) + if err != nil { + return nil, fmt.Errorf("prune sni_observations: %w", err) + } + result.SNIObservations, _ = r.RowsAffected() + + // ip_observations: same 7-day window as sni_observations. + r, err = s.readWrite.Exec(`DELETE FROM ip_observations WHERE last_seen < ?`, now-int64(pruneSNIWindow.Seconds())) + if err != nil { + return nil, fmt.Errorf("prune ip_observations: %w", err) + } + result.IPObservations, _ = r.RowsAffected() + + // killed_connections and zeek_alerts: audit records of rare, potentially forensically + // valuable events. 90-day window keeps a meaningful history without unbounded growth. + r, err = s.readWrite.Exec(`DELETE FROM killed_connections WHERE killed_at < ?`, now-int64(pruneKilledAlertsWindow.Seconds())) + if err != nil { + return nil, fmt.Errorf("prune killed_connections: %w", err) + } + result.KilledConnections, _ = r.RowsAffected() + + r, err = s.readWrite.Exec(`DELETE FROM zeek_alerts WHERE detected_at < ?`, now-int64(pruneKilledAlertsWindow.Seconds())) + if err != nil { + return nil, fmt.Errorf("prune zeek_alerts: %w", err) + } + result.ZeekAlerts, _ = r.RowsAffected() + + return result, nil +} diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 15acf98..3f6eac5 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -95,3 +95,944 @@ func TestFraudCheckCacheExpiresAfterWindow(t *testing.T) { t.Fatalf("GetRecentFraudCheck() = %#v, want nil for stale entry", entry) } } + +func TestAllowedIPDomainsMultiplePerIP(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + for _, domain := range []string{"github.com", "example.com", "EXAMPLE.NET"} { + if err := hexwallStore.UpsertAllowedIPDomain("203.0.113.30", domain); err != nil { + t.Fatalf("UpsertAllowedIPDomain(%q) error = %v", domain, err) + } + } + + domains, err := hexwallStore.DomainsForIP("203.0.113.30") + if err != nil { + t.Fatalf("DomainsForIP() error = %v", err) + } + + want := []string{"example.com", "example.net", "github.com"} + if len(domains) != len(want) { + t.Fatalf("DomainsForIP() = %v, want %v", domains, want) + } + for i, domain := range want { + if domains[i] != domain { + t.Fatalf("DomainsForIP()[%d] = %q, want %q", i, domains[i], domain) + } + } +} + +func TestDomainsForIPUnknownIP(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + if err := hexwallStore.UpsertAllowedIPDomain("203.0.113.30", "github.com"); err != nil { + t.Fatalf("UpsertAllowedIPDomain() error = %v", err) + } + + domains, err := hexwallStore.DomainsForIP("203.0.113.99") + if err != nil { + t.Fatalf("DomainsForIP() error = %v", err) + } + if len(domains) != 0 { + t.Fatalf("DomainsForIP() = %v, want empty for unknown ip", domains) + } +} + +func TestDomainsForIPExcludesStaleEntries(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + staleRefreshed := time.Now().Add(-refreshTrustWindow - time.Minute).Unix() + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO allowed_ip_domains (ip, domain, first_seen, last_refreshed) + VALUES (?, ?, ?, ?) + `, "203.0.113.40", "stale.example.com", staleRefreshed, staleRefreshed); err != nil { + t.Fatalf("insert stale allowed ip domain error = %v", err) + } + + if err := hexwallStore.UpsertAllowedIPDomain("203.0.113.40", "fresh.example.com"); err != nil { + t.Fatalf("UpsertAllowedIPDomain() error = %v", err) + } + + domains, err := hexwallStore.DomainsForIP("203.0.113.40") + if err != nil { + t.Fatalf("DomainsForIP() error = %v", err) + } + if len(domains) != 1 || domains[0] != "fresh.example.com" { + t.Fatalf("DomainsForIP() = %v, want only the fresh domain", domains) + } +} + +func TestUpsertAllowedIPDomainRefreshesWithoutDuplicating(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + originalSeen := time.Now().Add(-30 * time.Minute).Unix() + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO allowed_ip_domains (ip, domain, first_seen, last_refreshed) + VALUES (?, ?, ?, ?) + `, "203.0.113.50", "example.com", originalSeen, originalSeen); err != nil { + t.Fatalf("insert allowed ip domain error = %v", err) + } + + if err := hexwallStore.UpsertAllowedIPDomain("203.0.113.50", "example.com"); err != nil { + t.Fatalf("UpsertAllowedIPDomain() error = %v", err) + } + + var rowCount int + var firstSeen int64 + var lastRefreshed int64 + if err := hexwallStore.readOnly.QueryRow(` + SELECT COUNT(*), MIN(first_seen), MIN(last_refreshed) + FROM allowed_ip_domains + WHERE ip = ? + `, "203.0.113.50").Scan(&rowCount, &firstSeen, &lastRefreshed); err != nil { + t.Fatalf("query allowed ip domain error = %v", err) + } + + if rowCount != 1 { + t.Fatalf("row count = %d, want 1 after re-upsert", rowCount) + } + if firstSeen != originalSeen { + t.Fatalf("first_seen = %d, want preserved value %d", firstSeen, originalSeen) + } + if lastRefreshed <= originalSeen { + t.Fatalf("last_refreshed = %d, want a value newer than %d", lastRefreshed, originalSeen) + } +} + +func TestNewStoreBackfillsAllowedIPDomains(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + + // UpsertAllowedIP writes only the IP-level record, mimicking a database written + // before allowed_ip_domains existed. + if err := hexwallStore.UpsertAllowedIP("203.0.113.60", "example.com"); err != nil { + t.Fatalf("UpsertAllowedIP() error = %v", err) + } + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + + reopened, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() reopen error = %v", err) + } + defer func() { + if err := reopened.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + domains, err := reopened.DomainsForIP("203.0.113.60") + if err != nil { + t.Fatalf("DomainsForIP() error = %v", err) + } + if len(domains) != 1 || domains[0] != "example.com" { + t.Fatalf("DomainsForIP() = %v, want [example.com] from backfill", domains) + } +} + +func TestDomainCheckCacheRoundTrip(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + entry, err := hexwallStore.GetRecentDomainCheck("evil.example.com") + if err != nil { + t.Fatalf("GetRecentDomainCheck() error = %v", err) + } + if entry != nil { + t.Fatalf("GetRecentDomainCheck() = %#v, want nil before insert", entry) + } + + if err := hexwallStore.UpsertDomainCheck("evil.example.com", true); err != nil { + t.Fatalf("UpsertDomainCheck() error = %v", err) + } + + entry, err = hexwallStore.GetRecentDomainCheck("evil.example.com") + if err != nil { + t.Fatalf("GetRecentDomainCheck() after insert error = %v", err) + } + if entry == nil { + t.Fatal("GetRecentDomainCheck() = nil, want cached entry") + } + if !entry.ShouldBlock { + t.Fatal("GetRecentDomainCheck().ShouldBlock = false, want true") + } + if entry.CheckedAt <= 0 { + t.Fatalf("GetRecentDomainCheck().CheckedAt = %d, want positive unix timestamp", entry.CheckedAt) + } +} + +func TestDomainCheckCacheNormalizedKey(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + if err := hexwallStore.UpsertDomainCheck("EXAMPLE.COM", false); err != nil { + t.Fatalf("UpsertDomainCheck() error = %v", err) + } + + entry, err := hexwallStore.GetRecentDomainCheck(" example.com ") + if err != nil { + t.Fatalf("GetRecentDomainCheck() error = %v", err) + } + if entry == nil { + t.Fatal("GetRecentDomainCheck() = nil, normalized key lookup should match") + } + if entry.ShouldBlock { + t.Fatal("GetRecentDomainCheck().ShouldBlock = true, want false") + } +} + +func TestDomainCheckCacheExpiresAfterWindow(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + staleCheckedAt := time.Now().Add(-fraudCheckCacheWindow - time.Minute).Unix() + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO domain_checks (domain, should_block, checked_at) + VALUES (?, ?, ?) + `, "stale.example.com", 1, staleCheckedAt); err != nil { + t.Fatalf("insert stale domain check error = %v", err) + } + + entry, err := hexwallStore.GetRecentDomainCheck("stale.example.com") + if err != nil { + t.Fatalf("GetRecentDomainCheck() error = %v", err) + } + if entry != nil { + t.Fatalf("GetRecentDomainCheck() = %#v, want nil for stale entry", entry) + } +} + +func TestParseSNIOutcomeValid(t *testing.T) { + t.Parallel() + + tests := []struct { + input string + want SNIOutcome + }{ + {"policy-allowed", OutcomePolicyAllowed}, + {"policy-blocked", OutcomePolicyBlocked}, + {"ip-domain-match", OutcomeIPDomainMatch}, + {"unknown", OutcomeUnknown}, + } + + for _, tt := range tests { + got, ok := ParseSNIOutcome(tt.input) + if !ok { + t.Errorf("ParseSNIOutcome(%q) ok = false, want true", tt.input) + } + if got != tt.want { + t.Errorf("ParseSNIOutcome(%q) = %q, want %q", tt.input, got, tt.want) + } + } +} + +func TestParseSNIOutcomeInvalid(t *testing.T) { + t.Parallel() + + tests := []string{"", "bogus", "true", "false", "1", "0"} + for _, input := range tests { + got, ok := ParseSNIOutcome(input) + if ok { + t.Errorf("ParseSNIOutcome(%q) ok = true, want false", input) + } + if got != OutcomeUnknown { + t.Errorf("ParseSNIOutcome(%q) = %q, want %q", input, got, OutcomeUnknown) + } + } +} + +func TestSNIOutcomeValid(t *testing.T) { + t.Parallel() + + valid := []SNIOutcome{OutcomePolicyAllowed, OutcomePolicyBlocked, OutcomeIPDomainMatch, OutcomeUnknown} + for _, o := range valid { + if !o.Valid() { + t.Errorf("SNIOutcome(%q).Valid() = false, want true", o) + } + } + + invalid := []SNIOutcome{"bogus", "", "true"} + for _, o := range invalid { + if o.Valid() { + t.Errorf("SNIOutcome(%q).Valid() = true, want false", o) + } + } +} + +func TestLogSNIObservationInsertAndUpsert(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + // First insert. + if err := hexwallStore.LogSNIObservation("203.0.113.10", "example.com", "443", OutcomePolicyAllowed); err != nil { + t.Fatalf("LogSNIObservation() error = %v", err) + } + + var timesSeen int + var firstSeen, lastSeen int64 + var outcome string + err = hexwallStore.readOnly.QueryRow(` + SELECT times_seen, first_seen, last_seen, outcome + FROM sni_observations + WHERE ip = ? AND domain = ? AND local_port = ? + `, "203.0.113.10", "example.com", "443").Scan(×Seen, &firstSeen, &lastSeen, &outcome) + if err != nil { + t.Fatalf("query sni_observations error = %v", err) + } + if timesSeen != 1 { + t.Fatalf("times_seen = %d, want 1 after first insert", timesSeen) + } + if firstSeen <= 0 { + t.Fatalf("first_seen = %d, want positive unix timestamp", firstSeen) + } + if lastSeen != firstSeen { + t.Fatalf("last_seen = %d, want equal to first_seen on first insert, got %d", lastSeen, firstSeen) + } + if outcome != "policy-allowed" { + t.Fatalf("outcome = %q, want %q", outcome, "policy-allowed") + } + + // Second insert with different outcome — should upsert. + if err := hexwallStore.LogSNIObservation("203.0.113.10", "example.com", "443", OutcomeIPDomainMatch); err != nil { + t.Fatalf("LogSNIObservation() upsert error = %v", err) + } + + err = hexwallStore.readOnly.QueryRow(` + SELECT times_seen, first_seen, last_seen, outcome + FROM sni_observations + WHERE ip = ? AND domain = ? AND local_port = ? + `, "203.0.113.10", "example.com", "443").Scan(×Seen, &firstSeen, &lastSeen, &outcome) + if err != nil { + t.Fatalf("query sni_observations after upsert error = %v", err) + } + if timesSeen != 2 { + t.Fatalf("times_seen = %d, want 2 after upsert", timesSeen) + } + if outcome != "ip-domain-match" { + t.Fatalf("outcome = %q, want %q after upsert", outcome, "ip-domain-match") + } + if lastSeen < firstSeen { + t.Fatalf("last_seen = %d should not be older than first_seen = %d", lastSeen, firstSeen) + } + + // Third insert — times_seen should be 3. + if err := hexwallStore.LogSNIObservation("203.0.113.10", "example.com", "443", OutcomePolicyBlocked); err != nil { + t.Fatalf("LogSNIObservation() third insert error = %v", err) + } + + err = hexwallStore.readOnly.QueryRow(` + SELECT times_seen, outcome + FROM sni_observations + WHERE ip = ? AND domain = ? AND local_port = ? + `, "203.0.113.10", "example.com", "443").Scan(×Seen, &outcome) + if err != nil { + t.Fatalf("query sni_observations after third insert error = %v", err) + } + if timesSeen != 3 { + t.Fatalf("times_seen = %d, want 3 after third insert", timesSeen) + } + if outcome != "policy-blocked" { + t.Fatalf("outcome = %q, want %q after third insert", outcome, "policy-blocked") + } +} + +func TestLogSNIObservationNormalizesDomain(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + if err := hexwallStore.LogSNIObservation("203.0.113.10", " EXAMPLE.COM ", "443", OutcomePolicyAllowed); err != nil { + t.Fatalf("LogSNIObservation() error = %v", err) + } + + // Lookup by the normalized form. + var outcome string + err = hexwallStore.readOnly.QueryRow(` + SELECT outcome FROM sni_observations + WHERE ip = ? AND domain = ? AND local_port = ? + `, "203.0.113.10", "example.com", "443").Scan(&outcome) + if err != nil { + t.Fatalf("query sni_observations for normalized domain error = %v", err) + } + if outcome != "policy-allowed" { + t.Fatalf("outcome = %q, want %q", outcome, "policy-allowed") + } +} + +func TestNewStoreMigratesOldSNITable(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + // Create a store with the old schema manually. + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + + // Drop the new table and recreate with the old schema. + if _, err := hexwallStore.readWrite.Exec(`DROP TABLE sni_observations`); err != nil { + t.Fatalf("drop sni_observations error = %v", err) + } + if _, err := hexwallStore.readWrite.Exec(` + CREATE TABLE sni_observations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ip TEXT NOT NULL, domain TEXT NOT NULL, local_port TEXT NOT NULL, + known INTEGER NOT NULL, observed_at INTEGER NOT NULL + ) + `); err != nil { + t.Fatalf("create old sni_observations error = %v", err) + } + // Insert a row in the old format. + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO sni_observations (ip, domain, local_port, known, observed_at) + VALUES (?, ?, ?, ?, ?) + `, "203.0.113.20", "old.example.com", "443", 1, time.Now().Unix()); err != nil { + t.Fatalf("insert old sni_observations error = %v", err) + } + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + + // Reopen — migration should drop the old table and create the new one. + reopened, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() reopen error = %v", err) + } + defer func() { + if err := reopened.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + // The old row should be gone. + var count int + err = reopened.readOnly.QueryRow(`SELECT COUNT(*) FROM sni_observations`).Scan(&count) + if err != nil { + t.Fatalf("query sni_observations count error = %v", err) + } + if count != 0 { + t.Fatalf("row count = %d, want 0 after migration dropped old table", count) + } + + // The new schema should work. + if err := reopened.LogSNIObservation("203.0.113.20", "new.example.com", "443", OutcomePolicyAllowed); err != nil { + t.Fatalf("LogSNIObservation() after migration error = %v", err) + } + + err = reopened.readOnly.QueryRow(`SELECT COUNT(*) FROM sni_observations`).Scan(&count) + if err != nil { + t.Fatalf("query sni_observations count after insert error = %v", err) + } + if count != 1 { + t.Fatalf("row count = %d, want 1 after insert with new schema", count) + } +} + +func TestPruneRemovesStaleRows(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + now := time.Now().Unix() + + // Insert stale allowed_ip_domains (older than refreshTrustWindow). + staleRefreshed := now - int64(refreshTrustWindow.Seconds()) - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO allowed_ip_domains (ip, domain, first_seen, last_refreshed) + VALUES (?, ?, ?, ?) + `, "203.0.113.30", "stale.example.com", staleRefreshed, staleRefreshed); err != nil { + t.Fatalf("insert stale allowed_ip_domains error = %v", err) + } + + // Insert fresh allowed_ip_domains. + freshRefreshed := now - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO allowed_ip_domains (ip, domain, first_seen, last_refreshed) + VALUES (?, ?, ?, ?) + `, "203.0.113.30", "fresh.example.com", freshRefreshed, freshRefreshed); err != nil { + t.Fatalf("insert fresh allowed_ip_domains error = %v", err) + } + + // Insert stale sni_observations (older than 7 days). + staleSNI := now - int64(pruneSNIWindow.Seconds()) - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO sni_observations (ip, domain, local_port, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + `, "203.0.113.30", "stale.example.com", "443", "unknown", staleSNI, staleSNI, 5); err != nil { + t.Fatalf("insert stale sni_observations error = %v", err) + } + + // Insert fresh sni_observations. + freshSNI := now - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO sni_observations (ip, domain, local_port, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + `, "203.0.113.30", "fresh.example.com", "443", "unknown", freshSNI, freshSNI, 1); err != nil { + t.Fatalf("insert fresh sni_observations error = %v", err) + } + + // Insert stale killed_connections (older than 90 days). + staleKill := now - int64(pruneKilledAlertsWindow.Seconds()) - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO killed_connections (ip, pid, program, killed_at) + VALUES (?, ?, ?, ?) + `, "203.0.113.30", "1234", "curl", staleKill); err != nil { + t.Fatalf("insert stale killed_connections error = %v", err) + } + + // Insert fresh killed_connections. + freshKill := now - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO killed_connections (ip, pid, program, killed_at) + VALUES (?, ?, ?, ?) + `, "203.0.113.30", "5678", "wget", freshKill); err != nil { + t.Fatalf("insert fresh killed_connections error = %v", err) + } + + // Insert stale zeek_alerts. + staleZeek := now - int64(pruneKilledAlertsWindow.Seconds()) - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO zeek_alerts (src_ip, dst_ip, dst_port, sni, blocked, confidence, detected_at, action_taken) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, "10.0.0.1", "203.0.113.30", "443", "old.example.com", 0, "low", staleZeek, "logged"); err != nil { + t.Fatalf("insert stale zeek_alerts error = %v", err) + } + + // Insert fresh zeek_alerts. + freshZeek := now - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO zeek_alerts (src_ip, dst_ip, dst_port, sni, blocked, confidence, detected_at, action_taken) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, "10.0.0.2", "203.0.113.30", "443", "new.example.com", 1, "high", freshZeek, "logged"); err != nil { + t.Fatalf("insert fresh zeek_alerts error = %v", err) + } + + // Run prune. + result, err := hexwallStore.Prune() + if err != nil { + t.Fatalf("Prune() error = %v", err) + } + + if result.AllowedIPDomains != 1 { + t.Fatalf("Prune().AllowedIPDomains = %d, want 1", result.AllowedIPDomains) + } + if result.SNIObservations != 1 { + t.Fatalf("Prune().SNIObservations = %d, want 1", result.SNIObservations) + } + if result.KilledConnections != 1 { + t.Fatalf("Prune().KilledConnections = %d, want 1", result.KilledConnections) + } + if result.ZeekAlerts != 1 { + t.Fatalf("Prune().ZeekAlerts = %d, want 1", result.ZeekAlerts) + } + + // Verify fresh rows remain. + var count int + if err := hexwallStore.readOnly.QueryRow(`SELECT COUNT(*) FROM allowed_ip_domains`).Scan(&count); err != nil { + t.Fatalf("query allowed_ip_domains count error = %v", err) + } + if count != 1 { + t.Fatalf("allowed_ip_domains count = %d, want 1 fresh row remaining", count) + } + + if err := hexwallStore.readOnly.QueryRow(`SELECT COUNT(*) FROM sni_observations`).Scan(&count); err != nil { + t.Fatalf("query sni_observations count error = %v", err) + } + if count != 1 { + t.Fatalf("sni_observations count = %d, want 1 fresh row remaining", count) + } + + if err := hexwallStore.readOnly.QueryRow(`SELECT COUNT(*) FROM killed_connections`).Scan(&count); err != nil { + t.Fatalf("query killed_connections count error = %v", err) + } + if count != 1 { + t.Fatalf("killed_connections count = %d, want 1 fresh row remaining", count) + } + + if err := hexwallStore.readOnly.QueryRow(`SELECT COUNT(*) FROM zeek_alerts`).Scan(&count); err != nil { + t.Fatalf("query zeek_alerts count error = %v", err) + } + if count != 1 { + t.Fatalf("zeek_alerts count = %d, want 1 fresh row remaining", count) + } +} + +func TestPruneNoopWhenNothingExpired(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + now := time.Now().Unix() + + // Insert fresh rows in all pruneable tables. + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO allowed_ip_domains (ip, domain, first_seen, last_refreshed) + VALUES (?, ?, ?, ?) + `, "203.0.113.40", "fresh.example.com", now, now); err != nil { + t.Fatalf("insert fresh allowed_ip_domains error = %v", err) + } + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO sni_observations (ip, domain, local_port, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + `, "203.0.113.40", "fresh.example.com", "443", "unknown", now, now, 1); err != nil { + t.Fatalf("insert fresh sni_observations error = %v", err) + } + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO killed_connections (ip, pid, program, killed_at) + VALUES (?, ?, ?, ?) + `, "203.0.113.40", "1234", "curl", now); err != nil { + t.Fatalf("insert fresh killed_connections error = %v", err) + } + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO zeek_alerts (src_ip, dst_ip, dst_port, sni, blocked, confidence, detected_at, action_taken) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, "10.0.0.1", "203.0.113.40", "443", "fresh.example.com", 0, "low", now, "logged"); err != nil { + t.Fatalf("insert fresh zeek_alerts error = %v", err) + } + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO ip_observations (ip, program, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, ?) + `, "203.0.113.40", "curl", "unknown", now, now, 1); err != nil { + t.Fatalf("insert fresh ip_observations error = %v", err) + } + + result, err := hexwallStore.Prune() + if err != nil { + t.Fatalf("Prune() error = %v", err) + } + + total := result.AllowedIPDomains + result.SNIObservations + result.IPObservations + result.KilledConnections + result.ZeekAlerts + if total != 0 { + t.Fatalf("Prune() deleted %d total rows, want 0 when nothing is expired", total) + } +} + +func TestParseIPOutcomeValid(t *testing.T) { + t.Parallel() + + tests := []struct { + input string + want IPOutcome + }{ + {"ip-trusted", OutcomeIPTrusted}, + {"private-reserved", OutcomeIPReserved}, + {"deghost-blocked", OutcomeIPBlocked}, + {"unknown", OutcomeIPUnknown}, + } + + for _, tt := range tests { + got, ok := ParseIPOutcome(tt.input) + if !ok { + t.Errorf("ParseIPOutcome(%q) ok = false, want true", tt.input) + } + if got != tt.want { + t.Errorf("ParseIPOutcome(%q) = %q, want %q", tt.input, got, tt.want) + } + } +} + +func TestParseIPOutcomeInvalid(t *testing.T) { + t.Parallel() + + tests := []string{"", "bogus", "true", "false", "1", "0"} + for _, input := range tests { + got, ok := ParseIPOutcome(input) + if ok { + t.Errorf("ParseIPOutcome(%q) ok = true, want false", input) + } + if got != OutcomeIPUnknown { + t.Errorf("ParseIPOutcome(%q) = %q, want %q", input, got, OutcomeIPUnknown) + } + } +} + +func TestIPOutcomeValid(t *testing.T) { + t.Parallel() + + valid := []IPOutcome{OutcomeIPTrusted, OutcomeIPReserved, OutcomeIPBlocked, OutcomeIPUnknown} + for _, o := range valid { + if !o.Valid() { + t.Errorf("IPOutcome(%q).Valid() = false, want true", o) + } + } + + invalid := []IPOutcome{"bogus", "", "true"} + for _, o := range invalid { + if o.Valid() { + t.Errorf("IPOutcome(%q).Valid() = true, want false", o) + } + } +} + +func TestLogIPObservationInsertAndUpsert(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + // First insert. + if err := hexwallStore.LogIPObservation("203.0.113.10", "curl", OutcomeIPTrusted); err != nil { + t.Fatalf("LogIPObservation() error = %v", err) + } + + var timesSeen int + var firstSeen, lastSeen int64 + var outcome string + err = hexwallStore.readOnly.QueryRow(` + SELECT times_seen, first_seen, last_seen, outcome + FROM ip_observations + WHERE ip = ? AND program = ? + `, "203.0.113.10", "curl").Scan(×Seen, &firstSeen, &lastSeen, &outcome) + if err != nil { + t.Fatalf("query ip_observations error = %v", err) + } + if timesSeen != 1 { + t.Fatalf("times_seen = %d, want 1 after first insert", timesSeen) + } + if firstSeen <= 0 { + t.Fatalf("first_seen = %d, want positive unix timestamp", firstSeen) + } + if lastSeen != firstSeen { + t.Fatalf("last_seen = %d, want equal to first_seen on first insert, got %d", lastSeen, firstSeen) + } + if outcome != "ip-trusted" { + t.Fatalf("outcome = %q, want %q", outcome, "ip-trusted") + } + + // Second insert with different outcome — should upsert. + if err := hexwallStore.LogIPObservation("203.0.113.10", "curl", OutcomeIPBlocked); err != nil { + t.Fatalf("LogIPObservation() upsert error = %v", err) + } + + err = hexwallStore.readOnly.QueryRow(` + SELECT times_seen, first_seen, last_seen, outcome + FROM ip_observations + WHERE ip = ? AND program = ? + `, "203.0.113.10", "curl").Scan(×Seen, &firstSeen, &lastSeen, &outcome) + if err != nil { + t.Fatalf("query ip_observations after upsert error = %v", err) + } + if timesSeen != 2 { + t.Fatalf("times_seen = %d, want 2 after upsert", timesSeen) + } + if outcome != "deghost-blocked" { + t.Fatalf("outcome = %q, want %q after upsert", outcome, "deghost-blocked") + } + if lastSeen < firstSeen { + t.Fatalf("last_seen = %d should not be older than first_seen = %d", lastSeen, firstSeen) + } + + // Third insert — times_seen should be 3. + if err := hexwallStore.LogIPObservation("203.0.113.10", "curl", OutcomeIPReserved); err != nil { + t.Fatalf("LogIPObservation() third insert error = %v", err) + } + + err = hexwallStore.readOnly.QueryRow(` + SELECT times_seen, outcome + FROM ip_observations + WHERE ip = ? AND program = ? + `, "203.0.113.10", "curl").Scan(×Seen, &outcome) + if err != nil { + t.Fatalf("query ip_observations after third insert error = %v", err) + } + if timesSeen != 3 { + t.Fatalf("times_seen = %d, want 3 after third insert", timesSeen) + } + if outcome != "private-reserved" { + t.Fatalf("outcome = %q, want %q after third insert", outcome, "private-reserved") + } +} + +func TestLogIPObservationDifferentPrograms(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + // Same IP, different programs — should create separate rows. + if err := hexwallStore.LogIPObservation("203.0.113.20", "curl", OutcomeIPTrusted); err != nil { + t.Fatalf("LogIPObservation() error = %v", err) + } + if err := hexwallStore.LogIPObservation("203.0.113.20", "wget", OutcomeIPBlocked); err != nil { + t.Fatalf("LogIPObservation() error = %v", err) + } + + var count int + err = hexwallStore.readOnly.QueryRow(`SELECT COUNT(*) FROM ip_observations WHERE ip = ?`, "203.0.113.20").Scan(&count) + if err != nil { + t.Fatalf("query ip_observations count error = %v", err) + } + if count != 2 { + t.Fatalf("row count = %d, want 2 for same IP with different programs", count) + } +} + +func TestPruneRemovesStaleIPObservations(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "hexwall.db") + + hexwallStore, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + defer func() { + if err := hexwallStore.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + }() + + now := time.Now().Unix() + + // Insert stale ip_observations (older than 7 days). + staleIP := now - int64(pruneSNIWindow.Seconds()) - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO ip_observations (ip, program, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, ?) + `, "203.0.113.50", "curl", "unknown", staleIP, staleIP, 5); err != nil { + t.Fatalf("insert stale ip_observations error = %v", err) + } + + // Insert fresh ip_observations. + freshIP := now - 60 + if _, err := hexwallStore.readWrite.Exec(` + INSERT INTO ip_observations (ip, program, outcome, first_seen, last_seen, times_seen) + VALUES (?, ?, ?, ?, ?, ?) + `, "203.0.113.50", "wget", "unknown", freshIP, freshIP, 1); err != nil { + t.Fatalf("insert fresh ip_observations error = %v", err) + } + + // Run prune. + result, err := hexwallStore.Prune() + if err != nil { + t.Fatalf("Prune() error = %v", err) + } + + if result.IPObservations != 1 { + t.Fatalf("Prune().IPObservations = %d, want 1", result.IPObservations) + } + + // Verify fresh row remains. + var count int + if err := hexwallStore.readOnly.QueryRow(`SELECT COUNT(*) FROM ip_observations`).Scan(&count); err != nil { + t.Fatalf("query ip_observations count error = %v", err) + } + if count != 1 { + t.Fatalf("ip_observations count = %d, want 1 fresh row remaining", count) + } +} diff --git a/internal/zeek/zeek.go b/internal/zeek/zeek.go new file mode 100644 index 0000000..3a9cdc0 --- /dev/null +++ b/internal/zeek/zeek.go @@ -0,0 +1,360 @@ +// Package zeek tails Zeek logs to extract SNI/domain data and DNS-bypass notices. +package zeek + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "strings" + "sync" + "time" +) + +const ( + tickInterval = 5 * time.Second + entryExpiry = 2 * time.Minute +) + +// Config holds the configuration for the Zeek ssl.log tailer. +type Config struct { + LogPath string +} + +// Event is a parsed DNS bypass notice from Zeek. +type Event struct { + Timestamp time.Time + SrcIP string + DstIP string + DstPort string + SNI string +} + +type sniEntry struct { + domain string + observedAt time.Time +} + +// sslRecord is the subset of fields we extract from a Zeek SSL JSON log line. +type sslRecord struct { + OrigH string `json:"id.orig_h"` + OrigP any `json:"id.orig_p"` + RespH string `json:"id.resp_h"` + RespP any `json:"id.resp_p"` + ServerName string `json:"server_name"` +} + +// Client tails a Zeek ssl.log and maintains a short-lived in-memory cache +// mapping connection tuples to their observed SNI domain. +type Client struct { + logPath string + + mu sync.RWMutex + cache map[string]sniEntry + + offset int64 + pending []byte // bytes read past the last newline; held until the rest of the line arrives + stopCh chan struct{} + done chan struct{} +} + +// NewClient opens the ssl.log at the configured path and starts a +// background goroutine that tails new JSON lines. +// If cfg is nil or LogPath is empty, it returns nil without error. +func NewClient(cfg *Config) (*Client, error) { + if cfg == nil || strings.TrimSpace(cfg.LogPath) == "" { + return nil, nil + } + + logPath := strings.TrimSpace(cfg.LogPath) + + // Verify the file exists and is readable before starting the tailer. + f, err := os.Open(logPath) + if err != nil { + return nil, fmt.Errorf("failed to open zeek ssl.log %s: %w", logPath, err) + } + fi, err := f.Stat() + _ = f.Close() + if err != nil { + return nil, fmt.Errorf("failed to stat zeek ssl.log %s: %w", logPath, err) + } + + c := &Client{ + logPath: logPath, + cache: make(map[string]sniEntry), + offset: fi.Size(), + stopCh: make(chan struct{}), + done: make(chan struct{}), + } + + go c.tailLoop(context.Background()) + + slog.Info("zeek log watcher started", "path", logPath, "offset", c.offset) + + return c, nil +} + +// Close stops the background tailer and waits for it to finish. +func (c *Client) Close() { + if c == nil { + return + } + close(c.stopCh) + <-c.done +} + +// Lookup returns the SNI domain observed for the given connection tuple, +// or ("", false) if no matching record exists in the cache. +func (c *Client) Lookup(localPort, remoteIP, remotePort string) (string, bool) { + key := localPort + "|" + remoteIP + "|" + remotePort + + c.mu.RLock() + entry, ok := c.cache[key] + c.mu.RUnlock() + + if !ok { + return "", false + } + + if time.Since(entry.observedAt) > entryExpiry { + return "", false + } + + return entry.domain, true +} + +// ParseNoticeLine parses a single Zeek notice JSON line into an Event. +func ParseNoticeLine(line string) (Event, error) { + line = strings.TrimSpace(line) + if line == "" { + return Event{}, errors.New("empty notice line") + } + + var payload map[string]any + if err := json.Unmarshal([]byte(line), &payload); err != nil { + return Event{}, fmt.Errorf("parse notice line: %w", err) + } + + event := Event{Timestamp: time.Now()} + if ts, ok := payload["ts"].(string); ok { + if parsed, err := time.Parse(time.RFC3339Nano, ts); err == nil { + event.Timestamp = parsed + } + } + if ts, ok := payload["timestamp"].(string); ok { + if parsed, err := time.Parse(time.RFC3339Nano, ts); err == nil { + event.Timestamp = parsed + } + } + + event.SrcIP = stringValue(payload["src_ip"], payload["src"], payload["id.orig_h"]) + event.DstIP = stringValue(payload["dst_ip"], payload["dst"], payload["id.resp_h"]) + event.DstPort = stringValue(payload["dst_port"], payload["p"], payload["id.resp_p"]) + event.SNI = stringValue(payload["sni"], payload["server_name"]) + + if event.SrcIP == "" && event.DstIP == "" && event.SNI == "" { + return Event{}, errors.New("notice line did not contain recognizable fields") + } + + return event, nil +} + +// WatchNoticeLog tails a Zeek notice log and emits parsed Events. +func WatchNoticeLog(ctx context.Context, path string, events chan<- Event) error { + logPath := strings.TrimSpace(path) + if ctx == nil || logPath == "" { + return nil + } + + var offset int64 + poller := time.NewTicker(2 * time.Second) + defer poller.Stop() + + for { + select { + case <-ctx.Done(): + return nil + case <-poller.C: + file, err := os.Open(logPath) + if err != nil { + slog.Debug("zeek: cannot open notice log", "path", logPath, "err", err) + continue + } + + fi, err := file.Stat() + if err != nil { + _ = file.Close() + slog.Debug("zeek: cannot stat notice log", "path", logPath, "err", err) + continue + } + + if fi.Size() < offset { + offset = 0 + } + if fi.Size() > offset { + if _, err := file.Seek(offset, io.SeekStart); err != nil { + _ = file.Close() + slog.Debug("zeek: seek notice log failed", "path", logPath, "err", err) + continue + } + + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" { + continue + } + + event, err := ParseNoticeLine(line) + if err != nil { + continue + } + + select { + case events <- event: + default: + } + } + if err := scanner.Err(); err != nil { + slog.Debug("zeek: notice log scan failed", "path", logPath, "err", err) + } + } + _ = file.Close() + offset = fi.Size() + } + } +} + +func (c *Client) tailLoop(ctx context.Context) { + defer close(c.done) + + ticker := time.NewTicker(tickInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-c.stopCh: + return + case <-ticker.C: + c.processNewLines() + } + } +} + +func (c *Client) processNewLines() { + file, err := os.Open(c.logPath) + if err != nil { + slog.Debug("zeek: cannot open ssl.log", "path", c.logPath, "err", err) + return + } + defer func() { + if err := file.Close(); err != nil { + slog.Debug("zeek: failed to close ssl.log", "path", c.logPath, "err", err) + } + }() + + fi, err := file.Stat() + if err != nil { + slog.Debug("zeek: cannot stat ssl.log", "path", c.logPath, "err", err) + return + } + + // File was likely rotated — reset to the beginning and drop any + // partial line held over from the old file. + if fi.Size() < c.offset { + c.offset = 0 + c.pending = nil + } + + if fi.Size() == c.offset { + return + } + + if _, err := file.Seek(c.offset, io.SeekStart); err != nil { + slog.Debug("zeek: seek failed", "path", c.logPath, "err", err) + return + } + + data, err := io.ReadAll(file) + if err != nil { + slog.Debug("zeek: read failed", "path", c.logPath, "err", err) + return + } + + // We've now consumed everything up to fi.Size() from the file, even + // if the trailing bytes don't yet form a complete line — hold them in + // pending rather than losing them, since Zeek may still be mid-write + // on the last line when this tick fires. + c.offset = fi.Size() + + buf := append(c.pending, data...) + lastNewline := bytes.LastIndexByte(buf, '\n') + if lastNewline == -1 { + c.pending = buf + return + } + + complete := buf[:lastNewline] + c.pending = append([]byte(nil), buf[lastNewline+1:]...) + + lines := strings.Split(string(complete), "\n") + + c.mu.Lock() + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + + var rec sslRecord + if err := json.Unmarshal([]byte(line), &rec); err != nil { + continue + } + + if rec.ServerName == "" || rec.OrigH == "" || rec.RespH == "" { + continue + } + + origP := fmt.Sprint(rec.OrigP) + respP := fmt.Sprint(rec.RespP) + now := time.Now() + entry := sniEntry{domain: rec.ServerName, observedAt: now} + + // Store both orientations so Lookup works regardless of which + // side of the connection is the local endpoint. + c.cache[origP+"|"+rec.RespH+"|"+respP] = entry + c.cache[respP+"|"+rec.OrigH+"|"+origP] = entry + } + + // Prune expired entries. + cutoff := time.Now().Add(-entryExpiry) + for k, v := range c.cache { + if v.observedAt.Before(cutoff) { + delete(c.cache, k) + } + } + c.mu.Unlock() +} + +func stringValue(values ...any) string { + for _, value := range values { + switch v := value.(type) { + case string: + return strings.TrimSpace(v) + case fmt.Stringer: + return strings.TrimSpace(v.String()) + case nil: + continue + default: + return fmt.Sprint(v) + } + } + return "" +} diff --git a/internal/zeek/zeek_test.go b/internal/zeek/zeek_test.go new file mode 100644 index 0000000..2991a4b --- /dev/null +++ b/internal/zeek/zeek_test.go @@ -0,0 +1,168 @@ +package zeek + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +// newTestClient builds a Client pointed at path without starting the +// background tailLoop goroutine, so tests can call processNewLines directly +// at controlled points instead of racing a real ticker. +func newTestClient(t *testing.T, path string) *Client { + t.Helper() + return &Client{ + logPath: path, + cache: make(map[string]sniEntry), + } +} + +func writeFile(t *testing.T, path, content string) { + t.Helper() + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatalf("write file: %v", err) + } +} + +func TestParseNoticeLine(t *testing.T) { + line := `{"note":"DNSBypass::PossibleBypass","src_ip":"192.0.2.10","dst_ip":"203.0.113.20","dst_port":"443","sni":"evil.example","msg":"SNI evil.example from 192.0.2.10"}` + + event, err := ParseNoticeLine(line) + if err != nil { + t.Fatalf("ParseNoticeLine() error = %v", err) + } + + if event.SrcIP != "192.0.2.10" { + t.Fatalf("ParseNoticeLine().SrcIP = %q, want %q", event.SrcIP, "192.0.2.10") + } + if event.DstIP != "203.0.113.20" { + t.Fatalf("ParseNoticeLine().DstIP = %q, want %q", event.DstIP, "203.0.113.20") + } + if event.DstPort != "443" { + t.Fatalf("ParseNoticeLine().DstPort = %q, want %q", event.DstPort, "443") + } + if event.SNI != "evil.example" { + t.Fatalf("ParseNoticeLine().SNI = %q, want %q", event.SNI, "evil.example") + } + if event.Timestamp.IsZero() { + t.Fatal("ParseNoticeLine().Timestamp = zero time, want non-zero") + } + if event.Timestamp.After(time.Now().Add(time.Second)) { + t.Fatalf("ParseNoticeLine().Timestamp too far in the future: %v", event.Timestamp) + } +} + +func TestParseNoticeLineRejectsInvalidJSON(t *testing.T) { + _, err := ParseNoticeLine(`{"note":`) + if err == nil { + t.Fatal("ParseNoticeLine() error = nil, want error") + } +} + +func TestClientProcessNewLines_BasicParse(t *testing.T) { + path := filepath.Join(t.TempDir(), "ssl.log") + writeFile(t, path, `{"id.orig_h":"10.0.0.5","id.orig_p":51820,"id.resp_h":"203.0.113.20","id.resp_p":443,"server_name":"example.com"}`+"\n") + + c := newTestClient(t, path) + c.processNewLines() + + domain, ok := c.Lookup("51820", "203.0.113.20", "443") + if !ok { + t.Fatal("Lookup() ok = false, want true") + } + if domain != "example.com" { + t.Fatalf("Lookup() domain = %q, want %q", domain, "example.com") + } + + // The reverse orientation must also resolve, since Lookup doesn't know + // in advance which side of the connection is the local endpoint. + domain, ok = c.Lookup("443", "10.0.0.5", "51820") + if !ok { + t.Fatal("Lookup() reverse orientation ok = false, want true") + } + if domain != "example.com" { + t.Fatalf("Lookup() reverse orientation domain = %q, want %q", domain, "example.com") + } +} + +func TestClientProcessNewLines_SkipsIncompleteRecords(t *testing.T) { + path := filepath.Join(t.TempDir(), "ssl.log") + writeFile(t, path, `{"id.orig_h":"10.0.0.5","id.orig_p":51820,"id.resp_h":"203.0.113.20","id.resp_p":443}`+"\n"+ + `not json at all`+"\n"+ + `{"id.orig_h":"","id.orig_p":1,"id.resp_h":"203.0.113.21","id.resp_p":443,"server_name":"missing-orig-host.example"}`+"\n") + + c := newTestClient(t, path) + c.processNewLines() + + if len(c.cache) != 0 { + t.Fatalf("cache = %v, want empty (no valid records in input)", c.cache) + } +} + +func TestClientProcessNewLines_PartialLineHeldAcrossTicks(t *testing.T) { + path := filepath.Join(t.TempDir(), "ssl.log") + full := `{"id.orig_h":"10.0.0.5","id.orig_p":51820,"id.resp_h":"203.0.113.20","id.resp_p":443,"server_name":"example.com"}` + "\n" + + // Simulate Zeek flushing mid-record: only the first half of the line + // lands before the poll fires. + split := len(full) / 2 + writeFile(t, path, full[:split]) + + c := newTestClient(t, path) + c.processNewLines() + + if _, ok := c.Lookup("51820", "203.0.113.20", "443"); ok { + t.Fatal("Lookup() found a record from a partial line, want none yet") + } + if len(c.pending) == 0 { + t.Fatal("pending is empty, want the partial line to be held over") + } + + // The rest of the line arrives on the next tick. + writeFile(t, path, full) + c.processNewLines() + + domain, ok := c.Lookup("51820", "203.0.113.20", "443") + if !ok { + t.Fatal("Lookup() ok = false after completing the line, want true") + } + if domain != "example.com" { + t.Fatalf("Lookup() domain = %q, want %q", domain, "example.com") + } +} + +func TestClientProcessNewLines_Rotation(t *testing.T) { + path := filepath.Join(t.TempDir(), "ssl.log") + writeFile(t, path, `{"id.orig_h":"10.0.0.5","id.orig_p":51820,"id.resp_h":"203.0.113.20","id.resp_p":443,"server_name":"example.com"}`+"\n") + + c := newTestClient(t, path) + c.processNewLines() + + if _, ok := c.Lookup("51820", "203.0.113.20", "443"); !ok { + t.Fatal("Lookup() ok = false before rotation, want true") + } + + // Rotation: zeekctl closes the current log and starts a fresh, empty + // file at the same path — simulate the two ticks that straddle that. + writeFile(t, path, "") + c.processNewLines() + writeFile(t, path, `{"id.orig_h":"10.0.0.9","id.orig_p":9000,"id.resp_h":"203.0.113.30","id.resp_p":443,"server_name":"post-rotation.example"}`+"\n") + c.processNewLines() + + if domain, ok := c.Lookup("9000", "203.0.113.30", "443"); !ok || domain != "post-rotation.example" { + t.Fatalf("Lookup() after rotation = (%q, %v), want (%q, true)", domain, ok, "post-rotation.example") + } +} + +func TestClientLookup_Expiry(t *testing.T) { + c := newTestClient(t, "") + c.cache["51820|203.0.113.20|443"] = sniEntry{ + domain: "example.com", + observedAt: time.Now().Add(-entryExpiry - time.Second), + } + + if _, ok := c.Lookup("51820", "203.0.113.20", "443"); ok { + t.Fatal("Lookup() ok = true for an expired entry, want false") + } +} diff --git a/main.go b/main.go index 7199497..584e9d7 100644 --- a/main.go +++ b/main.go @@ -19,6 +19,7 @@ import ( "github.com/hexbytedev/hexwall/internal/pihole" "github.com/hexbytedev/hexwall/internal/somo" "github.com/hexbytedev/hexwall/internal/store" + "github.com/hexbytedev/hexwall/internal/zeek" ) const ( @@ -28,39 +29,94 @@ const ( connectionScanInterval = 10 * time.Second ) -var version = "dev" -var platform = runtime.GOOS + "/" + runtime.GOARCH +var ( + version = "dev" + platform = runtime.GOOS + "/" + runtime.GOARCH +) + +type runConfig struct { + dbPath string + hexwallDBPath string + zeekLogPath string + zeekNoticeLog string + enableZeek bool + enableDeghostIP bool + enableDeghostDomain bool + mode string + debug bool + showVersion bool +} func main() { os.Exit(run()) } -func run() int { +func parseFlags() (*runConfig, error) { dbPath := flag.String("db", "", "path to pihole-FTL.db (auto-detected if not set)") hexwallDB := flag.String("hexwall-db", "./hexwall.db", "path to local hexwall database") + zeekLogPath := flag.String("zeek-log", "", "path to zeek ssl.log for SNI-based domain verification (empty to skip)") + zeekNoticeLog := flag.String("zeek-notice-log", "/opt/zeek/logs/current/notice.log", "path to Zeek notice.log for DNS-bypass alerts") + enableZeek := flag.Bool("enable-zeek", false, "enable Zeek-based DNS-bypass detection") + enableDeghostIP := flag.Bool("enable-deghost-ip", true, "enable third-party IP reputation checks on the IP-only fallback path") + enableDeghostDomain := flag.Bool("enable-deghost-domain", true, "enable third-party domain reputation checks at rung 6") mode := flag.String("mode", monitor.ModeWatch, "monitor mode: watch (detect only) or enforce (kill + log)") - debug := flag.Bool("debug", false, "enable verbose per-connection scan logging") + debug := flag.Bool("debug", false, "enable debug-level logging, including verbose per-connection scan logging") showVersion := flag.Bool("version", false, "print version and exit") flag.Parse() - if *showVersion { - fmt.Printf("%s (%s)\n", version, platform) - return 0 + cfg := &runConfig{ + dbPath: strings.TrimSpace(*dbPath), + hexwallDBPath: strings.TrimSpace(*hexwallDB), + zeekLogPath: strings.TrimSpace(*zeekLogPath), + zeekNoticeLog: strings.TrimSpace(*zeekNoticeLog), + enableZeek: *enableZeek, + enableDeghostIP: *enableDeghostIP, + enableDeghostDomain: *enableDeghostDomain, + mode: strings.TrimSpace(*mode), + debug: *debug, + showVersion: *showVersion, + } + + if cfg.showVersion { + return cfg, nil } - selectedMode := strings.ToLower(strings.TrimSpace(*mode)) + selectedMode := strings.ToLower(cfg.mode) if selectedMode != monitor.ModeWatch && selectedMode != monitor.ModeEnforce { - slog.Error("invalid --mode value", "mode", *mode, "allowed", []string{monitor.ModeWatch, monitor.ModeEnforce}) + return nil, fmt.Errorf("invalid --mode value %q: allowed %q or %q", cfg.mode, monitor.ModeWatch, monitor.ModeEnforce) + } + cfg.mode = selectedMode + + if cfg.hexwallDBPath == "" { + return nil, fmt.Errorf("invalid --hexwall-db value %q", *hexwallDB) + } + + return cfg, nil +} + +func run() int { + cfg, err := parseFlags() + if err != nil { + slog.Error(err.Error()) return 1 } - resolvedDBPath := strings.TrimSpace(*dbPath) - hexwallDBPath := strings.TrimSpace(*hexwallDB) - if hexwallDBPath == "" { - slog.Error("invalid --hexwall-db value", "path", *hexwallDB) + if cfg == nil { return 1 } + // Install the log handler first so every later slog call honours --debug. + logLevel := slog.LevelInfo + if cfg.debug { + logLevel = slog.LevelDebug + } + slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}))) + + if cfg.showVersion { + fmt.Printf("%s (%s)\n", version, platform) + return 0 + } + // Cancel background work cleanly on Ctrl+C. ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() @@ -72,21 +128,21 @@ func run() int { } // 2. Resolve the Pi-hole database path from --db or auto-detection. - if resolvedDBPath == "" { + if cfg.dbPath == "" { detected, err := detector.FindDBPath() if err != nil { slog.Error("could not find pi-hole installation", "err", err) return 1 } - resolvedDBPath = detected - slog.Info("pi-hole database auto-detected", "path", resolvedDBPath) + cfg.dbPath = detected + slog.Info("pi-hole database auto-detected", "path", cfg.dbPath) } // 3. Open the Pi-hole database in read-only mode. - checker, err := pihole.NewChecker(&pihole.Config{DBPath: resolvedDBPath}) + checker, err := pihole.NewChecker(&pihole.Config{DBPath: cfg.dbPath}) if err != nil { - slog.Error("failed to open pi-hole database", "path", resolvedDBPath, "err", err) + slog.Error("failed to open pi-hole database", "path", cfg.dbPath, "err", err) return 1 } defer func() { @@ -95,10 +151,25 @@ func run() int { } }() + // 3b. Log policy list summary from gravity.db. + if counts, err := checker.PolicyCounts(); err != nil { + slog.Warn("failed to read pi-hole policy counts", "err", err) + } else if counts != nil { + slog.Info("pihole policy lists loaded", + "allow", counts["vw_allowlist"], + "deny", counts["vw_denylist"], + "gravity", counts["vw_gravity"], + "regex_allow", counts["vw_regex_allowlist"], + "regex_deny", counts["vw_regex_denylist"]) + if counts["vw_gravity"] == 0 { + slog.Warn("pi-hole gravity list is empty; blocklist may be broken or not yet downloaded") + } + } + // 4. Open the local hexwall database, creating it if needed. - hexwallStore, err := store.NewStore(hexwallDBPath) + hexwallStore, err := store.NewStore(cfg.hexwallDBPath) if err != nil { - slog.Error("failed to open hexwall database", "path", hexwallDBPath, "err", err) + slog.Error("failed to open hexwall database", "path", cfg.hexwallDBPath, "err", err) return 1 } defer func() { @@ -107,9 +178,67 @@ func run() int { } }() - slog.Info("hexwall database ready", "path", hexwallDBPath) + slog.Info("hexwall database ready", "path", cfg.hexwallDBPath) + + // 4a. Prune stale rows from prior runs at startup. + if result, err := hexwallStore.Prune(); err != nil { + slog.Warn("initial prune failed", "err", err) + } else if result != nil && (result.AllowedIPDomains+result.SNIObservations+result.IPObservations+result.KilledConnections+result.ZeekAlerts) > 0 { + slog.Info("pruned stale rows", + "allowed_ip_domains", result.AllowedIPDomains, + "sni_observations", result.SNIObservations, + "ip_observations", result.IPObservations, + "killed_connections", result.KilledConnections, + "zeek_alerts", result.ZeekAlerts) + } - deghostClient := deghost.NewClient(deghostBaseURL, deghostTimeout) + // 4b. Start the Zeek ssl.log tailer when a path is provided. + var zeekClient *zeek.Client + if cfg.zeekLogPath != "" { + zc, err := zeek.NewClient(&zeek.Config{ + LogPath: cfg.zeekLogPath, + }) + + if err != nil { + slog.Error("failed to start zeek log watcher", "path", cfg.zeekLogPath, "err", err) + return 1 + } + + defer zc.Close() + zeekClient = zc + } + + var zeekEvents = make(chan zeek.Event, 100) + if cfg.enableZeek && cfg.zeekNoticeLog != "" { + go func() { + if err := zeek.WatchNoticeLog(ctx, cfg.zeekNoticeLog, zeekEvents); err != nil { + slog.Error("zeek notice watcher stopped", "path", cfg.zeekNoticeLog, "err", err) + } + }() + go func() { + for { + select { + case <-ctx.Done(): + return + case event := <-zeekEvents: + monitor.HandleZeekEvent(checker, hexwallStore, cfg.mode, event) + } + } + }() + } + + // deghost is opt-in: it sends every unrecognized domain to a third-party API, + // which is a real privacy cost. Only construct the client when at least one check is enabled. + var deghostClient *deghost.Client + if cfg.enableDeghostIP || cfg.enableDeghostDomain { + deghostClient = deghost.NewClient(deghostBaseURL, deghostTimeout) + } else { + slog.Info("deghost disabled; unrecognized connections will not be escalated to third-party API") + } + + if cfg.mode == monitor.ModeEnforce && cfg.enableDeghostIP { + slog.Warn("enforce mode with IP deghost enabled: an external reputation verdict can now terminate a local process; shared cloud provider ranges are known to produce false positives on such feeds") + } // 5. Refresh trusted IPs before starting the monitor so the first tick does not kill legitimate connections. cache := pihole.NewIPCache(checker, hexwallStore) @@ -130,9 +259,33 @@ func run() int { } }() + // 6b. Periodically prune stale rows from the hexwall database. + go func() { + ticker := time.NewTicker(time.Hour) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if result, err := hexwallStore.Prune(); err != nil { + slog.Warn("periodic prune failed", "err", err) + } else if result != nil && (result.AllowedIPDomains+result.SNIObservations+result.IPObservations+result.KilledConnections+result.ZeekAlerts) > 0 { + slog.Info("pruned stale rows", + "allowed_ip_domains", result.AllowedIPDomains, + "sni_observations", result.SNIObservations, + "ip_observations", result.IPObservations, + "killed_connections", result.KilledConnections, + "zeek_alerts", result.ZeekAlerts) + } + } + } + }() + fmt.Println("Starting network monitor...") - fmt.Printf("> Connections will be checked every %s in %s mode.\n", connectionScanInterval, selectedMode) - if *debug { + fmt.Printf("> Connections will be checked every %s in %s mode.\n", connectionScanInterval, cfg.mode) + if cfg.debug { fmt.Println("> Debug logging is enabled for every scanned connection.") } @@ -146,7 +299,7 @@ func run() int { slog.Info("shutting down") return 0 case <-ticker.C: - monitor.RunScan(ctx, hexwallStore, deghostClient, selectedMode, *debug) + monitor.RunScan(ctx, checker, hexwallStore, deghostClient, cfg.enableDeghostIP, cfg.enableDeghostDomain, zeekClient, cfg.mode, cfg.debug) } } }