diff --git a/CLAUDE.md b/CLAUDE.md index c623059..8b1dd42 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,6 +2,10 @@ Project-specific rules for Claude when working in this repo. +## Repo + +The repo lives inside a Gitea instance; `tea` CLI is available for use by agents + ## Run `go vet` before every commit CI runs `go vet ./...` and will fail the build on any vet error. @@ -43,6 +47,8 @@ cp bin/restic-manager-agent \ /tmp/rm-smoke/data/agent-binaries/restic-manager-agent-linux-amd64 cp deploy/install/install.sh \ /tmp/rm-smoke/data/install/install.sh +cp deploy/install/install.ps1 \ + /tmp/rm-smoke/data/install/install.ps1 cp deploy/install/restic-manager-agent.service \ /tmp/rm-smoke/data/install/restic-manager-agent.service diff --git a/cmd/agent/announce.go b/cmd/agent/announce.go new file mode 100644 index 0000000..536baba --- /dev/null +++ b/cmd/agent/announce.go @@ -0,0 +1,262 @@ +// announce.go — agent-side announce-and-approve enrolment (P2-18c). +// +// Run path: when the agent has no AgentToken set but RM_SERVER is +// configured (and no -enroll-token was supplied), main() switches +// into announce mode: +// 1. Load (or mint+persist) an Ed25519 keypair in agent.yaml. +// 2. POST {hostname, os, arch, agent_version, restic_version, +// public_key} to /api/agents/announce. +// 3. Print the fingerprint to stderr in a copy-friendly banner so +// the operator can compare it against the dashboard. +// 4. Open /ws/agent/pending?pending_id=…, sign the nonce with our +// private key, wait for an `enrolled` message. +// 5. On enrolled: persist the bearer + repo creds, return; main() +// then drops into the normal WS run loop with the new bearer. +// 6. On reject: server closes the socket with code 4001; we exit +// with a clear message. +package main + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + stdhttp "net/http" + "os" + "strings" + "time" + + "github.com/coder/websocket" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/config" + "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/secrets" + "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/sysinfo" + "gitea.dcglab.co.uk/steve/restic-manager/internal/store" +) + +// announceRequest mirrors the server's announceRequest. Duplicated +// here so cmd/agent stays decoupled from the http package. +type announceRequest struct { + Hostname string `json:"hostname"` + OS string `json:"os"` + Arch string `json:"arch"` + AgentVersion string `json:"agent_version"` + ResticVersion string `json:"restic_version"` + PublicKey string `json:"public_key"` +} + +type announceResponse struct { + PendingID string `json:"pending_id"` + Fingerprint string `json:"fingerprint"` + HostnameCollision bool `json:"hostname_collision"` +} + +type pendingNonceMessage struct { + Type string `json:"type"` + Nonce string `json:"nonce"` +} + +type pendingSignedMessage struct { + Type string `json:"type"` + Signature string `json:"signature"` +} + +type pendingEnrolledMessage struct { + Type string `json:"type"` + HostID string `json:"host_id"` + Bearer string `json:"bearer"` +} + +// doAnnounce runs the full announce → wait-for-accept flow. On +// success, persists the bearer + host_id into cfg + writes secrets +// for the repo creds the admin supplied at accept time. Returns +// only after the bearer has landed (or on hard error / reject). +func doAnnounce(serverURL string, cfg *config.Config, agentVersion string) error { + ctx, cancel := context.WithTimeout(context.Background(), 24*time.Hour) + defer cancel() + + // Ensure we have a keypair. + priv, pub, err := loadOrMintAnnounceKey(cfg) + if err != nil { + return fmt.Errorf("announce: keypair: %w", err) + } + fingerprint := store.FingerprintForKey(pub) + + snap, err := sysinfo.Collect(ctx, cfg.ResticPath) + if err != nil { + return fmt.Errorf("announce: sysinfo: %w", err) + } + + // POST /api/agents/announce. + body, _ := json.Marshal(announceRequest{ + Hostname: snap.Hostname, OS: string(snap.OS), Arch: string(snap.Arch), + AgentVersion: agentVersion, ResticVersion: snap.ResticVersion, + PublicKey: base64.StdEncoding.EncodeToString(pub), + }) + req, _ := stdhttp.NewRequestWithContext(ctx, "POST", + strings.TrimRight(serverURL, "/")+"/api/agents/announce", + strings.NewReader(string(body))) + req.Header.Set("Content-Type", "application/json") + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + return fmt.Errorf("announce: POST: %w", err) + } + rawBody := readAllShort(res) + _ = res.Body.Close() + if res.StatusCode != stdhttp.StatusOK { + return fmt.Errorf("announce: server returned %d: %s", res.StatusCode, rawBody) + } + var ar announceResponse + if err := json.Unmarshal(rawBody, &ar); err != nil { + return fmt.Errorf("announce: parse response: %w", err) + } + + // Print the fingerprint banner. + fmt.Fprintln(os.Stderr, strings.Repeat("=", 64)) + fmt.Fprintln(os.Stderr, " Restic-manager: announce-and-approve enrolment") + fmt.Fprintln(os.Stderr, "") + fmt.Fprintln(os.Stderr, " Hostname : "+snap.Hostname) + fmt.Fprintln(os.Stderr, " Server : "+serverURL) + fmt.Fprintln(os.Stderr, " Pending ID : "+ar.PendingID) + fmt.Fprintln(os.Stderr, " Fingerprint : "+fingerprint) + if ar.HostnameCollision { + fmt.Fprintln(os.Stderr, "") + fmt.Fprintln(os.Stderr, " WARNING: another pending host already uses this hostname.") + fmt.Fprintln(os.Stderr, " Confirm the fingerprint above matches what you see in the UI.") + } + fmt.Fprintln(os.Stderr, "") + fmt.Fprintln(os.Stderr, " Compare the fingerprint with the one in the UI before accepting.") + fmt.Fprintln(os.Stderr, " Waiting for an admin to accept (1 hour timeout)…") + fmt.Fprintln(os.Stderr, strings.Repeat("=", 64)) + + // Open /ws/agent/pending and run the nonce-sign handshake. + wsURL := wsURLFromHTTP(serverURL) + "/ws/agent/pending?pending_id=" + ar.PendingID + dialCtx, dialCancel := context.WithTimeout(ctx, 30*time.Second) + c, dialRes, err := websocket.Dial(dialCtx, wsURL, nil) + dialCancel() + if err != nil { + return fmt.Errorf("announce: dial pending ws: %w", err) + } + if dialRes != nil && dialRes.Body != nil { + _ = dialRes.Body.Close() + } + defer func() { _ = c.CloseNow() }() + + // Read nonce. + rctx, rcancel := context.WithTimeout(ctx, 30*time.Second) + _, raw, err := c.Read(rctx) + rcancel() + if err != nil { + return fmt.Errorf("announce: read nonce: %w", err) + } + var nm pendingNonceMessage + if err := json.Unmarshal(raw, &nm); err != nil { + return fmt.Errorf("announce: parse nonce: %w", err) + } + nonce, err := base64.StdEncoding.DecodeString(nm.Nonce) + if err != nil { + return fmt.Errorf("announce: decode nonce: %w", err) + } + sig := ed25519.Sign(priv, nonce) + reply, _ := json.Marshal(pendingSignedMessage{ + Type: "signed_nonce", Signature: base64.StdEncoding.EncodeToString(sig), + }) + wctx, wcancel := context.WithTimeout(ctx, 10*time.Second) + if err := c.Write(wctx, websocket.MessageText, reply); err != nil { + wcancel() + return fmt.Errorf("announce: write signed nonce: %w", err) + } + wcancel() + + // Block until enrolled (or reject / disconnect). + rctx2, rcancel2 := context.WithTimeout(ctx, 1*time.Hour) + defer rcancel2() + _, raw2, err := c.Read(rctx2) + if err != nil { + // CloseError with our reject code 4001 = admin rejected. + var ce websocket.CloseError + if errors.As(err, &ce) && ce.Code == 4001 { + return errors.New("announce: rejected by admin") + } + return fmt.Errorf("announce: wait for enrolled: %w", err) + } + var em pendingEnrolledMessage + if err := json.Unmarshal(raw2, &em); err != nil { + return fmt.Errorf("announce: parse enrolled: %w", err) + } + if em.Type != "enrolled" || em.Bearer == "" { + return fmt.Errorf("announce: bad enrolled payload: %s", raw2) + } + + // Persist the bearer + host_id. + cfg.ServerURL = serverURL + cfg.HostID = em.HostID + cfg.AgentToken = em.Bearer + if err := cfg.EnsureSecretsKey(); err != nil { + return fmt.Errorf("announce: mint secrets key: %w", err) + } + // Note: repo creds aren't pushed in the enrolled message — the + // server pushes them via `config.update` on first WS hello. The + // secrets store will start empty and fill in then. + if err := cfg.Save(); err != nil { + return fmt.Errorf("announce: save config: %w", err) + } + // Touch the secrets store so it exists with the right perms. + keyBytes, _ := cfg.SecretsKeyBytes() + if _, err := secrets.New(cfg.ResolvedSecretsPath(), keyBytes); err != nil { + return fmt.Errorf("announce: open secrets store: %w", err) + } + fmt.Fprintln(os.Stderr, "Accepted. Bearer persisted; reconnecting via the standard WS.") + return nil +} + +// loadOrMintAnnounceKey returns the (priv, pub) keypair, generating +// + persisting one when AnnounceKey is empty. The private key holds +// the public half in its tail 32 bytes per ed25519 convention. +func loadOrMintAnnounceKey(cfg *config.Config) (ed25519.PrivateKey, ed25519.PublicKey, error) { + if cfg.AnnounceKey != "" { + raw, err := base64.StdEncoding.DecodeString(cfg.AnnounceKey) + if err != nil { + return nil, nil, fmt.Errorf("decode AnnounceKey: %w", err) + } + if len(raw) != ed25519.PrivateKeySize { + return nil, nil, fmt.Errorf("AnnounceKey must be %d bytes, got %d", + ed25519.PrivateKeySize, len(raw)) + } + priv := ed25519.PrivateKey(raw) + pub := priv.Public().(ed25519.PublicKey) + return priv, pub, nil + } + pub, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + return nil, nil, fmt.Errorf("generate keypair: %w", err) + } + cfg.AnnounceKey = base64.StdEncoding.EncodeToString(priv) + if err := cfg.Save(); err != nil { + return nil, nil, fmt.Errorf("persist AnnounceKey: %w", err) + } + return priv, pub, nil +} + +// wsURLFromHTTP swaps the http(s) scheme for ws(s). +func wsURLFromHTTP(httpURL string) string { + switch { + case strings.HasPrefix(httpURL, "https://"): + return "wss://" + strings.TrimPrefix(httpURL, "https://") + case strings.HasPrefix(httpURL, "http://"): + return "ws://" + strings.TrimPrefix(httpURL, "http://") + default: + return httpURL + } +} + +// readAllShort reads up to 64KB of the response body. The announce +// response is small; we cap to avoid pathological server replies. +func readAllShort(res *stdhttp.Response) []byte { + buf := make([]byte, 64*1024) + n, _ := res.Body.Read(buf) + return buf[:n] +} diff --git a/cmd/agent/main.go b/cmd/agent/main.go index d401640..ac43d3c 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -9,6 +9,7 @@ import ( "os" "os/signal" "strconv" + "sync" "syscall" "time" @@ -16,6 +17,7 @@ import ( "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/runner" "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/scheduler" "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/secrets" + "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/service" "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/sysinfo" "gitea.dcglab.co.uk/steve/restic-manager/internal/agent/wsclient" "gitea.dcglab.co.uk/steve/restic-manager/internal/api" @@ -32,6 +34,27 @@ func main() { } func run() error { + // Optional first positional verb for SCM control on Windows. + // `restic-manager-agent install|uninstall|start|stop` route into + // the service package; everything else falls through to the + // flag-driven default (which is what systemd / interactive runs + // hit). On non-Windows builds these verbs return a clear error. + if len(os.Args) > 1 { + switch os.Args[1] { + case "install": + return service.Install() + case "uninstall": + return service.Uninstall() + case "start": + return service.Start() + case "stop": + return service.Stop() + case "run": + // Strip the verb so flag.Parse sees the rest unchanged. + os.Args = append([]string{os.Args[0]}, os.Args[2:]...) + } + } + configPath := flag.String("config", config.DefaultPath(), "path to agent.yaml") enrollServer := flag.String("enroll-server", "", "server URL (used with -enroll-token to perform first-run enrollment)") enrollToken := flag.String("enroll-token", "", "one-time enrollment token (operator copies this from the UI)") @@ -58,8 +81,17 @@ func run() error { return doEnroll(*enrollServer, *enrollToken, cfg, version) } + // Announce-and-approve: -enroll-server set, no token, agent not + // yet enrolled. Run the announce flow inline; on success the cfg + // has the bearer + host_id and we drop into the normal run loop. + if !cfg.Enrolled() && *enrollServer != "" { + if err := doAnnounce(*enrollServer, cfg, version); err != nil { + return fmt.Errorf("announce: %w", err) + } + } + if !cfg.Enrolled() { - return fmt.Errorf("agent is not enrolled; run with -enroll-server and -enroll-token first (config %q)", *configPath) + return fmt.Errorf("agent is not enrolled; run with -enroll-server (and either -enroll-token or wait for admin to accept the announce) first (config %q)", *configPath) } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) @@ -170,6 +202,14 @@ type dispatcher struct { resticBin string secrets *secrets.Store scheduler *scheduler.Scheduler + + // Bandwidth caps in KB/s pushed via config.update. Mutated under + // bwMu by the config.update handler; read by runJob when building + // the runner. <=0 means "no cap" (do not pass --limit-* to restic). + // Per-job overrides on CommandRunPayload take precedence. + bwMu sync.Mutex + bwUpKBps int + bwDownKBps int } func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.Sender) error { @@ -263,6 +303,24 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S slog.Warn("ws agent: unknown config.update slot, ignoring", "slot", p.Slot) } + // Bandwidth caps ride independently of the slot — they're host- + // wide and apply to every restic invocation regardless of which + // credentials slot the job uses. nil pointer = no change in this + // push; non-nil = set to that value (≤0 clears the cap). + if p.BandwidthUpKBps != nil || p.BandwidthDownKBps != nil { + d.bwMu.Lock() + if p.BandwidthUpKBps != nil { + d.bwUpKBps = *p.BandwidthUpKBps + } + if p.BandwidthDownKBps != nil { + d.bwDownKBps = *p.BandwidthDownKBps + } + up, down := d.bwUpKBps, d.bwDownKBps + d.bwMu.Unlock() + slog.Info("ws agent: bandwidth caps updated", + "up_kbps", up, "down_kbps", down) + } + case api.MsgAgentUpdateAvail: var p api.AgentUpdateAvailablePayload _ = env.UnmarshalPayload(&p) @@ -295,11 +353,25 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc // not on r). If you find yourself adding a new JobKind that // needs delete authority, mirror the JobPrune pattern below // — don't try to overload r. + // Resolve bandwidth caps: per-job override (if set) wins over the + // host-wide caps last pushed via config.update. <=0 means no cap. + d.bwMu.Lock() + upKBps, downKBps := d.bwUpKBps, d.bwDownKBps + d.bwMu.Unlock() + if p.BandwidthUpKBps != nil { + upKBps = *p.BandwidthUpKBps + } + if p.BandwidthDownKBps != nil { + downKBps = *p.BandwidthDownKBps + } + r := runner.New(runner.Config{ - ResticBin: d.resticBin, - RepoURL: creds.URL, - RepoUsername: creds.Username, - RepoPassword: creds.Password, + ResticBin: d.resticBin, + RepoURL: creds.URL, + RepoUsername: creds.Username, + RepoPassword: creds.Password, + LimitUploadKBps: upKBps, + LimitDownloadKBps: downKBps, }, tx, time.Second) switch p.Kind { @@ -318,8 +390,9 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc } slog.Info("agent: accepting backup job", "job_id", p.JobID, "paths", paths, "excludes", p.Excludes, "tag", p.Tag) + hooks := runner.BackupHooks{Pre: p.PreHook, Post: p.PostHook} go func() { - if err := r.RunBackup(ctx, p.JobID, paths, p.Excludes, tags); err != nil { + if err := r.RunBackup(ctx, p.JobID, paths, p.Excludes, tags, hooks); err != nil { slog.Warn("agent: backup job failed", "job_id", p.JobID, "err", err) return } @@ -381,10 +454,12 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc runCreds = ac } prr := runner.New(runner.Config{ - ResticBin: d.resticBin, - RepoURL: runCreds.URL, - RepoUsername: runCreds.Username, - RepoPassword: runCreds.Password, + ResticBin: d.resticBin, + RepoURL: runCreds.URL, + RepoUsername: runCreds.Username, + RepoPassword: runCreds.Password, + LimitUploadKBps: upKBps, + LimitDownloadKBps: downKBps, }, tx, time.Second) slog.Info("agent: accepting prune job", "job_id", p.JobID, "admin_creds", p.RequiresAdminCreds) go func() { diff --git a/cmd/server/main.go b/cmd/server/main.go index c97b39d..a083a6d 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -156,6 +156,10 @@ func run() error { // shouldn't, but the queue exists either way). pendingDrainTick := time.NewTicker(30 * time.Second) defer pendingDrainTick.Stop() + // Pending-hosts expiry sweeper: drops announce rows past their 1h + // ceiling so the dashboard panel doesn't accumulate stale entries. + pendingExpiryTick := time.NewTicker(60 * time.Second) + defer pendingExpiryTick.Stop() mt := maintenance.New(st) go func() { for { @@ -176,6 +180,10 @@ func run() error { } case <-pendingDrainTick.C: srv.DrainAllDue(ctx) + case <-pendingExpiryTick.C: + if n, err := st.DeleteExpiredPendingHosts(ctx, time.Now().UTC()); err == nil && n > 0 { + slog.Info("expired pending hosts swept", "n", n) + } case <-maintenanceTick.C: decisions, err := mt.Decide(ctx, time.Now().UTC()) if err != nil { diff --git a/deploy/install/install.ps1 b/deploy/install/install.ps1 new file mode 100644 index 0000000..72b7c9d --- /dev/null +++ b/deploy/install/install.ps1 @@ -0,0 +1,133 @@ +# install.ps1 — Windows installer for the restic-manager agent (P2-17). +# +# Usage (Run as administrator): +# $env:RM_SERVER = "https://restic.lab.example" +# $env:RM_TOKEN = "" # omit for announce-and-approve +# iwr "$env:RM_SERVER/install/install.ps1" -UseBasicParsing | iex +# +# What it does: +# 1. checks for admin elevation +# 2. downloads the matching agent binary from the server +# 3. lays down C:\Program Files\restic-manager\ and +# C:\ProgramData\restic-manager\ (config + state) +# 4. registers the agent as a Windows service via the agent's own +# `install` subcommand (which uses the SCM API) +# 5. enrolls (token flow if RM_TOKEN set, otherwise announce flow) +# by spawning the agent with the right CLI flags and waits +# until config is written +# 6. surfaces (but does NOT disable) any existing scheduled tasks +# whose name contains "restic" so the operator can decide +# +# Idempotent — safe to re-run. + +[CmdletBinding()] +param( + [string]$Server = $env:RM_SERVER, + [string]$Token = $env:RM_TOKEN, + [string]$InstallDir = 'C:\Program Files\restic-manager', + [string]$DataDir = 'C:\ProgramData\restic-manager' +) + +$ErrorActionPreference = 'Stop' + +function Test-Admin { + $id = [System.Security.Principal.WindowsIdentity]::GetCurrent() + $pri = New-Object System.Security.Principal.WindowsPrincipal($id) + return $pri.IsInRole([System.Security.Principal.WindowsBuiltInRole]::Administrator) +} + +function Detect-Arch { + switch ($env:PROCESSOR_ARCHITECTURE) { + 'AMD64' { return 'amd64' } + 'ARM64' { return 'arm64' } + default { throw "unsupported PROCESSOR_ARCHITECTURE: $($env:PROCESSOR_ARCHITECTURE)" } + } +} + +function Detect-ResticTasks { + Write-Host '' + Write-Host '— Existing restic-named scheduled tasks (review manually) —' + try { + $tasks = Get-ScheduledTask -ErrorAction SilentlyContinue | + Where-Object { $_.TaskName -match 'restic' -or $_.TaskPath -match 'restic' } + if ($tasks) { + foreach ($t in $tasks) { + Write-Host " * $($t.TaskPath)$($t.TaskName) state=$($t.State)" + Write-Host " Disable with: Disable-ScheduledTask -TaskName '$($t.TaskName)' -TaskPath '$($t.TaskPath)'" + } + } else { + Write-Host ' (none found)' + } + } catch { + Write-Host ' (Get-ScheduledTask failed; review the Task Scheduler UI manually)' + } + Write-Host '' +} + +# --- preflight ------------------------------------------------------- + +if (-not (Test-Admin)) { + throw 'install.ps1: must be run from an elevated PowerShell (Run as administrator).' +} +if (-not $Server) { + throw 'install.ps1: -Server (or $env:RM_SERVER) is required, e.g. https://restic.lab.example' +} + +$arch = Detect-Arch +Write-Host "install.ps1: server=$Server arch=$arch" + +# --- directories ----------------------------------------------------- + +New-Item -ItemType Directory -Force -Path $InstallDir | Out-Null +New-Item -ItemType Directory -Force -Path $DataDir | Out-Null + +# --- download agent -------------------------------------------------- + +$agentExe = Join-Path $InstallDir 'restic-manager-agent.exe' +$tmpExe = "$agentExe.tmp" +$dlURL = "$Server/agent/binary?os=windows&arch=$arch" +Write-Host "install.ps1: downloading $dlURL" +Invoke-WebRequest -UseBasicParsing -Uri $dlURL -OutFile $tmpExe +# Atomic-ish replace: stop service if running so the .exe isn't busy. +try { Stop-Service -Name 'restic-manager-agent' -ErrorAction SilentlyContinue } catch {} +Move-Item -Force -Path $tmpExe -Destination $agentExe + +# --- enroll / announce ----------------------------------------------- + +$cfgPath = Join-Path $DataDir 'agent.yaml' +$args = @('-config', $cfgPath, '-enroll-server', $Server) +if ($Token) { + $args += @('-enroll-token', $Token) + Write-Host 'install.ps1: enrolling with one-time token' +} else { + Write-Host 'install.ps1: no RM_TOKEN — running announce-and-approve flow.' + Write-Host ' The fingerprint will print below. Compare it with the dashboard before clicking Accept.' +} +& $agentExe @args +if ($LASTEXITCODE -ne 0) { + throw "install.ps1: agent enrolment failed (exit $LASTEXITCODE)" +} + +# --- install + start service ---------------------------------------- + +# The 'install' subcommand registers the service via the SCM. If +# already registered, it errors loudly — re-run with -Force only if +# you've manually verified. +try { + & $agentExe install +} catch { + Write-Host "install.ps1: service may already be registered ($_); continuing." +} +try { + Start-Service -Name 'restic-manager-agent' +} catch { + Write-Host "install.ps1: Start-Service failed ($_); check Event Viewer." +} + +Detect-ResticTasks + +Write-Host '' +Write-Host 'install.ps1: done.' +Write-Host " config : $cfgPath" +Write-Host " binary : $agentExe" +Write-Host " service: restic-manager-agent (Get-Service to inspect)" diff --git a/internal/agent/config/config.go b/internal/agent/config/config.go index c10e20c..1e0cdd1 100644 --- a/internal/agent/config/config.go +++ b/internal/agent/config/config.go @@ -62,6 +62,13 @@ type Config struct { LegacyRepoURL string `yaml:"repo_url,omitempty"` LegacyRepoPassword string `yaml:"repo_password,omitempty"` + // AnnounceKey is the base64-encoded Ed25519 private key used by + // announce-and-approve enrolment (P2-18). Generated on first + // announce, persisted so the agent can re-attach to the same + // pending row across restarts. 64 bytes when decoded. + // Empty for token-flow enrolments. + AnnounceKey string `yaml:"announce_key,omitempty"` + // path is the file we loaded from. Used by Save. path string `yaml:"-"` } diff --git a/internal/agent/runner/hooks.go b/internal/agent/runner/hooks.go new file mode 100644 index 0000000..904b100 --- /dev/null +++ b/internal/agent/runner/hooks.go @@ -0,0 +1,106 @@ +// hooks.go — pre/post backup hooks for the agent runner (P2R-11). +// +// Hooks fire only for backup jobs (the runner's other kinds — +// init/forget/prune/check/unlock — call shell scripts that touch +// repo internals; running operator hooks for those would be +// surprising). Hook bodies arrive plaintext on the wire (server +// decrypted before the WS push). The agent never persists them +// to disk; they live in memory for the lifetime of one job. +// +// Failure semantics: +// - pre_hook non-zero exit aborts the backup: the runner returns +// the error, the job is recorded as failed, and the actual +// restic invocation never runs. +// - post_hook non-zero exit is logged with a warning prefix in +// the job log but does NOT change the job status — the operator +// wants the backup result preserved even if the cleanup step +// misbehaved. +// +// Streaming: each line of the hook's stdout/stderr is shipped as a +// log.stream envelope with payload prefixed `hook: ` so the live +// log viewer can visually separate it from restic's own output. +package runner + +import ( + "bufio" + "context" + "fmt" + "io" + "os/exec" + "runtime" + "sync/atomic" + "time" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" +) + +// runHook executes script via the host shell. status is the value +// passed as RM_JOB_STATUS in the env (empty for pre-hooks; the +// final job status — "succeeded" or "failed" — for post-hooks). +// Returns an error iff the hook exited non-zero. ctx cancellation +// kills the subprocess. +func (r *Runner) runHook(ctx context.Context, jobID, phase, script, status string, seq *atomic.Int64) error { + if script == "" { + return nil + } + shell, flag := defaultShell() + cmd := exec.CommandContext(ctx, shell, flag, script) + cmd.Env = []string{ + "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + } + if status != "" { + cmd.Env = append(cmd.Env, "RM_JOB_STATUS="+status) + } + cmd.Env = append(cmd.Env, "RM_JOB_ID="+jobID, "RM_HOOK_PHASE="+phase) + + stdout, err := cmd.StdoutPipe() + if err != nil { + return fmt.Errorf("hook %s: stdout pipe: %w", phase, err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + return fmt.Errorf("hook %s: stderr pipe: %w", phase, err) + } + if err := cmd.Start(); err != nil { + return fmt.Errorf("hook %s: start: %w", phase, err) + } + done := make(chan struct{}, 2) + go func() { r.pumpHookLines(stdout, "stdout", phase, jobID, seq); done <- struct{}{} }() + go func() { r.pumpHookLines(stderr, "stderr", phase, jobID, seq); done <- struct{}{} }() + <-done + <-done + if werr := cmd.Wait(); werr != nil { + return fmt.Errorf("hook %s exited non-zero: %w", phase, werr) + } + return nil +} + +// pumpHookLines streams lines as log.stream envelopes prefixed with +// "hook(): " so the live log can visually separate them. +func (r *Runner) pumpHookLines(rd io.Reader, stream, phase, jobID string, seq *atomic.Int64) { + scanner := bufio.NewScanner(rd) + scanner.Buffer(make([]byte, 0, 64*1024), 256*1024) + for scanner.Scan() { + line := "hook(" + phase + "): " + scanner.Text() + env, _ := api.Marshal(api.MsgLogStream, "", api.LogStreamLine{ + JobID: jobID, + Seq: seq.Add(1), + TS: time.Now().UTC(), + Stream: api.LogStream(stream), + Payload: line, + }) + _ = r.tx.Send(env) + } +} + +// defaultShell returns the (binary, single-arg-flag) pair to use for +// ` "