Pin three store invariants that no backend was checking

Makefile

@@ -1,4 +1,4 @@
-.PHONY: help test clean fmt lint lint-fix vet tidy deps check coverage t c f l lf v check-deps
+.PHONY: help test test-integration clean fmt lint lint-fix vet tidy deps check coverage t c f l lf v check-deps
 
 # Default target
 .DEFAULT_GOAL := help
@@ -7,6 +7,11 @@
 GO=go
 GOFLAGS=-v
 
+# Local integration run. The port is deliberately not 27017, so this does not
+# collide with a MongoDB you already have running.
+MONGO_PORT ?= 27019
+MONGO_CONTAINER ?= authsome-integration-mongo
+
 # Colors for output
 RED=\033[0;31m
 GREEN=\033[0;32m
@@ -102,6 +107,39 @@ test-race:
 	$(GO) test -race -v ./...
 	@echo "$(GREEN)✓ Race tests complete$(NC)"
 
+## test-integration: Run the Store Conformance CI job locally (needs Docker)
+# Same command the "Store Conformance (pg + mongo)" job runs, including its
+# -run pattern and its -p 1. Postgres comes up on its own through
+# testcontainers. Mongo needs a replica set, because the app and org cascades
+# run in transactions, so this starts one and tears it down again after.
+#
+# You get a new database every run, and you want that. Several of these tests
+# write fixed literals and never clean up after themselves, so a second run
+# against a database the first one dirtied fails on duplicate keys. CI never
+# sees it because CI gets a fresh container.
+test-integration:
+	@echo "$(BLUE)Starting MongoDB replica set on port $(MONGO_PORT)...$(NC)"
+	@docker rm -f $(MONGO_CONTAINER) >/dev/null 2>&1 || true
+	@docker run -d --name $(MONGO_CONTAINER) -p $(MONGO_PORT):27017 mongo:7 --replSet rs0 >/dev/null
+	@for i in $$(seq 1 30); do \
+		docker exec $(MONGO_CONTAINER) mongosh --quiet --eval 'db.runCommand({ping:1})' >/dev/null 2>&1 && break; \
+		sleep 2; \
+	done
+	@docker exec $(MONGO_CONTAINER) mongosh --quiet \
+		--eval 'rs.initiate({_id:"rs0",members:[{_id:0,host:"localhost:27017"}]})' >/dev/null 2>&1 || true
+	@for i in $$(seq 1 30); do \
+		docker exec $(MONGO_CONTAINER) mongosh --quiet --eval 'db.hello().isWritablePrimary' 2>/dev/null | grep -q true && break; \
+		sleep 2; \
+	done
+	@echo "$(BLUE)Running integration suite...$(NC)"
+	@AUTHSOME_MONGO_URI='mongodb://localhost:$(MONGO_PORT)/authsome_test?replicaSet=rs0&directConnection=true' \
+		$(GO) test -tags integration -p 1 -count=1 -timeout 15m \
+		-run '^TestConformance$$|^TestStoreConformance_|^TestMigration_' ./store/... ./plugins/...; \
+	status=$$?; \
+	docker rm -f $(MONGO_CONTAINER) >/dev/null 2>&1 || true; \
+	if [ $$status -ne 0 ]; then exit $$status; fi; \
+	printf '$(GREEN)✓ Integration suite complete$(NC)\n'
+
 ## coverage: Generate test coverage
 coverage:
 	@echo "$(BLUE)Generating coverage report...$(NC)"

plugins/sharedsignals/ssftest/event_cases.go

@@ -265,3 +265,44 @@ func testExpiredSignalIsNotActive(t *testing.T, f Fixture) {
 		"a risk signal that expired an hour ago is still constraining sign-in")
 	assert.True(t, sawLive, "a signal with an hour left must still apply")
 }
+
+// testActiveSignalsAreEnvironmentScoped keeps a signal raised in one
+// environment out of another. Risk decisions run per environment, so a
+// production compromise must not raise the risk on a development sign-in for
+// the same app and the same user, and vice versa. A query that filters on
+// app and user but forgets the environment predicate fails exactly here.
+func testActiveSignalsAreEnvironmentScoped(t *testing.T, f Fixture) {
+	ctx := context.Background()
+	s := seedStream(t, f)
+
+	// A second environment under the same app. Neither the signals table nor
+	// its indexes declare a foreign key on env_id, so this needs no row of
+	// its own on any backend.
+	otherEnv := id.NewEnvironmentID()
+
+	sig := &ssf.Signal{
+		ID: id.NewSSFSignalID(), AppID: f.AppID, EnvID: f.EnvID, UserID: f.UserID,
+		StreamID: s.ID, EventType: sessionRevoked, Severity: 100,
+		Reason: "credential-leak", EventAt: now(),
+		ExpiresAt: now().Add(time.Hour), CreatedAt: now(),
+	}
+	require.NoError(t, f.Store.CreateSignal(ctx, sig))
+
+	// time.Now() unqualified, matching the risk path's own call in risk.go.
+	other, err := f.Store.ListActiveSignals(ctx, f.AppID, otherEnv, f.UserID, time.Now())
+	require.NoError(t, err)
+	for _, got := range other {
+		assert.NotEqual(t, sig.ID, got.ID,
+			"a signal from another environment leaked into env %s", otherEnv)
+	}
+
+	mine, err := f.Store.ListActiveSignals(ctx, f.AppID, f.EnvID, f.UserID, time.Now())
+	require.NoError(t, err)
+	var found bool
+	for _, got := range mine {
+		if got.ID == sig.ID {
+			found = true
+		}
+	}
+	assert.True(t, found, "the signal must still be active in its own environment")
+}

plugins/sharedsignals/ssftest/ssftest.go

@@ -55,6 +55,7 @@ func RunConformance(t *testing.T, newFixture Factory, skip ...string) {
 		{"DeleteStream", testDeleteStream},
 		{"SubjectLinkUpsertIsIdempotent", testSubjectLinkUpsertIsIdempotent},
 		{"SubjectLinkLookupIsTenantScoped", testSubjectLinkLookupIsTenantScoped},
+		{"SubjectLinkUpsertIsConcurrencySafe", testSubjectLinkUpsertIsConcurrencySafe},
 		{"DuplicateJTIIsRejected", testDuplicateJTIIsRejected},
 		{"SameJTIOnAnotherStreamIsAllowed", testSameJTIOnAnotherStreamIsAllowed},
 		{"GetReceivedEventIsAppScoped", testGetReceivedEventIsAppScoped},
@@ -64,6 +65,7 @@ func RunConformance(t *testing.T, newFixture Factory, skip ...string) {
 		{"DeleteReceivedEventFreesTheJTI", testDeleteReceivedEventFreesTheJTI},
 		{"CountEventsSince", testCountEventsSince},
 		{"ExpiredSignalIsNotActive", testExpiredSignalIsNotActive},
+		{"ActiveSignalsAreEnvironmentScoped", testActiveSignalsAreEnvironmentScoped},
 	}
 	for _, tc := range cases {
 		if skipSet[tc.name] {

plugins/sharedsignals/ssftest/stream_cases.go

@@ -3,6 +3,7 @@ package ssftest
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -187,3 +188,63 @@ func testSubjectLinkLookupIsTenantScoped(t *testing.T, f Fixture) {
 	require.Error(t, err, "subject link lookup crossed a tenant boundary")
 	assert.True(t, errors.Is(err, ssf.ErrNotFound), "got %v", err)
 }
+
+// testSubjectLinkUpsertIsConcurrencySafe fires the same tuple at the store
+// from many goroutines at once. This is not ceremony and the goroutine count
+// is not arbitrary: a subject link is written on every SSO sign-in, so two
+// writes for one subject arriving together is what happens when somebody
+// opens two tabs. testSubjectLinkUpsertIsIdempotent proves the second call
+// updates rather than collides when it arrives afterwards. It says nothing
+// about what happens when both arrive at once.
+//
+// A read-then-write implementation loses that race. Both readers see "not
+// found", both insert, and the loser hits the unique index and surfaces a raw
+// constraint error instead of succeeding. So every call here must return nil,
+// and exactly one row must survive.
+func testSubjectLinkUpsertIsConcurrencySafe(t *testing.T, f Fixture) {
+	ctx := context.Background()
+	issuer := "https://" + unique("race") + ".test"
+	subject := unique("subject")
+
+	const n = 30
+	errs := make([]error, n)
+	var wg sync.WaitGroup
+	wg.Add(n)
+	for i := 0; i < n; i++ {
+		go func(i int) {
+			defer wg.Done()
+			errs[i] = f.Store.UpsertSubjectLink(ctx, &ssf.SubjectLink{
+				ID: id.NewSSFLinkID(), AppID: f.AppID, EnvID: f.EnvID,
+				Issuer: issuer, Subject: subject, UserID: id.NewUserID(),
+				Source: "verified", CreatedAt: now(), LastSeenAt: now(),
+			})
+		}(i)
+	}
+	wg.Wait()
+
+	for i, err := range errs {
+		assert.NoError(t, err, "concurrent upsert %d must not surface a constraint error", i)
+	}
+
+	// The loop above is also what establishes that exactly one row exists. A
+	// second row for this tuple could only come from a writer that hit the
+	// unique index on (app_id, env_id, issuer, subject) and reported it, and
+	// that writer would have failed the loop. On memory there is no index and
+	// the mutex does that work instead. The read below is a FindOne either
+	// way, so it is not what proves the count.
+	//
+	// What the storm leaves undetermined is which of the n writes landed
+	// last, because real concurrency gives no way to know. One more
+	// deterministic write after it settles pins that down.
+	last := id.NewUserID()
+	require.NoError(t, f.Store.UpsertSubjectLink(ctx, &ssf.SubjectLink{
+		ID: id.NewSSFLinkID(), AppID: f.AppID, EnvID: f.EnvID,
+		Issuer: issuer, Subject: subject, UserID: last,
+		Source: "verified", CreatedAt: now(), LastSeenAt: now(),
+	}))
+
+	got, err := f.Store.GetSubjectLink(ctx, f.AppID, f.EnvID, issuer, subject)
+	require.NoError(t, err)
+	assert.Equal(t, last, got.UserID,
+		"the upsert must be last-write-wins, not first-write-wins")
+}

plugins/sso/conformance_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/xraph/grove/drivers/sqlitedriver"
 	_ "github.com/xraph/grove/drivers/sqlitedriver/sqlitemigrate"
 
+	"github.com/xraph/authsome/id"
 	"github.com/xraph/authsome/plugins/sso"
 	"github.com/xraph/authsome/plugins/sso/ssotest"
 	"github.com/xraph/authsome/store"
@@ -57,6 +58,8 @@ func seedFixture(t *testing.T, core store.Store, plugin sso.Store) ssotest.Fixtu
 		EnvID:      tn.EnvID.String(),
 		OtherAppID: other.AppID,
 		OtherEnvID: other.EnvID.String(),
+		OrgID:      id.NewOrgID(),
+		OtherOrgID: id.NewOrgID(),
 	}
 }
 

plugins/sso/ssotest/cases.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/xraph/authsome/id"
+	"github.com/xraph/authsome/plugins/sso"
 )
 
 func testConnectionCRUD(t *testing.T, f Fixture) {
@@ -34,11 +35,16 @@ func testConnectionCRUD(t *testing.T, f Fixture) {
 func testConnectionNotFound(t *testing.T, f Fixture) {
 	ctx := context.Background()
 
+	// The sentinel, not a bare Error: callers branch on ErrConnectionNotFound
+	// to tell "no SSO configured for this domain" from a backend that failed,
+	// and the first sends the user to password login while the second must
+	// surface. A backend returning some other error passes assert.Error and
+	// breaks that branch.
 	_, err := f.Store.GetConnection(ctx, id.NewSSOConnectionID())
-	assert.Error(t, err, "an unknown connection id must not resolve")
+	assert.ErrorIs(t, err, sso.ErrConnectionNotFound, "an unknown connection id must not resolve")
 
 	_, err = f.Store.GetConnectionByDomain(ctx, f.AppID, unique("absent")+".test")
-	assert.Error(t, err, "an unconfigured domain must not resolve")
+	assert.ErrorIs(t, err, sso.ErrConnectionNotFound, "an unconfigured domain must not resolve")
 }
 
 // testDomainLookupIsAppScoped is the one that matters most in this store.
@@ -222,3 +228,54 @@ func testDeleteConnection(t *testing.T, f Fixture) {
 	_, err = f.Store.GetConnectionByDomain(ctx, f.AppID, c.Domain)
 	assert.Error(t, err, "a deleted connection must stop resolving by domain too")
 }
+
+// testDomainLookupIsOrgScoped covers the multi-tenant case one level below
+// testDomainLookupIsAppScoped. One company can run SSO for several of its own
+// organizations, and the unique index is on (app_id, org_id, domain) where
+// active, so the same domain is legitimately configured twice inside one app.
+// GetConnectionByDomainAndOrg has to return the right one. A lookup that
+// drops the org predicate resolves an email to whichever org happens to sort
+// first, sending that user to another org's identity provider.
+func testDomainLookupIsOrgScoped(t *testing.T, f Fixture) {
+	if f.OrgID.IsNil() || f.OtherOrgID.IsNil() {
+		t.Skip("fixture provides no second organization")
+	}
+	ctx := context.Background()
+
+	// One domain, two orgs, same app. Both rows are active, which the partial
+	// unique index permits precisely because org_id is part of the key.
+	domain := unique("shared") + ".test"
+
+	mine := newConnection(f.AppID, f.EnvID)
+	mine.Domain, mine.OrgID = domain, f.OrgID
+	require.NoError(t, f.Store.CreateConnection(ctx, mine))
+
+	theirs := newConnection(f.AppID, f.EnvID)
+	theirs.Domain, theirs.OrgID = domain, f.OtherOrgID
+	require.NoError(t, f.Store.CreateConnection(ctx, theirs),
+		"the same domain must be configurable in a second org of the same app")
+
+	got, err := f.Store.GetConnectionByDomainAndOrg(ctx, f.AppID, f.OrgID, domain)
+	require.NoError(t, err)
+	assert.Equal(t, mine.ID, got.ID, "domain lookup returned the wrong organization's connection")
+
+	got, err = f.Store.GetConnectionByDomainAndOrg(ctx, f.AppID, f.OtherOrgID, domain)
+	require.NoError(t, err)
+	assert.Equal(t, theirs.ID, got.ID, "domain lookup returned the wrong organization's connection")
+
+	// An org with nothing configured for the domain must come up empty rather
+	// than borrowing a sibling org's connection.
+	_, err = f.Store.GetConnectionByDomainAndOrg(ctx, f.AppID, id.NewOrgID(), domain)
+	assert.ErrorIs(t, err, sso.ErrConnectionNotFound)
+
+	// Deactivating one org's connection must leave the other untouched.
+	mine.Active = false
+	require.NoError(t, f.Store.UpdateConnection(ctx, mine))
+	_, err = f.Store.GetConnectionByDomainAndOrg(ctx, f.AppID, f.OrgID, domain)
+	assert.ErrorIs(t, err, sso.ErrConnectionNotFound)
+
+	still, err := f.Store.GetConnectionByDomainAndOrg(ctx, f.AppID, f.OtherOrgID, domain)
+	require.NoError(t, err)
+	assert.Equal(t, theirs.ID, still.ID,
+		"deactivating one organization's connection removed another's")
+}

plugins/sso/ssotest/ssotest.go

@@ -30,6 +30,12 @@ type Fixture struct {
 	OtherAppID id.AppID
 	// OtherEnvID is that second tenant's default environment.
 	OtherEnvID string
+	// OrgID and OtherOrgID are two organizations inside AppID. Connections
+	// route per organization as well as per app, so proving that lookup needs
+	// two orgs under one tenant rather than two tenants. org_id carries no
+	// foreign key on any backend, so these need no row of their own.
+	OrgID      id.OrgID
+	OtherOrgID id.OrgID
 }
 
 // Factory builds a fresh, empty, migrated fixture for a single test.
@@ -49,6 +55,7 @@ func RunConformance(t *testing.T, newFixture Factory, skip ...string) {
 		{"ConnectionCRUD", testConnectionCRUD},
 		{"ConnectionNotFound", testConnectionNotFound},
 		{"DomainLookupIsAppScoped", testDomainLookupIsAppScoped},
+		{"DomainLookupIsOrgScoped", testDomainLookupIsOrgScoped},
 		{"ProviderLookupIsAppScoped", testProviderLookupIsAppScoped},
 		{"ListConnectionsIsAppScoped", testListConnectionsIsAppScoped},
 		{"SAMLFieldsRoundTrip", testSAMLFieldsRoundTrip},