dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat(remote): route LXC/VM checks through the Proxmox host, not direct SSH

Browse Source 
The scheduler SSHed each guest directly and assumed a deployed probe script
plus working root SSH at the guest's address — false for headless (nfs-export),
keyless (teddycloud), mesh-only (rclone), and macOS (mac-mini) targets, which
left 49 enabled checks stuck "down" on a healthy fleet.

Extract the MCP run tool's resolveExecTarget into a shared internal/remote
package and make it the single execution path for both the scheduler and MCP.
LXC/VM checks now host-hop via pct exec / qm guest exec through the owning
Proxmox host (no per-guest lan_ip, sshd, or authorized key needed); hosts and
workstations resolve their address and user live, so mac-mini's `user: dtoro`
is honored without a re-seed. Address preference now prefers public_ipv4 over
mesh, so netbird-vps is probeable from the scheduler container.

cpu_check.sh gains a real Darwin branch (it reported cpu_pct 0 before).
checkdefaults.resolveSSHUser reads the top-level `user` attribute too.
A machine-target resolution failure is now logged before falling back to baked
config, so a broken probe-config is distinguishable from a real outage.
This commit is contained in:
authentik Default Admin
2026-07-29 13:37:00 +02:00
parent c10f6920cd
commit b8b4aa2aee
7 changed files with 629 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
internal/checkdefaults/defaults.go
View File

