From cd2a8d2b5e37f1b79a4b487760db5f3f7b3f818b Mon Sep 17 00:00:00 2001 From: "Cheng-Chen, Chen" Date: Thu, 22 Oct 2026 12:27:43 +0800 Subject: [PATCH] Preserve scoped stream authorization-loss audit causes --- .github/workflows/security-release-gate.yml | 6 +- COMPATIBILITY.md | 4 +- README.md | 4 +- auth/delegation.go | 29 +- auth/stream_audit.go | 84 +++++- auth/stream_cause_test.go | 311 ++++++++++++++++++++ delegation/classified_test.go | 152 ++++++++++ delegation/delegation.go | 83 ++++-- exec/exec.go | 3 +- logs/logs.go | 4 +- logs/scoped_logs_integration_test.go | 28 +- main.go | 2 +- main_test.go | 2 +- scripts/build | 2 +- scripts/package | 2 +- scripts/test | 2 +- stats/container_stats.go | 2 +- stats/host_stats.go | 2 +- stats/stats.go | 2 +- 19 files changed, 663 insertions(+), 61 deletions(-) create mode 100644 auth/stream_cause_test.go create mode 100644 delegation/classified_test.go diff --git a/.github/workflows/security-release-gate.yml b/.github/workflows/security-release-gate.yml index 144984cc..b0589019 100644 --- a/.github/workflows/security-release-gate.yml +++ b/.github/workflows/security-release-gate.yml @@ -16,7 +16,7 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 120 env: - CANDIDATE_VERSION: 0.38.6 + CANDIDATE_VERSION: 0.38.7 DAPPER_IMAGE: pasturestack-host-api-dapper:security-gate TRIVY_IMAGE: ghcr.io/aquasecurity/trivy:0.73.0@sha256:7cced7cae583819fc7806d4cbc0dbbc7cad18b99f7d3e235192e6da8c091045c COSIGN_IMAGE: ghcr.io/sigstore/cosign/cosign:v3.1.3@sha256:9e5c2f2edc34351160407ca3416c61855bdf9403c3c5936e0f0be7fc261611b8 @@ -36,8 +36,8 @@ jobs: test -z "$(git status --porcelain)" test "$(git rev-parse HEAD)" = "$GITHUB_SHA" # Exactly one reviewed maintenance commit after the existing published main. - test "$(git rev-list --count b124a9cce56cc3e1082d2f72434592bccb15f6ef..HEAD)" = 1 - test "$(git rev-list --count --merges b124a9cce56cc3e1082d2f72434592bccb15f6ef..HEAD)" = 0 + test "$(git rev-list --count 67d1e58d1999730f106ba76c3ee3cf03bbf495d6..HEAD)" = 1 + test "$(git rev-list --count --merges 67d1e58d1999730f106ba76c3ee3cf03bbf495d6..HEAD)" = 0 private_pattern=$(printf '%s|%s|%s' 'chen''21019' '10\.0\.0\.''125' 'C:\\''Users\\') if git grep -n -E "$private_pattern" -- .; then echo 'Private operator data is present in the public source tree.' >&2 diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md index 0e6ef634..05dc2fc8 100644 --- a/COMPATIBILITY.md +++ b/COMPATIBILITY.md @@ -6,7 +6,9 @@ Preferred settings use `platform-*` names. Historical `cattle-*` flags, legacy D Operator lifecycle messages support `en-US` and `zh-TW`. HTTP response bodies, log streams, statistics payloads, Docker output, identifiers, and protocol errors are not translated. -The source package version is `0.38.6`, built with Go 1.27.3 on a digest-pinned Ubuntu 26.04 image, a frozen bootstrap package snapshot, and checksum-pinned official security fixes. The final builder inventory records installed versions separately from the bootstrap lock. A published package must retain its reviewed producer revision, byte-identical packaging evidence, and successful security checks of the unpacked binary. Server packaging consumes `host-api-0.38.6.tar.gz`; its filename, executable layout, architecture, legacy SHA-1 compatibility files, and SHA-256 integrity files are release contracts. Product versions must remain pure numeric `MAJOR.MINOR.PATCH` values without branding or maintenance suffixes. +The source package version is `0.38.7`, built with Go 1.27.3 on a digest-pinned Ubuntu 26.04 image, a frozen bootstrap package snapshot, and checksum-pinned official security fixes. The final builder inventory records installed versions separately from the bootstrap lock. A published package must retain its reviewed producer revision, byte-identical packaging evidence, and successful security checks of the unpacked binary. Server packaging consumes `host-api-0.38.7.tar.gz`; its filename, executable layout, architecture, legacy SHA-1 compatibility files, and SHA-256 integrity files are release contracts. Product versions must remain pure numeric `MAJOR.MINOR.PATCH` values without branding or maintenance suffixes. + +The `key-audit-v1,key-delegation-v1` backend protocol and endpoint contracts remain unchanged. All non-allowed introspection results still close scoped streams. Confirmed authorization-loss evidence requires either expiry of the previously verified signed grant, or a configured-Engine 401/403 response carrying an allowlisted stable authorization code. Network failures, redirects, outages, malformed or mismatched responses are unknown, never confirmed revocation. Only authorized scoped transport cancellations receive a final recheck, inside the existing seven-second total report budget and without holding the audit mutex. A confirmed denial outranks a later disconnect or cancellation-induced Docker error; a completed success or genuine backend failure and the frozen first terminal record cannot be overwritten. Ordinary JWTs, Full/legacy traces, handshake admission, durable retry and ticket-bound agent/target/Host identities retain their contracts. No publication or native PASS is implied by the source version. Before release, validate token acquisition, event forwarding, logs, stats, exec, console, Docker socket proxy, container proxy, disabled-proxy behavior, key parsing, and restart behavior. diff --git a/README.md b/README.md index 9782a1a7..c1d1dad7 100644 --- a/README.md +++ b/README.md @@ -6,12 +6,14 @@ PastureStack is an independent community effort to preserve, audit, and moderniz ## Capabilities -The source package version is `0.38.6`, built with Go 1.27.3. Use a published release only after its source, reproducibility, vulnerability, and packaged-binary checks pass; a source version is not publication evidence. Target-bound API Key tickets require matching Engine introspection at handshake and during logs, exec, and statistics streams. Revocation or expiry closes the stream; it does not stop its container or kill an already started exec process. Legacy and full-without-expiry tokens retain their authentication contract. +The source package version is `0.38.7`, built with Go 1.27.3. Use a published release only after its source, reproducibility, vulnerability, and packaged-binary checks pass; a source version is not publication evidence. Target-bound API Key tickets require matching Engine introspection at handshake and during logs, exec, and statistics streams. Revocation or expiry closes the stream; it does not stop its container or kill an already started exec process. Legacy and full-without-expiry tokens retain their authentication contract. The authenticated backend connection declares the exact `key-audit-v1,key-delegation-v1` capabilities. A compatible Proxy requires that declaration for every verified API Key stream, including full keys, so an older embedded backend cannot bypass terminal audit. Ordinary non-Key streams retain their existing backend compatibility. Terminal evidence is sent with the host agent's credential. A private persistent spool reserves a PENDING record before execution for every verified API Key trace, preserves the first terminal result, and retries only evidence after outages or process restarts. Interrupted PENDING streams become `CANCELLED`, never a fabricated success. Original API Key and agent secrets are not persisted. Full keys retain their permission payload and expiry contract, but do not bypass audit durability; unavailable admission closes with retryable `AuditUnavailable` (503 semantics), not a permission denial. Ordinary non-Key tokens have no new audit-admission requirement. +For an authorized scoped stream, terminal classification preserves confirmed authorization loss even when a later transport close arrives. A transport cancellation receives one bounded recheck of the original signed grant before the terminal record is frozen; an unavailable or unverifiable check remains `StreamCancelled`, not confirmed revocation. This records authorization at finalization, not proof of which peer first closed the connection. Completed EOF or backend failures, ordinary JWTs and Full/legacy permission contracts are unchanged. + ## Configuration Preferred flags are `--platform-url`, `--platform-access-key`, and `--platform-secret-key`. Existing `--cattle-*` flags remain compatibility aliases. Set `PASTURESTACK_HOME` for the state root and `PASTURESTACK_LOCALE=en-US` or `zh-TW` for operator messages. `host-api --version` reports the pure numeric `MAJOR.MINOR.PATCH` build version. diff --git a/auth/delegation.go b/auth/delegation.go index 28592c77..ef1a0c87 100644 --- a/auth/delegation.go +++ b/auth/delegation.go @@ -37,6 +37,27 @@ func verifyDelegation(raw, path string, token *jwt.Token) (*jwt.Token, bool) { // The original signed token, not normalized claims, is rechecked for every tick. func StartStreamWatch(parent context.Context, raw string, verified *jwt.Token, deny func()) context.CancelFunc { + return startStreamWatchResult(parent, raw, verified, func(delegation.Result) { deny() }) +} + +// Watch closes immediately on either a confirmed denial or an unknown check. +// Only the former may become AuthorizationRevoked terminal evidence. +func (a *StreamAudit) Watch(parent context.Context, raw string, verified *jwt.Token, deny func()) context.CancelFunc { + return startStreamWatchResult(parent, raw, verified, func(result delegation.Result) { + a.authorizationResult(result) + deny() + }) +} + +func (a *StreamAudit) authorizationResult(result delegation.Result) { + if result == delegation.ConfirmedDenied { + a.Cancel("AuthorizationRevoked") + } else if result != delegation.Allowed { + a.Cancel("StreamCancelled") + } +} + +func startStreamWatchResult(parent context.Context, raw string, verified *jwt.Token, deny func(delegation.Result)) context.CancelFunc { initial, ok := MapClaims(verified) if ok { if _, scoped := initial[delegation.Claim]; !scoped { @@ -45,18 +66,18 @@ func StartStreamWatch(parent context.Context, raw string, verified *jwt.Token, d } token, err := ParseToken(raw, config.Config.ParsedPublicKey) if err != nil { - deny() + deny(delegation.Unknown) return func() {} } claims, ok := MapClaims(token) if !ok { - deny() + deny(delegation.Unknown) return func() {} } grant, err := delegation.Parse(map[string]interface{}(claims)) if err != nil { - deny() + deny(delegation.Unknown) return func() {} } - return delegation.New(config.Config.PlatformURL).Watch(parent, raw, grant, deny) + return delegation.New(config.Config.PlatformURL).WatchResult(parent, raw, grant, deny) } diff --git a/auth/stream_audit.go b/auth/stream_audit.go index 1dcea306..d5523f6a 100644 --- a/auth/stream_audit.go +++ b/auth/stream_audit.go @@ -22,6 +22,9 @@ import ( type StreamAudit struct { mu sync.Mutex enabled, required, cancelled bool + authorized, completed, frozen bool + reportOnce sync.Once + recheck func(context.Context) delegation.Result token, endpoint, accessKey, secretKey string outcome, failureCode string client *http.Client @@ -68,6 +71,12 @@ func BeginStreamAudit(raw string) *StreamAudit { return a } } + if a.required && a.enabled { + if grant, err := delegation.Parse(map[string]interface{}(claims)); err == nil && grant != nil { + verifier := delegation.New(config.Config.PlatformURL) + a.recheck = func(ctx context.Context) delegation.Result { return verifier.CheckResult(ctx, raw, grant) } + } + } if envelope, ok := claims["apiKeyAudit"].(map[string]interface{}); ok { a.expiresAt = receiptExpiry(envelope) } @@ -103,51 +112,94 @@ func (a *StreamAudit) Ready() bool { func (a *StreamAudit) Enabled() bool { return a.enabled } func (a *StreamAudit) Authorized() { a.mu.Lock() + defer a.mu.Unlock() + if a.frozen || a.completed || a.cancelled { + return + } + a.authorized = true a.outcome = "CANCELLED" a.failureCode = "ClientDisconnected" - a.mu.Unlock() } func (a *StreamAudit) Succeed() { a.mu.Lock() defer a.mu.Unlock() - if !a.cancelled { + if !a.cancelled && !a.frozen && !a.completed { a.outcome = "SUCCEEDED" a.failureCode = "" + a.completed = true } } func (a *StreamAudit) Fail(code string) { a.mu.Lock() defer a.mu.Unlock() - if !a.cancelled { + if !a.cancelled && !a.frozen && !a.completed { a.outcome = "FAILED" a.failureCode = code + a.completed = true } } func (a *StreamAudit) Cancel(code string) { a.mu.Lock() defer a.mu.Unlock() + if a.frozen || a.completed || (a.cancelled && cancellationPriority(code) <= cancellationPriority(a.failureCode)) { + return + } a.cancelled = true a.outcome = "CANCELLED" a.failureCode = code } -func (a *StreamAudit) Report() { - if !a.enabled { - return +func cancellationPriority(code string) int { + switch code { + case "AuthorizationRevoked": + return 3 + case "StreamCancelled": + return 2 + default: + return 1 } - queue := a.queue - if queue == nil { - queue, _ = configuredCompletionSpool() +} + +func (a *StreamAudit) finalize(ctx context.Context) { + a.mu.Lock() + check := a.recheck + transport := a.required && a.authorized && !a.completed && !a.frozen && + a.outcome == "CANCELLED" && a.failureCode != "AuthorizationRevoked" + a.mu.Unlock() + if transport { + result := delegation.Unknown + if check != nil { + result = check(ctx) // Never hold the audit mutex during network I/O. + } + a.authorizationResult(result) } - if queue == nil || queue.complete(a.record("TERMINAL")) != nil { - glog.Warning("API key stream terminal evidence could not be persisted") + a.mu.Lock() + a.frozen = true + a.mu.Unlock() +} + +func (a *StreamAudit) Report() { + if !a.enabled { return } - ctx, cancel := context.WithTimeout(context.Background(), 7*time.Second) - defer cancel() - if !queue.deliverOne(ctx, completionID(a.token)) { - glog.Warning("API key stream terminal audit remains pending delivery") - } + a.reportOnce.Do(func() { + // Rechecking and delivery share the existing total report budget. The + // stream is already closed; this cannot renew or reopen its authority. + ctx, cancel := context.WithTimeout(context.Background(), 7*time.Second) + defer cancel() + a.finalize(ctx) + queue := a.queue + if queue == nil { + queue, _ = configuredCompletionSpool() + } + if queue == nil || queue.complete(a.record("TERMINAL")) != nil { + glog.Warning("API key stream terminal evidence could not be persisted") + return + } + if !queue.deliverOne(ctx, completionID(a.token)) { + glog.Warning("API key stream terminal audit remains pending delivery") + } + }) } func (a *StreamAudit) deliver(ctx context.Context) bool { diff --git a/auth/stream_cause_test.go b/auth/stream_cause_test.go new file mode 100644 index 00000000..2264dfd3 --- /dev/null +++ b/auth/stream_cause_test.go @@ -0,0 +1,311 @@ +package auth + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/PastureStack/host-api/config" + "github.com/PastureStack/host-api/delegation" + "github.com/PastureStack/host-api/testutils" +) + +type causeFixture struct { + audit *StreamAudit + raw string + state atomic.Int32 + checks atomic.Int32 + deliveries atomic.Int32 + events chan map[string]string + rejectFirst bool + beforeReply func(int32) +} + +func newCauseFixture(t *testing.T, claim string, admitted bool) *causeFixture { + t.Helper() + prior := config.Config + t.Cleanup(func() { config.Config = prior }) + config.Config.ParsedPublicKey = testutils.ParseTestPublicKey() + config.Config.PlatformAccessKey = "unit-agent" + config.Config.PlatformSecretKey = "unit-agent-secret" + config.Config.CompletionSpoolDir = testCompletionDir(t) + config.Config.HostUuidCheck = true + config.Config.HostUuid = "unit-host" + f := &causeFixture{events: make(chan map[string]string, 8)} + now := time.Now().Unix() + expires := now + 120 + envelope := map[string]interface{}{"version": 1, "issuedAt": now, "keyId": 12, "revision": 3, + "principalId": 4, "accountId": 5, "requestId": "00000000-0000-4000-8000-000000000001", + "targetType": "container", "targetId": "1i9", "operation": "logs", "expiresAt": expires} + payload := map[string]interface{}{"hostUuid": "unit-host", "logs": map[string]interface{}{"Container": "unit-container", "Follow": true}} + claims := map[string]interface{}{"exp": expires} + if claim == delegation.Claim { + envelope["payload"] = payload + claims[delegation.Claim] = envelope + } else { + for key, value := range payload { + claims[key] = value + } + if claim == "apiKeyAudit" { + claims[claim] = envelope + } + } + f.raw = testutils.CreateTokenWithPayload(claims, testutils.ParseTestPrivateKey()) + digest := sha256.Sum256([]byte(f.raw)) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v2-beta/apiKeyDelegation/introspect": + f.checks.Add(1) + var body map[string]string + if json.NewDecoder(r.Body).Decode(&body) != nil || body["token"] != f.raw || r.URL.RawQuery != "" { + t.Error("final check lost original signed-ticket binding") + } + if f.state.Load() == 1 { + w.WriteHeader(http.StatusForbidden) + json.NewEncoder(w).Encode(map[string]string{"code": "ApiKeyPolicyChanged"}) + return + } + if f.state.Load() == 2 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + json.NewEncoder(w).Encode(map[string]interface{}{"allowed": true, "keyId": 12, "revision": 3, + "targetType": "container", "targetId": "1i9", "operation": "logs", "expiresAt": expires, + "tokenDigest": hex.EncodeToString(digest[:])}) + case "/v2-beta/apiKeyDelegationCompletion": + user, secret, ok := r.BasicAuth() + if !ok || user != "unit-agent" || secret != "unit-agent-secret" { + t.Error("completion lost host-agent authentication") + } + var body map[string]string + if json.NewDecoder(r.Body).Decode(&body) != nil || body["token"] != f.raw { + t.Error("completion changed original signed ticket") + } + delete(body, "token") // Never include the signed fixture ticket in diagnostics. + n := f.deliveries.Add(1) + f.events <- body + if f.beforeReply != nil { + f.beforeReply(n) + } + if f.rejectFirst && n == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + json.NewEncoder(w).Encode(map[string]bool{"accepted": true}) + default: + t.Error("unexpected endpoint") + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(server.Close) + config.Config.PlatformURL = server.URL + f.audit = BeginStreamAudit(f.raw) + if admitted { + if _, valid := GetAndCheckStreamToken(f.raw, "/v2-beta/logs/"); !valid { + t.Fatal("valid signed fixture failed admission") + } + } + if !f.audit.Ready() { + t.Fatal("private completion fixture not ready") + } + if admitted { + f.audit.Authorized() + } + f.checks.Store(0) + return f +} + +func (f *causeFixture) event(t *testing.T, outcome, code string) { + t.Helper() + select { + case event := <-f.events: + if event["outcome"] != outcome || event["failureCode"] != code { + t.Fatalf("terminal outcome=%s code=%s", event["outcome"], event["failureCode"]) + } + case <-time.After(5 * time.Second): + t.Fatal("terminal evidence missing") + } +} + +func TestStreamCauseProxyFirstRechecksOriginalSignedGrant(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + // The Proxy transport-close barrier precedes any Host watcher callback. + f.state.Store(1) + f.audit.Cancel("ClientDisconnected") + f.audit.Fail("DockerFailure") // Cancellation-induced pipe error is secondary. + f.audit.Report() + f.event(t, "CANCELLED", "AuthorizationRevoked") + if f.checks.Load() != 1 { + t.Fatal("transport cancellation did not receive exactly one final check") + } +} + +func TestStreamCauseHostFirstCannotBeDowngradedByDisconnect(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + f.audit.authorizationResult(delegation.ConfirmedDenied) + f.audit.Cancel("ClientDisconnected") + f.audit.authorizationResult(delegation.Unknown) + f.audit.Fail("DockerFailure") + f.audit.Succeed() + f.audit.Report() + f.event(t, "CANCELLED", "AuthorizationRevoked") + if f.checks.Load() != 0 { + t.Fatal("known denial was needlessly rechecked") + } +} + +func TestStreamCauseGenuineClientDisconnectWithLiveGrant(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + f.audit.Cancel("ClientDisconnected") + f.audit.Report() + f.event(t, "CANCELLED", "ClientDisconnected") + if f.checks.Load() != 1 { + t.Fatal("live-grant disconnect classification skipped the final check") + } +} + +func TestStreamCauseUnavailableEngineIsNotConfirmedRevocation(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + f.state.Store(2) + f.audit.Cancel("ClientDisconnected") + f.audit.Report() + f.event(t, "CANCELLED", "StreamCancelled") +} + +func TestStreamCauseExpiryOfPreviouslyAdmittedSignedGrant(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + token, err := ParseToken(f.raw, config.Config.ParsedPublicKey) + if err != nil { + t.Fatal("signed expiry fixture invalid") + } + claims, _ := MapClaims(token) + grant, err := delegation.Parse(map[string]interface{}(claims)) + if err != nil || grant == nil { + t.Fatal("signed grant missing") + } + v := delegation.New(config.Config.PlatformURL) + v.Now = func() time.Time { return time.Unix(grant.ExpiresAt, 0) } + f.audit.recheck = func(ctx context.Context) delegation.Result { return v.CheckResult(ctx, f.raw, grant) } + f.audit.Cancel("ClientDisconnected") + f.audit.Report() + f.event(t, "CANCELLED", "AuthorizationRevoked") + if f.checks.Load() != 0 { + t.Fatal("known local expiry contacted Engine") + } +} + +func TestStreamCauseCompletedEOFAndBackendFailureRemainTerminal(t *testing.T) { + for _, mode := range []string{"EOF", "DockerFailure"} { + t.Run(mode, func(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + outcome, code := "SUCCEEDED", "" + if mode == "EOF" { + f.audit.Succeed() + } else { + f.audit.Fail(mode) + outcome, code = "FAILED", mode + } + f.audit.Cancel("ClientDisconnected") + f.audit.authorizationResult(delegation.ConfirmedDenied) + f.audit.Report() + f.event(t, outcome, code) + if f.checks.Load() != 0 { + t.Fatal("completed backend work received a new authorization requirement") + } + }) + } +} + +func TestStreamCauseOrdinaryJWTFullTraceAndHandshakeHaveNoFinalCheck(t *testing.T) { + for _, claim := range []string{"legacy", "apiKeyAudit", delegation.Claim} { + t.Run(claim, func(t *testing.T) { + admitted := claim != delegation.Claim + f := newCauseFixture(t, claim, admitted) + if admitted { + f.audit.Cancel("ClientDisconnected") + } + f.audit.Report() + if claim == "apiKeyAudit" { + f.event(t, "CANCELLED", "ClientDisconnected") + } else if claim == delegation.Claim { + f.event(t, "FAILED", "HandshakeDenied") + } else if f.deliveries.Load() != 0 { + t.Fatal("ordinary JWT acquired an audit requirement") + } + if f.checks.Load() != 0 { + t.Fatal("unscoped or unadmitted stream acquired a final check") + } + }) + } +} + +func TestStreamCauseFinalizationHasNoNetworkMutexAndFreezesOnce(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + f.audit.recheck = func(ctx context.Context) delegation.Result { + deadline, ok := ctx.Deadline() + if !ok || time.Until(deadline) > 7*time.Second { + t.Error("final check escaped the existing total report budget") + } + close(entered) + select { + case <-release: + case <-ctx.Done(): + } + return delegation.Unknown + } + f.audit.Cancel("ClientDisconnected") + go func() { f.audit.Report(); close(done) }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("final check did not reach its entry barrier") + } + callback := make(chan struct{}) + go func() { f.audit.authorizationResult(delegation.ConfirmedDenied); close(callback) }() + select { + case <-callback: + case <-time.After(time.Second): + close(release) + <-done + t.Fatal("network check held the audit mutex") + } + close(release) + <-done + f.event(t, "CANCELLED", "AuthorizationRevoked") + f.audit.Cancel("ClientDisconnected") + f.audit.Fail("DockerFailure") + f.audit.Succeed() + f.audit.Report() + if record := f.audit.record("TERMINAL"); record.Outcome != "CANCELLED" || record.FailureCode != "AuthorizationRevoked" || f.deliveries.Load() != 1 { + t.Fatal("late callback or repeated Report changed the frozen terminal") + } +} + +func TestStreamCauseEvidenceRetryKeepsSameFrozenTicketAndOutcome(t *testing.T) { + f := newCauseFixture(t, delegation.Claim, true) + f.rejectFirst = true + f.beforeReply = func(n int32) { + if n == 1 { + record, _, err := f.audit.queue.read(f.audit.queue.path(completionID(f.raw))) + if err != nil || record.Token != f.raw || record.State != "TERMINAL" || record.FailureCode != "AuthorizationRevoked" { + t.Error("first delivery did not use the one persisted terminal") + } + f.audit.Cancel("ClientDisconnected") + f.audit.Fail("DockerFailure") + } + } + f.audit.authorizationResult(delegation.ConfirmedDenied) + f.audit.Report() + f.event(t, "CANCELLED", "AuthorizationRevoked") + f.event(t, "CANCELLED", "AuthorizationRevoked") + if f.deliveries.Load() != 2 || f.checks.Load() != 0 { + t.Fatal("retry changed terminal identity or repeated stream authorization") + } +} diff --git a/delegation/classified_test.go b/delegation/classified_test.go new file mode 100644 index 00000000..2be4cb94 --- /dev/null +++ b/delegation/classified_test.go @@ -0,0 +1,152 @@ +package delegation + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" +) + +type classifiedTransport func(*http.Request) (*http.Response, error) + +func (f classifiedTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func TestClassifiedChecksPreserveFailClosedWithoutInventingRevocation(t *testing.T) { + _, grant := fixture() + allowed, _ := json.Marshal(answer(grant, "signed-ticket")) + wrong := answer(grant, "signed-ticket") + wrong["revision"] = grant.Revision + 1 + wrongBody, _ := json.Marshal(wrong) + cases := []struct { + name string + status int + body string + want Result + }{ + {"allowed", 200, string(allowed), Allowed}, + {"policy-shrink", 403, `{"code":"ApiKeyPolicyChanged"}`, ConfirmedDenied}, + {"revoked", 403, `{"code":"ApiKeyRevoked"}`, ConfirmedDenied}, + {"expired", 401, `{"code":"ApiKeyExpired"}`, ConfirmedDenied}, + {"owner-lost", 403, `{"code":"OwnerPermissionDenied"}`, ConfirmedDenied}, + {"unclassified-403", 403, `{"code":"Forbidden"}`, Unknown}, + {"invalid-ticket-not-revoke", 403, `{"code":"ApiKeyDelegationInvalid"}`, Unknown}, + {"outage-with-code", 503, `{"code":"ApiKeyRevoked"}`, Unknown}, + {"redirect", 302, `{"code":"ApiKeyRevoked"}`, Unknown}, + {"bad-json", 403, `{"code":`, Unknown}, + {"trailing-json", 403, `{"code":"ApiKeyRevoked"}{}`, Unknown}, + {"wrong-binding", 200, string(wrongBody), Unknown}, + {"oversize", 403, strings.Repeat("x", 4097), Unknown}, + {"network", 0, "", Unknown}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + v := New("https://engine.example/v1") + v.Now = func() time.Time { return time.Unix(900, 0) } + v.Client.Transport = classifiedTransport(func(r *http.Request) (*http.Response, error) { + if r.URL.String() != "https://engine.example/v2-beta/apiKeyDelegation/introspect" || r.Method != http.MethodPost || r.URL.RawQuery != "" { + t.Error("classification changed trusted request binding") + } + var body map[string]string + if json.NewDecoder(r.Body).Decode(&body) != nil || body["token"] != "signed-ticket" || len(body) != 1 { + t.Error("classification did not send the original ticket") + } + if tc.status == 0 { + return nil, errors.New("synthetic transport unavailable") + } + headers := http.Header{} + if tc.status == http.StatusFound { + headers.Set("Location", "https://foreign.example/introspect") + } + return &http.Response{StatusCode: tc.status, Header: headers, Body: io.NopCloser(strings.NewReader(tc.body))}, nil + }) + if got := v.CheckResult(context.Background(), "signed-ticket", grant); got != tc.want { + t.Fatalf("classification=%d want=%d", got, tc.want) + } + if (v.Check(context.Background(), "signed-ticket", grant) == nil) != (tc.want == Allowed) { + t.Fatal("ordinary authorization stopped failing closed") + } + }) + } +} + +func TestClassifiedExpiryUsesPreviouslyVerifiedGrantWithoutNetwork(t *testing.T) { + _, grant := fixture() + v := New("https://engine.example") + v.Now = func() time.Time { return time.Unix(grant.ExpiresAt, 0) } + v.Client.Transport = classifiedTransport(func(*http.Request) (*http.Response, error) { + t.Fatal("known expiry made a network request") + return nil, errors.New("unexpected request") + }) + if v.CheckResult(context.Background(), "signed-ticket", grant) != ConfirmedDenied || v.Check(context.Background(), "signed-ticket", grant) == nil { + t.Fatal("expiry classification or fail-closed changed") + } + if v.CheckResult(context.Background(), "legacy", nil) != Allowed { + t.Fatal("legacy grant semantics changed") + } + t.Run("expires-during-valid-response", func(t *testing.T) { + var now atomic.Int64 + now.Store(900) + v := New("https://engine.example") + v.Now = func() time.Time { return time.Unix(now.Load(), 0) } + v.Client.Transport = classifiedTransport(func(*http.Request) (*http.Response, error) { + body, _ := json.Marshal(answer(grant, "signed-ticket")) + now.Store(grant.ExpiresAt) // Valid-reply barrier, not elapsed wall time. + return &http.Response{StatusCode: 200, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(string(body)))}, nil + }) + if v.CheckResult(context.Background(), "signed-ticket", grant) != ConfirmedDenied { + t.Fatal("known expiry during valid reply became unknown or allowed") + } + now.Store(900) + if v.Check(context.Background(), "signed-ticket", grant) == nil { + t.Fatal("ordinary authorization allowed a grant expiring during its reply") + } + }) +} + +func TestClassifiedWatcherClosesAtControlledTickAndExpiryBarriers(t *testing.T) { + _, grant := fixture() + for _, mode := range []string{"confirmed", "unknown", "expiry"} { + t.Run(mode, func(t *testing.T) { + v := New("https://engine.example") + v.Now = func() time.Time { return time.Unix(900, 0) } + v.Client.Transport = classifiedTransport(func(*http.Request) (*http.Response, error) { + if mode == "unknown" { + return nil, errors.New("synthetic outage") + } + return &http.Response{StatusCode: 403, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(`{"code":"ApiKeyRevoked"}`))}, nil + }) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + ticks, expiry := make(chan time.Time), make(chan time.Time) + denied := make(chan Result, 1) + done := make(chan struct{}) + go func() { + defer close(done) + v.runResult(ctx, "signed-ticket", grant, ticks, expiry, func(result Result) { denied <- result }) + }() + if mode == "expiry" { + expiry <- time.Unix(grant.ExpiresAt, 0) + } else { + ticks <- time.Unix(901, 0) + } + select { + case got := <-denied: + want := ConfirmedDenied + if mode == "unknown" { + want = Unknown + } + if got != want { + t.Fatal("watcher lost the denial classification") + } + case <-ctx.Done(): + t.Fatal("watcher did not close at the barrier") + } + <-done + }) + } +} diff --git a/delegation/delegation.go b/delegation/delegation.go index 159890d8..683d6812 100644 --- a/delegation/delegation.go +++ b/delegation/delegation.go @@ -123,6 +123,17 @@ type Verifier struct { Now func() time.Time } +// Result separates confirmed loss of authorization from an unverifiable check. +// Both non-Allowed results still fail closed for stream authorization. Callers +// must supply the grant from the original signature-verified ticket. +type Result uint8 + +const ( + Unknown Result = iota + Allowed + ConfirmedDenied +) + func New(platformURL string) *Verifier { v := &Verifier{Now: time.Now, Client: &http.Client{Timeout: 3 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}} u, err := url.Parse(platformURL) @@ -135,60 +146,94 @@ func New(platformURL string) *Verifier { } func (v *Verifier) Check(ctx context.Context, token string, g *Grant) error { + if v.CheckResult(ctx, token, g) != Allowed { + return ErrDenied + } + return nil +} + +func (v *Verifier) CheckResult(ctx context.Context, token string, g *Grant) Result { if g == nil { - return nil + return Allowed } - if v == nil || v.Endpoint == "" || v.Client == nil || v.Now == nil || token == "" || len(token) > MaxTokenBytes || v.Now().Unix() >= g.ExpiresAt { - return ErrDenied + if v == nil || v.Now == nil || token == "" || len(token) > MaxTokenBytes { + return Unknown + } + if v.Now().Unix() >= g.ExpiresAt { + return ConfirmedDenied + } + if v.Endpoint == "" || v.Client == nil { + return Unknown } encoded, _ := json.Marshal(map[string]string{"token": token}) req, err := http.NewRequestWithContext(ctx, http.MethodPost, v.Endpoint, bytes.NewReader(encoded)) if err != nil { - return ErrDenied + return Unknown } req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json") resp, err := v.Client.Do(req) if err != nil { - return ErrDenied + return Unknown } defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return ErrDenied - } data, err := io.ReadAll(io.LimitReader(resp.Body, 4097)) if err != nil || len(data) > 4096 { - return ErrDenied + return Unknown } var result map[string]interface{} decoder := json.NewDecoder(bytes.NewReader(data)) decoder.UseNumber() if decoder.Decode(&result) != nil { - return ErrDenied + return Unknown + } + var extra interface{} + if decoder.Decode(&extra) != io.EOF { + return Unknown + } + if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden { + // Only stable live-authorization codes from the configured Engine qualify. + // A proxy/gateway status, malformed grant or outage is not proof of revoke. + switch result["code"] { + case "ApiKeyRevoked", "ApiKeyPolicyChanged", "ApiKeyExpired", "OwnerPermissionDenied", + "KeyPolicyDenied", "KeyScopeDenied", "ApiKeyDelegationTargetChanged": + return ConfirmedDenied + } + return Unknown + } + if resp.StatusCode != http.StatusOK { + return Unknown } key, keyOK := Integer(result["keyId"]) revision, revisionOK := Integer(result["revision"]) expiry, expiryOK := Integer(result["expiresAt"]) digest := sha256.Sum256([]byte(token)) if result["allowed"] != true || !keyOK || key != g.KeyID || !revisionOK || revision != g.Revision || !expiryOK || expiry != g.ExpiresAt || - result["targetType"] != g.TargetType || result["targetId"] != g.TargetID || result["operation"] != g.Operation || result["tokenDigest"] != hex.EncodeToString(digest[:]) || v.Now().Unix() >= g.ExpiresAt { - return ErrDenied + result["targetType"] != g.TargetType || result["targetId"] != g.TargetID || result["operation"] != g.Operation || result["tokenDigest"] != hex.EncodeToString(digest[:]) { + return Unknown } - return nil + if v.Now().Unix() >= g.ExpiresAt { + return ConfirmedDenied + } + return Allowed } // Run accepts controllable barriers: tests do not sleep to simulate revocation. func (v *Verifier) Run(ctx context.Context, token string, g *Grant, ticks, expiry <-chan time.Time, deny func()) { + v.runResult(ctx, token, g, ticks, expiry, func(Result) { deny() }) +} + +func (v *Verifier) runResult(ctx context.Context, token string, g *Grant, ticks, expiry <-chan time.Time, deny func(Result)) { for { select { case <-ctx.Done(): return case <-expiry: - deny() + deny(ConfirmedDenied) return case <-ticks: - if v.Check(ctx, token, g) != nil { - deny() + if result := v.CheckResult(ctx, token, g); result != Allowed { + deny(result) return } } @@ -196,6 +241,10 @@ func (v *Verifier) Run(ctx context.Context, token string, g *Grant, ticks, expir } func (v *Verifier) Watch(parent context.Context, token string, g *Grant, deny func()) context.CancelFunc { + return v.WatchResult(parent, token, g, func(Result) { deny() }) +} + +func (v *Verifier) WatchResult(parent context.Context, token string, g *Grant, deny func(Result)) context.CancelFunc { ctx, cancel := context.WithCancel(parent) if g == nil { return cancel @@ -206,6 +255,6 @@ func (v *Verifier) Watch(parent context.Context, token string, g *Grant, deny fu lifetime = 0 } timer := time.NewTimer(lifetime) - go func() { defer ticker.Stop(); defer timer.Stop(); v.Run(ctx, token, g, ticker.C, timer.C, deny) }() + go func() { defer ticker.Stop(); defer timer.Stop(); v.runResult(ctx, token, g, ticker.C, timer.C, deny) }() return cancel } diff --git a/exec/exec.go b/exec/exec.go index ea9b2206..9cbaa455 100644 --- a/exec/exec.go +++ b/exec/exec.go @@ -78,8 +78,7 @@ func (h *ExecHandler) Handle(key string, initialMessage string, incomingMessages defer inputWriter.Close() ctx, cancel := context.WithCancel(context.Background()) defer cancel() - stop := auth.StartStreamWatch(ctx, tokenString, token, func() { - audit.Cancel("AuthorizationRevoked") + stop := audit.Watch(ctx, tokenString, token, func() { cancel() inputReader.Close() inputWriter.Close() diff --git a/logs/logs.go b/logs/logs.go index 41791ddb..ad9b7126 100644 --- a/logs/logs.go +++ b/logs/logs.go @@ -97,9 +97,9 @@ func (l *LogsHandler) Handle(key string, initialMessage string, incomingMessages defer cancel() watch := l.streamWatch if watch == nil { - watch = auth.StartStreamWatch + watch = audit.Watch } - stop := watch(ctx, tokenString, token, func() { audit.Cancel("AuthorizationRevoked"); cancel(); reader.Close(); writer.Close() }) + stop := watch(ctx, tokenString, token, func() { cancel(); reader.Close(); writer.Close() }) defer stop() containerRef, err := client.InspectContainer(container) diff --git a/logs/scoped_logs_integration_test.go b/logs/scoped_logs_integration_test.go index aa1236bc..76a10dad 100644 --- a/logs/scoped_logs_integration_test.go +++ b/logs/scoped_logs_integration_test.go @@ -9,6 +9,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "sync/atomic" "testing" "time" @@ -43,8 +44,14 @@ func TestScopedLogsActualDockerTerminalResults(t *testing.T) { } for _, mode := range []string{"EOF", "disconnect", "revoked", "handshake-denied"} { t.Run(mode, func(t *testing.T) { + containerConfig := &docker.Config{Image: "pasturestack/host-api-log-fixture:latest", Labels: map[string]string{"pasturestack.qa.fixture": "api-key-wp4-20261008"}} + if mode != "EOF" { + // Emit one real Docker-log barrier, then block on an owned FIFO. + // Follow must remain in flight until the handler closes transport. + containerConfig.Cmd = []string{"/bin/sh", "-c", "mkfifo /tmp/scoped-logs-hold && printf 'scoped-logs-follow-ready\n' && exec cat /tmp/scoped-logs-hold"} + } container, err := client.CreateContainer(docker.CreateContainerOptions{Name: "pasturestack-wp4-logs-" + mode, - Config: &docker.Config{Image: "pasturestack/host-api-log-fixture:latest", Labels: map[string]string{"pasturestack.qa.fixture": "api-key-wp4-20261008"}}}) + Config: containerConfig}) if err != nil { t.Fatal(err) } @@ -58,6 +65,7 @@ func TestScopedLogsActualDockerTerminalResults(t *testing.T) { raw := testutils.CreateTokenWithPayload(claims, testutils.ParseTestPrivateKey()) digest := sha256.Sum256([]byte(raw)) terminal := make(chan map[string]string, 1) + var revoked atomic.Bool server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/v2-beta/apiKeyDelegationCompletion" { user, password, ok := r.BasicAuth() @@ -74,13 +82,18 @@ func TestScopedLogsActualDockerTerminalResults(t *testing.T) { w.WriteHeader(403) return } + if revoked.Load() { + w.WriteHeader(http.StatusForbidden) + json.NewEncoder(w).Encode(map[string]string{"code": "ApiKeyPolicyChanged"}) + return + } json.NewEncoder(w).Encode(map[string]interface{}{"allowed": true, "keyId": 12, "revision": 3, "targetType": "container", "targetId": "1i9", "operation": "logs", "expiresAt": expires, "tokenDigest": hex.EncodeToString(digest[:])}) })) defer server.Close() config.Config.PlatformURL = server.URL incoming := make(chan string) defer func() { - if mode != "disconnect" { + if mode != "disconnect" && mode != "revoked" { close(incoming) } }() @@ -93,9 +106,8 @@ func TestScopedLogsActualDockerTerminalResults(t *testing.T) { done := make(chan struct{}) go func() { handler.Handle("fixture", "/v2-beta/logs/?token="+raw, incoming, response); close(done) }() if mode == "disconnect" || mode == "revoked" { - var deny func() select { - case deny = <-watcher: + case <-watcher: case <-time.After(10 * time.Second): t.Fatal("handshake did not reach live watcher") } @@ -108,10 +120,12 @@ func TestScopedLogsActualDockerTerminalResults(t *testing.T) { t.Fatal("Docker output barrier not reached") } if mode == "revoked" { - deny() - } else { - close(incoming) + revoked.Store(true) } + // Proxy-first: the real incoming goroutine records the transport + // cancellation before its pipe/context closes. Finalization then + // rechecks the same signed grant against the denied Engine reply. + close(incoming) } select { case <-done: diff --git a/main.go b/main.go index b2fa7d6d..26e9f5f0 100644 --- a/main.go +++ b/main.go @@ -30,7 +30,7 @@ import ( ) var ( - version = "0.38.5" + version = "0.38.7" showVersion = flag.Bool("version", false, "Print the host API version and exit") ) diff --git a/main_test.go b/main_test.go index e7e31f60..fc80bef5 100644 --- a/main_test.go +++ b/main_test.go @@ -25,7 +25,7 @@ func TestProxyConnectionURLValidatesAndEscapesToken(t *testing.T) { } parsed, _ := url.Parse(result) query := parsed.Query() - if query.Get("token") != "a+b&c" || query.Get("mode") != "host" || query.Get("hostApiVersion") != "0.38.5" || query.Get("hostApiCapabilities") != "key-audit-v1,key-delegation-v1" { + if query.Get("token") != "a+b&c" || query.Get("mode") != "host" || query.Get("hostApiVersion") != version || query.Get("hostApiCapabilities") != "key-audit-v1,key-delegation-v1" { t.Fatal("proxy connection did not preserve authority and declare producer capabilities") } for _, response := range []*platformapi.HostAPIProxyToken{ diff --git a/scripts/build b/scripts/build index bcdc3d3a..883e363b 100755 --- a/scripts/build +++ b/scripts/build @@ -8,7 +8,7 @@ if [ -f ./build/bootstrap.envs ];then . ./build/bootstrap.envs fi -BUILD_VERSION=${VERSION_OVERRIDE:-0.38.6} +BUILD_VERSION=${VERSION_OVERRIDE:-0.38.7} BUILD_VERSION=${BUILD_VERSION#v} if [[ ! "$BUILD_VERSION" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]; then echo 'VERSION_OVERRIDE must use MAJOR.MINOR.PATCH with no suffix' >&2 diff --git a/scripts/package b/scripts/package index 9b4483a3..2eec2165 100755 --- a/scripts/package +++ b/scripts/package @@ -8,7 +8,7 @@ cd "$(dirname "$0")/.." ARCH=${ARCH:?"ARCH not set"} SUFFIX="" [ "${ARCH}" != "amd64" ] && SUFFIX="_${ARCH}" -RELEASE_VERSION=${VERSION_OVERRIDE:-v0.38.6} +RELEASE_VERSION=${VERSION_OVERRIDE:-v0.38.7} RELEASE_VERSION=${RELEASE_VERSION#v} CONTENT=${PASTURESTACK_PACKAGE_ID:-$(git rev-parse --short=32 HEAD)} SOURCE_DATE_EPOCH=${SOURCE_DATE_EPOCH:-$(git show -s --format=%ct HEAD)} diff --git a/scripts/test b/scripts/test index 3c3dbb97..1ad33a7a 100755 --- a/scripts/test +++ b/scripts/test @@ -93,7 +93,7 @@ TEST_CONTAINER_NAME="pasturestack-host-api-test-$$" TEST_CONTAINER=$(docker run -i -d --name "${TEST_CONTAINER_NAME}" --label io.pasturestack.host-api-test=true busybox:1) mkdir -p dist -go test -mod=vendor -race -count=1 -timeout 120s -covermode=atomic \ +PASTURESTACK_QA_LOG_FIXTURE=1 go test -mod=vendor -race -count=1 -timeout 120s -covermode=atomic \ -coverprofile=dist/coverage.out -v ./... go tool cover -func=dist/coverage.out > dist/coverage-summary.txt cat dist/coverage-summary.txt diff --git a/stats/container_stats.go b/stats/container_stats.go index b0e4c5f4..3c065ed2 100644 --- a/stats/container_stats.go +++ b/stats/container_stats.go @@ -73,7 +73,7 @@ func (s *ContainerStatsHandler) Handle(key string, initialMessage string, incomi reader, writer := io.Pipe() defer reader.Close() defer writer.Close() - stop := auth.StartStreamWatch(ctx, tokenString, token, func() { audit.Cancel("AuthorizationRevoked"); cancel(); reader.Close(); writer.Close() }) + stop := audit.Watch(ctx, tokenString, token, func() { cancel(); reader.Close(); writer.Close() }) defer stop() go func(w *io.PipeWriter) { diff --git a/stats/host_stats.go b/stats/host_stats.go index c40c3e9e..69a6d242 100644 --- a/stats/host_stats.go +++ b/stats/host_stats.go @@ -54,7 +54,7 @@ func (s *HostStatsHandler) Handle(key string, initialMessage string, incomingMes reader, writer := io.Pipe() defer reader.Close() defer writer.Close() - stop := auth.StartStreamWatch(context.Background(), tokenString, token, func() { audit.Cancel("AuthorizationRevoked"); reader.Close(); writer.Close() }) + stop := audit.Watch(context.Background(), tokenString, token, func() { reader.Close(); writer.Close() }) defer stop() go func(w *io.PipeWriter) { diff --git a/stats/stats.go b/stats/stats.go index 26ae20af..e8f14bea 100644 --- a/stats/stats.go +++ b/stats/stats.go @@ -71,7 +71,7 @@ func (s *StatsHandler) Handle(key string, initialMessage string, incomingMessage reader, writer := io.Pipe() defer reader.Close() defer writer.Close() - stop := auth.StartStreamWatch(ctx, tokenString, token, func() { audit.Cancel("AuthorizationRevoked"); cancel(); reader.Close(); writer.Close() }) + stop := audit.Watch(ctx, tokenString, token, func() { cancel(); reader.Close(); writer.Close() }) defer stop() go func(w *io.PipeWriter) {