feat: build streams on krm-stream 0.7.0, and log every refusal once with its user

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
 

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

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 |

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))

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.

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,

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

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=

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)

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()

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
 	}
 

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())
+	}
+}

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)
 }

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) {