hub v0.44.0: PBS DR tier SLICE 1 — felhom-tenantsync surface (script+client) + hub provisioning flow (consume-once host secret, pbs_dr desired-state descriptor, fail-closed + idempotent, re-issue)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
2026-07-10 20:49:48 +02:00
parent 00afadc1fe
commit ce6a56691e
19 changed files with 1743 additions and 0 deletions
@@ -0,0 +1,219 @@
+// 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)")
+
+// 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)
+}
+
+// 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
+}
+
+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"`
+	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
+		}
+		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
+}
@@ -0,0 +1,300 @@
+package tenantsync
+
+// The tenancy SSH client against an IN-PROCESS x/crypto/ssh server (the wgsync client_test
+// pattern). Beyond the transport round-trip, these pin the package's two load-bearing contracts:
+// the typed token_exists outcome, and the secret-hygiene rule that NO error message ever embeds
+// stdout bytes (stdout is the secret channel).
+
+import (
+	"bytes"
+	"context"
+	"encoding/pem"
+	"errors"
+	"io"
+	"log"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"crypto/ed25519"
+	"crypto/rand"
+
+	"golang.org/x/crypto/ssh"
+)
+
+func testKeys(t *testing.T) ([]byte, ssh.Signer) {
+	t.Helper()
+	_, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("ed25519: %v", err)
+	}
+	block, err := ssh.MarshalPrivateKey(priv, "")
+	if err != nil {
+		t.Fatalf("MarshalPrivateKey: %v", err)
+	}
+	signer, err := ssh.NewSignerFromKey(priv)
+	if err != nil {
+		t.Fatalf("NewSignerFromKey: %v", err)
+	}
+	return pem.EncodeToMemory(block), signer
+}
+
+type testServer struct {
+	addr     string
+	mu       sync.Mutex
+	captured []byte
+	cmd      string
+}
+
+func startTestServer(t *testing.T, hostSigner, clientSigner ssh.Signer,
+	stdoutResp, stderrResp string, exitStatus uint32) *testServer {
+	t.Helper()
+	cfg := &ssh.ServerConfig{
+		PublicKeyCallback: func(conn ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) {
+			if bytes.Equal(key.Marshal(), clientSigner.PublicKey().Marshal()) {
+				return &ssh.Permissions{}, nil
+			}
+			return nil, io.EOF
+		},
+	}
+	cfg.AddHostKey(hostSigner)
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	t.Cleanup(func() { ln.Close() })
+	srv := &testServer{addr: ln.Addr().String()}
+
+	go func() {
+		conn, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		sconn, chans, reqs, err := ssh.NewServerConn(conn, cfg)
+		if err != nil {
+			return
+		}
+		defer sconn.Close()
+		go ssh.DiscardRequests(reqs)
+		for newCh := range chans {
+			if newCh.ChannelType() != "session" {
+				newCh.Reject(ssh.UnknownChannelType, "unsupported")
+				continue
+			}
+			ch, chReqs, err := newCh.Accept()
+			if err != nil {
+				continue
+			}
+			go func() {
+				for req := range chReqs {
+					if req.Type == "exec" {
+						var p struct{ Command string }
+						ssh.Unmarshal(req.Payload, &p)
+						srv.mu.Lock()
+						srv.cmd = p.Command
+						srv.mu.Unlock()
+						req.Reply(true, nil)
+						data, _ := io.ReadAll(ch)
+						srv.mu.Lock()
+						srv.captured = data
+						srv.mu.Unlock()
+						if stderrResp != "" {
+							ch.Stderr().Write([]byte(stderrResp))
+						}
+						if stdoutResp != "" {
+							ch.Write([]byte(stdoutResp))
+						}
+						ch.SendRequest("exit-status", false, ssh.Marshal(struct{ Status uint32 }{exitStatus}))
+						ch.Close()
+						return
+					}
+					req.Reply(false, nil)
+				}
+			}()
+		}
+	}()
+	return srv
+}
+
+func hostKeyLine(t *testing.T, s ssh.Signer) string {
+	t.Helper()
+	return strings.TrimSpace(string(ssh.MarshalAuthorizedKey(s.PublicKey())))
+}
+
+func newTestClient(t *testing.T, addr, hostKey string, clientPEM []byte) *Client {
+	t.Helper()
+	c, err := New(Config{
+		Addr: addr, User: "felhom-peersync", PrivateKey: clientPEM,
+		HostKeyLine: hostKey, Timeout: 5 * time.Second,
+	}, log.New(io.Discard, "", 0))
+	if err != nil {
+		t.Fatalf("tenantsync.New: %v", err)
+	}
+	return c
+}
+
+const okResp = `{"status":"ok","token_id":"felhom@pbs!peti","token_secret":"s3cr3t-uuid-value",` +
+	`"fingerprint":"aa:bb","datastore":"felhom-offsite","namespace":"peti"}`
+
+func TestProvision_RoundTripParsesResult(t *testing.T) {
+	clientPEM, clientSigner := testKeys(t)
+	_, hostSigner := testKeys(t)
+	srv := startTestServer(t, hostSigner, clientSigner, okResp, "", 0)
+	c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM)
+
+	res, err := c.Provision(context.Background(), "peti")
+	if err != nil {
+		t.Fatalf("Provision: %v", err)
+	}
+	if res.TokenID != "felhom@pbs!peti" || res.TokenSecret != "s3cr3t-uuid-value" ||
+		res.Fingerprint != "aa:bb" || res.Datastore != "felhom-offsite" || res.Namespace != "peti" {
+		t.Errorf("Result mis-parsed: %+v", res)
+	}
+	srv.mu.Lock()
+	defer srv.mu.Unlock()
+	if want := `{"customer_id":"peti","op":"provision"}`; string(srv.captured) != want {
+		t.Errorf("server received %q, want %q", srv.captured, want)
+	}
+	if srv.cmd != "felhom-tenantsync" {
+		t.Errorf("exec command = %q, want felhom-tenantsync", srv.cmd)
+	}
+}
+
+func TestProvision_TokenExistsIsTypedError(t *testing.T) {
+	clientPEM, clientSigner := testKeys(t)
+	_, hostSigner := testKeys(t)
+	// The script emits the error JSON on stdout AND exits 1 — both must map to ErrTokenExists.
+	srv := startTestServer(t, hostSigner, clientSigner,
+		`{"status":"error","code":"token_exists","error":"token felhom@pbs!peti already exists"}`, "", 1)
+	c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM)
+
+	_, err := c.Provision(context.Background(), "peti")
+	if !errors.Is(err, ErrTokenExists) {
+		t.Fatalf("err = %v, want ErrTokenExists", err)
+	}
+}
+
+func TestReissue_SendsReissueOp(t *testing.T) {
+	clientPEM, clientSigner := testKeys(t)
+	_, hostSigner := testKeys(t)
+	srv := startTestServer(t, hostSigner, clientSigner, okResp, "", 0)
+	c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM)
+
+	if _, err := c.Reissue(context.Background(), "peti"); err != nil {
+		t.Fatalf("Reissue: %v", err)
+	}
+	srv.mu.Lock()
+	defer srv.mu.Unlock()
+	if want := `{"customer_id":"peti","op":"reissue"}`; string(srv.captured) != want {
+		t.Errorf("server received %q, want %q", srv.captured, want)
+	}
+}
+
+func TestFingerprint_Op(t *testing.T) {
+	clientPEM, clientSigner := testKeys(t)
+	_, hostSigner := testKeys(t)
+	srv := startTestServer(t, hostSigner, clientSigner, `{"status":"ok","fingerprint":"cc:dd"}`, "", 0)
+	c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM)
+
+	fp, err := c.Fingerprint(context.Background())
+	if err != nil {
+		t.Fatalf("Fingerprint: %v", err)
+	}
+	if fp != "cc:dd" {
+		t.Errorf("fingerprint = %q, want cc:dd", fp)
+	}
+}
+
+// TestErrors_NeverEmbedStdout is the secret-hygiene contract: stdout may carry the token secret,
+// so NO error path may quote it. A malformed-but-secret-bearing stdout must yield an error that
+// does not contain the marker bytes. (wgsync quotes stdout in its malformed error — this package
+// deliberately must not; that divergence is the point of this test.)
+func TestErrors_NeverEmbedStdout(t *testing.T) {
+	cases := []struct {
+		name       string
+		stdout     string
+		exitStatus uint32
+	}{
+		{"malformed-ok-exit", `garbage SECRET-MARKER garbage`, 0},
+		{"malformed-fail-exit", `partial {"token_secret":"SECRET-MARKER"`, 1},
+		{"wrong-status", `{"status":"weird","token_secret":"SECRET-MARKER"}`, 0},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			clientPEM, clientSigner := testKeys(t)
+			_, hostSigner := testKeys(t)
+			srv := startTestServer(t, hostSigner, clientSigner, tc.stdout, "some stderr", tc.exitStatus)
+			c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM)
+
+			_, err := c.Provision(context.Background(), "peti")
+			if err == nil {
+				t.Fatal("Provision accepted a malformed response")
+			}
+			if strings.Contains(err.Error(), "SECRET-MARKER") {
+				t.Errorf("error %q embeds stdout bytes — the secret channel leaked into logs", err)
+			}
+		})
+	}
+}
+
+func TestOkResponseMissingFieldsRefused(t *testing.T) {
+	clientPEM, clientSigner := testKeys(t)
+	_, hostSigner := testKeys(t)
+	srv := startTestServer(t, hostSigner, clientSigner,
+		`{"status":"ok","token_id":"felhom@pbs!peti"}`, "", 0) // no secret/ns/fp/ds
+	c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM)
+
+	if _, err := c.Provision(context.Background(), "peti"); err == nil {
+		t.Fatal("Provision accepted an ok-response with missing fields")
+	}
+}
+
+func TestWrongHostKeyRefused(t *testing.T) {
+	clientPEM, clientSigner := testKeys(t)
+	_, hostSigner := testKeys(t)
+	_, otherSigner := testKeys(t)
+	srv := startTestServer(t, hostSigner, clientSigner, okResp, "", 0)
+	c := newTestClient(t, srv.addr, hostKeyLine(t, otherSigner), clientPEM)
+
+	_, err := c.Provision(context.Background(), "peti")
+	if err == nil {
+		t.Fatal("Provision succeeded against a server with the WRONG host key — the pin is dead")
+	}
+	srv.mu.Lock()
+	defer srv.mu.Unlock()
+	if len(srv.captured) != 0 {
+		t.Errorf("payload leaked to a mis-keyed server: %q", srv.captured)
+	}
+}
+
+func TestInvalidCustomerIDRefusedClientSide(t *testing.T) {
+	clientPEM, _ := testKeys(t)
+	_, hostSigner := testKeys(t)
+	// No server needed — the refusal must happen before any dial.
+	c := newTestClient(t, "127.0.0.1:1", hostKeyLine(t, hostSigner), clientPEM)
+	for _, bad := range []string{"", "-leading-dash", ".dot", "has space", strings.Repeat("x", 40)} {
+		if _, err := c.Provision(context.Background(), bad); err == nil {
+			t.Errorf("customer_id %q accepted", bad)
+		}
+	}
+}
+
+func TestNew_BadInputsFailEarly(t *testing.T) {
+	clientPEM, _ := testKeys(t)
+	_, hostSigner := testKeys(t)
+	hk := hostKeyLine(t, hostSigner)
+	logger := log.New(io.Discard, "", 0)
+
+	if _, err := New(Config{Addr: "x:22", User: "u", PrivateKey: []byte("not-a-key"), HostKeyLine: hk}, logger); err == nil {
+		t.Error("bad private key accepted")
+	}
+	if _, err := New(Config{Addr: "x:22", User: "u", PrivateKey: clientPEM, HostKeyLine: "not a key line"}, logger); err == nil {
+		t.Error("bad host key line accepted")
+	}
+	if _, err := New(Config{User: "u", PrivateKey: clientPEM, HostKeyLine: hk}, logger); err == nil {
+		t.Error("missing addr accepted")
+	}
+}