admin
6d7904786c
gates / gates (push) Successful in 7s
Link 7's only production caller was a --selftest reading R from an env var. Link 8 did not exist: that selftest writes the whole bundle JSON and its success message named "tunnel_token + pbs_token" -- accurate when written, a misstatement since v0.77.0 sealed the offsite repository password into the same bundle. It now names what THIS bundle carried and what it did not. POST /escrow/recover-offsite-password: the controller supplies R, the agent fetches this host's own blob from the hub (self-scoped by the per-host key), unseals it, and returns ONLY the offsite restic repository password plus its sha256. Not the tunnel token, not the PBS token, not the WG key -- the controller is a trust tier down and needs none of them. R: in memory for one call, cleared on every path, never on disk, never in argv, never logged, never echoed. A test redirects TMPDIR and asserts the tree is EMPTY afterwards -- emptiness rather than a content scan, because a content scan is defeated by a later call overwriting the leaked file, which is how the first version of that test passed its own red-proof while R sat on disk. Three distinct outcomes: no blob (404), a bundle that opens but predates the field (409), a code that does not open it (400, fail-closed at the KDF, nothing written). The wiring is asserted by an AST walk from func main() to the Options field, not by grep.
357 lines
14 KiB
Go
357 lines
14 KiB
Go
package hub
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/tls"
|
|
"crypto/x509"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.dooplex.hu/admin/felhom-agent/internal/config"
|
|
)
|
|
|
|
const reportPath = "/api/v1/host-report"
|
|
|
|
// Client posts host-reports to the hub. Auth is a per-host Bearer key. Transport is
|
|
// standard TLS (system roots, or a CAFile pool); verification is always on — the hub
|
|
// has a real cert (unlike the Proxmox self-signed path), so there is no insecure mode.
|
|
type Client struct {
|
|
baseURL string
|
|
apiKey string
|
|
hostID string // for the slice-10A desired-state/jobs paths (/hosts/{hostID}/…)
|
|
hc *http.Client
|
|
logger *slog.Logger
|
|
}
|
|
|
|
// NewClient builds a hub client from config (defaults applied). It never logs the key.
|
|
func NewClient(cfg config.HubConfig, logger *slog.Logger) (*Client, error) {
|
|
cfg = cfg.WithDefaults()
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
tlsCfg := &tls.Config{} // system roots
|
|
if cfg.CAFile != "" {
|
|
pem, err := os.ReadFile(cfg.CAFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("hub: reading ca_file: %w", err)
|
|
}
|
|
pool := x509.NewCertPool()
|
|
if !pool.AppendCertsFromPEM(pem) {
|
|
return nil, fmt.Errorf("hub: ca_file %q contained no usable certificates", cfg.CAFile)
|
|
}
|
|
tlsCfg.RootCAs = pool
|
|
}
|
|
hc := &http.Client{
|
|
Timeout: time.Duration(cfg.TimeoutSeconds) * time.Second,
|
|
Transport: &http.Transport{TLSClientConfig: tlsCfg},
|
|
}
|
|
return newClient(cfg.URL, cfg.APIKey, cfg.HostID, hc, logger), nil
|
|
}
|
|
|
|
// newClient is the shared constructor (tests inject a mock-transport *http.Client).
|
|
func newClient(baseURL, apiKey, hostID string, hc *http.Client, logger *slog.Logger) *Client {
|
|
return &Client{baseURL: strings.TrimRight(baseURL, "/"), apiKey: apiKey, hostID: hostID, hc: hc, logger: logger}
|
|
}
|
|
|
|
// TransportError is a network/connection failure (no HTTP response). It never
|
|
// contains the bearer token.
|
|
type TransportError struct{ Err error }
|
|
|
|
func (e *TransportError) Error() string { return "hub: transport error: " + e.Err.Error() }
|
|
func (e *TransportError) Unwrap() error { return e.Err }
|
|
|
|
// HTTPError is a non-2xx response. BodyTail is a short, token-free excerpt.
|
|
type HTTPError struct {
|
|
StatusCode int
|
|
BodyTail string
|
|
}
|
|
|
|
func (e *HTTPError) Error() string {
|
|
return fmt.Sprintf("hub: HTTP %d: %s", e.StatusCode, e.BodyTail)
|
|
}
|
|
|
|
// Report POSTs the host-report and returns the parsed control envelope. The report
|
|
// IS the heartbeat (locked decision 1). Errors are typed (transport vs HTTP) and
|
|
// never include the bearer token.
|
|
func (c *Client) Report(ctx context.Context, r *HostReport) (*ControlEnvelope, error) {
|
|
body, err := json.Marshal(r)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("hub: marshaling report: %w", err)
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+reportPath, bytes.NewReader(body))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("hub: building request: %w", err)
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+c.apiKey)
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("Accept", "application/json")
|
|
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return nil, &TransportError{Err: err} // token is in the request header, never the error
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return nil, &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)}
|
|
}
|
|
var env ControlEnvelope
|
|
if err := json.Unmarshal(raw, &env); err != nil {
|
|
return nil, fmt.Errorf("hub: decoding control envelope: %w", err)
|
|
}
|
|
return &env, nil
|
|
}
|
|
|
|
// FetchDesiredState GETs the host's authoritative desired-state (slice 10A — the "Down" channel's
|
|
// heavy payload). The agent calls this ONLY when the heartbeat envelope's DesiredGeneration has
|
|
// advanced past its cached one (the heartbeat stays light; the state moves on change). It is
|
|
// self-scoped server-side: the per-host key only ever reads ITS OWN host (the client uses its
|
|
// configured hostID). Errors are typed (transport vs HTTP) and never include the bearer token.
|
|
func (c *Client) FetchDesiredState(ctx context.Context) (*DesiredStateResponse, error) {
|
|
if c.hostID == "" {
|
|
return nil, fmt.Errorf("hub: FetchDesiredState requires a configured host_id")
|
|
}
|
|
url := c.baseURL + "/api/v1/hosts/" + c.hostID + "/desired-state"
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("hub: building desired-state request: %w", err)
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+c.apiKey)
|
|
req.Header.Set("Accept", "application/json")
|
|
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return nil, &TransportError{Err: err}
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return nil, &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)}
|
|
}
|
|
var out DesiredStateResponse
|
|
if err := json.Unmarshal(raw, &out); err != nil {
|
|
return nil, fmt.Errorf("hub: decoding desired-state: %w", err)
|
|
}
|
|
return &out, nil
|
|
}
|
|
|
|
// WGRegisterResponse is the hub's answer to a WG pubkey registration (S3; hub S2
|
|
// handleRegisterHostWG). Existed=true = idempotent re-register (nothing moved hub-side).
|
|
type WGRegisterResponse struct {
|
|
Pubkey string `json:"pubkey"`
|
|
AssignedIP string `json:"assigned_ip"` // "10.77.0.2/32"
|
|
Existed bool `json:"existed"`
|
|
Generation int64 `json:"generation"`
|
|
Sync string `json:"sync"` // hub→endpoint push status: ok | deferred:… | disabled | unchanged
|
|
}
|
|
|
|
// RegisterWG registers this host's WG public key with the hub (S3 — doc 06 §3.3 step 2; POST
|
|
// /hosts/{host_id}/wg, per-host key, self-scoped server-side). The hub allocates/keeps the /32,
|
|
// bumps the desired generation on real change, and pushes the peer to the endpoint. Errors are
|
|
// typed (transport vs HTTP: 403 auth, 404 unknown host, 409 conflict/endpoint-unset) and never
|
|
// include the bearer token. Only the PUBLIC key ever travels.
|
|
func (c *Client) RegisterWG(ctx context.Context, pubkey string) (*WGRegisterResponse, error) {
|
|
if c.hostID == "" {
|
|
return nil, fmt.Errorf("hub: RegisterWG requires a configured host_id")
|
|
}
|
|
body, err := json.Marshal(map[string]string{"pubkey": pubkey})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("hub: marshaling wg registration: %w", err)
|
|
}
|
|
url := c.baseURL + "/api/v1/hosts/" + c.hostID + "/wg"
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("hub: building wg-register request: %w", err)
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+c.apiKey)
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("Accept", "application/json")
|
|
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return nil, &TransportError{Err: err}
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return nil, &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)}
|
|
}
|
|
var out WGRegisterResponse
|
|
if err := json.Unmarshal(raw, &out); err != nil {
|
|
return nil, fmt.Errorf("hub: decoding wg-register response: %w", err)
|
|
}
|
|
return &out, nil
|
|
}
|
|
|
|
// ErrNoPBSSecret is the typed "404: no unconsumed PBS token secret staged for this host" outcome
|
|
// (PBS DR slice 2). Absent-or-already-consumed are indistinguishable by design (consume-once).
|
|
var ErrNoPBSSecret = fmt.Errorf("hub: no unconsumed PBS token secret staged for this host")
|
|
|
|
// ConsumePBSToken fetches this host's one-time PBS token secret — EXACTLY ONCE (PBS DR slice 2;
|
|
// POST /api/v1/hosts/{host_id}/pbs/consume-token, per-host key, self-scoped; NOTE the PLURAL
|
|
// /hosts/ — the slice-1 route). A 200 burns the secret hub-side: the caller MUST apply it or
|
|
// surface a loud consumed-but-failed state (never silent-retry). The secret is returned to the
|
|
// caller only — never logged, never in an error.
|
|
func (c *Client) ConsumePBSToken(ctx context.Context) (string, error) {
|
|
if c.hostID == "" {
|
|
return "", fmt.Errorf("hub: ConsumePBSToken requires a configured host_id")
|
|
}
|
|
url := c.baseURL + "/api/v1/hosts/" + c.hostID + "/pbs/consume-token"
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+c.apiKey)
|
|
req.Header.Set("Accept", "application/json")
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return "", &TransportError{Err: err}
|
|
}
|
|
defer resp.Body.Close()
|
|
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
|
|
if resp.StatusCode == http.StatusNotFound {
|
|
return "", ErrNoPBSSecret
|
|
}
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return "", &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)}
|
|
}
|
|
var out struct {
|
|
TokenSecret string `json:"token_secret"`
|
|
}
|
|
if err := json.Unmarshal(raw, &out); err != nil {
|
|
return "", fmt.Errorf("hub: decoding consume-token response (body withheld — secret channel)")
|
|
}
|
|
if out.TokenSecret == "" {
|
|
return "", fmt.Errorf("hub: consume-token returned an empty secret")
|
|
}
|
|
return out.TokenSecret, nil
|
|
}
|
|
|
|
// JobWire is one queued signed-op job as served by GET /hosts/{id}/jobs (slice 10A). The blob is
|
|
// OPAQUE to the hub — for slice 10B it is a base64 `SignedJobEnvelope` (op-blob + armored SSHSIG)
|
|
// the agent verifies before executing.
|
|
type JobWire struct {
|
|
JobID string `json:"job_id"`
|
|
BlobB64 string `json:"blob_b64"`
|
|
CreatedAt string `json:"created_at"`
|
|
}
|
|
|
|
// Jobs fetches this host's pending signed-op jobs (slice 10B). Self-scoped server-side (the
|
|
// per-host key only reads its own host). The agent verifies each before executing.
|
|
func (c *Client) Jobs(ctx context.Context) ([]JobWire, error) {
|
|
if c.hostID == "" {
|
|
return nil, fmt.Errorf("hub: Jobs requires a configured host_id")
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+"/api/v1/hosts/"+c.hostID+"/jobs", nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+c.apiKey)
|
|
req.Header.Set("Accept", "application/json")
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return nil, &TransportError{Err: err}
|
|
}
|
|
defer resp.Body.Close()
|
|
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return nil, &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)}
|
|
}
|
|
var out struct {
|
|
Jobs []JobWire `json:"jobs"`
|
|
}
|
|
if err := json.Unmarshal(raw, &out); err != nil {
|
|
return nil, fmt.Errorf("hub: decoding jobs: %w", err)
|
|
}
|
|
return out.Jobs, nil
|
|
}
|
|
|
|
// CompleteJob clears a processed job from the host's queue (slice 10B): DELETE
|
|
// /hosts/{id}/jobs/{job_id}, self-scoped. Called after a job is executed OR permanently rejected
|
|
// (the nonce is already durably burned on a passing verify, so re-processing is replay-safe).
|
|
func (c *Client) CompleteJob(ctx context.Context, jobID string) error {
|
|
if c.hostID == "" || jobID == "" {
|
|
return fmt.Errorf("hub: CompleteJob requires host_id + job_id")
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.baseURL+"/api/v1/hosts/"+c.hostID+"/jobs/"+jobID, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+c.apiKey)
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return &TransportError{Err: err}
|
|
}
|
|
defer resp.Body.Close()
|
|
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<10))
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func tail(b []byte, max int) string {
|
|
s := strings.TrimSpace(string(b))
|
|
if len(s) > max {
|
|
return s[:max] + "…"
|
|
}
|
|
return s
|
|
}
|
|
|
|
// IdentityEscrowResponse mirrors GET /api/v1/hosts/{host_id}/escrow (hub >= v0.94.0, R-199).
|
|
// Present=false is a CLEAN answer, not a fault: the host simply has no sealed bundle yet.
|
|
type IdentityEscrowResponse struct {
|
|
HostID string `json:"host_id"`
|
|
Present bool `json:"present"`
|
|
IdentityEscrowB64 string `json:"identity_escrow_b64"`
|
|
}
|
|
|
|
// FetchIdentityEscrow reads back THIS host's own opaque identity-escrow blob (R-199 link 6 — the
|
|
// mirror of UploadEscrow, self-scoped server-side by the per-host key). The bytes are ciphertext: they
|
|
// are useless without the customer's recovery code R, which neither the hub nor this agent ever holds.
|
|
//
|
|
// It is the ONLY retrieval this client performs, and it is deliberately narrow — no directive, no
|
|
// K-escrow, no key rotation. The operator-driven DR path (recovery-mode re-enroll) is a different
|
|
// endpoint with a different gate and is not reached from here.
|
|
//
|
|
// Errors are typed (transport vs HTTP) and never include the bearer token. The BLOB is never logged —
|
|
// only its length.
|
|
func (c *Client) FetchIdentityEscrow(ctx context.Context) (*IdentityEscrowResponse, error) {
|
|
if c.hostID == "" {
|
|
return nil, fmt.Errorf("hub: FetchIdentityEscrow requires a configured host_id")
|
|
}
|
|
url := c.baseURL + "/api/v1/hosts/" + c.hostID + "/escrow"
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("hub: building escrow-fetch request: %w", err)
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+c.apiKey)
|
|
req.Header.Set("Accept", "application/json")
|
|
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return nil, &TransportError{Err: err}
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return nil, &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)}
|
|
}
|
|
var out IdentityEscrowResponse
|
|
if err := json.Unmarshal(raw, &out); err != nil {
|
|
return nil, fmt.Errorf("hub: decoding escrow fetch: %w", err)
|
|
}
|
|
return &out, nil
|
|
}
|