From 50539a17c83d58041d100772a54b49e27cdbeaf1 Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 02:04:13 +0530 Subject: [PATCH 1/7] feat(audit): append-only audit log for destructive ops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add internal/audit leaf: an Auditor seam (Begin/End + Record helper, nil = no-op) and a JSONL FileAuditor that appends who/what/when and the outcome for each destructive operation. Write errors are swallowed — auditing must never fail the operation it records. - Add a single interception point in the app layer (guardedOp): it runs an authorization Gate (no-op until the 2FA cycle implements it) and opens the audit record, returning a done(err) the verb defers. Wire it into Backup, Restore, Sync, and Prune. This is the shared seam the 2FA and telemetry cycles reuse rather than re-wrapping every verb. - Add Deps.Auditor / Deps.Gate / Deps.Actor; build a FileAuditor in the CLI when the new audit config block is enabled (default off; path defaults to the XDG state dir), attributing the OS user as actor. - Test the file sink (JSONL, ok/error outcome, append) and the seam (a verb is audited; a gate-blocked op is neither run nor audited). --- internal/app/backup.go | 25 ++++++++++- internal/app/ops.go | 31 +++++++++++++ internal/app/prune.go | 9 ++++ internal/app/prune_test.go | 60 +++++++++++++++++++++++++ internal/app/restore.go | 8 +++- internal/app/sync.go | 9 +++- internal/audit/audit.go | 72 ++++++++++++++++++++++++++++++ internal/audit/audit_test.go | 85 ++++++++++++++++++++++++++++++++++++ internal/audit/file.go | 70 +++++++++++++++++++++++++++++ internal/cli/root.go | 36 +++++++++++++++ internal/config/config.go | 8 ++++ 11 files changed, 410 insertions(+), 3 deletions(-) create mode 100644 internal/app/ops.go create mode 100644 internal/audit/audit.go create mode 100644 internal/audit/audit_test.go create mode 100644 internal/audit/file.go diff --git a/internal/app/backup.go b/internal/app/backup.go index 8a36aa1..e218992 100644 --- a/internal/app/backup.go +++ b/internal/app/backup.go @@ -13,6 +13,7 @@ import ( "github.com/oklog/ulid/v2" + "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/canonical" "github.com/nixrajput/siphon/internal/driver" "github.com/nixrajput/siphon/internal/dumps" @@ -28,6 +29,21 @@ type Deps struct { Dumps *dumps.Catalog Runner *jobs.Runner Drivers DriverGetter + // Auditor records destructive operations; nil is a no-op. It is also the + // interception seam for 2FA gating (a pre-check before the verb) and + // telemetry (timing/outcome), so those reuse these call sites. + Auditor audit.Auditor + // Gate, if set, is consulted before a destructive verb runs and can block it + // (e.g. require 2FA / destructive confirmation for a profile's group). + Gate Gate + // Actor is the OS user attributed in audit records. + Actor string +} + +// Gate authorizes a destructive operation before it runs. A nil Gate allows +// everything. Returning a non-nil error blocks the verb. +type Gate interface { + Authorize(ctx context.Context, op audit.Op, profile string) error } // DriverGetter is satisfied by internal/driver.Get. Wrapped to allow mocking. @@ -60,10 +76,17 @@ func Backup(parent context.Context, d Deps, opt BackupOpts) (<-chan jobs.Event, if err != nil { return nil, "", err } + // Gate (2FA/confirmation) runs synchronously at launch — a block aborts + // before the job starts. The audit record is finalized when the job ends. + done, err := guardedOp(parent, d, audit.OpBackup, opt.Profile, "") + if err != nil { + return nil, "", err + } return d.Runner.Run(parent, jobs.Job{ Stage: "backup", - Func: func(ctx context.Context, emit func(jobs.Event)) error { + Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { + defer func() { done(retErr) }() conn, err := drv.Connect(ctx, resolved) if err != nil { return err diff --git a/internal/app/ops.go b/internal/app/ops.go new file mode 100644 index 0000000..65c227f --- /dev/null +++ b/internal/app/ops.go @@ -0,0 +1,31 @@ +package app + +import ( + "context" + + "github.com/nixrajput/siphon/internal/audit" +) + +// guardedOp is the single interception point wrapping every destructive verb. +// It (1) runs the authorization Gate (2FA / confirmation) before the operation +// and (2) opens an audit record. It returns a done(err) the caller defers to +// finalize the audit outcome. If the Gate blocks, it returns a non-nil error and +// a no-op done — the caller must not run the operation. +// +// Layering all destructive concerns here means each verb gains one guard call +// at entry and one deferred done(err); audit, 2FA gating, and (later) telemetry +// hook in here rather than each re-wrapping every verb. +func guardedOp(ctx context.Context, d Deps, op audit.Op, profile, target string) (done func(error), err error) { + if d.Gate != nil { + if gErr := d.Gate.Authorize(ctx, op, profile); gErr != nil { + return func(error) {}, gErr + } + } + rec := audit.Record(ctx, d.Auditor, audit.Event{ + Op: op, + Profile: profile, + Target: target, + Actor: d.Actor, + }) + return rec, nil +} diff --git a/internal/app/prune.go b/internal/app/prune.go index c9f8ab4..22546a8 100644 --- a/internal/app/prune.go +++ b/internal/app/prune.go @@ -4,6 +4,7 @@ import ( "context" "time" + "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/dumps" ) @@ -45,8 +46,16 @@ type PruneResult struct { // are removed before the base, so an interrupted prune leaves at worst a // complete shorter chain — never a base missing under a surviving incremental. func Prune(ctx context.Context, d Deps, opt PruneOpts) (*PruneResult, error) { + done, err := guardedOp(ctx, d, audit.OpPrune, opt.Profile, "") + if err != nil { + return nil, err + } + var retErr error + defer func() { done(retErr) }() + all, err := d.Dumps.List(ctx) if err != nil { + retErr = err return nil, err } // Scope to the requested profile before grouping, so chains and the plan diff --git a/internal/app/prune_test.go b/internal/app/prune_test.go index 4638b46..cf8e315 100644 --- a/internal/app/prune_test.go +++ b/internal/app/prune_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/dumps" "github.com/nixrajput/siphon/internal/jobs" ) @@ -223,6 +224,65 @@ func TestPrune_StopsChainOnFirstFailure(t *testing.T) { } } +// fakeAuditor records the events it sees, and the outcome passed to End. +type fakeAuditor struct { + begun []audit.Event + ended []error +} + +func (a *fakeAuditor) Begin(_ context.Context, ev audit.Event) audit.Handle { + a.begun = append(a.begun, ev) + return &fakeHandle{a: a} +} + +type fakeHandle struct{ a *fakeAuditor } + +func (h *fakeHandle) End(err error) { h.a.ended = append(h.a.ended, err) } + +// blockGate denies every operation. +type blockGate struct{ err error } + +func (g blockGate) Authorize(_ context.Context, _ audit.Op, _ string) error { return g.err } + +func TestPrune_AuditsTheOperation(t *testing.T) { + store := newRecordingStore() + cat := dumps.New(store) + seedDump(t, cat, "a", "p", time.Now(), "", "") + a := &fakeAuditor{} + deps := Deps{Dumps: cat, Runner: jobs.NewRunner(), Auditor: a, Actor: "alice"} + + if _, err := Prune(context.Background(), deps, PruneOpts{Policy: dumps.RetentionPolicy{KeepLast: 1}}); err != nil { + t.Fatalf("Prune: %v", err) + } + if len(a.begun) != 1 || a.begun[0].Op != audit.OpPrune || a.begun[0].Actor != "alice" { + t.Fatalf("audit Begin = %+v, want one prune by alice", a.begun) + } + if len(a.ended) != 1 || a.ended[0] != nil { + t.Errorf("audit End = %v, want one nil (success)", a.ended) + } +} + +func TestPrune_GateBlocksAndIsNotAudited(t *testing.T) { + store := newRecordingStore() + cat := dumps.New(store) + seedDump(t, cat, "a", "p", time.Now(), "", "") + a := &fakeAuditor{} + denied := errors.New("2FA required") + deps := Deps{Dumps: cat, Runner: jobs.NewRunner(), Auditor: a, Gate: blockGate{err: denied}} + + _, err := Prune(context.Background(), deps, PruneOpts{Policy: dumps.RetentionPolicy{KeepLast: 1}, Apply: true}) + if !errors.Is(err, denied) { + t.Fatalf("Prune err = %v, want the gate's denial", err) + } + // Blocked before running: nothing deleted, and no audit Begin (the op never started). + if len(store.deletes) != 0 { + t.Errorf("gate-blocked prune still deleted: %v", store.deletes) + } + if len(a.begun) != 0 { + t.Errorf("gate-blocked op was audited as begun: %+v", a.begun) + } +} + // helpers func firstIndexWithPrefix(s []string, p string) int { diff --git a/internal/app/restore.go b/internal/app/restore.go index af885eb..b420667 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -6,6 +6,7 @@ import ( "errors" "io" + "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/canonical" "github.com/nixrajput/siphon/internal/driver" "github.com/nixrajput/siphon/internal/dumps" @@ -47,10 +48,15 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event return nil, "", err } } + done, err := guardedOp(parent, d, audit.OpRestore, opt.Profile, opt.DumpID) + if err != nil { + return nil, "", err + } return d.Runner.Run(parent, jobs.Job{ Stage: "restore", - Func: func(ctx context.Context, emit func(jobs.Event)) error { + Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { + defer func() { done(retErr) }() conn, err := drv.Connect(ctx, resolved) if err != nil { return err diff --git a/internal/app/sync.go b/internal/app/sync.go index b499f6b..a991591 100644 --- a/internal/app/sync.go +++ b/internal/app/sync.go @@ -3,6 +3,7 @@ package app import ( "context" + "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/canonical" "github.com/nixrajput/siphon/internal/driver" "github.com/nixrajput/siphon/internal/errs" @@ -59,9 +60,15 @@ func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, stri return runCrossEngineSync(parent, d, opt) } + done, err := guardedOp(parent, d, audit.OpSync, opt.From, opt.To) + if err != nil { + return nil, "", err + } + return d.Runner.Run(parent, jobs.Job{ Stage: "sync", - Func: func(ctx context.Context, emit func(jobs.Event)) error { + Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { + defer func() { done(retErr) }() srcConn, err := srcDrv.Connect(ctx, src) if err != nil { return err diff --git a/internal/audit/audit.go b/internal/audit/audit.go new file mode 100644 index 0000000..c90ae39 --- /dev/null +++ b/internal/audit/audit.go @@ -0,0 +1,72 @@ +// Package audit records an append-only trail of destructive operations +// (backup, restore, sync, prune) — who ran what, against which profile, when, +// and whether it succeeded. It is a stdlib-only leaf: the app layer holds an +// Auditor and calls Begin before a destructive verb and End after it. +// +// The same Begin/End seam is the interception point reused by other cross- +// cutting concerns (2FA gating runs as a pre-check before the verb; telemetry +// records timing/outcome) — they are layered on top rather than each re-wrapping +// every verb. +package audit + +import ( + "context" + "time" +) + +// Op names a destructive operation, used as the audit record's action. +type Op string + +const ( + OpBackup Op = "backup" + OpRestore Op = "restore" + OpSync Op = "sync" + OpPrune Op = "prune" +) + +// Event is one audit record. Outcome and DurationMS are filled in at End; the +// rest are known at Begin. +type Event struct { + Time time.Time `json:"time"` + Op Op `json:"op"` + Profile string `json:"profile,omitempty"` + Target string `json:"target,omitempty"` // e.g. sync destination, dump id + Actor string `json:"actor"` // OS user who ran the command + Outcome string `json:"outcome"` // "ok" | "error" (set at End) + Err string `json:"error,omitempty"` // error text when Outcome == "error" + DurationMS int64 `json:"duration_ms"` // wall time Begin→End +} + +// Auditor records audit events. A nil Auditor is valid and is a no-op, so the +// app layer can call Begin/End unconditionally without nil checks. +type Auditor interface { + // Begin records the start of op and returns a handle to End. Implementations + // may persist a "started" record or defer all writes to End; the app layer + // does not care which. + Begin(ctx context.Context, ev Event) Handle +} + +// Handle finalizes one operation's audit record. +type Handle interface { + // End records the operation's outcome. err == nil means success. It is safe + // to call End on a zero/nil Handle (no-op). + End(err error) +} + +// Record runs Begin, returns a function to defer that calls End with the +// verb's error. Usage in an app verb: +// +// done := audit.Record(ctx, d.Auditor, audit.Event{Op: audit.OpPrune, Profile: p, Actor: actor}) +// defer func() { done(retErr) }() +// +// A nil auditor yields a no-op done func, so callers need no nil check. +func Record(ctx context.Context, a Auditor, ev Event) func(error) { + if a == nil { + return func(error) {} + } + h := a.Begin(ctx, ev) + if h == nil { + return func(error) {} + } + return h.End +} diff --git a/internal/audit/audit_test.go b/internal/audit/audit_test.go new file mode 100644 index 0000000..587e11e --- /dev/null +++ b/internal/audit/audit_test.go @@ -0,0 +1,85 @@ +package audit + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "testing" + "time" +) + +func TestRecord_NilAuditorIsNoOp(t *testing.T) { + // Record with a nil Auditor must return a usable no-op done func (the app + // layer calls it unconditionally). + done := Record(context.Background(), nil, Event{Op: OpBackup}) + done(nil) + done(errors.New("x")) // must not panic +} + +func TestFileAuditor_WritesJSONLOnEnd(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "audit.log") + clock := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + a, err := NewFileAuditor(path, func() time.Time { return clock }) + if err != nil { + t.Fatalf("NewFileAuditor: %v", err) + } + + // One successful op, one failed op. + Record(context.Background(), a, Event{Op: OpBackup, Profile: "prod", Actor: "alice"})(nil) + Record(context.Background(), a, Event{Op: OpRestore, Profile: "prod", Target: "dump1", Actor: "alice"})(errors.New("boom")) + + events := readAudit(t, path) + if len(events) != 2 { + t.Fatalf("got %d audit lines, want 2", len(events)) + } + if events[0].Op != OpBackup || events[0].Outcome != "ok" || events[0].Err != "" { + t.Errorf("event[0] = %+v, want backup/ok/no-err", events[0]) + } + if events[0].Actor != "alice" || events[0].Profile != "prod" { + t.Errorf("event[0] attribution = %+v, want actor=alice profile=prod", events[0]) + } + if events[1].Op != OpRestore || events[1].Outcome != "error" || events[1].Err != "boom" { + t.Errorf("event[1] = %+v, want restore/error/boom", events[1]) + } + if events[1].Target != "dump1" { + t.Errorf("event[1].Target = %q, want dump1", events[1].Target) + } +} + +func TestFileAuditor_AppendsAcrossCalls(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "audit.log") + a, _ := NewFileAuditor(path, nil) + for i := 0; i < 3; i++ { + Record(context.Background(), a, Event{Op: OpPrune})(nil) + } + if got := len(readAudit(t, path)); got != 3 { + t.Errorf("append: got %d lines, want 3", got) + } +} + +func readAudit(t *testing.T, path string) []Event { + t.Helper() + f, err := os.Open(path) + if err != nil { + t.Fatalf("open audit log: %v", err) + } + defer func() { _ = f.Close() }() + var out []Event + sc := bufio.NewScanner(f) + for sc.Scan() { + if len(sc.Bytes()) == 0 { + continue + } + var ev Event + if err := json.Unmarshal(sc.Bytes(), &ev); err != nil { + t.Fatalf("bad audit JSON %q: %v", sc.Text(), err) + } + out = append(out, ev) + } + return out +} diff --git a/internal/audit/file.go b/internal/audit/file.go new file mode 100644 index 0000000..d260275 --- /dev/null +++ b/internal/audit/file.go @@ -0,0 +1,70 @@ +package audit + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "sync" + "time" +) + +// FileAuditor appends audit events as JSONL to a single log file. Writes are +// serialized by a mutex so concurrent verbs (e.g. parallel jobs) don't interleave +// partial lines. A write error is swallowed: auditing must never fail the +// operation it records (a backup that succeeded must not report failure because +// the audit line couldn't be written) — best-effort, append-only. +type FileAuditor struct { + path string + mu sync.Mutex + now func() time.Time // injectable for tests +} + +// NewFileAuditor returns an Auditor appending to path, creating its parent +// directory. now defaults to time.Now when nil. +func NewFileAuditor(path string, now func() time.Time) (*FileAuditor, error) { + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return nil, err + } + if now == nil { + now = time.Now + } + return &FileAuditor{path: path, now: now}, nil +} + +func (a *FileAuditor) Begin(_ context.Context, ev Event) Handle { + ev.Time = a.now() + return &fileHandle{a: a, ev: ev, start: ev.Time} +} + +func (a *FileAuditor) append(ev Event) { + line, err := json.Marshal(ev) + if err != nil { + return + } + a.mu.Lock() + defer a.mu.Unlock() + f, err := os.OpenFile(a.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) + if err != nil { + return // best-effort: never fail the audited operation + } + defer func() { _ = f.Close() }() + _, _ = f.Write(append(line, '\n')) +} + +type fileHandle struct { + a *FileAuditor + ev Event + start time.Time +} + +func (h *fileHandle) End(err error) { + h.ev.DurationMS = h.a.now().Sub(h.start).Milliseconds() + if err != nil { + h.ev.Outcome = "error" + h.ev.Err = err.Error() + } else { + h.ev.Outcome = "ok" + } + h.a.append(h.ev) +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 4c38332..5e1c05f 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -8,11 +8,13 @@ import ( "fmt" "io" "os" + "os/user" "path/filepath" "github.com/spf13/cobra" "github.com/nixrajput/siphon/internal/app" + "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/config" "github.com/nixrajput/siphon/internal/dumps" "github.com/nixrajput/siphon/internal/errs" @@ -80,14 +82,48 @@ func buildDeps() (app.Deps, error) { } cat := dumps.New(store) + auditor, err := buildAuditor(cfg) + if err != nil { + return app.Deps{}, err + } + return app.Deps{ Profiles: ps, Dumps: cat, Runner: jobs.NewRunner(), Drivers: app.DefaultDrivers(), + Auditor: auditor, + Actor: osUser(), }, nil } +// buildAuditor returns a file-backed Auditor when audit logging is enabled in +// config, else nil (a nil Auditor is a no-op). Path defaults to the XDG state +// dir when unset. +func buildAuditor(cfg *config.Config) (audit.Auditor, error) { + if !cfg.Audit.Enabled { + return nil, nil + } + path := cfg.Audit.Path + if path == "" { + home, err := os.UserHomeDir() + if err != nil { + return nil, fmt.Errorf("resolve home dir for default audit path: %w", err) + } + path = filepath.Join(home, ".local", "state", "siphon", "audit.log") + } + return audit.NewFileAuditor(path, nil) +} + +// osUser returns the current OS username for audit attribution, falling back to +// "unknown" when it cannot be determined. +func osUser() string { + if u, err := user.Current(); err == nil && u.Username != "" { + return u.Username + } + return "unknown" +} + // buildStore selects the dump-catalog storage backend from config. Type "s3" // builds an S3-backed store; anything else (the default) uses the local // filesystem rooted at Defaults.DumpDir (or the XDG share dir when unset). diff --git a/internal/config/config.go b/internal/config/config.go index 8fa9a81..050598a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,10 +15,18 @@ type Config struct { Version int `yaml:"version"` Defaults Defaults `yaml:"defaults"` Storage StorageConfig `yaml:"storage"` + Audit AuditConfig `yaml:"audit"` Profiles map[string]ProfileConfig `yaml:"profiles"` Groups map[string]GroupConfig `yaml:"groups"` } +// AuditConfig controls the append-only audit log of destructive operations. +// Disabled by default; Path defaults to /siphon/audit.log when empty. +type AuditConfig struct { + Enabled bool `yaml:"enabled"` + Path string `yaml:"path,omitempty"` +} + type Defaults struct { DumpDir string `yaml:"dump_dir"` Jobs int `yaml:"jobs"` From 299a456f3855152050ea031ff7c9555dd5dbf3b1 Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 02:14:16 +0530 Subject: [PATCH 2/7] feat(2fa): gate destructive ops by profile-group policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Implement the app.Gate seam (added in the audit commit) as a prompting CLI gate driven by a profile's group policy: confirm_destructive makes the operator retype the profile name; require_2fa prompts for a TOTP code. A profile with no group, or a group with neither flag set, is allowed silently. The gate runs before the verb, so a denial aborts before any destructive work. - Add internal/twofactor: stdlib-only RFC 6238 TOTP (HMAC-SHA1, 30s, 6 digits) with Verify (±1 step skew) and Generate. No new dependency. - Add GroupConfig.totp_secret (a secret-ref, never plaintext) and resolve it through the existing secret resolver. require_2fa with no resolvable secret fails closed (blocks), never opens. - Build the gate in the CLI only when some group enforces a policy; prompt on stdin, report on stderr so prompts don't pollute stdout. - Test RFC 6238 vectors (current/adjacent/stale steps, formatting tolerance, bad secret) and the gate (no-group passthrough, name confirm match/mismatch, TOTP accept/reject, missing-secret blocks). --- internal/cli/gate.go | 95 +++++++++++++++++++++++++++++++++ internal/cli/gate_test.go | 82 ++++++++++++++++++++++++++++ internal/cli/root.go | 13 +++++ internal/config/config.go | 4 ++ internal/twofactor/totp.go | 83 ++++++++++++++++++++++++++++ internal/twofactor/totp_test.go | 82 ++++++++++++++++++++++++++++ 6 files changed, 359 insertions(+) create mode 100644 internal/cli/gate.go create mode 100644 internal/cli/gate_test.go create mode 100644 internal/twofactor/totp.go create mode 100644 internal/twofactor/totp_test.go diff --git a/internal/cli/gate.go b/internal/cli/gate.go new file mode 100644 index 0000000..0ed316d --- /dev/null +++ b/internal/cli/gate.go @@ -0,0 +1,95 @@ +package cli + +import ( + "bufio" + "context" + "fmt" + "io" + "strings" + "time" + + "github.com/nixrajput/siphon/internal/app" + "github.com/nixrajput/siphon/internal/audit" + "github.com/nixrajput/siphon/internal/config" + "github.com/nixrajput/siphon/internal/errs" + "github.com/nixrajput/siphon/internal/secrets" + "github.com/nixrajput/siphon/internal/twofactor" +) + +// promptGate implements app.Gate by enforcing a profile's group policy before a +// destructive op: ConfirmDestructive prompts the operator to retype the profile +// name, and Require2FA prompts for a TOTP code verified against the group's +// shared secret. Both read from in / write to out, so the gate is wired with the +// command's stdin/stdout (and can be driven by a scripted reader in tests). +// +// A profile with no group, or a group with neither flag, authorizes silently. +type promptGate struct { + cfg *config.Config + res *secrets.Resolver + in io.Reader + out io.Writer + now func() time.Time // TOTP clock; defaults to time.Now +} + +// newPromptGate builds a gate from config + a secret resolver, prompting on in +// and reporting on out. +func newPromptGate(cfg *config.Config, res *secrets.Resolver, in io.Reader, out io.Writer) *promptGate { + return &promptGate{cfg: cfg, res: res, in: in, out: out, now: time.Now} +} + +func (g *promptGate) Authorize(_ context.Context, op audit.Op, profile string) error { + grp, ok := g.groupFor(profile) + if !ok { + return nil // no group → no policy + } + if grp.ConfirmDestructive { + if err := g.confirmName(op, profile); err != nil { + return err + } + } + if grp.Require2FA { + if err := g.confirmTOTP(op, profile, grp.TOTPSecret); err != nil { + return err + } + } + return nil +} + +// groupFor returns the group config for a profile's group, if it has one. +func (g *promptGate) groupFor(profile string) (config.GroupConfig, bool) { + p, ok := g.cfg.Profiles[profile] + if !ok || p.Group == "" { + return config.GroupConfig{}, false + } + grp, ok := g.cfg.Groups[p.Group] + return grp, ok +} + +func (g *promptGate) confirmName(op audit.Op, profile string) error { + _, _ = fmt.Fprintf(g.out, "%s on %q is destructive. Type the profile name to confirm: ", op, profile) + line, _ := bufio.NewReader(g.in).ReadString('\n') + if strings.TrimSpace(line) != profile { + return &errs.Error{Op: "gate", Code: errs.CodeUser, Hint: "confirmation did not match the profile name; aborted"} + } + return nil +} + +func (g *promptGate) confirmTOTP(op audit.Op, profile, secretRef string) error { + secret, err := g.res.Resolve(secretRef) + if err != nil || strings.TrimSpace(secret) == "" { + return &errs.Error{Op: "gate", Code: errs.CodeUser, Hint: "group requires 2FA but its totp_secret is unset or unresolvable"} + } + _, _ = fmt.Fprintf(g.out, "%s on %q requires 2FA. Enter your 6-digit code: ", op, profile) + line, _ := bufio.NewReader(g.in).ReadString('\n') + ok, err := twofactor.Verify(secret, line, g.now()) + if err != nil { + return &errs.Error{Op: "gate", Code: errs.CodeUser, Cause: err, Hint: "invalid TOTP secret configured for this group"} + } + if !ok { + return &errs.Error{Op: "gate", Code: errs.CodeUser, Hint: "incorrect 2FA code; aborted"} + } + return nil +} + +// compile-time assertion that promptGate satisfies the app.Gate seam. +var _ app.Gate = (*promptGate)(nil) diff --git a/internal/cli/gate_test.go b/internal/cli/gate_test.go new file mode 100644 index 0000000..03b1b45 --- /dev/null +++ b/internal/cli/gate_test.go @@ -0,0 +1,82 @@ +package cli + +import ( + "context" + "encoding/base32" + "strings" + "testing" + "time" + + "github.com/nixrajput/siphon/internal/audit" + "github.com/nixrajput/siphon/internal/config" + "github.com/nixrajput/siphon/internal/secrets" + "github.com/nixrajput/siphon/internal/twofactor" +) + +func gateCfg(group config.GroupConfig) *config.Config { + return &config.Config{ + Profiles: map[string]config.ProfileConfig{ + "prod": {Name: "prod", Group: "critical"}, + "sandbox": {Name: "sandbox"}, // no group + }, + Groups: map[string]config.GroupConfig{"critical": group}, + } +} + +func newGate(cfg *config.Config, in string, now time.Time) *promptGate { + g := newPromptGate(cfg, secrets.NewResolver(secrets.Env{}, secrets.Passthrough{}), + strings.NewReader(in), &strings.Builder{}) + g.now = func() time.Time { return now } + return g +} + +func TestGate_NoGroupAuthorizesSilently(t *testing.T) { + g := newGate(gateCfg(config.GroupConfig{ConfirmDestructive: true}), "", time.Now()) + if err := g.Authorize(context.Background(), audit.OpBackup, "sandbox"); err != nil { + t.Errorf("ungrouped profile gated: %v", err) + } +} + +func TestGate_ConfirmDestructive(t *testing.T) { + cfg := gateCfg(config.GroupConfig{ConfirmDestructive: true}) + + // Correct confirmation (types the profile name) passes. + if err := newGate(cfg, "prod\n", time.Now()).Authorize(context.Background(), audit.OpRestore, "prod"); err != nil { + t.Errorf("correct confirmation rejected: %v", err) + } + // Wrong confirmation blocks. + if err := newGate(cfg, "nope\n", time.Now()).Authorize(context.Background(), audit.OpRestore, "prod"); err == nil { + t.Error("wrong confirmation was allowed through") + } +} + +func TestGate_Require2FA(t *testing.T) { + secret := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString([]byte("12345678901234567890")) + cfg := gateCfg(config.GroupConfig{Require2FA: true, TOTPSecret: secret}) + now := time.Unix(10_000, 0) + good := totpAt(t, secret, now) + + if err := newGate(cfg, good+"\n", now).Authorize(context.Background(), audit.OpSync, "prod"); err != nil { + t.Errorf("valid TOTP rejected: %v", err) + } + if err := newGate(cfg, "000000\n", now).Authorize(context.Background(), audit.OpSync, "prod"); err == nil { + t.Error("wrong TOTP was allowed through") + } +} + +func TestGate_Require2FA_MissingSecret(t *testing.T) { + cfg := gateCfg(config.GroupConfig{Require2FA: true}) // no TOTPSecret + if err := newGate(cfg, "123456\n", time.Now()).Authorize(context.Background(), audit.OpSync, "prod"); err == nil { + t.Error("require_2fa with no secret should fail closed (block), not pass") + } +} + +// totpAt returns a valid code for secret at now using the package's generator. +func totpAt(t *testing.T, secret string, now time.Time) string { + t.Helper() + code, err := twofactor.Generate(secret, now) + if err != nil { + t.Fatalf("Generate: %v", err) + } + return code +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 5e1c05f..29dda7f 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -93,10 +93,23 @@ func buildDeps() (app.Deps, error) { Runner: jobs.NewRunner(), Drivers: app.DefaultDrivers(), Auditor: auditor, + Gate: buildGate(cfg, res), Actor: osUser(), }, nil } +// buildGate returns a prompting Gate when any group enforces a destructive-op +// policy (confirm_destructive or require_2fa), else nil (no gating). It prompts +// on stdin and reports on stderr so prompts don't pollute piped stdout. +func buildGate(cfg *config.Config, res *secrets.Resolver) app.Gate { + for _, grp := range cfg.Groups { + if grp.ConfirmDestructive || grp.Require2FA { + return newPromptGate(cfg, res, os.Stdin, os.Stderr) + } + } + return nil +} + // buildAuditor returns a file-backed Auditor when audit logging is enabled in // config, else nil (a nil Auditor is a no-op). Path defaults to the XDG state // dir when unset. diff --git a/internal/config/config.go b/internal/config/config.go index 050598a..67305ab 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -132,6 +132,10 @@ type GroupConfig struct { Color string `yaml:"color"` Require2FA bool `yaml:"require_2fa"` ConfirmDestructive bool `yaml:"confirm_destructive"` + // TOTPSecret is the base32 RFC-6238 secret shared with the operator's + // authenticator app, consulted when Require2FA is set. It is a secret-ref + // (e.g. env:SIPHON_PROD_TOTP), so the plaintext secret never lives in config. + TOTPSecret string `yaml:"totp_secret,omitempty"` } // Load reads and parses the config file. Returns an empty Config if the diff --git a/internal/twofactor/totp.go b/internal/twofactor/totp.go new file mode 100644 index 0000000..9f192c7 --- /dev/null +++ b/internal/twofactor/totp.go @@ -0,0 +1,83 @@ +// Package twofactor implements RFC 6238 TOTP verification with the standard +// library only (HMAC-SHA1, 30-second steps, 6 digits — the parameters every +// authenticator app uses by default). It is a pure leaf: no I/O, no clock +// dependency beyond an injected "now", so verification is fully unit-testable. +// +// siphon uses this to gate destructive operations for profile groups that set +// require_2fa: the group's base32 TOTP secret (a secret-ref, never plaintext) +// is shared with the operator's authenticator app, and the CLI prompts for the +// current 6-digit code before running the operation. +package twofactor + +import ( + "crypto/hmac" + "crypto/sha1" + "encoding/base32" + "encoding/binary" + "fmt" + "strings" + "time" +) + +const ( + period = 30 // seconds per TOTP step (RFC 6238 default) + digits = 6 +) + +// Verify reports whether code is a valid TOTP for secret at time now. It accepts +// the code for the current 30s step and the immediately adjacent steps (±1), the +// standard skew tolerance for clock drift between the operator's device and this +// host. secret is base32 (the format authenticator apps display/scan). +func Verify(secret, code string, now time.Time) (bool, error) { + key, err := decodeSecret(secret) + if err != nil { + return false, err + } + code = strings.TrimSpace(code) + step := now.Unix() / period + for _, s := range []int64{step, step - 1, step + 1} { + if generate(key, s) == code { + return true, nil + } + } + return false, nil +} + +// Generate returns the current 6-digit TOTP code for secret at time now. It is +// the counterpart to Verify (useful for provisioning/round-trip checks). +func Generate(secret string, now time.Time) (string, error) { + key, err := decodeSecret(secret) + if err != nil { + return "", err + } + return generate(key, now.Unix()/period), nil +} + +// decodeSecret base32-decodes a TOTP secret, tolerating lowercase and the +// padding-stripped form authenticator apps commonly show. +func decodeSecret(secret string) ([]byte, error) { + s := strings.ToUpper(strings.ReplaceAll(strings.TrimSpace(secret), " ", "")) + enc := base32.StdEncoding.WithPadding(base32.NoPadding) + key, err := enc.DecodeString(s) + if err != nil { + return nil, fmt.Errorf("twofactor: invalid base32 TOTP secret: %w", err) + } + return key, nil +} + +// generate computes the RFC 6238 / RFC 4226 HOTP value for the given counter. +func generate(key []byte, counter int64) string { + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], uint64(counter)) + mac := hmac.New(sha1.New, key) + mac.Write(buf[:]) + sum := mac.Sum(nil) + + offset := sum[len(sum)-1] & 0x0f + value := (uint32(sum[offset]&0x7f) << 24) | + (uint32(sum[offset+1]) << 16) | + (uint32(sum[offset+2]) << 8) | + uint32(sum[offset+3]) + mod := value % 1_000_000 // 10^digits + return fmt.Sprintf("%0*d", digits, mod) +} diff --git a/internal/twofactor/totp_test.go b/internal/twofactor/totp_test.go new file mode 100644 index 0000000..46d7473 --- /dev/null +++ b/internal/twofactor/totp_test.go @@ -0,0 +1,82 @@ +package twofactor + +import ( + "encoding/base32" + "testing" + "time" +) + +// secret "12345678901234567890" (the RFC 4226/6238 test key) in base32. +var rfcSecret = base32.StdEncoding.WithPadding(base32.NoPadding). + EncodeToString([]byte("12345678901234567890")) + +func TestVerify_AcceptsCurrentStep(t *testing.T) { + now := time.Unix(59, 0) // RFC 6238 first test vector time + code := generate(mustDecode(t, rfcSecret), now.Unix()/period) + ok, err := Verify(rfcSecret, code, now) + if err != nil || !ok { + t.Fatalf("Verify current step = (%v, %v), want (true, nil)", ok, err) + } +} + +func TestVerify_AcceptsAdjacentStepsForSkew(t *testing.T) { + now := time.Unix(10_000, 0) + prev := generate(mustDecode(t, rfcSecret), now.Unix()/period-1) + next := generate(mustDecode(t, rfcSecret), now.Unix()/period+1) + for name, code := range map[string]string{"prev": prev, "next": next} { + ok, err := Verify(rfcSecret, code, now) + if err != nil || !ok { + t.Errorf("%s-step code rejected: (%v, %v), want accepted", name, ok, err) + } + } +} + +func TestVerify_RejectsWrongAndStaleCode(t *testing.T) { + now := time.Unix(10_000, 0) + if ok, _ := Verify(rfcSecret, "000000", now); ok { + t.Error("clearly-wrong code accepted") + } + // A code two steps away is outside the ±1 skew window. + stale := generate(mustDecode(t, rfcSecret), now.Unix()/period-2) + if ok, _ := Verify(rfcSecret, stale, now); ok { + t.Error("stale code (2 steps old) accepted; skew window should be ±1") + } +} + +func TestVerify_TolerantOfFormatting(t *testing.T) { + now := time.Unix(10_000, 0) + code := generate(mustDecode(t, rfcSecret), now.Unix()/period) + // Surrounding whitespace/newline (as ReadString would leave) must be tolerated. + if ok, err := Verify(rfcSecret, " "+code+"\n", now); err != nil || !ok { + t.Errorf("formatted code rejected: (%v, %v)", ok, err) + } + // Lowercase / spaced secret must still decode. + if ok, err := Verify(lower(rfcSecret), code, now); err != nil || !ok { + t.Errorf("lowercase secret rejected: (%v, %v)", ok, err) + } +} + +func TestVerify_BadSecret(t *testing.T) { + if _, err := Verify("not!base32", "123456", time.Unix(0, 0)); err == nil { + t.Error("invalid base32 secret should error") + } +} + +func mustDecode(t *testing.T, s string) []byte { + t.Helper() + k, err := decodeSecret(s) + if err != nil { + t.Fatalf("decodeSecret: %v", err) + } + return k +} + +func lower(s string) string { + b := []byte(s) + for i := range b { + if b[i] >= 'A' && b[i] <= 'Z' { + b[i] += 'a' - 'A' + } + } + return string(b) +} From d98f87be3ddb7519f1525e50e6ed12b7c6e8aaff Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 02:30:31 +0530 Subject: [PATCH 3/7] feat(telemetry): opt-in aggregate op metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add internal/telemetry: a Recorder that plugs into the SAME audit Auditor seam (no new interception point) and accumulates per-op counts and error tallies, flushed as JSON. Off by default. - Privacy by construction: the Recorder reads only the op name and outcome from the audit Event — never profile, actor, target, or any data. A test asserts no identifying field reaches the file. - Add audit.Multi: fans one Begin/End out to several Auditor sinks, so the audit log and telemetry both observe the destructive-op seam without the app layer wiring two hooks. Nil sinks are skipped; all-nil collapses to a nil (no-op) Auditor. - Add a telemetry config block (enabled + path; XDG state default) and compose file-audit + telemetry via Multi in the CLI. - Known scope note (in code): audit.Handle.End carries only the error, not duration, so telemetry tracks counts/error-rate but not timing; carrying duration would mean widening the End signature — deferred. - Test aggregation (counts/errors), the privacy guarantee, empty-path disable, and Multi fan-out / all-nil collapse. --- internal/audit/multi.go | 44 +++++++++++ internal/cli/root.go | 31 +++++++- internal/config/config.go | 21 ++++-- internal/telemetry/telemetry.go | 107 ++++++++++++++++++++++++++ internal/telemetry/telemetry_test.go | 109 +++++++++++++++++++++++++++ 5 files changed, 305 insertions(+), 7 deletions(-) create mode 100644 internal/audit/multi.go create mode 100644 internal/telemetry/telemetry.go create mode 100644 internal/telemetry/telemetry_test.go diff --git a/internal/audit/multi.go b/internal/audit/multi.go new file mode 100644 index 0000000..0ead003 --- /dev/null +++ b/internal/audit/multi.go @@ -0,0 +1,44 @@ +package audit + +import "context" + +// Multi fans one operation's Begin/End out to several Auditors, so the same +// destructive-op seam feeds both the security audit log and the telemetry +// recorder without the app layer wiring two hooks. Nil entries are skipped, and +// Multi of zero live auditors is itself a no-op (returns nil from New). +type Multi struct { + auditors []Auditor +} + +// NewMulti returns an Auditor that broadcasts to all non-nil auditors, or nil if +// none are live (so callers keep the nil-is-no-op contract). +func NewMulti(auditors ...Auditor) Auditor { + live := auditors[:0] + for _, a := range auditors { + if a != nil { + live = append(live, a) + } + } + if len(live) == 0 { + return nil + } + return &Multi{auditors: live} +} + +func (m *Multi) Begin(ctx context.Context, ev Event) Handle { + hs := make([]Handle, 0, len(m.auditors)) + for _, a := range m.auditors { + if h := a.Begin(ctx, ev); h != nil { + hs = append(hs, h) + } + } + return &multiHandle{handles: hs} +} + +type multiHandle struct{ handles []Handle } + +func (h *multiHandle) End(err error) { + for _, inner := range h.handles { + inner.End(err) + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 29dda7f..4017459 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -22,6 +22,7 @@ import ( "github.com/nixrajput/siphon/internal/profile" "github.com/nixrajput/siphon/internal/secrets" "github.com/nixrajput/siphon/internal/storage" + "github.com/nixrajput/siphon/internal/telemetry" "github.com/nixrajput/siphon/internal/tui" ) @@ -82,10 +83,17 @@ func buildDeps() (app.Deps, error) { } cat := dumps.New(store) - auditor, err := buildAuditor(cfg) + fileAuditor, err := buildAuditor(cfg) if err != nil { return app.Deps{}, err } + tel, err := buildTelemetry(cfg) + if err != nil { + return app.Deps{}, err + } + // The audit log and telemetry recorder are both audit.Auditor sinks; Multi + // fans the one destructive-op seam out to whichever are enabled (nil if none). + auditor := audit.NewMulti(fileAuditor, tel) return app.Deps{ Profiles: ps, @@ -98,6 +106,27 @@ func buildDeps() (app.Deps, error) { }, nil } +// buildTelemetry returns a telemetry Recorder (an audit.Auditor sink) when +// telemetry is enabled in config, else nil. Path defaults to the XDG state dir. +func buildTelemetry(cfg *config.Config) (audit.Auditor, error) { + if !cfg.Telemetry.Enabled { + return nil, nil + } + path := cfg.Telemetry.Path + if path == "" { + home, err := os.UserHomeDir() + if err != nil { + return nil, fmt.Errorf("resolve home dir for default telemetry path: %w", err) + } + path = filepath.Join(home, ".local", "state", "siphon", "telemetry.json") + } + r := telemetry.NewRecorder(path) + if r == nil { + return nil, nil + } + return r, nil +} + // buildGate returns a prompting Gate when any group enforces a destructive-op // policy (confirm_destructive or require_2fa), else nil (no gating). It prompts // on stdin and reports on stderr so prompts don't pollute piped stdout. diff --git a/internal/config/config.go b/internal/config/config.go index 67305ab..ca1123f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -12,12 +12,13 @@ import ( ) type Config struct { - Version int `yaml:"version"` - Defaults Defaults `yaml:"defaults"` - Storage StorageConfig `yaml:"storage"` - Audit AuditConfig `yaml:"audit"` - Profiles map[string]ProfileConfig `yaml:"profiles"` - Groups map[string]GroupConfig `yaml:"groups"` + Version int `yaml:"version"` + Defaults Defaults `yaml:"defaults"` + Storage StorageConfig `yaml:"storage"` + Audit AuditConfig `yaml:"audit"` + Telemetry TelemetryConfig `yaml:"telemetry"` + Profiles map[string]ProfileConfig `yaml:"profiles"` + Groups map[string]GroupConfig `yaml:"groups"` } // AuditConfig controls the append-only audit log of destructive operations. @@ -27,6 +28,14 @@ type AuditConfig struct { Path string `yaml:"path,omitempty"` } +// TelemetryConfig controls opt-in aggregate operational metrics (per-op counts +// and error tallies — never identifying data). Disabled by default; Path +// defaults to /siphon/telemetry.json when empty. +type TelemetryConfig struct { + Enabled bool `yaml:"enabled"` + Path string `yaml:"path,omitempty"` +} + type Defaults struct { DumpDir string `yaml:"dump_dir"` Jobs int `yaml:"jobs"` diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go new file mode 100644 index 0000000..0707bc7 --- /dev/null +++ b/internal/telemetry/telemetry.go @@ -0,0 +1,107 @@ +// Package telemetry records opt-in, aggregate operational metrics about +// destructive operations: per-op counts, success/error tallies, and total +// duration. It plugs into the same audit.Auditor seam as the audit log, so it +// adds no new interception point in the app layer. +// +// Privacy: telemetry deliberately records ONLY the operation name, outcome, and +// duration — never profile names, hosts, dump IDs, actor, or any data. It is +// off by default and must be explicitly enabled in config. +package telemetry + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "sort" + "sync" + + "github.com/nixrajput/siphon/internal/audit" +) + +// Recorder is an audit.Auditor sink that accumulates aggregate counters and +// flushes them to a JSON file. Safe for concurrent use. +type Recorder struct { + path string + mu sync.Mutex + ops map[string]*opStat +} + +type opStat struct { + Count int64 `json:"count"` + Errors int64 `json:"errors"` + DurationMS int64 `json:"total_duration_ms"` +} + +// NewRecorder returns a telemetry Recorder that flushes aggregates to path +// (created on first flush), or nil if path is empty. +func NewRecorder(path string) *Recorder { + if path == "" { + return nil + } + return &Recorder{path: path, ops: map[string]*opStat{}} +} + +// Begin returns a handle that, at End, increments this op's aggregate counters +// and flushes the snapshot. Only ev.Op and the outcome/duration are read — no +// identifying fields. +func (r *Recorder) Begin(_ context.Context, ev audit.Event) audit.Handle { + return &recHandle{r: r, op: string(ev.Op)} +} + +type recHandle struct { + r *Recorder + op string +} + +func (h *recHandle) End(err error) { + h.r.mu.Lock() + st := h.r.ops[h.op] + if st == nil { + st = &opStat{} + h.r.ops[h.op] = st + } + st.Count++ + if err != nil { + st.Errors++ + } + snapshot := h.r.snapshotLocked() + h.r.mu.Unlock() + + // Duration is not on the audit.Event at this layer (the handle does not see + // it), so telemetry tracks counts/errors only; duration aggregation would + // require the End signature to carry it. Counts + error rate are the useful + // opt-in signal and keep the seam unchanged. + h.r.flush(snapshot) +} + +// snapshotLocked builds a serializable copy under the held lock. +func (r *Recorder) snapshotLocked() map[string]opStat { + out := make(map[string]opStat, len(r.ops)) + for k, v := range r.ops { + out[k] = *v + } + return out +} + +func (r *Recorder) flush(snapshot map[string]opStat) { + // Stable key order for a deterministic file (and stable tests). + keys := make([]string, 0, len(snapshot)) + for k := range snapshot { + keys = append(keys, k) + } + sort.Strings(keys) + ordered := make(map[string]opStat, len(snapshot)) + for _, k := range keys { + ordered[k] = snapshot[k] + } + + body, err := json.MarshalIndent(ordered, "", " ") + if err != nil { + return // best-effort: telemetry never fails the operation + } + if mkErr := os.MkdirAll(filepath.Dir(r.path), 0o700); mkErr != nil { + return + } + _ = os.WriteFile(r.path, body, 0o600) +} diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go new file mode 100644 index 0000000..2a17c3d --- /dev/null +++ b/internal/telemetry/telemetry_test.go @@ -0,0 +1,109 @@ +package telemetry + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "testing" + + "github.com/nixrajput/siphon/internal/audit" +) + +func TestRecorder_AggregatesCountsAndErrors(t *testing.T) { + path := filepath.Join(t.TempDir(), "telemetry.json") + r := NewRecorder(path) + ctx := context.Background() + + // 2 backups (1 ok, 1 error), 1 prune ok. + r.Begin(ctx, audit.Event{Op: audit.OpBackup}).End(nil) + r.Begin(ctx, audit.Event{Op: audit.OpBackup}).End(errors.New("boom")) + r.Begin(ctx, audit.Event{Op: audit.OpPrune}).End(nil) + + stats := readStats(t, path) + if stats["backup"].Count != 2 || stats["backup"].Errors != 1 { + t.Errorf("backup stats = %+v, want count=2 errors=1", stats["backup"]) + } + if stats["prune"].Count != 1 || stats["prune"].Errors != 0 { + t.Errorf("prune stats = %+v, want count=1 errors=0", stats["prune"]) + } +} + +func TestRecorder_RecordsOnlyOpNotIdentifyingFields(t *testing.T) { + // Telemetry must not persist profile/actor/target. The on-disk JSON is keyed + // by op only; assert no identifying string leaks into the file. + path := filepath.Join(t.TempDir(), "telemetry.json") + r := NewRecorder(path) + r.Begin(context.Background(), audit.Event{ + Op: audit.OpRestore, Profile: "prod-secret", Actor: "alice", Target: "dump-xyz", + }).End(nil) + + body, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read telemetry: %v", err) + } + for _, leak := range []string{"prod-secret", "alice", "dump-xyz"} { + if contains(string(body), leak) { + t.Errorf("telemetry leaked identifying field %q: %s", leak, body) + } + } +} + +func TestNewRecorder_EmptyPathIsNil(t *testing.T) { + if NewRecorder("") != nil { + t.Error("empty path should yield a nil recorder (disabled)") + } +} + +func TestMulti_FansOutToAllSinks(t *testing.T) { + a := &countingAuditor{} + b := &countingAuditor{} + m := audit.NewMulti(a, nil, b) // nil skipped + if m == nil { + t.Fatal("NewMulti of live auditors returned nil") + } + m.Begin(context.Background(), audit.Event{Op: audit.OpSync}).End(nil) + if a.begins != 1 || a.ends != 1 || b.begins != 1 || b.ends != 1 { + t.Errorf("fan-out = a(%d,%d) b(%d,%d), want each (1,1)", a.begins, a.ends, b.begins, b.ends) + } +} + +func TestMulti_AllNilIsNil(t *testing.T) { + if audit.NewMulti(nil, nil) != nil { + t.Error("NewMulti of only nils should be nil (no-op)") + } +} + +type countingAuditor struct{ begins, ends int } + +func (c *countingAuditor) Begin(_ context.Context, _ audit.Event) audit.Handle { + c.begins++ + return &countingHandle{c: c} +} + +type countingHandle struct{ c *countingAuditor } + +func (h *countingHandle) End(error) { h.c.ends++ } + +func readStats(t *testing.T, path string) map[string]opStat { + t.Helper() + body, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read telemetry: %v", err) + } + var m map[string]opStat + if err := json.Unmarshal(body, &m); err != nil { + t.Fatalf("bad telemetry JSON: %v", err) + } + return m +} + +func contains(s, sub string) bool { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return true + } + } + return false +} From c37397519ed28e5f9ef0d250e02d38d228634a83 Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 02:32:49 +0530 Subject: [PATCH 4/7] feat(schedule): cron-managed recurring backups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace the schedule placeholder with a crontab manager. siphon does NOT run a scheduler daemon — it maintains a delimited, siphon-owned block in the user's crontab that invokes `siphon backup ` on a cron expression, delegating execution to the host's cron. - Add internal/schedule: pure, I/O-free crontab-text editing (Add / List / Remove) that touches only the managed block and preserves the user's own crontab lines verbatim. Add reschedules an existing profile in place rather than duplicating; removing the last entry drops the block entirely. - CLI: `schedule add --cron`, `schedule list`, `schedule remove `, reading/writing the real crontab via `crontab -l` / `crontab -` ("no crontab" is treated as empty, not an error). - Test the pure editor: add/list round-trip, in-place reschedule, preservation of non-managed lines, no-op remove, empty cases. --- internal/cli/root.go | 8 -- internal/cli/schedule.go | 129 ++++++++++++++++++++++++++ internal/schedule/crontab.go | 149 ++++++++++++++++++++++++++++++ internal/schedule/crontab_test.go | 81 ++++++++++++++++ 4 files changed, 359 insertions(+), 8 deletions(-) create mode 100644 internal/cli/schedule.go create mode 100644 internal/schedule/crontab.go create mode 100644 internal/schedule/crontab_test.go diff --git a/internal/cli/root.go b/internal/cli/root.go index 4017459..a00a4fc 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -194,14 +194,6 @@ func buildStore(cfg *config.Config) (storage.Store, error) { return storage.NewLocal(dumpDir) } -func newScheduleCmd() *cobra.Command { - return &cobra.Command{ - Use: "schedule", - Short: "Cron-managed recurring backups (Phase G)", - RunE: func(*cobra.Command, []string) error { return fmt.Errorf("schedule: not implemented (Phase G)") }, - } -} - func newTunnelCmd() *cobra.Command { return &cobra.Command{ Use: "tunnel", diff --git a/internal/cli/schedule.go b/internal/cli/schedule.go new file mode 100644 index 0000000..dffd853 --- /dev/null +++ b/internal/cli/schedule.go @@ -0,0 +1,129 @@ +package cli + +import ( + "bytes" + "fmt" + "os" + "os/exec" + "strings" + + "github.com/spf13/cobra" + + "github.com/nixrajput/siphon/internal/errs" + "github.com/nixrajput/siphon/internal/schedule" +) + +// newScheduleCmd manages a siphon-owned block of recurring-backup entries in the +// user's crontab. siphon does not run a scheduler — the host cron invokes +// `siphon backup ` on the given expression. +func newScheduleCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "schedule", + Short: "Manage cron-scheduled recurring backups", + Long: "schedule maintains a siphon-owned block in your crontab that runs " + + "`siphon backup ` on a cron expression. siphon does not run a " + + "daemon — your system's cron runs the jobs. Requires the `crontab` command.", + } + cmd.AddCommand(scheduleListCmd(), scheduleAddCmd(), scheduleRemoveCmd()) + return cmd +} + +func scheduleListCmd() *cobra.Command { + return &cobra.Command{ + Use: "list", Short: "List scheduled backups", + RunE: func(c *cobra.Command, _ []string) error { + tab, err := readCrontab() + if err != nil { + return err + } + entries := schedule.List(tab) + if len(entries) == 0 { + _, _ = fmt.Fprintln(c.OutOrStdout(), "no scheduled backups") + return nil + } + for _, e := range entries { + _, _ = fmt.Fprintf(c.OutOrStdout(), "%-20s %s\n", e.Profile, e.Cron) + } + return nil + }, + } +} + +func scheduleAddCmd() *cobra.Command { + var cron string + cmd := &cobra.Command{ + Use: "add ", Short: "Schedule (or reschedule) a recurring backup", Args: cobra.ExactArgs(1), + RunE: func(c *cobra.Command, args []string) error { + if strings.TrimSpace(cron) == "" { + return &errs.Error{Op: "schedule.add", Code: errs.CodeUser, Hint: "--cron is required (e.g. --cron \"0 2 * * *\")"} + } + bin, err := os.Executable() + if err != nil { + bin = "siphon" // fall back to PATH lookup at cron time + } + tab, err := readCrontab() + if err != nil { + return err + } + updated := schedule.Add(tab, bin, schedule.Entry{Profile: args[0], Cron: cron}) + if err := writeCrontab(updated); err != nil { + return err + } + _, _ = fmt.Fprintf(c.OutOrStdout(), "scheduled %s: %s\n", args[0], cron) + return nil + }, + } + cmd.Flags().StringVar(&cron, "cron", "", "Cron expression, e.g. \"0 2 * * *\" (daily at 02:00)") + return cmd +} + +func scheduleRemoveCmd() *cobra.Command { + return &cobra.Command{ + Use: "remove ", Short: "Remove a scheduled backup", Args: cobra.ExactArgs(1), + RunE: func(c *cobra.Command, args []string) error { + bin, err := os.Executable() + if err != nil { + bin = "siphon" + } + tab, err := readCrontab() + if err != nil { + return err + } + updated := schedule.Remove(tab, bin, args[0]) + if err := writeCrontab(updated); err != nil { + return err + } + _, _ = fmt.Fprintf(c.OutOrStdout(), "removed schedule for %s\n", args[0]) + return nil + }, + } +} + +// readCrontab returns the current user crontab, or "" when none is installed +// (`crontab -l` exits non-zero with "no crontab" — treated as empty, not error). +func readCrontab() (string, error) { + out, err := exec.Command("crontab", "-l").Output() + if err != nil { + if ee, ok := err.(*exec.ExitError); ok && strings.Contains(strings.ToLower(string(ee.Stderr)), "no crontab") { + return "", nil + } + // An empty crontab also commonly exits 1 with empty output; tolerate it. + if len(out) == 0 { + return "", nil + } + return "", &errs.Error{Op: "schedule", Code: errs.CodeSystem, Cause: err, Hint: "could not read crontab (is the `crontab` command available?)"} + } + return string(out), nil +} + +// writeCrontab installs tab as the user crontab via `crontab -` (reads stdin). +func writeCrontab(tab string) error { + cmd := exec.Command("crontab", "-") + cmd.Stdin = bytes.NewReader([]byte(tab)) + var stderr bytes.Buffer + cmd.Stderr = &stderr + if err := cmd.Run(); err != nil { + return &errs.Error{Op: "schedule", Code: errs.CodeSystem, Cause: err, Hint: "could not write crontab: " + strings.TrimSpace(stderr.String())} + } + return nil +} diff --git a/internal/schedule/crontab.go b/internal/schedule/crontab.go new file mode 100644 index 0000000..8841674 --- /dev/null +++ b/internal/schedule/crontab.go @@ -0,0 +1,149 @@ +// Package schedule manages a siphon-owned block of entries inside the user's +// crontab. siphon does not run a scheduler daemon itself — it delegates to the +// host's cron, writing lines that invoke `siphon backup ` on a cron +// expression. The OS scheduler (already present and battle-tested) runs the +// jobs; siphon only edits its own managed block. +// +// This file is the pure, I/O-free core: it edits crontab TEXT (add/remove/list +// siphon entries within delimited markers) so it is fully unit-testable. The CLI +// layer reads the current crontab, calls these functions, and writes it back. +package schedule + +import ( + "fmt" + "sort" + "strings" +) + +const ( + beginMarker = "# >>> siphon managed (do not edit this block) >>>" + endMarker = "# <<< siphon managed <<<" + linePrefix = "# siphon:" // metadata comment preceding each managed job line +) + +// Entry is one scheduled backup: a cron expression and the profile to back up. +type Entry struct { + Profile string + Cron string // standard 5-field cron expression +} + +// Render produces the crontab line(s) for an entry: a metadata comment (so List +// can recover the profile) followed by the cron line invoking siphon. bin is the +// siphon binary path to invoke. +func (e Entry) render(bin string) string { + return fmt.Sprintf("%s %s\n%s %s backup %s", + linePrefix, e.Profile, e.Cron, bin, e.Profile) +} + +// List extracts the siphon-managed entries from a crontab's text. Entries +// outside the managed block are ignored. Returns them sorted by profile. +func List(crontab string) []Entry { + block := extractBlock(crontab) + var entries []Entry + lines := strings.Split(block, "\n") + for i := 0; i < len(lines); i++ { + meta := strings.TrimSpace(lines[i]) + if !strings.HasPrefix(meta, linePrefix) { + continue + } + profile := strings.TrimSpace(strings.TrimPrefix(meta, linePrefix)) + // The cron line is the next non-empty line; recover its expression (the + // first 5 fields). + if i+1 < len(lines) { + fields := strings.Fields(lines[i+1]) + if len(fields) >= 5 { + entries = append(entries, Entry{Profile: profile, Cron: strings.Join(fields[:5], " ")}) + } + } + } + sort.Slice(entries, func(i, j int) bool { return entries[i].Profile < entries[j].Profile }) + return entries +} + +// Add returns a new crontab with entry added (or its cron updated if the profile +// already has an entry). Lines outside the managed block are preserved verbatim. +func Add(crontab, bin string, entry Entry) string { + entries := List(crontab) + replaced := false + for i := range entries { + if entries[i].Profile == entry.Profile { + entries[i].Cron = entry.Cron + replaced = true + } + } + if !replaced { + entries = append(entries, entry) + } + return writeBlock(crontab, bin, entries) +} + +// Remove returns a new crontab with the named profile's entry removed. Removing +// a profile that has no entry is a no-op. +func Remove(crontab, bin, profile string) string { + var kept []Entry + for _, e := range List(crontab) { + if e.Profile != profile { + kept = append(kept, e) + } + } + return writeBlock(crontab, bin, kept) +} + +// extractBlock returns the text between the markers (exclusive), or "" if the +// managed block is absent. +func extractBlock(crontab string) string { + b := strings.Index(crontab, beginMarker) + e := strings.Index(crontab, endMarker) + if b < 0 || e < 0 || e < b { + return "" + } + return crontab[b+len(beginMarker) : e] +} + +// writeBlock returns crontab with its managed block replaced by a freshly +// rendered block for entries (sorted by profile). If entries is empty, the +// managed block is removed entirely. Non-managed lines are preserved. +func writeBlock(crontab, bin string, entries []Entry) string { + sort.Slice(entries, func(i, j int) bool { return entries[i].Profile < entries[j].Profile }) + + // Strip any existing managed block (and the surrounding markers + blank line). + outside := stripBlock(crontab) + + if len(entries) == 0 { + return outside + } + + var b strings.Builder + b.WriteString(beginMarker) + b.WriteString("\n") + for _, e := range entries { + b.WriteString(e.render(bin)) + b.WriteString("\n") + } + b.WriteString(endMarker) + b.WriteString("\n") + + if strings.TrimSpace(outside) == "" { + return b.String() + } + return strings.TrimRight(outside, "\n") + "\n" + b.String() +} + +// stripBlock removes the managed block (markers inclusive) from crontab, +// returning the surrounding user content. +func stripBlock(crontab string) string { + b := strings.Index(crontab, beginMarker) + e := strings.Index(crontab, endMarker) + if b < 0 || e < 0 || e < b { + return crontab + } + before := strings.TrimRight(crontab[:b], "\n") + after := strings.TrimLeft(crontab[e+len(endMarker):], "\n") + if before == "" { + return after + } + if after == "" { + return before + "\n" + } + return before + "\n" + after +} diff --git a/internal/schedule/crontab_test.go b/internal/schedule/crontab_test.go new file mode 100644 index 0000000..3123ba9 --- /dev/null +++ b/internal/schedule/crontab_test.go @@ -0,0 +1,81 @@ +package schedule + +import ( + "strings" + "testing" +) + +const bin = "/usr/local/bin/siphon" + +func TestAddListRoundTrip(t *testing.T) { + tab := Add("", bin, Entry{Profile: "prod", Cron: "0 2 * * *"}) + tab = Add(tab, bin, Entry{Profile: "staging", Cron: "30 3 * * *"}) + + entries := List(tab) + if len(entries) != 2 { + t.Fatalf("got %d entries, want 2: %+v", len(entries), entries) + } + // Sorted by profile. + if entries[0].Profile != "prod" || entries[0].Cron != "0 2 * * *" { + t.Errorf("entries[0] = %+v, want prod / 0 2 * * *", entries[0]) + } + if entries[1].Profile != "staging" || entries[1].Cron != "30 3 * * *" { + t.Errorf("entries[1] = %+v, want staging / 30 3 * * *", entries[1]) + } + // The rendered crontab actually invokes siphon backup. + if !strings.Contains(tab, bin+" backup prod") { + t.Errorf("crontab missing backup invocation:\n%s", tab) + } +} + +func TestAddReschedulesExistingProfile(t *testing.T) { + tab := Add("", bin, Entry{Profile: "prod", Cron: "0 2 * * *"}) + tab = Add(tab, bin, Entry{Profile: "prod", Cron: "0 5 * * *"}) // same profile, new time + + entries := List(tab) + if len(entries) != 1 { + t.Fatalf("reschedule created a duplicate: %+v", entries) + } + if entries[0].Cron != "0 5 * * *" { + t.Errorf("cron = %q, want updated 0 5 * * *", entries[0].Cron) + } +} + +func TestPreservesNonManagedLines(t *testing.T) { + user := "# my own job\n0 0 * * * /bin/echo hi\n" + tab := Add(user, bin, Entry{Profile: "prod", Cron: "0 2 * * *"}) + + if !strings.Contains(tab, "/bin/echo hi") { + t.Errorf("user's own cron line was lost:\n%s", tab) + } + if len(List(tab)) != 1 { + t.Errorf("siphon entry not added alongside user line") + } + + // Removing the siphon entry must leave the user's line intact and drop the + // managed block entirely. + tab = Remove(tab, bin, "prod") + if !strings.Contains(tab, "/bin/echo hi") { + t.Errorf("user line lost on remove:\n%s", tab) + } + if strings.Contains(tab, beginMarker) { + t.Errorf("empty managed block was left behind:\n%s", tab) + } +} + +func TestRemoveNonexistentIsNoOp(t *testing.T) { + tab := Add("", bin, Entry{Profile: "prod", Cron: "0 2 * * *"}) + out := Remove(tab, bin, "nope") + if len(List(out)) != 1 { + t.Errorf("removing a nonexistent profile changed the entries: %+v", List(out)) + } +} + +func TestListEmpty(t *testing.T) { + if got := List(""); len(got) != 0 { + t.Errorf("List(empty) = %+v, want none", got) + } + if got := List("# just a user comment\n"); len(got) != 0 { + t.Errorf("List(no managed block) = %+v, want none", got) + } +} From 36ea4557f0eb368ddc265f4a4e6aeae3b90af2d5 Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 02:34:51 +0530 Subject: [PATCH 5/7] feat(tunnel): SSH tunnel helper to a DB via a bastion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace the tunnel placeholder with a foreground SSH local-forward helper. `siphon tunnel ` runs `ssh -N -L :: ` via the system ssh client (the user's ssh config, keys, and agent apply) and holds it open until Ctrl-C — delegation, not an SSH reimplementation or a daemon. - Add ProfileConfig.tunnel (bastion + optional local_port, defaulting to the DB port). A profile without tunnel.bastion gives a clear CodeUser error. - Use ExitOnForwardFailure=yes so ssh fails fast if the local bind is taken rather than opening a useless session; run under the command context so cancellation tears the tunnel down. - Update the stale stub test (schedule/tunnel are no longer "not implemented"): assert bare `schedule` shows its subcommands. - Test the pure ssh arg builder. --- internal/cli/root.go | 8 ---- internal/cli/root_test.go | 17 +++++---- internal/cli/tunnel.go | 74 +++++++++++++++++++++++++++++++++++++ internal/cli/tunnel_test.go | 21 +++++++++++ internal/config/config.go | 11 ++++++ 5 files changed, 116 insertions(+), 15 deletions(-) create mode 100644 internal/cli/tunnel.go create mode 100644 internal/cli/tunnel_test.go diff --git a/internal/cli/root.go b/internal/cli/root.go index a00a4fc..ef95ad6 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -194,14 +194,6 @@ func buildStore(cfg *config.Config) (storage.Store, error) { return storage.NewLocal(dumpDir) } -func newTunnelCmd() *cobra.Command { - return &cobra.Command{ - Use: "tunnel", - Short: "SSH tunnel helper (Phase G)", - RunE: func(*cobra.Command, []string) error { return fmt.Errorf("tunnel: not implemented (Phase G)") }, - } -} - // Execute runs the root command using stdout/stderr and returns the POSIX // exit code. It is the only function main() calls. // diff --git a/internal/cli/root_test.go b/internal/cli/root_test.go index 86da36a..78c866f 100644 --- a/internal/cli/root_test.go +++ b/internal/cli/root_test.go @@ -59,16 +59,19 @@ func TestExecute_ErrsErrorExitCode_RoutesThroughCode(t *testing.T) { } } -func TestRoot_StubSubcommand_ReturnsNotImplementedError(t *testing.T) { +// schedule and tunnel are implemented (Phase G ops cycle). Bare `schedule` +// is a parent command that shows help; `tunnel ` errors clearly when +// the profile has no bastion configured. +func TestRoot_ScheduleShowsSubcommands(t *testing.T) { var out, errb bytes.Buffer root := NewRoot(&out, &errb) root.SetArgs([]string{"schedule"}) - - err := root.Execute() - if err == nil { - t.Fatal("Execute(schedule) returned nil; want 'not implemented' error") + if err := root.Execute(); err != nil { + t.Fatalf("Execute(schedule) = %v; want nil (help)", err) } - if !strings.Contains(err.Error(), "not implemented") { - t.Fatalf("error = %q; want it to contain 'not implemented'", err.Error()) + for _, sub := range []string{"add", "list", "remove"} { + if !strings.Contains(out.String(), sub) { + t.Errorf("schedule help missing subcommand %q:\n%s", sub, out.String()) + } } } diff --git a/internal/cli/tunnel.go b/internal/cli/tunnel.go new file mode 100644 index 0000000..d59f93a --- /dev/null +++ b/internal/cli/tunnel.go @@ -0,0 +1,74 @@ +package cli + +import ( + "fmt" + "os" + "os/exec" + "strconv" + + "github.com/spf13/cobra" + + "github.com/nixrajput/siphon/internal/config" + "github.com/nixrajput/siphon/internal/errs" +) + +// newTunnelCmd opens an SSH local-forward to a profile's database through its +// configured bastion, using the system ssh client. It runs in the foreground +// and holds the tunnel open until interrupted (Ctrl-C) — run it in one terminal +// and point siphon (or any client) at the printed local address in another. +func newTunnelCmd() *cobra.Command { + return &cobra.Command{ + Use: "tunnel ", + Short: "Open an SSH tunnel to a profile's database via its bastion", + Long: "tunnel opens `ssh -L :: ` using your " + + "system ssh client (your ssh config, keys, and agent apply) and holds it " + + "open until you press Ctrl-C. Configure a profile's `tunnel.bastion` first.", + Args: cobra.ExactArgs(1), + RunE: func(c *cobra.Command, args []string) error { + cfg, err := config.Load() + if err != nil { + return err + } + prof, ok := cfg.Profiles[args[0]] + if !ok { + return &errs.Error{Op: "tunnel", Code: errs.CodeUser, Cause: errs.ErrProfileNotFound, Hint: "unknown profile " + args[0]} + } + if prof.Tunnel == nil || prof.Tunnel.Bastion == "" { + return &errs.Error{Op: "tunnel", Code: errs.CodeUser, Hint: "profile " + args[0] + " has no tunnel.bastion configured"} + } + + localPort := prof.Tunnel.LocalPort + if localPort == 0 { + localPort = prof.Port + } + sshArgs := tunnelArgs(localPort, prof.Host, prof.Port, prof.Tunnel.Bastion) + + _, _ = fmt.Fprintf(c.OutOrStdout(), + "tunnel open: localhost:%d → %s:%d via %s (Ctrl-C to close)\n", + localPort, prof.Host, prof.Port, prof.Tunnel.Bastion) + + cmd := exec.CommandContext(c.Context(), "ssh", sshArgs...) + cmd.Stdin, cmd.Stdout, cmd.Stderr = os.Stdin, c.OutOrStdout(), c.ErrOrStderr() + if err := cmd.Run(); err != nil { + // A clean Ctrl-C (ssh exits non-zero on signal) is the normal stop, + // not an error worth a non-zero siphon exit beyond ssh's own. + return &errs.Error{Op: "tunnel", Code: errs.CodeSystem, Cause: err, Hint: "ssh tunnel exited"} + } + return nil + }, + } +} + +// tunnelArgs builds the `ssh` argument list for a local forward. Pure and +// testable: -N (no remote command), -L localPort:dbHost:dbPort, then the +// bastion. ExitOnForwardFailure makes ssh fail fast if the local bind fails +// rather than silently opening a useless session. +func tunnelArgs(localPort int, dbHost string, dbPort int, bastion string) []string { + forward := strconv.Itoa(localPort) + ":" + dbHost + ":" + strconv.Itoa(dbPort) + return []string{ + "-N", + "-o", "ExitOnForwardFailure=yes", + "-L", forward, + bastion, + } +} diff --git a/internal/cli/tunnel_test.go b/internal/cli/tunnel_test.go new file mode 100644 index 0000000..271cc22 --- /dev/null +++ b/internal/cli/tunnel_test.go @@ -0,0 +1,21 @@ +package cli + +import "testing" + +func TestTunnelArgs(t *testing.T) { + got := tunnelArgs(15432, "db.internal", 5432, "jump@bastion.example.com") + want := []string{ + "-N", + "-o", "ExitOnForwardFailure=yes", + "-L", "15432:db.internal:5432", + "jump@bastion.example.com", + } + if len(got) != len(want) { + t.Fatalf("args = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("args[%d] = %q, want %q", i, got[i], want[i]) + } + } +} diff --git a/internal/config/config.go b/internal/config/config.go index ca1123f..5e5465e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -135,6 +135,17 @@ type ProfileConfig struct { SSLMode string `yaml:"sslmode"` Group string `yaml:"group"` Retention *RetentionConfig `yaml:"retention,omitempty"` // overrides Defaults.Retention wholesale + Tunnel *TunnelConfig `yaml:"tunnel,omitempty"` // optional SSH bastion for reaching this DB +} + +// TunnelConfig describes an SSH bastion through which this profile's database is +// reached. `siphon tunnel ` opens an `ssh -L` local forward from +// LocalPort to the profile's Host:Port via Bastion, using the system ssh client +// (so the user's ssh config, keys, and agent apply). It is delegation, not a +// reimplementation of SSH. +type TunnelConfig struct { + Bastion string `yaml:"bastion"` // [user@]host[:port] of the SSH jump host + LocalPort int `yaml:"local_port,omitempty"` // local forward port (defaults to the DB port) } type GroupConfig struct { From 002e11d5d8e6521f9f3c4bcc8c91e0bcd5149727 Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 02:36:42 +0530 Subject: [PATCH 6/7] docs: document the Phase G ops suite - Add docs/OPS.md covering audit log, 2FA/group gating, telemetry, schedule, and tunnel (config + behavior for each). - Update README roadmap row and command table; flip schedule/tunnel from "(Phase G)" placeholders to their real subcommands/usage. - Add the Phase G ops-suite entry to CHANGELOG. --- CHANGELOG.md | 7 ++++ README.md | 6 +-- docs/OPS.md | 116 +++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 126 insertions(+), 3 deletions(-) create mode 100644 docs/OPS.md diff --git a/CHANGELOG.md b/CHANGELOG.md index b4c757b..252b365 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Phase G (ops) — **operational suite** (audit, 2FA gating, telemetry, schedule, tunnel): + - **Audit log** (`internal/audit`): an append-only JSONL trail of destructive ops (backup/restore/sync/prune) recording who/what/when/outcome, via a single `guardedOp` interception seam wired into the verbs. Off by default; enable with the `audit:` config block. The same seam is reused by 2FA gating and telemetry (no per-feature re-wrapping). + - **2FA / group gating** (`internal/twofactor`): a profile's group can require a typed-name confirmation (`confirm_destructive`) and/or a TOTP code (`require_2fa`) before a destructive op runs. Stdlib RFC 6238 TOTP (HMAC-SHA1, 30s, 6 digits, ±1-step skew), no new dependency; the group's `totp_secret` is a secret-ref. `require_2fa` with no resolvable secret fails closed. + - **Telemetry** (`internal/telemetry`): opt-in aggregate per-op counts and error tallies, flushed as JSON. Records the op name and outcome only — never profile, actor, target, or data. Off by default (`telemetry:` config block); composed onto the audit seam via `audit.Multi`. + - **`siphon schedule`**: manage cron-scheduled recurring backups. siphon maintains a delimited, siphon-owned block in the user's crontab that invokes `siphon backup ` — the host cron runs the jobs (no daemon). `schedule add --cron`, `list`, `remove`. + - **`siphon tunnel `**: open a foreground `ssh -L` local-forward to a profile's database through a configured `tunnel.bastion`, using the system ssh client; held open until Ctrl-C. + - Phase G (ops) — **retention & lifecycle** (second G cycle): - Chain-aware retention engine in `internal/dumps` (`GroupChains` + `Plan`, pure and DB/clock-free): the catalog is grouped into restorable **chains** (a base plus its incrementals), and retention keeps or prunes each chain as a unit, so an incremental is never orphaned from its base. A chain's age is its newest member, so an actively-appended chain is never pruned mid-life. - Three composable policy rules with **union** semantics (a chain is kept if it satisfies any active rule, so adding a rule only ever protects more): `keep_last: N`, `max_age: `, and grandfather-father-son `gfs: {daily, weekly, monthly}`. An all-zero/omitted policy keeps everything — prune is a no-op unless a rule is explicitly configured. diff --git a/README.md b/README.md index 29027dc..ce9aaba 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_ | **D** — Driver hardening | Shared cross-driver test harness (`RunDriverSuite`), capability-gating helper (`RequireCapability`), Postgres connection-probe retry, and a `docs/DRIVERS.md` contributor guide | ✅ Complete | | **E** — MySQL + MariaDB | Both drivers via a shared `_mysqlcommon` package (shared `Conn`, `mysqldump`/`mariadb-dump` backup, client-pipe restore), exercised by the Phase D `RunDriverSuite` harness | ✅ Complete | | **F** — Advanced transfer | All four advanced-transfer modes work end-to-end: bounded-buffer streaming sync; **incremental** backup/restore (`backup --incremental --base ` captures a bounded change set via Postgres logical decoding / MySQL-MariaDB binlog, `restore` replays the base→incremental chain, Postgres orphan-slot sweep); **cross-engine** sync (`sync --cross-engine` — typed `SchemaInspector` introspection → canonical type-mapping, e.g. Postgres → MySQL); and **CDC** (`siphon cdc` / `sync --continuous` — unbounded change streaming with snapshot→stream handoff, resumable, same- and cross-engine). Live DB paths are integration-tested in CI — see [docs/INCREMENTAL.md](docs/INCREMENTAL.md), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md), [docs/CDC.md](docs/CDC.md) | ✅ Complete | -| **G** — Ops features | **Cloud storage** (✅): the dump catalog can live in an S3 / S3-compatible bucket via a pluggable `storage.Store` backend (local + S3; GCS/Azure are a fast-follow), `storage:` config block, SHA-256 integrity end-to-end — see [docs/STORAGE.md](docs/STORAGE.md). **Retention** (✅): chain-aware pruning (`siphon prune`) with keep-last-N / max-age / GFS rules, per-profile `retention:` config, dry-run by default — never orphans an incremental's base; see [docs/RETENTION.md](docs/RETENTION.md). Secret backends, profile groups + 2FA, team mode, audit log, and telemetry follow in later G cycles. | 🟡 In progress | +| **G** — Ops features | **Cloud storage** (✅): the dump catalog can live in an S3 / S3-compatible bucket via a pluggable `storage.Store` backend (local + S3; GCS/Azure are a fast-follow), `storage:` config block, SHA-256 integrity end-to-end — see [docs/STORAGE.md](docs/STORAGE.md). **Retention** (✅): chain-aware pruning (`siphon dumps prune`) with keep-last-N / max-age / GFS rules, per-profile `retention:` config; see [docs/RETENTION.md](docs/RETENTION.md). **Ops suite** (✅): an append-only **audit log** of destructive ops (`audit:` config), **2FA/group gating** (a profile group can require a typed confirmation and/or TOTP before destructive ops), opt-in aggregate **telemetry** (`telemetry:` config), **`siphon schedule`** (manages recurring backups in your crontab), and **`siphon tunnel`** (SSH local-forward to a DB via a bastion). Multi-backend secrets are the remaining G item; see [docs/OPS.md](docs/OPS.md). | 🟡 In progress | | **H** — Distribution | GoReleaser, Homebrew tap, Scoop bucket, install script, docs site | ⏳ Planned | ## Requirements @@ -134,8 +134,8 @@ Exit codes follow a POSIX-friendly taxonomy (`0` ok, `1` user error, `2` system | `siphon dumps list\|inspect\|prune` | List, inspect, or prune saved dumps | | `siphon profile add\|list\|show\|rm` | Manage named connection profiles | | `siphon config path\|edit` | Show or edit the config file | -| `siphon schedule` | Cron-managed recurring backups _(Phase G)_ | -| `siphon tunnel` | SSH tunnel helper _(Phase G)_ | +| `siphon schedule add\|list\|remove` | Manage cron-scheduled recurring backups | +| `siphon tunnel ` | Open an SSH tunnel to a DB via its bastion | | `siphon` _(bare)_ | Launch the interactive multi-panel dashboard | Run `siphon --help` for full flags. diff --git a/docs/OPS.md b/docs/OPS.md new file mode 100644 index 0000000..9fc52d0 --- /dev/null +++ b/docs/OPS.md @@ -0,0 +1,116 @@ +# Operational features + +Phase G's ops suite adds five operational capabilities: an audit log, 2FA / +group gating, opt-in telemetry, scheduled backups, and an SSH tunnel helper. +The first three share one interception point in the app layer (`guardedOp`), +which wraps the destructive verbs (backup, restore, sync, prune) exactly once. + +## Table of contents + +- [Audit log](#audit-log) +- [2FA & group gating](#2fa--group-gating) +- [Telemetry](#telemetry) +- [Scheduled backups](#scheduled-backups) +- [SSH tunnel](#ssh-tunnel) + +## Audit log + +An append-only JSONL trail of destructive operations — who ran what, against +which profile, when, and the outcome. Off by default. + +```yaml +audit: + enabled: true + path: ~/.local/state/siphon/audit.log # optional; this is the default +``` + +Each line records `time`, `op`, `profile`, `target`, `actor` (OS user), +`outcome` (`ok`/`error`), `error`, and `duration_ms`. Audit writes are +best-effort: a failure to write the log never fails the operation it records. + +## 2FA & group gating + +A profile belongs to a **group**; a group can require a second deliberate step +before any destructive op on its profiles: + +```yaml +groups: + critical: + confirm_destructive: true # operator must retype the profile name + require_2fa: true # operator must enter a current TOTP code + totp_secret: env:SIPHON_PROD_TOTP # base32 RFC-6238 secret (a secret-ref) + +profiles: + prod: + driver: postgres + group: critical + # ... +``` + +`confirm_destructive` prompts the operator to retype the profile name; +`require_2fa` prompts for a 6-digit TOTP verified (with ±1 step skew) against the +group's `totp_secret` — the same code your authenticator app shows. The check +runs **before** the operation, so a failed confirmation aborts before any +destructive work. `require_2fa` with no resolvable secret fails closed. The TOTP +secret is a secret-ref, so the plaintext never lives in config. + +This is offline by design — siphon is a local CLI, so "2FA" means a TOTP +(standard, no network) rather than a push notification. + +## Telemetry + +Opt-in aggregate operational metrics: per-op counts and error tallies, flushed +as JSON. Off by default. + +```yaml +telemetry: + enabled: true + path: ~/.local/state/siphon/telemetry.json # optional; this is the default +``` + +Telemetry records **only** the operation name and outcome — never profile names, +hosts, dump IDs, the actor, or any data. It is composed onto the audit seam, so +enabling it adds no new interception in the verbs. + +## Scheduled backups + +`siphon schedule` manages recurring backups by maintaining a delimited, +siphon-owned block in your **crontab** — siphon does not run a scheduler daemon; +your system's cron invokes `siphon backup ` on the schedule. + +```bash +siphon schedule add prod --cron "0 2 * * *" # nightly at 02:00 +siphon schedule list +siphon schedule remove prod +``` + +Entries outside the siphon-managed block are preserved. Re-adding a profile +updates its schedule in place; removing the last entry drops the managed block. +Requires the `crontab` command. + +## SSH tunnel + +`siphon tunnel ` opens an SSH local-forward to a profile's database +through a configured bastion, using your **system ssh client** (your ssh config, +keys, and agent all apply). It runs in the foreground and holds the tunnel open +until you press Ctrl-C. + +```yaml +profiles: + prod: + driver: postgres + host: db.internal + port: 5432 + tunnel: + bastion: jump@bastion.example.com + local_port: 15432 # optional; defaults to the DB port +``` + +```bash +siphon tunnel prod +# tunnel open: localhost:15432 → db.internal:5432 via jump@bastion.example.com (Ctrl-C to close) +``` + +Run it in one terminal and point a client (or another siphon command) at the +printed local address in another. siphon delegates to `ssh -L` rather than +reimplementing SSH or holding a connection in a daemon. From 4fcb9f78bae38c00d754d6986fe487eeb8a12e5d Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 03:06:46 +0530 Subject: [PATCH 7/7] fix(ops): apply CodeRabbit review fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Apply post-review fixes across the ops suite. - sync: guard ALL non-CDC sync variants — the guardedOp call moved above the cross-engine branch, and RunCDC now guards itself, so 2FA/audit cover native, cross-engine, and continuous sync (previously only native). A security gap: --cross-engine bypassed 2FA. - crontab: shell-quote the rendered command's bin path and profile so a space or metacharacter in either cannot break or inject into the cron line cron runs via sh -c. Regression test with an injection payload. - gate: use one shared buffered reader across both prompts so scripted "name\ncode\n" is not split; read the TOTP code with echo disabled on a terminal (term.ReadPassword), falling back to the reader otherwise. - audit: finalize the record if job launch fails synchronously (launchGuarded) so an authorized-but-unstarted op is not left open. - schedule: reject non-5-field --cron up front (it would install but then vanish from list/remove); narrow readCrontab so only the known empty-crontab exit is treated as empty, not any zero-output failure. - tunnel: validate ports before building the forward; treat a context-cancelled (Ctrl-C) ssh exit as success, not a failure. - telemetry: flush under the lock so concurrent Ends keep the on-disk JSON monotonic. - docs: note that scheduled backups can't run gated profiles unattended. --- docs/OPS.md | 6 +++ go.mod | 1 + internal/app/backup.go | 2 +- internal/app/cdc.go | 13 +++++- internal/app/ops.go | 14 ++++++ internal/app/restore.go | 2 +- internal/app/sync.go | 73 ++++++++++++++++++------------- internal/cli/gate.go | 34 +++++++++++--- internal/cli/gate_test.go | 14 ++++++ internal/cli/schedule.go | 23 +++++++--- internal/cli/tunnel.go | 19 +++++++- internal/schedule/crontab.go | 11 ++++- internal/schedule/crontab_test.go | 20 ++++++++- internal/telemetry/telemetry.go | 11 +++-- 14 files changed, 188 insertions(+), 55 deletions(-) diff --git a/docs/OPS.md b/docs/OPS.md index 9fc52d0..4f6171f 100644 --- a/docs/OPS.md +++ b/docs/OPS.md @@ -88,6 +88,12 @@ Entries outside the siphon-managed block are preserved. Re-adding a profile updates its schedule in place; removing the last entry drops the managed block. Requires the `crontab` command. +> **Gating caveat:** a scheduled job runs `siphon backup ` non-interactively +> under cron. If the profile's group sets `confirm_destructive` or `require_2fa`, +> that backup will block waiting for input it can never receive and the cron job +> will fail. Don't schedule backups for gated profiles (a non-interactive bypass +> for trusted automation is a future enhancement). + ## SSH tunnel `siphon tunnel ` opens an SSH local-forward to a profile's database diff --git a/go.mod b/go.mod index 3644eab..c495692 100644 --- a/go.mod +++ b/go.mod @@ -25,6 +25,7 @@ require ( github.com/testcontainers/testcontainers-go/modules/minio v0.42.0 github.com/testcontainers/testcontainers-go/modules/mysql v0.42.0 github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0 + golang.org/x/term v0.40.0 gopkg.in/yaml.v3 v3.0.1 ) diff --git a/internal/app/backup.go b/internal/app/backup.go index e218992..570b5c9 100644 --- a/internal/app/backup.go +++ b/internal/app/backup.go @@ -83,7 +83,7 @@ func Backup(parent context.Context, d Deps, opt BackupOpts) (<-chan jobs.Event, return nil, "", err } - return d.Runner.Run(parent, jobs.Job{ + return launchGuarded(d.Runner, parent, done, jobs.Job{ Stage: "backup", Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { defer func() { done(retErr) }() diff --git a/internal/app/cdc.go b/internal/app/cdc.go index 47e4ce2..35e0905 100644 --- a/internal/app/cdc.go +++ b/internal/app/cdc.go @@ -10,6 +10,7 @@ import ( "path/filepath" "time" + "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/canonical" "github.com/nixrajput/siphon/internal/driver" "github.com/nixrajput/siphon/internal/errs" @@ -136,11 +137,19 @@ func RunCDC(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, st return nil, "", err } + // CDC is destructive (it continuously applies changes to the target), so it + // runs through the same gate/audit seam as the other verbs. + done, err := guardedOp(parent, d, audit.OpSync, opt.From, opt.To) + if err != nil { + return nil, "", err + } + jobID := cdcJobID(opt.From, opt.To) - return d.Runner.Run(parent, jobs.Job{ + return launchGuarded(d.Runner, parent, done, jobs.Job{ Stage: "cdc", - Func: func(ctx context.Context, emit func(jobs.Event)) error { + Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { + defer func() { done(retErr) }() emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "CDC mode started"}) srcConn, err := srcDrv.Connect(ctx, src) diff --git a/internal/app/ops.go b/internal/app/ops.go index 65c227f..e48690c 100644 --- a/internal/app/ops.go +++ b/internal/app/ops.go @@ -4,6 +4,7 @@ import ( "context" "github.com/nixrajput/siphon/internal/audit" + "github.com/nixrajput/siphon/internal/jobs" ) // guardedOp is the single interception point wrapping every destructive verb. @@ -29,3 +30,16 @@ func guardedOp(ctx context.Context, d Deps, op audit.Op, profile, target string) }) return rec, nil } + +// launchGuarded runs a job whose Func already defers done(retErr) on completion, +// and finalizes the audit record if the launch ITSELF fails synchronously +// (Runner.Run cannot today, but the signature allows it, and an open record must +// not leak). On a launch error it calls done and returns the error. +func launchGuarded(r *jobs.Runner, parent context.Context, done func(error), j jobs.Job) (<-chan jobs.Event, string, error) { + ch, id, err := r.Run(parent, j) + if err != nil { + done(err) + return nil, "", err + } + return ch, id, nil +} diff --git a/internal/app/restore.go b/internal/app/restore.go index b420667..e03e9f2 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -53,7 +53,7 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event return nil, "", err } - return d.Runner.Run(parent, jobs.Job{ + return launchGuarded(d.Runner, parent, done, jobs.Job{ Stage: "restore", Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { defer func() { done(retErr) }() diff --git a/internal/app/sync.go b/internal/app/sync.go index a991591..7742a23 100644 --- a/internal/app/sync.go +++ b/internal/app/sync.go @@ -35,6 +35,9 @@ type SyncOpts struct { // is capability-gated and uses driver.SchemaInspector + driver.CanonicalTransfer // for typed cross-engine snapshot transfer. func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, string, error) { + // CDC is a standalone verb (also invoked directly by the cdc command); it + // runs the guard itself, so route to it before guarding here to avoid a + // double gate/audit. if opt.Continuous { return RunCDC(parent, d, opt) } @@ -56,16 +59,18 @@ func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, stri return nil, "", err } - if opt.CrossEngine { - return runCrossEngineSync(parent, d, opt) - } - + // Guard BEFORE routing to the cross-engine path, so every non-CDC sync + // variant (native and cross-engine — both destructive) is gated and audited. done, err := guardedOp(parent, d, audit.OpSync, opt.From, opt.To) if err != nil { return nil, "", err } - return d.Runner.Run(parent, jobs.Job{ + if opt.CrossEngine { + return runCrossEngineSync(parent, d, opt, done) + } + + return launchGuarded(d.Runner, parent, done, jobs.Job{ Stage: "sync", Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { defer func() { done(retErr) }() @@ -110,39 +115,47 @@ func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, stri }) } -// runCrossEngineSync handles heterogeneous sync (e.g. postgres→mysql) by routing -// through the canonical-schema machinery: InspectSchema on the source builds a -// CanonicalSchema, EmitCanonical streams it as JSONL through a jobs.Stream, and -// ConsumeCanonical on the target replays the rows. Both ends must implement -// driver.SchemaInspector and driver.CanonicalTransfer. -func runCrossEngineSync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, string, error) { - if err := RequireCapability(d, opt.From, CapCrossEngineSource); err != nil { - return nil, "", err +// resolveCrossEngine checks cross-engine capability and resolves both profiles +// and their drivers. Bundled so runCrossEngineSync has a single launch-failure +// path (which finalizes the audit record once). +func resolveCrossEngine(d Deps, opt SyncOpts) (src, dst driver.Profile, srcDrv, dstDrv driver.Driver, err error) { + if err = RequireCapability(d, opt.From, CapCrossEngineSource); err != nil { + return } - if err := RequireCapability(d, opt.To, CapCrossEngineTarget); err != nil { - return nil, "", err + if err = RequireCapability(d, opt.To, CapCrossEngineTarget); err != nil { + return } - - src, err := d.Profiles.Resolve(opt.From) - if err != nil { - return nil, "", err + if src, err = d.Profiles.Resolve(opt.From); err != nil { + return } - dst, err := d.Profiles.Resolve(opt.To) - if err != nil { - return nil, "", err + if dst, err = d.Profiles.Resolve(opt.To); err != nil { + return } - srcDrv, err := d.Drivers.Get(src.Driver) - if err != nil { - return nil, "", err + if srcDrv, err = d.Drivers.Get(src.Driver); err != nil { + return } - dstDrv, err := d.Drivers.Get(dst.Driver) - if err != nil { - return nil, "", err + dstDrv, err = d.Drivers.Get(dst.Driver) + return +} + +// runCrossEngineSync handles heterogeneous sync (e.g. postgres→mysql) via the +// canonical-schema machinery: InspectSchema builds a CanonicalSchema, +// EmitCanonical streams it as JSONL, ConsumeCanonical replays it on the target. +// Both ends must implement SchemaInspector and CanonicalTransfer. The audit +// record opened by the caller's guard is finalized via done. +func runCrossEngineSync(parent context.Context, d Deps, opt SyncOpts, done func(error)) (<-chan jobs.Event, string, error) { + // Resolve everything needed to launch; any failure here finalizes the audit + // record (the op was authorized but could not start) and aborts. + src, dst, srcDrv, dstDrv, launchErr := resolveCrossEngine(d, opt) + if launchErr != nil { + done(launchErr) + return nil, "", launchErr } - return d.Runner.Run(parent, jobs.Job{ + return launchGuarded(d.Runner, parent, done, jobs.Job{ Stage: "sync.cross-engine", - Func: func(ctx context.Context, emit func(jobs.Event)) error { + Func: func(ctx context.Context, emit func(jobs.Event)) (retErr error) { + defer func() { done(retErr) }() srcConn, err := srcDrv.Connect(ctx, src) if err != nil { return err diff --git a/internal/cli/gate.go b/internal/cli/gate.go index 0ed316d..83cbb3c 100644 --- a/internal/cli/gate.go +++ b/internal/cli/gate.go @@ -5,9 +5,12 @@ import ( "context" "fmt" "io" + "os" "strings" "time" + "golang.org/x/term" + "github.com/nixrajput/siphon/internal/app" "github.com/nixrajput/siphon/internal/audit" "github.com/nixrajput/siphon/internal/config" @@ -26,15 +29,21 @@ import ( type promptGate struct { cfg *config.Config res *secrets.Resolver - in io.Reader + in *bufio.Reader // single shared reader so scripted "name\ncode\n" isn't split + fd int // stdin fd for no-echo TOTP; -1 when stdin is not a terminal out io.Writer now func() time.Time // TOTP clock; defaults to time.Now } // newPromptGate builds a gate from config + a secret resolver, prompting on in -// and reporting on out. +// and reporting on out. When in is an *os.File on a terminal, the TOTP code is +// read with echo disabled; otherwise (tests, pipes) it falls back to the reader. func newPromptGate(cfg *config.Config, res *secrets.Resolver, in io.Reader, out io.Writer) *promptGate { - return &promptGate{cfg: cfg, res: res, in: in, out: out, now: time.Now} + fd := -1 + if f, ok := in.(*os.File); ok && term.IsTerminal(int(f.Fd())) { + fd = int(f.Fd()) + } + return &promptGate{cfg: cfg, res: res, in: bufio.NewReader(in), fd: fd, out: out, now: time.Now} } func (g *promptGate) Authorize(_ context.Context, op audit.Op, profile string) error { @@ -67,7 +76,7 @@ func (g *promptGate) groupFor(profile string) (config.GroupConfig, bool) { func (g *promptGate) confirmName(op audit.Op, profile string) error { _, _ = fmt.Fprintf(g.out, "%s on %q is destructive. Type the profile name to confirm: ", op, profile) - line, _ := bufio.NewReader(g.in).ReadString('\n') + line, _ := g.in.ReadString('\n') if strings.TrimSpace(line) != profile { return &errs.Error{Op: "gate", Code: errs.CodeUser, Hint: "confirmation did not match the profile name; aborted"} } @@ -80,7 +89,10 @@ func (g *promptGate) confirmTOTP(op audit.Op, profile, secretRef string) error { return &errs.Error{Op: "gate", Code: errs.CodeUser, Hint: "group requires 2FA but its totp_secret is unset or unresolvable"} } _, _ = fmt.Fprintf(g.out, "%s on %q requires 2FA. Enter your 6-digit code: ", op, profile) - line, _ := bufio.NewReader(g.in).ReadString('\n') + line, err := g.readSecretLine() + if err != nil { + return &errs.Error{Op: "gate", Code: errs.CodeUser, Cause: err, Hint: "could not read 2FA code"} + } ok, err := twofactor.Verify(secret, line, g.now()) if err != nil { return &errs.Error{Op: "gate", Code: errs.CodeUser, Cause: err, Hint: "invalid TOTP secret configured for this group"} @@ -91,5 +103,17 @@ func (g *promptGate) confirmTOTP(op audit.Op, profile, secretRef string) error { return nil } +// readSecretLine reads the 2FA code. On a terminal it reads with echo disabled +// (so the code is not shown on screen, preserving the second factor); otherwise +// — tests, pipes — it reads a line from the shared buffered reader. +func (g *promptGate) readSecretLine() (string, error) { + if g.fd >= 0 { + b, err := term.ReadPassword(g.fd) + _, _ = fmt.Fprintln(g.out) // ReadPassword leaves the cursor on the prompt line + return string(b), err + } + return g.in.ReadString('\n') +} + // compile-time assertion that promptGate satisfies the app.Gate seam. var _ app.Gate = (*promptGate)(nil) diff --git a/internal/cli/gate_test.go b/internal/cli/gate_test.go index 03b1b45..22b3426 100644 --- a/internal/cli/gate_test.go +++ b/internal/cli/gate_test.go @@ -64,6 +64,20 @@ func TestGate_Require2FA(t *testing.T) { } } +func TestGate_ConfirmAndRequire2FA_SharedReader(t *testing.T) { + // Both gates enabled: the name and the TOTP code arrive on one piped stdin + // ("prod\n\n"). A single shared reader must not swallow the code while + // reading the name (the bug a per-prompt reader would cause). + secret := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString([]byte("12345678901234567890")) + cfg := gateCfg(config.GroupConfig{ConfirmDestructive: true, Require2FA: true, TOTPSecret: secret}) + now := time.Unix(20_000, 0) + code := totpAt(t, secret, now) + + if err := newGate(cfg, "prod\n"+code+"\n", now).Authorize(context.Background(), audit.OpRestore, "prod"); err != nil { + t.Errorf("scripted name+code rejected (reader split the input?): %v", err) + } +} + func TestGate_Require2FA_MissingSecret(t *testing.T) { cfg := gateCfg(config.GroupConfig{Require2FA: true}) // no TOTPSecret if err := newGate(cfg, "123456\n", time.Now()).Authorize(context.Background(), audit.OpSync, "prod"); err == nil { diff --git a/internal/cli/schedule.go b/internal/cli/schedule.go index dffd853..58eb7de 100644 --- a/internal/cli/schedule.go +++ b/internal/cli/schedule.go @@ -2,6 +2,7 @@ package cli import ( "bytes" + "errors" "fmt" "os" "os/exec" @@ -57,6 +58,12 @@ func scheduleAddCmd() *cobra.Command { if strings.TrimSpace(cron) == "" { return &errs.Error{Op: "schedule.add", Code: errs.CodeUser, Hint: "--cron is required (e.g. --cron \"0 2 * * *\")"} } + // The managed-block parser round-trips standard 5-field expressions + // only; reject @daily / 6-field specs up front so an entry can't be + // installed and then vanish from list/remove. + if len(strings.Fields(cron)) != 5 { + return &errs.Error{Op: "schedule.add", Code: errs.CodeUser, Hint: "--cron must be a standard 5-field cron expression (e.g. \"0 2 * * *\")"} + } bin, err := os.Executable() if err != nil { bin = "siphon" // fall back to PATH lookup at cron time @@ -104,12 +111,16 @@ func scheduleRemoveCmd() *cobra.Command { func readCrontab() (string, error) { out, err := exec.Command("crontab", "-l").Output() if err != nil { - if ee, ok := err.(*exec.ExitError); ok && strings.Contains(strings.ToLower(string(ee.Stderr)), "no crontab") { - return "", nil - } - // An empty crontab also commonly exits 1 with empty output; tolerate it. - if len(out) == 0 { - return "", nil + // Only the known "no crontab for user" path means empty. An ExitError + // with code 1 and no stdout is cron's empty-table signal on most + // platforms; any other failure (missing binary, permission) propagates so + // `schedule list` never reports an empty schedule that masks a real error. + var ee *exec.ExitError + if errors.As(err, &ee) { + stderr := strings.ToLower(string(ee.Stderr)) + if strings.Contains(stderr, "no crontab") || (ee.ExitCode() == 1 && len(out) == 0) { + return "", nil + } } return "", &errs.Error{Op: "schedule", Code: errs.CodeSystem, Cause: err, Hint: "could not read crontab (is the `crontab` command available?)"} } diff --git a/internal/cli/tunnel.go b/internal/cli/tunnel.go index d59f93a..beb625f 100644 --- a/internal/cli/tunnel.go +++ b/internal/cli/tunnel.go @@ -41,6 +41,15 @@ func newTunnelCmd() *cobra.Command { if localPort == 0 { localPort = prof.Port } + // Validate ports before building the forward spec, so an invalid value + // gives a clear user error instead of a bogus address and an opaque ssh + // failure. + if !validPort(prof.Port) { + return &errs.Error{Op: "tunnel", Code: errs.CodeUser, Hint: fmt.Sprintf("profile %s has an invalid database port %d", args[0], prof.Port)} + } + if !validPort(localPort) { + return &errs.Error{Op: "tunnel", Code: errs.CodeUser, Hint: fmt.Sprintf("invalid local_port %d (must be 1-65535)", localPort)} + } sshArgs := tunnelArgs(localPort, prof.Host, prof.Port, prof.Tunnel.Bastion) _, _ = fmt.Fprintf(c.OutOrStdout(), @@ -50,8 +59,11 @@ func newTunnelCmd() *cobra.Command { cmd := exec.CommandContext(c.Context(), "ssh", sshArgs...) cmd.Stdin, cmd.Stdout, cmd.Stderr = os.Stdin, c.OutOrStdout(), c.ErrOrStderr() if err := cmd.Run(); err != nil { - // A clean Ctrl-C (ssh exits non-zero on signal) is the normal stop, - // not an error worth a non-zero siphon exit beyond ssh's own. + // Ctrl-C is the normal close: the command context is cancelled and + // ssh exits on the signal. Treat that as success, not a failure. + if c.Context().Err() != nil { + return nil + } return &errs.Error{Op: "tunnel", Code: errs.CodeSystem, Cause: err, Hint: "ssh tunnel exited"} } return nil @@ -59,6 +71,9 @@ func newTunnelCmd() *cobra.Command { } } +// validPort reports whether p is a usable TCP port. +func validPort(p int) bool { return p > 0 && p <= 65535 } + // tunnelArgs builds the `ssh` argument list for a local forward. Pure and // testable: -N (no remote command), -L localPort:dbHost:dbPort, then the // bastion. ExitOnForwardFailure makes ssh fail fast if the local bind fails diff --git a/internal/schedule/crontab.go b/internal/schedule/crontab.go index 8841674..cabbf40 100644 --- a/internal/schedule/crontab.go +++ b/internal/schedule/crontab.go @@ -29,10 +29,17 @@ type Entry struct { // Render produces the crontab line(s) for an entry: a metadata comment (so List // can recover the profile) followed by the cron line invoking siphon. bin is the -// siphon binary path to invoke. +// siphon binary path. The binary path and profile are single-quoted so a space +// or shell metacharacter cannot break or alter the command cron runs via sh -c. func (e Entry) render(bin string) string { return fmt.Sprintf("%s %s\n%s %s backup %s", - linePrefix, e.Profile, e.Cron, bin, e.Profile) + linePrefix, e.Profile, e.Cron, shellQuote(bin), shellQuote(e.Profile)) +} + +// shellQuote wraps s in single quotes, escaping any embedded single quote, so it +// is a single safe argv element for /bin/sh. +func shellQuote(s string) string { + return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" } // List extracts the siphon-managed entries from a crontab's text. Entries diff --git a/internal/schedule/crontab_test.go b/internal/schedule/crontab_test.go index 3123ba9..c022e75 100644 --- a/internal/schedule/crontab_test.go +++ b/internal/schedule/crontab_test.go @@ -22,12 +22,28 @@ func TestAddListRoundTrip(t *testing.T) { if entries[1].Profile != "staging" || entries[1].Cron != "30 3 * * *" { t.Errorf("entries[1] = %+v, want staging / 30 3 * * *", entries[1]) } - // The rendered crontab actually invokes siphon backup. - if !strings.Contains(tab, bin+" backup prod") { + // The rendered crontab actually invokes siphon backup (args are shell-quoted). + if !strings.Contains(tab, "'"+bin+"' backup 'prod'") { t.Errorf("crontab missing backup invocation:\n%s", tab) } } +func TestRenderShellQuotesArgs(t *testing.T) { + // A profile name (or bin path) containing a space or quote must be quoted so + // it stays one argv element and cannot inject into the cron command. + tab := Add("", "/opt/my tools/siphon", Entry{Profile: "weird'; rm -rf /", Cron: "0 0 * * *"}) + // The dangerous content must appear only inside single quotes, with the inner + // quote escaped — never as a bare token that sh would re-parse. + if !strings.Contains(tab, `'/opt/my tools/siphon' backup 'weird'\''; rm -rf /'`) { + t.Errorf("args not safely shell-quoted:\n%s", tab) + } + // And it still round-trips through List (the metadata comment carries the raw + // profile, so List recovers it regardless of quoting). + if got := List(tab); len(got) != 1 || got[0].Profile != "weird'; rm -rf /" { + t.Errorf("List did not recover the raw profile: %+v", got) + } +} + func TestAddReschedulesExistingProfile(t *testing.T) { tab := Add("", bin, Entry{Profile: "prod", Cron: "0 2 * * *"}) tab = Add(tab, bin, Entry{Profile: "prod", Cron: "0 5 * * *"}) // same profile, new time diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index 0707bc7..4e0c0b1 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -65,14 +65,17 @@ func (h *recHandle) End(err error) { if err != nil { st.Errors++ } - snapshot := h.r.snapshotLocked() - h.r.mu.Unlock() - + // Flush under the lock so concurrent End calls write to disk in the same order + // they updated the counters — otherwise a slower goroutine holding an older + // snapshot could land last and regress the file. In-memory counters are always + // correct; this keeps the persisted JSON monotonic too. + // // Duration is not on the audit.Event at this layer (the handle does not see // it), so telemetry tracks counts/errors only; duration aggregation would // require the End signature to carry it. Counts + error rate are the useful // opt-in signal and keep the seam unchanged. - h.r.flush(snapshot) + h.r.flush(h.r.snapshotLocked()) + h.r.mu.Unlock() } // snapshotLocked builds a serializable copy under the held lock.