feat: sweep for cross-account patterns nightly, since every other control watches one subject

cmd/consumer/main.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"context"
+	"github.com/warmbly/warmbly/internal/app/correlate"
+	"github.com/warmbly/warmbly/internal/app/orgrisk"
 	"log"
 	"os"
 	"os/signal"
@@ -429,6 +431,12 @@ func main() {
 	// risk_pool worker when the band changes. Skipped if AssignmentService
 	// or WorkerRepo are nil.
 	go jobsService.StartRiskRebalancer(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

@@ -71,6 +71,19 @@ Creating an account records the address it was created from, the browser's user-
 
 Nothing here refuses a signup on its own. A single weak signal is wrong too often, and turning a false positive into a rejected account leaves a real customer with no way through.
 
+## 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,125 @@
+// Package correlate runs the nightly cross-account sweep.
+//
+// Every other abuse control watches one subject: a rate limit watches a user,
+// warmup health a mailbox, verification an address. An actor spreading the same
+// behaviour across several accounts stays under all of them. This is the pass
+// that looks at the group.
+package correlate
+
+import (
+	"context"
+	"fmt"
+	"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. Each finding is filed as a signal on every member
+// organization; the band it produces is the risk service's decision, not this
+// package's.
+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.record(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.record(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 {
+		for _, b := range bursts {
+			s.signal(ctx, b.OrganizationIDs, "mailbox_burst", weightMailboxBurst,
+				fmt.Sprintf("connected %d or more mailboxes within %s", MailboxBurstCount, MailboxBurstWindow))
+			recorded++
+		}
+	}
+
+	if recorded > 0 {
+		log.Info().Int("findings", recorded).Msg("correlation sweep recorded cross-account findings")
+	}
+}
+
+func (s *Service) record(ctx context.Context, clusters []repository.Cluster, key string, weight int, format string) int {
+	n := 0
+	for _, c := range clusters {
+		if len(c.OrganizationIDs) < MinClusterMembers {
+			continue
+		}
+		// "%d OTHER workspaces", so the sentence reads correctly for a member.
+		detail := fmt.Sprintf(format, len(c.OrganizationIDs)-1)
+		s.signal(ctx, c.OrganizationIDs, key, weight, detail)
+		n++
+	}
+	return n
+}
+
+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/repository/correlation_live_test.go

@@ -0,0 +1,163 @@
+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)
+		}
+	}
+}

internal/repository/pg_correlation.go

@@ -0,0 +1,130 @@
+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.
+		   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)
+		   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()
+}