complete MCP tool surface — Matrix approval webhook loop + token verification

Plan #6 (MCP Tool Completion / bin/homelab Migration) done. - Approval records created for gated request_execution actions - Notifier sends Matrix messages with HMAC approval tokens - Stores matrix_event_id, polls /relations/{id}/m.annotation for ✅/❌ - Reaction detection triggers DecideApproval API call - Token verification added to DecideApproval endpoint - Migration 013: matrix_event_id + alert_sent_at on approvals - AGENTS.md: 21-tool surface documented, stale homelab CLI refs removed - Plan index updated, audit cross-reference refreshed
2026-07-08 11:02:06 +02:00
parent 5b22f2367b
commit 7c6cffb5f5
12 changed files with 907 additions and 235 deletions
--- a/internal/httpapi/phase3.go
+++ b/internal/httpapi/phase3.go
@@ -2,12 +2,16 @@ package httpapi

 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"os"
 	"strings"
 	"time"

+	"github.com/dtoro/oikos/internal/db"
 	"github.com/dtoro/oikos/internal/db/sqlcgen"
 	"github.com/dtoro/oikos/internal/domain"
 	"github.com/dtoro/oikos/internal/httpapi/gen"
@@ -15,8 +19,177 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgtype"
+	"golang.org/x/crypto/ssh"
 )

+var (
+	_sshUser string
+	_sshKey  []byte
+)
+
+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)
+		}
+	}
+}
+
+func sshExec(ctx context.Context, host, user, command string) (string, error) {
+	initSSH()
+	if len(_sshKey) == 0 {
+		return "", fmt.Errorf("no SSH key available")
+	}
+	if user == "" {
+		user = _sshUser
+	}
+
+	addr := host + ":22"
+	signer, err := ssh.ParsePrivateKey(_sshKey)
+	if err != nil {
+		return "", fmt.Errorf("parse key: %w", err)
+	}
+
+	cfg := &ssh.ClientConfig{
+		User:            user,
+		Auth:            []ssh.AuthMethod{ssh.PublicKeys(signer)},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		Timeout:         10 * time.Second,
+	}
+
+	client, err := ssh.Dial("tcp", addr, cfg)
+	if err != nil {
+		return "", fmt.Errorf("dial %s: %w", host, err)
+	}
+	defer client.Close()
+
+	session, err := client.NewSession()
+	if err != nil {
+		return "", fmt.Errorf("session: %w", err)
+	}
+	defer session.Close()
+
+	out, err := session.CombinedOutput(command)
+	if err != nil && out == nil {
+		return "", fmt.Errorf("exec: %w", err)
+	}
+	return strings.TrimSpace(string(out)), nil
+}
+
+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)
+}
+
+// executeApprovedAction runs a gated action after operator approval.
+// Runs in a background goroutine to not block the HTTP response.
+func 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, err := resolveHostSSH(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, fmt.Sprintf(`{"error":"%s"}`, err.Error()))
+		return
+	}
+
+	parts := strings.SplitN(actionStr, ":", 3)
+	if len(parts) < 2 {
+		slog.Error("httpapi: malformed action string", "action", actionStr)
+		return
+	}
+	action, params := parts[0], parts[1]
+	if len(parts) == 3 {
+		params = parts[1] + ":" + parts[2]
+	}
+
+	startedAt := time.Now()
+	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 = sshExec(ctx, host, user, cmd)
+
+	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 = sshExec(ctx, host, user, cmd)
+
+	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, fmt.Sprintf(`{"error":"unknown action: %s"}`, action))
+		return
+	}
+
+	durationMs := int(time.Since(startedAt).Milliseconds())
+	result := fmt.Sprintf(`{"output":"%s"}`, strings.ReplaceAll(output, "\n", "\\n"))
+	status := "completed"
+	verified := true
+	if err != nil {
+		result = fmt.Sprintf(`{"output":"%s","error":"%s"}`, strings.ReplaceAll(output, "\n", "\\n"), err.Error())
+		status = "failed"
+		verified = false
+	}
+
+	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, result, durationMs, verified, startedAt, time.Now())
+
+	slog.Info("httpapi: approved action executed",
+		"execution_id", execID, "action", action, "status", status, "duration_ms", durationMs)
+}
+
 // ─── Checks ────────────────────────────────────────────────────────────

 func (s *Server) ListChecks(ctx context.Context, req gen.ListChecksRequestObject) (gen.ListChecksResponseObject, error) {
@@ -715,6 +888,28 @@ func (s *Server) DecideApproval(ctx context.Context, req gen.DecideApprovalReque

 	q := sqlcgen.New(tx)

+	// Verify HMAC token if provided (single-use, S5).
+	if req.Body.Token != nil && *req.Body.Token != "" {
+		var tokenHash *string
+		var apprStatus string
+		var expiresAt time.Time
+		err := tx.QueryRow(ctx,
+			"SELECT token_hash, status, expires_at FROM approvals WHERE entity_id = $1",
+			id).Scan(&tokenHash, &apprStatus, &expiresAt)
+		if err != nil || tokenHash == nil {
+			return nil, fmt.Errorf("%w: approval not found", domain.ErrNotFound)
+		}
+		if apprStatus != "pending" {
+			return nil, fmt.Errorf("%w: approval already decided", domain.ErrInvalidTransition)
+		}
+		if expiresAt.Before(time.Now()) {
+			return nil, fmt.Errorf("%w: approval token expired", domain.ErrInvalidTransition)
+		}
+		if *tokenHash != hashToken(*req.Body.Token) {
+			return nil, fmt.Errorf("%w: invalid approval token", domain.ErrInvalidInput)
+		}
+	}
+
 	// Map decision to status.
 	var status string
 	switch req.Body.Decision {
@@ -752,6 +947,29 @@ func (s *Server) DecideApproval(ctx context.Context, req gen.DecideApprovalReque
 		return nil, auditErr
 	}

+	// On approve: execute the linked gated command.
+	if status == "approved" {
+		var execID, targetID uuid.UUID
+		var actionStr, targetSlug string
+		err := tx.QueryRow(ctx, `
+			SELECT e.entity_id, e.target_entity_id, e.action
+			FROM executions e
+			WHERE e.approval_id = $1 AND e.status = 'pending_approval'
+			LIMIT 1`, id).Scan(&execID, &targetID, &actionStr)
+		if err == nil {
+			// Resolve target entity slug from targetID.
+			_ = tx.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", targetID).Scan(&targetSlug)
+
+			go executeApprovedAction(context.Background(), s.pool, execID, targetSlug, actionStr)
+			_, _ = tx.Exec(ctx, `UPDATE executions SET status = 'approved', risk_class = 'config_mutation' WHERE entity_id = $1`, execID)
+
+			slog.Info("httpapi: approved execution queued",
+				"execution_id", execID, "target", targetSlug, "action", actionStr)
+		} else {
+			slog.Warn("httpapi: no pending execution found for approval", "approval_id", id, "error", err)
+		}
+	}
+
 	if err := tx.Commit(ctx); err != nil {
 		return nil, err
 	}
@@ -1838,3 +2056,8 @@ func parseIntOrZero(s string) (int, error) {
 	}
 	return n, nil
 }
+
+func hashToken(token string) string {
+	h := sha256.Sum256([]byte(token))
+	return hex.EncodeToString(h[:])
+}