From 64914d8e7d6cb82ffe8c8234ac0387d7d1a641f2 Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Tue, 3 Nov 2026 16:54:06 +0100 Subject: [PATCH] feat(openbao): preflight one server replacement from live evidence Add a dispatch-only, protected Replace OpenBao Server workflow whose only mode, preflight, captures each expected server's health through its own port-forward and one authenticated Raft configuration read, then runs the health and quorum judges for the named server. It changes nothing. The Raft read uses a new workload-less openbao-replacement-observer service account and Kubernetes-auth role whose only policy reads the Raft configuration and revokes its own token. Fixes #5836 Part of #4970 Co-Authored-By: Claude Opus 5.5 --- .github/workflows/replace-openbao-server.yaml | 153 ++++++++ .../infrastructure/vault-config/job.yaml | 25 ++ .../vault-config/kustomization.yaml | 1 + .../service-account-replacement-observer.yaml | 11 + scripts/capture-openbao-replacement/README.md | 44 +++ scripts/capture-openbao-replacement/main.go | 328 +++++++++++++++++ .../capture-openbao-replacement/main_test.go | 334 ++++++++++++++++++ .../workflow_test.go | 113 ++++++ 8 files changed, 1009 insertions(+) create mode 100644 .github/workflows/replace-openbao-server.yaml create mode 100644 k8s/bases/infrastructure/vault-config/service-account-replacement-observer.yaml create mode 100644 scripts/capture-openbao-replacement/README.md create mode 100644 scripts/capture-openbao-replacement/main.go create mode 100644 scripts/capture-openbao-replacement/main_test.go create mode 100644 scripts/capture-openbao-replacement/workflow_test.go diff --git a/.github/workflows/replace-openbao-server.yaml b/.github/workflows/replace-openbao-server.yaml new file mode 100644 index 0000000000..85a58ca755 --- /dev/null +++ b/.github/workflows/replace-openbao-server.yaml @@ -0,0 +1,153 @@ +name: Replace OpenBao Server + +# Read-only first slice of the single-server replacement procedure (#4970, +# #5836): it judges whether replacing the named server is safe right now and +# changes nothing. Only `preflight` exists until the replacement step lands. + +on: + workflow_dispatch: + inputs: + mode: + description: Judge whether the named server can be replaced now, without changing anything + required: true + type: choice + default: preflight + options: [preflight] + server: + description: The one server a later replacement would take out + required: true + type: choice + options: [openbao-0, openbao-1, openbao-2] + +permissions: {} + +concurrency: + group: prod-deploy + cancel-in-progress: false + queue: max + +jobs: + preflight: + name: Judge OpenBao health and quorum before a replacement + runs-on: ubuntu-latest + timeout-minutes: 15 + # Keep prod secrets, reviewers, wait timers and branch policy without + # recording a deployment: this run deploys nothing. + environment: + name: prod + deployment: false + permissions: + contents: read # checkout reviewed main and verify it is still current + env: + GH_TOKEN: ${{ github.token }} + REPLACEMENT_MODE: ${{ inputs.mode }} + REPLACEMENT_SERVER: ${{ inputs.server }} + steps: + - name: Require a first main dispatch before credentials + shell: bash + run: | + set -euo pipefail + [[ "$GITHUB_REF" == refs/heads/main && "$GITHUB_RUN_ATTEMPT" == 1 ]] + [[ "$REPLACEMENT_MODE" == preflight ]] + [[ "$REPLACEMENT_SERVER" =~ ^openbao-[0-9]$ ]] + + - name: Checkout the exact reviewed dispatch commit + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.sha }} + persist-credentials: false + + - name: Require the validated checkout to remain current main + run: ./scripts/verify-prod-recovery-source.sh "$GITHUB_SHA" + + - name: Setup Go + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + cache: false + + - name: Require the collector and both judges to pass offline + run: go test -race -count=1 ./scripts/capture-openbao-replacement ./scripts/judge-openbao-health ./scripts/judge-openbao-quorum ./internal/strictjson + + # The judges' expectations come from the reviewed render, never from what + # the cluster reports: the prod replica count and the chart's release. + - name: Derive the expected servers and release from the reviewed render + shell: bash + run: | + set -euo pipefail + replicas="$(yq -er '.data.openbao_replicas' k8s/clusters/prod/bootstrap/config-map.yaml)" + [[ "$replicas" =~ ^[3-9]$ ]] + servers="$(seq -f 'openbao-%g' 0 $((replicas - 1)) | paste -sd, -)" + [[ ",$servers," == *",$REPLACEMENT_SERVER,"* ]] + release=k8s/bases/infrastructure/controllers/openbao/helm-release.yaml + repository=k8s/bases/infrastructure/controllers/openbao/helm-repository.yaml + chart="$(yq -er '.spec.chart.spec.version' "$release")" + url="$(yq -er '.spec.url' "$repository")" + [[ "$url" == https://openbao.github.io/openbao-helm ]] + version="$(helm show chart openbao --repo "$url" --version "$chart" | yq -er '.appVersion')" + version="${version#v}" + [[ "$version" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]] + { + echo "OPENBAO_SERVERS=$servers" + echo "OPENBAO_VERSION=$version" + } >> "$GITHUB_ENV" + echo "Expecting $replicas servers on release $version" + + # Built before the capture so compile time cannot age the evidence the + # judges measure against --max-age. + - name: Build the collector and judges + shell: bash + run: | + set -euo pipefail + mkdir -p "$RUNNER_TEMP/bin" + for tool in capture-openbao-replacement judge-openbao-health judge-openbao-quorum; do + go build -o "$RUNNER_TEMP/bin/$tool" "./scripts/$tool" + done + + - name: Restore the established protected production kubeconfig + shell: bash + env: + KUBE_CONFIG: ${{ secrets.KUBE_CONFIG }} + run: | + set -euo pipefail + test -n "$KUBE_CONFIG" + umask 077 + mkdir -p ~/.kube + printf '%s' "$KUBE_CONFIG" > ~/.kube/config + + - name: Select the stable production API endpoint + shell: bash + env: + HCLOUD_TOKEN: ${{ secrets.HCLOUD_TOKEN }} + run: ./scripts/use-prod-stable-api-endpoint.sh >/dev/null + + # Reads pods and the two OpenBao endpoints only. The captures stay in a + # private directory; only the collector's closed status line and the + # judges' verdict lines, which carry no body, token or address, are logged. + - name: Capture and judge health and quorum for the named server + shell: bash + run: | + set -euo pipefail + private="$(mktemp -d)" + trap 'rm -rf "$private"' EXIT + bin="$RUNNER_TEMP/bin" + "$bin/capture-openbao-replacement" --expect-servers "$OPENBAO_SERVERS" --out "$private" + status=0 + health=0 + "$bin/judge-openbao-health" --input "$private/health.json" \ + --expect-servers "$OPENBAO_SERVERS" --expect-version "$OPENBAO_VERSION" --max-age 5m || health=$? + quorum=0 + "$bin/judge-openbao-quorum" --input "$private/quorum.json" \ + --expect-servers "$OPENBAO_SERVERS" --replace "$REPLACEMENT_SERVER" --max-age 5m || quorum=$? + # UNKNOWN (2) outranks HOLD (1), which outranks PROVEN (0). + for code in "$health" "$quorum"; do + if (( code > status )); then status=$code; fi + done + case "$status" in + 0) verdict=PROVEN ;; + 1) verdict=HOLD ;; + *) verdict=UNKNOWN ;; + esac + echo "OPENBAO_REPLACEMENT_PREFLIGHT=$verdict server=$REPLACEMENT_SERVER" + echo "OpenBao replacement preflight for \`$REPLACEMENT_SERVER\`: **$verdict**" >> "$GITHUB_STEP_SUMMARY" + exit "$status" diff --git a/k8s/bases/infrastructure/vault-config/job.yaml b/k8s/bases/infrastructure/vault-config/job.yaml index 55dc07cc80..20558ff2ae 100644 --- a/k8s/bases/infrastructure/vault-config/job.yaml +++ b/k8s/bases/infrastructure/vault-config/job.yaml @@ -985,6 +985,18 @@ spec: } POLICY + # Read-only replacement preflight (#5836): the Raft membership the + # quorum judge decides on, and revoking its own token when done. + # Health needs no token, so it is not granted here. + bao policy write openbao-replacement-observer - <<'POLICY' + path "sys/storage/raft/configuration" { + capabilities = ["read"] + } + path "auth/token/revoke-self" { + capabilities = ["update"] + } + POLICY + bao policy write vault-snapshot-readonly - <<'POLICY' path "sys/health" { capabilities = ["read"] @@ -1266,6 +1278,19 @@ spec: policies=vault-snapshot-readonly \ ttl=15m + # Minted per run by the protected replacement preflight (#5836); + # no default policy, and a token that outlives a run expires fast. + bao write auth/kubernetes/role/openbao-replacement-observer - <<'REPLACEMENT_OBSERVER_ROLE' + { + "bound_service_account_names": ["openbao-replacement-observer"], + "bound_service_account_namespaces": ["openbao"], + "token_policies": ["openbao-replacement-observer"], + "token_no_default_policy": true, + "token_ttl": 300, + "token_max_ttl": 300 + } + REPLACEMENT_OBSERVER_ROLE + # Weekly umami blue/green credential rotation (#4107). bao write auth/kubernetes/role/umami-db-rotator \ bound_service_account_names=umami-db-rotator \ diff --git a/k8s/bases/infrastructure/vault-config/kustomization.yaml b/k8s/bases/infrastructure/vault-config/kustomization.yaml index 6e355d2b38..4631e1f31d 100644 --- a/k8s/bases/infrastructure/vault-config/kustomization.yaml +++ b/k8s/bases/infrastructure/vault-config/kustomization.yaml @@ -3,6 +3,7 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - service-account.yaml + - service-account-replacement-observer.yaml - role.yaml - role-binding.yaml - job.yaml diff --git a/k8s/bases/infrastructure/vault-config/service-account-replacement-observer.yaml b/k8s/bases/infrastructure/vault-config/service-account-replacement-observer.yaml new file mode 100644 index 0000000000..906bbb838c --- /dev/null +++ b/k8s/bases/infrastructure/vault-config/service-account-replacement-observer.yaml @@ -0,0 +1,11 @@ +# Identity of the read-only OpenBao replacement preflight (#5836). No workload +# runs as it: the protected `Replace OpenBao Server` workflow mints one +# short-lived token for it per run. OpenBao's `openbao-replacement-observer` +# Kubernetes-auth role (vault-config) binds to exactly this SA and grants only +# the Raft configuration read and revoking its own token. +apiVersion: v1 +kind: ServiceAccount +metadata: + name: openbao-replacement-observer + namespace: openbao +automountServiceAccountToken: false diff --git a/scripts/capture-openbao-replacement/README.md b/scripts/capture-openbao-replacement/README.md new file mode 100644 index 0000000000..7331e3d857 --- /dev/null +++ b/scripts/capture-openbao-replacement/README.md @@ -0,0 +1,44 @@ +# capture-openbao-replacement + +Captures, from production, the two observations the single-server OpenBao replacement procedure +(#4970) judges before it replaces anything. It changes nothing and decides nothing: the +[health judge](../judge-openbao-health/README.md) and the +[quorum judge](../judge-openbao-quorum/README.md) decide from what it writes. + +```sh +capture-openbao-replacement \ + --expect-servers openbao-0,openbao-1,openbao-2 \ + --out +``` + +The protected [`Replace OpenBao Server`](../../.github/workflows/replace-openbao-server.yaml) +workflow runs it in `preflight` mode and then runs both judges for the server it was asked about. +The expected servers and release come from the reviewed render, never from the cluster. + +| Exit | Line | Meaning | +|---|---|---| +| 0 | `OPENBAO_REPLACEMENT_EVIDENCE=CAPTURED` | `health.json` and `quorum.json` are in `--out`, readable only by the owner. | +| 2 | `OPENBAO_REPLACEMENT_EVIDENCE=UNVERIFIED stage=` | A read failed, so nothing was written. Invalid arguments also exit 2. | + +The stage is one of `pod-identity`, `server-health`, `identity-token`, `observer-login`, +`raft-configuration`, `token-revoke` or `private-capture`. There is no HOLD: a sealed server or +a non-200 Raft read is recorded as it was observed, and the judges turn it into HOLD or UNKNOWN. + +## What it reads + +- **Health:** each expected server through a port-forward to that pod alone, so a body is always + filed under the server that produced it. Each read is bound to the pod UID seen before the + first read; a UID that differs afterwards fails the capture, so a pod replaced mid-capture + cannot lend its evidence to its successor. +- **Raft configuration:** one authenticated read through `service/openbao-active`. The command + mints a ten-minute token for the workload-less `openbao-replacement-observer` service account, + logs in to the Kubernetes-auth role of the same name, reads once and revokes the OpenBao token. + A token it cannot revoke fails the capture. The role carries no default policy, its token lives + five minutes at most, and its only policy reads `sys/storage/raft/configuration` and revokes + itself (`k8s/bases/infrastructure/vault-config/job.yaml`). + +It prints its one status line and never a token, address or response body. + +Run the regressions with `go test ./scripts/capture-openbao-replacement`. They stand in for the +cluster and the servers, run the real judges on the captures, and pin the workflow's protection +and the kubectl verbs the command may use. diff --git a/scripts/capture-openbao-replacement/main.go b/scripts/capture-openbao-replacement/main.go new file mode 100644 index 0000000000..8a79eb804e --- /dev/null +++ b/scripts/capture-openbao-replacement/main.go @@ -0,0 +1,328 @@ +// Command capture-openbao-replacement captures, from production, the two +// observations the single-server OpenBao replacement procedure (#4970, #5836) +// judges before it replaces anything: each expected server's health read +// through its own port-forward, and one authenticated Raft configuration read. +// +// It changes nothing. It reads pods, mints one short-lived token for the +// workload-less openbao-replacement-observer service account, logs in to the +// Kubernetes-auth role that can read only the Raft configuration, reads it once +// and revokes the OpenBao token. It writes health.json and quorum.json into +// --out for judge-openbao-health and judge-openbao-quorum, which decide; this +// command decides nothing about health or quorum. +// +// Every server's read is bound to the pod UID seen before and after it, so a +// pod replaced mid-capture cannot lend its evidence to its successor. It prints +// one closed status line and never a token, address or response body. +// +// Exit codes: 0 CAPTURED, 2 UNVERIFIED (a read failed or the arguments are +// invalid). There is no HOLD: holding is the judges' verdict. +package main + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "regexp" + "strings" + "time" +) + +const ( + namespace = "openbao" + kubeContext = "admin@prod" + observerAccount = "openbao-replacement-observer" + activeService = "service/openbao-active" + // maxBodyBytes matches the judges' own bound on one observation document. + maxBodyBytes = 1 << 20 + // The token API refuses durations under ten minutes; the OpenBao token + // the login returns is bounded separately by the role's TTL. + tokenDuration = "10m" + + usage = "usage: capture-openbao-replacement --expect-servers openbao-0,openbao-1,openbao-2 --out " +) + +var ( + serverName = regexp.MustCompile(`^openbao-[0-9]{1,2}$`) + forwarding = regexp.MustCompile(`^Forwarding from 127\.0\.0\.1:([0-9]{1,5}) -> 8200$`) +) + +// stage names the first read that failed; only these words reach the log. +type stage string + +func (s stage) Error() string { return string(s) } + +const ( + stagePods stage = "pod-identity" + stageHealth stage = "server-health" + stageIdentity stage = "identity-token" + stageLogin stage = "observer-login" + stageRaft stage = "raft-configuration" + stageRevoke stage = "token-revoke" + stageWrite stage = "private-capture" +) + +// cluster is everything the command asks of production, so the regressions +// can stand in for kubectl and the servers behind each port-forward. +type cluster interface { + podUID(ctx context.Context, name string) (string, error) + createToken(ctx context.Context) (string, error) + // forward returns the base URL of a local port-forward to target and a + // function that ends it. + forward(ctx context.Context, target string) (string, func(), error) + client() *http.Client +} + +type healthObservation struct { + Name string `json:"name"` + PodUID string `json:"podUID"` + HealthStatus int `json:"healthStatus"` + Health json.RawMessage `json:"health"` +} + +type healthCapture struct { + StartedAt string `json:"startedAt"` + CapturedAt string `json:"capturedAt"` + Servers []healthObservation `json:"servers"` +} + +type quorumCapture struct { + StartedAt string `json:"startedAt"` + CapturedAt string `json:"capturedAt"` + Status int `json:"status"` + Configuration json.RawMessage `json:"configuration"` +} + +func stamp(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) } + +// request performs one bounded read and returns its status and body. A body +// over the bound is a failed read, never a truncated one. +func request(ctx context.Context, c *http.Client, method, url, token string, body []byte) (int, []byte, error) { + var reader io.Reader + if body != nil { + reader = bytes.NewReader(body) + } + req, err := http.NewRequestWithContext(ctx, method, url, reader) + if err != nil { + return 0, nil, err + } + if token != "" { + req.Header.Set("X-Vault-Token", token) + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + res, err := c.Do(req) + if err != nil { + return 0, nil, err + } + defer func() { _ = res.Body.Close() }() + b, err := io.ReadAll(io.LimitReader(res.Body, maxBodyBytes+1)) + if err != nil || len(b) > maxBodyBytes { + return 0, nil, errors.New("unreadable response") + } + return res.StatusCode, b, nil +} + +// captureHealth reads every server's health through a port-forward to that +// pod alone, so a body always comes from the server it is filed under. +func captureHealth(ctx context.Context, k cluster, servers []string, now func() time.Time) (healthCapture, error) { + before := make(map[string]string, len(servers)) + for _, name := range servers { + uid, err := k.podUID(ctx, name) + if err != nil || uid == "" { + return healthCapture{}, stagePods + } + before[name] = uid + } + capture := healthCapture{StartedAt: stamp(now())} + for _, name := range servers { + base, stop, err := k.forward(ctx, "pod/"+name) + if err != nil { + return healthCapture{}, stageHealth + } + status, body, err := request(ctx, k.client(), http.MethodGet, base+"/v1/sys/health?standbyok=true", "", nil) + stop() + if err != nil || !json.Valid(body) { + return healthCapture{}, stageHealth + } + capture.Servers = append(capture.Servers, healthObservation{Name: name, PodUID: before[name], HealthStatus: status, Health: body}) + } + capture.CapturedAt = stamp(now()) + for _, name := range servers { + if uid, err := k.podUID(ctx, name); err != nil || uid != before[name] { + return healthCapture{}, stagePods + } + } + return capture, nil +} + +// captureQuorum logs the observer in through the active server, reads the Raft +// configuration once and revokes the token before reporting anything. +func captureQuorum(ctx context.Context, k cluster, now func() time.Time) (capture quorumCapture, err error) { + jwt, err := k.createToken(ctx) + if err != nil || strings.TrimSpace(jwt) == "" { + return quorumCapture{}, stageIdentity + } + base, stop, err := k.forward(ctx, activeService) + if err != nil { + return quorumCapture{}, stageLogin + } + defer stop() + login, _ := json.Marshal(map[string]string{"role": observerAccount, "jwt": strings.TrimSpace(jwt)}) + status, body, err := request(ctx, k.client(), http.MethodPost, base+"/v1/auth/kubernetes/login", "", login) + var auth struct { + Auth struct { + ClientToken string `json:"client_token"` + } `json:"auth"` + } + if err != nil || status != http.StatusOK || json.Unmarshal(body, &auth) != nil || auth.Auth.ClientToken == "" { + return quorumCapture{}, stageLogin + } + token := auth.Auth.ClientToken + defer func() { + // A capture whose token outlives it is not reported as captured. + status, _, revokeErr := request(context.WithoutCancel(ctx), k.client(), http.MethodPost, base+"/v1/auth/token/revoke-self", token, []byte("{}")) + if err == nil && (revokeErr != nil || status != http.StatusNoContent) { + capture, err = quorumCapture{}, stageRevoke + } + }() + capture.StartedAt = stamp(now()) + status, body, err = request(ctx, k.client(), http.MethodGet, base+"/v1/sys/storage/raft/configuration", token, nil) + if err != nil || !json.Valid(body) { + return quorumCapture{}, stageRaft + } + capture.CapturedAt = stamp(now()) + capture.Status, capture.Configuration = status, body + return capture, nil +} + +func writePrivate(dir, name string, value any) error { + b, err := json.Marshal(value) + if err != nil { + return err + } + return os.WriteFile(filepath.Join(dir, name), b, 0o600) +} + +func parseServers(list string) ([]string, bool) { + servers := strings.Split(list, ",") + seen := map[string]bool{} + for _, s := range servers { + if !serverName.MatchString(s) || seen[s] { + return nil, false + } + seen[s] = true + } + return servers, len(servers) >= 3 +} + +func run(ctx context.Context, args []string, stdout, stderr io.Writer, k cluster, now func() time.Time) int { + flags := flag.NewFlagSet("capture-openbao-replacement", flag.ContinueOnError) + flags.SetOutput(io.Discard) + list := flags.String("expect-servers", "", "comma-separated server names from the reviewed render") + out := flags.String("out", "", "existing private directory for health.json and quorum.json") + if flags.Parse(args) != nil || flags.NArg() != 0 { + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } + servers, ok := parseServers(*list) + info, err := os.Stat(*out) + if !ok || *out == "" || err != nil || !info.IsDir() || info.Mode().Perm()&0o077 != 0 { + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } + fail := func(s error) int { + _, _ = fmt.Fprintf(stdout, "OPENBAO_REPLACEMENT_EVIDENCE=UNVERIFIED stage=%s\n", s) + return 2 + } + health, err := captureHealth(ctx, k, servers, now) + if err != nil { + return fail(err) + } + quorum, err := captureQuorum(ctx, k, now) + if err != nil { + return fail(err) + } + if writePrivate(*out, "health.json", health) != nil || writePrivate(*out, "quorum.json", quorum) != nil { + return fail(stageWrite) + } + _, _ = fmt.Fprintln(stdout, "OPENBAO_REPLACEMENT_EVIDENCE=CAPTURED") + return 0 +} + +// kubectl is the production cluster, reached with the protected kubeconfig. +type kubectl struct{ http *http.Client } + +func (kubectl) args(rest ...string) []string { + return append([]string{"--context=" + kubeContext, "--request-timeout=20s", "--namespace=" + namespace}, rest...) +} + +func (k kubectl) output(ctx context.Context, rest ...string) ([]byte, error) { + var out bytes.Buffer + cmd := exec.CommandContext(ctx, "kubectl", k.args(rest...)...) + cmd.Stdout, cmd.Stderr = &out, io.Discard + err := cmd.Run() + return out.Bytes(), err +} + +func (k kubectl) podUID(ctx context.Context, name string) (string, error) { + b, err := k.output(ctx, "get", "pod", name, "--output=jsonpath={.metadata.uid}") + return strings.TrimSpace(string(b)), err +} + +func (k kubectl) createToken(ctx context.Context) (string, error) { + b, err := k.output(ctx, "create", "token", observerAccount, "--duration="+tokenDuration) + return string(b), err +} + +func (k kubectl) forward(ctx context.Context, target string) (string, func(), error) { + ctx, cancel := context.WithCancel(ctx) + cmd := exec.CommandContext(ctx, "kubectl", k.args("port-forward", "--address=127.0.0.1", target, ":8200")...) + cmd.Stderr = io.Discard + pipe, err := cmd.StdoutPipe() + if err != nil { + cancel() + return "", nil, err + } + if err := cmd.Start(); err != nil { + cancel() + return "", nil, err + } + stop := func() { cancel(); _ = cmd.Wait() } + port := make(chan string, 1) + go func() { + lines := bufio.NewScanner(pipe) + for lines.Scan() { + if m := forwarding.FindStringSubmatch(lines.Text()); m != nil { + port <- m[1] + break + } + } + _, _ = io.Copy(io.Discard, pipe) + }() + select { + case p := <-port: + return "http://127.0.0.1:" + p, stop, nil + case <-time.After(20 * time.Second): + stop() + return "", nil, errors.New("port-forward did not start") + } +} + +func (k kubectl) client() *http.Client { return k.http } + +func main() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + c := &http.Client{Timeout: 15 * time.Second, Transport: &http.Transport{Proxy: nil}} + os.Exit(run(ctx, os.Args[1:], os.Stdout, os.Stderr, kubectl{http: c}, time.Now)) +} diff --git a/scripts/capture-openbao-replacement/main_test.go b/scripts/capture-openbao-replacement/main_test.go new file mode 100644 index 0000000000..4960a01163 --- /dev/null +++ b/scripts/capture-openbao-replacement/main_test.go @@ -0,0 +1,334 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" +) + +const secretToken = "synthetic-openbao-token-only" + +// fake stands in for kubectl and for each server behind a port-forward. +type fake struct { + mu sync.Mutex + uids map[string]string + uidsAfter map[string]string // UIDs reported once the health reads have begun + healthy map[string]int // health status per pod + reads map[string]int + forwards []string + loginCode int + raftCode int + revoked bool + revokeErr bool + jwt string + tokenErr bool + servers []*httptest.Server +} + +func newFake() *fake { + return &fake{ + uids: map[string]string{"openbao-0": "uid-0", "openbao-1": "uid-1", "openbao-2": "uid-2"}, + uidsAfter: map[string]string{}, + healthy: map[string]int{"openbao-0": 200, "openbao-1": 200, "openbao-2": 200}, + reads: map[string]int{}, + loginCode: 200, + raftCode: 200, + jwt: "synthetic-service-account-jwt\n", + } +} + +func (f *fake) close() { + for _, s := range f.servers { + s.Close() + } +} + +func (f *fake) podUID(_ context.Context, name string) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.forwards) > 0 { + if uid, ok := f.uidsAfter[name]; ok { + return uid, nil + } + } + uid, ok := f.uids[name] + if !ok { + return "", errors.New("not found") + } + return uid, nil +} + +func (f *fake) createToken(context.Context) (string, error) { + if f.tokenErr { + return "", errors.New("forbidden") + } + return f.jwt, nil +} + +func (f *fake) forward(_ context.Context, target string) (string, func(), error) { + f.mu.Lock() + f.forwards = append(f.forwards, target) + f.mu.Unlock() + mux := http.NewServeMux() + if name, ok := strings.CutPrefix(target, "pod/"); ok { + mux.HandleFunc("/v1/sys/health", func(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + f.reads[name]++ + code := f.healthy[name] + f.mu.Unlock() + if r.URL.Query().Get("standbyok") != "true" || r.Header.Get("X-Vault-Token") != "" { + code = 400 + } + w.WriteHeader(code) + standby := "true" + if name == "openbao-0" { + standby = "false" + } + _, _ = io.WriteString(w, `{"initialized":true,"sealed":false,"standby":`+standby+`,"server_time_utc":`+strconv.FormatInt(time.Now().Unix(), 10)+`,"version":"2.6.3","cluster_id":"c","node":"`+name+`"}`) + }) + } else if target == activeService { + mux.HandleFunc("/v1/auth/kubernetes/login", func(w http.ResponseWriter, r *http.Request) { + var body map[string]string + if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&body) != nil || body["role"] != observerAccount || body["jwt"] != "synthetic-service-account-jwt" { + w.WriteHeader(400) + return + } + w.WriteHeader(f.loginCode) + _, _ = io.WriteString(w, `{"auth":{"client_token":"`+secretToken+`"}}`) + }) + mux.HandleFunc("/v1/sys/storage/raft/configuration", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.Header.Get("X-Vault-Token") != secretToken { + w.WriteHeader(403) + _, _ = io.WriteString(w, `{"errors":["permission denied"]}`) + return + } + if f.raftCode < 0 { + w.WriteHeader(502) + _, _ = io.WriteString(w, "bad gateway") + return + } + w.WriteHeader(f.raftCode) + _, _ = io.WriteString(w, `{"data":{"config":{"index":0,"servers":[{"node_id":"openbao-0","address":"10.0.0.1:8201","leader":true,"voter":true,"protocol_version":"3"},{"node_id":"openbao-1","address":"10.0.0.2:8201","leader":false,"voter":true,"protocol_version":"3"},{"node_id":"openbao-2","address":"10.0.0.3:8201","leader":false,"voter":true,"protocol_version":"3"}]}}}`) + }) + mux.HandleFunc("/v1/auth/token/revoke-self", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.Header.Get("X-Vault-Token") != secretToken || f.revokeErr { + w.WriteHeader(403) + return + } + f.mu.Lock() + f.revoked = true + f.mu.Unlock() + w.WriteHeader(204) + }) + } else { + return "", nil, errors.New("unexpected target") + } + s := httptest.NewServer(mux) + f.servers = append(f.servers, s) + return s.URL, func() {}, nil +} + +func (f *fake) client() *http.Client { return &http.Client{Timeout: 5 * time.Second} } + +func clock() func() time.Time { + t := time.Date(2026, 11, 3, 16, 0, 0, 0, time.UTC) + return func() time.Time { t = t.Add(time.Second); return t } +} + +func privateDir(t *testing.T) string { + t.Helper() + dir := filepath.Join(t.TempDir(), "private") + if err := os.Mkdir(dir, 0o700); err != nil { + t.Fatal(err) + } + return dir +} + +func capture(t *testing.T, f *fake, dir string, extra ...string) (int, string, string) { + t.Helper() + var stdout, stderr bytes.Buffer + args := append([]string{"--expect-servers", "openbao-0,openbao-1,openbao-2", "--out", dir}, extra...) + code := run(context.Background(), args, &stdout, &stderr, f, clock()) + for _, out := range []string{stdout.String(), stderr.String()} { + if strings.Contains(out, secretToken) || strings.Contains(out, "synthetic-service-account-jwt") || strings.Contains(out, "10.0.0.1") { + t.Fatalf("a credential or address reached the log: %q", out) + } + } + return code, stdout.String(), stderr.String() +} + +func TestCapturesBothObservationsForTheJudges(t *testing.T) { + f := newFake() + defer f.close() + dir := privateDir(t) + code, out, _ := capture(t, f, dir) + if code != 0 || out != "OPENBAO_REPLACEMENT_EVIDENCE=CAPTURED\n" { + t.Fatalf("code=%d out=%q", code, out) + } + if !f.revoked { + t.Fatal("the observer token was not revoked") + } + var health healthCapture + b, _ := os.ReadFile(filepath.Join(dir, "health.json")) + if json.Unmarshal(b, &health) != nil || len(health.Servers) != 3 || health.StartedAt >= health.CapturedAt { + t.Fatalf("health capture: %s", b) + } + for i, s := range health.Servers { + var body map[string]any + if json.Unmarshal(s.Health, &body) != nil || body["node"] != s.Name || s.PodUID != f.uids[s.Name] || s.HealthStatus != 200 || s.Name != []string{"openbao-0", "openbao-1", "openbao-2"}[i] { + t.Fatalf("observation %d not bound to its own server: %+v", i, s) + } + } + var quorum quorumCapture + b, _ = os.ReadFile(filepath.Join(dir, "quorum.json")) + if json.Unmarshal(b, &quorum) != nil || quorum.Status != 200 || !bytes.Contains(quorum.Configuration, []byte(`"node_id":"openbao-0"`)) || quorum.StartedAt >= quorum.CapturedAt { + t.Fatalf("quorum capture: %s", b) + } + for _, name := range []string{"health.json", "quorum.json"} { + info, err := os.Stat(filepath.Join(dir, name)) + if err != nil || info.Mode().Perm() != 0o600 { + t.Fatalf("%s is not private", name) + } + } + want := []string{"pod/openbao-0", "pod/openbao-1", "pod/openbao-2", activeService} + if strings.Join(f.forwards, " ") != strings.Join(want, " ") { + t.Fatalf("forwards %v", f.forwards) + } +} + +func TestRecordsANonSuccessHealthStatusForTheJudge(t *testing.T) { + f := newFake() + defer f.close() + f.healthy["openbao-1"] = 503 + dir := privateDir(t) + if code, _, _ := capture(t, f, dir); code != 0 { + t.Fatal("a sealed server is the judge's finding, not a failed capture") + } + var health healthCapture + b, _ := os.ReadFile(filepath.Join(dir, "health.json")) + if json.Unmarshal(b, &health) != nil || health.Servers[1].HealthStatus != 503 { + t.Fatalf("status not preserved: %s", b) + } +} + +func TestFailedReadsAreUnverifiedAndWriteNothing(t *testing.T) { + for _, tc := range []struct { + name string + set func(*fake) + stage stage + }{ + {"missing pod", func(f *fake) { delete(f.uids, "openbao-2") }, stagePods}, + {"pod replaced mid-capture", func(f *fake) { f.uidsAfter["openbao-1"] = "uid-1-new" }, stagePods}, + {"token mint refused", func(f *fake) { f.tokenErr = true }, stageIdentity}, + {"empty token", func(f *fake) { f.jwt = "\n" }, stageIdentity}, + {"login refused", func(f *fake) { f.loginCode = 403 }, stageLogin}, + {"raft read unreadable", func(f *fake) { f.raftCode = -1 }, stageRaft}, + {"revoke refused", func(f *fake) { f.revokeErr = true }, stageRevoke}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newFake() + defer f.close() + tc.set(f) + dir := privateDir(t) + code, out, _ := capture(t, f, dir) + if code != 2 || out != "OPENBAO_REPLACEMENT_EVIDENCE=UNVERIFIED stage="+string(tc.stage)+"\n" { + t.Fatalf("code=%d out=%q", code, out) + } + if entries, _ := os.ReadDir(dir); len(entries) != 0 { + t.Fatal("a failed capture left evidence for the judges") + } + }) + } +} + +func TestRaftReadStatusIsPassedThroughNotJudged(t *testing.T) { + // A non-200 raft read with a valid body still reaches the judge, which owns + // the UNKNOWN verdict for it; only an unreadable body fails the capture. + f := newFake() + defer f.close() + f.raftCode = 429 + dir := privateDir(t) + if code, _, _ := capture(t, f, dir); code != 0 { + t.Fatal("capture refused a readable raft response") + } + var quorum quorumCapture + b, _ := os.ReadFile(filepath.Join(dir, "quorum.json")) + if json.Unmarshal(b, &quorum) != nil || quorum.Status != 429 { + t.Fatalf("status not preserved: %s", b) + } +} + +func TestRefusesInvalidArguments(t *testing.T) { + f := newFake() + defer f.close() + open := filepath.Join(t.TempDir(), "open") + if err := os.Mkdir(open, 0o755); err != nil { + t.Fatal(err) + } + for _, args := range [][]string{ + {}, + {"--expect-servers", "openbao-0,openbao-1", "--out", privateDir(t)}, + {"--expect-servers", "openbao-0,openbao-0,openbao-1", "--out", privateDir(t)}, + {"--expect-servers", "openbao-0,openbao-1,other", "--out", privateDir(t)}, + {"--expect-servers", "openbao-0,openbao-1,openbao-2", "--out", open}, + {"--expect-servers", "openbao-0,openbao-1,openbao-2", "--out", filepath.Join(t.TempDir(), "absent")}, + {"--expect-servers", "openbao-0,openbao-1,openbao-2", "--out", privateDir(t), "extra"}, + {"--expect-servers", "openbao-0,openbao-1,openbao-2", "--out", privateDir(t), "--unknown"}, + } { + var stdout, stderr bytes.Buffer + if code := run(context.Background(), args, &stdout, &stderr, f, clock()); code != 2 || stdout.Len() != 0 { + t.Fatalf("%v: code=%d out=%q", args, code, stdout.String()) + } + } + if len(f.forwards) != 0 { + t.Fatal("invalid arguments reached the cluster") + } +} + +func TestForwardingLineIsParsedStrictly(t *testing.T) { + for line, ok := range map[string]bool{ + "Forwarding from 127.0.0.1:41234 -> 8200": true, + "Forwarding from [::1]:41234 -> 8200": false, + "Forwarding from 0.0.0.0:41234 -> 8200": false, + "Forwarding from 127.0.0.1:41234 -> 8201": false, + " Forwarding from 127.0.0.1:41234 -> 8200": false, + } { + if (forwarding.FindStringSubmatch(line) != nil) != ok { + t.Fatalf("%q", line) + } + } +} + +// The judges decide; this proves they can read what the collector writes, so +// an envelope drift fails here rather than as UNKNOWN in production. +func TestJudgesAcceptTheCaptures(t *testing.T) { + f := newFake() + defer f.close() + dir := privateDir(t) + var stdout, stderr bytes.Buffer + args := []string{"--expect-servers", "openbao-0,openbao-1,openbao-2", "--out", dir} + if code := run(context.Background(), args, &stdout, &stderr, f, time.Now); code != 0 { + t.Fatalf("capture: %q", stdout.String()) + } + for _, judge := range [][]string{ + {"../judge-openbao-health", "--input", filepath.Join(dir, "health.json"), "--expect-servers", "openbao-0,openbao-1,openbao-2", "--expect-version", "2.6.3", "--max-age", "5m"}, + {"../judge-openbao-quorum", "--input", filepath.Join(dir, "quorum.json"), "--expect-servers", "openbao-0,openbao-1,openbao-2", "--replace", "openbao-2", "--max-age", "5m"}, + } { + out, err := exec.Command("go", append([]string{"run"}, judge...)...).CombinedOutput() + if err != nil || !bytes.Contains(out, []byte("PROVEN")) { + t.Fatalf("%s: %v %s", judge[0], err, out) + } + } +} diff --git a/scripts/capture-openbao-replacement/workflow_test.go b/scripts/capture-openbao-replacement/workflow_test.go new file mode 100644 index 0000000000..2cffac7459 --- /dev/null +++ b/scripts/capture-openbao-replacement/workflow_test.go @@ -0,0 +1,113 @@ +package main + +import ( + "os" + "path/filepath" + "reflect" + "regexp" + "strings" + "testing" + + "gopkg.in/yaml.v3" +) + +func obj(v any) map[string]any { m, _ := v.(map[string]any); return m } +func str(v any) string { s, _ := v.(string); return s } +func array(v any) []any { a, _ := v.([]any); return a } + +func field(m map[string]any, keys ...string) any { + var v any = m + for _, key := range keys { + v = obj(v)[key] + } + return v +} + +func workflow(t *testing.T) map[string]any { + t.Helper() + b, err := os.ReadFile(filepath.Join("..", "..", ".github", "workflows", "replace-openbao-server.yaml")) + if err != nil { + t.Fatal(err) + } + var w map[string]any + if yaml.Unmarshal(b, &w) != nil { + t.Fatal("invalid workflow") + } + return w +} + +// The preflight must stay manual, protected, serialized with production +// deploys and unable to change anything, until a reviewed change adds the +// replacement step (#4970). +func TestPreflightStaysManualProtectedAndReadOnly(t *testing.T) { + w := workflow(t) + if len(obj(w["on"])) != 1 || obj(w["on"])["workflow_dispatch"] == nil || len(obj(w["permissions"])) != 0 || + field(w, "concurrency", "group") != "prod-deploy" || field(w, "concurrency", "cancel-in-progress") != false || field(w, "concurrency", "queue") != "max" { + t.Fatal("manual or serialization boundary changed") + } + mode := obj(field(w, "on", "workflow_dispatch", "inputs", "mode")) + if mode["default"] != "preflight" || !reflect.DeepEqual(mode["options"], []any{"preflight"}) { + t.Fatal("a mode other than the read-only preflight appeared") + } + jobs := obj(w["jobs"]) + if len(jobs) != 1 { + t.Fatal("unexpected execution lane") + } + j := obj(jobs["preflight"]) + if !reflect.DeepEqual(j["environment"], map[string]any{"name": "prod", "deployment": false}) || !reflect.DeepEqual(j["permissions"], map[string]any{"contents": "read"}) { + t.Fatal("protection or GitHub authority changed") + } + mutating := regexp.MustCompile(`\b(kubectl|bao|vault)\b[^\n]*\b(delete|patch|apply|scale|rollout|edit|replace|annotate|label|drain|cordon|write)\b`) + source, tested, restored, captured := false, false, false, false + for _, v := range array(j["steps"]) { + s := obj(v) + body := str(s["run"]) + if mutating.MatchString(body) { + t.Fatalf("a step can change production: %q", s["name"]) + } + if strings.Contains(body, "verify-prod-recovery-source.sh") { + source = true + } + if strings.Contains(body, "go test -race -count=1 ./scripts/capture-openbao-replacement ./scripts/judge-openbao-health ./scripts/judge-openbao-quorum") { + if restored { + t.Fatal("regressions after credentials") + } + tested = true + } + if field(s, "env", "KUBE_CONFIG") != nil { + if !tested || !source { + t.Fatal("credentials before exact-source and regressions") + } + restored = true + } + if strings.Contains(body, "capture-openbao-replacement\" --expect-servers") { + if !restored || !strings.Contains(body, "judge-openbao-health") || !strings.Contains(body, "judge-openbao-quorum") || !strings.Contains(body, "--replace \"$REPLACEMENT_SERVER\"") { + t.Fatal("capture is not judged by both judges for the named server") + } + captured = true + } + } + if !source || !tested || !restored || !captured { + t.Fatal("missing protected controls") + } +} + +// The collector may read pods, mint the observer's token and port-forward; +// nothing else reaches kubectl. +func TestCollectorUsesOnlyReadingKubectlVerbs(t *testing.T) { + b, err := os.ReadFile("main.go") + if err != nil { + t.Fatal(err) + } + calls := regexp.MustCompile(`k\.(?:output|args)\((?:ctx, )?"([a-z-]+)"`).FindAllStringSubmatch(string(b), -1) + seen := map[string]bool{} + for _, c := range calls { + seen[c[1]] = true + } + if !reflect.DeepEqual(seen, map[string]bool{"get": true, "create": true, "port-forward": true}) { + t.Fatalf("kubectl verbs changed: %v", seen) + } + if !strings.Contains(string(b), `"create", "token", observerAccount`) { + t.Fatal("create is used for something other than the observer token") + } +}