Files
oikos/internal/actuator/ssh.go
dtoro 75c0848a6f
Some checks failed
ci / build-test (push) Has been cancelled
ci / docker-build (push) Has been cancelled
ci / web (push) Has been cancelled
Desktop App / Build Linux (amd64) (push) Has been cancelled
Desktop App / Attach to Release (push) Has been cancelled
0.29.0 — code-quality refactor (plan E1–E5): file splits, sqlc migration, SSH unification, test coverage
E1: split monolithic files — cmd/nomos (main.go → server.go + mcp.go + workers.go),
    internal/mcp/tools.go → entity_tools/ops_tools/knowledge_tools/analysis_tools,
    internal/httpapi/impl.go → domain files (entities, events, signals, ontology,
    fleet_health, client_context, client_lifecycle, entity_mutations, query_audit).
E2: migrate raw pool.Exec queries to sqlc (entities/relationships queries + generated).
E3: unify SSH — consolidate crypto/ssh dial into actuator/client.go (+client_test).
E4/E5: add tests — db/lifecycle, checkdefaults/build, ontology/preconditions, policy/risk.
2026-08-08 22:47:06 +02:00

246 lines
6.8 KiB
Go

// Package actuator provides SSH-based skill procedure execution for the Oikos
// Phase 3 actuator loop. It runs stored skill procedures over SSH with a
// restricted key, classifies SSH errors into retryable/fatal/timeout, and
// supports step-by-step procedure verification.
package actuator
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net"
"strings"
"sync"
"time"
"golang.org/x/crypto/ssh"
)
// ─── Procedure types ──────────────────────────────────────────────────────
// Procedure represents a parsed skill procedure from JSON config.
type Procedure struct {
Steps []Step `json:"steps"`
}
// Step is a single step within a procedure.
type Step struct {
Runner string `json:"runner"` // "shell", "script", "verify"
Target string `json:"target,omitempty"` // hostname/IP (empty = local)
Command string `json:"command"` // shell command or script path
TimeoutS int `json:"timeout_s,omitempty"` // per-step timeout in seconds
}
// SSHResult holds the outcome of an SSH execution.
type SSHResult struct {
Output string `json:"output"`
Duration time.Duration `json:"duration"`
Verified bool `json:"verified"`
Err error `json:"error,omitempty"`
}
// ─── Error classification ─────────────────────────────────────────────────
// SSHErrorClass categorises SSH errors.
type SSHErrorClass int
const (
SSHErrorUnknown SSHErrorClass = iota
SSHErrorNetwork // dial/connect timeout — retryable
SSHErrorAuth // auth failure — fatal
SSHErrorTimeout // command timed out
SSHErrorRemote // remote command returned non-zero
SSHErrorOther // other non-retryable
)
func (c SSHErrorClass) String() string {
switch c {
case SSHErrorNetwork:
return "network"
case SSHErrorAuth:
return "auth"
case SSHErrorTimeout:
return "timed_out"
case SSHErrorRemote:
return "remote"
case SSHErrorOther:
return "other"
default:
return "unknown"
}
}
// classifySSHError maps an SSH error to a class for retry/fatal decisions.
func classifySSHError(err error) SSHErrorClass {
if err == nil {
return SSHErrorOther
}
if errors.Is(err, context.DeadlineExceeded) {
return SSHErrorTimeout
}
var netErr net.Error
if errors.As(err, &netErr) {
if netErr.Timeout() {
return SSHErrorNetwork
}
return SSHErrorNetwork
}
if strings.Contains(err.Error(), "unable to authenticate") ||
strings.Contains(err.Error(), "no supported methods remain") ||
strings.Contains(err.Error(), "ssh: handshake failed") ||
strings.Contains(err.Error(), "publickey") ||
strings.Contains(err.Error(), "permission denied") {
return SSHErrorAuth
}
var exitErr *ssh.ExitError
if errors.As(err, &exitErr) {
return SSHErrorRemote
}
return SSHErrorOther
}
// ─── SSH execution ────────────────────────────────────────────────────────
// SSHConfig holds connection parameters for SSH sessions.
type SSHConfig struct {
Host string
Port int
User string
KeyPath string
Timeout time.Duration
}
// ExecuteProcedure runs a complete procedure over SSH, step by step.
// Returns the combined result, duration, and verified status.
//
// Context cancellation aborts the running session. Returns the last
// successfully completed step's output on partial failure.
func ExecuteProcedure(
ctx context.Context,
cfg SSHConfig,
proc Procedure,
) SSHResult {
start := time.Now()
// Parse the SSH key
signer, err := LoadSigner(cfg.KeyPath)
if err != nil {
return SSHResult{
Err: err,
Duration: time.Since(start),
Verified: false,
}
}
client, err := Dial(ctx, DialOptions{
Host: cfg.Host, Port: cfg.Port, User: cfg.User,
Signer: signer, Timeout: cfg.Timeout,
})
if err != nil {
class := classifySSHError(err)
return SSHResult{
Err: fmt.Errorf("ssh dial (%s): %w", class, err),
Duration: time.Since(start),
Verified: false,
}
}
defer client.Close()
// Execute each step in sequence
var lastOutput string
verified := true
for i, step := range proc.Steps {
// Check context before each step
if ctx.Err() != nil {
return SSHResult{
Output: lastOutput,
Duration: time.Since(start),
Verified: false,
Err: fmt.Errorf("cancelled before step %d: %w", i, ctx.Err()),
}
}
timeout := time.Duration(step.TimeoutS) * time.Second
if timeout <= 0 {
timeout = 30 * time.Second
}
stepCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
output, err := runSSHCommand(stepCtx, client, step.Command)
if err != nil {
class := classifySSHError(err)
// Verify steps that fail are not counted as verified failures
if step.Runner == "verify" {
verified = false
}
// Non-verify step failure is a real failure
if step.Runner != "verify" {
return SSHResult{
Output: lastOutput,
Duration: time.Since(start),
Err: fmt.Errorf("step %d (%s) failed (%s): %w", i, step.Runner, class, err),
Verified: false,
}
}
}
lastOutput = output
slog.Debug("ssh step completed",
"step", i,
"runner", step.Runner,
"duration", time.Since(start).Round(time.Millisecond),
)
}
return SSHResult{
Output: lastOutput,
Duration: time.Since(start),
Verified: verified,
}
}
// runSSHCommand executes a single command over an established SSH session via
// the shared RunCombinedOutput primitive (context-aware abort + combined output).
func runSSHCommand(ctx context.Context, client *ssh.Client, command string) (string, error) {
out, err := RunCombinedOutput(ctx, client, command)
return string(out), err
}
// ─── Procedure parsing ────────────────────────────────────────────────────
// ParseProcedure deserialises a JSON procedure (from skill.procedure).
func ParseProcedure(data []byte) (Procedure, error) {
var proc Procedure
if err := json.Unmarshal(data, &proc); err != nil {
return Procedure{}, fmt.Errorf("parse procedure: %w", err)
}
return proc, nil
}
// ─── Global SSH client options ────────────────────────────────────────────
var (
mu sync.Mutex
// defaultSSHTimeout is the default dial timeout for SSH connections.
defaultSSHTimeout = 10 * time.Second
)
// SetDefaultSSHTimeout overrides the default SSH dial timeout. Not safe for
// concurrent use during active execution.
func SetDefaultSSHTimeout(d time.Duration) {
mu.Lock()
defer mu.Unlock()
defaultSSHTimeout = d
}