Merge remote-tracking branch 'origin/main' into feat/149-ato-signals

# Conflicts:
#	docs/content/docs/guides/security.mdx

cmd/consumer/main.go

@@ -8,6 +8,9 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/warmbly/warmbly/internal/app/correlate"
+	"github.com/warmbly/warmbly/internal/app/orgrisk"
+
 	"github.com/aws/aws-sdk-go-v2/aws"
 	awsconf "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/getsentry/sentry-go"
@@ -433,6 +436,12 @@ func main() {
 	// Same cadence, different question: risk_band picks the worker, the
 	// lifecycle picks whether the mailbox is in cold rotation at all.
 	go jobsService.StartLifecycleRebalancer(ctx, 1*time.Hour)
+	// The cross-account sweep. Nightly rather than hourly: it looks for
+	// patterns that form over days and it touches every organization.
+	go correlate.NewService(
+		repository.NewCorrelationRepository(primaryDB),
+		orgrisk.NewService(repository.NewOrgRiskRepository(primaryDB)),
+	).Start(ctx, 24*time.Hour)
 
 	// Tracking consumer (opens/clicks): a second subscription on the shared bus
 	// for the tracking topic. It records open/click engagement and fires INSTANT

docs/content/docs/guides/security.mdx

@@ -91,6 +91,19 @@ Repeated anomalies inside a fortnight are recorded against the workspace's postu
 This uses the optional MaxMind city database. Without it configured, sign-ins are recorded but never judged, which is the safe direction: a false challenge locks a real person out of their own account.
 </Callout>
 
+## Cross-account patterns
+
+Individually, every abuse control watches one thing: a rate limit watches one account, warmup health one mailbox, verification one address. Someone spreading the same behaviour across several accounts stays under all of them.
+
+A nightly sweep looks at the group instead. It groups workspaces that were opened from the same address, or by the same email identity once plus-tags and Gmail dots are collapsed, and notices a workspace that connects an unusual number of mailboxes at once.
+
+Findings are recorded as evidence on the [workspace posture](/guides/deliverability/), never acted on alone:
+
+- **Three workspaces is the floor.** Two sharing an address is a coincidence.
+- **Private and loopback addresses are ignored**, since every self-hosted install signs up over a LAN.
+- **Sharing an address carries little weight** on its own. Offices, co-working spaces and VPNs all produce it honestly.
+- **Only the last 30 days count.** Two workspaces opened from one office a year apart are not related.
+
 ## Strongest setup
 
 1. **Add a passkey** so daily sign-in is fast and phishing-resistant.

internal/app/correlate/correlate.go

@@ -0,0 +1,150 @@
+// Package correlate runs the nightly cross-account sweep: the abuse pass that
+// watches a group of accounts rather than one subject.
+package correlate
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog/log"
+
+	"github.com/warmbly/warmbly/internal/app/orgrisk"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// Weights. Small on purpose: sharing an address is common (an office, a
+// co-working space, a VPN), so this contributes evidence rather than a verdict.
+const (
+	weightSharedIP       = 15
+	weightSharedIdentity = 25
+	weightMailboxBurst   = 15
+
+	// LookbackWindow bounds how far back accounts are correlated. Two
+	// organizations opened from one office a year apart are not a cluster.
+	LookbackWindow = 30 * 24 * time.Hour
+	// MinClusterMembers is the group size below which nothing is recorded.
+	MinClusterMembers = 3
+	// MailboxBurstCount and MailboxBurstWindow describe a sending fleet being
+	// stood up rather than a business connecting its inboxes.
+	MailboxBurstCount  = 15
+	MailboxBurstWindow = 24 * time.Hour
+)
+
+// Service runs the sweep.
+type Service struct {
+	repo    repository.CorrelationRepository
+	orgRisk orgrisk.Service
+}
+
+func NewService(repo repository.CorrelationRepository, risk orgrisk.Service) *Service {
+	return &Service{repo: repo, orgRisk: risk}
+}
+
+// Start runs the sweep on an interval until the context ends.
+func (s *Service) Start(ctx context.Context, interval time.Duration) {
+	if s == nil || s.repo == nil || s.orgRisk == nil {
+		return
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.Run(ctx)
+		}
+	}
+}
+
+// Run performs one sweep, filing each finding as a signal on every member
+// organization. Signals are retracted from organizations that no longer match
+// before current ones are recorded, so a score can fall as well as climb.
+func (s *Service) Run(ctx context.Context) {
+	since := time.Now().Add(-LookbackWindow)
+	recorded := 0
+
+	if clusters, err := s.repo.ClustersBySignupIP(ctx, MinClusterMembers, since); err != nil {
+		log.Warn().Err(err).Msg("correlation sweep: signup-ip clustering failed")
+	} else {
+		recorded += s.apply(ctx, clusters, "cluster_signup_ip", weightSharedIP,
+			"opened alongside %d other workspaces from one address")
+	}
+
+	if clusters, err := s.repo.ClustersBySignupIdentity(ctx, MinClusterMembers, since); err != nil {
+		log.Warn().Err(err).Msg("correlation sweep: identity clustering failed")
+	} else {
+		recorded += s.apply(ctx, clusters, "cluster_signup_identity", weightSharedIdentity,
+			"opened alongside %d other workspaces by the same email identity")
+	}
+
+	if bursts, err := s.repo.OrgsConnectingMailboxesFast(ctx, MailboxBurstCount, MailboxBurstWindow); err != nil {
+		log.Warn().Err(err).Msg("correlation sweep: mailbox burst query failed")
+	} else {
+		recorded += s.apply(ctx, bursts, "mailbox_burst", weightMailboxBurst,
+			fmt.Sprintf("connected %d or more mailboxes within %s", MailboxBurstCount, MailboxBurstWindow))
+	}
+
+	if recorded > 0 {
+		log.Info().Int("findings", recorded).Msg("correlation sweep recorded cross-account findings")
+	}
+}
+
+// apply records the current matches for one detector and retracts the signal
+// from every organization that carried it but no longer matches.
+func (s *Service) apply(ctx context.Context, clusters []repository.Cluster, key string, weight int, format string) int {
+	matched := make(map[uuid.UUID]string)
+	for _, c := range clusters {
+		if len(c.OrganizationIDs) < 2 {
+			// A burst is a single organization; a cluster needs its floor.
+			if key != "mailbox_burst" || len(c.OrganizationIDs) == 0 {
+				continue
+			}
+		} else if len(c.OrganizationIDs) < MinClusterMembers {
+			continue
+		}
+		detail := format
+		if strings.Contains(format, "%d") {
+			// "%d OTHER workspaces", so the sentence reads correctly to a member.
+			detail = fmt.Sprintf(format, len(c.OrganizationIDs)-1)
+		}
+		for _, orgID := range c.OrganizationIDs {
+			matched[orgID] = detail
+		}
+	}
+
+	// Retract first, so an organization that dropped out of every cluster does
+	// not keep the weight.
+	if previous, err := s.orgRisk.OrgsWithSignal(ctx, key); err != nil {
+		log.Warn().Str("signal", key).Msg("correlation sweep: could not list previous holders")
+	} else {
+		for _, orgID := range previous {
+			if _, still := matched[orgID]; still {
+				continue
+			}
+			if _, cerr := s.orgRisk.ClearSignal(ctx, orgID, key); cerr != nil {
+				log.Warn().Str("organization_id", orgID.String()).Str("signal", key).
+					Msg("correlation sweep: could not retract a finding that no longer holds")
+			}
+		}
+	}
+
+	for orgID, detail := range matched {
+		s.signal(ctx, []uuid.UUID{orgID}, key, weight, detail)
+	}
+	return len(matched)
+}
+
+func (s *Service) signal(ctx context.Context, orgs []uuid.UUID, key string, weight int, detail string) {
+	for _, orgID := range orgs {
+		if _, err := s.orgRisk.RecordSignal(ctx, orgID, orgrisk.Signal{
+			Key: key, Weight: weight, Detail: detail,
+		}); err != nil {
+			log.Warn().Str("organization_id", orgID.String()).Str("signal", key).
+				Msg("correlation sweep: could not record signal")
+		}
+	}
+}

