diff --git a/cmd/felis/api.go b/cmd/felis/api.go index 6a42d1d..0305a7c 100644 --- a/cmd/felis/api.go +++ b/cmd/felis/api.go @@ -30,6 +30,7 @@ import ( "felis.lolicon.best/internal/reaper" "felis.lolicon.best/internal/registryprune" "felis.lolicon.best/internal/restore" + "felis.lolicon.best/internal/retention" "felis.lolicon.best/internal/submit" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" @@ -84,6 +85,14 @@ func cmdAPI(args []string, stdout, stderr io.Writer) int { return 1 } + // Load already refused a malformed [audit] retention. + auditRetention, _ := cfg.Audit.RetentionPeriod() + if auditRetention == 0 { + fmt.Fprintln(stdout, "felis api: audit rows are kept forever ([audit] retention = \"forever\")") + } else { + fmt.Fprintf(stdout, "felis api: audit rows older than %d days are deleted ([audit] retention; export them first with felis db audit-export)\n", int(auditRetention/(24*time.Hour))) + } + ctx := ctrl.SetupSignalHandler() // Before anything serves: an api on a schema it was not built for answers with @@ -421,6 +430,7 @@ func cmdAPI(args []string, stdout, stderr io.Writer) int { go pruner.Loop(ctx, registryPruneInterval) } go reapRejectedContexts(ctx, submissions, stderr) + go retention.Loop(ctx, drv.DB(), retention.Policy{Audit: auditRetention}, retentionInterval, slog.Default()) servers := []*http.Server{internalSrv, externalSrv} if httpsSrv != nil { @@ -698,6 +708,11 @@ func reapRejectedContexts(ctx context.Context, m *submit.Manager, stderr io.Writ } } +// retentionInterval spaces the runs that delete spent sign-in rows and audit rows +// past [audit] retention. The rows are spent for weeks before they go, so a few +// runs a day keep the tables flat. +const retentionInterval = 6 * time.Hour + // registryPruneInterval spaces the registry pruner's runs. The registry-gc // sidecar sweeps once a day, so pruning more often only changes which sweep frees // a layer. diff --git a/cmd/felis/db.go b/cmd/felis/db.go index 8050221..9627c61 100644 --- a/cmd/felis/db.go +++ b/cmd/felis/db.go @@ -15,6 +15,7 @@ import ( "felis.lolicon.best/internal/config" "felis.lolicon.best/internal/dbbackup" + "felis.lolicon.best/internal/retention" ) const dbUsage = `usage: @@ -24,6 +25,7 @@ const dbUsage = `usage: felis db verify [-dir dir] felis db list [-dir dir] felis db check [-dir dir] [-max-age 26h] + felis db audit-export [-config path] [-since date] [-until date] [-out file] ` // defaultKeep is how many bundles of a label a backup leaves behind. Manual @@ -58,6 +60,8 @@ func cmdDB(args []string, stdout, stderr io.Writer) int { return dbList(fs, dir, rest, stdout, stderr) case "check": return dbCheck(fs, dir, rest, stdout, stderr) + case "audit-export": + return dbAuditExport(fs, rest, stdout, stderr) case "-h", "--help", "help": fmt.Fprint(stdout, dbUsage) return 0 @@ -262,6 +266,88 @@ func dbList(fs *flag.FlagSet, dir *string, args []string, stdout, stderr io.Writ return 0 } +// dbAuditExport writes audit rows to a file (or stdout) as JSON lines, so an +// install can keep them past [audit] retention, after which felis-api deletes them. +func dbAuditExport(fs *flag.FlagSet, args []string, stdout, stderr io.Writer) int { + cfgPath := fs.String("config", "/etc/felis/felis.toml", "path to felis.toml") + sinceFlag := fs.String("since", "", "first day (or RFC 3339 instant) to export, inclusive; empty starts at the oldest row") + untilFlag := fs.String("until", "", "day (or RFC 3339 instant) to stop before, exclusive; empty runs to the newest row") + out := fs.String("out", "", "file to write (created 0600, never overwritten); empty writes to stdout") + if err := fs.Parse(args); err != nil { + return 2 + } + if fs.NArg() > 0 { + fmt.Fprint(stderr, dbUsage) + return 2 + } + since, err := parseExportBound(*sinceFlag) + if err != nil { + fmt.Fprintf(stderr, "felis db audit-export: -since: %v\n", err) + return 2 + } + until, err := parseExportBound(*untilFlag) + if err != nil { + fmt.Fprintf(stderr, "felis db audit-export: -until: %v\n", err) + return 2 + } + if !since.IsZero() && !until.IsZero() && !until.After(since) { + fmt.Fprintf(stderr, "felis db audit-export: -until %s is not after -since %s\n", *untilFlag, *sinceFlag) + return 2 + } + url, err := dbDatabaseURL(*cfgPath) + if err != nil { + fmt.Fprintf(stderr, "felis db audit-export: %v\n", err) + return 1 + } + w := stdout + var f *os.File + if *out != "" { + if f, err = os.OpenFile(*out, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600); err != nil { + fmt.Fprintf(stderr, "felis db audit-export: %v\n", err) + return 1 + } + w = f + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + defer cancel() + n, err := exportAudit(ctx, url, since, until, w) + if f != nil { + if cerr := f.Close(); err == nil { + err = cerr + } + } + if err != nil { + fmt.Fprintf(stderr, "felis db audit-export: %v (%d rows written)\n", err, n) + return 1 + } + fmt.Fprintf(stderr, "felis db audit-export: %d audit rows written\n", n) + return 0 +} + +func exportAudit(ctx context.Context, url string, since, until time.Time, w io.Writer) (int, error) { + drv, err := openStore(ctx, url, false) + if err != nil { + return 0, fmt.Errorf("open database: %w", err) + } + defer drv.Close() + return retention.ExportAudit(ctx, drv.DB(), since, until, w) +} + +// parseExportBound reads a -since/-until value: a day (midnight UTC) or an +// RFC 3339 instant; empty is an open bound. +func parseExportBound(v string) (time.Time, error) { + if v == "" { + return time.Time{}, nil + } + if t, err := time.Parse(time.DateOnly, v); err == nil { + return t, nil + } + if t, err := time.Parse(time.RFC3339, v); err == nil { + return t.UTC(), nil + } + return time.Time{}, fmt.Errorf("%q is neither a day (2026-01-31) nor an RFC 3339 instant (2026-01-31T12:00:00Z)", v) +} + func humanBytes(n int64) string { const unit = 1024 if n < unit { diff --git a/cmd/felis/db_test.go b/cmd/felis/db_test.go index 7831143..0fa40d1 100644 --- a/cmd/felis/db_test.go +++ b/cmd/felis/db_test.go @@ -8,6 +8,7 @@ import ( "io" "strings" "testing" + "time" "felis.lolicon.best/internal/store" ) @@ -149,3 +150,36 @@ func TestPreMigrateBackupOnlyGuardsAPopulatedDatabase(t *testing.T) { } } } + +// audit-export takes a day or an RFC 3339 instant for each bound, and refuses a +// malformed or inverted window before it opens the config or the database. +func TestAuditExportBounds(t *testing.T) { + for _, tc := range []struct{ in, want string }{ + {"", "0001-01-01T00:00:00Z"}, + {"2026-01-31", "2026-01-31T00:00:00Z"}, + {"2026-01-31T12:30:00+08:00", "2026-01-31T04:30:00Z"}, + } { + got, err := parseExportBound(tc.in) + if err != nil || got.Format(time.RFC3339) != tc.want { + t.Errorf("parseExportBound(%q) = %v, %v; want %s", tc.in, got, err, tc.want) + } + } + if _, err := parseExportBound("31/01/2026"); err == nil || err.Error() != `"31/01/2026" is neither a day (2026-01-31) nor an RFC 3339 instant (2026-01-31T12:00:00Z)` { + t.Errorf("parseExportBound(31/01/2026) err = %v", err) + } + for _, tc := range []struct { + args []string + wantErr string + }{ + {[]string{"db", "audit-export", "-since", "yesterday"}, `felis db audit-export: -since: "yesterday" is neither`}, + {[]string{"db", "audit-export", "-until", "2026-13-01"}, `felis db audit-export: -until: "2026-13-01" is neither`}, + {[]string{"db", "audit-export", "-since", "2026-02-01", "-until", "2026-02-01"}, "felis db audit-export: -until 2026-02-01 is not after -since 2026-02-01"}, + {[]string{"db", "audit-export", "extra"}, "felis db audit-export [-config path]"}, + } { + var out, errBuf bytes.Buffer + code := run(append(tc.args, "-config", "/nonexistent/felis.toml"), &out, &errBuf) + if code != 2 || !strings.Contains(errBuf.String(), tc.wantErr) { + t.Errorf("%v: exit %d, stderr %q; want 2 and %q", tc.args, code, errBuf.String(), tc.wantErr) + } + } +} diff --git a/cmd/felis/reaper.go b/cmd/felis/reaper.go index f0196ea..302adcd 100644 --- a/cmd/felis/reaper.go +++ b/cmd/felis/reaper.go @@ -317,27 +317,9 @@ func resolveWorldDir(ctx context.Context, cl client.Client, namespace, worldsRoo } } -// parseSpanDuration parses the human spans used in felis.toml's [archive] table: -// "3mo" (months≈30d), "15d" (days), or any time.ParseDuration unit ("12h"). -func parseSpanDuration(s string) (time.Duration, error) { - s = strings.TrimSpace(s) - switch { - case strings.HasSuffix(s, "mo"): - n, err := strconv.Atoi(strings.TrimSuffix(s, "mo")) - if err != nil { - return 0, err - } - return time.Duration(n) * 30 * 24 * time.Hour, nil - case strings.HasSuffix(s, "d"): - n, err := strconv.Atoi(strings.TrimSuffix(s, "d")) - if err != nil { - return 0, err - } - return time.Duration(n) * 24 * time.Hour, nil - default: - return time.ParseDuration(s) - } -} +// parseSpanDuration parses the human spans used in felis.toml's [archive] table +// (config.ParseSpan). +func parseSpanDuration(s string) (time.Duration, error) { return config.ParseSpan(s) } // parseByteSize parses a Kubernetes-style quantity ("200Gi", "10G") into bytes. // An empty string means unlimited (0). diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index a8794e3..deddb21 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -2031,6 +2031,33 @@ A failed audit write does not fail the action; it logs `audit: lost ...` in `felis-api` and counts in `felis_audit_write_failures_total` (`FelisAuditWriteFailing`). The cause is almost always PostgreSQL (§16). +### How long rows are kept + +`felis-api` prunes the database a minute after it starts and every 6 hours +after that, and logs `retention: pruned spent rows` with a count per table: + +| Rows | Deleted | +|---|---| +| sessions | 30 days after they expired or were signed out | +| email codes, passkey challenges, setup links, op-login requests | 30 days after they expired or were used | +| `/felis link` bind codes | 30 days after they expired | +| account migrations that never completed | 30 days after their last step (completed ones stay) | +| wrong-code windows (`otp_failure_windows`) | 30 days after they began | +| `audit_logs` | once older than `[audit] retention` | + +`[audit] retention` defaults to `365d`; it takes days (`90d`), months of 30 +days (`18mo`) or `forever`, and refuses anything under `30d`. The daily +`felis db backup` bundles (§16) still hold the rows for as long as the bundles +are kept. To keep audit rows past the retention, export them before they go: + +```sh +sudo felis db audit-export -until 2026-01-01 -out /root/audit-2025.jsonl +``` + +`-since` and `-until` take a day (UTC midnight) or an RFC 3339 instant; the +window is `[since, until)`. Each line is one row as JSON, oldest first. The +file is created `0600` and an existing file is never overwritten. + ### Optional: a Cloudflare rate limiting rule in front The limits above live in the API, so they hold on any edge. Behind Cloudflare @@ -2080,3 +2107,4 @@ for 10 seconds (the Free plan's limits). | Right code refused; `otp_account_locked` / `FelisOTPAccountLocked` | §17 | | `FelisSignInFailures` / who is guessing, from where | §17 | | `FelisAuditWriteFailing` | §17 | +| How long sessions, codes and audit rows are kept; export audit rows | §17 | diff --git a/internal/config/config.go b/internal/config/config.go index 5a574ef..b0c4828 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -9,7 +9,9 @@ import ( "net" "net/url" "regexp" + "strconv" "strings" + "time" "github.com/BurntSushi/toml" ) @@ -25,6 +27,7 @@ type Config struct { Archive ArchiveConfig `toml:"archive"` Offsite OffsiteConfig `toml:"offsite"` SMTP SMTPConfig `toml:"smtp"` + Audit AuditConfig `toml:"audit"` // AuthSources is the [[auth_source]] array-of-tables: the third-party Yggdrasil // roots the Felis-nano hasJoined multiplexer federates over, in priority order // (config order = priority, so array-of-tables not a map — a map would lose order @@ -256,6 +259,63 @@ type ArchiveS3Config struct { SecretKeyRef string `toml:"secret_key_ref"` } +// AuditConfig is the [audit] table. Retention is how long felis-api keeps audit +// rows before deleting them ("365d", "18mo", or "forever" to keep every row); +// empty means DefaultAuditRetention. Export what must outlive it with +// `felis db audit-export` first. +type AuditConfig struct { + Retention string `toml:"retention"` +} + +// DefaultAuditRetention keeps a year of audit rows; MinAuditRetention is the +// shortest an install may set, since the manual-backup cooldown and an incident +// investigation both read recent rows. +const ( + DefaultAuditRetention = 365 * 24 * time.Hour + MinAuditRetention = 30 * 24 * time.Hour +) + +// RetentionPeriod resolves Retention: 0 keeps every row. +func (a AuditConfig) RetentionPeriod() (time.Duration, error) { + switch v := strings.TrimSpace(a.Retention); v { + case "": + return DefaultAuditRetention, nil + case "forever": + return 0, nil + default: + d, err := ParseSpan(v) + if err != nil { + return 0, fmt.Errorf("config: [audit] retention %q must be a span such as 365d or 18mo, or forever", a.Retention) + } + if d < MinAuditRetention { + return 0, fmt.Errorf("config: [audit] retention %q is shorter than the 30d minimum", a.Retention) + } + return d, nil + } +} + +// ParseSpan parses the human spans felis.toml uses for retention periods: +// "3mo" (months of 30 days), "15d" (days), or any time.ParseDuration unit ("12h"). +func ParseSpan(s string) (time.Duration, error) { + s = strings.TrimSpace(s) + switch { + case strings.HasSuffix(s, "mo"): + n, err := strconv.Atoi(strings.TrimSuffix(s, "mo")) + if err != nil { + return 0, err + } + return time.Duration(n) * 30 * 24 * time.Hour, nil + case strings.HasSuffix(s, "d"): + n, err := strconv.Atoi(strings.TrimSuffix(s, "d")) + if err != nil { + return 0, err + } + return time.Duration(n) * 24 * time.Hour, nil + default: + return time.ParseDuration(s) + } +} + // OffsiteConfig is the [offsite] table: the S3-compatible bucket, away from // this machine, that `felis offsite sync` (felis-offsite.timer on the host) // copies every world archive and the newest database bundles into, encrypted @@ -488,6 +548,9 @@ func (c *Config) Validate() error { if err := c.Offsite.validate(); err != nil { return err } + if _, err := c.Audit.RetentionPeriod(); err != nil { + return err + } if h := c.Auth.ClientIPHeader; strings.ContainsAny(h, " :\t\r\n") { return fmt.Errorf("config: [auth] client_ip_header %q must be a bare header name such as CF-Connecting-IP or X-Forwarded-For", h) } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index a7c66e1..a35f0fb 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -5,6 +5,7 @@ import ( "path/filepath" "strings" "testing" + "time" "felis.lolicon.best/internal/config" ) @@ -695,3 +696,50 @@ prefix = "site-a" } } } + +// TestAuditRetention: [audit] retention resolves to a year when unset, to 0 for +// "forever", and to the named span otherwise; a span under 30 days or one that +// does not parse fails at load. +func TestAuditRetention(t *testing.T) { + const head = ` +[server] +root_domain = "mc.example.net" +[database] +url = "postgres://felis@db/felis" +` + const day = 24 * time.Hour + for _, tc := range []struct { + name, table string + want time.Duration + }{ + {"unset", "", 365 * day}, + {"empty", "[audit]\nretention = \"\"\n", 365 * day}, + {"forever", "[audit]\nretention = \"forever\"\n", 0}, + {"days", "[audit]\nretention = \"90d\"\n", 90 * day}, + {"months", "[audit]\nretention = \"18mo\"\n", 540 * day}, + {"hours", "[audit]\nretention = \"720h\"\n", 30 * day}, + {"exactly the minimum", "[audit]\nretention = \"30d\"\n", 30 * day}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg, err := config.Load(writeTOML(t, head+tc.table)) + if err != nil { + t.Fatalf("Load: %v", err) + } + got, err := cfg.Audit.RetentionPeriod() + if err != nil || got != tc.want { + t.Fatalf("RetentionPeriod = %v, %v; want %v", got, err, tc.want) + } + }) + } + for _, tc := range []struct{ value, wantErr string }{ + {"29d", `config: [audit] retention "29d" is shorter than the 30d minimum`}, + {"719h", `config: [audit] retention "719h" is shorter than the 30d minimum`}, + {"a year", `config: [audit] retention "a year" must be a span such as 365d or 18mo, or forever`}, + {"Forever", `config: [audit] retention "Forever" must be a span such as 365d or 18mo, or forever`}, + } { + _, err := config.Load(writeTOML(t, head+"[audit]\nretention = \""+tc.value+"\"\n")) + if err == nil || err.Error() != tc.wantErr { + t.Errorf("retention %q: err = %v, want %q", tc.value, err, tc.wantErr) + } + } +} diff --git a/internal/pgint/retention_test.go b/internal/pgint/retention_test.go new file mode 100644 index 0000000..06795c4 --- /dev/null +++ b/internal/pgint/retention_test.go @@ -0,0 +1,215 @@ +//go:build pgint + +package pgint + +import ( + "bytes" + "context" + "regexp" + "testing" + "time" + + "felis.lolicon.best/internal/retention" +) + +func mustExec(t *testing.T, q string, args ...any) { + t.Helper() + if _, err := db.Exec(q, args...); err != nil { + t.Fatalf("%s: %v", q, err) + } +} + +func rowExists(t *testing.T, q string, args ...any) bool { + t.Helper() + var ok bool + if err := db.QueryRow(`SELECT EXISTS (`+q+`)`, args...).Scan(&ok); err != nil { + t.Fatalf("%s: %v", q, err) + } + return ok +} + +// Prune deletes each sign-in row a month after it was spent (expired, used or +// revoked) and keeps the ones spent more recently and the live ones; audit rows +// go once they are older than the policy's retention. +func TestPruneDropsRowsSpentLongerThanGrace(t *testing.T) { + ctx := context.Background() + now := mustNow() + old := now.Add(-31 * 24 * time.Hour) // spent past the 30-day grace + recent := now.Add(-29 * 24 * time.Hour) // spent inside it + future := now.Add(time.Hour) + u := newUser(t, "user", "prune") + src1, src2, src3 := newUser(t, "user", "mig1"), newUser(t, "user", "mig2"), newUser(t, "user", "mig3") + tag := suffix(t) + id := func(s string) string { return s + "-" + tag } + + for _, r := range []struct { + name string + expires, used, rev any + }{ + {"s-expired-old", old, nil, nil}, + {"s-revoked-old", future, nil, old}, + {"s-expired-recent", recent, nil, nil}, + {"s-revoked-recent", future, nil, recent}, + {"s-live", future, nil, nil}, + } { + mustExec(t, `INSERT INTO sessions (token_hash, user_id, expires_at, revoked_at) VALUES ($1, $2, $3, $4)`, + id(r.name), u.ID, r.expires, r.rev) + } + for _, r := range []struct { + name string + expires, used any + }{ + {"expired-old", old, nil}, + {"used-old", future, old}, + {"expired-recent", recent, nil}, + {"used-recent", future, recent}, + {"live", future, nil}, + } { + mustExec(t, `INSERT INTO email_otps (id, user_id, email, code_hash, purpose, expires_at, consumed_at) + VALUES ($1, $2, 'p@example.net', 'h', 'login_email', $3, $4)`, id("otp-"+r.name), u.ID, r.expires, r.used) + mustExec(t, `INSERT INTO webauthn_challenges (id, user_id, purpose, session_data, expires_at, consumed_at) + VALUES ($1, $2, 'passkey_register', '\x00', $3, $4)`, id("wc-"+r.name), u.ID, r.expires, r.used) + mustExec(t, `INSERT INTO webauthn_discoverable_challenges (id, session_data, expires_at, consumed_at) + VALUES ($1, '\x00', $2, $3)`, id("wd-"+r.name), r.expires, r.used) + mustExec(t, `INSERT INTO setup_tokens (token_hash, user_id, expires_at, consumed_at) VALUES ($1, $2, $3, $4)`, + id("st-"+r.name), u.ID, r.expires, r.used) + mustExec(t, `INSERT INTO op_login_requests (id, user_id, email, expires_at, consumed_at) + VALUES ($1, $2, 'staff@example.net', $3, $4)`, id("op-"+r.name), u.ID, r.expires, r.used) + } + for _, r := range []struct { + name string + expires time.Time + }{{"lc-old", old}, {"lc-recent", recent}, {"lc-live", future}} { + mustExec(t, `INSERT INTO account_link_codes (code, mc_uuid, expires_at) VALUES ($1, $2, $3)`, + id(r.name), testUUID(t), r.expires) + } + // A migration abandoned a month ago goes; one touched recently stays, and a + // redeemed one is the record of the move and stays however old it is. + mustExec(t, `INSERT INTO account_migrations (id, source_user_id, state, created_at, updated_at) VALUES ($1, $2, 'initiated', $3, $3)`, + id("mig-stale"), src1.ID, old) + mustExec(t, `INSERT INTO account_migrations (id, source_user_id, state, created_at, updated_at) VALUES ($1, $2, 'confirmed', $3, $4)`, + id("mig-fresh"), src2.ID, old, recent) + mustExec(t, `INSERT INTO account_migrations (id, source_user_id, state, redeemed_at, created_at, updated_at) VALUES ($1, $2, 'redeemed', $3, $3, $3)`, + id("mig-redeemed"), src3.ID, old) + mustExec(t, `INSERT INTO otp_failure_windows (user_id, purpose, window_start, failures) VALUES ($1, 'login_email', $2, 3)`, u.ID, old) + mustExec(t, `INSERT INTO otp_failure_windows (user_id, purpose, window_start, failures) VALUES ($1, 'onboard_email', $2, 3)`, u.ID, recent) + mustExec(t, `INSERT INTO audit_logs (actor, source, action, request_id, created_at) VALUES ('pgint', 'external', 'prune.test', $1, $2)`, + id("audit-61d"), now.Add(-61*24*time.Hour)) + mustExec(t, `INSERT INTO audit_logs (actor, source, action, request_id, created_at) VALUES ('pgint', 'external', 'prune.test', $1, $2)`, + id("audit-59d"), now.Add(-59*24*time.Hour)) + + if _, err := retention.Prune(ctx, db, now, retention.Policy{Audit: 60 * 24 * time.Hour}); err != nil { + t.Fatalf("Prune: %v", err) + } + + for _, c := range []struct { + q, key string + want bool + }{ + {`SELECT 1 FROM sessions WHERE token_hash = $1`, "s-expired-old", false}, + {`SELECT 1 FROM sessions WHERE token_hash = $1`, "s-revoked-old", false}, + {`SELECT 1 FROM sessions WHERE token_hash = $1`, "s-expired-recent", true}, + {`SELECT 1 FROM sessions WHERE token_hash = $1`, "s-revoked-recent", true}, + {`SELECT 1 FROM sessions WHERE token_hash = $1`, "s-live", true}, + {`SELECT 1 FROM account_link_codes WHERE code = $1`, "lc-old", false}, + {`SELECT 1 FROM account_link_codes WHERE code = $1`, "lc-recent", true}, + {`SELECT 1 FROM account_link_codes WHERE code = $1`, "lc-live", true}, + {`SELECT 1 FROM account_migrations WHERE id = $1`, "mig-stale", false}, + {`SELECT 1 FROM account_migrations WHERE id = $1`, "mig-fresh", true}, + {`SELECT 1 FROM account_migrations WHERE id = $1`, "mig-redeemed", true}, + {`SELECT 1 FROM audit_logs WHERE request_id = $1`, "audit-61d", false}, + {`SELECT 1 FROM audit_logs WHERE request_id = $1`, "audit-59d", true}, + } { + if got := rowExists(t, c.q, id(c.key)); got != c.want { + t.Errorf("%s: present = %v, want %v", c.key, got, c.want) + } + } + for _, table := range []struct{ name, q, prefix string }{ + {"email_otps", `SELECT 1 FROM email_otps WHERE id = $1`, "otp-"}, + {"webauthn_challenges", `SELECT 1 FROM webauthn_challenges WHERE id = $1`, "wc-"}, + {"webauthn_discoverable_challenges", `SELECT 1 FROM webauthn_discoverable_challenges WHERE id = $1`, "wd-"}, + {"setup_tokens", `SELECT 1 FROM setup_tokens WHERE token_hash = $1`, "st-"}, + {"op_login_requests", `SELECT 1 FROM op_login_requests WHERE id = $1`, "op-"}, + } { + for _, c := range []struct { + name string + want bool + }{{"expired-old", false}, {"used-old", false}, {"expired-recent", true}, {"used-recent", true}, {"live", true}} { + if got := rowExists(t, table.q, id(table.prefix+c.name)); got != c.want { + t.Errorf("%s %s: present = %v, want %v", table.name, c.name, got, c.want) + } + } + } + if rowExists(t, `SELECT 1 FROM otp_failure_windows WHERE user_id = $1 AND purpose = 'login_email'`, u.ID) { + t.Error("an OTP failure window that began a month ago survived") + } + if !rowExists(t, `SELECT 1 FROM otp_failure_windows WHERE user_id = $1 AND purpose = 'onboard_email'`, u.ID) { + t.Error("an OTP failure window inside the grace was deleted") + } +} + +// With no audit retention every audit row stays, and a table with more spent rows +// than one DELETE takes is emptied of them in the same run. +func TestPruneKeepsAuditForeverAndDrainsLargeBacklogs(t *testing.T) { + ctx := context.Background() + now := mustNow() + tag := suffix(t) + mustExec(t, `INSERT INTO audit_logs (actor, source, action, request_id, created_at) VALUES ('pgint', 'external', 'prune.test', $1, $2)`, + "ancient-"+tag, now.Add(-3000*24*time.Hour)) + mustExec(t, `INSERT INTO account_link_codes (code, mc_uuid, expires_at) + SELECT $1 || n, gen_random_uuid(), $2 FROM generate_series(1, 5003) AS n`, "bulk-"+tag+"-", now.Add(-40*24*time.Hour)) + + res, err := retention.Prune(ctx, db, now, retention.Policy{}) + if err != nil { + t.Fatalf("Prune: %v", err) + } + if !rowExists(t, `SELECT 1 FROM audit_logs WHERE request_id = $1`, "ancient-"+tag) { + t.Error("an audit row was deleted with retention off") + } + if _, ok := res["audit_logs"]; ok { + t.Errorf("audit_logs was pruned with retention off: %v", res) + } + var left int + if err := db.QueryRow(`SELECT count(*) FROM account_link_codes WHERE code LIKE $1`, "bulk-"+tag+"-%").Scan(&left); err != nil { + t.Fatal(err) + } + if left != 0 { + t.Errorf("%d of 5003 spent bind codes left after one run", left) + } + if res["account_link_codes"] < 5003 { + t.Errorf("account_link_codes deleted = %d, want at least 5003", res["account_link_codes"]) + } +} + +var idRE = regexp.MustCompile(`"id":\d+`) + +// ExportAudit writes the rows inside [since, until) as JSON lines, oldest first, +// with the optional columns left out when empty. +func TestExportAuditWritesTheWindow(t *testing.T) { + ctx := context.Background() + u := newUser(t, "user", "export") + tag := suffix(t) + base := time.Date(2031, 3, 4, 0, 0, 0, 0, time.UTC) + mustExec(t, `INSERT INTO audit_logs (actor, source, action, request_id, created_at) VALUES ('before', 'external', 'x.before', $1, $2)`, + "exp-"+tag, base.Add(-time.Second)) + mustExec(t, `INSERT INTO audit_logs (actor, actor_user_id, source, action, server_name, request_id, client_ip, user_agent, payload, created_at) + VALUES ('alice@example.net', $1, 'external', 'server.patch', 'survival', $2, '192.0.2.7', 'curl/8', '{"display_name":"Survival"}', $3)`, + u.ID, "exp-"+tag, base.Add(2*time.Hour)) + mustExec(t, `INSERT INTO audit_logs (actor, source, action, created_at) VALUES ('internal', 'internal:velocity', 'server.wake', $1)`, + base.Add(time.Hour)) + mustExec(t, `INSERT INTO audit_logs (actor, source, action, request_id, created_at) VALUES ('after', 'external', 'x.after', $1, $2)`, + "exp-"+tag, base.Add(24*time.Hour)) + + var buf bytes.Buffer + n, err := retention.ExportAudit(ctx, db, base, base.Add(24*time.Hour), &buf) + if err != nil { + t.Fatalf("ExportAudit: %v", err) + } + // Row ids are serial; blank them to compare the rest literally. + got := idRE.ReplaceAllString(buf.String(), `"id":0`) + want := `{"id":0,"created_at":"2031-03-04T01:00:00Z","actor":"internal","source":"internal:velocity","action":"server.wake"}` + "\n" + + `{"id":0,"created_at":"2031-03-04T02:00:00Z","actor":"alice@example.net","actor_user_id":"` + u.ID + `","source":"external","action":"server.patch","server_name":"survival","request_id":"exp-` + tag + `","client_ip":"192.0.2.7","user_agent":"curl/8","payload":{"display_name":"Survival"}}` + "\n" + if n != 2 || got != want { + t.Fatalf("ExportAudit wrote %d rows:\n%s\nwant 2:\n%s", n, got, want) + } +} diff --git a/internal/retention/retention.go b/internal/retention/retention.go new file mode 100644 index 0000000..b5f6e79 --- /dev/null +++ b/internal/retention/retention.go @@ -0,0 +1,203 @@ +// Package retention deletes the control-plane rows that stopped mattering, so +// the sign-in tables and the audit trail do not grow for as long as an install +// runs. +// +// Every sign-in door leaves a row behind: a session per login, an email code or +// passkey challenge per attempt, a bind code per /felis link, an op-login request +// per staff sign-in, a setup token per invite. Each is spent once it expires, +// is used, or is revoked, and a spent row only holds a hash nothing accepts any +// more. Prune drops those rows Grace after they were spent: long enough that an +// incident can still line up a session's device and address with the audit +// trail, short enough that the tables stay the size of what is live. +// +// Audit rows are kept for the install's [audit] retention and then deleted, oldest +// first. `felis db audit-export` (ExportAudit) writes the rows to a file first +// for an install that must keep them longer. +package retention + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "io" + "log/slog" + "time" +) + +// Grace is how long a spent sign-in row outlives the moment it was spent. +const Grace = 30 * 24 * time.Hour + +// batch bounds one DELETE, so a first run on an install that never pruned stays +// clear of the pool's statement timeout and holds its row locks briefly. +const batch = 5000 + +// sweep is one table's spent rows: every statement takes the cutoff as $1 and a +// row limit as $2, and deletes at most $2 rows spent before the cutoff. +type sweep struct { + table string + where string +} + +// sweeps name each sign-in table's spent rows. A session is spent once it +// expired or was revoked; a single-use code or challenge once it expired or was +// redeemed; a bind code once it expired (redeeming deletes it); a migration that +// never completed once nobody touched it for Grace (a redeemed one is the record +// of which account moved where and is kept); an OTP failure window once it began +// Grace ago (a window lasts a day). +var sweeps = []sweep{ + {"sessions", "expires_at < $1 OR revoked_at < $1"}, + {"email_otps", "expires_at < $1 OR consumed_at < $1"}, + {"webauthn_challenges", "expires_at < $1 OR consumed_at < $1"}, + {"webauthn_discoverable_challenges", "expires_at < $1 OR consumed_at < $1"}, + {"setup_tokens", "expires_at < $1 OR consumed_at < $1"}, + {"op_login_requests", "expires_at < $1 OR consumed_at < $1"}, + {"account_link_codes", "expires_at < $1"}, + {"account_migrations", "state <> 'redeemed' AND updated_at < $1"}, + {"otp_failure_windows", "window_start < $1"}, +} + +// Policy is what Prune keeps. +type Policy struct { + // Audit is how long audit rows are kept; 0 keeps every row. + Audit time.Duration +} + +// Result counts the rows Prune deleted, by table. +type Result map[string]int64 + +// Total is the number of rows deleted across all tables. +func (r Result) Total() int64 { + var n int64 + for _, v := range r { + n += v + } + return n +} + +// Prune deletes the spent sign-in rows and the audit rows past the policy's +// retention. It stops at the first failing table and returns what it deleted +// so far with the error; the next run picks up from there. +func Prune(ctx context.Context, db *sql.DB, now time.Time, p Policy) (Result, error) { + res := Result{} + for _, s := range sweeps { + n, err := deleteBatched(ctx, db, s.table, s.where, now.Add(-Grace)) + res[s.table] = n + if err != nil { + return res, fmt.Errorf("prune %s: %w", s.table, err) + } + } + if p.Audit > 0 { + n, err := deleteBatched(ctx, db, "audit_logs", "created_at < $1", now.Add(-p.Audit)) + res["audit_logs"] = n + if err != nil { + return res, fmt.Errorf("prune audit_logs: %w", err) + } + } + return res, nil +} + +func deleteBatched(ctx context.Context, db *sql.DB, table, where string, cutoff time.Time) (int64, error) { + q := fmt.Sprintf(`DELETE FROM %[1]s WHERE ctid IN (SELECT ctid FROM %[1]s WHERE %[2]s LIMIT $2)`, table, where) + var total int64 + for { + r, err := db.ExecContext(ctx, q, cutoff, batch) + if err != nil { + return total, err + } + n, err := r.RowsAffected() + if err != nil { + return total, err + } + total += n + if n < batch { + return total, nil + } + } +} + +// Loop runs Prune a minute after it starts and then every interval until ctx +// ends, logging what each run deleted. +func Loop(ctx context.Context, db *sql.DB, p Policy, every time.Duration, log *slog.Logger) { + t := time.NewTimer(time.Minute) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + } + res, err := Prune(ctx, db, time.Now(), p) + if err != nil { + log.Error("retention: prune failed", "err", err, "deleted", res.Total()) + } else if res.Total() > 0 { + args := []any{"deleted", res.Total()} + for _, s := range sweeps { + if res[s.table] > 0 { + args = append(args, s.table, res[s.table]) + } + } + if res["audit_logs"] > 0 { + args = append(args, "audit_logs", res["audit_logs"]) + } + log.Info("retention: pruned spent rows", args...) + } + t.Reset(every) + } +} + +// AuditRow is one audit_logs row as ExportAudit writes it. +type AuditRow struct { + ID int64 `json:"id"` + CreatedAt time.Time `json:"created_at"` + Actor string `json:"actor"` + ActorUserID string `json:"actor_user_id,omitempty"` + Source string `json:"source"` + Action string `json:"action"` + ServerName string `json:"server_name,omitempty"` + RequestID string `json:"request_id,omitempty"` + ClientIP string `json:"client_ip,omitempty"` + UserAgent string `json:"user_agent,omitempty"` + Payload json.RawMessage `json:"payload,omitempty"` +} + +// ExportAudit writes the audit rows created in [since, until) to w as JSON lines, +// oldest first; a zero bound is open. It returns how many rows it wrote. +func ExportAudit(ctx context.Context, db *sql.DB, since, until time.Time, w io.Writer) (int, error) { + const q = `SELECT id, created_at, actor, COALESCE(actor_user_id, ''), source, action, + COALESCE(server_name, ''), COALESCE(request_id, ''), COALESCE(host(client_ip), ''), + COALESCE(user_agent, ''), COALESCE(payload::text, '') + FROM audit_logs + WHERE ($1::timestamptz IS NULL OR created_at >= $1) AND ($2::timestamptz IS NULL OR created_at < $2) + ORDER BY created_at, id` + rows, err := db.QueryContext(ctx, q, nullTime(since), nullTime(until)) + if err != nil { + return 0, err + } + defer rows.Close() + enc := json.NewEncoder(w) + n := 0 + for rows.Next() { + var ( + r AuditRow + payload string + ) + if err := rows.Scan(&r.ID, &r.CreatedAt, &r.Actor, &r.ActorUserID, &r.Source, &r.Action, + &r.ServerName, &r.RequestID, &r.ClientIP, &r.UserAgent, &payload); err != nil { + return n, err + } + r.CreatedAt = r.CreatedAt.UTC() + if payload != "" { + r.Payload = json.RawMessage(payload) + } + if err := enc.Encode(r); err != nil { + return n, err + } + n++ + } + return n, rows.Err() +} + +func nullTime(t time.Time) sql.NullTime { + return sql.NullTime{Time: t, Valid: !t.IsZero()} +}