felhom-agent/internal/hub/client.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
}