From 02a1ab524aa1e4691c8d7357ac07a17e960c19ed Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 28 Aug 2026 11:58:49 -0700 Subject: [PATCH 1/3] feat: sweep for cross-account patterns nightly, since every other control watches one subject --- cmd/consumer/main.go | 8 + docs/content/docs/guides/security.mdx | 13 ++ internal/app/correlate/correlate.go | 125 ++++++++++++++ internal/repository/correlation_live_test.go | 163 +++++++++++++++++++ internal/repository/pg_correlation.go | 130 +++++++++++++++ 5 files changed, 439 insertions(+) create mode 100644 internal/app/correlate/correlate.go create mode 100644 internal/repository/correlation_live_test.go create mode 100644 internal/repository/pg_correlation.go diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 7472cc33..e1cf4484 100644 --- a/cmd/consumer/main.go +++ b/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 diff --git a/docs/content/docs/guides/security.mdx b/docs/content/docs/guides/security.mdx index 3b81ac8b..d2de5076 100644 --- a/docs/content/docs/guides/security.mdx +++ b/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. diff --git a/internal/app/correlate/correlate.go b/internal/app/correlate/correlate.go new file mode 100644 index 00000000..0e982d01 --- /dev/null +++ b/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") + } + } +} diff --git a/internal/repository/correlation_live_test.go b/internal/repository/correlation_live_test.go new file mode 100644 index 00000000..3f231981 --- /dev/null +++ b/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) + } + } +} diff --git a/internal/repository/pg_correlation.go b/internal/repository/pg_correlation.go new file mode 100644 index 00000000..641c52d8 --- /dev/null +++ b/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() +} From e31286d9732e3bb167948735b45928f8842b197a Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 28 Aug 2026 12:12:28 -0700 Subject: [PATCH 2/3] feat: retract findings that no longer hold, exclude private IPv6, and fix the containment operator that let loopback through: a cluster ageing out of the lookback or a mailbox burst ending left its weight on the workspace permanently, so the score could only ever climb, and each detector now clears itself from organizations that stopped matching before recording the current ones; the private-address filter covered only IPv4, so an IPv6-only self-hosted install would have clustered as a ring; and it used << rather than <<=, which is strictly-contained and therefore excludes an address from the prefix describing exactly it, so ::1 was never inside ::1/128 --- cmd/consumer/main.go | 5 +- internal/app/correlate/correlate.go | 64 +++++++++++++++----- internal/app/orgrisk/service.go | 11 ++++ internal/app/orgrisk/service_test.go | 3 + internal/repository/correlation_live_test.go | 24 ++++++++ internal/repository/pg_correlation.go | 16 +++-- internal/repository/pg_org_risk.go | 23 +++++++ 7 files changed, 124 insertions(+), 22 deletions(-) diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index e1cf4484..28e7114a 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -2,14 +2,15 @@ package main import ( "context" - "github.com/warmbly/warmbly/internal/app/correlate" - "github.com/warmbly/warmbly/internal/app/orgrisk" "log" "os" "os/signal" "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" diff --git a/internal/app/correlate/correlate.go b/internal/app/correlate/correlate.go index 0e982d01..7964d7bd 100644 --- a/internal/app/correlate/correlate.go +++ b/internal/app/correlate/correlate.go @@ -9,6 +9,7 @@ package correlate import ( "context" "fmt" + "strings" "time" "github.com/google/uuid" @@ -66,6 +67,11 @@ func (s *Service) Start(ctx context.Context, interval time.Duration) { // 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. +// +// Every signal is RETRACTED from organizations that no longer match before the +// current matches are recorded. Without that a cluster which aged out of the +// lookback, or a mailbox burst that ended, would leave its weight on the +// workspace permanently and the score could only ever climb. func (s *Service) Run(ctx context.Context) { since := time.Now().Add(-LookbackWindow) recorded := 0 @@ -73,25 +79,22 @@ func (s *Service) Run(ctx context.Context) { 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, + 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.record(ctx, clusters, "cluster_signup_identity", weightSharedIdentity, + 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 { - for _, b := range bursts { - s.signal(ctx, b.OrganizationIDs, "mailbox_burst", weightMailboxBurst, - fmt.Sprintf("connected %d or more mailboxes within %s", MailboxBurstCount, MailboxBurstWindow)) - recorded++ - } + recorded += s.apply(ctx, bursts, "mailbox_burst", weightMailboxBurst, + fmt.Sprintf("connected %d or more mailboxes within %s", MailboxBurstCount, MailboxBurstWindow)) } if recorded > 0 { @@ -99,18 +102,49 @@ func (s *Service) Run(ctx context.Context) { } } -func (s *Service) record(ctx context.Context, clusters []repository.Cluster, key string, weight int, format string) int { - n := 0 +// 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) < MinClusterMembers { + 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 } - // "%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++ + 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 + } } - return n + + // 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) { diff --git a/internal/app/orgrisk/service.go b/internal/app/orgrisk/service.go index b525e9c9..d19a918a 100644 --- a/internal/app/orgrisk/service.go +++ b/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") diff --git a/internal/app/orgrisk/service_test.go b/internal/app/orgrisk/service_test.go index 758390a0..d7e1fbbb 100644 --- a/internal/app/orgrisk/service_test.go +++ b/internal/app/orgrisk/service_test.go @@ -135,6 +135,9 @@ func (s *stubRiskRepo) UpdateOrgRiskSignals(_ context.Context, _ uuid.UUID, copy := *s.risk return ©, 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 diff --git a/internal/repository/correlation_live_test.go b/internal/repository/correlation_live_test.go index 3f231981..a3b86a35 100644 --- a/internal/repository/correlation_live_test.go +++ b/internal/repository/correlation_live_test.go @@ -161,3 +161,27 @@ func TestLiveCorrelationNeedsMoreThanTwo(t *testing.T) { } } } + +// 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) + } + } + } +} diff --git a/internal/repository/pg_correlation.go b/internal/repository/pg_correlation.go index 641c52d8..e3fa34f9 100644 --- a/internal/repository/pg_correlation.go +++ b/internal/repository/pg_correlation.go @@ -53,11 +53,17 @@ func (r *correlationRepository) ClustersBySignupIP(ctx context.Context, minMembe 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) + -- 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 diff --git a/internal/repository/pg_org_risk.go b/internal/repository/pg_org_risk.go index 7b024793..e4154e7f 100644 --- a/internal/repository/pg_org_risk.go +++ b/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() +} From 61f7b9cd3e776784cf51536b59580ae80dfd8bd8 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 28 Aug 2026 12:21:59 -0700 Subject: [PATCH 3/3] feat: condense the correlate package and Run comments to the local constraint, moving the cross-account rationale out of production code --- internal/app/correlate/correlate.go | 19 +++++-------------- 1 file changed, 5 insertions(+), 14 deletions(-) diff --git a/internal/app/correlate/correlate.go b/internal/app/correlate/correlate.go index 7964d7bd..3d2a4974 100644 --- a/internal/app/correlate/correlate.go +++ b/internal/app/correlate/correlate.go @@ -1,9 +1,5 @@ -// 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 runs the nightly cross-account sweep: the abuse pass that +// watches a group of accounts rather than one subject. package correlate import ( @@ -64,14 +60,9 @@ func (s *Service) Start(ctx context.Context, interval time.Duration) { } } -// 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. -// -// Every signal is RETRACTED from organizations that no longer match before the -// current matches are recorded. Without that a cluster which aged out of the -// lookback, or a mailbox burst that ended, would leave its weight on the -// workspace permanently and the score could only ever climb. +// 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