From e8f56a1ef497266e4ace2329fc13a2e266384f21 Mon Sep 17 00:00:00 2001 From: Gary Hansen Date: Wed, 8 Jul 2026 02:43:50 +1000 Subject: [PATCH 1/3] feat(receiver): storage layer and webhook payload compat Dual-dialect store (SQLite via modernc.org, MySQL via go-sql-driver, both pure Go) with order-tolerant start/complete upserts, filtered and paginated listing, and aggregate queries (per-day, top domains, query types, statuses, duration percentiles, top clients). Sender gains optional EXPLOREDNS_WEBHOOK_TOKEN bearer auth; a round-trip test pins receiver structs byte-compatible with the sender payloads. Note: go directive moves to 1.25.0, required by modernc.org/sqlite. CI reads the version from go.mod so GOTOOLCHAIN=local stays satisfied. Co-Authored-By: Claude Fable 5 --- go.mod | 25 +- go.sum | 71 +++- internal/receiver/store/dialect.go | 103 +++++ internal/receiver/store/events.go | 39 ++ internal/receiver/store/query.go | 322 +++++++++++++++ internal/receiver/store/store.go | 217 ++++++++++ internal/receiver/store/store_test.go | 544 ++++++++++++++++++++++++++ web/api/handler.go | 2 +- web/api/webhook.go | 10 +- web/api/webhook_compat_test.go | 126 ++++++ web/api/webhook_test.go | 32 +- 11 files changed, 1468 insertions(+), 23 deletions(-) create mode 100644 internal/receiver/store/dialect.go create mode 100644 internal/receiver/store/events.go create mode 100644 internal/receiver/store/query.go create mode 100644 internal/receiver/store/store.go create mode 100644 internal/receiver/store/store_test.go create mode 100644 web/api/webhook_compat_test.go diff --git a/go.mod b/go.mod index d12f85c..008327e 100644 --- a/go.mod +++ b/go.mod @@ -1,16 +1,27 @@ module gitea.hansenits.com.au/hits/ExploreDNS -go 1.24.0 +go 1.25.0 require ( + github.com/go-sql-driver/mysql v1.10.0 github.com/miekg/dns v1.1.72 - golang.org/x/net v0.48.0 + golang.org/x/net v0.54.0 + modernc.org/sqlite v1.53.0 ) require ( - golang.org/x/mod v0.31.0 // indirect - golang.org/x/sync v0.19.0 // indirect - golang.org/x/sys v0.39.0 // indirect - golang.org/x/text v0.32.0 // indirect - golang.org/x/tools v0.40.0 // indirect + filippo.io/edwards25519 v1.2.0 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/mod v0.36.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.44.0 // indirect + golang.org/x/text v0.37.0 // indirect + golang.org/x/tools v0.45.0 // indirect + modernc.org/libc v1.73.4 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.11.0 // indirect ) diff --git a/go.sum b/go.sum index 04b719e..88c35a0 100644 --- a/go.sum +++ b/go.sum @@ -1,16 +1,63 @@ +filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo= +filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/go-sql-driver/mysql v1.10.0 h1:Q+1LV8DkHJvSYAdR83XzuhDaTykuDx0l6fkXxoWCWfw= +github.com/go-sql-driver/mysql v1.10.0/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/miekg/dns v1.1.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI= github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs= -golang.org/x/mod v0.31.0 h1:HaW9xtz0+kOcWKwli0ZXy79Ix+UW/vOfmWI5QVd2tgI= -golang.org/x/mod v0.31.0/go.mod h1:43JraMp9cGx1Rx3AqioxrbrhNsLl2l/iNAvuBkrezpg= -golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= -golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= -golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk= -golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU= -golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY= -golang.org/x/tools v0.40.0 h1:yLkxfA+Qnul4cs9QA3KnlFu0lVmd8JJfoq+E41uSutA= -golang.org/x/tools v0.40.0/go.mod h1:Ik/tzLRlbscWpqqMRjyWYDisX8bG13FrdXp3o4Sr9lc= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4= +golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ= +golang.org/x/net v0.54.0 h1:2zJIZAxAHV/OHCDTCOHAYehQzLfSXuf/5SoL/Dv6w/w= +golang.org/x/net v0.54.0/go.mod h1:Sj4oj8jK6XmHpBZU/zWHw3BV3abl4Kvi+Ut7cQcY+cQ= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= +golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= +golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8= +golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0= +modernc.org/cc/v4 v4.28.4 h1:Hd/4Es+MBj+/7hSdZaisNyu6bv3V0Dp2MdllyfqaH+c= +modernc.org/cc/v4 v4.28.4/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= +modernc.org/ccgo/v4 v4.34.4 h1:OVnSOWQjVKOYkFxoHYB+qQmSHK5gqMqARM+K9DpR/Ws= +modernc.org/ccgo/v4 v4.34.4/go.mod h1:qdKqE8FNIYyysougB1RX9MxCzp5oJOcQXSobANJ4TuE= +modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= +modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/gc/v3 v3.1.3 h1:6QAplYyVO+KdPW3pGnqmJDUxtkec8ooEWvks/hhU3lc= +modernc.org/gc/v3 v3.1.3/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= +modernc.org/libc v1.73.4 h1:+ra4Ui8ngyt8HDcO1FTDPWlkAh6yOdaO2yAoh8MddQA= +modernc.org/libc v1.73.4/go.mod h1:DXZ3eO8qMCNn2SnmTNCiC71nJ9Rcq3PsnpU6Vc4rWK8= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= +modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg= +modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= +modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= +modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= +modernc.org/sqlite v1.53.0 h1:20WG8N9q4ji/dEqGk4uiI0c6OPjSeLTNYGFCc3+7c1M= +modernc.org/sqlite v1.53.0/go.mod h1:xoEpOIpGrgT48H5iiyt/YXPCZPEzlfmfFwtk8Lklw8s= +modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= +modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/internal/receiver/store/dialect.go b/internal/receiver/store/dialect.go new file mode 100644 index 0000000..28daf16 --- /dev/null +++ b/internal/receiver/store/dialect.go @@ -0,0 +1,103 @@ +package store + +import ( + "context" + "database/sql" + "strings" +) + +// dialect abstracts the SQL syntax differences between the sqlite and +// mysql backends. Both drivers use '?' placeholders, and all datetimes are +// bound and scanned as "2006-01-02 15:04:05.000000" UTC strings so value +// handling stays identical; only the constructs below diverge. +type dialect interface { + name() string + + // datetimeType is the column type for datetime values. sqlite has no + // datetime type (TEXT affinity stores our formatted strings verbatim, + // which sort and compare lexically); mysql needs DATETIME(6) because + // plain DATETIME truncates the sub-second precision we write. + datetimeType() string + + // upsert appends the insert-or-update clause to an INSERT statement, + // overwriting cols with the values from the attempted insert. sqlite + // spells this ON CONFLICT(id) DO UPDATE SET col=excluded.col; mysql + // spells it ON DUPLICATE KEY UPDATE col=VALUES(col) (VALUES() is + // deprecated in MySQL 8.0.20+ but is the only form MariaDB and older + // MySQL also accept). + upsert(insert string, cols []string) string + + // createIndex creates an index if it does not exist. sqlite supports + // CREATE INDEX IF NOT EXISTS; mysql has no IF NOT EXISTS for indexes, + // so existence is checked via information_schema first. + createIndex(ctx context.Context, db *sql.DB, index, table, column string) error + + // likeContains returns a "col contains substr" predicate plus its + // bind argument, with LIKE wildcards in substr escaped. The ESCAPE + // literal differs: mysql string literals treat backslash as an escape + // character (so the SQL needs '\\'), sqlite ones do not (so it needs + // '\', and has no default escape character at all). + likeContains(col, substr string) (predicate string, arg string) +} + +// escapeLike backslash-escapes LIKE pattern metacharacters in s. +func escapeLike(s string) string { + r := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`) + return r.Replace(s) +} + +type sqliteDialect struct{} + +func (sqliteDialect) name() string { return "sqlite" } +func (sqliteDialect) datetimeType() string { return "TEXT" } + +func (sqliteDialect) upsert(insert string, cols []string) string { + sets := make([]string, len(cols)) + for i, c := range cols { + sets[i] = c + "=excluded." + c + } + return insert + " ON CONFLICT(id) DO UPDATE SET " + strings.Join(sets, ", ") +} + +func (sqliteDialect) createIndex(ctx context.Context, db *sql.DB, index, table, column string) error { + _, err := db.ExecContext(ctx, + "CREATE INDEX IF NOT EXISTS "+index+" ON "+table+" ("+column+")") + return err +} + +func (sqliteDialect) likeContains(col, substr string) (string, string) { + return col + ` LIKE ? ESCAPE '\'`, "%" + escapeLike(substr) + "%" +} + +type mysqlDialect struct{} + +func (mysqlDialect) name() string { return "mysql" } +func (mysqlDialect) datetimeType() string { return "DATETIME(6)" } + +func (mysqlDialect) upsert(insert string, cols []string) string { + sets := make([]string, len(cols)) + for i, c := range cols { + sets[i] = c + "=VALUES(" + c + ")" + } + return insert + " ON DUPLICATE KEY UPDATE " + strings.Join(sets, ", ") +} + +func (mysqlDialect) createIndex(ctx context.Context, db *sql.DB, index, table, column string) error { + var n int + err := db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM information_schema.statistics + WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?`, + table, index).Scan(&n) + if err != nil { + return err + } + if n > 0 { + return nil + } + _, err = db.ExecContext(ctx, "CREATE INDEX "+index+" ON "+table+" ("+column+")") + return err +} + +func (mysqlDialect) likeContains(col, substr string) (string, string) { + return col + ` LIKE ? ESCAPE '\\'`, "%" + escapeLike(substr) + "%" +} diff --git a/internal/receiver/store/events.go b/internal/receiver/store/events.go new file mode 100644 index 0000000..fa12d55 --- /dev/null +++ b/internal/receiver/store/events.go @@ -0,0 +1,39 @@ +package store + +import ( + "encoding/json" + "time" +) + +// StartEvent and CompleteEvent are the receiver-side decodings of the +// webhook payloads posted by the main app. Field names and JSON tags must +// stay in sync with webhookStartEvent/webhookCompleteEvent in +// web/api/webhook.go; a compatibility test in web/api enforces this. + +// StartEvent mirrors web/api webhookStartEvent. +type StartEvent struct { + Event string `json:"event"` + ID string `json:"id"` + Domain string `json:"domain"` + QueryType string `json:"query_type"` + AllRoots bool `json:"all_roots"` + ClientIP string `json:"client_ip"` + StartedAt time.Time `json:"started_at"` +} + +// CompleteEvent mirrors web/api webhookCompleteEvent. Summary is kept as +// raw JSON: the receiver stores it verbatim and never interprets it. +type CompleteEvent struct { + Event string `json:"event"` + ID string `json:"id"` + Domain string `json:"domain"` + QueryType string `json:"query_type"` + ClientIP string `json:"client_ip"` + StartedAt time.Time `json:"started_at"` + DoneAt time.Time `json:"done_at"` + DurationMS int64 `json:"duration_ms"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + ResultCount int `json:"result_count"` + Summary json.RawMessage `json:"summary"` +} diff --git a/internal/receiver/store/query.go b/internal/receiver/store/query.go new file mode 100644 index 0000000..c8a438c --- /dev/null +++ b/internal/receiver/store/query.go @@ -0,0 +1,322 @@ +package store + +import ( + "context" + "database/sql" + "math" + "sort" + "strings" + "time" +) + +// Traversal is one stored traversal row. Pointer fields are NULL until the +// complete event arrives. +type Traversal struct { + ID string + Domain string + QueryType string + AllRoots bool + ClientIP string + StartedAt time.Time + DoneAt *time.Time + DurationMS *int64 + Status string + Error string + ResultCount *int + Summary string + FirstSeen time.Time + LastSeen time.Time +} + +// ListFilter narrows ListTraversals. Zero values mean "no constraint"; +// From/To bound started_at inclusively. +type ListFilter struct { + Domain string // substring match + Status string + From time.Time + To time.Time +} + +const defaultListLimit = 50 + +// where renders the filter as a WHERE clause (or "") plus bind args. +func (f ListFilter) where(d dialect) (string, []any) { + var preds []string + var args []any + if f.Domain != "" { + p, arg := d.likeContains("domain", f.Domain) + preds = append(preds, p) + args = append(args, arg) + } + if f.Status != "" { + preds = append(preds, "status = ?") + args = append(args, f.Status) + } + if !f.From.IsZero() { + preds = append(preds, "started_at >= ?") + args = append(args, fmtTime(f.From)) + } + if !f.To.IsZero() { + preds = append(preds, "started_at <= ?") + args = append(args, fmtTime(f.To)) + } + if len(preds) == 0 { + return "", nil + } + return " WHERE " + strings.Join(preds, " AND "), args +} + +// ListTraversals returns one page of matching traversals ordered newest +// first, plus the total match count. limit <= 0 selects a default page size. +func (s *Store) ListTraversals(ctx context.Context, f ListFilter, limit, offset int) ([]Traversal, int, error) { + where, args := f.where(s.d) + + var total int + if err := s.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM traversals"+where, args...).Scan(&total); err != nil { + return nil, 0, err + } + + if limit <= 0 { + limit = defaultListLimit + } + if offset < 0 { + offset = 0 + } + q := `SELECT id, domain, query_type, all_roots, client_ip, started_at, done_at, + duration_ms, status, error, result_count, summary, first_seen, last_seen + FROM traversals` + where + ` ORDER BY started_at DESC, id LIMIT ? OFFSET ?` + rows, err := s.db.QueryContext(ctx, q, append(args, limit, offset)...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var out []Traversal + for rows.Next() { + var ( + tr Traversal + startedAt, doneAt sql.NullString + errMsg, summary sql.NullString + durationMS sql.NullInt64 + resultCount sql.NullInt64 + firstSeen, lastSeen string + ) + if err := rows.Scan(&tr.ID, &tr.Domain, &tr.QueryType, &tr.AllRoots, &tr.ClientIP, + &startedAt, &doneAt, &durationMS, &tr.Status, &errMsg, &resultCount, + &summary, &firstSeen, &lastSeen); err != nil { + return nil, 0, err + } + tr.Error = errMsg.String + tr.Summary = summary.String + if durationMS.Valid { + v := durationMS.Int64 + tr.DurationMS = &v + } + if resultCount.Valid { + v := int(resultCount.Int64) + tr.ResultCount = &v + } + if startedAt.Valid { + if tr.StartedAt, err = parseDBTime(startedAt.String); err != nil { + return nil, 0, err + } + } + if doneAt.Valid { + t, err := parseDBTime(doneAt.String) + if err != nil { + return nil, 0, err + } + tr.DoneAt = &t + } + if tr.FirstSeen, err = parseDBTime(firstSeen); err != nil { + return nil, 0, err + } + if tr.LastSeen, err = parseDBTime(lastSeen); err != nil { + return nil, 0, err + } + out = append(out, tr) + } + return out, total, rows.Err() +} + +// Totals summarises whole-table traversal volume. Errors counts all-time +// error-status rows so callers can derive an error rate. +type Totals struct { + AllTime int + Last24h int + Last7d int + DistinctDomains int + DistinctClients int + Errors int +} + +// Totals returns whole-table totals with the recent windows measured back +// from now. COALESCE covers the empty table, where SUM yields NULL. +func (s *Store) Totals(ctx context.Context, now time.Time) (Totals, error) { + var t Totals + err := s.db.QueryRowContext(ctx, `SELECT COUNT(*), + COALESCE(SUM(CASE WHEN started_at >= ? THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN started_at >= ? THEN 1 ELSE 0 END), 0), + COUNT(DISTINCT domain), + COUNT(DISTINCT client_ip), + COALESCE(SUM(CASE WHEN status = ? THEN 1 ELSE 0 END), 0) + FROM traversals`, + fmtTime(now.Add(-24*time.Hour)), fmtTime(now.AddDate(0, 0, -7)), StatusError). + Scan(&t.AllTime, &t.Last24h, &t.Last7d, &t.DistinctDomains, &t.DistinctClients, &t.Errors) + return t, err +} + +// DayStats is one calendar day's traversal volume. +type DayStats struct { + Day string // "2006-01-02" (UTC) + Total int + Errors int +} + +// StatsPerDay returns per-day totals and error counts for the last days +// calendar days (UTC), oldest first. Days with no traffic are omitted. +func (s *Store) StatsPerDay(ctx context.Context, days int) ([]DayStats, error) { + if days <= 0 { + days = 1 + } + midnight := time.Now().UTC().Truncate(24 * time.Hour) + cutoff := midnight.AddDate(0, 0, -(days - 1)) + // substr on the stored value yields "YYYY-MM-DD" in both backends + // (MySQL casts DATETIME to its string form implicitly). + rows, err := s.db.QueryContext(ctx, `SELECT substr(started_at, 1, 10) AS day, + COUNT(*), SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) + FROM traversals WHERE started_at >= ? GROUP BY day ORDER BY day`, + StatusError, fmtTime(cutoff)) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []DayStats + for rows.Next() { + var d DayStats + if err := rows.Scan(&d.Day, &d.Total, &d.Errors); err != nil { + return nil, err + } + out = append(out, d) + } + return out, rows.Err() +} + +// NameCount is a generic (value, count) aggregate row. +type NameCount struct { + Name string + Count int +} + +// TopDomains returns the most-queried domains since the given time (zero = +// all time), most frequent first, at most limit rows. +func (s *Store) TopDomains(ctx context.Context, since time.Time, limit int) ([]NameCount, error) { + return s.countBy(ctx, "domain", since, limit) +} + +// TopClientIPs returns the most active client IPs since the given time. +func (s *Store) TopClientIPs(ctx context.Context, since time.Time, limit int) ([]NameCount, error) { + return s.countBy(ctx, "client_ip", since, limit) +} + +// QueryTypeCounts returns traversal counts per query type since the given +// time, most frequent first. +func (s *Store) QueryTypeCounts(ctx context.Context, since time.Time) ([]NameCount, error) { + return s.countBy(ctx, "query_type", since, 0) +} + +// StatusCounts returns traversal counts per status since the given time, +// most frequent first. +func (s *Store) StatusCounts(ctx context.Context, since time.Time) ([]NameCount, error) { + return s.countBy(ctx, "status", since, 0) +} + +// countBy groups rows by col and counts them. col is always one of the +// fixed column names above, never user input. +func (s *Store) countBy(ctx context.Context, col string, since time.Time, limit int) ([]NameCount, error) { + q := "SELECT " + col + ", COUNT(*) AS n FROM traversals" + var args []any + if !since.IsZero() { + q += " WHERE started_at >= ?" + args = append(args, fmtTime(since)) + } + q += " GROUP BY " + col + " ORDER BY n DESC, " + col + if limit > 0 { + q += " LIMIT ?" + args = append(args, limit) + } + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []NameCount + for rows.Next() { + var nc NameCount + if err := rows.Scan(&nc.Name, &nc.Count); err != nil { + return nil, err + } + out = append(out, nc) + } + return out, rows.Err() +} + +// DurationStats summarises completed-traversal durations. +type DurationStats struct { + Count int + AvgMS float64 + P50MS int64 + P95MS int64 +} + +// Durations returns duration statistics for traversals started since the +// given time (zero = all time). Percentiles are computed in Go (nearest +// rank) because SQL percentile support differs between the backends. +func (s *Store) Durations(ctx context.Context, since time.Time) (DurationStats, error) { + q := "SELECT duration_ms FROM traversals WHERE duration_ms IS NOT NULL" + var args []any + if !since.IsZero() { + q += " AND started_at >= ?" + args = append(args, fmtTime(since)) + } + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return DurationStats{}, err + } + defer rows.Close() + + var values []int64 + var sum int64 + for rows.Next() { + var v int64 + if err := rows.Scan(&v); err != nil { + return DurationStats{}, err + } + values = append(values, v) + sum += v + } + if err := rows.Err(); err != nil { + return DurationStats{}, err + } + if len(values) == 0 { + return DurationStats{}, nil + } + sort.Slice(values, func(i, j int) bool { return values[i] < values[j] }) + return DurationStats{ + Count: len(values), + AvgMS: float64(sum) / float64(len(values)), + P50MS: percentile(values, 0.50), + P95MS: percentile(values, 0.95), + }, nil +} + +// percentile returns the nearest-rank percentile of sorted values. +func percentile(sorted []int64, q float64) int64 { + rank := int(math.Ceil(q * float64(len(sorted)))) + if rank < 1 { + rank = 1 + } + return sorted[rank-1] +} diff --git a/internal/receiver/store/store.go b/internal/receiver/store/store.go new file mode 100644 index 0000000..428cb75 --- /dev/null +++ b/internal/receiver/store/store.go @@ -0,0 +1,217 @@ +// Package store persists webhook telemetry events from the ExploreDNS API +// server into MySQL or SQLite through a shared database/sql layer. +package store + +import ( + "context" + "database/sql" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/go-sql-driver/mysql" + _ "modernc.org/sqlite" +) + +// Traversal status values, mirroring the job statuses in web/api. +const ( + StatusRunning = "running" + StatusComplete = "complete" + StatusError = "error" +) + +// schemaVersion is recorded in the schema_version table on first open so +// future releases can detect and migrate older layouts. +const schemaVersion = 1 + +// dbTimeLayout is the canonical datetime encoding: fixed-width so sqlite +// TEXT comparisons order chronologically, and a valid MySQL DATETIME(6) +// literal. +const dbTimeLayout = "2006-01-02 15:04:05.000000" + +// Store persists traversal telemetry events. +type Store struct { + db *sql.DB + d dialect +} + +// OpenSQLite opens (creating if needed) a SQLite-backed store at path, +// creating parent directories first. +func OpenSQLite(path string) (*Store, error) { + if dir := filepath.Dir(path); dir != "." && dir != "" { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("create sqlite directory: %w", err) + } + } + db, err := sql.Open("sqlite", path) + if err != nil { + return nil, fmt.Errorf("open sqlite %s: %w", path, err) + } + // A single connection serialises writers so concurrent ingests never + // see SQLITE_BUSY. + db.SetMaxOpenConns(1) + return open(db, sqliteDialect{}) +} + +// OpenMySQL opens a MySQL-backed store using a go-sql-driver DSN +// (user:pass@tcp(host:port)/dbname). +func OpenMySQL(dsn string) (*Store, error) { + if _, err := mysql.ParseDSN(dsn); err != nil { + return nil, fmt.Errorf("parse mysql dsn: %w", err) + } + db, err := sql.Open("mysql", dsn) + if err != nil { + return nil, fmt.Errorf("open mysql: %w", err) + } + return open(db, mysqlDialect{}) +} + +// RedactMySQLDSN returns dsn with any password replaced, safe for logging. +func RedactMySQLDSN(dsn string) string { + cfg, err := mysql.ParseDSN(dsn) + if err != nil { + return "(unparsable DSN)" + } + if cfg.Passwd != "" { + cfg.Passwd = "xxxxx" + } + return cfg.FormatDSN() +} + +func open(db *sql.DB, d dialect) (*Store, error) { + s := &Store{db: db, d: d} + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := s.init(ctx); err != nil { + db.Close() + return nil, fmt.Errorf("init %s schema: %w", d.name(), err) + } + return s, nil +} + +// Backend names the database backend ("sqlite" or "mysql"). +func (s *Store) Backend() string { return s.d.name() } + +// Close releases the underlying database handle. +func (s *Store) Close() error { return s.db.Close() } + +// init creates the schema if missing and stamps schema_version. +func (s *Store) init(ctx context.Context) error { + dt := s.d.datetimeType() + stmts := []string{ + `CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)`, + fmt.Sprintf(`CREATE TABLE IF NOT EXISTS traversals ( + id VARCHAR(64) NOT NULL PRIMARY KEY, + domain VARCHAR(255) NOT NULL DEFAULT '', + query_type VARCHAR(16) NOT NULL DEFAULT '', + all_roots BOOLEAN NOT NULL DEFAULT 0, + client_ip VARCHAR(64) NOT NULL DEFAULT '', + started_at %s NULL, + done_at %s NULL, + duration_ms BIGINT NULL, + status VARCHAR(16) NOT NULL DEFAULT '', + error TEXT NULL, + result_count INT NULL, + summary TEXT NULL, + first_seen %s NOT NULL, + last_seen %s NOT NULL + )`, dt, dt, dt, dt), + } + for _, q := range stmts { + if _, err := s.db.ExecContext(ctx, q); err != nil { + return err + } + } + for _, idx := range [][2]string{ + {"idx_traversals_started_at", "started_at"}, + {"idx_traversals_domain", "domain"}, + {"idx_traversals_client_ip", "client_ip"}, + {"idx_traversals_status", "status"}, + } { + if err := s.d.createIndex(ctx, s.db, idx[0], "traversals", idx[1]); err != nil { + return err + } + } + // Stamp the version on first creation only. Checked in Go because + // MySQL and sqlite disagree on FROM-less SELECT ... WHERE syntax. + var n int + if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM schema_version`).Scan(&n); err != nil { + return err + } + if n == 0 { + _, err := s.db.ExecContext(ctx, `INSERT INTO schema_version (version) VALUES (?)`, schemaVersion) + return err + } + return nil +} + +// RecordStart upserts a "start" event. Duplicate deliveries refresh the +// request attributes but never touch status or completion fields, so a +// retried start arriving after the complete event cannot clobber the +// terminal state. +func (s *Store) RecordStart(ctx context.Context, ev StartEvent) error { + now := fmtTime(time.Now()) + q := s.d.upsert(`INSERT INTO traversals + (id, domain, query_type, all_roots, client_ip, started_at, status, first_seen, last_seen) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + []string{"domain", "query_type", "all_roots", "client_ip", "started_at", "last_seen"}) + _, err := s.db.ExecContext(ctx, q, + ev.ID, ev.Domain, ev.QueryType, ev.AllRoots, ev.ClientIP, + dbTime(ev.StartedAt), StatusRunning, now, now) + return err +} + +// RecordComplete upserts a "complete" event. It creates the row when the +// start event has not arrived (out-of-order delivery) and overwrites the +// completion fields on duplicates; all_roots and first_seen are start-only +// and left untouched. +func (s *Store) RecordComplete(ctx context.Context, ev CompleteEvent) error { + now := fmtTime(time.Now()) + q := s.d.upsert(`INSERT INTO traversals + (id, domain, query_type, client_ip, started_at, done_at, duration_ms, + status, error, result_count, summary, first_seen, last_seen) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + []string{"domain", "query_type", "client_ip", "started_at", "done_at", + "duration_ms", "status", "error", "result_count", "summary", "last_seen"}) + var summary any + if len(ev.Summary) > 0 { + summary = string(ev.Summary) + } + _, err := s.db.ExecContext(ctx, q, + ev.ID, ev.Domain, ev.QueryType, ev.ClientIP, + dbTime(ev.StartedAt), dbTime(ev.DoneAt), ev.DurationMS, + ev.Status, nullIfEmpty(ev.Error), ev.ResultCount, summary, now, now) + return err +} + +// fmtTime encodes t for storage. +func fmtTime(t time.Time) string { return t.UTC().Format(dbTimeLayout) } + +// dbTime encodes t for a nullable datetime column; zero times become NULL +// (MySQL DATETIME cannot hold year 1). +func dbTime(t time.Time) any { + if t.IsZero() { + return nil + } + return fmtTime(t) +} + +func nullIfEmpty(s string) any { + if s == "" { + return nil + } + return s +} + +// parseDBTime decodes a stored datetime. Besides the canonical layout it +// accepts second precision and RFC 3339 in case the MySQL DSN enables +// parseTime (database/sql then hands strings back in RFC 3339). +func parseDBTime(s string) (time.Time, error) { + for _, layout := range []string{dbTimeLayout, "2006-01-02 15:04:05", time.RFC3339Nano} { + if t, err := time.ParseInLocation(layout, s, time.UTC); err == nil { + return t, nil + } + } + return time.Time{}, fmt.Errorf("unrecognised datetime %q", s) +} diff --git a/internal/receiver/store/store_test.go b/internal/receiver/store/store_test.go new file mode 100644 index 0000000..52bfc68 --- /dev/null +++ b/internal/receiver/store/store_test.go @@ -0,0 +1,544 @@ +package store + +import ( + "context" + "encoding/json" + "path/filepath" + "testing" + "time" +) + +func newTestStore(t *testing.T) *Store { + t.Helper() + s, err := OpenSQLite(filepath.Join(t.TempDir(), "receiver.db")) + if err != nil { + t.Fatalf("OpenSQLite: %v", err) + } + t.Cleanup(func() { s.Close() }) + return s +} + +func startEvent(id string) StartEvent { + return StartEvent{ + Event: "start", + ID: id, + Domain: "example.com", + QueryType: "A", + AllRoots: true, + ClientIP: "203.0.113.9", + StartedAt: time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC), + } +} + +func completeEvent(id string) CompleteEvent { + return CompleteEvent{ + Event: "complete", + ID: id, + Domain: "example.com", + QueryType: "A", + ClientIP: "203.0.113.9", + StartedAt: time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC), + DoneAt: time.Date(2026, 7, 1, 10, 0, 3, 0, time.UTC), + DurationMS: 3000, + Status: StatusComplete, + ResultCount: 4, + Summary: json.RawMessage(`{"answers":[{"probability":1,"records":["example.com 300 IN A 192.0.2.1"]}]}`), + } +} + +// getRow fetches a single traversal by exact id via ListTraversals. +func getRow(t *testing.T, s *Store, id string) Traversal { + t.Helper() + rows, _, err := s.ListTraversals(context.Background(), ListFilter{}, 1000, 0) + if err != nil { + t.Fatalf("ListTraversals: %v", err) + } + for _, r := range rows { + if r.ID == id { + return r + } + } + t.Fatalf("row %q not found", id) + return Traversal{} +} + +func TestRecordStartThenComplete(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + if err := s.RecordStart(ctx, startEvent("t1")); err != nil { + t.Fatalf("RecordStart: %v", err) + } + row := getRow(t, s, "t1") + if row.Status != StatusRunning { + t.Errorf("Status = %q, want %q", row.Status, StatusRunning) + } + if !row.AllRoots { + t.Error("AllRoots = false, want true") + } + if row.DoneAt != nil || row.DurationMS != nil || row.ResultCount != nil { + t.Errorf("completion fields set before complete: %+v", row) + } + if !row.StartedAt.Equal(startEvent("t1").StartedAt) { + t.Errorf("StartedAt = %v, want %v", row.StartedAt, startEvent("t1").StartedAt) + } + + ev := completeEvent("t1") + if err := s.RecordComplete(ctx, ev); err != nil { + t.Fatalf("RecordComplete: %v", err) + } + row = getRow(t, s, "t1") + if row.Status != StatusComplete { + t.Errorf("Status = %q, want %q", row.Status, StatusComplete) + } + if row.DoneAt == nil || !row.DoneAt.Equal(ev.DoneAt) { + t.Errorf("DoneAt = %v, want %v", row.DoneAt, ev.DoneAt) + } + if row.DurationMS == nil || *row.DurationMS != 3000 { + t.Errorf("DurationMS = %v, want 3000", row.DurationMS) + } + if row.ResultCount == nil || *row.ResultCount != 4 { + t.Errorf("ResultCount = %v, want 4", row.ResultCount) + } + if row.Summary != string(ev.Summary) { + t.Errorf("Summary = %q, want %q", row.Summary, ev.Summary) + } + if !row.AllRoots { + t.Error("AllRoots clobbered by complete event") + } +} + +func TestCompleteBeforeStart(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + if err := s.RecordComplete(ctx, completeEvent("t1")); err != nil { + t.Fatalf("RecordComplete: %v", err) + } + row := getRow(t, s, "t1") + if row.Status != StatusComplete { + t.Fatalf("Status = %q, want %q", row.Status, StatusComplete) + } + + // The delayed start must fill in start-only fields without touching + // the terminal state. + if err := s.RecordStart(ctx, startEvent("t1")); err != nil { + t.Fatalf("RecordStart: %v", err) + } + row = getRow(t, s, "t1") + if row.Status != StatusComplete { + t.Errorf("Status = %q after late start, want %q", row.Status, StatusComplete) + } + if row.DoneAt == nil || row.DurationMS == nil { + t.Errorf("completion fields lost after late start: %+v", row) + } + if !row.AllRoots { + t.Error("AllRoots not filled in by late start") + } +} + +func TestDuplicateStartAfterComplete(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + if err := s.RecordStart(ctx, startEvent("t1")); err != nil { + t.Fatalf("RecordStart: %v", err) + } + if err := s.RecordComplete(ctx, completeEvent("t1")); err != nil { + t.Fatalf("RecordComplete: %v", err) + } + if err := s.RecordStart(ctx, startEvent("t1")); err != nil { + t.Fatalf("duplicate RecordStart: %v", err) + } + + row := getRow(t, s, "t1") + if row.Status != StatusComplete { + t.Errorf("Status = %q after duplicate start, want %q", row.Status, StatusComplete) + } + if row.DoneAt == nil { + t.Error("DoneAt lost after duplicate start") + } + if row.Summary == "" { + t.Error("Summary lost after duplicate start") + } +} + +func TestDuplicateCompleteIdempotent(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + ev := completeEvent("t1") + ev.Status = StatusError + ev.Error = "traversal timed out after 5m0s" + for i := 0; i < 3; i++ { + if err := s.RecordComplete(ctx, ev); err != nil { + t.Fatalf("RecordComplete #%d: %v", i, err) + } + } + + rows, total, err := s.ListTraversals(ctx, ListFilter{}, 10, 0) + if err != nil { + t.Fatalf("ListTraversals: %v", err) + } + if total != 1 || len(rows) != 1 { + t.Fatalf("total = %d, len = %d, want 1 row", total, len(rows)) + } + if rows[0].Status != StatusError || rows[0].Error != ev.Error { + t.Errorf("row = %q/%q, want %q/%q", rows[0].Status, rows[0].Error, StatusError, ev.Error) + } +} + +// seedRows inserts a deterministic mixed dataset for list/aggregate tests. +func seedRows(t *testing.T, s *Store) { + t.Helper() + ctx := context.Background() + base := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + rows := []struct { + id, domain, qtype, ip, status string + day int + durMS int64 + }{ + {"a1", "example.com", "A", "203.0.113.1", StatusComplete, 0, 100}, + {"a2", "example.com", "AAAA", "203.0.113.1", StatusComplete, 0, 200}, + {"a3", "sub.example.com", "A", "203.0.113.2", StatusError, 1, 300}, + {"a4", "other.net", "MX", "203.0.113.3", StatusComplete, 1, 400}, + {"a5", "other.net", "A", "203.0.113.1", StatusComplete, 2, 500}, + {"a6", "under_score.org", "A", "203.0.113.4", StatusComplete, 2, 600}, + } + for i, r := range rows { + started := base.AddDate(0, 0, r.day).Add(time.Duration(i) * time.Minute) + ev := CompleteEvent{ + ID: r.id, Domain: r.domain, QueryType: r.qtype, ClientIP: r.ip, + StartedAt: started, DoneAt: started.Add(time.Duration(r.durMS) * time.Millisecond), + DurationMS: r.durMS, Status: r.status, ResultCount: 1, + } + if r.status == StatusError { + ev.Error = "lookup failed" + } + if err := s.RecordComplete(ctx, ev); err != nil { + t.Fatalf("seed %s: %v", r.id, err) + } + } + // One still-running traversal on day 2. + if err := s.RecordStart(ctx, StartEvent{ + ID: "a7", Domain: "running.io", QueryType: "A", ClientIP: "203.0.113.5", + StartedAt: base.AddDate(0, 0, 2).Add(time.Hour), + }); err != nil { + t.Fatalf("seed a7: %v", err) + } +} + +func TestListTraversalsFiltersAndPagination(t *testing.T) { + s := newTestStore(t) + seedRows(t, s) + ctx := context.Background() + + rows, total, err := s.ListTraversals(ctx, ListFilter{}, 3, 0) + if err != nil { + t.Fatalf("ListTraversals: %v", err) + } + if total != 7 || len(rows) != 3 { + t.Fatalf("total = %d, len = %d, want 7 and 3", total, len(rows)) + } + // Newest first: a7 (day2+1h) then a6, a5. + if rows[0].ID != "a7" || rows[1].ID != "a6" || rows[2].ID != "a5" { + t.Errorf("page 1 order = %s,%s,%s, want a7,a6,a5", rows[0].ID, rows[1].ID, rows[2].ID) + } + rows, _, err = s.ListTraversals(ctx, ListFilter{}, 3, 3) + if err != nil { + t.Fatalf("ListTraversals page 2: %v", err) + } + if rows[0].ID != "a4" || rows[1].ID != "a3" || rows[2].ID != "a2" { + t.Errorf("page 2 order = %s,%s,%s, want a4,a3,a2", rows[0].ID, rows[1].ID, rows[2].ID) + } + + // Domain substring matches example.com and sub.example.com. + rows, total, err = s.ListTraversals(ctx, ListFilter{Domain: "example"}, 0, 0) + if err != nil { + t.Fatalf("domain filter: %v", err) + } + if total != 3 { + t.Errorf("domain filter total = %d, want 3", total) + } + for _, r := range rows { + if r.Domain != "example.com" && r.Domain != "sub.example.com" { + t.Errorf("domain filter matched %q", r.Domain) + } + } + + // LIKE metacharacters in the filter must be literal, not wildcards. + _, total, err = s.ListTraversals(ctx, ListFilter{Domain: "%"}, 0, 0) + if err != nil { + t.Fatalf("percent filter: %v", err) + } + if total != 0 { + t.Errorf("%% filter total = %d, want 0", total) + } + _, total, err = s.ListTraversals(ctx, ListFilter{Domain: "under_score"}, 0, 0) + if err != nil { + t.Fatalf("underscore filter: %v", err) + } + if total != 1 { + t.Errorf("under_score filter total = %d, want 1", total) + } + + _, total, err = s.ListTraversals(ctx, ListFilter{Status: StatusError}, 0, 0) + if err != nil { + t.Fatalf("status filter: %v", err) + } + if total != 1 { + t.Errorf("status filter total = %d, want 1", total) + } + + base := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + rows, total, err = s.ListTraversals(ctx, ListFilter{ + From: base.AddDate(0, 0, 1), + To: base.AddDate(0, 0, 2).Add(-time.Second), + }, 0, 0) + if err != nil { + t.Fatalf("time filter: %v", err) + } + if total != 2 { + t.Fatalf("time filter total = %d, want 2 (got %+v)", total, rows) + } + if rows[0].ID != "a4" || rows[1].ID != "a3" { + t.Errorf("time filter = %s,%s, want a4,a3", rows[0].ID, rows[1].ID) + } +} + +func TestStatsPerDay(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + for i, spec := range []struct { + daysAgo int + status string + }{ + {0, StatusComplete}, {0, StatusError}, {1, StatusComplete}, {5, StatusComplete}, + } { + ev := completeEvent(string(rune('a' + i))) + ev.StartedAt = now.AddDate(0, 0, -spec.daysAgo) + ev.DoneAt = ev.StartedAt.Add(time.Second) + ev.Status = spec.status + if err := s.RecordComplete(ctx, ev); err != nil { + t.Fatalf("seed: %v", err) + } + } + + days, err := s.StatsPerDay(ctx, 3) + if err != nil { + t.Fatalf("StatsPerDay: %v", err) + } + if len(days) != 2 { + t.Fatalf("len = %d, want 2 (%+v)", len(days), days) + } + yesterday := now.AddDate(0, 0, -1).Format("2006-01-02") + today := now.Format("2006-01-02") + if days[0].Day != yesterday || days[0].Total != 1 || days[0].Errors != 0 { + t.Errorf("day[0] = %+v, want %s total 1 errors 0", days[0], yesterday) + } + if days[1].Day != today || days[1].Total != 2 || days[1].Errors != 1 { + t.Errorf("day[1] = %+v, want %s total 2 errors 1", days[1], today) + } +} + +func TestTopDomains(t *testing.T) { + s := newTestStore(t) + seedRows(t, s) + + top, err := s.TopDomains(context.Background(), time.Time{}, 2) + if err != nil { + t.Fatalf("TopDomains: %v", err) + } + want := []NameCount{{"example.com", 2}, {"other.net", 2}} + if len(top) != 2 || top[0] != want[0] || top[1] != want[1] { + t.Errorf("TopDomains = %+v, want %+v", top, want) + } + + // Window excludes day 0 rows (a1, a2), so example.com drops out and + // other.net (a4, a5) leads. + since := time.Date(2026, 7, 2, 0, 0, 0, 0, time.UTC) + top, err = s.TopDomains(context.Background(), since, 10) + if err != nil { + t.Fatalf("TopDomains since: %v", err) + } + if len(top) != 4 || top[0] != (NameCount{"other.net", 2}) { + t.Errorf("TopDomains since = %+v, want other.net x2 leading 4 domains", top) + } +} + +func TestQueryTypeCounts(t *testing.T) { + s := newTestStore(t) + seedRows(t, s) + + counts, err := s.QueryTypeCounts(context.Background(), time.Time{}) + if err != nil { + t.Fatalf("QueryTypeCounts: %v", err) + } + want := []NameCount{{"A", 5}, {"AAAA", 1}, {"MX", 1}} + if len(counts) != 3 || counts[0] != want[0] || counts[1] != want[1] || counts[2] != want[2] { + t.Errorf("QueryTypeCounts = %+v, want %+v", counts, want) + } +} + +func TestStatusCounts(t *testing.T) { + s := newTestStore(t) + seedRows(t, s) + + counts, err := s.StatusCounts(context.Background(), time.Time{}) + if err != nil { + t.Fatalf("StatusCounts: %v", err) + } + want := []NameCount{{StatusComplete, 5}, {StatusError, 1}, {StatusRunning, 1}} + if len(counts) != 3 || counts[0] != want[0] || counts[1] != want[1] || counts[2] != want[2] { + t.Errorf("StatusCounts = %+v, want %+v", counts, want) + } +} + +func TestTopClientIPs(t *testing.T) { + s := newTestStore(t) + seedRows(t, s) + + top, err := s.TopClientIPs(context.Background(), time.Time{}, 1) + if err != nil { + t.Fatalf("TopClientIPs: %v", err) + } + if len(top) != 1 || top[0].Name != "203.0.113.1" || top[0].Count != 3 { + t.Errorf("TopClientIPs = %+v, want 203.0.113.1 x3", top) + } +} + +func TestDurations(t *testing.T) { + s := newTestStore(t) + seedRows(t, s) + + // Durations 100..600; the running row has none and is excluded. + stats, err := s.Durations(context.Background(), time.Time{}) + if err != nil { + t.Fatalf("Durations: %v", err) + } + if stats.Count != 6 { + t.Fatalf("Count = %d, want 6", stats.Count) + } + if stats.AvgMS != 350 { + t.Errorf("AvgMS = %v, want 350", stats.AvgMS) + } + if stats.P50MS != 300 { + t.Errorf("P50MS = %d, want 300", stats.P50MS) + } + if stats.P95MS != 600 { + t.Errorf("P95MS = %d, want 600", stats.P95MS) + } + + // Empty window. + stats, err = s.Durations(context.Background(), time.Date(2030, 1, 1, 0, 0, 0, 0, time.UTC)) + if err != nil { + t.Fatalf("Durations empty: %v", err) + } + if stats.Count != 0 || stats.AvgMS != 0 || stats.P50MS != 0 || stats.P95MS != 0 { + t.Errorf("empty Durations = %+v, want zeros", stats) + } +} + +func TestTotals(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + + tot, err := s.Totals(ctx, now) + if err != nil { + t.Fatalf("Totals empty: %v", err) + } + if tot != (Totals{}) { + t.Errorf("empty Totals = %+v, want zeros", tot) + } + + for _, spec := range []struct { + id, domain, ip, status string + ago time.Duration + }{ + {"t1", "a.com", "203.0.113.1", StatusComplete, time.Hour}, + {"t2", "a.com", "203.0.113.2", StatusError, 2 * time.Hour}, + {"t3", "b.net", "203.0.113.1", StatusComplete, 48 * time.Hour}, + {"t4", "c.org", "203.0.113.3", StatusComplete, 10 * 24 * time.Hour}, + } { + ev := completeEvent(spec.id) + ev.Domain = spec.domain + ev.ClientIP = spec.ip + ev.Status = spec.status + ev.StartedAt = now.Add(-spec.ago) + ev.DoneAt = ev.StartedAt.Add(time.Second) + if err := s.RecordComplete(ctx, ev); err != nil { + t.Fatalf("seed %s: %v", spec.id, err) + } + } + + tot, err = s.Totals(ctx, now) + if err != nil { + t.Fatalf("Totals: %v", err) + } + want := Totals{AllTime: 4, Last24h: 2, Last7d: 3, DistinctDomains: 3, DistinctClients: 3, Errors: 1} + if tot != want { + t.Errorf("Totals = %+v, want %+v", tot, want) + } +} + +func TestOpenSQLiteCreatesParentDirs(t *testing.T) { + path := filepath.Join(t.TempDir(), "nested", "dir", "receiver.db") + s, err := OpenSQLite(path) + if err != nil { + t.Fatalf("OpenSQLite: %v", err) + } + defer s.Close() + if s.Backend() != "sqlite" { + t.Errorf("Backend = %q, want sqlite", s.Backend()) + } + + // Reopening must not fail on the existing schema and must keep the + // stamped version. + s.Close() + s2, err := OpenSQLite(path) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer s2.Close() + var v int + if err := s2.db.QueryRow(`SELECT version FROM schema_version`).Scan(&v); err != nil { + t.Fatalf("schema_version: %v", err) + } + if v != schemaVersion { + t.Errorf("schema version = %d, want %d", v, schemaVersion) + } +} + +func TestRedactMySQLDSN(t *testing.T) { + got := RedactMySQLDSN("user:s3cret@tcp(db.example.com:3306)/exploredns") + if got != "user:xxxxx@tcp(db.example.com:3306)/exploredns" { + t.Errorf("RedactMySQLDSN = %q", got) + } + if got := RedactMySQLDSN("::::"); got != "(unparsable DSN)" { + t.Errorf("RedactMySQLDSN(bad) = %q", got) + } +} + +// TestMySQLDialectSQL pins the MySQL-side SQL text, which unit tests cannot +// execute without a server. +func TestMySQLDialectSQL(t *testing.T) { + d := mysqlDialect{} + got := d.upsert("INSERT INTO traversals (id, domain) VALUES (?, ?)", []string{"domain", "last_seen"}) + want := "INSERT INTO traversals (id, domain) VALUES (?, ?)" + + " ON DUPLICATE KEY UPDATE domain=VALUES(domain), last_seen=VALUES(last_seen)" + if got != want { + t.Errorf("upsert = %q, want %q", got, want) + } + pred, arg := d.likeContains("domain", `50%_o\ff`) + if pred != `domain LIKE ? ESCAPE '\\'` { + t.Errorf("likeContains predicate = %q", pred) + } + if arg != `%50\%\_o\\ff%` { + t.Errorf("likeContains arg = %q", arg) + } + if d.datetimeType() != "DATETIME(6)" { + t.Errorf("datetimeType = %q", d.datetimeType()) + } +} diff --git a/web/api/handler.go b/web/api/handler.go index 07f4b87..1fd4ea5 100644 --- a/web/api/handler.go +++ b/web/api/handler.go @@ -276,7 +276,7 @@ func newHandler(ctx context.Context) *Handler { maxRunning: envInt("EXPLOREDNS_MAX_JOBS", defaultMaxRunningJobs), version: "dev", limiter: newRateLimiter(limit, window), - webhook: newWebhookReporter(os.Getenv("EXPLOREDNS_WEBHOOK_URL")), + webhook: newWebhookReporter(os.Getenv("EXPLOREDNS_WEBHOOK_URL"), os.Getenv("EXPLOREDNS_WEBHOOK_TOKEN")), fp: fingerprint.New(), } diff --git a/web/api/webhook.go b/web/api/webhook.go index 3c8be5f..9a92887 100644 --- a/web/api/webhook.go +++ b/web/api/webhook.go @@ -52,19 +52,22 @@ type webhookCompleteEvent struct { // single retry, and failures are logged but never surface to callers. type webhookReporter struct { url string + token string client *http.Client timeout time.Duration retryDelay time.Duration } // newWebhookReporter returns a reporter for url, or nil when url is empty -// (webhook reporting disabled). A nil reporter is safe to call. -func newWebhookReporter(url string) *webhookReporter { +// (webhook reporting disabled). A nil reporter is safe to call. A non-empty +// token is sent as an Authorization bearer token on every delivery. +func newWebhookReporter(url, token string) *webhookReporter { if url == "" { return nil } return &webhookReporter{ url: url, + token: token, client: &http.Client{}, timeout: 5 * time.Second, retryDelay: 2 * time.Second, @@ -106,6 +109,9 @@ func (wr *webhookReporter) post(event string, body []byte) error { } req.Header.Set("Content-Type", "application/json") req.Header.Set("X-ExploreDNS-Event", event) + if wr.token != "" { + req.Header.Set("Authorization", "Bearer "+wr.token) + } resp, err := wr.client.Do(req) if err != nil { diff --git a/web/api/webhook_compat_test.go b/web/api/webhook_compat_test.go new file mode 100644 index 0000000..77f669f --- /dev/null +++ b/web/api/webhook_compat_test.go @@ -0,0 +1,126 @@ +package api + +import ( + "bytes" + "encoding/json" + "testing" + "time" + + rstore "gitea.hansenits.com.au/hits/ExploreDNS/internal/receiver/store" +) + +// These tests pin the webhook payload contract between this package (the +// sender) and internal/receiver/store (the receiver): every field a sender +// struct marshals must decode into the receiver struct and vice versa. +// DisallowUnknownFields turns any renamed or missing field into a failure. + +// decodeStrict unmarshals data into v, failing on unknown fields. +func decodeStrict(t *testing.T, data []byte, v any) { + t.Helper() + dec := json.NewDecoder(bytes.NewReader(data)) + dec.DisallowUnknownFields() + if err := dec.Decode(v); err != nil { + t.Fatalf("decode %s into %T: %v", data, v, err) + } +} + +func TestWebhookStartEventReceiverCompat(t *testing.T) { + sent := webhookStartEvent{ + Event: webhookEventStart, + ID: "job-1", + Domain: "example.com", + QueryType: "AAAA", + AllRoots: true, + ClientIP: "203.0.113.9", + StartedAt: time.Date(2026, 7, 1, 10, 0, 0, 123456789, time.UTC), + } + raw, err := json.Marshal(sent) + if err != nil { + t.Fatalf("marshal sender: %v", err) + } + + var got rstore.StartEvent + decodeStrict(t, raw, &got) + want := rstore.StartEvent{ + Event: sent.Event, + ID: sent.ID, + Domain: sent.Domain, + QueryType: sent.QueryType, + AllRoots: sent.AllRoots, + ClientIP: sent.ClientIP, + StartedAt: sent.StartedAt, + } + if got != want { + t.Errorf("receiver decoded %+v, want %+v", got, want) + } + + // Round-trip back into the sender struct so receiver-only fields + // would also fail. + back, err := json.Marshal(got) + if err != nil { + t.Fatalf("marshal receiver: %v", err) + } + var sent2 webhookStartEvent + decodeStrict(t, back, &sent2) + if sent2 != sent { + t.Errorf("sender round-trip %+v, want %+v", sent2, sent) + } +} + +func TestWebhookCompleteEventReceiverCompat(t *testing.T) { + sent := webhookCompleteEvent{ + Event: webhookEventComplete, + ID: "job-1", + Domain: "example.com", + QueryType: "A", + ClientIP: "203.0.113.9", + StartedAt: time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC), + DoneAt: time.Date(2026, 7, 1, 10, 0, 3, 500000000, time.UTC), + DurationMS: 3500, + Status: statusError, + Error: "traversal timed out after 5m0s", + ResultCount: 7, + Summary: &Summary{ + Answers: []SummaryAnswer{{Probability: 0.75, Records: []string{"example.com 300 IN A 192.0.2.1"}}}, + ByStatus: []SummaryStatus{{Status: "servfail", Probability: 0.25}}, + }, + } + raw, err := json.Marshal(sent) + if err != nil { + t.Fatalf("marshal sender: %v", err) + } + + var got rstore.CompleteEvent + decodeStrict(t, raw, &got) + if got.Event != sent.Event || got.ID != sent.ID || got.Domain != sent.Domain || + got.QueryType != sent.QueryType || got.ClientIP != sent.ClientIP || + !got.StartedAt.Equal(sent.StartedAt) || !got.DoneAt.Equal(sent.DoneAt) || + got.DurationMS != sent.DurationMS || got.Status != sent.Status || + got.Error != sent.Error || got.ResultCount != sent.ResultCount { + t.Errorf("receiver decoded %+v, want %+v", got, sent) + } + + // The receiver keeps Summary as raw JSON; it must match the sender's + // marshalled Summary byte for byte. + wantSummary, err := json.Marshal(sent.Summary) + if err != nil { + t.Fatalf("marshal summary: %v", err) + } + if !bytes.Equal(got.Summary, wantSummary) { + t.Errorf("receiver Summary = %s, want %s", got.Summary, wantSummary) + } + + back, err := json.Marshal(got) + if err != nil { + t.Fatalf("marshal receiver: %v", err) + } + var sent2 webhookCompleteEvent + decodeStrict(t, back, &sent2) + back2, err := json.Marshal(sent2) + if err != nil { + t.Fatalf("re-marshal sender: %v", err) + } + if !bytes.Equal(back2, raw) { + t.Errorf("sender round-trip = %s, want %s", back2, raw) + } +} diff --git a/web/api/webhook_test.go b/web/api/webhook_test.go index a01998d..19b3376 100644 --- a/web/api/webhook_test.go +++ b/web/api/webhook_test.go @@ -235,7 +235,37 @@ func TestWebhook_RetriesOnceOnFailure(t *testing.T) { func TestWebhook_NilReporterSafe(t *testing.T) { var wr *webhookReporter wr.send("start", map[string]string{"event": "start"}) // must not panic - if newWebhookReporter("") != nil { + if newWebhookReporter("", "token") != nil { t.Fatal("empty URL should disable the webhook reporter") } } + +// TestWebhook_BearerToken verifies the Authorization header is sent exactly +// when a token is configured (EXPLOREDNS_WEBHOOK_TOKEN on the real path). +func TestWebhook_BearerToken(t *testing.T) { + for _, tc := range []struct { + name, token, wantAuth string + }{ + {"token set", "s3cret", "Bearer s3cret"}, + {"token unset", "", ""}, + } { + t.Run(tc.name, func(t *testing.T) { + done := make(chan string, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + done <- r.Header.Get("Authorization") + })) + defer ts.Close() + + newWebhookReporter(ts.URL, tc.token).send("start", map[string]string{"event": "start"}) + + select { + case got := <-done: + if got != tc.wantAuth { + t.Fatalf("Authorization = %q, want %q", got, tc.wantAuth) + } + case <-time.After(5 * time.Second): + t.Fatal("webhook was not delivered") + } + }) + } +} -- 2.54.0 From beb595442fc5d4377a0b267721a3ceb6ff268679 Mon Sep 17 00:00:00 2001 From: Gary Hansen Date: Wed, 8 Jul 2026 02:43:51 +1000 Subject: [PATCH 2/3] feat(receiver): ingest server and admin dashboard POST /webhook (bearer-token auth, strict validation, 1MB cap) and /healthz stay open; everything under /admin requires basic auth (RECEIVER_ADMIN_USER/PASSWORD, refuses to start without a password, constant-time compares). Admin JSON APIs for the traversal log (filters, pagination) and stats, plus an embedded dashboard: stat cards, four Chart.js charts (lazy CDN load with graceful degradation), filterable log with expandable summaries, auto-refresh. Fetches resolve against location.origin so credentialed bookmark URLs work. Co-Authored-By: Claude Fable 5 --- cmd/exploredns-receiver/main.go | 86 ++++ internal/receiver/server/admin.go | 280 ++++++++++++ internal/receiver/server/admin_test.go | 482 ++++++++++++++++++++ internal/receiver/server/server.go | 222 ++++++++++ internal/receiver/server/server_test.go | 209 +++++++++ internal/receiver/server/static/admin.html | 486 +++++++++++++++++++++ 6 files changed, 1765 insertions(+) create mode 100644 cmd/exploredns-receiver/main.go create mode 100644 internal/receiver/server/admin.go create mode 100644 internal/receiver/server/admin_test.go create mode 100644 internal/receiver/server/server.go create mode 100644 internal/receiver/server/server_test.go create mode 100644 internal/receiver/server/static/admin.html diff --git a/cmd/exploredns-receiver/main.go b/cmd/exploredns-receiver/main.go new file mode 100644 index 0000000..ff5fca9 --- /dev/null +++ b/cmd/exploredns-receiver/main.go @@ -0,0 +1,86 @@ +// Command exploredns-receiver stores usage webhooks posted by the +// ExploreDNS API server in MySQL or SQLite. +// +// Configuration (environment): +// +// RECEIVER_ADDR listen address (default ":8080") +// RECEIVER_MYSQL_DSN go-sql-driver DSN; when set, events go to MySQL +// RECEIVER_SQLITE_PATH SQLite fallback path (default "data/exploredns-receiver.db") +// RECEIVER_INGEST_TOKEN bearer token required on POST /webhook (open when unset) +// RECEIVER_ADMIN_USER basic-auth username for /admin (default "admin") +// RECEIVER_ADMIN_PASSWORD basic-auth password for /admin (required) +package main + +import ( + "fmt" + "log" + "os" + "os/signal" + "syscall" + "time" + + "gitea.hansenits.com.au/hits/ExploreDNS/internal/receiver/server" + "gitea.hansenits.com.au/hits/ExploreDNS/internal/receiver/store" +) + +// version is stamped at build time via -ldflags "-X main.version=...". +var version = "dev" + +func main() { + addr := envOr("RECEIVER_ADDR", ":8080") + + adminPass := os.Getenv("RECEIVER_ADMIN_PASSWORD") + if adminPass == "" { + fmt.Fprintln(os.Stderr, "Error: RECEIVER_ADMIN_PASSWORD is not set; refusing to start with an unprotected admin interface") + os.Exit(1) + } + + var ( + st *store.Store + backend string + err error + ) + if dsn := os.Getenv("RECEIVER_MYSQL_DSN"); dsn != "" { + st, err = store.OpenMySQL(dsn) + backend = "mysql " + store.RedactMySQLDSN(dsn) + } else { + path := envOr("RECEIVER_SQLITE_PATH", "data/exploredns-receiver.db") + st, err = store.OpenSQLite(path) + backend = "sqlite " + path + } + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + + srv := server.New(addr, st) + srv.SetVersion(version) + srv.SetIngestToken(os.Getenv("RECEIVER_INGEST_TOKEN")) + srv.SetAdminAuth(envOr("RECEIVER_ADMIN_USER", "admin"), adminPass) + if err := srv.Start(); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + + log.Printf("ExploreDNS receiver %s listening on %s, storing to %s", version, srv.Addr(), backend) + + quit := make(chan os.Signal, 1) + signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + <-quit + + log.Println("Shutting down...") + if err := srv.Shutdown(15 * time.Second); err != nil { + log.Printf("Shutdown error: %v", err) + } + if err := st.Close(); err != nil { + log.Printf("Close store error: %v", err) + } +} + +// envOr returns the environment variable name, or def when unset or empty. +func envOr(name, def string) string { + if v := os.Getenv(name); v != "" { + return v + } + return def +} diff --git a/internal/receiver/server/admin.go b/internal/receiver/server/admin.go new file mode 100644 index 0000000..4306aed --- /dev/null +++ b/internal/receiver/server/admin.go @@ -0,0 +1,280 @@ +package server + +import ( + "crypto/sha256" + "crypto/subtle" + _ "embed" + "encoding/json" + "log" + "net/http" + "strconv" + "time" + + "gitea.hansenits.com.au/hits/ExploreDNS/internal/receiver/store" +) + +//go:embed static/admin.html +var adminHTML []byte + +const ( + maxListLimit = 200 + defaultStatsDays = 30 + maxStatsDays = 365 + topLimit = 10 +) + +// requireAdmin gates next behind HTTP basic auth against the configured +// admin credentials. An empty configured password locks the subtree +// entirely rather than opening it. +func (h *handler) requireAdmin(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + user, pass, ok := r.BasicAuth() + // Evaluate both comparisons unconditionally to keep timing uniform. + userOK := secretEqual(user, h.adminUser) + passOK := secretEqual(pass, h.adminPass) + if !ok || !userOK || !passOK || h.adminPass == "" { + w.Header().Set("WWW-Authenticate", `Basic realm="ExploreDNS receiver admin", charset="UTF-8"`) + writeError(w, http.StatusUnauthorized, "authentication required") + return + } + next.ServeHTTP(w, r) + }) +} + +// secretEqual compares two strings in constant time; both sides are hashed +// first so length differences do not leak. +func secretEqual(got, want string) bool { + g := sha256.Sum256([]byte(got)) + w := sha256.Sum256([]byte(want)) + return subtle.ConstantTimeCompare(g[:], w[:]) == 1 +} + +// adminPage handles GET /admin, serving the embedded admin UI. +func (h *handler) adminPage(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Write(adminHTML) //nolint:errcheck +} + +// traversalItem is the admin API JSON shape of one stored traversal. +// Completion fields are omitted while a traversal is still running. +type traversalItem struct { + ID string `json:"id"` + Domain string `json:"domain"` + QueryType string `json:"query_type"` + AllRoots bool `json:"all_roots"` + ClientIP string `json:"client_ip"` + StartedAt string `json:"started_at,omitempty"` + DoneAt string `json:"done_at,omitempty"` + DurationMS *int64 `json:"duration_ms,omitempty"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + ResultCount *int `json:"result_count,omitempty"` + Summary json.RawMessage `json:"summary,omitempty"` + ReceivedAt string `json:"received_at"` +} + +func toItem(tr store.Traversal) traversalItem { + it := traversalItem{ + ID: tr.ID, Domain: tr.Domain, QueryType: tr.QueryType, AllRoots: tr.AllRoots, + ClientIP: tr.ClientIP, StartedAt: rfc3339(tr.StartedAt), Status: tr.Status, + Error: tr.Error, DurationMS: tr.DurationMS, ResultCount: tr.ResultCount, + ReceivedAt: rfc3339(tr.FirstSeen), + } + if tr.DoneAt != nil { + it.DoneAt = rfc3339(*tr.DoneAt) + } + if tr.Summary != "" { + it.Summary = json.RawMessage(tr.Summary) + } + return it +} + +// rfc3339 renders t as UTC RFC 3339, or "" for the zero time. +func rfc3339(t time.Time) string { + if t.IsZero() { + return "" + } + return t.UTC().Format(time.RFC3339) +} + +// adminTraversals handles GET /admin/api/traversals. +func (h *handler) adminTraversals(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + + limit := 50 + if v := q.Get("limit"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n < 1 { + writeError(w, http.StatusBadRequest, "invalid limit "+strconv.Quote(v)) + return + } + limit = min(n, maxListLimit) + } + offset := 0 + if v := q.Get("offset"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n < 0 { + writeError(w, http.StatusBadRequest, "invalid offset "+strconv.Quote(v)) + return + } + offset = n + } + f := store.ListFilter{Domain: q.Get("domain"), Status: q.Get("status")} + if v := q.Get("from"); v != "" { + t, err := time.Parse(time.RFC3339, v) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid from: "+err.Error()) + return + } + f.From = t + } + if v := q.Get("to"); v != "" { + t, err := time.Parse(time.RFC3339, v) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid to: "+err.Error()) + return + } + f.To = t + } + + rows, total, err := h.st.ListTraversals(r.Context(), f, limit, offset) + if err != nil { + log.Printf("receiver admin: list traversals: %v", err) + writeError(w, http.StatusInternalServerError, "list traversals failed") + return + } + items := make([]traversalItem, 0, len(rows)) + for _, tr := range rows { + items = append(items, toItem(tr)) + } + writeJSON(w, http.StatusOK, map[string]any{"total": total, "items": items}) +} + +type statsResponse struct { + Days int `json:"days"` + Totals statsTotals `json:"totals"` + PerDay []dayJSON `json:"per_day"` + TopDomains []nameCountJSON `json:"top_domains"` + QueryTypes []nameCountJSON `json:"query_types"` + Statuses []nameCountJSON `json:"statuses"` + Durations durationsJSON `json:"durations"` + TopClients []nameCountJSON `json:"top_clients"` +} + +type statsTotals struct { + AllTime int `json:"all_time"` + Last24h int `json:"last_24h"` + Last7d int `json:"last_7d"` + DistinctDomains int `json:"distinct_domains"` + DistinctClients int `json:"distinct_clients"` + ErrorRate float64 `json:"error_rate"` +} + +type dayJSON struct { + Day string `json:"day"` + Total int `json:"total"` + Errors int `json:"errors"` +} + +type nameCountJSON struct { + Name string `json:"name"` + Count int `json:"count"` +} + +type durationsJSON struct { + Count int `json:"count"` + AvgMS float64 `json:"avg_ms"` + P50MS int64 `json:"p50_ms"` + P95MS int64 `json:"p95_ms"` +} + +func nameCounts(rows []store.NameCount) []nameCountJSON { + out := make([]nameCountJSON, 0, len(rows)) + for _, r := range rows { + out = append(out, nameCountJSON{Name: r.Name, Count: r.Count}) + } + return out +} + +// adminStats handles GET /admin/api/stats. The totals block uses fixed +// windows; everything else covers the last ?days calendar days (UTC), +// matching the StatsPerDay window. +func (h *handler) adminStats(w http.ResponseWriter, r *http.Request) { + days := defaultStatsDays + if v := r.URL.Query().Get("days"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n < 1 { + writeError(w, http.StatusBadRequest, "invalid days "+strconv.Quote(v)) + return + } + days = min(n, maxStatsDays) + } + + ctx := r.Context() + now := time.Now().UTC() + since := now.Truncate(24*time.Hour).AddDate(0, 0, -(days - 1)) + + fail := func(what string, err error) { + log.Printf("receiver admin: %s: %v", what, err) + writeError(w, http.StatusInternalServerError, what+" failed") + } + totals, err := h.st.Totals(ctx, now) + if err != nil { + fail("totals", err) + return + } + perDay, err := h.st.StatsPerDay(ctx, days) + if err != nil { + fail("per-day stats", err) + return + } + topDomains, err := h.st.TopDomains(ctx, since, topLimit) + if err != nil { + fail("top domains", err) + return + } + queryTypes, err := h.st.QueryTypeCounts(ctx, since) + if err != nil { + fail("query types", err) + return + } + statuses, err := h.st.StatusCounts(ctx, since) + if err != nil { + fail("statuses", err) + return + } + dur, err := h.st.Durations(ctx, since) + if err != nil { + fail("durations", err) + return + } + topClients, err := h.st.TopClientIPs(ctx, since, topLimit) + if err != nil { + fail("top clients", err) + return + } + + resp := statsResponse{ + Days: days, + Totals: statsTotals{ + AllTime: totals.AllTime, + Last24h: totals.Last24h, + Last7d: totals.Last7d, + DistinctDomains: totals.DistinctDomains, + DistinctClients: totals.DistinctClients, + }, + PerDay: make([]dayJSON, 0, len(perDay)), + TopDomains: nameCounts(topDomains), + QueryTypes: nameCounts(queryTypes), + Statuses: nameCounts(statuses), + Durations: durationsJSON{Count: dur.Count, AvgMS: dur.AvgMS, P50MS: dur.P50MS, P95MS: dur.P95MS}, + TopClients: nameCounts(topClients), + } + if totals.AllTime > 0 { + resp.Totals.ErrorRate = float64(totals.Errors) / float64(totals.AllTime) + } + for _, d := range perDay { + resp.PerDay = append(resp.PerDay, dayJSON{Day: d.Day, Total: d.Total, Errors: d.Errors}) + } + writeJSON(w, http.StatusOK, resp) +} diff --git a/internal/receiver/server/admin_test.go b/internal/receiver/server/admin_test.go new file mode 100644 index 0000000..86e3beb --- /dev/null +++ b/internal/receiver/server/admin_test.go @@ -0,0 +1,482 @@ +package server + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.hansenits.com.au/hits/ExploreDNS/internal/receiver/store" +) + +const ( + testAdminUser = "admin" + testAdminPass = "swordfish" +) + +// get performs a GET with optional basic-auth credentials. +func get(h http.Handler, path, user, pass string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, path, nil) + if user != "" || pass != "" { + req.SetBasicAuth(user, pass) + } + w := httptest.NewRecorder() + h.ServeHTTP(w, req) + return w +} + +func TestStartRequiresAdminPassword(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "receiver.db")) + if err != nil { + t.Fatalf("OpenSQLite: %v", err) + } + defer st.Close() + + srv := New("127.0.0.1:0", st) + err = srv.Start() + if err == nil { + srv.Shutdown(time.Second) //nolint:errcheck + t.Fatal("Start succeeded without an admin password") + } + if !strings.Contains(err.Error(), "RECEIVER_ADMIN_PASSWORD") { + t.Errorf("error = %q, want mention of RECEIVER_ADMIN_PASSWORD", err) + } +} + +func TestAdminAuthMatrix(t *testing.T) { + paths := []string{"/admin", "/admin/api/traversals", "/admin/api/stats"} + creds := []struct { + name string + user, pass string + want int + }{ + {"no credentials", "", "", http.StatusUnauthorized}, + {"wrong user", "root", testAdminPass, http.StatusUnauthorized}, + {"wrong password", testAdminUser, "nope", http.StatusUnauthorized}, + {"correct credentials", testAdminUser, testAdminPass, http.StatusOK}, + } + h, _ := newTestHandler(t, "") + for _, path := range paths { + for _, tc := range creds { + t.Run(path+" "+tc.name, func(t *testing.T) { + w := get(h, path, tc.user, tc.pass) + if w.Code != tc.want { + t.Fatalf("status = %d, want %d (%s)", w.Code, tc.want, w.Body) + } + if tc.want == http.StatusUnauthorized { + if got := w.Header().Get("WWW-Authenticate"); !strings.Contains(got, `Basic realm=`) { + t.Errorf("WWW-Authenticate = %q, want Basic realm", got) + } + } + }) + } + } +} + +// TestAdminSubtreeAuthenticated pins that unknown and unauthenticated paths +// under /admin still 401 rather than falling through to a handler. +func TestAdminSubtreeAuthenticated(t *testing.T) { + h, _ := newTestHandler(t, "") + for _, path := range []string{"/admin/", "/admin/secret", "/admin/api/other"} { + if got := get(h, path, "", "").Code; got != http.StatusUnauthorized { + t.Errorf("GET %s without creds = %d, want 401", path, got) + } + } +} + +// TestAdminLockedWithoutPassword pins that a handler built without an admin +// password rejects everything, even blank credentials. +func TestAdminLockedWithoutPassword(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "receiver.db")) + if err != nil { + t.Fatalf("OpenSQLite: %v", err) + } + defer st.Close() + h := newHandler(st, "test-version", "", "admin", "") + + req := httptest.NewRequest(http.MethodGet, "/admin", nil) + req.SetBasicAuth("admin", "") + w := httptest.NewRecorder() + h.ServeHTTP(w, req) + if w.Code != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", w.Code) + } +} + +// TestWebhookAndHealthzStayOpen asserts the sender and k8s probe endpoints +// need no admin credentials. +func TestWebhookAndHealthzStayOpen(t *testing.T) { + h, _ := newTestHandler(t, "") + if got := post(h, startJSON, nil).Code; got != http.StatusNoContent { + t.Errorf("POST /webhook without creds = %d, want 204", got) + } + if got := get(h, "/healthz", "", "").Code; got != http.StatusOK { + t.Errorf("GET /healthz without creds = %d, want 200", got) + } +} + +// seedList inserts the deterministic fixed-date dataset used by the +// traversals endpoint tests (same shape as the store package's seed). +func seedList(t *testing.T, st *store.Store) { + t.Helper() + ctx := context.Background() + base := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + rows := []struct { + id, domain, qtype, ip, status string + day int + durMS int64 + }{ + {"a1", "example.com", "A", "203.0.113.1", store.StatusComplete, 0, 100}, + {"a2", "example.com", "AAAA", "203.0.113.1", store.StatusComplete, 0, 200}, + {"a3", "sub.example.com", "A", "203.0.113.2", store.StatusError, 1, 300}, + {"a4", "other.net", "MX", "203.0.113.3", store.StatusComplete, 1, 400}, + {"a5", "other.net", "A", "203.0.113.1", store.StatusComplete, 2, 500}, + {"a6", "under_score.org", "A", "203.0.113.4", store.StatusComplete, 2, 600}, + } + for i, r := range rows { + started := base.AddDate(0, 0, r.day).Add(time.Duration(i) * time.Minute) + ev := store.CompleteEvent{ + ID: r.id, Domain: r.domain, QueryType: r.qtype, ClientIP: r.ip, + StartedAt: started, DoneAt: started.Add(time.Duration(r.durMS) * time.Millisecond), + DurationMS: r.durMS, Status: r.status, ResultCount: 1, + Summary: json.RawMessage(`{"answers":[{"probability":1}]}`), + } + if r.status == store.StatusError { + ev.Error = "lookup failed" + } + if err := st.RecordComplete(ctx, ev); err != nil { + t.Fatalf("seed %s: %v", r.id, err) + } + } + if err := st.RecordStart(ctx, store.StartEvent{ + ID: "a7", Domain: "running.io", QueryType: "A", ClientIP: "203.0.113.5", + StartedAt: base.AddDate(0, 0, 2).Add(time.Hour), + }); err != nil { + t.Fatalf("seed a7: %v", err) + } +} + +type listResponse struct { + Total int `json:"total"` + Items []struct { + ID string `json:"id"` + Domain string `json:"domain"` + QueryType string `json:"query_type"` + ClientIP string `json:"client_ip"` + StartedAt string `json:"started_at"` + DoneAt string `json:"done_at"` + DurationMS *int64 `json:"duration_ms"` + Status string `json:"status"` + Error string `json:"error"` + ResultCount *int `json:"result_count"` + Summary json.RawMessage `json:"summary"` + ReceivedAt string `json:"received_at"` + } `json:"items"` +} + +func listGET(t *testing.T, h http.Handler, query string) listResponse { + t.Helper() + w := get(h, "/admin/api/traversals"+query, testAdminUser, testAdminPass) + if w.Code != http.StatusOK { + t.Fatalf("GET %s = %d (%s)", query, w.Code, w.Body) + } + var resp listResponse + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode %s: %v", query, err) + } + return resp +} + +func TestAdminTraversalsEndpoint(t *testing.T) { + h, st := newTestHandler(t, "") + seedList(t, st) + + // Default page: everything, newest first. + resp := listGET(t, h, "") + if resp.Total != 7 || len(resp.Items) != 7 { + t.Fatalf("total = %d, len = %d, want 7 and 7", resp.Total, len(resp.Items)) + } + first := resp.Items[0] + if first.ID != "a7" || first.Status != store.StatusRunning { + t.Errorf("first item = %s/%s, want a7/running", first.ID, first.Status) + } + if first.DurationMS != nil || first.DoneAt != "" || len(first.Summary) != 0 { + t.Errorf("running item has completion fields: %+v", first) + } + if first.StartedAt != "2026-07-03T01:00:00Z" { + t.Errorf("started_at = %q, want 2026-07-03T01:00:00Z", first.StartedAt) + } + if first.ReceivedAt == "" { + t.Error("received_at empty") + } + + // Completed rows carry duration, result count, and raw summary JSON. + last := resp.Items[6] + if last.ID != "a1" || last.DurationMS == nil || *last.DurationMS != 100 || + last.ResultCount == nil || *last.ResultCount != 1 { + t.Errorf("oldest item = %+v, want a1 with duration 100 and 1 result", last) + } + if string(last.Summary) != `{"answers":[{"probability":1}]}` { + t.Errorf("summary = %s", last.Summary) + } + + // Pagination. + resp = listGET(t, h, "?limit=3&offset=3") + if resp.Total != 7 || len(resp.Items) != 3 { + t.Fatalf("page 2 total = %d, len = %d, want 7 and 3", resp.Total, len(resp.Items)) + } + if resp.Items[0].ID != "a4" || resp.Items[1].ID != "a3" || resp.Items[2].ID != "a2" { + t.Errorf("page 2 order = %s,%s,%s, want a4,a3,a2", + resp.Items[0].ID, resp.Items[1].ID, resp.Items[2].ID) + } + + // Filters. + if resp := listGET(t, h, "?domain=example"); resp.Total != 3 { + t.Errorf("domain filter total = %d, want 3", resp.Total) + } + resp = listGET(t, h, "?status=error") + if resp.Total != 1 || resp.Items[0].ID != "a3" || resp.Items[0].Error != "lookup failed" { + t.Errorf("status filter = %+v, want a3 with error", resp) + } + resp = listGET(t, h, "?from=2026-07-02T00:00:00Z&to=2026-07-02T23:59:59Z") + if resp.Total != 2 { + t.Errorf("time window total = %d, want 2", resp.Total) + } + + // Limit above the cap succeeds and is clamped rather than rejected. + if resp := listGET(t, h, "?limit=1000"); resp.Total != 7 { + t.Errorf("clamped limit total = %d, want 7", resp.Total) + } +} + +func TestAdminTraversalsBadParams(t *testing.T) { + h, _ := newTestHandler(t, "") + for _, q := range []string{"?limit=abc", "?limit=0", "?offset=-1", "?from=notatime", "?to=2026-13-99"} { + if got := get(h, "/admin/api/traversals"+q, testAdminUser, testAdminPass).Code; got != http.StatusBadRequest { + t.Errorf("GET %s = %d, want 400", q, got) + } + } + if got := get(h, "/admin/api/stats?days=x", testAdminUser, testAdminPass).Code; got != http.StatusBadRequest { + t.Errorf("GET stats?days=x = %d, want 400", got) + } +} + +type statsBody struct { + Days int `json:"days"` + Totals struct { + AllTime int `json:"all_time"` + Last24h int `json:"last_24h"` + Last7d int `json:"last_7d"` + DistinctDomains int `json:"distinct_domains"` + DistinctClients int `json:"distinct_clients"` + ErrorRate float64 `json:"error_rate"` + } `json:"totals"` + PerDay []struct { + Day string `json:"day"` + Total int `json:"total"` + Errors int `json:"errors"` + } `json:"per_day"` + TopDomains []struct { + Name string `json:"name"` + Count int `json:"count"` + } `json:"top_domains"` + QueryTypes []struct { + Name string `json:"name"` + Count int `json:"count"` + } `json:"query_types"` + Statuses []struct { + Name string `json:"name"` + Count int `json:"count"` + } `json:"statuses"` + Durations struct { + Count int `json:"count"` + AvgMS float64 `json:"avg_ms"` + P50MS int64 `json:"p50_ms"` + P95MS int64 `json:"p95_ms"` + } `json:"durations"` + TopClients []struct { + Name string `json:"name"` + Count int `json:"count"` + } `json:"top_clients"` +} + +func statsGET(t *testing.T, h http.Handler, query string) statsBody { + t.Helper() + w := get(h, "/admin/api/stats"+query, testAdminUser, testAdminPass) + if w.Code != http.StatusOK { + t.Fatalf("GET stats%s = %d (%s)", query, w.Code, w.Body) + } + var resp statsBody + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode stats: %v", err) + } + return resp +} + +func TestAdminStatsEndpoint(t *testing.T) { + h, st := newTestHandler(t, "") + ctx := context.Background() + now := time.Now().UTC() + + complete := func(id, domain, qtype, ip, status string, ago time.Duration, durMS int64) { + t.Helper() + ev := store.CompleteEvent{ + ID: id, Domain: domain, QueryType: qtype, ClientIP: ip, + StartedAt: now.Add(-ago), DoneAt: now.Add(-ago).Add(time.Duration(durMS) * time.Millisecond), + DurationMS: durMS, Status: status, ResultCount: 1, + } + if status == store.StatusError { + ev.Error = "boom" + } + if err := st.RecordComplete(ctx, ev); err != nil { + t.Fatalf("seed %s: %v", id, err) + } + } + complete("r1", "a.com", "A", "203.0.113.1", store.StatusComplete, time.Hour, 100) + complete("r2", "a.com", "A", "203.0.113.2", store.StatusError, 2*time.Hour, 200) + complete("r3", "b.net", "A", "203.0.113.1", store.StatusComplete, 30*time.Hour, 300) + complete("r4", "c.org", "MX", "203.0.113.3", store.StatusComplete, 8*24*time.Hour, 400) + if err := st.RecordStart(ctx, store.StartEvent{ + ID: "r5", Domain: "a.com", QueryType: "A", ClientIP: "203.0.113.1", + StartedAt: now.Add(-time.Hour), + }); err != nil { + t.Fatalf("seed r5: %v", err) + } + + s := statsGET(t, h, "?days=30") + if s.Days != 30 { + t.Errorf("days = %d, want 30", s.Days) + } + tot := s.Totals + if tot.AllTime != 5 || tot.Last24h != 3 || tot.Last7d != 4 || + tot.DistinctDomains != 3 || tot.DistinctClients != 3 { + t.Errorf("totals = %+v, want all 5, 24h 3, 7d 4, domains 3, clients 3", tot) + } + if tot.ErrorRate != 0.2 { + t.Errorf("error_rate = %v, want 0.2", tot.ErrorRate) + } + + var perDayTotal, perDayErrors int + for _, d := range s.PerDay { + perDayTotal += d.Total + perDayErrors += d.Errors + } + if perDayTotal != 5 || perDayErrors != 1 { + t.Errorf("per_day sums = %d/%d, want 5/1 (%+v)", perDayTotal, perDayErrors, s.PerDay) + } + + if len(s.TopDomains) != 3 || s.TopDomains[0].Name != "a.com" || s.TopDomains[0].Count != 3 { + t.Errorf("top_domains = %+v, want a.com x3 leading 3", s.TopDomains) + } + if len(s.QueryTypes) != 2 || s.QueryTypes[0].Name != "A" || s.QueryTypes[0].Count != 4 || + s.QueryTypes[1].Name != "MX" || s.QueryTypes[1].Count != 1 { + t.Errorf("query_types = %+v, want A x4, MX x1", s.QueryTypes) + } + if len(s.Statuses) != 3 || s.Statuses[0].Name != store.StatusComplete || s.Statuses[0].Count != 3 { + t.Errorf("statuses = %+v, want complete x3 leading", s.Statuses) + } + d := s.Durations + if d.Count != 4 || d.AvgMS != 250 || d.P50MS != 200 || d.P95MS != 400 { + t.Errorf("durations = %+v, want count 4 avg 250 p50 200 p95 400", d) + } + if len(s.TopClients) != 3 || s.TopClients[0].Name != "203.0.113.1" || s.TopClients[0].Count != 3 { + t.Errorf("top_clients = %+v, want 203.0.113.1 x3 leading 3", s.TopClients) + } + + // A 7-day window drops r4 from the windowed aggregates but not from the + // fixed totals. + s = statsGET(t, h, "?days=7") + if s.Totals.AllTime != 5 { + t.Errorf("7d all_time = %d, want 5", s.Totals.AllTime) + } + if s.Durations.Count != 3 || s.Durations.AvgMS != 200 { + t.Errorf("7d durations = %+v, want count 3 avg 200", s.Durations) + } + if len(s.QueryTypes) != 1 || s.QueryTypes[0].Count != 4 { + t.Errorf("7d query_types = %+v, want A x4 only", s.QueryTypes) + } + + // days above the cap clamps to 365. + if s := statsGET(t, h, "?days=9999"); s.Days != 365 { + t.Errorf("days = %d, want 365", s.Days) + } + // default is 30. + if s := statsGET(t, h, ""); s.Days != 30 { + t.Errorf("default days = %d, want 30", s.Days) + } +} + +func TestAdminStatsEmptyStore(t *testing.T) { + h, _ := newTestHandler(t, "") + s := statsGET(t, h, "") + if s.Totals.AllTime != 0 || s.Totals.ErrorRate != 0 { + t.Errorf("empty totals = %+v, want zeros", s.Totals) + } + // Arrays must be present (possibly empty), never null. + w := get(h, "/admin/api/stats", testAdminUser, testAdminPass) + body := w.Body.String() + for _, key := range []string{`"per_day":[]`, `"top_domains":[]`, `"query_types":[]`, `"statuses":[]`, `"top_clients":[]`} { + if !strings.Contains(body, key) { + t.Errorf("stats body missing %s: %s", key, body) + } + } +} + +// TestStaticAdmin_Markup asserts the admin page ships the stat cards, +// charts, log table, and controls the API is built for. +func TestStaticAdmin_Markup(t *testing.T) { + h, _ := newTestHandler(t, "") + w := get(h, "/admin", testAdminUser, testAdminPass) + if w.Code != http.StatusOK { + t.Fatalf("GET /admin = %d", w.Code) + } + if ct := w.Header().Get("Content-Type"); !strings.HasPrefix(ct, "text/html") { + t.Errorf("Content-Type = %q, want text/html", ct) + } + page := w.Body.String() + for _, want := range []string{ + `id="statAllTime"`, `id="statLast24h"`, `id="statLast7d"`, + `id="statDomains"`, `id="statClients"`, `id="statErrorRate"`, + `id="perDayChart"`, `id="statusChart"`, `id="domainsChart"`, `id="typesChart"`, + `cdn.jsdelivr.net/npm/chart.js@4`, // Chart.js 4, lazy-loaded + `id="logTable"`, `id="logBody"`, + `ReceivedDomainTypeStatusDurationClient IPResults`, + `id="filterDomain"`, `id="filterStatus"`, `id="filterFrom"`, `id="filterTo"`, `id="applyFilters"`, + `id="prevPage"`, `id="nextPage"`, `id="pageInfo"`, + `id="autoRefresh"`, `id="daysSelect"`, + `/admin/api/stats`, `/admin/api/traversals`, + } { + if !strings.Contains(page, want) { + t.Errorf("admin.html missing %q", want) + } + } +} + +// TestStaticAdmin_ScriptSyntax runs node --check over the inline admin +// script so syntax errors fail in CI, not in the browser. +func TestStaticAdmin_ScriptSyntax(t *testing.T) { + nodeBin, err := exec.LookPath("node") + if err != nil { + t.Skip("node not installed") + } + page := string(adminHTML) + start := strings.Index(page, "") + if start < 0 || end < 0 || end < start { + t.Fatal("admin.html has no inline script") + } + src := page[start+len(" + + -- 2.54.0 From c9a963ecdd58d2fa99fbb91613c2e8f43ca8eaf7 Mon Sep 17 00:00:00 2001 From: Gary Hansen Date: Wed, 8 Jul 2026 02:43:51 +1000 Subject: [PATCH 3/3] chore(receiver): packaging, k8s manifests, CI/release, docs Dockerfile.receiver (CGO-free, /data volume), receiver image in CI and tag releases, receiver binary in release archives, make build-receiver, example k8s manifests (deployment/service/ingress/secret/pvc) under deploy/k8s/receiver/, and README coverage including sender/receiver token pairing. Co-Authored-By: Claude Fable 5 --- .gitea/workflows/ci.yml | 10 +++ .gitea/workflows/release.yml | 23 +++++-- Dockerfile.cli | 2 +- Dockerfile.receiver | 26 ++++++++ Dockerfile.web | 2 +- Makefile | 8 ++- README.md | 98 ++++++++++++++++++++++++++--- deploy/k8s/receiver/deployment.yaml | 57 +++++++++++++++++ deploy/k8s/receiver/ingress.yaml | 33 ++++++++++ deploy/k8s/receiver/namespace.yaml | 6 ++ deploy/k8s/receiver/pvc.yaml | 16 +++++ deploy/k8s/receiver/secret.yaml | 25 ++++++++ deploy/k8s/receiver/service.yaml | 16 +++++ 13 files changed, 306 insertions(+), 16 deletions(-) create mode 100644 Dockerfile.receiver create mode 100644 deploy/k8s/receiver/deployment.yaml create mode 100644 deploy/k8s/receiver/ingress.yaml create mode 100644 deploy/k8s/receiver/namespace.yaml create mode 100644 deploy/k8s/receiver/pvc.yaml create mode 100644 deploy/k8s/receiver/secret.yaml create mode 100644 deploy/k8s/receiver/service.yaml diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 07edd46..676f979 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -67,3 +67,13 @@ jobs: tags: | gitea.hansenits.com.au/hits/exploredns-web:latest gitea.hansenits.com.au/hits/exploredns-web:${{ github.sha }} + + - name: Build and push receiver image + uses: docker/build-push-action@v6 + with: + context: . + file: Dockerfile.receiver + push: true + tags: | + gitea.hansenits.com.au/hits/exploredns-receiver:latest + gitea.hansenits.com.au/hits/exploredns-receiver:${{ github.sha }} diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml index c322e0a..406da0b 100644 --- a/.gitea/workflows/release.yml +++ b/.gitea/workflows/release.yml @@ -1,8 +1,8 @@ name: Release # Builds release artifacts when a version tag (v*) is pushed: -# - cross-compiled CLI + server binaries attached to the Gitea release -# - version-tagged docker images for the CLI and web server +# - cross-compiled CLI + server + receiver binaries attached to the Gitea release +# - version-tagged docker images for the CLI, web server, and receiver # The owner usually creates the Gitea release by hand with notes; this # workflow attaches assets to it (creating a bare release only when none # exists) and skips assets that are already attached, so re-runs are safe. @@ -42,10 +42,13 @@ jobs: CGO_ENABLED=0 GOOS="$GOOS" GOARCH="$GOARCH" \ go build -trimpath -ldflags "-s -w -X main.version=${TAG}" \ -o "${OUT}/exploredns-server${EXT}" ./cmd/server + CGO_ENABLED=0 GOOS="$GOOS" GOARCH="$GOARCH" \ + go build -trimpath -ldflags "-s -w -X main.version=${TAG}" \ + -o "${OUT}/exploredns-receiver${EXT}" ./cmd/exploredns-receiver if [ "$GOOS" = "windows" ]; then - (cd "$OUT" && zip -q "../exploredns_${TAG}_${GOOS}_${GOARCH}.zip" exploredns.exe exploredns-server.exe) + (cd "$OUT" && zip -q "../exploredns_${TAG}_${GOOS}_${GOARCH}.zip" exploredns.exe exploredns-server.exe exploredns-receiver.exe) else - tar -czf "dist/exploredns_${TAG}_${GOOS}_${GOARCH}.tar.gz" -C "$OUT" exploredns exploredns-server + tar -czf "dist/exploredns_${TAG}_${GOOS}_${GOARCH}.tar.gz" -C "$OUT" exploredns exploredns-server exploredns-receiver fi rm -rf "$OUT" done @@ -130,3 +133,15 @@ jobs: tags: | gitea.hansenits.com.au/hits/exploredns-web:latest gitea.hansenits.com.au/hits/exploredns-web:${{ github.ref_name }} + + - name: Build and push receiver image + uses: docker/build-push-action@v6 + with: + context: . + file: Dockerfile.receiver + push: true + build-args: | + VERSION=${{ github.ref_name }} + tags: | + gitea.hansenits.com.au/hits/exploredns-receiver:latest + gitea.hansenits.com.au/hits/exploredns-receiver:${{ github.ref_name }} diff --git a/Dockerfile.cli b/Dockerfile.cli index fd40a4c..109bc94 100644 --- a/Dockerfile.cli +++ b/Dockerfile.cli @@ -1,5 +1,5 @@ # Build stage -FROM golang:1.24-alpine AS builder +FROM golang:1.25-alpine AS builder WORKDIR /src diff --git a/Dockerfile.receiver b/Dockerfile.receiver new file mode 100644 index 0000000..1404efe --- /dev/null +++ b/Dockerfile.receiver @@ -0,0 +1,26 @@ +# Build stage +FROM golang:1.25-alpine AS builder + +WORKDIR /src + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +ARG VERSION=dev +RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w -X main.version=${VERSION}" -o /out/exploredns-receiver ./cmd/exploredns-receiver + +# Final stage +FROM alpine:3.21 + +RUN apk --no-cache add ca-certificates + +COPY --from=builder /out/exploredns-receiver /usr/local/bin/exploredns-receiver + +ENV RECEIVER_SQLITE_PATH=/data/exploredns-receiver.db +VOLUME /data + +EXPOSE 8080 + +ENTRYPOINT ["exploredns-receiver"] diff --git a/Dockerfile.web b/Dockerfile.web index 7636194..aac819b 100644 --- a/Dockerfile.web +++ b/Dockerfile.web @@ -1,5 +1,5 @@ # Build stage -FROM golang:1.24-alpine AS builder +FROM golang:1.25-alpine AS builder WORKDIR /src diff --git a/Makefile b/Makefile index d73dcb0..25e092d 100644 --- a/Makefile +++ b/Makefile @@ -1,12 +1,13 @@ BINARY_NAME=exploredns SERVER_BINARY_NAME=exploredns-server +RECEIVER_BINARY_NAME=exploredns-receiver BUILD_DIR=bin GO=go GOFLAGS=-v VERSION?=$(shell git describe --tags --always 2>/dev/null || echo dev) LDFLAGS=-ldflags "-X main.version=$(VERSION)" -.PHONY: build build-server build-all test lint clean cover deploy deploy-status +.PHONY: build build-server build-receiver build-all test lint clean cover deploy deploy-status build: $(GO) build $(GOFLAGS) $(LDFLAGS) -o $(BUILD_DIR)/$(BINARY_NAME) ./cmd/exploredns @@ -14,7 +15,10 @@ build: build-server: $(GO) build $(GOFLAGS) $(LDFLAGS) -o $(BUILD_DIR)/$(SERVER_BINARY_NAME) ./cmd/server -build-all: build build-server +build-receiver: + $(GO) build $(GOFLAGS) $(LDFLAGS) -o $(BUILD_DIR)/$(RECEIVER_BINARY_NAME) ./cmd/exploredns-receiver + +build-all: build build-server build-receiver test: $(GO) test -v -race -coverprofile=coverage.out ./... diff --git a/README.md b/README.md index 692dcad..209e60a 100644 --- a/README.md +++ b/README.md @@ -297,6 +297,7 @@ variables at startup: | `EXPLOREDNS_CORS_ORIGIN` | *(unset)* | Off by default (the SPA is same-origin). Set an origin — or `*` for development — to enable cross-origin API access. | | `EXPLOREDNS_RATE_LIMIT` | `30/1h` | Per-client-IP token-bucket limit on `POST /api/traverse` in `N/duration` form (e.g. `10/10m`); invalid values fall back to the default. Over-limit requests get `429`. Buckets refill continuously. Direct localhost connections are exempt (dev loop, tests), but proxied requests are always limited by the real client IP from `Fly-Client-IP` / `X-Forwarded-For`. | | `EXPLOREDNS_WEBHOOK_URL` | *(unset)* | Off by default. When set, the server POSTs a usage-reporting JSON event to this URL on every traversal start and completion (see below). | +| `EXPLOREDNS_WEBHOOK_TOKEN` | *(unset)* | Optional bearer token for webhook deliveries. When set, every webhook POST carries `Authorization: Bearer `; pair it with the receiver's `RECEIVER_INGEST_TOKEN`. | ### Usage reporting @@ -312,12 +313,17 @@ traversal, each with header `X-ExploreDNS-Event` naming the event: `X-Forwarded-For` entry, else the connection address). Delivery is fire-and-forget: a 5-second timeout, one retry after 2 seconds, and failures are logged without ever affecting the traversal or the API response. On -Fly.io, configure it as a secret rather than in `fly.toml`: +Fly.io, configure it (and the optional bearer token) as secrets rather than +in `fly.toml`: ```sh -fly secrets set EXPLOREDNS_WEBHOOK_URL=https://example.com/hook +fly secrets set EXPLOREDNS_WEBHOOK_URL=https://example.com/hook \ + EXPLOREDNS_WEBHOOK_TOKEN=some-long-random-string ``` +This repo ships a matching receiver for these events — see +[Usage telemetry receiver](#usage-telemetry-receiver). + --- ## Deploying to Fly.io @@ -366,9 +372,9 @@ flyctl tokens create deploy -x 999999h Pushing a `v*` tag triggers the full release pipeline: -1. `.gitea/workflows/release.yml` (`binaries` job) cross-compiles the CLI - and server for linux/amd64, linux/arm64, darwin/amd64, darwin/arm64 and - windows/amd64, packages them as +1. `.gitea/workflows/release.yml` (`binaries` job) cross-compiles the CLI, + server, and receiver for linux/amd64, linux/arm64, darwin/amd64, + darwin/arm64 and windows/amd64, packages them as `exploredns___.tar.gz` (`.zip` on Windows) plus a `SHA256SUMS` file, and attaches everything to the Gitea release for the tag. Create the release with notes by hand before (or after) pushing @@ -376,8 +382,8 @@ Pushing a `v*` tag triggers the full release pipeline: a bare one only when none exists, and skips already-attached assets so re-runs are safe. 2. `.gitea/workflows/release.yml` (`docker` job) pushes - `gitea.hansenits.com.au/hits/exploredns-cli` and `…/exploredns-web` - images tagged `` and `latest`. + `gitea.hansenits.com.au/hits/exploredns-cli`, `…/exploredns-web`, and + `…/exploredns-receiver` images tagged `` and `latest`. 3. `.gitea/workflows/deploy.yml` deploys the web server to Fly.io. All binaries are stamped with the tag via @@ -420,6 +426,78 @@ with `--show-servers`. Suitable for piping into `jq`. --- +## Usage telemetry receiver + +`cmd/exploredns-receiver` is a small companion service that receives the +usage webhooks described above (`start`/`complete` events from +`EXPLOREDNS_WEBHOOK_URL`), stores them in MySQL or SQLite, and serves a +basic-auth-protected admin dashboard (`/admin`) plus JSON API +(`/admin/api/traversals`, `/admin/api/stats`) over the collected data. It +is a separate binary intended to run wherever you keep long-lived storage +(e.g. a home Kubernetes cluster) while the public web server stays +stateless. + +Endpoints: `POST /webhook` (ingest, bearer-token protected when configured), +`GET /healthz` (liveness/readiness), `GET /admin` and `GET /admin/api/*` +(basic auth, always required). + +### Configuration + +| Variable | Default | Meaning | +|---|---|---| +| `RECEIVER_ADDR` | `:8080` | Listen address. | +| `RECEIVER_MYSQL_DSN` | *(unset)* | [go-sql-driver DSN](https://github.com/go-sql-driver/mysql#dsn-data-source-name) (`user:pass@tcp(host:3306)/dbname`). When set, events are stored in MySQL and the SQLite settings are ignored. | +| `RECEIVER_SQLITE_PATH` | `data/exploredns-receiver.db` | SQLite database path, used when no MySQL DSN is set (the container image defaults it to `/data/exploredns-receiver.db`). Parent directories are created automatically. | +| `RECEIVER_INGEST_TOKEN` | *(unset)* | When set, `POST /webhook` requires `Authorization: Bearer `. Leave unset only on trusted networks. | +| `RECEIVER_ADMIN_USER` | `admin` | Basic-auth username for `/admin`. | +| `RECEIVER_ADMIN_PASSWORD` | *(required)* | Basic-auth password for `/admin`; the receiver refuses to start without it. | + +Both storage backends share one portable schema; pure-Go drivers +(`modernc.org/sqlite`, `github.com/go-sql-driver/mysql`) keep the binary +CGO-free. SQLite is the zero-setup default; point `RECEIVER_MYSQL_DSN` at +an external MySQL when you want the data outside the pod/VM. + +### Running with Docker + +```sh +docker run -d --name exploredns-receiver \ + -p 8080:8080 \ + -v exploredns-receiver-data:/data \ + -e RECEIVER_ADMIN_PASSWORD=change-me \ + -e RECEIVER_INGEST_TOKEN=some-long-random-string \ + gitea.hansenits.com.au/hits/exploredns-receiver:latest +``` + +The image stores SQLite data under the `/data` volume; add +`-e RECEIVER_MYSQL_DSN=...` to use MySQL instead. + +### Running on Kubernetes + +[deploy/k8s/receiver/](deploy/k8s/receiver/) contains commented template +manifests: a single-replica deployment (SQLite on a 1Gi PVC mounted at +`/data`, probes on `/healthz`), ClusterIP service, ingress with TLS +placeholders, and a secret template for the `RECEIVER_*` variables. Edit +the placeholder host/credentials, then: + +```sh +kubectl apply -f deploy/k8s/receiver/ +``` + +Keep one replica while on SQLite; MySQL removes that constraint. + +### Pairing with the web server + +Set the same token on both ends so the receiver only accepts events from +your server — e.g. on Fly.io: + +```sh +fly secrets set EXPLOREDNS_WEBHOOK_URL=https://receiver.example.com/webhook \ + EXPLOREDNS_WEBHOOK_TOKEN=some-long-random-string +# receiver side: RECEIVER_INGEST_TOKEN=some-long-random-string +``` + +--- + ## Comparison with dnstraverse | Feature | dnstraverse (Ruby) | ExploreDNS (Go) | @@ -440,13 +518,16 @@ with `--show-servers`. Suitable for piping into `jq`. ``` cmd/exploredns/ CLI entry point and flag parsing cmd/server/ HTTP API server entry point +cmd/exploredns-receiver/ Usage telemetry receiver entry point internal/config/ Configuration types, validation, and usage text internal/dns/ DNS query layer, root discovery, transport internal/traverse/ Core traversal engine, referral resolution, caching internal/fingerprint/ DNS server version fingerprinting (version.bind CHAOS) internal/output/ Result formatting — text tree and JSON renderers internal/integration/ End-to-end integration tests +internal/receiver/ Telemetry receiver: HTTP server, admin UI, event store web/api/ HTTP handler, job store, SSE streaming, static assets +deploy/k8s/receiver/ Kubernetes manifest templates for the receiver ``` --- @@ -456,7 +537,8 @@ web/api/ HTTP handler, job store, SSE streaming, static assets ```sh make build # compile CLI binary to bin/exploredns make build-server # compile server binary to bin/exploredns-server -make build-all # compile both binaries +make build-receiver # compile telemetry receiver to bin/exploredns-receiver +make build-all # compile all three binaries make test # run all unit and integration tests make lint # run go vet make clean # remove build artefacts diff --git a/deploy/k8s/receiver/deployment.yaml b/deploy/k8s/receiver/deployment.yaml new file mode 100644 index 0000000..321d375 --- /dev/null +++ b/deploy/k8s/receiver/deployment.yaml @@ -0,0 +1,57 @@ +# TEMPLATE — single-replica receiver deployment. Keep replicas at 1 while +# using the SQLite backend: the database file on the RWO volume supports only +# one writer. With RECEIVER_MYSQL_DSN you may scale out and drop the volume. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: exploredns-receiver + namespace: exploredns-receiver + labels: + app: exploredns-receiver +spec: + replicas: 1 + strategy: + type: Recreate # RWO volume: never run old and new pods concurrently + selector: + matchLabels: + app: exploredns-receiver + template: + metadata: + labels: + app: exploredns-receiver + spec: + containers: + - name: receiver + image: gitea.hansenits.com.au/hits/exploredns-receiver:latest + ports: + - name: http + containerPort: 8080 + envFrom: + - secretRef: + name: exploredns-receiver + volumeMounts: + - name: data + mountPath: /data + livenessProbe: + httpGet: + path: /healthz + port: http + initialDelaySeconds: 5 + periodSeconds: 15 + readinessProbe: + httpGet: + path: /healthz + port: http + initialDelaySeconds: 2 + periodSeconds: 10 + resources: + requests: + cpu: 50m + memory: 64Mi + limits: + cpu: 500m + memory: 256Mi + volumes: + - name: data + persistentVolumeClaim: + claimName: exploredns-receiver-data diff --git a/deploy/k8s/receiver/ingress.yaml b/deploy/k8s/receiver/ingress.yaml new file mode 100644 index 0000000..c7d9c61 --- /dev/null +++ b/deploy/k8s/receiver/ingress.yaml @@ -0,0 +1,33 @@ +# TEMPLATE — replace receiver.example.com with your real host and wire up +# TLS for your cluster (the webhook bearer token and admin password travel +# in headers, so plain HTTP is not acceptable across the internet). +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: exploredns-receiver + namespace: exploredns-receiver + annotations: {} + # nginx ingress controller: + # cert-manager.io/cluster-issuer: letsencrypt + # nginx.ingress.kubernetes.io/proxy-body-size: 1m + # + # traefik: + # traefik.ingress.kubernetes.io/router.entrypoints: websecure + # traefik.ingress.kubernetes.io/router.tls: "true" +spec: + # ingressClassName: nginx + tls: + - hosts: + - receiver.example.com + secretName: exploredns-receiver-tls # created by cert-manager or by hand + rules: + - host: receiver.example.com + http: + paths: + - path: / + pathType: Prefix + backend: + service: + name: exploredns-receiver + port: + name: http diff --git a/deploy/k8s/receiver/namespace.yaml b/deploy/k8s/receiver/namespace.yaml new file mode 100644 index 0000000..8c8f219 --- /dev/null +++ b/deploy/k8s/receiver/namespace.yaml @@ -0,0 +1,6 @@ +# TEMPLATE — optional. Skip this file (and drop the namespace fields from the +# other manifests) to deploy into an existing namespace. +apiVersion: v1 +kind: Namespace +metadata: + name: exploredns-receiver diff --git a/deploy/k8s/receiver/pvc.yaml b/deploy/k8s/receiver/pvc.yaml new file mode 100644 index 0000000..bab2037 --- /dev/null +++ b/deploy/k8s/receiver/pvc.yaml @@ -0,0 +1,16 @@ +# TEMPLATE — backing storage for the SQLite database (RECEIVER_SQLITE_PATH +# defaults to /data/exploredns-receiver.db in the container image). Not needed +# when RECEIVER_MYSQL_DSN is set, but harmless to keep. Set storageClassName +# if your cluster has no default class. +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: exploredns-receiver-data + namespace: exploredns-receiver +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 1Gi + # storageClassName: standard diff --git a/deploy/k8s/receiver/secret.yaml b/deploy/k8s/receiver/secret.yaml new file mode 100644 index 0000000..5d6abce --- /dev/null +++ b/deploy/k8s/receiver/secret.yaml @@ -0,0 +1,25 @@ +# TEMPLATE — fill in real values before applying, or create the secret +# imperatively instead and never commit credentials: +# +# kubectl -n exploredns-receiver create secret generic exploredns-receiver \ +# --from-literal=RECEIVER_ADMIN_USER=admin \ +# --from-literal=RECEIVER_ADMIN_PASSWORD='change-me' \ +# --from-literal=RECEIVER_INGEST_TOKEN='change-me-too' +# +# The deployment loads every key here as an environment variable (envFrom). +apiVersion: v1 +kind: Secret +metadata: + name: exploredns-receiver + namespace: exploredns-receiver +type: Opaque +stringData: + RECEIVER_ADMIN_USER: admin + RECEIVER_ADMIN_PASSWORD: change-me + # Bearer token the main app must send on POST /webhook. Must match the + # sender's EXPLOREDNS_WEBHOOK_TOKEN. Leave unset to accept unauthenticated + # posts (not recommended for an internet-facing receiver). + RECEIVER_INGEST_TOKEN: change-me-too + # Uncomment to store events in an external MySQL instead of the SQLite + # file on the PVC (go-sql-driver DSN). + # RECEIVER_MYSQL_DSN: "user:pass@tcp(mysql.example.com:3306)/exploredns" diff --git a/deploy/k8s/receiver/service.yaml b/deploy/k8s/receiver/service.yaml new file mode 100644 index 0000000..34be9c1 --- /dev/null +++ b/deploy/k8s/receiver/service.yaml @@ -0,0 +1,16 @@ +# TEMPLATE — cluster-internal service in front of the receiver pod. +apiVersion: v1 +kind: Service +metadata: + name: exploredns-receiver + namespace: exploredns-receiver + labels: + app: exploredns-receiver +spec: + type: ClusterIP + selector: + app: exploredns-receiver + ports: + - name: http + port: 8080 + targetPort: http -- 2.54.0