diff --git a/Taskfile.yml b/Taskfile.yml index a61bc39..a63da6e 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -7,8 +7,8 @@ vars: IMAGE: '{{.IMAGE | default "ghcr.io/configbutler/krm-foyer"}}' # The krm-stream browser bundle the hello example vendors, and npm's integrity for # its package. Bump both together, then `task vendor-krm-stream`. - KRM_STREAM_VERSION: 0.6.0 - KRM_STREAM_INTEGRITY: sha512-8vGYVE3np6Ec0bs+Bd0/Y7+PqPp4sQ2YRwbR3Y/VrPPeiIxcHyKt8GHtHa9uUqCp59YXgPALMIJwRh/RXIxhFQ== + KRM_STREAM_VERSION: 0.7.0 + KRM_STREAM_INTEGRITY: sha512-LFXyJIv4UmlBBEJEdx68wzwaqUIVai5HbDe70ACTWIpFZexlJbd/wyhzryoW5uBg55y/f+dZ4bapajybPJ7c0w== VERSION: sh: git describe --tags --always --dirty 2>/dev/null || echo dev diff --git a/docs/design.md b/docs/design.md index 33d3fb4..31f4ccc 100644 --- a/docs/design.md +++ b/docs/design.md @@ -170,7 +170,7 @@ subresources, which return whatever their backend sends: A held-back response is answered as an [interruption](#interruptions): code gets a `Status` with status 502 and the reason, and a person browsing gets a page that explains it. For a redirect, both carry the target; for HTML, the page shows its source -as escaped text. Every interruption is logged. +as escaped text. Every interruption is logged ([how](#interruptions)). ### Interruptions @@ -220,6 +220,13 @@ Rules: API server's answer and reaches the tab as its JSON. Interruptions are only what krm-foyer itself decided. For a 403, [`/_foyer/access`](#what-may-i-do) is where a person finds out why. +- **Every refusal is one log line**, whoever refused: `"msg":"refused"`, with `by` + (`krm-foyer` or `kubernetes`), the `user` as the issuer named them once the request + has a session, the `route`, method and path (never the query), and the reason. That is + an interruption below 500; the API server's 401, 403, 409 or 422 on `/k8s`; and a + stream's `FORBIDDEN`, `UNAUTHENTICATED` or `SCOPE_INVALID`, with the scope asked for. + An interruption from 500 is a failure, not a refusal, and is logged as + `"msg":"interruption"`. What the API server wrote is never logged. - **Permission is a link, never an action.** Following a redirect is a navigation the user starts; nothing is re-sent, and no request with a body is ever continued. An upstream HTML page is never rendered on the origin, with or without consent: consent diff --git a/docs/investigations/krm-stream-shared-watches.md b/docs/investigations/krm-stream-shared-watches.md index 036f021..52550a7 100644 --- a/docs/investigations/krm-stream-shared-watches.md +++ b/docs/investigations/krm-stream-shared-watches.md @@ -9,6 +9,24 @@ that sharing costs. For each we say what krm-foyer does meanwhile. Everything wa checked against krm-stream main at `5ca19ed` (gateway 0.6.0) on 2026-10-04, by reading the source and by tests in krm-foyer that reproduce each problem. +## Status (2026-10-04): asks 9, 10 and 13 are in krm-stream 0.7.0 + +krm-stream took ask 9, the "require a write timeout" form of ask 10, and the documentation +part of ask 13 ([its proposal 0008](https://github.com/ConfigButler/krm-stream/blob/main/docs/proposals/0008-shared-watch-hardening.md)), +and krm-foyer moved to 0.7.0: + +- **Ask 9:** krm-foyer's upstream backend opens the shared watch in `Watch` again; the + workaround that opened it on the first `Next` is gone. The test that reproduced the + problem stays, and fails against 0.6.0. +- **Ask 10:** nothing changed in krm-foyer, which always set a write timeout. A test now + checks that every route in front of a stream lets it flush and set write deadlines. +- **Ask 13:** the decision cache's key cites krm-stream's documented review attributes; the + test that fails if a review starts carrying a selector stays, since the key is + krm-foyer's to get right. + +Asks 11 and 12 are deferred, with what would reopen them recorded in the proposal. +krm-foyer keeps rechecking on the timer and reusing decisions in its own authorizer. + ## The asks at a glance | # | Ask | Priority | What krm-foyer does meanwhile | diff --git a/docs/roadmap.md b/docs/roadmap.md index 78d274c..f9a3e5b 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -77,8 +77,9 @@ Each step makes pending specs real and ends with `task verify` green. opening and every recheck, reused briefly for exactly the same question. Metrics count watches by identity, the streams on shared watches and every access decision. Every stream's writes are bounded, so a browser that stops reading cannot hold off - a recheck, and a shared watch slow to open holds up no other scope; both are asked - of krm-stream as well ([requests](investigations/krm-stream-shared-watches.md)). + a recheck, and a shared watch slow to open holds up no other scope; krm-stream 0.7.0 + took both from krm-foyer's [requests](investigations/krm-stream-shared-watches.md), + so the second is krm-stream's own now. The hello example's notes are shared, and the rehearsal runs both ways. A [decision guide](watches.md) says when to share, stream per-user, or use a native watch, which stays available but is not where the work goes. @@ -295,8 +296,9 @@ Security items need tests that try to get past the boundary. ### Seeing what happened -- [ ] One log line per refusal (policy denial, upstream 401, 403, 409 or 422) with - subject, route and reason +- [x] One log line per refusal (krm-foyer's own, the API server's 401, 403, 409 or 422, + and a stream's refusals) with user, route and reason, and who refused (unit + tests, each checked against a build broken on purpose) - [x] Metrics on every bound, requests in flight, why responses are cut short, and interruptions by reason, on a listener of their own (step 4; see [metrics](bounds.md#metrics)) diff --git a/docs/watches.md b/docs/watches.md index 8e54b05..63bb271 100644 --- a/docs/watches.md +++ b/docs/watches.md @@ -126,7 +126,10 @@ A grant taken away ends a shared stream within: The last line is there because krm-stream delivers events and rechecks one at a time: a recheck waits for the write in progress. Without a bound on writes, a browser that stopped reading would hold its stream open, unchecked, until the 30-minute response -duration. With it, the write fails and the stream ends. +duration. With it, the write fails and the stream ends. Since 0.7.0 krm-stream refuses +timed rechecks without a write timeout, and its +[revocation budget](https://github.com/ConfigButler/krm-stream/blob/main/docs/auth.md#revocation-budget) +breaks down the same parts. Logout and session expiry are not affected: they end every stream within the session-check interval, as before. diff --git a/examples/hello/web/krm-stream.js b/examples/hello/web/krm-stream.js index 725b6c7..6088230 100644 --- a/examples/hello/web/krm-stream.js +++ b/examples/hello/web/krm-stream.js @@ -1108,7 +1108,7 @@ function resourceStreamURL(base, scope) { } // dist/version.js -var VERSION = "0.6.0"; +var VERSION = "0.7.0"; var PROTOCOL_VERSION = 1; export { DEFAULT_EDITABLE_REGIONS, diff --git a/go.mod b/go.mod index 6b30f86..b105546 100644 --- a/go.mod +++ b/go.mod @@ -3,8 +3,8 @@ module github.com/ConfigButler/krm-foyer go 1.27.1 require ( - github.com/ConfigButler/krm-stream/gateway v0.6.0 - github.com/ConfigButler/krm-stream/gateway/kube v0.6.0 + github.com/ConfigButler/krm-stream/gateway v0.7.0 + github.com/ConfigButler/krm-stream/gateway/kube v0.7.0 github.com/chromedp/cdproto v0.0.0-20260714215040-dc233986426f github.com/chromedp/chromedp v0.16.0 github.com/coreos/go-oidc/v3 v3.21.0 diff --git a/go.sum b/go.sum index 333085f..12eabf0 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ -github.com/ConfigButler/krm-stream/gateway v0.6.0 h1:PJk7Q2Ec3DPG3qloAPxhbJq39jlObirYGTREzOK7U2U= -github.com/ConfigButler/krm-stream/gateway v0.6.0/go.mod h1:sVRhasWxXotQQHsyY4mKvvT0npkupDjrvg/i3Elq5+E= -github.com/ConfigButler/krm-stream/gateway/kube v0.6.0 h1:SNEmImC9oJiPxneFA8M9dIvtk1hC2AYw1qsTlCfAAqQ= -github.com/ConfigButler/krm-stream/gateway/kube v0.6.0/go.mod h1:h/WjlrcDwfmyF1Vv4YtzqhOXDvIdjGaOyHkbotU128c= +github.com/ConfigButler/krm-stream/gateway v0.7.0 h1:Pb1fWshxt44D+LMcD7sIrzo3lJQyKtYivIfRuXpBLxI= +github.com/ConfigButler/krm-stream/gateway v0.7.0/go.mod h1:sVRhasWxXotQQHsyY4mKvvT0npkupDjrvg/i3Elq5+E= +github.com/ConfigButler/krm-stream/gateway/kube v0.7.0 h1:8Zm8+il5xgFlRWLWehm1P7e4qSentcMzkRa5AcbqyS4= +github.com/ConfigButler/krm-stream/gateway/kube v0.7.0/go.mod h1:4NPwh1gb4ewfGunozln3G8W4y0//VPirwx+AdeNsmuo= github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0= github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= diff --git a/internal/auth/auth.go b/internal/auth/auth.go index 6e4379b..9b54b1d 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -471,7 +471,11 @@ func (a *Auth) Token(r *http.Request) (gate.Credential, *interruption.Interrupti if err != nil { return gate.Credential{}, a.refusal(err) } - return gate.Credential{Token: s.IDToken, Session: session.Handle(r), Live: func(ctx context.Context) bool { + user := s.Email + if user == "" { + user = s.Subject + } + return gate.Credential{Token: s.IDToken, User: user, Session: session.Handle(r), Live: func(ctx context.Context) bool { err := a.cfg.Sessions.Check(r.WithContext(ctx)) if err != nil && !errors.Is(err, session.ErrNoSession) { a.logger.Warn("session store failed during a session check", "err", err) diff --git a/internal/auth/credentials_test.go b/internal/auth/credentials_test.go index bca0335..4b6f7aa 100644 --- a/internal/auth/credentials_test.go +++ b/internal/auth/credentials_test.go @@ -169,6 +169,10 @@ func TestLiveFollowsTheSession(t *testing.T) { if !cred.Live(t.Context()) { t.Fatal("a live session's credential is not live") } + // Log lines name the user as the issuer did. + if cred.User != alice { + t.Errorf("User = %q, want %q", cred.User, alice) + } // The session's name for bounds is stable, its own, and not its ID. again, _ := h.auth.Token(cookieRequest(b)) other := h.browser() diff --git a/internal/gate/gate.go b/internal/gate/gate.go index 46922b4..fba473c 100644 --- a/internal/gate/gate.go +++ b/internal/gate/gate.go @@ -33,6 +33,10 @@ type Credentials interface { type Credential struct { // Token is the user's bearer token. Token string + // User is who the issuer says the user is, for log lines: their email, or their + // subject when the issuer gave none. The API server decides who the token + // belongs to. + User string // Session names the session the token came from, for the bounds kept per // session. It is opaque, and never logged or sent anywhere. Session string @@ -186,13 +190,59 @@ func (g *Gate) Metrics() *metrics.Metrics { return g.metrics } // Interrupt answers r with i, instead of the API server, and logs and counts it. func (g *Gate) Interrupt(w http.ResponseWriter, r *http.Request, i *interruption.Interruption) { - path, _, _ := strings.Cut(r.RequestURI, "?") - g.logger.Info("interruption", "status", i.Status, "reason", i.Reason, "message", i.Message, - "method", r.Method, "path", path) + g.interrupt(w, r, "", i) +} + +// Interrupt answers a's request with i, as Gate.Interrupt does, and logs its user. +func (a *Admission) Interrupt(w http.ResponseWriter, i *interruption.Interruption) { + a.gate.interrupt(w, a.Request, a.User, i) +} + +// An interruption below 500 is krm-foyer refusing the request, and is logged as a +// refusal; one from 500 is a failure, which the browser may try again. +func (g *Gate) interrupt(w http.ResponseWriter, r *http.Request, user string, i *interruption.Interruption) { + attrs := []any{"status", i.Status, "reason", i.Reason, "message", i.Message} + if i.Status < http.StatusInternalServerError { + g.refused(r, ByFoyer, user, attrs...) + } else { + g.logger.Info("interruption", append(attrs, request(r, user)...)...) + } g.metrics.Interruption(i.Reason) i.Serve(w, r) } +// Who refused a request. +const ( + ByFoyer = "krm-foyer" + ByKubernetes = "kubernetes" +) + +// Refused logs that a's request was refused by, for the reason attrs give. Every +// refusal is one line, whoever refused: an operator finds who was refused what, +// and by whom, in one place. +func (a *Admission) Refused(by string, attrs ...any) { + a.gate.refused(a.Request, by, a.User, attrs...) +} + +func (g *Gate) refused(r *http.Request, by, user string, attrs ...any) { + g.logger.Info("refused", append(append([]any{"by", by}, attrs...), request(r, user)...)...) +} + +// request is what a log line says about r: its method, path and route, and its +// user once known. The query is left out, as it may name anything. +func request(r *http.Request, user string) []any { + path, _, _ := strings.Cut(r.RequestURI, "?") + route := "/" + if first, _, _ := strings.Cut(strings.TrimPrefix(path, "/"), "/"); first != "" { + route += first + } + attrs := []any{"method", r.Method, "path", path, "route", route} + if user != "" { + attrs = append(attrs, "user", user) + } + return attrs +} + // Admission is a request the gate let through. Close it when the response has // ended, exactly once. type Admission struct { @@ -233,12 +283,12 @@ func (g *Gate) admit(w http.ResponseWriter, r *http.Request, slots *concurrency) return nil } if ok, wait := g.rate.allow(cred.Session); !ok { - g.Interrupt(w, r, g.rate.refuse(wait)) + g.interrupt(w, r, cred.User, g.rate.refuse(wait)) return nil } release, refused := slots.acquire(cred.Session) if refused != nil { - g.Interrupt(w, r, refused) + g.interrupt(w, r, cred.User, refused) return nil } diff --git a/internal/proxy/interruption_test.go b/internal/proxy/interruption_test.go index 8a0e447..8bc3755 100644 --- a/internal/proxy/interruption_test.go +++ b/internal/proxy/interruption_test.go @@ -268,17 +268,70 @@ func TestSignInReturnsToTheURL(t *testing.T) { } } -// Every interruption is logged with the path that was asked for, including the -// ones decided after the API server answered. +// Every interruption is logged once, with the path that was asked for, including +// the ones decided after the API server answered. One below 500 is krm-foyer +// refusing the request, logged as a refusal like the API server's; one from 500 is +// a failure. The user is named once the request has one. func TestInterruptionsAreLoggedWithThePath(t *testing.T) { for _, tc := range interruptionCases() { t.Run(tc.name, func(t *testing.T) { f := tc.foyer(t) f.request(t, http.MethodGet, tc.target, nil, tc.header) path, _, _ := strings.Cut(tc.target, "?") - if logs := f.logs.String(); !strings.Contains(logs, `"msg":"interruption"`) || !strings.Contains(logs, `"path":"`+path+`"`) { - t.Errorf("logged:\n%s", logs) + want := map[string]any{"msg": "interruption", "path": path, "route": "/k8s", "status": float64(tc.status)} + if tc.status < http.StatusInternalServerError { + want["msg"], want["by"] = "refused", gate.ByFoyer + } + lines := logLines(t, f.logs.String(), want) + if len(lines) != 1 { + t.Fatalf("%d lines like %v; logged:\n%s", len(lines), want, f.logs.String()) + } + // Past the credential: the bound counts per session, and the API + // server's answers come after the gate. + admitted := tc.occupy || tc.upstream != nil + if user, ok := lines[0]["user"]; admitted != ok || (ok && user != testUser) { + t.Errorf("user = %v; want it named only past the credential", user) } }) } } + +// The API server refusing the user is logged once, as krm-foyer's own refusals +// are: who, what, and why. Any other answer is not a refusal, and is not logged. +func TestKubernetesRefusalsAreLogged(t *testing.T) { + api := newAPIServer(t, func(w http.ResponseWriter, r *http.Request) { + status, _ := strconv.Atoi(r.URL.Query().Get("answer")) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = io.WriteString(w, `{"kind":"Status","message":"`+userToken+`"}`) + }) + f := newFoyer(t, api, credentials{token: userToken}) + for status, reason := range map[int]string{ + http.StatusUnauthorized: "Unauthorized", http.StatusForbidden: "Forbidden", + http.StatusConflict: "Conflict", http.StatusUnprocessableEntity: "Unprocessable Entity", + http.StatusOK: "", http.StatusNotFound: "", http.StatusTooManyRequests: "", http.StatusInternalServerError: "", + } { + before := len(logLines(t, f.logs.String(), nil)) + target := "/k8s/api/v1/namespaces/a/configmaps?answer=" + strconv.Itoa(status) + if resp := f.get(t, target); resp.StatusCode != status { + t.Fatalf("GET %s = %d", target, resp.StatusCode) + } + lines := logLines(t, f.logs.String(), nil)[before:] + if reason == "" { + if len(lines) != 0 { + t.Errorf("%d is not a refusal, but logged %v", status, lines) + } + continue + } + want := map[string]any{ + "msg": "refused", "by": gate.ByKubernetes, "user": testUser, "status": float64(status), "reason": reason, + "method": http.MethodGet, "path": "/k8s/api/v1/namespaces/a/configmaps", "route": "/k8s", + } + if len(lines) != 1 || len(logLines(t, f.logs.String(), want)) != 1 { + t.Errorf("%d: logged %v; want one line like %v", status, lines, want) + } + } + if strings.Contains(f.logs.String(), userToken) { + t.Errorf("what the API server wrote reached the log:\n%s", f.logs.String()) + } +} diff --git a/internal/proxy/lifecycle_test.go b/internal/proxy/lifecycle_test.go index 874a79d..d80c571 100644 --- a/internal/proxy/lifecycle_test.go +++ b/internal/proxy/lifecycle_test.go @@ -280,5 +280,5 @@ func TestUpstreamErrorAfterACutAborts(t *testing.T) { t.Errorf("something was written before the abort: %d %q", w.Code, w.Body) } }() - p.upstreamError(w, r, context.Canceled) + _ = p.upstreamError(r, context.Canceled) } diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 51b5f9f..4a22e4c 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -165,15 +165,30 @@ func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { if err := checkResponse(resp); err != nil { return err } + if refusals[resp.StatusCode] { + a.Refused(gate.ByKubernetes, "status", resp.StatusCode, "reason", http.StatusText(resp.StatusCode)) + } return p.limitBytes(a, resp) }, // ReverseProxy hands its error handler the outgoing request, whose headers // are the allowlisted ones. How to answer depends on the browser's request. - ErrorHandler: func(w http.ResponseWriter, _ *http.Request, err error) { p.upstreamError(w, r, err) }, + ErrorHandler: func(w http.ResponseWriter, _ *http.Request, err error) { + if refused := p.upstreamError(r, err); refused != nil { + a.Interrupt(w, refused) + } + }, } rp.ServeHTTP(&headWriter{ResponseWriter: w}, r) } +// refusals are the API server's answers that refuse the user, rather than fail: +// not signed in, not allowed, in conflict, or invalid. Each is logged, as +// krm-foyer's own refusals are. +var refusals = map[int]bool{ + http.StatusUnauthorized: true, http.StatusForbidden: true, + http.StatusConflict: true, http.StatusUnprocessableEntity: true, +} + // checkResponse applies the upstream response rules (docs/design.md). An error it // returns is an *interruption.Interruption, answered by upstreamError. func checkResponse(resp *http.Response) error { @@ -253,24 +268,26 @@ func heldBack(field, value string) *interruption.Interruption { } } -func (p *Proxy) upstreamError(w http.ResponseWriter, r *http.Request, err error) { +// upstreamError is the interruption to answer r with for err, or nil when nobody +// is left to answer. It panics with http.ErrAbortHandler when the gate cut r short. +func (p *Proxy) upstreamError(r *http.Request, err error) *interruption.Interruption { var refused *interruption.Interruption - if !errors.As(err, &refused) { - if gate.CutShort(r.Context()) { - // The request may have reached Kubernetes, so no interruption may say it - // did not, and an empty answer would pass for a complete one. - panic(http.ErrAbortHandler) - } - if r.Context().Err() != nil { - return // the browser went away; nobody is left to answer - } - p.logger.Warn("API server unreachable", "cause", upstream.Class(err)) - refused = &interruption.Interruption{ - Status: http.StatusBadGateway, Reason: "BadGateway", - Message: "the API server could not be reached", - } + if errors.As(err, &refused) { + return refused + } + if gate.CutShort(r.Context()) { + // The request may have reached Kubernetes, so no interruption may say it + // did not, and an empty answer would pass for a complete one. + panic(http.ErrAbortHandler) + } + if r.Context().Err() != nil { + return nil // the browser went away; nobody is left to answer + } + p.logger.Warn("API server unreachable", "cause", upstream.Class(err)) + return &interruption.Interruption{ + Status: http.StatusBadGateway, Reason: "BadGateway", + Message: "the API server could not be reached", } - p.interrupt(w, r, refused) } func (p *Proxy) interrupt(w http.ResponseWriter, r *http.Request, i *interruption.Interruption) { diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index ddbbcf1..12483b0 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -24,6 +24,9 @@ import ( const userToken = "user-token-4f1c9a" //nolint:gosec // a marker to search for, not a credential +// testUser is who the issuer says the test's user is. +const testUser = "alice@example.test" + // credentials is the test's session: it hands the proxy a fixed answer. It exists // only in this test file; the binary has no way to be given a token but a session. // Its session stays live unless live says otherwise. @@ -44,7 +47,30 @@ func (c credentials) Token(*http.Request) (gate.Credential, *interruption.Interr if session == "" { session = "s1" } - return gate.Credential{Token: c.token, Live: live, Session: session}, c.refused + return gate.Credential{Token: c.token, User: testUser, Live: live, Session: session}, c.refused +} + +// logLines is what krm-foyer logged, a line at a time, for those that match all of +// want. +func logLines(t *testing.T, logs string, want map[string]any) []map[string]any { + t.Helper() + var found []map[string]any + for line := range strings.Lines(logs) { + var l map[string]any + if err := json.Unmarshal([]byte(line), &l); err != nil { + continue // not krm-foyer's: Go's own, in plain text + } + matches := true + for k, v := range want { + if got, ok := l[k]; !ok || got != v { + matches = false + } + } + if matches { + found = append(found, l) + } + } + return found } // apiServer stands in for the API server and records what reached it. @@ -865,8 +891,8 @@ func TestTokenNeverLeaves(t *testing.T) { if strings.Contains(f.logs.String(), userToken) { t.Error("the token is in krm-foyer's logs") } - if !strings.Contains(f.logs.String(), `"msg":"interruption"`) { - t.Error("interruptions are not logged") + if !strings.Contains(f.logs.String(), `"msg":"refused"`) || !strings.Contains(f.logs.String(), `"msg":"interruption"`) { + t.Error("refusals and interruptions are not logged") } } diff --git a/internal/server/server_test.go b/internal/server/server_test.go index ba4c959..d205c48 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -5,6 +5,8 @@ import ( "net/http/httptest" "strings" "testing" + + "github.com/ConfigButler/krm-stream/gateway" ) func get(t *testing.T, path string) *httptest.ResponseRecorder { @@ -182,3 +184,38 @@ func TestStreamRoute(t *testing.T) { t.Errorf("without streams, /stream/v1 = %d", rec.Code) } } + +// A stream gets a response writer it can flush and give write deadlines, over +// HTTP/1.1 and HTTP/2, through every route krm-foyer puts in front of it. The write +// timeout that bounds a stalled browser, and with it a shared stream's recheck, +// depends on both; a wrapper that hid them would end every stream at once. +func TestStreamsCanFlushAndSetWriteDeadlines(t *testing.T) { + got := make(chan error, 1) + h := New(Config{Version: "test", Kubernetes: http.NotFoundHandler(), Stream: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + got <- gateway.CheckHTTPStreaming(w) + })}) + h1 := httptest.NewServer(h) + t.Cleanup(h1.Close) + h2 := httptest.NewUnstartedServer(h) + h2.EnableHTTP2 = true + h2.StartTLS() + t.Cleanup(h2.Close) + + for name, srv := range map[string]*httptest.Server{"HTTP/1.1": h1, "HTTP/2": h2} { + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, srv.URL+"/stream/v1?resource=notes&version=v1", nil) + if err != nil { + t.Fatal(err) + } + resp, err := srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if want := name[len("HTTP/")] - '0'; resp.ProtoMajor != int(want) { + t.Errorf("%s: served over %s", name, resp.Proto) + } + if err := <-got; err != nil { + t.Errorf("%s: %v", name, err) + } + } +} diff --git a/internal/stream/shared.go b/internal/stream/shared.go index 01f4cce..2f1c7f6 100644 --- a/internal/stream/shared.go +++ b/internal/stream/shared.go @@ -292,11 +292,13 @@ func (d *decisions) Authorize(ctx context.Context, p gateway.Principal, scope ga if err != nil { return gateway.Unauthenticated("not authenticated") } - // The question is what the reviews ask: the subject, and the group, version, - // resource, namespace and name of the scope. A label selector is not part of it - // (RBAC cannot grant by label, and the reviews do not send one), so streams that - // differ only in their selector share a decision. TestASharedStreamAsksAboutEachUser - // fails if the reviews start sending one. + // The question is what the reviews ask, as krm-stream documents it + // (docs/auth.md, "What the SubjectAccessReview asks"): the subject, and the + // group, version, resource, namespace and name of the scope, for list and then + // watch. One decision answers both verbs, as Authorize asks both. Selectors and + // subresource are not sent, so streams that differ only in their selector share + // a decision; TestTheReviewsAskAboutNoSelector fails if that changes. There is + // one target, so decisions need no keeping apart by API server. key, err := json.Marshal(struct { Subject kube.Subject Group, Version, Resource, Namespace, Name string @@ -393,47 +395,28 @@ type sharedUpstream struct { metrics *metrics.Metrics } -// Watch returns at once, and the watch is opened on the first Next. krm-stream calls -// Watch holding a lock every shared scope takes, and with a context nobody cancels -// but the scope's own: opened here, one API server slow to answer would hold up the -// streams of every other scope, and a stream that gave up waiting could not end the -// request. Opened in Next, it is the scope's alone, and its last stream leaving -// cancels it (docs/investigations/krm-stream-shared-watches.md, ask 9). +// Watch opens the scope's watch. krm-stream opens it outside the lock its scopes +// share, with a context its waiting streams share, so a slow opening holds up only +// its own scope, and its last stream leaving cancels it (krm-stream 0.7.0, from +// docs/investigations/krm-stream-shared-watches.md, ask 9). func (b *sharedUpstream) Watch(ctx context.Context, scope gateway.Scope) (gateway.Watcher, error) { - return &sharedWatcher{upstream: b, ctx: ctx, scope: scope}, nil + open, err := b.kube.Watch(ctx, scope) + if err != nil { + return nil, ownRefusal(err) + } + return &sharedWatcher{Watcher: open, done: b.metrics.UpstreamWatch(metrics.IdentityShared)}, nil } +// sharedWatcher counts its watch until it is stopped, and reports the shared +// identity being refused as krm-foyer's own fault. type sharedWatcher struct { - upstream *sharedUpstream - ctx context.Context // the scope's: cancelled when its last stream leaves - scope gateway.Scope - - mu sync.Mutex - open gateway.Watcher // nil until opened - done func() - stopped bool + gateway.Watcher + stop sync.Once + done func() } func (w *sharedWatcher) Next(ctx context.Context) (gateway.WatchEvent, error) { - w.mu.Lock() - open := w.open - w.mu.Unlock() - if open == nil { - opened, err := w.upstream.kube.Watch(w.ctx, w.scope) - if err != nil { - return gateway.WatchEvent{}, ownRefusal(err) - } - w.mu.Lock() - if w.stopped { - w.mu.Unlock() - opened.Stop() - return gateway.WatchEvent{}, context.Canceled - } - w.open, w.done = opened, w.upstream.metrics.UpstreamWatch(metrics.IdentityShared) - open = opened - w.mu.Unlock() - } - ev, err := open.Next(ctx) + ev, err := w.Watcher.Next(ctx) if ev.Type == gateway.WatchError { ev.Err = ownRefusal(ev.Err) } @@ -441,16 +424,10 @@ func (w *sharedWatcher) Next(ctx context.Context) (gateway.WatchEvent, error) { } func (w *sharedWatcher) Stop() { - w.mu.Lock() - defer w.mu.Unlock() - if w.stopped { - return - } - w.stopped = true - if w.open != nil { - w.open.Stop() + w.stop.Do(func() { + w.Watcher.Stop() w.done() - } + }) } // errSharedRefused is the cause of an INTERNAL sent for the shared identity being diff --git a/internal/stream/shared_test.go b/internal/stream/shared_test.go index 1aed0ff..f3de8ab 100644 --- a/internal/stream/shared_test.go +++ b/internal/stream/shared_test.go @@ -334,6 +334,9 @@ func TestASharedStreamAsksAboutEachUser(t *testing.T) { if strings.Contains(carol.raw.String(), "note first") { t.Errorf("carol was sent the shared watch's notes: %s", carol.raw.String()) } + if refusals := logLines(f.logs.String(), map[string]any{"msg": "refused", "by": gate.ByKubernetes, "code": "FORBIDDEN"}); len(refusals) != 1 { + t.Errorf("carol's refusal logged %d times as Kubernetes':\n%s", len(refusals), f.logs.String()) + } var carols []authorizationv1.SubjectAccessReviewSpec for _, spec := range api.subjectReviews() { @@ -360,8 +363,9 @@ func TestASharedStreamAsksAboutEachUser(t *testing.T) { } // Decisions are reused across label selectors because the reviews ask about none -// (see decisions.Authorize). This fails if they start to, for a stream that asks for -// a selector: the key must then take it. +// (see decisions.Authorize). krm-stream documents that since 0.7.0 and announces a +// change to it; this fails if one comes anyway, for a stream that asks for a +// selector: the key must then take it. func TestTheReviewsAskAboutNoSelector(t *testing.T) { api := newSharedAPI(t) f := sharedFoyer(t, api, byHeader{}, nil, time.Minute) @@ -373,9 +377,9 @@ func TestTheReviewsAskAboutNoSelector(t *testing.T) { t.Fatalf("a review without resource attributes: %+v", spec) } verbs[attrs.Verb] = true - if attrs.LabelSelector != nil || attrs.FieldSelector != nil { - t.Errorf("the %s review asks about selectors (%+v, %+v): key decisions on them", - attrs.Verb, attrs.LabelSelector, attrs.FieldSelector) + if attrs.LabelSelector != nil || attrs.FieldSelector != nil || attrs.Subresource != "" { + t.Errorf("the %s review asks about selectors or a subresource (%+v, %+v, %q): key decisions on them", + attrs.Verb, attrs.LabelSelector, attrs.FieldSelector, attrs.Subresource) } } if !verbs["list"] || !verbs["watch"] { @@ -774,8 +778,13 @@ func TestAStuckOpeningHoldsUpNoOtherScope(t *testing.T) { return } stuck <- struct{}{} - <-r.Context().Done() // no answer, not even headers, until krm-foyer gives up - released <- struct{}{} + // No answer, not even headers, until krm-foyer gives up; or until the test + // ends, so a krm-foyer that never gives up fails it rather than hangs it. + select { + case <-r.Context().Done(): + released <- struct{}{} + case <-t.Context().Done(): + } } m := metrics.New() f := sharedFoyer(t, api, byHeader{}, m, time.Minute) diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 82c0a99..08d9a42 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -21,6 +21,7 @@ import ( "github.com/ConfigButler/krm-stream/gateway" "github.com/ConfigButler/krm-stream/gateway/kube" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/client-go/rest" "github.com/ConfigButler/krm-foyer/internal/gate" @@ -136,7 +137,7 @@ func (s *Streams) ServeHTTP(w http.ResponseWriter, r *http.Request) { // unnamed target, the one cluster, exists. Scopes: gateway.ScopePolicy{Targets: []string{""}, AnyResource: true, AllowLabelSelector: true}, Projections: gateway.ProjectionPolicyFunc(project), - Diagnostics: s.diagnose, + Diagnostics: func(d gateway.Diagnostic) { s.diagnose(a, d) }, // Every write to the browser is bounded, so one that stopped reading ends its // stream rather than holding it, and a shared stream's rechecks, which wait // for the write in progress. @@ -181,11 +182,27 @@ func project(_ context.Context, _ gateway.Principal, _ gateway.Scope, requested // diagnose logs what went wrong behind an error the browser was sent: its code and // the kind of failure, never its text, which the API server or whatever answered -// instead wrote (see package upstream). A recovered resync is routine. -func (s *Streams) diagnose(d gateway.Diagnostic) { - if d.Code == gateway.CodeResyncRequired { +// instead wrote (see package upstream). A refusal is logged as every refusal is: by +// Kubernetes when it did not let the user list and watch, or did not accept their +// token; by krm-foyer when the scope is not one it serves. A recovered resync is +// routine. +func (s *Streams) diagnose(a *gate.Admission, d gateway.Diagnostic) { + switch d.Code { + case gateway.CodeResyncRequired: s.logger.Debug("stream resynchronised", "cause", class(d.Err)) return + case gateway.CodeForbidden, gateway.CodeUnauthenticated, gateway.CodeScopeInvalid: + // FORBIDDEN and UNAUTHENTICATED are Kubernetes' answers: a 403 or 401, or a + // SubjectAccessReview's denial. SCOPE_INVALID is the API server's only when it + // does not serve the resource; otherwise the gateway refused the scope. + var status apierrors.APIStatus + by := gate.ByFoyer + if d.Code != gateway.CodeScopeInvalid || errors.As(d.Err, &status) { + by = gate.ByKubernetes + } + a.Refused(by, "code", string(d.Code), "group", d.Scope.Group, "resource", d.Scope.Resource, + "namespace", d.Scope.Namespace, "name", d.Scope.Name) + return } s.logger.Warn("API server could not serve a stream", "code", string(d.Code), "terminal", d.Terminal, "cause", class(d.Err)) diff --git a/internal/stream/stream_test.go b/internal/stream/stream_test.go index b3674a9..ad53506 100644 --- a/internal/stream/stream_test.go +++ b/internal/stream/stream_test.go @@ -51,9 +51,12 @@ func (c credentials) Token(*http.Request) (gate.Credential, *interruption.Interr if session == "" { session = "s1" } - return gate.Credential{Token: c.token, Session: session, Live: live}, c.refused + return gate.Credential{Token: c.token, User: testUser, Session: session, Live: live}, c.refused } +// testUser is who the issuer says every test user is. +const testUser = "alice@example.test" + // byHeader hands each request the token named by its Test-Token header, as if each // came from a session of its own. type byHeader struct{} @@ -468,6 +471,33 @@ func TestScopesTheGatewayRefuses(t *testing.T) { if n := len(api.received()); n != 0 { t.Errorf("%d requests reached the API server", n) } + refusals := logLines(f.logs.String(), map[string]any{"msg": "refused", "by": gate.ByFoyer, + "code": "SCOPE_INVALID", "user": testUser, "route": "/stream", "path": "/stream/v1"}) + if len(refusals) != 7 { + t.Errorf("%d refusals logged by krm-foyer, want 7:\n%s", len(refusals), f.logs.String()) + } +} + +// logLines is what krm-foyer logged, a line at a time, for those that match all of +// want. +func logLines(logs string, want map[string]any) []map[string]any { + var found []map[string]any + for line := range strings.Lines(logs) { + var l map[string]any + if json.Unmarshal([]byte(line), &l) != nil { + continue + } + matches := true + for k, v := range want { + if l[k] != v { + matches = false + } + } + if matches { + found = append(found, l) + } + } + return found } // Each built-in projection may be asked for: a projection hides nothing from a @@ -519,6 +549,12 @@ func TestKubernetesRefusalsKeepTheirMeaning(t *testing.T) { if host := strings.TrimPrefix(api.URL, "https://"); strings.Contains(s.raw.String(), host) { t.Errorf("the API server's address reached the browser: %s", s.raw.String()) } + // Logged once, as every refusal is: Kubernetes' own, and what was asked for. + want := map[string]any{"msg": "refused", "by": gate.ByKubernetes, "code": tc.code, "user": testUser, + "route": "/stream", "group": "hello.krm-foyer.example", "resource": "notes", "namespace": "hello"} + if n := len(logLines(f.logs.String(), want)); n != 1 || len(logLines(f.logs.String(), map[string]any{"msg": "refused"})) != 1 { + t.Errorf("want one line like %v; logged:\n%s", want, f.logs.String()) + } }) } }