Phase G ops suite: audit, 2FA, telemetry, schedule, tunnel

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 <profile>` — the host cron runs the jobs (no daemon). `schedule add <profile> --cron`, `list`, `remove`.
+  - **`siphon tunnel <profile>`**: 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: <duration>`, 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.

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 <id>` 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 <profile>`            | Open an SSH tunnel to a DB via its bastion                   |
 | `siphon` _(bare)_                    | Launch the interactive multi-panel dashboard                 |
 
 Run `siphon <command> --help` for full flags.

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 <profile>` 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 <profile>` 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.

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

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
+}

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

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 {

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

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