From d4d559049ad70cf4c76846b37a21fb3db53dd88f Mon Sep 17 00:00:00 2001 From: mintaka Date: Thu, 1 Oct 2026 20:35:13 -0400 Subject: [PATCH] fix(runnerhub): scope Runner-originated binding calls to the session's tenant (RIG-4146) Runners are shared across tenants, so the Runner door carries no tenant and binding reads/deletes on deliverAck, forgeNotificationAck and dropLostSession ran under the bootstrap tenant. Resolve the session's tenant under the system role (SessionBindingTenant), then act under WithTenant. A session bound in two tenants on one runner is refused. Co-authored-by: Matt Wilkinson --- go/internal/runnerhub/binding_cache_test.go | 9 ++ go/internal/runnerhub/hub.go | 13 +++ go/internal/runnerhub/relay_comms.go | 35 ++++++++ .../runnerhub/runner_tenant_pgtest_test.go | 84 +++++++++++++++++++ go/internal/store/db/querier.go | 4 + go/internal/store/db/session_bindings.sql.go | 32 +++++++ .../store/queries/session_bindings.sql | 6 ++ go/internal/store/session_bindings.go | 23 +++++ .../store/session_bindings_pgtest_test.go | 29 +++++++ 9 files changed, 235 insertions(+) create mode 100644 go/internal/runnerhub/runner_tenant_pgtest_test.go diff --git a/go/internal/runnerhub/binding_cache_test.go b/go/internal/runnerhub/binding_cache_test.go index 94e3dbe8d..7a5f7cb44 100644 --- a/go/internal/runnerhub/binding_cache_test.go +++ b/go/internal/runnerhub/binding_cache_test.go @@ -151,6 +151,15 @@ func (f *fakeBindingStore) EffectiveTenant(context.Context) store.TenantID { const testRunnerID = "runner-1" // seed inserts a binding directly (test setup), bypassing the displacement path. +func (f *fakeBindingStore) SessionBindingTenant(_ context.Context, sessionID, runnerID string) (store.TenantID, error) { + f.mu.Lock() + defer f.mu.Unlock() + if b, ok := f.bindings[sessionID]; ok && b.RunnerID == runnerID { + return f.tenant, nil + } + return "", store.ErrNotFound +} + func (f *fakeBindingStore) seed(sessionID string) { f.mu.Lock() defer f.mu.Unlock() diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index c2f14c643..19a50858c 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -216,6 +216,9 @@ type SessionBindingStore interface { // EffectiveTenant names the tenant a request-scoped call resolves against, // so a post-write BindingChange publishes on that tenant's routing subject. EffectiveTenant(ctx context.Context) store.TenantID + // SessionBindingTenant names the tenant binding sessionID to runnerID. The hub + // calls it under the system role: the Runner door carries no tenant. + SessionBindingTenant(ctx context.Context, sessionID, runnerID string) (store.TenantID, error) } // RoutingFabric is the binding-cache invalidation seam (RIG-3108 / §T4): a @@ -811,6 +814,11 @@ func (h *Hub) deliverAck(ctx context.Context, ev RunnerEvent, ack *compassv1inte // BEFORE the system-role escalation below. The binding read is single-valued only // because RLS narrows it to one tenant; a BYPASSRLS read could return a row from an // ARBITRARY tenant. Resolving here keeps it tenant-scoped and fail-closed. + ctx, scoped := h.runnerSessionCtx(ctx, ev.RunnerID, ev.SessionID) + if !scoped { + h.countDroppedAck(ev, "acking session is bound in several tenants") + return + } agent, ok := h.accountForRunnerSession(ctx, ev.RunnerID, ev.SessionID) if !ok { h.countDroppedAck(ev, "no agent account bound to the acking session") @@ -860,6 +868,11 @@ func (h *Hub) forgeNotificationAck(ctx context.Context, ev RunnerEvent, ack *com // account on the REQUEST ctx, BEFORE the system-role escalation, so the // binding read stays tenant-scoped and cannot return a foreign tenant's row // under BYPASSRLS. Only the cursor advance below runs under the system role. + ctx, scoped := h.runnerSessionCtx(ctx, ev.RunnerID, ev.SessionID) + if !scoped { + h.countDroppedAck(ev, "acking session is bound in several tenants") + return + } agent, ok := h.accountForRunnerSession(ctx, ev.RunnerID, ev.SessionID) if !ok { h.countDroppedAck(ev, "no agent account bound to the acking session") diff --git a/go/internal/runnerhub/relay_comms.go b/go/internal/runnerhub/relay_comms.go index 04080497b..6dcb27a8f 100644 --- a/go/internal/runnerhub/relay_comms.go +++ b/go/internal/runnerhub/relay_comms.go @@ -866,6 +866,10 @@ func commsCallError(err error) *compassv1internal.CommsCallError { // unknown, then reports the account so it can be woken. Only the owning Runner's // refusal counts, so a foreign Runner cannot unbind another's session. func (h *Hub) dropLostSession(ctx context.Context, runnerID, sessionID string) { + ctx, scoped := h.runnerSessionCtx(ctx, runnerID, sessionID) + if !scoped { + return + } account, ok := h.accountForRunnerSession(ctx, runnerID, sessionID) if !ok { return @@ -880,3 +884,34 @@ func (h *Hub) dropLostSession(ctx context.Context, runnerID, sessionID string) { lost.OnSessionLost(sessionID, account) } } + +// runnerSessionCtx scopes a Runner-originated ctx to the tenant that binds +// sessionID to runnerID. Runners are shared across tenants, so the door sets none; +// without this, binding reads and deletes run under the bootstrap tenant. The +// tenant is read under the system role; everything after runs tenant-scoped. +// An ambiguous session fails closed. A miss or a store fault keeps ctx, whose +// own binding read then fails closed for a non-bootstrap session. +func (h *Hub) runnerSessionCtx(ctx context.Context, runnerID, sessionID string) (context.Context, bool) { + if _, ok := store.TenantFromContext(ctx); ok { + return ctx, true + } + h.mu.Lock() + bindings := h.bindings + h.mu.Unlock() + if bindings == nil || runnerID == "" || sessionID == "" { + return ctx, true + } + tenant, err := bindings.SessionBindingTenant(store.WithSystemRole(ctx), sessionID, runnerID) + switch { + case err == nil: + return store.WithTenant(ctx, tenant), true + case errors.Is(err, store.ErrConflict): + h.log.Error("session bound in several tenants; refusing the runner call", + "session_id", sessionID, "runner_id", runnerID) + return ctx, false + case !errors.Is(err, store.ErrNotFound): + h.log.Warn("resolve session tenant failed; continuing unscoped", + "session_id", sessionID, "error", err) + } + return ctx, true +} diff --git a/go/internal/runnerhub/runner_tenant_pgtest_test.go b/go/internal/runnerhub/runner_tenant_pgtest_test.go new file mode 100644 index 000000000..8faeecb21 --- /dev/null +++ b/go/internal/runnerhub/runner_tenant_pgtest_test.go @@ -0,0 +1,84 @@ +//go:build pgtest && unix + +package runnerhub + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/pgtest" + "github.com/RigelBuild/compass/go/internal/store" +) + +// A Runner is shared across tenants and its door carries none. A lost session of a +// non-bootstrap tenant, read cold from the table, must still be released durably and +// woken: the hub resolves the session's tenant before it reads or deletes. +func TestDropLostSessionScopesToTheSessionTenant(t *testing.T) { + ctx := context.Background() + dsn := pgtest.RequireDSN(t) + st, err := store.Open(ctx, dsn) + if err != nil { + t.Fatalf("store.Open: %v", err) + } + t.Cleanup(st.Close) + if _, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: "admin", DisplayName: "admin"}); err != nil { + t.Fatalf("BootstrapAdmin: %v", err) + } + + tenantB := seedTenantRow(t, ctx, dsn, "tenant-b") + ctxB := store.WithTenant(ctx, tenantB) + owner, err := st.CreateUser(ctxB, store.NewUser{Handle: "owner-b", DisplayName: "owner-b"}) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + agent, err := st.CreateAgent(ctxB, owner.ID, store.NewAgent{Handle: "agent-b", DisplayName: "agent-b"}) + if err != nil { + t.Fatalf("CreateAgent: %v", err) + } + if _, err := st.RecordSessionBinding(ctxB, "sess-b", agent.ID, "runner-1"); err != nil { + t.Fatalf("RecordSessionBinding: %v", err) + } + + // A fresh hub: its cache is cold, as after a Server restart. + hub := newHubOnly() + hub.SetSessionBindingStore(st) + sink := &recordingLostSink{} + hub.SetSessionLostSink(sink) + hub.enroll(ctx, "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) + // enroll's reap ran unscoped and cannot see tenant B's row; re-record it so the + // test does not depend on that. + if _, err := st.RecordSessionBinding(ctxB, "sess-b", agent.ID, "runner-1"); err != nil { + t.Fatalf("RecordSessionBinding after enroll: %v", err) + } + + hub.dropLostSession(ctx, "runner-1", "sess-b") + + if len(sink.lost) != 1 || sink.lost[0] != agent.ID { + t.Fatalf("lost = %v, want [%s]: the tenant-B session was not resolved, so no wake", sink.lost, agent.ID) + } + if _, _, err := st.ResolveSessionBinding(ctxB, "sess-b"); err == nil { + t.Fatal("tenant B's durable binding survived the drop; a cache miss would resurrect it") + } +} + +// seedTenantRow inserts a non-bootstrap tenant. tenants is RLS-exempt, so a plain +// connection may write it. +func seedTenantRow(t *testing.T, ctx context.Context, dsn, slug string) store.TenantID { + t.Helper() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatalf("pgx.Connect: %v", err) + } + defer func() { _ = conn.Close(ctx) }() // test cleanup; nothing actionable remains + id := slug + "-id" + if _, err := conn.Exec(ctx, + "INSERT INTO tenants (id, slug, display_name, created_at_unix_ms) VALUES ($1, $2, $3, $4)", + id, slug, slug, time.Now().UnixMilli()); err != nil { + t.Fatalf("seed tenant %q: %v", slug, err) + } + return store.TenantID(id) +} diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index 8b7dcd7f9..56f0fb701 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -557,6 +557,10 @@ type Querier interface { // A MISS is not an error: a first-ever bind returns pgx.ErrNoRows and the Store // maps that to the empty displaced id. SessionBindingForUpdate(ctx context.Context, agentAccountID string) (string, error) + // The one query here meant for the system role: a Runner-originated call carries + // no tenant, so the hub reads the session's tenant cross-tenant, then acts under it. + // :many so a session id minted in two tenants is refused, not resolved arbitrarily. + SessionBindingTenants(ctx context.Context, arg SessionBindingTenantsParams) ([]string, error) SessionMaxEntrySeq(ctx context.Context, sessionID string) (int64, error) SessionTranscript(ctx context.Context, sessionID string) ([]SessionTranscriptRow, error) // Agent-activity queries (sqlc adoption T5, RIG-3034). These replace the inline diff --git a/go/internal/store/db/session_bindings.sql.go b/go/internal/store/db/session_bindings.sql.go index 232f908d6..8cf7d54c4 100644 --- a/go/internal/store/db/session_bindings.sql.go +++ b/go/internal/store/db/session_bindings.sql.go @@ -237,3 +237,35 @@ func (q *Queries) SessionBindingForUpdate(ctx context.Context, agentAccountID st err := row.Scan(&session_id) return session_id, err } + +const sessionBindingTenants = `-- name: SessionBindingTenants :many +SELECT tenant_id FROM session_bindings WHERE session_id = $1 AND runner_id = $2 +` + +type SessionBindingTenantsParams struct { + SessionID string + RunnerID string +} + +// The one query here meant for the system role: a Runner-originated call carries +// no tenant, so the hub reads the session's tenant cross-tenant, then acts under it. +// :many so a session id minted in two tenants is refused, not resolved arbitrarily. +func (q *Queries) SessionBindingTenants(ctx context.Context, arg SessionBindingTenantsParams) ([]string, error) { + rows, err := q.db.Query(ctx, sessionBindingTenants, arg.SessionID, arg.RunnerID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []string + for rows.Next() { + var tenant_id string + if err := rows.Scan(&tenant_id); err != nil { + return nil, err + } + items = append(items, tenant_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/go/internal/store/queries/session_bindings.sql b/go/internal/store/queries/session_bindings.sql index 246b3f1bc..2c4a2558e 100644 --- a/go/internal/store/queries/session_bindings.sql +++ b/go/internal/store/queries/session_bindings.sql @@ -140,3 +140,9 @@ DELETE FROM session_bindings WHERE session_id = $1; DELETE FROM session_bindings WHERE runner_id = $1 RETURNING session_id, agent_account_id; + +-- The one query here meant for the system role: a Runner-originated call carries +-- no tenant, so the hub reads the session's tenant cross-tenant, then acts under it. +-- :many so a session id minted in two tenants is refused, not resolved arbitrarily. +-- name: SessionBindingTenants :many +SELECT tenant_id FROM session_bindings WHERE session_id = $1 AND runner_id = $2; diff --git a/go/internal/store/session_bindings.go b/go/internal/store/session_bindings.go index 99062838e..e1340e211 100644 --- a/go/internal/store/session_bindings.go +++ b/go/internal/store/session_bindings.go @@ -276,3 +276,26 @@ func (s *Store) DeleteSessionBindingsForRunner(ctx context.Context, runnerID str }) return bindings, nil } + +// SessionBindingTenant names the tenant whose binding maps sessionID to runnerID. +// It is the system-role read a Runner-originated call needs first: the Runner +// door carries no tenant, so the hub resolves it here, then acts under WithTenant. +// ErrNotFound when no tenant binds the pair; ErrConflict when several do, so an +// ambiguous session never resolves to an arbitrary tenant. +func (s *Store) SessionBindingTenant(ctx context.Context, sessionID, runnerID string) (TenantID, error) { + if sessionID == "" || runnerID == "" { + return "", fmt.Errorf("%w: session id and runner id are required", ErrInvalidArgument) + } + tenants, err := s.q.SessionBindingTenants(ctx, db.SessionBindingTenantsParams{SessionID: sessionID, RunnerID: runnerID}) + if err != nil { + return "", fmt.Errorf("store: resolve session binding tenant: %w", err) + } + switch len(tenants) { + case 0: + return "", fmt.Errorf("%w: session %q is not bound to runner %q", ErrNotFound, sessionID, runnerID) + case 1: + return TenantID(tenants[0]), nil + default: + return "", fmt.Errorf("%w: session %q is bound in %d tenants", ErrConflict, sessionID, len(tenants)) + } +} diff --git a/go/internal/store/session_bindings_pgtest_test.go b/go/internal/store/session_bindings_pgtest_test.go index 13c500747..f313002f1 100644 --- a/go/internal/store/session_bindings_pgtest_test.go +++ b/go/internal/store/session_bindings_pgtest_test.go @@ -1116,3 +1116,32 @@ func TestRecordSessionBindingReportsDisplacedExactlyOnceAgainstARunnerSweep(t *t } t.Logf("%d iterations: the displaced session was reported to exactly one caller every time", iterations) } + +// TestSessionBindingTenantResolvesAcrossTenantsUnderSystemRole pins the read a +// Runner-originated call scopes itself with: under the system role it names the +// binding's tenant, a session minted in two tenants on one runner is a conflict, +// and a pair nobody binds is not-found. +func TestSessionBindingTenantResolvesAcrossTenantsUnderSystemRole(t *testing.T) { + s := newTestStore(t) + tenantB := seedTenant(t, s, "tenant-b") + ctxA := context.Background() + ctxB := WithTenant(context.Background(), tenantB) + sys := WithSystemRole(context.Background()) + + agentA := mustAgent(t, s, mustUser(t, s, "owner-a").ID, "agent-a") + agentB := seedTenantAgent(t, ctxB, s, "agent-b") + mustBind(t, ctxB, s, "sess-b", agentB.ID, "runner-1") + + got, err := s.SessionBindingTenant(sys, "sess-b", "runner-1") + if err != nil || got != tenantB { + t.Fatalf("SessionBindingTenant(sess-b) = (%q, %v), want (%q, nil)", got, err, tenantB) + } + if _, err := s.SessionBindingTenant(sys, "sess-b", "runner-2"); !errors.Is(err, ErrNotFound) { + t.Fatalf("SessionBindingTenant(sess-b, foreign runner) err = %v, want ErrNotFound", err) + } + + mustBind(t, ctxA, s, "sess-b", agentA.ID, "runner-1") + if _, err := s.SessionBindingTenant(sys, "sess-b", "runner-1"); !errors.Is(err, ErrConflict) { + t.Fatalf("SessionBindingTenant(sess-b bound in two tenants) err = %v, want ErrConflict", err) + } +}