test(19.7-09): gate workspace check and unreleased-pin policy, TestSessionReaper and TestAuthTime
- check-phase19.7.sh: workspace_check (framework to this checkout, the OAuth server plugin's user plugin to USER_PLUGIN) in the plugins, consumers and docker stages; consumer_with_user_plugin builds fonoteka.go and sm-bm-app against USER_PLUGIN through a scratch go.work; demo_pins accepts unreleased commits its canonical checkout holds and refuses any drift with PHASE197_RELEASED=1; self-test cases for each, one PASS line each - TestSessionReaper pins A-CR-01, A-IN-01 and A-IN-06 (memory and Redis stores, store errors, the GET stream re-check, the ephemeral label) - TestAuthTime pins auth_time at mint and through every refresh path - GO_TESTS and GO_TAGGED_TESTS require them with the plan 06/07 smokes
This commit is contained in:
182
modules/bouncer/auth_time_test.go
Normal file
182
modules/bouncer/auth_time_test.go
Normal file
@@ -0,0 +1,182 @@
|
||||
package bouncer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
)
|
||||
|
||||
// TestAuthTime pins the auth_time claim of plan 07 (P-29, C-CR-02's
|
||||
// framework half): every sign-in mint stamps it equal to iat for both
|
||||
// audiences, every refresh path copies it unchanged across rotations, a
|
||||
// token without it refreshes without it, and VerifyAuthTime keeps the
|
||||
// checks of VerifyClaims and VerifyClaimsAudience.
|
||||
func TestAuthTime(t *testing.T) {
|
||||
const iss = "https://app.test/_user/api/v1/login"
|
||||
const refreshIss = "https://app.test/_user/api/v1/refresh"
|
||||
|
||||
if ClaimAuthTime != "auth_time" {
|
||||
t.Fatalf("ClaimAuthTime = %q, want auth_time", ClaimAuthTime)
|
||||
}
|
||||
|
||||
t.Run("mint stamps auth_time equal to iat for both audiences", func(t *testing.T) {
|
||||
for _, aud := range []string{AudienceUser, AudienceBackend} {
|
||||
tok, _, err := MintAudience(secret, "42", iss, time.Hour, aud)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
claims := decodeClaims(t, tok, secret)
|
||||
at := claimUnix(t, claims, ClaimAuthTime)
|
||||
iat := claimUnix(t, claims, "iat")
|
||||
if at.Unix() != iat.Unix() {
|
||||
t.Fatalf("%s: auth_time %d != iat %d", aud, at.Unix(), iat.Unix())
|
||||
}
|
||||
_, got, ok, err := VerifyAuthTime(tok, secret, aud)
|
||||
if err != nil || !ok || got.Unix() != iat.Unix() {
|
||||
t.Fatalf("%s: VerifyAuthTime = %v, %v, %v", aud, got, ok, err)
|
||||
}
|
||||
}
|
||||
tok, _, err := Mint(secret, "42", iss, time.Hour)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, _, ok, err := VerifyAuthTime(tok, secret, AudienceUser); err != nil || !ok {
|
||||
t.Fatalf("Mint: ok=%v err=%v", ok, err)
|
||||
}
|
||||
})
|
||||
|
||||
signedAt := time.Now().Add(-90 * time.Minute).Truncate(time.Second)
|
||||
users := memUsers{byID: map[uint]*Principal{42: {ID: 42, Backend: true}}}
|
||||
refreshers := []struct {
|
||||
name string
|
||||
audience string
|
||||
refresh func(tok string) (string, error)
|
||||
}{
|
||||
{"Refresh", AudienceUser, func(tok string) (string, error) {
|
||||
return Refresh(secret, tok, 4*time.Hour, NewMemoryBlacklist(), 0, refreshIss)
|
||||
}},
|
||||
{"RefreshAudience", AudienceBackend, func(tok string) (string, error) {
|
||||
return RefreshAudience(secret, tok, AudienceBackend, 4*time.Hour, NewMemoryBlacklist(), 0, refreshIss)
|
||||
}},
|
||||
{"RefreshAudienceFor", AudienceBackend, func(tok string) (string, error) {
|
||||
return RefreshAudienceFor(context.Background(), users, secret, tok, AudienceBackend, 4*time.Hour, NewMemoryBlacklist(), 0, refreshIss)
|
||||
}},
|
||||
}
|
||||
for _, r := range refreshers {
|
||||
t.Run(r.name+" carries auth_time unchanged across two rotations", func(t *testing.T) {
|
||||
tok, _, err := mintAudience(secret, "42", iss, 30*time.Minute, r.audience, signedAt, signedAt)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
prevIat := signedAt
|
||||
for i := 1; i <= 2; i++ {
|
||||
next, err := r.refresh(tok)
|
||||
if err != nil {
|
||||
t.Fatalf("rotation %d: %v", i, err)
|
||||
}
|
||||
_, at, ok, err := VerifyAuthTime(next, secret, r.audience)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("rotation %d: ok=%v err=%v", i, ok, err)
|
||||
}
|
||||
if at.Unix() != signedAt.Unix() {
|
||||
t.Fatalf("rotation %d: auth_time %d, want %d", i, at.Unix(), signedAt.Unix())
|
||||
}
|
||||
iat := claimUnix(t, decodeClaims(t, next, secret), "iat")
|
||||
if !iat.After(prevIat) && i == 1 {
|
||||
t.Fatalf("rotation %d: iat %s did not move past %s", i, iat, prevIat)
|
||||
}
|
||||
prevIat = iat
|
||||
tok = next
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("a legacy token without auth_time refreshes without one", func(t *testing.T) {
|
||||
legacy := sign(t, jwt.SigningMethodHS256, jwt.MapClaims{
|
||||
"sub": "42", "aud": AudienceBackend, "jti": "legacy-backend",
|
||||
"iat": time.Now().Add(-10 * time.Minute).Unix(),
|
||||
"exp": time.Now().Add(time.Minute).Unix(),
|
||||
}, []byte(secret))
|
||||
for name, refresh := range map[string]func(string) (string, error){
|
||||
"RefreshAudience": func(tok string) (string, error) {
|
||||
return RefreshAudience(secret, tok, AudienceBackend, time.Hour, NewMemoryBlacklist(), 0, refreshIss)
|
||||
},
|
||||
"RefreshAudienceFor": func(tok string) (string, error) {
|
||||
return RefreshAudienceFor(context.Background(), users, secret, tok, AudienceBackend, time.Hour, NewMemoryBlacklist(), 0, refreshIss)
|
||||
},
|
||||
} {
|
||||
next, err := refresh(legacy)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", name, err)
|
||||
}
|
||||
if _, has := decodeClaims(t, next, secret)[ClaimAuthTime]; has {
|
||||
t.Fatalf("%s stamped auth_time on a legacy token", name)
|
||||
}
|
||||
if _, _, ok, err := VerifyAuthTime(next, secret, AudienceBackend); err != nil || ok {
|
||||
t.Fatalf("%s: VerifyAuthTime ok=%v err=%v, want ok=false and no error", name, ok, err)
|
||||
}
|
||||
}
|
||||
user := sign(t, jwt.SigningMethodHS256, jwt.MapClaims{
|
||||
"sub": "42", "jti": "legacy-user",
|
||||
"iat": time.Now().Add(-10 * time.Minute).Unix(),
|
||||
"exp": time.Now().Add(time.Minute).Unix(),
|
||||
}, []byte(secret))
|
||||
next, err := Refresh(secret, user, time.Hour, NewMemoryBlacklist(), 0, refreshIss)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, has := decodeClaims(t, next, secret)[ClaimAuthTime]; has {
|
||||
t.Fatal("Refresh stamped auth_time on a legacy token")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("VerifyAuthTime refuses what VerifyClaims refuses", func(t *testing.T) {
|
||||
good, _, err := MintAudience(secret, "42", iss, time.Hour, AudienceUser)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, _, _, err := VerifyAuthTime(good, "another-secret", AudienceUser); err == nil {
|
||||
t.Fatal("bad signature accepted")
|
||||
}
|
||||
if _, _, _, err := VerifyAuthTime(good, secret, AudienceBackend); err == nil {
|
||||
t.Fatal("wrong audience accepted")
|
||||
}
|
||||
backend, _, err := MintAudience(secret, "42", iss, time.Hour, AudienceBackend)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, _, _, err := VerifyAuthTime(backend, secret, AudienceUser); err == nil {
|
||||
t.Fatal("a backend token accepted for the frontend audience")
|
||||
}
|
||||
past := time.Now().Add(-time.Hour)
|
||||
expired, _, err := mintAudience(secret, "42", iss, time.Minute, AudienceUser, past, past)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, _, _, err := VerifyAuthTime(expired, secret, AudienceUser); err == nil {
|
||||
t.Fatal("expired token accepted")
|
||||
}
|
||||
if _, _, _, err := VerifyAuthTime(good, secret, " "); err == nil {
|
||||
t.Fatal("empty audience accepted")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a frontend token without aud is accepted as VerifyClaims does", func(t *testing.T) {
|
||||
at := time.Now().Add(-5 * time.Minute).Unix()
|
||||
noAud := sign(t, jwt.SigningMethodHS256, jwt.MapClaims{
|
||||
"sub": "42", "jti": "no-aud",
|
||||
"iat": time.Now().Unix(),
|
||||
"exp": time.Now().Add(time.Hour).Unix(),
|
||||
ClaimAuthTime: at,
|
||||
}, []byte(secret))
|
||||
if _, _, _, _, err := VerifyClaims(noAud, secret); err != nil {
|
||||
t.Fatalf("VerifyClaims refused the no-aud token: %v", err)
|
||||
}
|
||||
sub, got, ok, err := VerifyAuthTime(noAud, secret, AudienceUser)
|
||||
if err != nil || !ok || sub != "42" || got.Unix() != at {
|
||||
t.Fatalf("VerifyAuthTime(no aud) = %q, %v, %v, %v", sub, got, ok, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -67,3 +67,21 @@ func (p *Pier) StopReaper() {
|
||||
p.stopReap()
|
||||
}
|
||||
}
|
||||
|
||||
// SessionDone answers a channel closed once the local session's SDK server
|
||||
// session has ended (ss.Wait returned), and false when the session is not
|
||||
// connected on this instance.
|
||||
func (p *Pier) SessionDone(sid string) (<-chan struct{}, bool) {
|
||||
p.sessions.mu.Lock()
|
||||
ls := p.sessions.local[sid]
|
||||
p.sessions.mu.Unlock()
|
||||
if ls == nil || ls.ss == nil {
|
||||
return nil, false
|
||||
}
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
_ = ls.ss.Wait()
|
||||
close(done)
|
||||
}()
|
||||
return done, true
|
||||
}
|
||||
|
||||
249
modules/pier/reaper_test.go
Normal file
249
modules/pier/reaper_test.go
Normal file
@@ -0,0 +1,249 @@
|
||||
package pier_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/pact"
|
||||
)
|
||||
|
||||
// logBuffer is a goroutine-safe slog destination for audit-line checks.
|
||||
type logBuffer struct {
|
||||
mu sync.Mutex
|
||||
buf bytes.Buffer
|
||||
}
|
||||
|
||||
func (b *logBuffer) Write(p []byte) (int, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.Write(p)
|
||||
}
|
||||
|
||||
func (b *logBuffer) String() string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.String()
|
||||
}
|
||||
|
||||
// captureLogs routes the process logger (and so diary's pier component)
|
||||
// into a buffer for the rest of the test.
|
||||
func captureLogs(t *testing.T) *logBuffer {
|
||||
t.Helper()
|
||||
buf := &logBuffer{}
|
||||
previous := slog.Default()
|
||||
slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})))
|
||||
t.Cleanup(func() { slog.SetDefault(previous) })
|
||||
return buf
|
||||
}
|
||||
|
||||
// TestSessionReaper pins the session reaper of plan 06 (A-CR-01, A-IN-01,
|
||||
// A-IN-06): a stateful session whose store record expired is closed on the
|
||||
// instance holding it (its SDK session ends, the id answers 404, the audit
|
||||
// line says expired); on a shared Redis store an instance keeps a copy only
|
||||
// while the record lives; a store error never closes a copy; a GET stream
|
||||
// whose record vanished ends through the re-check; and a tool call served
|
||||
// without a session logs the session as ephemeral.
|
||||
func TestSessionReaper(t *testing.T) {
|
||||
plugin := acmePlugin{tools: []pact.MCPTool{echoTool()}}
|
||||
|
||||
t.Run("memory store reaps an idle session past the TTL", func(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
in := startInstance(t, newApp(t, appSpec{
|
||||
mcp: "enabled: false\nsessions:\n store: memory\n idle_ttl: 30m\n",
|
||||
guards: []pact.MCPGuard{&acmeGuard{}},
|
||||
}), plugin)
|
||||
p := in.pier()
|
||||
t.Cleanup(p.StopReaper)
|
||||
logs := captureLogs(t)
|
||||
|
||||
sid := openSession(t, ctx, in, "oas_at_test")
|
||||
done, ok := p.SessionDone(sid)
|
||||
if !ok {
|
||||
t.Fatal("the session is not connected on the instance that opened it")
|
||||
}
|
||||
if n := p.ReapNow(ctx); n != 0 {
|
||||
t.Fatalf("a pass within the TTL reaped %d sessions", n)
|
||||
}
|
||||
if !p.AdvanceStoreClock(31 * time.Minute) {
|
||||
t.Fatal("the instance has no memory store")
|
||||
}
|
||||
if n := p.ReapNow(ctx); n != 1 {
|
||||
t.Fatalf("reaped = %d, want 1", n)
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("ss.Wait did not return after the reap")
|
||||
}
|
||||
if n := p.LocalSessions(); n != 0 {
|
||||
t.Fatalf("local sessions after the reap = %d, want 0", n)
|
||||
}
|
||||
if gone := rpc(t, ctx, in.url(), "oas_at_test", sid, callBody); gone.status != http.StatusNotFound {
|
||||
t.Fatalf("call on the reaped session = %d %s", gone.status, gone.body)
|
||||
}
|
||||
logged := logs.String()
|
||||
if !strings.Contains(logged, "step=expired") || !strings.Contains(logged, "session="+sid[:8]) {
|
||||
t.Fatalf("no expired audit line for %s: %s", sid[:8], logged)
|
||||
}
|
||||
if strings.Contains(logged, "oas_at_test") {
|
||||
t.Fatalf("the audit lines carry the token: %s", logged)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("redis instance keeps its copy only while the record lives", func(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
a, b, mr := twoInstances(t, appSpec{
|
||||
mcp: "enabled: false\nsessions:\n store: redis\n idle_ttl: 30m\n",
|
||||
guards: []pact.MCPGuard{&acmeGuard{}},
|
||||
}, plugin)
|
||||
t.Cleanup(a.pier().StopReaper)
|
||||
t.Cleanup(b.pier().StopReaper)
|
||||
|
||||
sid := openSession(t, ctx, a, "oas_at_test")
|
||||
// B serves one request and holds a rehydrated copy; A keeps serving.
|
||||
if call := rpc(t, ctx, b.url(), "oas_at_test", sid, callBody); call.status != http.StatusOK {
|
||||
t.Fatalf("call on B = %d %s", call.status, call.body)
|
||||
}
|
||||
if n := b.pier().LocalSessions(); n != 1 {
|
||||
t.Fatalf("B local sessions = %d, want 1", n)
|
||||
}
|
||||
mr.FastForward(20 * time.Minute)
|
||||
if call := rpc(t, ctx, a.url(), "oas_at_test", sid, callBody); call.status != http.StatusOK {
|
||||
t.Fatalf("call on A = %d %s", call.status, call.body)
|
||||
}
|
||||
mr.FastForward(20 * time.Minute)
|
||||
if n := b.pier().ReapNow(ctx); n != 0 {
|
||||
t.Fatalf("B reaped a session A kept alive (%d closed)", n)
|
||||
}
|
||||
if n := b.pier().LocalSessions(); n != 1 {
|
||||
t.Fatalf("B local sessions while the record lives = %d, want 1", n)
|
||||
}
|
||||
mr.FastForward(31 * time.Minute)
|
||||
if n := b.pier().ReapNow(ctx); n != 1 {
|
||||
t.Fatalf("B reaped %d sessions after the record expired, want 1", n)
|
||||
}
|
||||
if n := a.pier().ReapNow(ctx); n != 1 {
|
||||
t.Fatalf("A reaped %d sessions after the record expired, want 1", n)
|
||||
}
|
||||
if a.pier().LocalSessions() != 0 || b.pier().LocalSessions() != 0 {
|
||||
t.Fatalf("copies left: A %d, B %d", a.pier().LocalSessions(), b.pier().LocalSessions())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a store error keeps the copy", func(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
a, _, mr := twoInstances(t, appSpec{
|
||||
mcp: "enabled: false\nsessions:\n store: redis\n idle_ttl: 30m\n",
|
||||
guards: []pact.MCPGuard{&acmeGuard{}},
|
||||
}, plugin)
|
||||
t.Cleanup(a.pier().StopReaper)
|
||||
logs := captureLogs(t)
|
||||
openSession(t, ctx, a, "oas_at_test")
|
||||
openSession(t, ctx, a, "oas_at_test")
|
||||
mr.SetError("ERR the store is down")
|
||||
n := a.pier().ReapNow(ctx)
|
||||
mr.SetError("")
|
||||
if n != 0 {
|
||||
t.Fatalf("a store error closed %d sessions", n)
|
||||
}
|
||||
if got := a.pier().LocalSessions(); got != 2 {
|
||||
t.Fatalf("local sessions after a failing pass = %d, want 2", got)
|
||||
}
|
||||
if c := strings.Count(logs.String(), "session reaper could not read the store"); c != 1 {
|
||||
t.Fatalf("store error logged %d times in one pass, want once: %s", c, logs.String())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a GET stream whose record vanished ends through the re-check", func(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
guard := &acmeGuard{}
|
||||
in := startInstance(t, newApp(t, appSpec{
|
||||
mcp: "enabled: false\nsessions:\n store: memory\n idle_ttl: 30m\n",
|
||||
guards: []pact.MCPGuard{guard},
|
||||
}), plugin)
|
||||
p := in.pier()
|
||||
t.Cleanup(p.StopReaper)
|
||||
p.SetRecheck(50 * time.Millisecond)
|
||||
logs := captureLogs(t)
|
||||
|
||||
sid := openSession(t, ctx, in, "oas_at_test")
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, in.url(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer oas_at_test")
|
||||
req.Header.Set("Accept", "text/event-stream")
|
||||
req.Header.Set("Mcp-Session-Id", sid)
|
||||
req.Header.Set("Mcp-Protocol-Version", "2025-11-25")
|
||||
res, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if res.StatusCode != http.StatusOK {
|
||||
t.Fatalf("GET stream = %d", res.StatusCode)
|
||||
}
|
||||
ended := make(chan []byte, 1)
|
||||
go func() {
|
||||
raw, _ := io.ReadAll(res.Body)
|
||||
ended <- raw
|
||||
}()
|
||||
select {
|
||||
case <-ended:
|
||||
t.Fatal("the stream ended while its record lived")
|
||||
case <-time.After(300 * time.Millisecond):
|
||||
}
|
||||
// The record expires; no event is published, so only the re-check
|
||||
// can notice.
|
||||
if !p.AdvanceStoreClock(31 * time.Minute) {
|
||||
t.Fatal("the instance has no memory store")
|
||||
}
|
||||
select {
|
||||
case raw := <-ended:
|
||||
if strings.Contains(string(raw), "ended") {
|
||||
t.Fatalf("the stream delivered an event before closing: %s", raw)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the stream outlived its record")
|
||||
}
|
||||
if n := p.LocalSessions(); n != 0 {
|
||||
t.Fatalf("local sessions after the stream closed = %d, want 0", n)
|
||||
}
|
||||
if !strings.Contains(logs.String(), "step=stream_closed") {
|
||||
t.Fatalf("no stream_closed audit line: %s", logs.String())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a call without a session logs ephemeral", func(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
in := startInstance(t, newApp(t, appSpec{
|
||||
mcp: "enabled: false\nsessions:\n store: memory\n",
|
||||
guards: []pact.MCPGuard{&acmeGuard{}},
|
||||
}), plugin)
|
||||
t.Cleanup(in.pier().StopReaper)
|
||||
logs := captureLogs(t)
|
||||
call := rpc(t, ctx, in.url(), "oas_at_test", "", callBody)
|
||||
if call.status != http.StatusOK || !strings.Contains(call.body, `\"principal\":7`) {
|
||||
t.Fatalf("call without a session = %d %s", call.status, call.body)
|
||||
}
|
||||
logged := logs.String()
|
||||
if !strings.Contains(logged, "pier: tool call") || !strings.Contains(logged, "session=ephemeral") {
|
||||
t.Fatalf("the tool call is not logged as ephemeral: %s", logged)
|
||||
}
|
||||
if n := in.pier().LocalSessions(); n != 0 {
|
||||
t.Fatalf("an ephemeral call left %d local sessions", n)
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user