feat(retention): felis-api 定时清理过期会话、验证码、挑战、绑定码等表,审计按 [audit] retention 保留,新增 felis db audit-export 导出归档

This commit is contained in:
Lemon-miaow committed 2026-09-25 15:53:34 +08:00
1 parent 38288e1c60
commit 084ba1ed9e
9 files changed
+695 -21

No files matched your search

+15
View File
@@ -30,6 +30,7 @@ import (
"felis.lolicon.best/internal/reaper" "felis.lolicon.best/internal/reaper"
"felis.lolicon.best/internal/registryprune" "felis.lolicon.best/internal/registryprune"
"felis.lolicon.best/internal/restore" "felis.lolicon.best/internal/restore"
"felis.lolicon.best/internal/retention"
"felis.lolicon.best/internal/submit" "felis.lolicon.best/internal/submit"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -84,6 +85,14 @@ func cmdAPI(args []string, stdout, stderr io.Writer) int {
return 1 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() ctx := ctrl.SetupSignalHandler()
// Before anything serves: an api on a schema it was not built for answers with // 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 pruner.Loop(ctx, registryPruneInterval)
} }
go reapRejectedContexts(ctx, submissions, stderr) go reapRejectedContexts(ctx, submissions, stderr)
go retention.Loop(ctx, drv.DB(), retention.Policy{Audit: auditRetention}, retentionInterval, slog.Default())
servers := []*http.Server{internalSrv, externalSrv} servers := []*http.Server{internalSrv, externalSrv}
if httpsSrv != nil { 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 // registryPruneInterval spaces the registry pruner's runs. The registry-gc
// sidecar sweeps once a day, so pruning more often only changes which sweep frees // sidecar sweeps once a day, so pruning more often only changes which sweep frees
// a layer. // a layer.
+86
View File
@@ -15,6 +15,7 @@ import (
"felis.lolicon.best/internal/config" "felis.lolicon.best/internal/config"
"felis.lolicon.best/internal/dbbackup" "felis.lolicon.best/internal/dbbackup"
"felis.lolicon.best/internal/retention"
) )
const dbUsage = `usage: const dbUsage = `usage:
@@ -24,6 +25,7 @@ const dbUsage = `usage:
felis db verify [-dir dir] <bundle> felis db verify [-dir dir] <bundle>
felis db list [-dir dir] felis db list [-dir dir]
felis db check [-dir dir] [-max-age 26h] 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 // 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) return dbList(fs, dir, rest, stdout, stderr)
case "check": case "check":
return dbCheck(fs, dir, rest, stdout, stderr) return dbCheck(fs, dir, rest, stdout, stderr)
case "audit-export":
return dbAuditExport(fs, rest, stdout, stderr)
case "-h", "--help", "help": case "-h", "--help", "help":
fmt.Fprint(stdout, dbUsage) fmt.Fprint(stdout, dbUsage)
return 0 return 0
@@ -262,6 +266,88 @@ func dbList(fs *flag.FlagSet, dir *string, args []string, stdout, stderr io.Writ
return 0 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 { func humanBytes(n int64) string {
const unit = 1024 const unit = 1024
if n < unit { if n < unit {
+34
View File
@@ -8,6 +8,7 @@ import (
"io" "io"
"strings" "strings"
"testing" "testing"
"time"
"felis.lolicon.best/internal/store" "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)
}
}
}
+3 -21
View File
@@ -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: // parseSpanDuration parses the human spans used in felis.toml's [archive] table
// "3mo" (months≈30d), "15d" (days), or any time.ParseDuration unit ("12h"). // (config.ParseSpan).
func parseSpanDuration(s string) (time.Duration, error) { func parseSpanDuration(s string) (time.Duration, error) { return config.ParseSpan(s) }
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)
}
}
// parseByteSize parses a Kubernetes-style quantity ("200Gi", "10G") into bytes. // parseByteSize parses a Kubernetes-style quantity ("200Gi", "10G") into bytes.
// An empty string means unlimited (0). // An empty string means unlimited (0).
+28
View File
@@ -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` `felis-api` and counts in `felis_audit_write_failures_total`
(`FelisAuditWriteFailing`). The cause is almost always PostgreSQL (§16). (`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 ### Optional: a Cloudflare rate limiting rule in front
The limits above live in the API, so they hold on any edge. Behind Cloudflare 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 | | Right code refused; `otp_account_locked` / `FelisOTPAccountLocked` | §17 |
| `FelisSignInFailures` / who is guessing, from where | §17 | | `FelisSignInFailures` / who is guessing, from where | §17 |
| `FelisAuditWriteFailing` | §17 | | `FelisAuditWriteFailing` | §17 |
| How long sessions, codes and audit rows are kept; export audit rows | §17 |
+63
View File
@@ -9,7 +9,9 @@ import (
"net" "net"
"net/url" "net/url"
"regexp" "regexp"
"strconv"
"strings" "strings"
"time"
"github.com/BurntSushi/toml" "github.com/BurntSushi/toml"
) )
@@ -25,6 +27,7 @@ type Config struct {
Archive ArchiveConfig `toml:"archive"` Archive ArchiveConfig `toml:"archive"`
Offsite OffsiteConfig `toml:"offsite"` Offsite OffsiteConfig `toml:"offsite"`
SMTP SMTPConfig `toml:"smtp"` SMTP SMTPConfig `toml:"smtp"`
Audit AuditConfig `toml:"audit"`
// AuthSources is the [[auth_source]] array-of-tables: the third-party Yggdrasil // AuthSources is the [[auth_source]] array-of-tables: the third-party Yggdrasil
// roots the Felis-nano hasJoined multiplexer federates over, in priority order // 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 // (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"` 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 // OffsiteConfig is the [offsite] table: the S3-compatible bucket, away from
// this machine, that `felis offsite sync` (felis-offsite.timer on the host) // this machine, that `felis offsite sync` (felis-offsite.timer on the host)
// copies every world archive and the newest database bundles into, encrypted // 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 { if err := c.Offsite.validate(); err != nil {
return err return err
} }
if _, err := c.Audit.RetentionPeriod(); err != nil {
return err
}
if h := c.Auth.ClientIPHeader; strings.ContainsAny(h, " :\t\r\n") { 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) 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)
} }
+48
View File
@@ -5,6 +5,7 @@ import (
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"time"
"felis.lolicon.best/internal/config" "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)
}
}
}
+215
View File
@@ -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, '[email protected]', '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, '[email protected]', $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 ('[email protected]', $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":"[email protected]","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)
}
}
+203
View File
@@ -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()}
}