302 lines
13 KiB
Go
302 lines
13 KiB
Go
// Package tenantsync drives the offsite endpoint's per-customer PBS tenancy surface (PBS DR tier
|
||
// SLICE 1) — the structural twin of internal/wgsync: SSH with a PINNED host key (exact-match or
|
||
// refuse, no fallback) to a forced-command script (`felhom-tenantsync` — its OWN key + sudoers
|
||
// line; the peersync surface is untouched). JSON on stdin, JSON on stdout, one op per session.
|
||
//
|
||
// SECRET HYGIENE (load-bearing divergence from wgsync): the script's stdout carries the one-time
|
||
// PBS token secret. It is parsed into Result and handed to the caller for the consume-once store
|
||
// write — it is NEVER logged, and error messages NEVER embed stdout bytes (stderr only). Do not
|
||
// "improve" the diagnostics by quoting the response.
|
||
package tenantsync
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"net"
|
||
"regexp"
|
||
"strings"
|
||
"time"
|
||
|
||
"golang.org/x/crypto/ssh"
|
||
)
|
||
|
||
// ErrTokenExists is the typed "provision refused: the token already exists" outcome — the hub
|
||
// treats it as a state mismatch (a descriptor should exist; re-issue is the explicit recovery).
|
||
var ErrTokenExists = errors.New("tenantsync: token already exists on the endpoint (re-issue is the explicit path)")
|
||
|
||
// ErrUsageUnsupported is the typed "endpoint script predates the usage op" outcome (v0.65.0): an
|
||
// endpoint STILL on tenantsync ≤ v1.1.0 answers `{"op":"usage"}` with its generic `bad_request
|
||
// "unknown op"`. It is the GRACEFUL-DEGRADATION signal — the PBS checker maps it to an "unavailable"
|
||
// snapshot state (an expected pre-update condition), NOT an error/degraded state, so no alert fires
|
||
// and the gauge honestly says "usage not available (endpoint update pending)". The moment ep0 gets
|
||
// v1.2.0, the next poll succeeds — no hub redeploy.
|
||
var ErrUsageUnsupported = errors.New("tenantsync: endpoint does not support the usage op (script update pending)")
|
||
|
||
// ErrSelfCheckFailed is the typed "the new token could not list its own namespace, so the endpoint rolled it back"
|
||
// outcome (code self_check_failed). R-1295: PBS's proxy caches permissions for up to 5 s and an ACL-only change does
|
||
// not clear that cache, so another client's request between the token create and its grant leaves the token refused
|
||
// (403) for up to ~5 s — measured 4.2–5.0 s on PBS 4.2.5 under load, 9 of 20 first checks failed. The rollback leaves
|
||
// no token, so a later attempt starts clean: the hub retries it after a pause (web/pbsdr.go).
|
||
var ErrSelfCheckFailed = errors.New("tenantsync: the new token failed its own-namespace self-check and was rolled back")
|
||
|
||
// customerIDRe mirrors the script's validation — refuse client-side before a wasted SSH round-trip.
|
||
var customerIDRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.-]{0,30}$`)
|
||
|
||
// Config configures the SSH client. Addr/User/HostKeyLine are typically the SAME values as the
|
||
// peersync client (same box, same low-priv user, same pinned host key); PrivateKey is tenantsync's
|
||
// OWN key (the authorized_keys line selects the forced command).
|
||
type Config struct {
|
||
Addr string // "host:22"
|
||
User string // "felhom-peersync"
|
||
PrivateKey []byte // PEM private key (from the mounted Secret file)
|
||
HostKeyLine string // single authorized_keys-format line of the endpoint's host pubkey
|
||
Timeout time.Duration // default 60s (tenancy ops run several PBS commands)
|
||
}
|
||
|
||
// Result is the script's ok-response: the descriptor fields + the ONE-TIME token secret.
|
||
// TokenSecret is transient custody — store it consume-once immediately, never log the struct.
|
||
type Result struct {
|
||
TokenID string `json:"token_id"`
|
||
TokenSecret string `json:"token_secret"`
|
||
Fingerprint string `json:"fingerprint"`
|
||
Datastore string `json:"datastore"`
|
||
Namespace string `json:"namespace"`
|
||
}
|
||
|
||
// Client is a pinned-host-key SSH per-op executor. Construct with New (parses keys up front).
|
||
type Client struct {
|
||
addr string
|
||
user string
|
||
signer ssh.Signer
|
||
hostKey ssh.PublicKey
|
||
timeout time.Duration
|
||
logger *log.Logger
|
||
}
|
||
|
||
// New builds a Client, failing early on an unparsable private key or host-key line.
|
||
func New(cfg Config, logger *log.Logger) (*Client, error) {
|
||
if cfg.Addr == "" || cfg.User == "" {
|
||
return nil, fmt.Errorf("tenantsync: Addr and User are required")
|
||
}
|
||
signer, err := ssh.ParsePrivateKey(cfg.PrivateKey)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("tenantsync: parse private key: %w", err)
|
||
}
|
||
hostKey, _, _, _, err := ssh.ParseAuthorizedKey([]byte(cfg.HostKeyLine))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("tenantsync: parse host key line: %w", err)
|
||
}
|
||
timeout := cfg.Timeout
|
||
if timeout == 0 {
|
||
timeout = 60 * time.Second
|
||
}
|
||
if logger == nil {
|
||
logger = log.Default()
|
||
}
|
||
return &Client{
|
||
addr: cfg.Addr, user: cfg.User, signer: signer,
|
||
hostKey: hostKey, timeout: timeout, logger: logger,
|
||
}, nil
|
||
}
|
||
|
||
// Provision creates the customer's namespace + privilege-separated token on the endpoint.
|
||
// An already-existing token is the typed ErrTokenExists (never silently re-keyed).
|
||
func (c *Client) Provision(ctx context.Context, customerID string) (*Result, error) {
|
||
return c.tenancyOp(ctx, "provision", customerID)
|
||
}
|
||
|
||
// Reissue explicitly re-keys the customer's token (delete + recreate + re-grant, script-side).
|
||
func (c *Client) Reissue(ctx context.Context, customerID string) (*Result, error) {
|
||
return c.tenancyOp(ctx, "reissue", customerID)
|
||
}
|
||
|
||
// Deprovision DESTROYS the customer's PBS namespace, all its backup groups, and its token — the
|
||
// customer-RESET teardown (v0.61.0, operator ack-gated). The shared felhom@pbs user is never touched
|
||
// (co-tenants ride it). Idempotent: a missing tenant is a clean success (existed=false). This carries
|
||
// NO secret, so it does not route through tenancyOp's token-field validation.
|
||
func (c *Client) Deprovision(ctx context.Context, customerID string) (existed bool, err error) {
|
||
if !customerIDRe.MatchString(customerID) {
|
||
return false, fmt.Errorf("tenantsync: invalid customer_id %q", customerID)
|
||
}
|
||
payload, err := json.Marshal(map[string]string{"op": "deprovision", "customer_id": customerID})
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
stdout, stderr, runErr := c.exec(ctx, payload)
|
||
resp, err := parseResponse(stdout, stderr, runErr)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
if resp.Namespace == "" {
|
||
return false, fmt.Errorf("tenantsync: deprovision response missing namespace")
|
||
}
|
||
c.logger.Printf("[INFO] tenantsync: deprovision ok for %s (ns=%s, existed=%t)",
|
||
customerID, resp.Namespace, resp.Deleted)
|
||
return resp.Deleted, nil
|
||
}
|
||
|
||
// Fingerprint returns the endpoint PBS's API cert fingerprint (the descriptor field).
|
||
func (c *Client) Fingerprint(ctx context.Context) (string, error) {
|
||
stdout, stderr, runErr := c.exec(ctx, []byte(`{"op":"fingerprint"}`))
|
||
resp, err := parseResponse(stdout, stderr, runErr)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if resp.Fingerprint == "" {
|
||
return "", fmt.Errorf("tenantsync: fingerprint op returned an empty fingerprint")
|
||
}
|
||
return resp.Fingerprint, nil
|
||
}
|
||
|
||
// BoxUsage is the endpoint PBS datastore's raw fill (bytes), from the read-only usage op (v1.2.0).
|
||
type BoxUsage struct {
|
||
Total int64
|
||
Used int64
|
||
Avail int64
|
||
// R-1165 (tenantsync v1.3.0): the top-level namespaces no prune job covers. UnprunedChecked=false when the
|
||
// endpoint could not read its job list, or predates v1.3.0 — then Unpruned says nothing.
|
||
Unpruned []string
|
||
UnprunedChecked bool
|
||
}
|
||
|
||
// Usage returns the felhom-offsite datastore's total/used/avail from the endpoint — a read-only op
|
||
// (no token, no namespace, no mutation), the structural twin of Fingerprint. Against an endpoint
|
||
// still on tenantsync ≤ v1.1.0 the op is unknown → ErrUsageUnsupported (graceful degradation).
|
||
func (c *Client) Usage(ctx context.Context) (BoxUsage, error) {
|
||
stdout, stderr, runErr := c.exec(ctx, []byte(`{"op":"usage"}`))
|
||
resp, err := parseResponse(stdout, stderr, runErr)
|
||
if err != nil {
|
||
return BoxUsage{}, err
|
||
}
|
||
if resp.Total <= 0 {
|
||
return BoxUsage{}, fmt.Errorf("tenantsync: usage op returned a non-positive total")
|
||
}
|
||
return BoxUsage{Total: resp.Total, Used: resp.Used, Avail: resp.Avail, Unpruned: resp.Unpruned, UnprunedChecked: resp.UnprunedChecked}, nil
|
||
}
|
||
|
||
func (c *Client) tenancyOp(ctx context.Context, op, customerID string) (*Result, error) {
|
||
if !customerIDRe.MatchString(customerID) {
|
||
return nil, fmt.Errorf("tenantsync: invalid customer_id %q", customerID)
|
||
}
|
||
payload, err := json.Marshal(map[string]string{"op": op, "customer_id": customerID})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
stdout, stderr, runErr := c.exec(ctx, payload)
|
||
resp, err := parseResponse(stdout, stderr, runErr)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if resp.TokenSecret == "" || resp.TokenID == "" || resp.Namespace == "" || resp.Fingerprint == "" || resp.Datastore == "" {
|
||
// Field NAMES only — never values (TokenSecret).
|
||
return nil, fmt.Errorf("tenantsync: %s response is missing required fields", op)
|
||
}
|
||
c.logger.Printf("[INFO] tenantsync: %s ok for %s (ns=%s, token_id=%s; secret withheld from logs)",
|
||
op, customerID, resp.Namespace, resp.TokenID)
|
||
return &resp.Result, nil
|
||
}
|
||
|
||
// response is the script's stdout contract — ok carries the Result fields, error carries code+error.
|
||
type response struct {
|
||
Status string `json:"status"`
|
||
Code string `json:"code"`
|
||
Error string `json:"error"`
|
||
Deleted bool `json:"deleted"` // deprovision op: whether the namespace existed (was destroyed)
|
||
// usage op (v1.2.0): the felhom-offsite datastore's raw fill, in BYTES.
|
||
Total int64 `json:"total"`
|
||
Used int64 `json:"used"`
|
||
Avail int64 `json:"avail"`
|
||
// usage op (v1.3.0, R-1165): namespaces with no prune job, and whether that list was read.
|
||
Unpruned []string `json:"unpruned"`
|
||
UnprunedChecked bool `json:"unpruned_checked"`
|
||
Result
|
||
}
|
||
|
||
// parseResponse turns (stdout, stderr, runErr) into a typed outcome. The script emits its error
|
||
// JSON on stdout and exits 1, so a run error is parsed for the typed code FIRST; only when stdout
|
||
// carries no usable JSON does the raw failure (with stderr, NEVER stdout) surface.
|
||
func parseResponse(stdout, stderr []byte, runErr error) (*response, error) {
|
||
var resp response
|
||
parseOK := json.Unmarshal(bytes.TrimSpace(stdout), &resp) == nil
|
||
if parseOK && resp.Status == "error" {
|
||
if resp.Code == "token_exists" {
|
||
return nil, ErrTokenExists
|
||
}
|
||
if resp.Code == "self_check_failed" {
|
||
return nil, fmt.Errorf("%w: %s", ErrSelfCheckFailed, resp.Error)
|
||
}
|
||
// An endpoint ≤ v1.1.0 has no usage op → its generic `bad_request "unknown op"`. Map it to the
|
||
// typed graceful-degradation signal (the exact err_json string is the script's fixed contract).
|
||
if resp.Code == "bad_request" && strings.Contains(resp.Error, "unknown op") {
|
||
return nil, ErrUsageUnsupported
|
||
}
|
||
return nil, fmt.Errorf("tenantsync: endpoint refused: %s (code %s)", resp.Error, resp.Code)
|
||
}
|
||
if runErr != nil {
|
||
return nil, fmt.Errorf("tenantsync: remote op failed: %w (stderr: %s)",
|
||
runErr, strings.TrimSpace(string(stderr)))
|
||
}
|
||
if !parseOK || resp.Status != "ok" {
|
||
// stdout may carry the secret — report shape only, never bytes.
|
||
return nil, fmt.Errorf("tenantsync: malformed endpoint response (%d stdout bytes; stderr: %s)",
|
||
len(bytes.TrimSpace(stdout)), strings.TrimSpace(string(stderr)))
|
||
}
|
||
return &resp, nil
|
||
}
|
||
|
||
// exec runs one forced-command session: payload on stdin, returns stdout/stderr. The connection
|
||
// discipline (pinned key, constrained HostKeyAlgorithms, deadline both sides of the handshake)
|
||
// is wgsync.Push's, verbatim — that shape is live-proven against the real sshd.
|
||
func (c *Client) exec(ctx context.Context, payload []byte) (stdout, stderr []byte, err error) {
|
||
sshCfg := &ssh.ClientConfig{
|
||
User: c.user,
|
||
Auth: []ssh.AuthMethod{ssh.PublicKeys(c.signer)},
|
||
HostKeyCallback: ssh.FixedHostKey(c.hostKey),
|
||
// Constrain negotiation to the PINNED key's algorithm — a multi-hostkey sshd (stock:
|
||
// ECDSA + ed25519) otherwise presents a different type and FixedHostKey refuses a
|
||
// legitimate server (wgsync S1 live finding).
|
||
HostKeyAlgorithms: []string{c.hostKey.Type()},
|
||
Timeout: c.timeout,
|
||
}
|
||
dialer := net.Dialer{Timeout: c.timeout}
|
||
conn, err := dialer.DialContext(ctx, "tcp", c.addr)
|
||
if err != nil {
|
||
return nil, nil, fmt.Errorf("tenantsync: dial %s: %w", c.addr, err)
|
||
}
|
||
if dl, ok := ctx.Deadline(); ok {
|
||
conn.SetDeadline(dl)
|
||
} else {
|
||
conn.SetDeadline(time.Now().Add(c.timeout))
|
||
}
|
||
sconn, chans, reqs, err := ssh.NewClientConn(conn, c.addr, sshCfg)
|
||
if err != nil {
|
||
conn.Close()
|
||
return nil, nil, fmt.Errorf("tenantsync: ssh handshake %s: %w", c.addr, err)
|
||
}
|
||
client := ssh.NewClient(sconn, chans, reqs)
|
||
defer client.Close()
|
||
conn.SetDeadline(time.Time{})
|
||
if dl, ok := ctx.Deadline(); ok {
|
||
conn.SetDeadline(dl)
|
||
}
|
||
|
||
session, err := client.NewSession()
|
||
if err != nil {
|
||
return nil, nil, fmt.Errorf("tenantsync: session: %w", err)
|
||
}
|
||
defer session.Close()
|
||
|
||
var outBuf, errBuf bytes.Buffer
|
||
session.Stdin = bytes.NewReader(payload)
|
||
session.Stdout = &outBuf
|
||
session.Stderr = &errBuf
|
||
|
||
// The forced command overrides this string; it documents intent on the wire.
|
||
runErr := session.Run("felhom-tenantsync")
|
||
return outBuf.Bytes(), errBuf.Bytes(), runErr
|
||
}
|