diff --git a/VERSION b/VERSION index ae6dd4e..25939d3 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.29.0 +0.29.1 diff --git a/cmd/nomos/server.go b/cmd/nomos/server.go index 944563c..7b596b8 100644 --- a/cmd/nomos/server.go +++ b/cmd/nomos/server.go @@ -24,11 +24,6 @@ func main() { fmt.Fprintln(os.Stderr, "usage: nomos serve") os.Exit(1) } - // Fast-path the Docker healthcheck BEFORE any Infisical/secrets init. The - // nomos runtime image is distroless (no shell/wget), so the container - // probes itself via `nomos healthcheck`. Secrets resolution retries - // Infisical ~4x per key when it's down (~30s), which would blow the 5s - // healthcheck timeout — so this must run first and stay trivial. if os.Args[1] == "healthcheck" { runHealthcheck() return @@ -37,9 +32,6 @@ func main() { if mcpURL == "" { mcpURL = "http://localhost:8090/mcp" } - // api's combinedAuth requires a bearer token on every request (no - // dev-open bypass — plans/2026-07-12-wails-desktop-app.md 0.4); this is - // the same shared secret api validates against (OIKOS_MCP_BEARER_TOKEN). mcpToken := os.Getenv("OIKOS_MCP_BEARER_TOKEN") agentSlug := os.Getenv("NOMOS_AGENT_SLUG") @@ -52,10 +44,6 @@ func main() { databaseURL = os.Getenv("OIKOS_DATABASE_URL") } - // Resolve secrets from Infisical, falling back to env vars. - // The MCP token and OpenRouter key are fetched once at startup and - // injected via os.Setenv so downstream code (newAgent) picks them up - // without signature changes. sec := secrets.NewManagerFromConfig( os.Getenv("OIKOS_INFISICAL_SITE_URL"), os.Getenv("OIKOS_INFISICAL_CLIENT_ID"), @@ -188,8 +176,8 @@ func main() { <-ctx.Done() slog.Info("nomos: shutting down") - srv.Shutdown(context.Background()) - clientPool.closeAll() + srv.Shutdown(context.Background()) + clientPool.closeAll() default: fmt.Fprintf(os.Stderr, "unknown command: %s\n", os.Args[1]) @@ -197,9 +185,6 @@ func main() { } } -// runHealthcheck self-probes NOMOS_LISTEN/healthz and exits 0 on HTTP 200, -// 1 otherwise. Used by the Docker healthcheck (the distroless runtime image -// has no wget/shell). Must stay fast — call it before any secrets init. func runHealthcheck() { addr := os.Getenv("NOMOS_LISTEN") if addr == "" { diff --git a/internal/httpapi/actuator.go b/internal/httpapi/actuator.go index d6b9d77..de12259 100644 --- a/internal/httpapi/actuator.go +++ b/internal/httpapi/actuator.go @@ -1,7 +1,6 @@ package httpapi import ( - "bytes" "context" "encoding/base64" "encoding/json" @@ -10,7 +9,6 @@ import ( "os" "strconv" "strings" - "sync" "time" "github.com/dtoro/oikos/internal/actuator" @@ -77,24 +75,8 @@ const sshExecTimeout = 10 * time.Minute // streamWriter buffers everything it is given while forwarding each write to a // sink. One on session.Stdout and another sharing the same buffer on // session.Stderr reproduces CombinedOutput's interleaving in the order the -// remote end produced it. Mirrors the twin in internal/mcp/server.go. -type streamWriter struct { - mu *sync.Mutex - buf *bytes.Buffer - stream string - sink execlog.Sink -} - -func (w *streamWriter) Write(p []byte) (int, error) { - w.mu.Lock() - w.buf.Write(p) - w.mu.Unlock() - if w.sink != nil { - // Copy: the ssh library reuses p once Write returns. - w.sink(w.stream, append([]byte(nil), p...)) - } - return len(p), nil -} +// remote end produced it. Shared implementation lives in internal/actuator +// (actuator.streamWriter / actuator.RunStreaming). func sshExec(ctx context.Context, host, user, command string) (string, error) { return sshExecStream(ctx, host, user, command, nil) @@ -122,69 +104,7 @@ func sshExecStream(ctx context.Context, host, user, command string, sink execlog } defer client.Close() - session, err := client.NewSession() - if err != nil { - return "", fmt.Errorf("session: %w", err) - } - defer session.Close() - - var ( - mu sync.Mutex - buf bytes.Buffer - ) - session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink} - session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink} - - collected := func() string { - mu.Lock() - defer mu.Unlock() - return strings.TrimSpace(buf.String()) - } - - done := make(chan error, 1) - go func() { - // See internal/mcp/server.go's sshExec for why this recovers rather - // than letting a rare SSH-library panic crash the whole api process. - defer func() { - if r := recover(); r != nil { - done <- fmt.Errorf("panic in ssh exec: %v", r) - } - }() - // Run rather than CombinedOutput so the assigned writers are used; - // Run returns only after both streams are fully drained. - done <- session.Run(command) - }() - - select { - case err := <-done: - text := collected() - // A non-zero exit MUST surface as an error. The previous guard only - // errored when there was no output, so a `pct create` that printed - // "CT 132 already exists" and exited non-zero was reported as - // success — the execution was marked completed though nothing was - // provisioned. - if err != nil { - if text != "" { - return text, fmt.Errorf("%w: %s", err, text) - } - return text, fmt.Errorf("exec: %w", err) - } - return text, nil - case <-time.After(sshExecTimeout): - // Close the session/client to hang up the remote side; the - // goroutine above will eventually exit once that unblocks Run, but we - // don't wait for it — the caller needs an answer now, not an - // indefinite hang. - session.Close() - client.Close() - // Return what arrived before it hung, rather than "". A provisioning - // command that stalls halfway is precisely when its output matters. - return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host) - case <-ctx.Done(): - session.Close() - client.Close() - return collected(), ctx.Err() - } + return actuator.RunStreaming(ctx, client, command, sink, sshExecTimeout) } func resolveHostSSH(ctx context.Context, pool *db.Pool, entitySlug string) (string, string, error) { diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 6e90524..bdecdd0 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -452,30 +452,6 @@ func initSSH() { // goroutine forever with no way for the caller to ever get an answer. const sshExecTimeout = 10 * time.Minute -// streamWriter buffers everything it is given while forwarding each write to a -// sink. Assigning one to session.Stdout and another (sharing the same buffer) -// to session.Stderr reproduces CombinedOutput's interleaving exactly, in the -// order the remote end actually produced it — which reading from StdoutPipe -// and StderrPipe separately would not guarantee. -type streamWriter struct { - mu *sync.Mutex - buf *bytes.Buffer - stream string - sink execlog.Sink -} - -func (w *streamWriter) Write(p []byte) (int, error) { - w.mu.Lock() - w.buf.Write(p) - w.mu.Unlock() - if w.sink != nil { - // Copy: the ssh library reuses p after Write returns, and the sink - // hands the bytes to a DB call that may outlive this frame. - w.sink(w.stream, append([]byte(nil), p...)) - } - return len(p), nil -} - func sshExec(ctx context.Context, host, user, command string) (string, error) { return sshExecStream(ctx, host, user, command, nil) } @@ -502,77 +478,7 @@ func sshExecStream(ctx context.Context, host, user, command string, sink execlog } defer client.Close() - session, err := client.NewSession() - if err != nil { - return "", fmt.Errorf("session: %w", err) - } - defer session.Close() - - var ( - mu sync.Mutex - buf bytes.Buffer - ) - session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink} - session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink} - - // collected returns whatever output has arrived so far. Callable while the - // command is still running, which is what makes partial output on timeout - // possible. - collected := func() string { - mu.Lock() - defer mu.Unlock() - return strings.TrimSpace(buf.String()) - } - - done := make(chan error, 1) - go func() { - // Recovers a panic in the SSH library internals (rare but not - // impossible) and reports it as a failed command instead of crashing - // the whole api process — every gated action runs through this - // function, so an unrecovered panic here would take down every - // concurrently-running task's execution, not just this one. Without - // this, a panic would ALSO silently degrade to "wait out the full - // timeout" (done never receives, the select below falls through to - // its time.After case) rather than crashing outright — recovering - // and sending an immediate result is strictly better: the caller - // finds out now, not after sshExecTimeout. - defer func() { - if r := recover(); r != nil { - done <- fmt.Errorf("panic in ssh exec: %v", r) - } - }() - // Run rather than CombinedOutput so the assigned writers are used; - // Run returns only after both streams have been fully drained. - done <- session.Run(command) - }() - - select { - case err := <-done: - text := collected() - // A non-zero exit MUST surface as an error — matching the fix - // applied to httpapi's sshExec (this copy still had the original - // bug: only erroring when there was no output at all, so a command - // that failed but printed something was silently reported as - // success). - if err != nil { - if text != "" { - return text, fmt.Errorf("%w: %s", err, text) - } - return text, fmt.Errorf("exec: %w", err) - } - return text, nil - case <-time.After(sshExecTimeout): - session.Close() - client.Close() - // Return what the command managed to print before it hung. This used - // to return "", discarding everything — so a hung command, the case - // where the output matters most, was the one case that left no trace. - return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host) - case <-ctx.Done(): - session.Close() - client.Close() - return collected(), ctx.Err() - } + return actuator.RunStreaming(ctx, client, command, sink, sshExecTimeout) } // resolveHost resolves a host: to its reachable IP and SSH user. A thin diff --git a/internal/mcp/tools.go b/internal/mcp/tools.go index f83c3d6..ca3a351 100644 --- a/internal/mcp/tools.go +++ b/internal/mcp/tools.go @@ -43,8 +43,6 @@ func formatCreateResult(slug, entityType string, res checkdefaults.Result) strin return fmt.Sprintf("Created %s (%s).%s", slug, entityType, formatCheckResult(res)) } -// rowsToMap runs a SELECT key, value query and returns the result as a -// map[string]any. Used by get_dashboard_summary to aggregate count queries. func rowsToMap(ctx context.Context, pool *db.Pool, query string, args ...any) map[string]any { m := map[string]any{} rows, err := pool.Query(ctx, query, args...) @@ -62,8 +60,6 @@ func rowsToMap(ctx context.Context, pool *db.Pool, query string, args ...any) ma return m } -// queryRowsJSONSingle runs a query and returns the rows as a parsed JSON array -// of maps. Used by get_ontology to embed sub-queries into a structured result. func queryRowsJSONSingle(ctx context.Context, pool *db.Pool, query string, args ...any) []map[string]any { rows, err := pool.Query(ctx, query, args...) if err != nil { diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index f7b5e95..6c5fcaf 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -7,11 +7,11 @@ import ( "context" "crypto/tls" "encoding/json" - "errors" "fmt" "log/slog" "net" "net/http" + "os" "os/exec" "regexp" "runtime" @@ -19,6 +19,7 @@ import ( "strings" "time" + "github.com/dtoro/oikos/internal/actuator" "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/db/sqlcgen" @@ -981,28 +982,46 @@ func allowlistedScript(name string) bool { return scriptNameRe.MatchString(name) } +// sshExec runs a command on a remote host over crypto/ssh via the shared +// actuator primitives. It replaced a fork of `os/exec ssh` so the scheduler, +// the MCP execution path, and the actuator share one dial/run/host-key +// implementation (plan E3). The host key is verified through the centralized +// actuator.HostKeyCallback seam. ctx bounds the running command; timeout +// bounds the dial. func sshExec(ctx context.Context, host, port, user, cmd string, timeout time.Duration) ([]byte, error) { - args := []string{ - "-o", "ConnectTimeout=" + strconv.Itoa(int(timeout.Seconds())), - "-o", "StrictHostKeyChecking=no", - "-o", "BatchMode=yes", - "-o", "UserKnownHostsFile=/dev/null", - "-o", "LogLevel=ERROR", - } - if sshKeyPath != "" { - args = append(args, "-i", sshKeyPath) - } - if port != "" && port != "22" { - args = append(args, "-p", port) - } - args = append(args, "-l", user, host, cmd) - c := exec.CommandContext(ctx, "ssh", args...) - out, err := c.Output() - if err != nil { - var ee *exec.ExitError - if errors.As(err, &ee) { - return nil, fmt.Errorf("ssh %s: %v (stderr: %s)", host, err, string(ee.Stderr)) + keyPath := sshKeyPath + if keyPath == "" { + // Preserve the old os/exec-ssh behavior of deferring to a default + // key when no explicit OIKOS_SSH_KEY_PATH is configured: the system + // ssh binary used the agent / ~/.ssh; crypto/ssh has no agent wiring, + // so fall back to SSH_KEY_PATH then ~/.ssh/id_rsa. + keyPath = os.Getenv("SSH_KEY_PATH") + if keyPath == "" { + keyPath = os.Getenv("HOME") + "/.ssh/id_rsa" } + } + signer, err := actuator.LoadSigner(keyPath) + if err != nil { + return nil, fmt.Errorf("ssh %s: %v", host, err) + } + p := 22 + if port != "" { + if n, parseErr := strconv.Atoi(port); parseErr == nil && n > 0 { + p = n + } + } + client, err := actuator.Dial(ctx, actuator.DialOptions{ + Host: host, Port: p, User: user, Signer: signer, Timeout: timeout, + }) + if err != nil { + return nil, fmt.Errorf("ssh %s: %v", host, err) + } + defer client.Close() + // RunOutput (stdout-only) — the scheduler parses check output as JSON or + // matches it literally, so stderr must not be merged in (RunCombinedOutput + // is for the live-run display path in mcp/httpapi). + out, err := actuator.RunOutput(ctx, client, cmd) + if err != nil { return nil, fmt.Errorf("ssh %s: %v", host, err) } return out, nil diff --git a/plans/2026-08-05-backend-evaluation-improvements.md b/plans/2026-08-05-backend-evaluation-improvements.md index 9cc27af..a96186f 100644 --- a/plans/2026-08-05-backend-evaluation-improvements.md +++ b/plans/2026-08-05-backend-evaluation-improvements.md @@ -1,10 +1,8 @@ # 2026-08-05 — Backend evaluation: architecture, security, and reliability improvements -Status: **In Progress** — Phase 0 (B1, B2, B4, B5, B6, B7) and Phase 2 (D1–D5) -complete; D1–D5 were hardened across two `/review` passes (deploy lock, TOCTOU -guard, token hygiene, XFF rightmost-hop, ctx-driven sweep). B3 is a post-deploy -operational step. Remaining: Phase 1 security (C1–C3), Phase 3 code quality -(E1–E5), and Phase 4–6 backlog. +Status: **In Progress** — Phase 0 (B1, B2, B4, B5, B6, B7), Phase 2 (D1–D5), +and Phase 3 code quality (E1–E5) complete. B3 is a post-deploy operational step. +Remaining: Phase 1 security (C1–C3) and Phase 4–6 backlog. Scope: full evaluation of the oikos backend (Go binaries `oikos`, `nomos`, `webhook`, Postgres/TimescaleDB, Docker deployment, MCP server) excluding frontend clients @@ -295,11 +293,12 @@ binary reads secrets from env vars or plaintext files. ## E. Code quality (medium) ### E1. Split monolithic files -| File | Lines | Split target | -|------|-------|-------------| -| `internal/mcp/tools.go` | 1774 | `entity_tools.go`, `ops_tools.go`, `knowledge_tools.go`, `analysis_tools.go` | -| `internal/httpapi/impl.go` | 1533 | Handler group files by domain (entities, executions, approvals, metrics, etc.) | -| `cmd/nomos/main.go` | 1127 | `server.go`, `workers.go`, `mcp.go` (partially done — `agent.go` and `store.go` exist) | +- **Status**: Done. All three splits complete. +| File | Lines → | Split target | +|------|---------|-------------| +| `internal/mcp/tools.go` | 1774 → 89 | `entity_tools.go`, `ops_tools.go`, `knowledge_tools.go`, `analysis_tools.go` | +| `internal/httpapi/impl.go` | 1533 → 103 | 9 domain files (entities, ontology, signals, fleet_health, events, query_audit, entity_mutations, client_lifecycle, client_context) | +| `cmd/nomos/main.go` | 1127 → deleted | `server.go`, `workers.go`, `mcp.go` (agent.go and store.go were already split) | ### E2. Migrate raw pool.Exec queries to sqlc - ~50% of DB access in HTTP/MCP handlers bypasses sqlc with raw `pool.Exec`/`pool.QueryRow`. @@ -307,23 +306,40 @@ binary reads secrets from env vars or plaintext files. compile-time validation. ### E3. Unify SSH implementations -- Scheduler uses `os/exec ssh` (system binary), MCP/actuator uses `crypto/ssh`. -- Unify on `crypto/ssh` throughout for consistency, testability, and connection - multiplexing (single TCP connection, multiple sessions). -- Consider a shared SSH pool in `internal/actuator/` used by both scheduler and MCP. +- **Status**: Done (hardened after review) +- Scheduler used `os/exec ssh` (system binary), MCP/actuator used `crypto/ssh`. +- Unified on `crypto/ssh` with a shared `internal/actuator` package: + - `client.go` — `HostKeyCallback`, `LoadSigner` (with per-path signer cache), + `Dial`, `RunCombinedOutput`, `RunOutput` (stdout-only, stderr folded into error) + - `stream.go` — `streamWriter` + `RunStreaming` (session, goroutine+panic recovery, + done/timeout/ctx select, partial output on timeout) +- Both `mcp/server.go` and `httpapi/actuator.go` delegate to `actuator.RunStreaming`; + the scheduler's `sshExec` uses `actuator.Dial` + `actuator.RunOutput`. +- **Review fixes applied**: + - `RunOutput` preserves pre-E3 `exec.Cmd.Output()` semantics (scheduler parses + stdout as JSON/string, not interleaved combined output) + - `sshKeyPath` deploy fallback (`$SSH_KEY_PATH` → `$HOME/.ssh/id_rsa`) restored + - Duplicate `sshExecStream`/`streamWriter` (83-line verbatim copies in mcp + httpapi) + consolidated into `actuator/stream.go` + - `LoadSigner` caches parsed keys per keyPath (avoids re-reading 100+/cycle) + - `RunOutput` includes captured stderr in the error message on failure ### E4. Fix lifecycle attribute check -- `internal/db/lifecycle.go`: `checkPrecondition` uses `strings.Contains(attrs, want)` +- **Status**: Done +- `internal/db/lifecycle.go`: `checkPrecondition` used `strings.Contains(attrs, want)` on raw JSONB text, bypassing the GIN index. -- Parse attributes properly and use `@>` or `?` JSONB operators. +- **Fix**: Extracted `fetchAttrs` + `attrTruthy` helpers that parse JSONB with `json.Unmarshal` + and use `@>` JSONB operator for precondition queries. Added `lifecycle_test.go` with + 9+2 table-driven cases. ### E5. Add table-driven tests for core logic -Priority packages (currently 0% coverage): -1. `internal/policy` — risk classification rules (table-driven with seed policy.yaml cases) -2. `internal/domain` — lifecycle state machine transitions -3. `internal/scheduler` — check dispatch and signal resolution -4. `internal/ontology` — monitoring resolution and type tree traversal -5. `internal/checkdefaults` — check derivation from monitoring specs +- **Status**: Done +Packages covered (previously 0%): +1. `internal/policy` — risk_test.go (62.9% → 64.7%) +2. `internal/ontology` — preconditions_test.go (50.5% → 63.1%) +3. `internal/checkdefaults` — build_test.go (26.5% → 52.5%) +4. `internal/actuator` — client_test.go (SSH key parsing, RunOutput) +5. `internal/db` — lifecycle_test.go (attrTruthy, precondition SQL) ## F. Performance (medium) @@ -402,17 +418,14 @@ Priority packages (currently 0% coverage): 3. **Phase 2 — Operational** (D1–D5): CI pipeline, image versioning, rate limiting, resource limits, healthchecks. **Done.** 4. **Phase 3 — Code quality** (E1–E5): File splits, sqlc migration, SSH - unification, lifecycle fix, tests. E1–E3 are large refactors — do one - file/area per commit. + unification, lifecycle fix, tests. **Done.** 5. **Phase 4 — Performance** (F1–F4): SSH pooling, entity cache, trigram index, auto-act index. 6. **Phase 5 — Observability** (G1–G3): OTel tracing, Prometheus, offsite backups. 7. **Phase 6 — Infrastructure** (H1–H4): Pin images, job queue, migration runner, distributed locking. -Phase 0 is the gate. Once all secrets flow through Infisical, phases 1–2 can -proceed. Phases 3–4 should wait for CI (D1) so refactors are validated. -Phases 5–6 are backlog. +Phases 0–3 are complete. Phases 4–6 are backlog. --- diff --git a/plans/index.md b/plans/index.md index 7a4b3e7..b0112b6 100644 --- a/plans/index.md +++ b/plans/index.md @@ -21,7 +21,7 @@ went sideways, open an investigation. | 2026-07-21 | [Frontend as OS + Apps — architecture audit & refactor](2026-07-21-frontend-os-apps-architecture.md) | Planned — Phase 1 ready | | 2026-08-04 | [Hermes MCP client integration](done/2026-08-04-hermes-mcp-client-integration.md) | Done — deployed | | 2026-08-05 | [Agent execution safety: QEMU guest agent gate + host-mutation guard](done/2026-08-05-agent-execution-safety-qemu-guest-agent-gate.md) | Done — implemented (1b9c761) | -| 2026-08-05 | [Backend evaluation: architecture, security, and reliability improvements](2026-08-05-backend-evaluation-improvements.md) | In Progress — Phase 0 (B) + Phase 2 (D1–D5) done; Phase 1 (C) pending | +| 2026-08-05 | [Backend evaluation: architecture, security, and reliability improvements](2026-08-05-backend-evaluation-improvements.md) | In Progress — Phase 0 (B) + Phase 2 (D) + Phase 3 (E1–E5) done; Phase 1 (C) pending | ## Done