dtoro
60c0432d8b
Seed ingest/export moves behind ports.SeedRepository (SeedRepo in the postgres adapter; knowledge ingest absorbed from internal/knowledge, package deleted). pct_create flow (defaults, template/VMID/gateway pre-flights, pct create, graph registration) moves from httpapi's approved-execution path into app.ProvisioningService + the ssh provisioner adapter; CLI seed/export/secret become adapters over the services. EntityCreateInput gains EnrolledAt. Plan status corrected: phases 0-7 shipped, 8 + 9 gates open. VERSION 0.34.1.
412 lines
17 KiB
Go
412 lines
17 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/dtoro/oikos/internal/actuator"
|
|
"github.com/dtoro/oikos/internal/adapters/postgres"
|
|
"github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen"
|
|
"github.com/dtoro/oikos/internal/core/app"
|
|
"github.com/dtoro/oikos/internal/execlog"
|
|
"github.com/dtoro/oikos/internal/observability"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
var (
|
|
_sshUser string
|
|
_sshKey []byte
|
|
)
|
|
|
|
// flexBool accepts a JSON bool, number (0/1), or string ("true"/"1"/"yes").
|
|
// LLMs routinely emit `"privileged": 0` instead of `false`; a strict `bool`
|
|
// field made the approved pct_create execution fail to parse *after* the
|
|
// operator had already approved it — the container was never created and the
|
|
// operator saw "queued" with no result. This type tolerates the common shapes.
|
|
type flexBool bool
|
|
|
|
func (b *flexBool) UnmarshalJSON(data []byte) error {
|
|
s := strings.TrimSpace(strings.Trim(string(data), `"`))
|
|
switch strings.ToLower(s) {
|
|
case "true", "1", "yes", "on":
|
|
*b = true
|
|
case "false", "0", "no", "off", "", "null":
|
|
*b = false
|
|
default:
|
|
return fmt.Errorf("cannot parse %q as bool", s)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func initSSH() {
|
|
if _sshUser == "" {
|
|
_sshUser = os.Getenv("OIKOS_SSH_USER")
|
|
if _sshUser == "" {
|
|
_sshUser = "root"
|
|
}
|
|
}
|
|
if len(_sshKey) == 0 {
|
|
keyPath := os.Getenv("OIKOS_SSH_KEY_PATH")
|
|
if keyPath == "" {
|
|
keyPath = "/etc/oikos/ssh_key"
|
|
}
|
|
var err error
|
|
_sshKey, err = os.ReadFile(keyPath)
|
|
if err != nil {
|
|
slog.Warn("httpapi ssh: cannot read key", "path", keyPath, "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// sshExecTimeout bounds how long a single remote command may run. Without
|
|
// this, a hung remote command (e.g. a piped install script stuck retrying
|
|
// DNS against a misconfigured gateway) blocks the executing goroutine
|
|
// forever: the execution never leaves 'approved'/'running', the operator
|
|
// sees an unkillable spinner, and get_execution_status has nothing new to
|
|
// report. Generous enough for a real apt/docker install; not infinite.
|
|
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. Shared implementation lives in internal/actuator
|
|
// (actuator.streamWriter / actuator.RunStreaming).
|
|
|
|
// sshExecStream runs a command and reports its combined output, forwarding
|
|
// each chunk to sink as it arrives. A nil sink behaves exactly as before.
|
|
func sshExecStream(ctx context.Context, host, user, command string, sink execlog.Sink) (string, error) {
|
|
initSSH()
|
|
if len(_sshKey) == 0 {
|
|
return "", fmt.Errorf("no SSH key available")
|
|
}
|
|
if user == "" {
|
|
user = _sshUser
|
|
}
|
|
|
|
signer, err := actuator.LoadSignerFromBytes(_sshKey)
|
|
if err != nil {
|
|
return "", fmt.Errorf("parse key: %w", err)
|
|
}
|
|
|
|
client, err := actuator.Dial(ctx, actuator.DialOptions{Host: host, User: user, Signer: signer})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer client.Close()
|
|
|
|
return actuator.RunStreaming(ctx, client, command, sink, sshExecTimeout)
|
|
}
|
|
|
|
func resolveHostSSH(ctx context.Context, pool *db.Pool, entitySlug string) (string, string, error) {
|
|
var attrs string
|
|
err := pool.QueryRow(ctx, "SELECT attributes::text FROM entities WHERE slug = $1", entitySlug).Scan(&attrs)
|
|
if err != nil {
|
|
return "", "", fmt.Errorf("entity not found: %s", entitySlug)
|
|
}
|
|
|
|
var m map[string]interface{}
|
|
if err := json.Unmarshal([]byte(attrs), &m); err != nil {
|
|
return "", "", fmt.Errorf("parse attributes: %w", err)
|
|
}
|
|
|
|
sshUser := _sshUser
|
|
if sshUser == "" {
|
|
sshUser = "root"
|
|
}
|
|
|
|
if ip, ok := m["lan_ip"].(string); ok && ip != "" {
|
|
return ip, sshUser, nil
|
|
}
|
|
if mesh, ok := m["mesh"].(map[string]interface{}); ok {
|
|
for _, proto := range []string{"netbird", "tailscale"} {
|
|
if p, ok := mesh[proto].(map[string]interface{}); ok {
|
|
if ip, ok := p["ip"].(string); ok && ip != "" {
|
|
return ip, sshUser, nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return "", "", fmt.Errorf("no IP found for %s", entitySlug)
|
|
}
|
|
|
|
// resolveRunTarget mirrors internal/mcp.resolveExecTarget for the approved-
|
|
// execution side: any target slug (host: or lxc:) resolves to the SSH
|
|
// endpoint that runs the command plus a wrap function that turns a plain
|
|
// shell command into what actually needs to be sent — identity for a host,
|
|
// `pct exec <pve_id>` for an LXC. Kept as a small duplicate rather than a
|
|
// cross-package import to avoid coupling httpapi to mcp for one helper.
|
|
func resolveRunTarget(ctx context.Context, pool *db.Pool, targetSlug string) (host, user string, wrap func(string) string, err error) {
|
|
if strings.HasPrefix(targetSlug, "host:") {
|
|
host, user, err = resolveHostSSH(ctx, pool, targetSlug)
|
|
return host, user, func(cmd string) string { return cmd }, err
|
|
}
|
|
if strings.HasPrefix(targetSlug, "lxc:") {
|
|
var pveID, hostAttr string
|
|
// COALESCE the host column: many older LXC entities (seeded from
|
|
// inventory, not provisioned by pct_create) have pve_id but no host
|
|
// attribute at all. Scanning a SQL NULL into a plain string errors
|
|
// the whole row, wrongly reporting "missing pve_id" even when it was
|
|
// present — COALESCE avoids the NULL, "" is handled below.
|
|
if qerr := pool.QueryRow(ctx, "SELECT attributes->>'pve_id', COALESCE(attributes->>'host', '') FROM entities WHERE slug = $1", targetSlug).Scan(&pveID, &hostAttr); qerr != nil || pveID == "" {
|
|
return "", "", nil, fmt.Errorf("LXC not found or missing pve_id: %s", targetSlug)
|
|
}
|
|
hostSlug := hostAttr
|
|
if hostSlug == "" {
|
|
hostSlug = "hubris"
|
|
}
|
|
if !strings.HasPrefix(hostSlug, "host:") {
|
|
hostSlug = "host:" + hostSlug
|
|
}
|
|
host, user, err = resolveHostSSH(ctx, pool, hostSlug)
|
|
id := pveID
|
|
return host, user, func(cmd string) string {
|
|
b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
|
|
return fmt.Sprintf("pct exec %s -- bash -c 'echo %s | base64 -d | bash'", id, b64)
|
|
}, err
|
|
}
|
|
return "", "", nil, fmt.Errorf("unsupported target %q: must be host:<slug> or lxc:<slug>", targetSlug)
|
|
}
|
|
|
|
// executeApprovedAction runs a gated action after operator approval.
|
|
// Runs in a background goroutine to not block the HTTP response.
|
|
// emitExecutionEvent records an execution lifecycle event for SSE fan-out so
|
|
// the control room can watch approved actions run to completion live.
|
|
func emitExecutionEvent(ctx context.Context, pool *db.Pool, execID uuid.UUID, status string, detail map[string]any) {
|
|
severity := "info"
|
|
if status == "failed" {
|
|
severity = "warning"
|
|
}
|
|
// The correlation id was hardcoded to "", so execution events could not be
|
|
// tied back to the session that caused them — the one join you want when
|
|
// asking "what did this agent turn actually do?". It is already on the
|
|
// execution row; read it rather than threading it through eleven callers.
|
|
var correlationID string
|
|
if err := pool.QueryRow(ctx,
|
|
`SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); err != nil {
|
|
correlationID = ""
|
|
}
|
|
_ = observability.Event(ctx, sqlcgen.New(pool), "execution."+status, &execID, severity, "actuator", correlationID, detail)
|
|
if status == "completed" || status == "failed" || status == "cancelled" {
|
|
closePlanStepForExecution(ctx, pool, execID, status)
|
|
}
|
|
}
|
|
|
|
// closePlanStepForExecution auto-closes a task plan step whose linked execution
|
|
// just reached a terminal state, so the task board advances even if the agent
|
|
// doesn't call update_plan_step itself (belt and suspenders — the agent links
|
|
// the step to the execution when it starts it; the api finishes it here). Emits
|
|
// plan.step.finished correlated to the step's session. No-op for the vast
|
|
// majority of executions, which aren't plan steps.
|
|
func closePlanStepForExecution(ctx context.Context, pool *db.Pool, execID uuid.UUID, execStatus string) {
|
|
stepStatus := "done"
|
|
if execStatus == "failed" || execStatus == "cancelled" {
|
|
stepStatus = "failed"
|
|
}
|
|
var stepID, sessionID string
|
|
var seq int
|
|
if err := pool.QueryRow(ctx, `
|
|
UPDATE session_plan_steps SET status = $2, finished_at = now()
|
|
WHERE execution_id = $1 AND status NOT IN ('done', 'failed', 'skipped')
|
|
RETURNING id::text, session_id::text, seq`, execID, stepStatus).Scan(&stepID, &sessionID, &seq); err != nil {
|
|
return // no matching open step
|
|
}
|
|
_ = observability.Event(ctx, sqlcgen.New(pool), "plan.step.finished", &execID, "info", "actuator", sessionID,
|
|
map[string]any{"step_id": stepID, "seq": seq, "status": stepStatus, "execution_id": execID.String()})
|
|
}
|
|
|
|
func (s *Server) executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, targetSlug string, actionStr string) {
|
|
slog.Info("httpapi: executing approved action", "execution_id", execID, "target", targetSlug, "action", actionStr)
|
|
|
|
host, user, wrap, err := resolveRunTarget(ctx, pool, targetSlug)
|
|
if err != nil {
|
|
slog.Error("httpapi: resolve host for approved execution", "error", err, "target", targetSlug)
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
|
|
execID, jsonErr("%s", err.Error()))
|
|
emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": err.Error()})
|
|
return
|
|
}
|
|
|
|
idx := strings.Index(actionStr, ":")
|
|
if idx < 0 {
|
|
slog.Error("httpapi: malformed action string (no colon)", "action", actionStr)
|
|
return
|
|
}
|
|
action, params := actionStr[:idx], actionStr[idx+1:]
|
|
|
|
startedAt := time.Now()
|
|
// Persist started_at now, not at the end. It was captured here but only
|
|
// written in the terminal UPDATE, so a running execution reported
|
|
// started_at = NULL for its entire life — the UI could not show how long
|
|
// anything had been going, which is exactly when you want to know.
|
|
if _, err := pool.Exec(ctx,
|
|
`UPDATE executions SET status = 'running', started_at = $2 WHERE entity_id = $1`,
|
|
execID, startedAt); err != nil {
|
|
slog.Error("httpapi: mark execution running", "error", err, "execution_id", execID)
|
|
}
|
|
|
|
// Stream output for the actions whose output an operator actually watches:
|
|
// a long apt upgrade, a pct create, an arbitrary approved `run`. The small
|
|
// internal lookups further down (listing template cache, pvesh nextid) stay
|
|
// unstreamed — they are plumbing, and logging them would bury the command
|
|
// the operator approved.
|
|
var correlationID string
|
|
if qerr := pool.QueryRow(ctx,
|
|
`SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); qerr != nil {
|
|
correlationID = ""
|
|
}
|
|
sink, flushLogs := execlog.New(ctx, pool, execID, correlationID)
|
|
defer flushLogs()
|
|
|
|
var output, cmd string
|
|
|
|
switch action {
|
|
case "systemctl":
|
|
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
|
switch {
|
|
case strings.HasPrefix(params, "enable:"):
|
|
svc = strings.TrimPrefix(params, "enable:")
|
|
cmd = fmt.Sprintf("systemctl enable %s --now 2>&1; sleep 1; systemctl is-active %s", svc, svc)
|
|
case strings.HasPrefix(params, "disable:"):
|
|
svc = strings.TrimPrefix(params, "disable:")
|
|
cmd = fmt.Sprintf("systemctl disable %s --now 2>&1; sleep 1; systemctl is-active %s", svc, svc)
|
|
default:
|
|
cmd = fmt.Sprintf("systemctl %s %s 2>&1", params, svc)
|
|
}
|
|
output, err = sshExecStream(ctx, host, user, cmd, sink)
|
|
|
|
case "apt_upgrade":
|
|
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
|
cmd = fmt.Sprintf("apt update -qq 2>&1 >/dev/null && apt upgrade -y -qq 2>&1; echo '---'; systemctl is-active %s || true", svc)
|
|
output, err = sshExecStream(ctx, host, user, cmd, sink)
|
|
|
|
case "pct_create":
|
|
var cfg struct {
|
|
VMID int `json:"vmid"`
|
|
Hostname string `json:"hostname"`
|
|
Cores int `json:"cores"`
|
|
Memory int `json:"memory"`
|
|
DiskGB int `json:"disk_gb"`
|
|
IP string `json:"ip"`
|
|
GW string `json:"gw"`
|
|
Bridge string `json:"bridge"` // e.g. vmbr0/vmbr1 — which bridge actually reaches the target subnet on this host varies per host, don't assume vmbr0
|
|
Storage string `json:"storage"`
|
|
Template string `json:"template"`
|
|
Privileged flexBool `json:"privileged"`
|
|
Nesting flexBool `json:"nesting"`
|
|
Mounts []string `json:"mounts"`
|
|
Nameserver string `json:"nameserver"`
|
|
Searchdomain string `json:"searchdomain"`
|
|
// No services/post_install here anymore — pct_create is atomic
|
|
// (create + start + register only). Installing packages and
|
|
// running setup scripts is the agent's job via follow-up `run`
|
|
// calls against lxc:<hostname>, so each step is individually
|
|
// observable and recoverable instead of one opaque multi-minute
|
|
// black box. See the comment above the removed post-create block.
|
|
}
|
|
if err := json.Unmarshal([]byte(params), &cfg); err != nil {
|
|
slog.Error("httpapi: pct_create parse params", "error", err, "params", params)
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
|
|
execID, jsonErr("invalid pct_create params: %v", err))
|
|
emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": err.Error()})
|
|
return
|
|
}
|
|
// The flow (spec defaults, template/VMID pre-flights, pct create,
|
|
// graph registration) lives in ProvisioningService + the ssh
|
|
// provisioner adapter since Phase 7; this handler only maps the
|
|
// wire payload and streams the create output to the execution log.
|
|
outcome, perr := s.provisioning.CreateLXC(ctx, app.CreateLXCCmd{
|
|
HostSlug: targetSlug,
|
|
Hostname: cfg.Hostname,
|
|
VMID: cfg.VMID,
|
|
Cores: cfg.Cores,
|
|
MemoryMB: cfg.Memory,
|
|
DiskGB: cfg.DiskGB,
|
|
IP: cfg.IP,
|
|
GW: cfg.GW,
|
|
Bridge: cfg.Bridge,
|
|
Storage: cfg.Storage,
|
|
Template: cfg.Template,
|
|
Privileged: bool(cfg.Privileged),
|
|
Nesting: bool(cfg.Nesting),
|
|
Mounts: cfg.Mounts,
|
|
Nameserver: cfg.Nameserver,
|
|
Searchdomain: cfg.Searchdomain,
|
|
Sink: sink,
|
|
})
|
|
output = outcome.Output
|
|
err = perr
|
|
if perr == nil {
|
|
emitExecutionEvent(ctx, pool, execID, "executing", map[string]any{
|
|
"lxc_slug": outcome.Slug, "vmid": outcome.VMID, "host": targetSlug,
|
|
})
|
|
}
|
|
|
|
case "run":
|
|
// The general gated primitive: arbitrary shell against any host or
|
|
// LXC, approved and classified by internal/policy.ClassifyCommand at
|
|
// request time (see mcp/server.go's "run" tool). No fixed action
|
|
// enum — new capability doesn't require new Go code here.
|
|
var cfg struct {
|
|
Command string `json:"command"`
|
|
Purpose string `json:"purpose"`
|
|
}
|
|
if perr := json.Unmarshal([]byte(params), &cfg); perr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
|
|
execID, jsonErr("invalid run params: %v", perr))
|
|
emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": perr.Error()})
|
|
return
|
|
}
|
|
cmd = wrap(cfg.Command)
|
|
output, err = sshExecStream(ctx, host, user, cmd, sink)
|
|
|
|
default:
|
|
slog.Error("httpapi: unknown gated action for approved execution", "action", action, "execution_id", execID)
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
|
|
execID, jsonErr("unknown action: %s", action))
|
|
return
|
|
}
|
|
|
|
durationMs := int(time.Since(startedAt).Milliseconds())
|
|
status := "completed"
|
|
verified := true
|
|
// Build result via json.Marshal, not string interpolation. Command output
|
|
// (apt/pct) contains quotes, backslashes and control chars; the old
|
|
// fmt.Sprintf only escaped "\n", producing invalid JSON that failed the
|
|
// ::jsonb cast — so this UPDATE was silently discarded and the execution
|
|
// was stuck at "approved" forever even though provisioning succeeded.
|
|
resMap := map[string]any{"output": output}
|
|
if err != nil {
|
|
resMap["error"] = err.Error()
|
|
status = "failed"
|
|
verified = false
|
|
}
|
|
resultJSON, _ := json.Marshal(resMap)
|
|
|
|
if _, uerr := pool.Exec(ctx, `UPDATE executions SET status=$2, result=$3::jsonb, duration_ms=$4, verified=$5, started_at=$6, completed_at=$7 WHERE entity_id=$1`,
|
|
execID, status, resultJSON, durationMs, verified, startedAt, time.Now()); uerr != nil {
|
|
slog.Error("httpapi: finalize execution status", "error", uerr, "execution_id", execID, "intended_status", status)
|
|
}
|
|
|
|
emitExecutionEvent(ctx, pool, execID, status, map[string]any{
|
|
"action": action, "target": targetSlug, "duration_ms": durationMs,
|
|
})
|
|
|
|
slog.Info("httpapi: approved action executed",
|
|
"execution_id", execID, "action", action, "status", status, "duration_ms", durationMs)
|
|
}
|
|
|
|
// jsonErr builds a valid {"error": "..."} JSON payload for an execution's
|
|
// result column. Always use this instead of fmt.Sprintf'ing JSON by hand —
|
|
// error text and command output routinely contain quotes/backslashes that
|
|
// break a hand-built string and fail the ::jsonb cast.
|
|
func jsonErr(format string, args ...any) []byte {
|
|
b, _ := json.Marshal(map[string]any{"error": fmt.Sprintf(format, args...)})
|
|
return b
|
|
}
|