From ec119566fd09059bbd869ad5dd1d3a22e593292d Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 15 Aug 2026 20:56:28 +0200 Subject: [PATCH] scratch matrix approval notifier, add chat-native MCP approval tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes the entire Matrix-based notifier (internal/notifier/) that polled for pending approvals, sent Matrix alerts, and checked for reaction-based approve/deny. Approval decisions now work on any chat platform (Hermes desktop, Telegram, Discord, WhatsApp, CLI) via two new MCP tools: - list_approvals — query pending/recent approvals by status or entity - decide_approval — approve/deny via same API endpoint as UI + nomos Config fields removed: MatrixHomeserver, MatrixUserID, MatrixToken, MatrixRoomID, ApprovalHMACSecret. Docker notifier: service removed. Approval HMAC token generation removed (unused by code). The existing chat-assent path in nomos (cmd/nomos/assent.go) and the control-room Approve button keep working unchanged — both call the shared POST /api/v1/approvals/{id}/decision endpoint. --- AGENTS.md | 9 +- CLIENTS.md | 2 +- CONTRIBUTING.md | 1 - README.md | 12 +- cmd/oikos/main.go | 16 +- docker-compose.yml | 38 ---- internal/config/config.go | 26 +-- internal/health/health.go | 2 +- internal/mcp/ops_tools.go | 93 +++++++++ internal/notifier/notifier.go | 300 ----------------------------- internal/notifier/notifier_test.go | 187 ------------------ internal/scheduler/backup.go | 2 +- scripts/cutover-checklist.md | 2 +- scripts/deploy.sh | 8 +- scripts/seed-secrets.sh | 4 - 15 files changed, 115 insertions(+), 587 deletions(-) delete mode 100644 internal/notifier/notifier.go delete mode 100644 internal/notifier/notifier_test.go diff --git a/AGENTS.md b/AGENTS.md index 7ce7f29e..553e8a02 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -56,7 +56,7 @@ Endpoint: `https://mcp.hubris.network/mcp`. Every call needs enrollment and `/healthz` (see "Authentication" below for where the token comes from). -Available tools (65 total — the authoritative list; do not hardcode the count +Available tools (67 total — the authoritative list; do not hardcode the count elsewhere; regenerate from `internal/mcp/` when tools change): 💡 Slug param alias: all entity-lookup tools now accept `slug` in addition to @@ -103,6 +103,8 @@ elsewhere; regenerate from `internal/mcp/` when tools change): list_executions(entity_slug, status, limit=25) — cursor-paginated execution history list_entity_sessions(entity_slug) — active Nomos sessions linked to an entity get_dashboard_summary() — fleet overview: counts, health, signals, approvals + list_approvals(status, entity_slug, limit) — list pending/recent approvals; filter by status (pending, approved, denied) or entity + decide_approval(approval_id, decision) — approve or deny a pending execution; calls the same API endpoint as the Approve button in the UI get_secret(key, path, environment) — retrieve a secret from the Infisical vault list_secrets(path_prefix) — list secret keys in the Infisical vault set_secret(key, value, path, environment) — store/update a secret (requires approval) @@ -187,13 +189,14 @@ The DB is the truth. The old wiki files are archived at `archive/knowledge/` - **Actions** (restart, logs, apt, pct exec, or anything else): Nomos calls `run` (the general execution primitive) via MCP. `reversible_low`/read-only actions execute immediately; `config_mutation` and `destructive` actions are queued for - operator approval via Matrix or the control-room UI's Operations page. + operator approval via the App button in the control-room UI or via + `list_approvals`/`decide_approval` MCP tools from any agent. - **Secrets**: managed by Infisical (`oikos secret` subcommand for migration). Never hardcode secrets — use env vars from `.env`. - **Mutations** (restart, edit configs, etc.): classified against `seeds/policy.yaml`. `reversible_low` actions auto-execute; `config_mutation`/`destructive` actions require approval — granted by - the operator via Matrix reply or the control-room UI, not a CLI flag. + the operator via the App button or `decide_approval` MCP tool call. See OIKOS.md. ## 7. Communication mode diff --git a/CLIENTS.md b/CLIENTS.md index e37398ed..553b1c9c 100644 --- a/CLIENTS.md +++ b/CLIENTS.md @@ -23,7 +23,7 @@ Docker stack on mac-mini and exposes an MCP server + REST API. | Change history | MCP `get_change_history` | | State snapshot (health, disk, drift) | MCP `get_state_snapshot` | | Secrets (Infisical) | REST API + `oikos secret` CLI | -| Approval tokens | Matrix via notifier | +| Approve/deny pending executions | MCP `list_approvals`, `decide_approval` (chat-native, works on any platform) | | Run a command on a host/LXC (policy-gated) | MCP `run` | | Record a discovered fact/relationship | MCP `update_entity_attributes`, `create_relationship`, `upsert_knowledge` | diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 79009418..9fd26e3d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -74,7 +74,6 @@ internal/ All Go packages scheduler/ Observe loop, probes, signals actuator/ SSH execution learning/ Pattern recognition, anomaly detection - notifier/ Matrix notifications, approval tokens policy/ Risk classifier secrets/ Infisical + SOPS backend domain/ Core types: entities, approvals, signals, patterns diff --git a/README.md b/README.md index 86bba4ac..a9a8ee65 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ learns from outcomes, and escalates when uncertain. ## Quick start ```bash -# Dev stack (postgres + api + scheduler + notifier). The api/nomos +# Dev stack (postgres + api + scheduler). The api/nomos # services need a shared token — every route requires a real bearer # credential, there's no dev-open bypass. OIKOS_MCP_BEARER_TOKEN=dev-token docker compose --profile dev up -d @@ -42,8 +42,8 @@ cd web && OIKOS_API_TOKEN=dev-token npm run dev # http://localhost:5173 Workstation ─── │ nomos (8092) ──MCP── api (8090) │ (mesh) │ MCP gateway REST + MCP │ │ │ - │ scheduler ── notifier ── postgres │ - │ (observe) (Matrix) (Timescale)│ + │ scheduler ─── postgres │ + │ (observe) (Timescale) │ └──────────────────────────────────┘ ``` @@ -51,7 +51,6 @@ cd web && OIKOS_API_TOKEN=dev-token npm run dev # http://localhost:5173 |-----------|------|------| | `oikos api` | 8090 | REST API + MCP server (tool list in [AGENTS.md §3](AGENTS.md#3-the-mcp-server)) | | `oikos scheduler` | — | Probe runner, signal lifecycle, metrics | -| `oikos notifier` | — | Approval tokens, Matrix alerts | | `nomos serve` | 8092 | MCP client gateway, query routing | ## Phases @@ -60,7 +59,7 @@ cd web && OIKOS_API_TOKEN=dev-token npm run dev # http://localhost:5173 |-------|--------|-------------| | 1 — Ontology + DB | ✅ | TimescaleDB, migrations, seeds, blast_radius | | 2 — API | ✅ | OpenAPI-first REST + MCP, auth, SSE, audit | -| 3 — Control loop | ✅ | Scheduler, actuator, learning, classifier, notifier | +| 3 — Control loop | ✅ | Scheduler, actuator, learning, classifier | | 4 — Nomos agent | ✅ | Standalone MCP client gateway, agent activity | | 5 — Secrets | ✅ | Infisical backend + SOPS fallback, rotation runbooks | | 6 — Deploy | ✅ | CI pipeline, cutover checklist, watchdog, rollback | @@ -100,7 +99,6 @@ oikos seed # ingest ontology/inventory/policy seeds oikos export # export DB state to YAML oikos api # serve REST + MCP oikos scheduler # run observe loop -oikos notifier # run notification loop oikos all # all roles in one process oikos secret list # enumerate SOPS secrets oikos secret migrate # SOPS → Infisical @@ -123,7 +121,7 @@ cmd/nomos/ Nomos MCP client gateway cmd/webhook/ Gitea deploy-webhook receiver (push-to-deploy on mac-mini) cmd/desktop/ Wails desktop wrapper around the SPA internal/ Go packages (actuator, checkdefaults, config, db, domain, - httpapi, knowledge, learning, mcp, notifier, observability, + httpapi, knowledge, learning, mcp, observability, ontology, policy, safego, scheduler, secrets) web/ Control-room SPA (Svelte 5) — standalone, not embedded api/openapi.yaml API contract (OpenAPI 3.1) diff --git a/cmd/oikos/main.go b/cmd/oikos/main.go index 6168ebd5..2fd2dd8e 100644 --- a/cmd/oikos/main.go +++ b/cmd/oikos/main.go @@ -15,7 +15,6 @@ import ( "github.com/dtoro/oikos/internal/execworker" "github.com/dtoro/oikos/internal/httpapi" "github.com/dtoro/oikos/internal/knowledge" - "github.com/dtoro/oikos/internal/notifier" "github.com/dtoro/oikos/internal/observability" "github.com/dtoro/oikos/internal/scheduler" "github.com/dtoro/oikos/internal/secrets" @@ -23,7 +22,6 @@ import ( ) var schedulerRunner = scheduler.RunnerForMain() -var notifierRunner = notifier.RunnerForMain() var execWorkerRunner = execworker.RunnerForMain() func main() { @@ -55,8 +53,6 @@ func main() { ) if sec != nil { overlays := secrets.ConfigOverlays(map[string]func(string){ - "matrix_token": func(v string) { cfg.MatrixToken = v }, - "approval_hmac-secret": func(v string) { cfg.ApprovalHMACSecret = v }, "mcp_bearer-token": func(v string) { cfg.MCPBearerToken = v }, "api_token": func(v string) { cfg.APIToken = v }, "oidc_client-secret": func(v string) { cfg.OIDCClientSecret = v }, @@ -65,7 +61,7 @@ func main() { slog.Info("secrets resolved from Infisical", "count", n) secrets.VerifyExpectedSecrets(ctx, sec, []string{ - "matrix_token", "approval_hmac-secret", "mcp_bearer-token", + "approval_hmac-secret", "mcp_bearer-token", "api_token", "openrouter_api-key", "webhook_hmac-secret", }) } @@ -95,8 +91,6 @@ func main() { } case "scheduler": runWithPool(ctx, cfg, "scheduler", schedulerRunner) - case "notifier": - runWithPool(ctx, cfg, "notifier", notifierRunner) case "execution-worker": runWithPool(ctx, cfg, "execution-worker", execWorkerRunner) case "all": @@ -113,10 +107,9 @@ func main() { } go schedulerRunner(ctx, pool, cfg) - go notifierRunner(ctx, pool, cfg) go execWorkerRunner(ctx, pool, cfg) - slog.Info("all: starting api with scheduler + notifier + execution-worker in background") + slog.Info("all: starting api with scheduler + execution-worker in background") if err := httpapi.ListenAndServe(ctx, pool, cfg); err != nil { slog.Error("api failed", "error", err) os.Exit(1) @@ -145,7 +138,6 @@ Roles: export Export DB state back to seed YAMLs (DR / version control) api Run the REST + MCP API server scheduler Run the observe loop - notifier Run the notification service (Matrix alerts) all Run all roles in one process (dev mode) secret Secret management (Infisical: get, set, list, verify, audit, migrate, export-sops) knowledge Convert wiki to knowledge seed (one-shot) @@ -386,8 +378,6 @@ func runSecret(ctx context.Context, cfg config.Config) { // expectedSecrets is the set of keys that should exist in Infisical // for a fully-migrated deployment. var expectedSecrets = []string{ - "matrix_token", - "approval_hmac-secret", "mcp_bearer-token", "api_token", "openrouter_api-key", @@ -429,8 +419,6 @@ func runSecretAudit(ctx context.Context, cfg config.Config) { backend := newInfisicalBackendOrFail(cfg) envValues := map[string]string{ - "matrix_token": cfg.MatrixToken, - "approval_hmac-secret": cfg.ApprovalHMACSecret, "mcp_bearer-token": cfg.MCPBearerToken, "api_token": cfg.APIToken, "oidc_client-secret": cfg.OIDCClientSecret, diff --git a/docker-compose.yml b/docker-compose.yml index 1281d3b2..cdb21174 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -157,44 +157,6 @@ services: retries: 3 start_period: 90s - # Notifier (Phase 3) — Matrix alerts - notifier: - image: oikos-notifier:${OIKOS_VERSION:-latest} - build: - context: . - dockerfile: compose/oikos/Dockerfile - restart: unless-stopped - profiles: ["dev", "full"] - depends_on: - seed: - condition: service_completed_successfully - environment: - OIKOS_DATABASE_URL: postgres://oikos:${OIKOS_DB_PASSWORD:-oikos_dev}@postgres:5432/oikos?sslmode=disable - OIKOS_DEBUG: "true" - OIKOS_APPROVAL_HMAC_SECRET: ${OIKOS_APPROVAL_HMAC_SECRET:-dev-secret} - OIKOS_MATRIX_HOMESERVER: ${OIKOS_MATRIX_HOMESERVER:-https://matrix.hubris.network} - OIKOS_MATRIX_USER: ${OIKOS_MATRIX_USER:-@hermes:hubris.network} - OIKOS_MATRIX_TOKEN: ${OIKOS_MATRIX_TOKEN} - OIKOS_MATRIX_ROOM: ${OIKOS_MATRIX_ROOM:-!alerts:hubris.network} - # Liveness probe (plan D5): bumps each approval/reaction tick. - OIKOS_HEALTH_LISTEN: ":8094" - OIKOS_INFISICAL_SITE_URL: ${OIKOS_INFISICAL_SITE_URL:-} - OIKOS_INFISICAL_CLIENT_ID: ${OIKOS_INFISICAL_CLIENT_ID:-} - OIKOS_INFISICAL_CLIENT_SECRET: ${OIKOS_INFISICAL_CLIENT_SECRET:-} - OIKOS_INFISICAL_PROJECT_ID: ${OIKOS_INFISICAL_PROJECT_ID:-} - OIKOS_INFISICAL_ENV: ${OIKOS_INFISICAL_ENV:-dev} - command: ["notifier"] - stop_signal: SIGTERM - stop_grace_period: 30s - mem_limit: 128m - cpus: 0.5 - healthcheck: - test: ["CMD", "wget", "-q", "-O", "-", "http://127.0.0.1:8094/healthz"] - interval: 30s - timeout: 5s - retries: 3 - start_period: 120s - # Execution worker (Phase 6) — Postgres-backed job queue execution-worker: image: oikos-execution-worker:${OIKOS_VERSION:-latest} diff --git a/internal/config/config.go b/internal/config/config.go index 7556e32b..da4934d9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -37,7 +37,7 @@ type Config struct { APIRateBurst int // Health probe HTTP listener (plan D5). Background-loop roles (scheduler, - // notifier) expose a staleness-aware /healthz here. Empty disables the + // execution-worker) expose a staleness-aware /healthz here. Empty disables the // health server (local/non-docker runs). HealthListen string @@ -53,12 +53,6 @@ type Config struct { // Scheduler (Phase 3) SchedulerInterval time.Duration // check loop interval (default 30s) - // Notifier (Phase 3) - MatrixHomeserver string // Matrix server URL - MatrixUserID string // bot user ID (e.g. @oikos:matrix.hubris.network) - MatrixToken string // Matrix access token - MatrixRoomID string // alert room ID - // Actuator (Phase 3) SSHKeyPath string // path to the restricted SSH key SSHUser string // SSH user on targets (default "oikos") @@ -68,9 +62,6 @@ type Config struct { // Learning (Phase 3) LearningInterval time.Duration // pattern extraction interval (default 3600s) - // Approval HMAC secret (Phase 3) - ApprovalHMACSecret string - // Nomos agent entity ID (Phase 4) NomosAgentID string NomosAgentSlug string @@ -148,18 +139,6 @@ func FromEnv() Config { c.SchedulerInterval = d } } - if v := os.Getenv("OIKOS_MATRIX_HOMESERVER"); v != "" { - c.MatrixHomeserver = v - } - if v := os.Getenv("OIKOS_MATRIX_USER"); v != "" { - c.MatrixUserID = v - } - if v := os.Getenv("OIKOS_MATRIX_TOKEN"); v != "" { - c.MatrixToken = v - } - if v := os.Getenv("OIKOS_MATRIX_ROOM"); v != "" { - c.MatrixRoomID = v - } if v := os.Getenv("OIKOS_SSH_KEY_PATH"); v != "" { c.SSHKeyPath = v } @@ -177,9 +156,6 @@ func FromEnv() Config { c.LearningInterval = d } } - if v := os.Getenv("OIKOS_APPROVAL_HMAC_SECRET"); v != "" { - c.ApprovalHMACSecret = v - } if v := os.Getenv("OIKOS_NOMOS_AGENT_ID"); v != "" { c.NomosAgentID = v } diff --git a/internal/health/health.go b/internal/health/health.go index 636f4583..e45c247d 100644 --- a/internal/health/health.go +++ b/internal/health/health.go @@ -1,5 +1,5 @@ // Package health provides a staleness-aware liveness probe for background- -// loop services (scheduler, notifier) that don't otherwise serve HTTP. +// loop services (scheduler, execution-worker) that don't otherwise serve HTTP. // // The owning loop calls Probe.Bump() on each iteration. A /healthz endpoint // returns 200 while the last bump is within the staleness window, and 503 diff --git a/internal/mcp/ops_tools.go b/internal/mcp/ops_tools.go index 1a4bd545..c83efd9e 100644 --- a/internal/mcp/ops_tools.go +++ b/internal/mcp/ops_tools.go @@ -1,10 +1,12 @@ package mcp import ( + "bytes" "context" "encoding/json" "fmt" "net/http" + "os" "strings" "time" @@ -724,5 +726,96 @@ func OpsTools(pool *db.Pool, agentID uuid.UUID, sec secretBackend) []toolReg { purpose := fmt.Sprintf("push %s into %s at %s", sourcePath, targetSlug, destPath) return classifyAndGate(ctx, pool, agentID, hostEntityID, "host:"+hostSlug, cmd, purpose, "config_mutation", sessionID), nil }}, + // ── Approval Management (replaces Matrix notifier) ───────────── + {tool: &mcp.Tool{Name: "list_approvals", Description: "List pending and recent approvals. Returns approval ID, action, risk class, target slug, status, and timing. Filter by status (pending, approved, denied) or entity slug to scope. Use after a `run` returns 'requires approval' to see what's pending so you can present it to the operator for a decision.", + InputSchema: objSchema( + prop{"status", "string", "Optional: filter by status (pending, approved, denied, expired, revoked)"}, + prop{"entity_slug", "string", "Optional: filter by target entity slug"}, + prop{"limit", "integer", "Max rows (default 20)"}), + }, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + args := argsMap(req) + status, _ := args["status"].(string) + entSlug, _ := args["entity_slug"].(string) + lim := int(getFloat(args, "limit", 20)) + if lim < 1 { + lim = 1 + } + if lim > 100 { + lim = 100 + } + var statusPtr, slugPtr *string + if status != "" { + statusPtr = &status + } + if entSlug != "" { + slugPtr = &entSlug + } + return queryRows(ctx, pool, ` + SELECT a.entity_id, e.slug AS target_slug, a.action, a.risk_class, + a.kind, a.status, a.expires_at, a.decided_at, a.created_at + FROM approvals a + JOIN entities e ON e.id = COALESCE(a.subject_entity_id, a.entity_id) + WHERE ($1::text IS NULL OR a.status = $1) + AND ($2::text IS NULL OR e.slug = $2) + ORDER BY a.created_at DESC LIMIT $3`, statusPtr, slugPtr, lim), nil + }}, + {tool: &mcp.Tool{Name: "decide_approval", Description: "Approve or deny a pending execution approval. Call this after presenting the command details to the operator and getting their explicit authorization (\"go ahead\", \"yes\", \"proceed\", or for destructive actions \"I confirm ...\"). The operator's typed approval in any chat (Hermes desktop, Telegram, Discord, WhatsApp) works — this tool is the agent-side action to record the decision and trigger the execution. Returns the new approval status and execution result once done.", + InputSchema: objSchema( + prop{"approval_id", "string", "Approval entity UUID from list_approvals or a prior run result (e.g. 'execution X queued')"}, + prop{"decision", "string", "Decision: 'approve' to authorize the action, 'deny' to reject it. Destructive actions still need explicit typed confirmation from the operator."}), + }, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + args := argsMap(req) + appID, _ := args["approval_id"].(string) + decision, _ := args["decision"].(string) + if appID == "" || decision == "" { + return textResult("error: approval_id and decision are required"), nil + } + if decision != "approve" && decision != "deny" { + return textResult("error: decision must be 'approve' or 'deny'"), nil + } + + // Validate the approval exists and is still pending + var status string + if err := pool.QueryRow(ctx, `SELECT status FROM approvals WHERE entity_id = $1`, appID).Scan(&status); err != nil { + return textResult(fmt.Sprintf("approval not found: %s", appID)), nil + } + if status != "pending" { + return textResult(fmt.Sprintf("approval %s is already %s — cannot decide again", appID, status)), nil + } + + // Call the HTTP API decision endpoint (same path as the UI Approve button + // and nomos chat-assent), so all approval paths share one code path for + // execution dispatch, session management, events, and audit trail. + apiBase := os.Getenv("OIKOS_API_BASE") + if apiBase == "" { + apiBase = "http://api:8090" + } + body, _ := json.Marshal(map[string]string{"decision": decision}) + client := &http.Client{Timeout: 30 * time.Second} + hreq, hreqErr := http.NewRequestWithContext(ctx, http.MethodPost, + apiBase+"/api/v1/approvals/"+appID+"/decision", bytes.NewReader(body)) + if hreqErr != nil { + return textResult(fmt.Sprintf("error: %v", hreqErr)), nil + } + hreq.Header.Set("Content-Type", "application/json") + if token := os.Getenv("OIKOS_MCP_BEARER_TOKEN"); token != "" { + hreq.Header.Set("Authorization", "Bearer "+token) + } + resp, reqErr := client.Do(hreq) + if reqErr != nil { + return textResult(fmt.Sprintf("error: API call failed: %v", reqErr)), nil + } + defer resp.Body.Close() + + var result map[string]any + if err := json.NewDecoder(resp.Body).Decode(&result); err == nil && len(result) > 0 { + b, _ := json.MarshalIndent(result, "", " ") + return textResult(fmt.Sprintf("%s: approval %s -> %s\nResponse: %s", decision, appID, decision, string(b))), nil + } + if resp.StatusCode == http.StatusOK { + return textResult(fmt.Sprintf("%s: approval %s decided successfully.", decision, appID)), nil + } + return textResult(fmt.Sprintf("%s: API returned status %d for approval %s", decision, resp.StatusCode, appID)), nil + }}, } } diff --git a/internal/notifier/notifier.go b/internal/notifier/notifier.go deleted file mode 100644 index f6437aec..00000000 --- a/internal/notifier/notifier.go +++ /dev/null @@ -1,300 +0,0 @@ -// Package notifier handles alerts and approval requests via Matrix. -// Uses the DB as the rendezvous — no service-to-service calls (SA7/A7). -// Pending approvals survive restarts of either side. -package notifier - -import ( - "bytes" - "context" - "crypto/hmac" - "crypto/sha256" - "encoding/hex" - "encoding/json" - "fmt" - "log/slog" - "net/http" - "strings" - "time" - - "github.com/dtoro/oikos/internal/config" - "github.com/dtoro/oikos/internal/db" - "github.com/dtoro/oikos/internal/health" - "github.com/google/uuid" -) - -// Run starts the notifier loop. Blocks until ctx is cancelled. -func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { - slog.Info("notifier: starting") - - // Liveness probe (plan D5): the notifier ticks every 15s (approvals) and - // 30s (reactions). 2 min staleness covers a slow Matrix round-trip plus a - // missed tick without false-failing. - probe := health.New(2 * time.Minute) - probe.Serve(ctx, cfg.HealthListen) - - processPendingApprovals(ctx, pool, cfg) - probe.Bump() - - ticker := time.NewTicker(15 * time.Second) - defer ticker.Stop() - - reactionTimer := time.NewTicker(30 * time.Second) - defer reactionTimer.Stop() - - for { - select { - case <-ctx.Done(): - slog.Info("notifier: shutting down") - return - case <-ticker.C: - processPendingApprovals(ctx, pool, cfg) - probe.Bump() - case <-reactionTimer.C: - pollReactions(ctx, pool, cfg) - probe.Bump() - } - } -} - -// RunnerForMain provides the run function for registration in main. -func RunnerForMain() func(context.Context, *db.Pool, config.Config) { - return Run -} - -type pendingApproval struct { - ID uuid.UUID - Action string - RiskClass string - TokenHash *string - AlertSentAt *time.Time - MatrixEventID *string - ExpiresAt time.Time -} - -// processPendingApprovals finds pending approvals, generates tokens, and sends Matrix alerts. -func processPendingApprovals(ctx context.Context, pool *db.Pool, cfg config.Config) { - rows, err := pool.Query(ctx, ` - SELECT entity_id, action, risk_class, token_hash, alert_sent_at, - matrix_event_id, expires_at - FROM approvals - WHERE status = 'pending' AND expires_at > now() - ORDER BY created_at`) - if err != nil { - slog.Error("notifier: list approvals", "error", err) - return - } - defer rows.Close() - - var approvals []pendingApproval - for rows.Next() { - var a pendingApproval - if err := rows.Scan(&a.ID, &a.Action, &a.RiskClass, &a.TokenHash, - &a.AlertSentAt, &a.MatrixEventID, &a.ExpiresAt); err != nil { - slog.Error("notifier: scan approval", "error", err) - continue - } - approvals = append(approvals, a) - } - - for _, a := range approvals { - if a.ExpiresAt.Before(time.Now()) { - pool.Exec(ctx, "UPDATE approvals SET status = 'expired' WHERE entity_id = $1", a.ID) - continue - } - - if a.TokenHash == nil || *a.TokenHash == "" { - token := generateApprovalToken(a.ID, cfg.ApprovalHMACSecret) - tokenHash := hashToken(token) - pool.Exec(ctx, "UPDATE approvals SET token_hash = $2 WHERE entity_id = $1", a.ID, tokenHash) - a.TokenHash = &tokenHash - } - - if a.AlertSentAt != nil { - continue - } - - eventID, err := sendMatrixAlert(ctx, cfg, a.ID, a.Action, a.RiskClass) - if err != nil { - slog.Error("notifier: send Matrix alert", "error", err, "approval_id", a.ID) - continue - } - pool.Exec(ctx, - "UPDATE approvals SET matrix_event_id = $2, alert_sent_at = now() WHERE entity_id = $1", - a.ID, eventID) - } -} - -// pollReactions checks Matrix for ✅/❌ reactions on sent approval messages. -func pollReactions(ctx context.Context, pool *db.Pool, cfg config.Config) { - if cfg.MatrixHomeserver == "" || cfg.MatrixToken == "" { - return - } - - rows, err := pool.Query(ctx, ` - SELECT entity_id, matrix_event_id - FROM approvals - WHERE status = 'pending' - AND matrix_event_id IS NOT NULL - AND alert_sent_at IS NOT NULL - AND expires_at > now() - ORDER BY created_at`) - if err != nil { - slog.Error("notifier: query approvals for reactions", "error", err) - return - } - defer rows.Close() - - for rows.Next() { - var approvalID uuid.UUID - var matrixEventID string - if err := rows.Scan(&approvalID, &matrixEventID); err != nil { - continue - } - - decision := checkReaction(ctx, cfg, cfg.MatrixRoomID, matrixEventID) - if decision == "" { - continue - } - - slog.Info("notifier: reaction detected", - "approval_id", approvalID, "decision", decision) - - callDecideApproval(ctx, cfg, approvalID, decision) - } -} - -// checkReaction queries Matrix for annotations (reactions) on a message. -func checkReaction(ctx context.Context, cfg config.Config, roomID, eventID string) string { - url := fmt.Sprintf("%s/_matrix/client/v3/rooms/%s/relations/%s/m.annotation", - cfg.MatrixHomeserver, roomID, eventID) - - req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) - if err != nil { - return "" - } - req.Header.Set("Authorization", "Bearer "+cfg.MatrixToken) - - resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req) - if err != nil { - slog.Warn("notifier: Matrix relations query failed", "error", err) - return "" - } - defer resp.Body.Close() - - var result struct { - Chunk []struct { - Type string `json:"type"` - Content struct { - RelatesTo map[string]string `json:"m.relates_to"` - } `json:"content"` - } `json:"chunk"` - } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return "" - } - - for _, ev := range result.Chunk { - if ev.Type != "m.reaction" { - continue - } - key := ev.Content.RelatesTo["key"] - switch { - case strings.Contains(key, "\u2705"), strings.Contains(key, "\U0001F44D"), - key == "✅", key == "👍", key == "approve": - return "approve" - case strings.Contains(key, "\u274C"), strings.Contains(key, "\U0001F44E"), - key == "❌", key == "👎", key == "deny": - return "deny" - } - } - return "" -} - -// callDecideApproval calls the oikos API to record a decision. -func callDecideApproval(ctx context.Context, cfg config.Config, approvalID uuid.UUID, decision string) { - body := map[string]string{"decision": decision} - data, _ := json.Marshal(body) - - url := fmt.Sprintf("http://api:8090/api/v1/approvals/%s/decision", approvalID) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(data)) - if err != nil { - slog.Error("notifier: build decide request", "error", err) - return - } - req.Header.Set("Content-Type", "application/json") - if cfg.APIToken != "" { - req.Header.Set("Authorization", "Bearer "+cfg.APIToken) - } - - resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req) - if err != nil { - slog.Error("notifier: decide API call", "error", err) - return - } - resp.Body.Close() - - slog.Info("notifier: decided via reaction", - "approval_id", approvalID, "decision", decision, "status", resp.StatusCode) -} - -// sendMatrixAlert posts an approval request message and returns the event ID. -func sendMatrixAlert(ctx context.Context, cfg config.Config, approvalID uuid.UUID, action, riskClass string) (string, error) { - body := fmt.Sprintf( - "🔐 **Approval required**\n\n"+ - "**Action:** %s\n**Risk class:** %s\n**ID:** `%s`\n**Expires:** 1h\n\n"+ - "React ✅ to approve or ❌ to deny.", - action, riskClass, approvalID, - ) - - msg := map[string]any{"msgtype": "m.text", "body": body} - data, err := json.Marshal(msg) - if err != nil { - return "", fmt.Errorf("marshal: %w", err) - } - - txnID := fmt.Sprintf("approval-%s-%d", approvalID, time.Now().UnixNano()) - url := fmt.Sprintf("%s/_matrix/client/v3/rooms/%s/send/m.room.message/%s", - cfg.MatrixHomeserver, cfg.MatrixRoomID, txnID) - - req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(data)) - if err != nil { - return "", fmt.Errorf("build request: %w", err) - } - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", "Bearer "+cfg.MatrixToken) - - resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req) - if err != nil { - return "", fmt.Errorf("send: %w", err) - } - defer resp.Body.Close() - - var mxResp struct { - EventID string `json:"event_id"` - } - json.NewDecoder(resp.Body).Decode(&mxResp) - - if mxResp.EventID == "" { - return "", fmt.Errorf("no event_id (status %d)", resp.StatusCode) - } - - slog.Info("notifier: alert sent", - "approval_id", approvalID, "event_id", mxResp.EventID) - return mxResp.EventID, nil -} - -// generateApprovalToken creates a single-use HMAC token. -func generateApprovalToken(approvalID uuid.UUID, secret string) string { - if secret == "" { - secret = "dev-secret-do-not-use-in-prod" - } - mac := hmac.New(sha256.New, []byte(secret)) - mac.Write([]byte(approvalID.String())) - mac.Write([]byte(fmt.Sprintf("%d", time.Now().UnixNano()))) - return hex.EncodeToString(mac.Sum(nil)) -} - -func hashToken(token string) string { - h := sha256.Sum256([]byte(token)) - return hex.EncodeToString(h[:]) -} diff --git a/internal/notifier/notifier_test.go b/internal/notifier/notifier_test.go deleted file mode 100644 index 35de6bb5..00000000 --- a/internal/notifier/notifier_test.go +++ /dev/null @@ -1,187 +0,0 @@ -package notifier - -import ( - "crypto/hmac" - "crypto/sha256" - "encoding/hex" - "fmt" - "regexp" - "strings" - "testing" - "time" - - "github.com/google/uuid" -) - -var hex64Re = regexp.MustCompile(`^[0-9a-f]{64}$`) - -func TestHashToken(t *testing.T) { - // sha256("") == e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 - emptyHash := "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" - - t.Run("determinism same input same output", func(t *testing.T) { - a := hashToken("approval-token-123") - b := hashToken("approval-token-123") - if a != b { - t.Fatalf("hashToken not deterministic: %q vs %q", a, b) - } - }) - - t.Run("empty string known sha256", func(t *testing.T) { - got := hashToken("") - if got != emptyHash { - t.Fatalf("hashToken(\"\") = %q, want %q", got, emptyHash) - } - }) - - t.Run("different inputs different outputs", func(t *testing.T) { - a := hashToken("one") - b := hashToken("two") - if a == b { - t.Fatalf("hashToken collided for different inputs: %q", a) - } - }) - - t.Run("output is valid 64-char hex", func(t *testing.T) { - for _, in := range []string{"", "abc", "some-longer-token-value-xyz"} { - got := hashToken(in) - if !hex64Re.MatchString(got) { - t.Fatalf("hashToken(%q) = %q, not 64-char lowercase hex", in, got) - } - // also must decode cleanly to 32 bytes - b, err := hex.DecodeString(got) - if err != nil { - t.Fatalf("hashToken(%q) decode error: %v", in, err) - } - if len(b) != 32 { - t.Fatalf("hashToken(%q) decoded len = %d, want 32", in, len(b)) - } - } - }) -} - -func TestGenerateApprovalToken(t *testing.T) { - id := uuid.New() - - t.Run("output is 64-char hex", func(t *testing.T) { - tok := generateApprovalToken(id, "super-secret") - if !hex64Re.MatchString(tok) { - t.Fatalf("generateApprovalToken = %q, not 64-char lowercase hex", tok) - } - b, err := hex.DecodeString(tok) - if err != nil { - t.Fatalf("decode error: %v", err) - } - if len(b) != 32 { - t.Fatalf("decoded len = %d, want 32 (sha256)", len(b)) - } - }) - - t.Run("empty secret falls back to dev secret no panic", func(t *testing.T) { - tok := generateApprovalToken(id, "") - if tok == "" { - t.Fatal("empty secret produced empty token") - } - if !hex64Re.MatchString(tok) { - t.Fatalf("empty-secret token %q not 64-char hex", tok) - } - }) - - // Non-determinism: time.Now().UnixNano() is embedded in the HMAC message, - // so two calls with identical inputs produce different tokens (unless the - // clock has nanosecond-identical reads, which we do not assert against). - t.Run("same inputs twice produce different tokens (time-based)", func(t *testing.T) { - a := generateApprovalToken(id, "stable-secret") - b := generateApprovalToken(id, "stable-secret") - if a == b { - // Not a hard failure (clock granularity), but document expectation. - t.Logf("note: two immediate calls returned identical token %q — clock resolution collapsed", a) - } - }) - - t.Run("different secrets produce different tokens", func(t *testing.T) { - a := generateApprovalToken(id, "secret-a") - b := generateApprovalToken(id, "secret-b") - if a == b { - t.Fatalf("different secrets produced same token %q", a) - } - }) - - // HMAC correctness: re-derive the token with the same secret + approvalID - // using a freshly captured timestamp window is impossible because we don't - // observe the embedded timestamp. Instead, verify the token is a valid - // HMAC-SHA256 by brute-forcing a small time window around now: reconstruct - // mac(secret, approvalID || ts) for ts in [now-N, now] and confirm one - // matches. This proves the token genuinely is an HMAC over (approvalID, ts) - // with the supplied secret. - t.Run("token is HMAC-SHA256 over approvalID+timestamp with secret", func(t *testing.T) { - secret := "hmac-verify-secret" - before := nowNanos() - tok := generateApprovalToken(id, secret) - after := nowNanos() - - // The token's embedded ts is captured inside generateApprovalToken, - // which is called after `before` was sampled — so ts ∈ [before, after]. - // Add a tiny ±band to absorb scheduler jitter on loaded runners. - lo := before - 10_000 - hi := after + 10_000 - matched := false - for ts := lo; ts <= hi; ts++ { - mac := hmac.New(sha256.New, []byte(secret)) - mac.Write([]byte(id.String())) - mac.Write([]byte(formatInt(ts))) - cand := hex.EncodeToString(mac.Sum(nil)) - if hmac.Equal([]byte(cand), []byte(tok)) { - matched = true - break - } - } - if !matched { - t.Fatalf("token %q did not match any HMAC in window [%d,%d]; not a valid HMAC-SHA256 over approvalID+ts", tok, lo, hi) - } - }) - - t.Run("empty secret HMAC uses dev fallback secret", func(t *testing.T) { - before := nowNanos() - tok := generateApprovalToken(id, "") - after := nowNanos() - dev := "dev-secret-do-not-use-in-prod" - lo := before - 10_000 - hi := after + 10_000 - matched := false - for ts := lo; ts <= hi; ts++ { - mac := hmac.New(sha256.New, []byte(dev)) - mac.Write([]byte(id.String())) - mac.Write([]byte(formatInt(ts))) - cand := hex.EncodeToString(mac.Sum(nil)) - if hmac.Equal([]byte(cand), []byte(tok)) { - matched = true - break - } - } - if !matched { - t.Fatalf("empty-secret token %q did not match dev-fallback HMAC", tok) - } - }) - - // sanity: token should not leak the secret in plaintext - t.Run("token does not contain secret substring", func(t *testing.T) { - secret := "leakcheck-secret-xyz" - tok := generateApprovalToken(id, secret) - if strings.Contains(tok, secret) { - t.Fatalf("token %q contains secret substring %q", tok, secret) - } - }) -} - -// nowNanos returns the current nanosecond count, matching the time source -// used by generateApprovalToken (time.Now().UnixNano()). -func nowNanos() int64 { - return time.Now().UnixNano() -} - -// formatInt mirrors fmt.Sprintf("%d", ...) used by the production code so the -// re-derivation in tests is byte-identical. -func formatInt(n int64) string { - return fmt.Sprintf("%d", n) -} diff --git a/internal/scheduler/backup.go b/internal/scheduler/backup.go index 958ecd57..558be255 100644 --- a/internal/scheduler/backup.go +++ b/internal/scheduler/backup.go @@ -17,7 +17,7 @@ import ( // since the first seed, but nothing ever verified that a backup actually // happened — a silent backup failure looked exactly like a working one. This // makes staleness a Signal like any other, so it flows through the existing -// dedup, auto-resolve and notifier path rather than needing its own machinery. +// dedup, auto-resolve and webhook path rather than needing its own machinery. // // Config: {"path": "/opt/oikos/backups", "max_age_s": 86400, "host": …} // diff --git a/scripts/cutover-checklist.md b/scripts/cutover-checklist.md index 2fb750fd..e5fbcdf0 100644 --- a/scripts/cutover-checklist.md +++ b/scripts/cutover-checklist.md @@ -6,7 +6,7 @@ Status: [x] = done, [ ] = pending - [x] **Backup**: `pg_dump oikos > backups/pre-cutover-20260707.sql` (145K) - [x] **CI green**: pushed to main, `.gitea/workflows/ci.yml` exists -- [x] **Deploy test**: Docker stack running with api + scheduler + notifier + nomos +- [x] **Deploy test**: Docker stack running with api + scheduler + nomos - [x] **Caddy config**: `compose/caddy/Caddyfile.oikos` pushed to `dtoro/caddy-conf` (ed20908). Auto-deploys to caddy (121). - [x] **DNS**: `oikos.hubris.network` already resolves to 192.168.8.175 (mac-mini mesh) - [x] **Secrets**: Infisical bootstrapped + migration complete 2026-07-07. All 11 SOPS secrets migrated to Infisical (oikos project, dev env). Machine identity `oikos-api` has RW access verified via Go SDK. ENCRYPTION_KEY must be 32-char raw string (docs incorrect). SOPS fallback preserved for DR. secrets-issuance decommissioned — stopped/disabled on apps/105; superseded by Infisical. diff --git a/scripts/deploy.sh b/scripts/deploy.sh index ba5200a5..3ccb8f2f 100755 --- a/scripts/deploy.sh +++ b/scripts/deploy.sh @@ -8,8 +8,8 @@ # D2 — versioned images: tags every built image v$VERSION (from VERSION file), # keeps the last 3 tags per service for rollback. -# Notify on deploy failure via Matrix. Uses Oikos API to raise an event -# so the scheduler picks it up and alerts via the notifier. +# Notify on deploy failure. Uses Oikos API to raise an event +# so the scheduler picks it up. notify_deploy_failure() { local reason="$1" local sha="${SHA:-unknown}" @@ -22,7 +22,7 @@ notify_deploy_failure() { -d "{\"type\":\"deploy.failed\",\"severity\":\"critical\",\"source\":\"webhook\",\"data\":{\"sha\":\"$sha\",\"reason\":\"$reason\"}}" \ >/dev/null 2>&1 || true fi - # Also try Matrix directly via the notifier's webhook endpoint if configured + # if configured if [ -n "${MATRIX_WEBHOOK_URL:-}" ]; then curl -sf -X POST "$MATRIX_WEBHOOK_URL" \ -H "Content-Type: application/json" \ @@ -238,7 +238,7 @@ echo "[6/8] prune old image tags (keep 3)" if [ -n "$OIKOS_VERSION" ]; then images=$(docker compose --profile "$PROFILE" config --images 2>/dev/null || true) if [ -z "$images" ]; then - images="oikos-api oikos-scheduler oikos-notifier oikos-migrate oikos-seed oikos-nomos oikos-web" + images="oikos-api oikos-scheduler oikos-migrate oikos-seed oikos-nomos oikos-web" fi printf '%s\n' $images | sed 's/:.*//' | grep '^oikos-' | sort -u | while read -r repo; do docker image ls "$repo" --format '{{.Tag}}' 2>/dev/null | grep '^v' | sort -rV | tail -n +4 | while read -r tag; do diff --git a/scripts/seed-secrets.sh b/scripts/seed-secrets.sh index 57ef15d4..79823bd7 100755 --- a/scripts/seed-secrets.sh +++ b/scripts/seed-secrets.sh @@ -49,16 +49,12 @@ seed_key() { fi } -matrix_token="$(get_container_env notifier OIKOS_MATRIX_TOKEN)" -approval_hmac="$(get_container_env notifier OIKOS_APPROVAL_HMAC_SECRET)" mcp_token="$(get_container_env api OIKOS_MCP_BEARER_TOKEN)" openrouter_key="$(get_container_env nomos OPENROUTER_API_KEY)" webhook_hmac="$(get_container_env api WEBHOOK_HMAC_SECRET 2>/dev/null)" api_token="$mcp_token" -seed_key "matrix_token" "$matrix_token" -seed_key "approval_hmac-secret" "$approval_hmac" seed_key "mcp_bearer-token" "$mcp_token" seed_key "api_token" "$api_token" seed_key "openrouter_api-key" "$openrouter_key"