@@ -380,6 +380,12 @@ func resolveHost(attrs map[string]any) string {
if ip, ok := attrs["lan_ip"].(string); ok && ip != "" {
return ip
}
// public_ipv4 before mesh: the scheduler container has no mesh interface,
// so a standalone-server reachable only by mesh IP (netbird-vps) is
// unprobeable even though a public IPv4 is available.
if ip, ok := attrs["public_ipv4"].(string); ok && ip != "" {
return ip
}
if mesh, ok := attrs["mesh"].(map[string]any); ok {
if nb, ok := mesh["netbird"].(map[string]any); ok {
if ip, ok := nb["ip"].(string); ok && ip != "" {
@@ -409,6 +415,13 @@ func resolveSSHUser(attrs map[string]any) string {
return u
}
}
// Workstations carry their login as a top-level `user` attribute
// (mac-mini: user: dtoro) rather than under ssh.user. Take it only when
// no explicit ssh.user was set, so a host that genuinely wants root still
// gets root.
if u, ok := attrs["user"].(string); ok && u != "" {
return u
}
return "root"
}

144
internal/mcp/server.go
View File

@@ -5,7 +5,6 @@ package mcp
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"fmt"
"html"
@@ -25,6 +24,7 @@ import (
"github.com/dtoro/oikos/internal/execlog"
"github.com/dtoro/oikos/internal/observability"
"github.com/dtoro/oikos/internal/policy"
"github.com/dtoro/oikos/internal/remote"
"github.com/google/jsonschema-go/jsonschema"
"github.com/google/uuid"
"github.com/modelcontextprotocol/go-sdk/mcp"
@@ -461,31 +461,12 @@ func sshExecStream(ctx context.Context, host, user, command string, sink execlog
}
}
func resolveHost(ctx context.Context, pool *db.Pool, entitySlug string) (hostIP string, sshUser string, err 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)
}
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)
// resolveHost resolves a host:<slug> to its reachable IP and SSH user. A thin
// wrapper over the shared resolver (internal/remote), kept so slug-based
// callers keep working; the shared resolver also prefers public_ipv4 over
// mesh and honors a per-entity ssh.user.
func resolveHost(ctx context.Context, pool *db.Pool, entitySlug string) (hostIP string, sshUserOut string, err error) {
return remote.ResolveHost(ctx, pool, entitySlug, sshUser)
}
// htmlTagRe strips HTML tags for the naive text extraction in httpGet.
@@ -564,109 +545,28 @@ func isPrivateHost(host string) bool {
// connection: identity for a host, `pct exec <pve_id> -- ...` for an LXC,
// `qm guest exec <pve_id> -- ...` for a VM.
//
// The lxc.attributes.host value is stored WITHOUT a "host:" prefix (e.g.
// "strong", not "host:strong") — see pct_create's entity registration. The
// pre-existing pct_exec handler queried resolveHost with that bare value
// directly, which can never match a "host:*" slug and always fails; this
// prefixes it correctly.
//
// vm: support (2026-07-18): VMs in inventory.yaml carry `pve_id` and a `host`
// attribute (or a `hosts` relationship) just like LXCs, but they're reached
// via `qm guest exec` instead of `pct exec`. Previously the agent had to
// SSH-hop via `host:hubris` to reach a VM (e.g. `ssh root@<vm_ip> '...'`),
// which broke on nested shell quoting and forced manual escaping workarounds
// — see plans/2026-07-18-session-review-three-sessions.md P1.6. A VM's
// `host` attribute is optional: if absent, fall back to looking up the
// `hosts` relationship on the VM entity, then to hubris (the documented
// default Proxmox host) — same fallback chain as LXCs.
// Delegates to the shared resolver (internal/remote), the single path used by
// both the MCP `run` tool and the scheduler's checks. The historical notes
// (host attr without prefix, vm host-resolution chain, nested-quoting
// handling via base64) all still hold — they now live in remote.guestWrap.
func resolveExecTarget(ctx context.Context, pool *db.Pool, targetSlug string) (host, user string, wrap func(cmd string) string, err error) {
if strings.HasPrefix(targetSlug, "host:") {
host, user, err = resolveHost(ctx, pool, targetSlug)
return host, user, func(cmd string) string { return cmd }, err
et, err := remote.ResolveExecTarget(ctx, pool, targetSlug, sshUser)
if err != nil {
return "", "", nil, 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 := resolveProxmoxHostSlug(ctx, pool, targetSlug, hostAttr)
host, user, err = resolveHost(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
}
if strings.HasPrefix(targetSlug, "vm:") {
// VMs: same host-resolution chain as LXCs (attributes.host →
// `hosts` relationship → hubris default), but reached via
// `qm guest exec` instead of `pct exec`. Requires the QEMU
// guest agent running inside the VM (the standard Proxmox
// setup; ZimaOS/HAOS in this fleet already have it).
var pveID, hostAttr string
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("VM not found or missing pve_id: %s", targetSlug)
}
hostSlug := resolveProxmoxHostSlug(ctx, pool, targetSlug, hostAttr)
host, user, err = resolveHost(ctx, pool, hostSlug)
id := pveID
return host, user, func(cmd string) string {
b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
// `qm guest exec <id> -- /bin/bash -c '...'` returns JSON by
// default; pipe through `jq -r .out` if available, else cat.
// The base64 round-trip mirrors the LXC path so nested quoting
// (the original VM-target pain point — session 55927f0a) is
// handled identically to LXC dispatch.
return fmt.Sprintf(
"qm guest exec %s -- /bin/bash -c 'echo %s | base64 -d | bash' | jq -r '.out // .err // empty' 2>/dev/null || qm guest exec %s -- /bin/bash -c 'echo %s | base64 -d | bash'",
id, b64, id, b64)
}, err
}
return "", "", nil, fmt.Errorf("unsupported target %q: must be host:<slug>, lxc:<slug>, or vm:<slug>", targetSlug)
return et.Host, et.User, et.Wrap, nil
}
// resolveProxmoxHostSlug resolves the Proxmox host slug that owns a given
// LXC/VM target. Resolution order:
// 1. hostAttr if non-empty (the entity's attributes.host — stored without
// "host:" prefix in inventory.yaml and pct_create).
// 2. the `hosts` relationship on the entity (e.g. host:hubris → vm:zimaos),
// looked up in the relationships table — the canonical graph source.
// 3. "hubris" as a documented default Proxmox host fallback.
//
// Returns a slug with the "host:" prefix attached, ready for resolveHost.
// Extracted from the inline LXC path (2026-07-18) so the VM path shares the
// same chain — see plans/2026-07-18-session-review-three-sessions.md P1.6.
// LXC/VM target (see internal/remote.ResolveProxmoxHostSlug for the chain).
// This slug-based wrapper looks up the entity id so slug callers keep working;
// the shared resolver takes an id directly.
func resolveProxmoxHostSlug(ctx context.Context, pool *db.Pool, entitySlug, hostAttr string) string {
hostSlug := strings.TrimSpace(hostAttr)
if hostSlug == "" {
// Fall back to the `hosts` relationship — the graph edge from
// the Proxmox host to this LXC/VM. This is the canonical source
// for "who owns this VM" in inventory.yaml; the `host` attribute
// is a denormalized shortcut that not every entity has.
var relHostSlug string
// hosts relationship: source=host, target=lxc/vm. Look up the
// source slug given the target.
if err := pool.QueryRow(ctx, `
SELECT e.slug FROM relationships r
JOIN entities e ON e.id = r.source_id
WHERE r.target_id = (SELECT id FROM entities WHERE slug = $1)
AND r.type = 'hosts' AND r.valid_to IS NULL
LIMIT 1`, entitySlug).Scan(&relHostSlug); err == nil && relHostSlug != "" {
hostSlug = relHostSlug
}
var id uuid.UUID
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", entitySlug).Scan(&id); err != nil {
id = uuid.Nil
}
if hostSlug == "" {
hostSlug = "hubris" // documented default Proxmox host when unset
}
if !strings.HasPrefix(hostSlug, "host:") {
hostSlug = "host:" + hostSlug
}
return hostSlug
return remote.ResolveProxmoxHostSlug(ctx, pool, id, hostAttr)
}
// classifyAndGate is the shared classify→execute-or-queue path for every

257
internal/remote/remote.go Normal file
View File

@@ -0,0 +1,257 @@
// Package remote resolves how to execute a command on a target entity and
// turns a plain shell command into whatever must be sent over the SSH
// connection that reaches it.
//
// The canonical access model: a host or workstation is reached by direct SSH
// to its address; an LXC or VM is NEVER SSH'd into directly — it is reached
// through its owning Proxmox host via `pct exec` / `qm guest exec`. One SSH
// credential per host (the host's root key), no per-guest keys, sshd, or
// lan_ip required for execution. Network probes (http/ping) still hit a
// guest's lan_ip directly; only command execution host-hops.
//
// This is the single resolver shared by the scheduler's check execution and
// the MCP `run` tool. Previously they diverged — the scheduler SSHed guests
// directly (broken for headless/keyless/mesh-only guests), while MCP
// host-hopped (working). Keeping one path keeps them in lockstep.
package remote
import (
"context"
"encoding/base64"
"encoding/json"
"fmt"
"strings"
"github.com/dtoro/oikos/internal/db"
"github.com/google/uuid"
)
// DefaultUser is the SSH user when an entity declares no ssh.user. The
// Proxmox hosts and their guests are all administered as root.
const DefaultUser = "root"
// ExecTarget is a resolved execution endpoint: the SSH address and user to
// connect to, plus Wrap, which rewrites a plain command for transport.
type ExecTarget struct {
Host string
User string
// Wrap turns a plain shell command into the form that must be sent over
// the SSH connection to this target: the identity function for a host,
// `pct exec <id> -- bash -c 'echo <b64> | base64 -d | bash'` for an LXC,
// the `qm guest exec` equivalent for a VM. The base64 round-trip keeps
// nested quoting identical across both guest kinds.
Wrap func(cmd string) string
}
// IsGuest reports whether an entity type is reached via pct/qm exec through a
// Proxmox host rather than by direct SSH. docker-container is reached via its
// host's docker socket, not pct, so it is not a guest here.
func IsGuest(entityType string) bool {
return entityType == "lxc" || entityType == "vm"
}
// ResolveHost resolves a `host:<slug>` to its reachable network address and
// SSH user. Address preference: lan_ip, then public_ipv4, then mesh IP, then
// mesh fqdn. Preferring public_ipv4 over mesh matters because the scheduler
// container has no mesh interface — a standalone-server with only a mesh IP
// (netbird-vps) was unreachable, and a public_ipv4 was sitting unused.
//
// fallbackUser is used when the entity declares no ssh.user; callers pass
// their configured default (the scheduler uses "root", the MCP run tool uses
// its configured OIKOS_SSH_USER).
func ResolveHost(ctx context.Context, pool *db.Pool, hostSlug, fallbackUser string) (addr, user string, err error) {
var raw string
if err = pool.QueryRow(ctx, "SELECT attributes::text FROM entities WHERE slug = $1", hostSlug).Scan(&raw); err != nil {
return "", "", fmt.Errorf("entity not found: %s", hostSlug)
}
var m map[string]any
if err = json.Unmarshal([]byte(raw), &m); err != nil {
return "", "", fmt.Errorf("parse attributes for %s: %w", hostSlug, err)
}
if v, ok := m["lan_ip"].(string); ok && v != "" {
addr = v
} else if v, ok := m["public_ipv4"].(string); ok && v != "" {
addr = v
} else if mesh, ok := m["mesh"].(map[string]any); ok {
if nb, ok := mesh["netbird"].(map[string]any); ok {
if v, ok := nb["ip"].(string); ok && v != "" {
addr = v
} else if v, ok := nb["fqdn"].(string); ok && v != "" {
addr = v
}
}
}
if addr == "" {
return "", "", fmt.Errorf("no IP found for %s", hostSlug)
}
user = fallbackUser
if ssh, ok := m["ssh"].(map[string]any); ok {
if u, ok := ssh["user"].(string); ok && u != "" {
user = u
}
}
return addr, user, nil
}
// ResolveProxmoxHostSlug resolves the Proxmox host slug that owns a guest.
// Resolution order: the hostAttr if non-empty (the entity's attributes.host,
// stored without the "host:" prefix), the `hosts` relationship on the guest
// (the canonical graph edge), then "hubris" as the documented default.
//
// entityID is the guest's entity id; the relationship lookup uses it
// directly rather than a slug subquery.
func ResolveProxmoxHostSlug(ctx context.Context, pool *db.Pool, entityID uuid.UUID, hostAttr string) string {
hostSlug := strings.TrimSpace(hostAttr)
if hostSlug == "" {
var relHostSlug string
if err := pool.QueryRow(ctx, `
SELECT e.slug FROM relationships r
JOIN entities e ON e.id = r.source_id
WHERE r.target_id = $1
AND r.type = 'hosts' AND r.valid_to IS NULL
LIMIT 1`, entityID).Scan(&relHostSlug); err == nil && relHostSlug != "" {
hostSlug = relHostSlug
}
}
if hostSlug == "" {
hostSlug = "hubris" // documented default Proxmox host when unset
}
if !strings.HasPrefix(hostSlug, "host:") {
hostSlug = "host:" + hostSlug
}
return hostSlug
}
// ResolveExecTarget resolves a target slug (host:, lxc:, or vm:) to its
// execution endpoint. This is the slug-based entry used by the MCP `run` tool.
func ResolveExecTarget(ctx context.Context, pool *db.Pool, targetSlug, fallbackUser string) (ExecTarget, error) {
switch {
case strings.HasPrefix(targetSlug, "host:"):
addr, user, err := ResolveHost(ctx, pool, targetSlug, fallbackUser)
if err != nil {
return ExecTarget{}, err
}
return ExecTarget{Host: addr, User: user, Wrap: func(cmd string) string { return cmd }}, nil
case strings.HasPrefix(targetSlug, "lxc:"), strings.HasPrefix(targetSlug, "vm:"):
var (
id uuid.UUID
pveID string
typ string
hostAttr string
)
if err := pool.QueryRow(ctx,
"SELECT id, type, attributes->>'pve_id', COALESCE(attributes->>'host','') FROM entities WHERE slug = $1",
targetSlug).Scan(&id, &typ, &pveID, &hostAttr); err != nil || pveID == "" {
return ExecTarget{}, fmt.Errorf("guest not found or missing pve_id: %s", targetSlug)
}
hostSlug := ResolveProxmoxHostSlug(ctx, pool, id, hostAttr)
addr, user, err := ResolveHost(ctx, pool, hostSlug, fallbackUser)
if err != nil {
return ExecTarget{}, err
}
return ExecTarget{Host: addr, User: user, Wrap: guestWrap(typ, pveID)}, nil
}
return ExecTarget{}, fmt.Errorf("unsupported target %q: must be host:<slug>, lxc:<slug>, or vm:<slug>", targetSlug)
}
// ResolveExecTargetForCheck resolves an execution endpoint keyed by the
// target's id and type — the data the scheduler has at check-execution time
// (check_defs carry target_id + target_type, not a slug). Guests route via
// pct/qm exec; everything else (hosts, workstations, services resolved to
// their hosting machine) is reached by direct SSH to the entity's own address.
func ResolveExecTargetForCheck(ctx context.Context, pool *db.Pool, targetID uuid.UUID, targetType, fallbackUser string) (ExecTarget, error) {
if IsGuest(targetType) {
var (
pveID string
hostAttr string
)
if err := pool.QueryRow(ctx,
"SELECT attributes->>'pve_id', COALESCE(attributes->>'host','') FROM entities WHERE id = $1",
targetID).Scan(&pveID, &hostAttr); err != nil || pveID == "" {
return ExecTarget{}, fmt.Errorf("guest %s missing pve_id", targetID)
}
hostSlug := ResolveProxmoxHostSlug(ctx, pool, targetID, hostAttr)
addr, user, err := ResolveHost(ctx, pool, hostSlug, fallbackUser)
if err != nil {
return ExecTarget{}, err
}
return ExecTarget{Host: addr, User: user, Wrap: guestWrap(targetType, pveID)}, nil
}
// Host-like target: reach it directly at its own address. Services and
// other non-host entities that reach here should already have had their
// host address baked into check config at seed time; this path covers
// host/workstation targets whose address is resolved live.
addr, user, err := resolveHostByID(ctx, pool, targetID, fallbackUser)
if err != nil {
return ExecTarget{}, err
}
return ExecTarget{Host: addr, User: user, Wrap: func(cmd string) string { return cmd }}, nil
}
// resolveHostByID is ResolveHost keyed by entity id.
func resolveHostByID(ctx context.Context, pool *db.Pool, id uuid.UUID, fallbackUser string) (addr, user string, err error) {
var raw string
if err = pool.QueryRow(ctx, "SELECT attributes::text FROM entities WHERE id = $1", id).Scan(&raw); err != nil {
return "", "", fmt.Errorf("entity %s not found", id)
}
var m map[string]any
if err = json.Unmarshal([]byte(raw), &m); err != nil {
return "", "", fmt.Errorf("parse attributes: %w", err)
}
for _, key := range []string{"lan_ip", "public_ipv4", "mesh_ip"} {
if v, ok := m[key].(string); ok && v != "" {
addr = v
break
}
}
if addr == "" {
if mesh, ok := m["mesh"].(map[string]any); ok {
if nb, ok := mesh["netbird"].(map[string]any); ok {
if v, ok := nb["ip"].(string); ok && v != "" {
addr = v
} else if v, ok := nb["fqdn"].(string); ok && v != "" {
addr = v
}
}
}
}
if addr == "" {
return "", "", fmt.Errorf("no IP found for entity %s", id)
}
user = fallbackUser
if ssh, ok := m["ssh"].(map[string]any); ok {
if u, ok := ssh["user"].(string); ok && u != "" {
user = u
}
}
if u, ok := m["user"].(string); ok && u != "" && user == fallbackUser {
// Workstations carry their login as a top-level `user` attribute
// (mac-mini: user: dtoro), not under ssh.user. Take it only when no
// explicit ssh.user was set, so a host that genuinely wants root still
// gets root.
user = u
}
return addr, user, nil
}
// guestWrap builds the pct/qm exec wrapper for a guest of the given type.
func guestWrap(entityType, pveID string) func(cmd string) string {
if entityType == "vm" {
return func(cmd string) string {
b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
// `qm guest exec` returns JSON; pipe through jq for a clean stdout,
// falling back to the raw form. Mirrors the LXC base64 round-trip.
return fmt.Sprintf(
"qm guest exec %s -- /bin/bash -c 'echo %s | base64 -d | bash' | jq -r '.out // .err // empty' 2>/dev/null || qm guest exec %s -- /bin/bash -c 'echo %s | base64 -d | bash'",
pveID, b64, pveID, b64)
}
}
return func(cmd string) string {
b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
return fmt.Sprintf("pct exec %s -- bash -c 'echo %s | base64 -d | bash'", pveID, b64)
}
}

172
internal/remote/remote_test.go Normal file
View File

@@ -0,0 +1,172 @@
package remote
import (
"context"
"encoding/base64"
"strings"
"testing"
"github.com/dtoro/oikos/internal/db"
"github.com/google/uuid"
)
// guestWrap and IsGuest are pure logic — always tested. The DB-backed
// resolvers are integration tests guarded by OIKOS_TEST_DATABASE_URL, the
// same convention as internal/scheduler/coverage_test.go.
func TestIsGuest(t *testing.T) {
cases := map[string]bool{
"lxc": true, "vm": true,
"proxmox-host": false, "workstation": false,
"service": false, "docker-container": false,
}
for typ, want := range cases {
if got := IsGuest(typ); got != want {
t.Errorf("IsGuest(%q) = %v, want %v", typ, got, want)
}
}
}
func TestGuestWrapLXC(t *testing.T) {
w := guestWrap("lxc", "132")
out := w("/opt/oikos/checks/cpu_check.sh 'svc'")
if !strings.Contains(out, "pct exec 132 -- bash -c ") {
t.Fatalf("lxc wrap must use pct exec: %q", out)
}
if strings.Contains(out, "qm guest exec") {
t.Fatalf("lxc wrap must not use qm: %q", out)
}
// The base64 payload must round-trip to the original command.
i := strings.Index(out, "echo ")
j := strings.LastIndex(out, " | base64 -d | bash")
if i < 0 || j < 0 || j <= i {
t.Fatalf("cannot locate base64 payload in %q", out)
}
dec, err := base64.StdEncoding.DecodeString(out[i+len("echo ") : j])
if err != nil {
t.Fatalf("decode payload: %v", err)
}
if string(dec) != "/opt/oikos/checks/cpu_check.sh 'svc'" {
t.Fatalf("round-trip mismatch: %q", string(dec))
}
}
func TestGuestWrapVM(t *testing.T) {
w := guestWrap("vm", "108")
out := w("uname -a")
if !strings.Contains(out, "qm guest exec 108") {
t.Fatalf("vm wrap must use qm guest exec: %q", out)
}
}
func TestResolveExecTargetUnsupported(t *testing.T) {
// No DB needed: an unsupported slug prefix errors before any query.
if _, err := ResolveExecTarget(context.Background(), nil, "service:gitea", DefaultUser); err == nil {
t.Fatal("expected error for unsupported target prefix")
}
}
// --- integration tests (require a real Postgres) ---
func newRemotePool(t *testing.T) *db.Pool {
t.Helper()
base := testDatabaseURL(t)
return createTestDB(t, base)
}
func testDatabaseURL(t *testing.T) string {
t.Helper()
u := getenvOrDefault("OIKOS_TEST_DATABASE_URL", "")
if u == "" {
t.Skip("OIKOS_TEST_DATABASE_URL not set — skipping integration test")
}
return u
}
func TestResolveHostPrefersLAN(t *testing.T) {
pool := newRemotePool(t)
ctx := context.Background()
mustExec(t, pool, ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at)
VALUES ($1,'host:x','proxmox-host','x','active','{"lan_ip":"10.0.0.1","public_ipv4":"1.2.3.4","mesh":{"netbird":{"ip":"100.64.0.1"}}}'::jsonb,1,now(),now())`,
uuid.New())
addr, user, err := ResolveHost(ctx, pool, "host:x", DefaultUser)
if err != nil {
t.Fatalf("ResolveHost: %v", err)
}
if addr != "10.0.0.1" {
t.Errorf("addr = %q, want lan_ip 10.0.0.1", addr)
}
if user != "root" {
t.Errorf("user = %q, want root", user)
}
}
func TestResolveHostFallsBackToPublicIPv4(t *testing.T) {
// netbird-vps: no lan_ip, has public_ipv4 + mesh ip. Must prefer
// public_ipv4 — the scheduler container has no mesh interface.
pool := newRemotePool(t)
ctx := context.Background()
mustExec(t, pool, ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at)
VALUES ($1,'host:vps','standalone-server','vps','active','{"public_ipv4":"82.165.190.79","mesh":{"netbird":{"ip":"100.122.165.149"}}}'::jsonb,1,now(),now())`,
uuid.New())
addr, _, err := ResolveHost(ctx, pool, "host:vps", DefaultUser)
if err != nil {
t.Fatalf("ResolveHost: %v", err)
}
if addr != "82.165.190.79" {
t.Errorf("addr = %q, want public_ipv4 (mesh unreachable from container)", addr)
}
}
func TestResolveExecTargetForCheckLXCRoutesViaHost(t *testing.T) {
// An LXC guest with a `hosts` edge to a proxmox host must resolve to the
// HOST's address (the host-hop target), wrapped as `pct exec`.
pool := newRemotePool(t)
ctx := context.Background()
hostID := uuid.New()
guestID := uuid.New()
mustExec(t, pool, ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at)
VALUES ($1,'host:hubris','proxmox-host','hubris','active','{"lan_ip":"192.168.8.77"}'::jsonb,1,now(),now())`, hostID)
mustExec(t, pool, ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at)
VALUES ($1,'lxc:rclone','lxc','rclone','active','{"pve_id":"132"}'::jsonb,1,now(),now())`, guestID)
mustExec(t, pool, ctx, `INSERT INTO relationships (source_id, target_id, type, valid_from, created_at)
VALUES ($1,$2,'hosts',now(),now())`, hostID, guestID)
et, err := ResolveExecTargetForCheck(ctx, pool, guestID, "lxc", DefaultUser)
if err != nil {
t.Fatalf("ResolveExecTargetForCheck: %v", err)
}
if et.Host != "192.168.8.77" {
t.Errorf("Host = %q, want proxmox host lan_ip 192.168.8.77 (host-hop)", et.Host)
}
out := et.Wrap("/opt/oikos/checks/cpu_check.sh")
if !strings.Contains(out, "pct exec 132") {
t.Errorf("guest wrap must use pct exec 132, got %q", out)
}
}
func TestResolveExecTargetForCheckHostIsDirect(t *testing.T) {
// A host-like target resolves to its own address with identity wrap.
pool := newRemotePool(t)
ctx := context.Background()
hid := uuid.New()
mustExec(t, pool, ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at)
VALUES ($1,'ws:mini','workstation','mini','active','{"lan_ip":"192.168.178.182","user":"dtoro"}'::jsonb,1,now(),now())`, hid)
et, err := ResolveExecTargetForCheck(ctx, pool, hid, "workstation", DefaultUser)
if err != nil {
t.Fatalf("ResolveExecTargetForCheck: %v", err)
}
if et.Host != "192.168.178.182" {
t.Errorf("Host = %q, want 192.168.178.182", et.Host)
}
// Workstation's top-level `user` must be honored (the mac-mini fix).
if et.User != "dtoro" {
t.Errorf("User = %q, want dtoro (top-level user attr)", et.User)
}
if cmd := et.Wrap("uptime"); cmd != "uptime" {
t.Errorf("host wrap must be identity, got %q", cmd)
}
}

68
internal/remote/testutil_test.go Normal file
View File

@@ -0,0 +1,68 @@
package remote
import (
"context"
"fmt"
"math/rand"
"os"
"strings"
"testing"
"github.com/dtoro/oikos/internal/db"
"github.com/jackc/pgx/v5"
)
// createTestDB provisions a throwaway migrated database off baseURL, the same
// convention as internal/scheduler/coverage_test.go. The base URL must point
// at a Postgres superuser-capable connection.
func createTestDB(t *testing.T, baseURL string) *db.Pool {
t.Helper()
ctx := context.Background()
admin, err := pgx.Connect(ctx, baseURL)
if err != nil {
t.Fatalf("connect admin: %v", err)
}
dbName := fmt.Sprintf("oikos_rem_%08x", rand.Int63())
if _, err := admin.Exec(ctx, "CREATE DATABASE "+dbName); err != nil {
admin.Close(ctx)
t.Fatalf("create test db: %v", err)
}
admin.Close(ctx)
at := strings.LastIndex(baseURL, "/")
testURL := baseURL[:at+1] + dbName
if q := strings.Index(baseURL[at:], "?"); q >= 0 {
testURL += baseURL[at+q:]
}
pool, err := db.New(ctx, testURL)
if err != nil {
t.Fatalf("connect test db: %v", err)
}
if err := pool.Migrate(ctx); err != nil {
t.Fatalf("migrate: %v", err)
}
t.Cleanup(func() {
pool.Close()
if admin, err := pgx.Connect(ctx, baseURL); err == nil {
admin.Exec(ctx, "DROP DATABASE IF EXISTS "+dbName+" WITH (FORCE)")
admin.Close(ctx)
}
})
return pool
}
func getenvOrDefault(key, def string) string {
if v := os.Getenv(key); v != "" {
return v
}
return def
}
func mustExec(t *testing.T, pool *db.Pool, ctx context.Context, q string, args ...any) {
t.Helper()
if _, err := pool.Exec(ctx, q, args...); err != nil {
t.Fatalf("exec %s: %v", q, err)
}
}

93
internal/scheduler/scheduler.go
View File

@@ -23,6 +23,7 @@ import (
"github.com/dtoro/oikos/internal/db"
"github.com/dtoro/oikos/internal/db/sqlcgen"
"github.com/dtoro/oikos/internal/observability"
"github.com/dtoro/oikos/internal/remote"
"github.com/google/uuid"
"golang.org/x/sync/errgroup"
"golang.org/x/sys/unix"
@@ -111,7 +112,7 @@ func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDef
q := sqlcgen.New(pool)
start := time.Now()
result := executeCheck(ctx, cd)
result := executeCheck(ctx, pool, cd)
// Stamp the run before processing the result: due-ness must advance even
// when a check fails, or a permanently failing check would be re-run on
@@ -277,8 +278,10 @@ type checkResult struct {
err error
}
// executeCheck dispatches to the appropriate checker by kind.
func executeCheck(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
// executeCheck dispatches to the appropriate checker by kind. pool is needed
// by the ssh-script path, which resolves the target's execution endpoint
// (guests route through their Proxmox host; see internal/remote).
func executeCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
switch cd.Kind {
case "http":
return checkHTTP(ctx, cd)
@@ -291,7 +294,7 @@ func executeCheck(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) check
case "ping":
return checkPing(ctx, cd)
case "ssh-script":
return checkSSHScript(ctx, cd)
return checkSSHScript(ctx, pool, cd)
case "backup-freshness":
return checkBackupFreshness(ctx, cd)
default:
@@ -706,8 +709,16 @@ func parsePingLatency(output []byte) float64 {
return val
}
// checkSSHScript executes an allowlisted script on a remote host via SSH.
func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
// checkSSHScript executes an allowlisted script on a remote target via SSH.
//
// Routing follows the canonical access model (internal/remote): an LXC or VM
// is NEVER SSH'd into directly — it is reached through its Proxmox host via
// `pct exec`/`qm guest exec`, so a guest needs no lan_ip, sshd, or authorized
// key of its own. Hosts and workstations are reached by direct SSH, resolved
// live so a workstation's login (mac-mini: `user: dtoro`) is honored without
// a re-seed. Services and other entities fall back to the host address baked
// into check config at seed time (their hosting container's address).
func checkSSHScript(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
cfg := struct {
Host string `json:"host"`
Port int `json:"port"`
@@ -725,12 +736,6 @@ func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) che
if cfg.Host == "" || cfg.Script == "" {
return checkResult{health: "healthy"}
}
if cfg.Port == 0 {
cfg.Port = 22
}
if cfg.User == "" {
cfg.User = sshUser
}
if !allowlistedScript(cfg.Script) {
return checkResult{
@@ -753,13 +758,45 @@ func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) che
// the remote command. The script name itself is allowlisted above.
scriptPath += " '" + strings.ReplaceAll(cfg.Args, "'", `'\''`) + "'"
}
port := strconv.Itoa(cfg.Port)
output, err := sshExec(ctx, cfg.Host, port, cfg.User, scriptPath, timeout)
// Resolve the execution endpoint. Guests host-hop; host/workstation types
// resolve their address + user live; everything else uses the baked config.
host, port, user := cfg.Host, strconv.Itoa(oru(cfg.Port, 22)), orStr(cfg.User, sshUser)
wrap := func(cmd string) string { return cmd }
targetType := ""
if cd.TargetType != nil {
targetType = *cd.TargetType
}
if cd.TargetID != nil {
switch {
case remote.IsGuest(targetType):
et, err := remote.ResolveExecTargetForCheck(ctx, pool, *cd.TargetID, targetType, sshUser)
if err != nil {
return checkResult{
health: "down", signalKind: "ssh-script",
evidence: fmt.Sprintf("route guest %s: %v", cd.EntitySlug, err), err: err,
}
}
host, port, user, wrap = et.Host, "22", et.User, et.Wrap
case isMachine(targetType):
et, err := remote.ResolveExecTargetForCheck(ctx, pool, *cd.TargetID, targetType, sshUser)
if err == nil {
host, port, user, wrap = et.Host, "22", et.User, et.Wrap
} else {
// Log the resolution failure so an opaque ssh "down" doesn't
// hide that the real cause was host/user resolution (e.g. a
// missing attribute), then fall back to the baked config below.
slog.Warn("scheduler: machine target resolution failed, using baked config",
"entity", cd.EntitySlug, "target_type", targetType, "error", err)
}
}
}
output, err := sshExec(ctx, host, port, user, wrap(scriptPath), timeout)
if err != nil {
return checkResult{
health: "down", signalKind: "ssh-script",
evidence: fmt.Sprintf("ssh %s:%s %s: %v", cfg.Host, strconv.Itoa(cfg.Port), cfg.Script, err),
evidence: fmt.Sprintf("ssh %s:%s %s: %v", host, port, cfg.Script, err),
err: err,
}
}
@@ -797,6 +834,32 @@ func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) che
}
}
// isMachine reports whether a target type is a physical/virtual machine that
// should be reached by direct SSH at its own resolved address (rather than the
// baked hosting-container address a service uses). These are the machine
// subtypes in the ontology.
func isMachine(entityType string) bool {
switch entityType {
case "proxmox-host", "standalone-server", "workstation", "appliance":
return true
}
return false
}
// oru returns v when nonzero, else def. orStr returns v when non-empty, else def.
func oru(v, def int) int {
if v != 0 {
return v
}
return def
}
func orStr(v, def string) string {
if v != "" {
return v
}
return def
}
var scriptNameRe = regexp.MustCompile(`^[a-z][a-z0-9_-]+\.sh$`)
func allowlistedScript(name string) bool {