diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index db4196f8..8c46fca5 100644 --- a/cmd/consumer/main.go +++ b/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 diff --git a/docs/content/docs/guides/security.mdx b/docs/content/docs/guides/security.mdx index 2a4e6017..93edc139 100644 --- a/docs/content/docs/guides/security.mdx +++ b/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. +## 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..3d2a4974 --- /dev/null +++ b/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") + } + } +} 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 new file mode 100644 index 00000000..a3b86a35 --- /dev/null +++ b/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) + } + } + } +} diff --git a/internal/repository/pg_correlation.go b/internal/repository/pg_correlation.go new file mode 100644 index 00000000..e3fa34f9 --- /dev/null +++ b/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() +} 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() +}