internal/app/orgrisk/service.go

@@ -58,6 +58,9 @@ type Service interface {
 	RecordSignal(ctx context.Context, orgID uuid.UUID, sig Signal) (*models.OrgRisk, *errx.Error)
 	// ClearSignal removes a detector's finding, for when it no longer holds.
 	ClearSignal(ctx context.Context, orgID uuid.UUID, key string) (*models.OrgRisk, *errx.Error)
+	// OrgsWithSignal lists organizations carrying a detector's finding, so a
+	// recurring sweep can retract the ones that no longer match.
+	OrgsWithSignal(ctx context.Context, key string) ([]uuid.UUID, *errx.Error)
 	// SetState is an operator's manual override, which outranks the score.
 	SetState(ctx context.Context, orgID uuid.UUID, state models.OrgRiskState, reason string) (*models.OrgRisk, *errx.Error)
 }
@@ -137,6 +140,14 @@ func (s *service) ClearSignal(ctx context.Context, orgID uuid.UUID, key string)
 	})
 }
 
+func (s *service) OrgsWithSignal(ctx context.Context, key string) ([]uuid.UUID, *errx.Error) {
+	ids, err := s.repo.OrgsWithSignal(ctx, key)
+	if err != nil {
+		return nil, errx.InternalError()
+	}
+	return ids, nil
+}
+
 func (s *service) SetState(ctx context.Context, orgID uuid.UUID, state models.OrgRiskState, reason string) (*models.OrgRisk, *errx.Error) {
 	if !state.Valid() {
 		return nil, errx.New(errx.BadRequest, "unknown risk state")

internal/app/orgrisk/service_test.go

@@ -135,6 +135,9 @@ func (s *stubRiskRepo) UpdateOrgRiskSignals(_ context.Context, _ uuid.UUID,
 	copy := *s.risk
 	return &copy, nil
 }
+func (s *stubRiskRepo) OrgsWithSignal(context.Context, string) ([]uuid.UUID, error) {
+	return nil, nil
+}
 func (s *stubRiskRepo) SetOrgRiskState(_ context.Context, _ uuid.UUID, state models.OrgRiskState, reason string) (*models.OrgRisk, error) {
 	s.risk.State, s.risk.Reason = state, reason
 	copy := *s.risk

internal/repository/correlation_live_test.go

@@ -0,0 +1,187 @@
+package repository
+
+import (
+	"context"
+	"net/mail"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+// Issue #148: the cross-account view. These prove the queries against the real
+// schema, including the case that would hurt most: NOT clustering ordinary
+// customers who happen to share an office address.
+//
+//	WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \
+//	  go test ./internal/repository/ -run LiveCorrelation -v
+
+// makeOrg creates an owner with the given signup metadata and their workspace.
+func makeOrg(t *testing.T, ip, normalized string) uuid.UUID {
+	t.Helper()
+	handle, pool := liveContactDB(t)
+	ctx := context.Background()
+
+	addr, err := mail.ParseAddress("corr-" + uuid.New().String()[:8] + "@test.local")
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	u, err := NewUserRepostory(handle, nil).CreateUser(ctx, addr, "hash")
+	if err != nil {
+		t.Fatalf("create user: %v", err)
+	}
+	if _, err := pool.Exec(ctx,
+		`UPDATE users SET signup_ip = NULLIF($2,'')::inet, signup_email_normalized = NULLIF($3,'') WHERE id = $1`,
+		u.ID, ip, normalized); err != nil {
+		t.Fatalf("set signup metadata: %v", err)
+	}
+	org := uuid.New()
+	if _, err := pool.Exec(ctx,
+		`INSERT INTO organizations (id, name, slug, owner_user_id) VALUES ($1, 'Corr', $2, $3)`,
+		org, "corr-"+org.String()[:8], u.ID); err != nil {
+		t.Fatalf("create org: %v", err)
+	}
+	t.Cleanup(func() {
+		c := context.Background()
+		if _, err := pool.Exec(c, `DELETE FROM email_accounts WHERE organization_id = $1`, org); err != nil {
+			t.Errorf("cleanup mailboxes: %v", err)
+		}
+		if _, err := pool.Exec(c, `DELETE FROM organizations WHERE id = $1`, org); err != nil {
+			t.Errorf("cleanup org: %v", err)
+		}
+		if _, err := pool.Exec(c, `DELETE FROM users WHERE id = $1`, u.ID); err != nil {
+			t.Errorf("cleanup user: %v", err)
+		}
+	})
+	return org
+}
+
+func contains(ids []uuid.UUID, want uuid.UUID) bool {
+	for _, id := range ids {
+		if id == want {
+			return true
+		}
+	}
+	return false
+}
+
+func TestLiveCorrelationClustersBySignupIP(t *testing.T) {
+	handle, _ := liveContactDB(t)
+	repo := NewCorrelationRepository(handle)
+
+	ip := "198.51.100.77"
+	a, b, c := makeOrg(t, ip, ""), makeOrg(t, ip, ""), makeOrg(t, ip, "")
+	// A fourth from elsewhere must not be swept in.
+	other := makeOrg(t, "203.0.113.9", "")
+
+	clusters, err := repo.ClustersBySignupIP(context.Background(), 3, time.Now().Add(-time.Hour))
+	if err != nil {
+		t.Fatalf("ClustersBySignupIP: %v", err)
+	}
+	var found *Cluster
+	for i := range clusters {
+		if clusters[i].Key == ip {
+			found = &clusters[i]
+		}
+	}
+	if found == nil {
+		t.Fatalf("no cluster for %s; got %d clusters", ip, len(clusters))
+	}
+	for _, want := range []uuid.UUID{a, b, c} {
+		if !contains(found.OrganizationIDs, want) {
+			t.Errorf("cluster is missing %s", want)
+		}
+	}
+	if contains(found.OrganizationIDs, other) {
+		t.Error("an organization from a different address was swept into the cluster")
+	}
+}
+
+// The failure that would hurt most: a self-hosted install signing up over a
+// LAN is the normal case, and every such install shares 192.168.x.x.
+func TestLiveCorrelationIgnoresPrivateAddresses(t *testing.T) {
+	handle, _ := liveContactDB(t)
+	repo := NewCorrelationRepository(handle)
+
+	for i := 0; i < 4; i++ {
+		makeOrg(t, "192.168.1.50", "")
+	}
+	clusters, err := repo.ClustersBySignupIP(context.Background(), 3, time.Now().Add(-time.Hour))
+	if err != nil {
+		t.Fatalf("ClustersBySignupIP: %v", err)
+	}
+	for _, c := range clusters {
+		if c.Key == "192.168.1.50" {
+			t.Error("a LAN address was treated as a cluster; every self-hosted install shares one")
+		}
+	}
+}
+
+func TestLiveCorrelationClustersByIdentity(t *testing.T) {
+	handle, _ := liveContactDB(t)
+	repo := NewCorrelationRepository(handle)
+
+	identity := "oneperson" + uuid.New().String()[:6] + "@gmail.com"
+	a, b, c := makeOrg(t, "", identity), makeOrg(t, "", identity), makeOrg(t, "", identity)
+
+	clusters, err := repo.ClustersBySignupIdentity(context.Background(), 3, time.Now().Add(-time.Hour))
+	if err != nil {
+		t.Fatalf("ClustersBySignupIdentity: %v", err)
+	}
+	for _, cl := range clusters {
+		if cl.Key != identity {
+			continue
+		}
+		for _, want := range []uuid.UUID{a, b, c} {
+			if !contains(cl.OrganizationIDs, want) {
+				t.Errorf("cluster is missing %s", want)
+			}
+		}
+		return
+	}
+	t.Fatalf("no cluster for %s", identity)
+}
+
+// Two organizations is a coincidence; the floor is three.
+func TestLiveCorrelationNeedsMoreThanTwo(t *testing.T) {
+	handle, _ := liveContactDB(t)
+	repo := NewCorrelationRepository(handle)
+
+	ip := "198.51.100.88"
+	makeOrg(t, ip, "")
+	makeOrg(t, ip, "")
+
+	clusters, err := repo.ClustersBySignupIP(context.Background(), 3, time.Now().Add(-time.Hour))
+	if err != nil {
+		t.Fatalf("ClustersBySignupIP: %v", err)
+	}
+	for _, c := range clusters {
+		if c.Key == ip {
+			t.Errorf("two organizations were reported as a cluster: %+v", c)
+		}
+	}
+}
+
+// IPv6 LANs cluster just as wrongly as IPv4 ones: a self-hosted install
+// reached over ULA or link-local is the normal case, not a ring.
+func TestLiveCorrelationIgnoresPrivateIPv6(t *testing.T) {
+	handle, _ := liveContactDB(t)
+	repo := NewCorrelationRepository(handle)
+
+	for _, ip := range []string{"fd00::1", "fe80::1", "::1"} {
+		for i := 0; i < 3; i++ {
+			makeOrg(t, ip, "")
+		}
+	}
+	clusters, err := repo.ClustersBySignupIP(context.Background(), 3, time.Now().Add(-time.Hour))
+	if err != nil {
+		t.Fatalf("ClustersBySignupIP: %v", err)
+	}
+	for _, c := range clusters {
+		for _, priv := range []string{"fd00::1", "fe80::1", "::1"} {
+			if c.Key == priv {
+				t.Errorf("private IPv6 %s was treated as a cluster", priv)
+			}
+		}
+	}
+}

internal/repository/pg_correlation.go

@@ -0,0 +1,136 @@
+package repository
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+)
+
+// Cluster is a set of organizations that share something an unrelated set of
+// customers would not.
+type Cluster struct {
+	// Key is the shared value, for the evidence record.
+	Key string
+	// OrganizationIDs are every member. A cluster of one is not a cluster and
+	// is never returned.
+	OrganizationIDs []uuid.UUID
+}
+
+// CorrelationRepository finds organizations linked across accounts.
+//
+// Every per-entity control in the platform watches one subject, so an actor
+// spreading activity across several accounts sits under all of them. These
+// queries are the view that sees the group.
+type CorrelationRepository interface {
+	// ClustersBySignupIP groups organizations whose owners signed up from the
+	// same address. Private and loopback addresses are excluded: a self-hosted
+	// install signing up over a LAN is the normal case, not a signal.
+	ClustersBySignupIP(ctx context.Context, minMembers int, since time.Time) ([]Cluster, error)
+	// ClustersBySignupIdentity groups organizations whose owners' addresses
+	// collapse to the same person once plus-tags and Gmail dots are removed.
+	ClustersBySignupIdentity(ctx context.Context, minMembers int, since time.Time) ([]Cluster, error)
+	// OrgsConnectingMailboxesFast finds organizations that added an unusual
+	// number of mailboxes in a short window, which is what setting up a
+	// throwaway sending fleet looks like.
+	OrgsConnectingMailboxesFast(ctx context.Context, minMailboxes int, within time.Duration) ([]Cluster, error)
+}
+
+type correlationRepository struct {
+	DB *db.DB
+}
+
+func NewCorrelationRepository(database *db.DB) CorrelationRepository {
+	return &correlationRepository{DB: database}
+}
+
+func (r *correlationRepository) ClustersBySignupIP(ctx context.Context, minMembers int, since time.Time) ([]Cluster, error) {
+	return r.clusters(ctx, `
+		SELECT host(u.signup_ip) AS key, array_agg(DISTINCT o.id) AS orgs
+		  FROM organizations o
+		  JOIN users u ON u.id = o.owner_user_id
+		 WHERE u.signup_ip IS NOT NULL
+		   -- A LAN or loopback signup is a self-hosted install, not a signal.
+		   -- Both families: an IPv6-only LAN clusters just as wrongly.
+		   -- <<= not <<: the strict operator excludes an address from a prefix
+		   -- that describes exactly it, so ::1 is not "inside" ::1/128.
+		   AND NOT (u.signup_ip <<= '10.0.0.0/8'::inet
+		         OR u.signup_ip <<= '172.16.0.0/12'::inet
+		         OR u.signup_ip <<= '192.168.0.0/16'::inet
+		         OR u.signup_ip <<= '127.0.0.0/8'::inet
+		         OR u.signup_ip <<= '169.254.0.0/16'::inet
+		         OR u.signup_ip <<= '::1/128'::inet
+		         OR u.signup_ip <<= 'fc00::/7'::inet
+		         OR u.signup_ip <<= 'fe80::/10'::inet)
+		   AND o.created_at >= $2
+		 GROUP BY 1
+		HAVING COUNT(DISTINCT o.id) >= $1
+	`, minMembers, since)
+}
+
+func (r *correlationRepository) ClustersBySignupIdentity(ctx context.Context, minMembers int, since time.Time) ([]Cluster, error) {
+	return r.clusters(ctx, `
+		SELECT u.signup_email_normalized AS key, array_agg(DISTINCT o.id) AS orgs
+		  FROM organizations o
+		  JOIN users u ON u.id = o.owner_user_id
+		 WHERE u.signup_email_normalized IS NOT NULL
+		   AND u.signup_email_normalized <> ''
+		   AND o.created_at >= $2
+		 GROUP BY 1
+		HAVING COUNT(DISTINCT o.id) >= $1
+	`, minMembers, since)
+}
+
+func (r *correlationRepository) OrgsConnectingMailboxesFast(ctx context.Context, minMailboxes int, within time.Duration) ([]Cluster, error) {
+	rows, err := r.DB.Pool.Query(ctx, `
+		SELECT o.id::text, COUNT(*) AS n
+		  FROM email_accounts ea
+		  JOIN organizations o ON o.id = ea.organization_id
+		 WHERE ea.created_at >= NOW() - $2::interval
+		 GROUP BY o.id
+		HAVING COUNT(*) >= $1
+	`, minMailboxes, within.String())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []Cluster
+	for rows.Next() {
+		var id string
+		var n int
+		if err := rows.Scan(&id, &n); err != nil {
+			return nil, err
+		}
+		orgID, perr := uuid.Parse(id)
+		if perr != nil {
+			continue
+		}
+		out = append(out, Cluster{Key: id, OrganizationIDs: []uuid.UUID{orgID}})
+	}
+	return out, rows.Err()
+}
+
+func (r *correlationRepository) clusters(ctx context.Context, query string, minMembers int, since time.Time) ([]Cluster, error) {
+	if minMembers < 2 {
+		// A cluster of one is just an organization.
+		minMembers = 2
+	}
+	rows, err := r.DB.Pool.Query(ctx, query, minMembers, since)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []Cluster
+	for rows.Next() {
+		var c Cluster
+		if err := rows.Scan(&c.Key, &c.OrganizationIDs); err != nil {
+			return nil, err
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}

internal/repository/pg_org_risk.go

@@ -25,6 +25,10 @@ type OrgRiskRepository interface {
 	// SetOrgRiskState is an operator override. It leaves the signals alone:
 	// the evidence that led here is still the evidence.
 	SetOrgRiskState(ctx context.Context, orgID uuid.UUID, state models.OrgRiskState, reason string) (*models.OrgRisk, error)
+	// OrgsWithSignal lists organizations currently carrying a detector's
+	// finding. A recurring sweep needs this to retract findings that no longer
+	// hold; without it a signal recorded once would never decay.
+	OrgsWithSignal(ctx context.Context, key string) ([]uuid.UUID, error)
 }
 
 type orgRiskRepository struct {
@@ -165,3 +169,22 @@ func buildOrgRisk(orgID uuid.UUID, state string, score int, reason *string, rawS
 	}
 	return risk
 }
+
+func (r *orgRiskRepository) OrgsWithSignal(ctx context.Context, key string) ([]uuid.UUID, error) {
+	rows, err := r.DB.Pool.Query(ctx,
+		`SELECT id FROM organizations WHERE risk_signals ? $1`, key)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []uuid.UUID
+	for rows.Next() {
+		var id uuid.UUID
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		out = append(out, id)
+	}
+	return out, rows.Err()
+}