feat(security-contacts): resolve GitHub handles to emails using cdp identities [CM-1323]

scripts/services/security-contacts-worker.yaml

@@ -6,7 +6,7 @@ x-env-args: &env-args
   SERVICE: security-contacts-worker
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
-  CROWD_TEMPORAL_TASKQUEUE: packages-worker
+  CROWD_TEMPORAL_TASKQUEUE: security-contacts-worker
   CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
 
 services:

services/apps/packages_worker/package.json

@@ -43,9 +43,9 @@
     "start:go-worker": "CROWD_TEMPORAL_TASKQUEUE=go-worker SERVICE=go-worker tsx src/bin/go-worker.ts",
     "dev:go-worker": "CROWD_TEMPORAL_TASKQUEUE=go-worker SERVICE=go-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9241 src/bin/go-worker.ts",
     "dev:go-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=go-worker SERVICE=go-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9241 src/bin/go-worker.ts",
-    "start:security-contacts-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker SERVICE=security-contacts-worker tsx src/bin/security-contacts-worker.ts",
-    "dev:security-contacts-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker SERVICE=security-contacts-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9243 src/bin/security-contacts-worker.ts",
-    "dev:security-contacts-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=packages-worker SERVICE=security-contacts-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9243 src/bin/security-contacts-worker.ts",
+    "start:security-contacts-worker": "CROWD_TEMPORAL_TASKQUEUE=security-contacts-worker SERVICE=security-contacts-worker tsx src/bin/security-contacts-worker.ts",
+    "dev:security-contacts-worker": "CROWD_TEMPORAL_TASKQUEUE=security-contacts-worker SERVICE=security-contacts-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9243 src/bin/security-contacts-worker.ts",
+    "dev:security-contacts-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=security-contacts-worker SERVICE=security-contacts-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9243 src/bin/security-contacts-worker.ts",
     "start:nuget-worker": "CROWD_TEMPORAL_TASKQUEUE=nuget-worker SERVICE=nuget-worker tsx src/bin/nuget-worker.ts",
     "dev:nuget-worker": "CROWD_TEMPORAL_TASKQUEUE=nuget-worker SERVICE=nuget-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9242 src/bin/nuget-worker.ts",
     "dev:nuget-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=nuget-worker SERVICE=nuget-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9242 src/bin/nuget-worker.ts",

services/apps/packages_worker/src/config.ts

@@ -18,6 +18,16 @@ export function getPackagesDbConfig() {
   }
 }
 
+export function getCdpDbConfig() {
+  return {
+    host: requireEnv('CROWD_DB_READ_HOST'),
+    port: requireEnvInt('CROWD_DB_PORT'),
+    database: requireEnv('CROWD_DB_DATABASE'),
+    user: requireEnv('CROWD_DB_USERNAME'),
+    password: requireEnv('CROWD_DB_PASSWORD'),
+  }
+}
+
 export function getGithubAppConfig() {
   const rawPrivateKey = requireEnv('CROWD_GITHUB_PRIVATE_KEY')
   const privateKeyPem = Buffer.from(rawPrivateKey, 'base64').toString('ascii')

services/apps/packages_worker/src/db.ts

@@ -1,13 +1,18 @@
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { DbConnection, getDbConnection } from '@crowd/database'
 
-import { getPackagesDbConfig } from './config'
+import { getCdpDbConfig, getPackagesDbConfig } from './config'
 
 export async function getPackagesDb() {
   const conn = await getDbConnection(getPackagesDbConfig())
   return pgpQx(conn)
 }
 
+export async function getCdpDb() {
+  const conn = await getDbConnection(getCdpDbConfig())
+  return pgpQx(conn)
+}
+
 // Raw pg-promise connection for the same pool getPackagesDb() wraps
 // (getDbConnection caches per host:database). Needed for COPY FROM STDIN via
 // pg-copy-streams, which requires direct access to the underlying pg client.

services/apps/packages_worker/src/security-contacts/activities.ts

@@ -1,7 +1,7 @@
 import { getServiceChildLogger } from '@crowd/logging'
 
 import { getSecurityContactsConfig } from '../config'
-import { getPackagesDb } from '../db'
+import { getCdpDb, getPackagesDb } from '../db'
 
 import { IngestSingleResult, ingestSecurityContactsForPurl } from './ingestSingle'
 import { BatchResult, processBatch } from './processBatch'
@@ -11,8 +11,9 @@ const log = getServiceChildLogger('security-contacts-activity')
 export async function processSecurityContactsBatch(): Promise<BatchResult> {
   const config = getSecurityContactsConfig()
   const qx = await getPackagesDb()
+  const cdpQx = await getCdpDb()
 
-  const result = await processBatch(qx, config)
+  const result = await processBatch(qx, cdpQx, config)
   log.info({ ...result }, 'Security contacts batch activity complete')
   return result
 }
@@ -22,8 +23,9 @@ export async function ingestSecurityContactsForPurlActivity(
 ): Promise<IngestSingleResult> {
   const config = getSecurityContactsConfig()
   const qx = await getPackagesDb()
+  const cdpQx = await getCdpDb()
 
-  const result = await ingestSecurityContactsForPurl(qx, config, purl)
+  const result = await ingestSecurityContactsForPurl(qx, cdpQx, config, purl)
   log.info({ purl, ...result }, 'On-demand security contacts ingest activity complete')
   return result
 }

services/apps/packages_worker/src/security-contacts/ingestSingle.ts

@@ -80,6 +80,7 @@ function toTarget(row: SingleRepoRow): RepoTarget {
  */
 export async function ingestSecurityContactsForPurl(
   qx: QueryExecutor,
+  cdpQx: QueryExecutor,
   config: Config,
   purl: string,
 ): Promise<IngestSingleResult> {
@@ -92,7 +93,7 @@ export async function ingestSecurityContactsForPurl(
   const target = toTarget(row)
   const baseDeps = buildBaseDeps(config)
   try {
-    await processRepo(target, baseDeps, qx)
+    await processRepo(target, baseDeps, qx, cdpQx)
   } catch (err) {
     log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed')
     await markRepoAttempted(qx, target.repoId).catch(() => undefined)

services/apps/packages_worker/src/security-contacts/processBatch.ts

@@ -16,6 +16,7 @@ import { extractSecurityMd } from './extractors/securityMd'
 import { extractSecurityTxt } from './extractors/securityTxt'
 import { githubApiGet } from './githubToken'
 import { reconcile } from './reconcile'
+import { resolveCdpEmails } from './resolveCdpEmails'
 import {
   Extractor,
   ExtractorDeps,
@@ -120,6 +121,7 @@ export async function processRepo(
   target: RepoTarget,
   baseDeps: Omit<ExtractorDeps, 'repoTree'>,
   qx: QueryExecutor,
+  cdpQx: QueryExecutor,
 ): Promise<void> {
   // One tree fetch per repo, shared by extractors that probe well-known paths.
   let repoTree: ExtractorDeps['repoTree'] = { paths: null }
@@ -161,11 +163,27 @@ export async function processRepo(
     contacts = contacts.filter((c) => c.channel !== 'github-pvr')
   }
 
+  const handleContacts = contacts.filter((c) => c.channel === 'github-handle')
+  if (handleContacts.length > 0) {
+    try {
+      contacts.push(...(await resolveCdpEmails(cdpQx, handleContacts)))
+    } catch (err) {
+      log.warn(
+        { repoId: target.repoId, errMsg: (err as Error).message },
+        'CDP email resolution failed — proceeding without resolved emails',
+      )
+    }
+  }
+
   const scored = reconcile(contacts)
   await writeContacts(qx, target.repoId, scored, policies)
 }
 
-export async function processBatch(qx: QueryExecutor, config: Config): Promise<BatchResult> {
+export async function processBatch(
+  qx: QueryExecutor,
+  cdpQx: QueryExecutor,
+  config: Config,
+): Promise<BatchResult> {
   const batch = await fetchBatch(qx)
   if (batch.length === 0) return { processed: 0 }
 
@@ -191,7 +209,7 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise<B
         )
       }
       try {
-        await processRepo(target, deps, qx)
+        await processRepo(target, deps, qx, cdpQx)
       } catch (err) {
         log.error(
           { repoId: target.repoId, errMsg: (err as Error).message },

services/apps/packages_worker/src/security-contacts/reconcile.ts

@@ -8,8 +8,6 @@ import {
   SourceTier,
 } from './types'
 
-const MAX_CONTACTS = 5
-
 const ROLE_PRIORITY: Record<ContactRole, number> = {
   'security-team': 5,
   maintainer: 4,
@@ -143,5 +141,5 @@ export function reconcile(contacts: RawContact[], now: Date = new Date()): Score
       a.value.localeCompare(b.value),
   )
 
-  return scored.slice(0, MAX_CONTACTS)
+  return scored
 }

services/apps/packages_worker/src/security-contacts/resolveCdpEmails.ts

@@ -0,0 +1,59 @@
+import {
+  findMembersByGithubHandles,
+  findResolvableEmailsForMembers,
+} from '@crowd/data-access-layer/src/members/identities'
+import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+
+import { ProvenanceEntry, RawContact } from './types'
+
+function latestTimestamp(provenance: ProvenanceEntry[]): string {
+  const times = provenance.map((p) => p.declaredAt ?? p.fetchedAt)
+  return times.length === 0
+    ? new Date().toISOString()
+    : times.reduce((a, b) => (new Date(b).getTime() > new Date(a).getTime() ? b : a))
+}
+
+export async function resolveCdpEmails(
+  cdpQx: QueryExecutor,
+  handleContacts: RawContact[],
+): Promise<RawContact[]> {
+  if (handleContacts.length === 0) return []
+
+  const handles = [...new Set(handleContacts.map((c) => c.value.toLowerCase()))]
+  const members = await findMembersByGithubHandles(cdpQx, handles)
+  if (members.length === 0) return []
+
+  const memberIdsByHandle = new Map<string, string[]>()
+  for (const m of members) {
+    const key = m.githubHandle.toLowerCase()
+    memberIdsByHandle.set(key, [...(memberIdsByHandle.get(key) ?? []), m.memberId])
+  }
+
+  const emails = await findResolvableEmailsForMembers(cdpQx, [
+    ...new Set(members.map((m) => m.memberId)),
+  ])
+  const emailsByMember = new Map<string, { verified: string[]; unverified: string[] }>()
+  for (const e of emails) {
+    const bucket = emailsByMember.get(e.memberId) ?? { verified: [], unverified: [] }
+    ;(e.verified ? bucket.verified : bucket.unverified).push(e.email)
+    emailsByMember.set(e.memberId, bucket)
+  }
+
+  return handleContacts.flatMap((contact) => {
+    const fetchedAt = latestTimestamp(contact.provenance)
+    const memberIds = memberIdsByHandle.get(contact.value.toLowerCase()) ?? []
+    return memberIds.flatMap((memberId) => {
+      const bucket = emailsByMember.get(memberId)
+      if (!bucket) return []
+      const useVerified = bucket.verified.length > 0
+      const source = useVerified ? 'cdp-verified' : 'cdp-unverified'
+      return (useVerified ? bucket.verified : bucket.unverified).map((email) => ({
+        channel: 'email' as const,
+        value: email,
+        role: contact.role,
+        tier: contact.tier,
+        provenance: [{ source, sourceTier: contact.tier, fetchedAt }],
+      }))
+    })
+  })
+}

services/apps/packages_worker/src/security-contacts/schedule.ts

@@ -13,7 +13,7 @@ function scheduleAction() {
     type: 'startWorkflow' as const,
     workflowType: ingestSecurityContacts,
     workflowId: 'security-contacts-daily',
-    taskQueue: 'packages-worker',
+    taskQueue: 'security-contacts-worker',
     workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT,
     retry: {
       initialInterval: '30 seconds',

services/apps/packages_worker/src/security-contacts/score.ts

@@ -3,7 +3,7 @@ import {
   securityContactConfidenceBand,
 } from '@crowd/data-access-layer/src/osspckgs/api'
 
-import { ContactChannel, ProvenanceEntry, RawContact, SourceTier } from './types'
+import { ProvenanceEntry, RawContact, SourceTier } from './types'
 
 const WEIGHTS = { tier: 0.55, channel: 0.2, freshness: 0.15, corroboration: 0.1 }
 
@@ -12,6 +12,9 @@ const TIER_SCORE: Record<SourceTier, number> = { A: 1.0, B: 0.7, C: 0.4, D: 0.2
 // github-handle contacts carry no resolved email; nudge them below an equivalent email
 const HANDLE_ONLY_PENALTY = 0.05
 
+const CDP_UNVERIFIED_CHANNEL_QUALITY = 0.35
+const CDP_UNVERIFIED_PENALTY = 0.25
+
 const FRESH_DAYS = 90
 const STALE_DAYS = 730
 const DAY_MS = 24 * 60 * 60 * 1000
@@ -55,10 +58,11 @@ function emailQuality(value: string): number {
   return 0.6
 }
 
-function channelQuality(channel: ContactChannel, value: string): number {
-  switch (channel) {
+function channelQuality(contact: RawContact): number {
+  if (isCdpUnverified(contact.provenance)) return CDP_UNVERIFIED_CHANNEL_QUALITY
+  switch (contact.channel) {
     case 'email':
-      return emailQuality(value)
+      return emailQuality(contact.value)
     case 'github-pvr':
       return 0.95
     case 'web-form':
@@ -81,23 +85,35 @@ function freshnessScore(provenance: ProvenanceEntry[], now: Date): number {
   return 1 - (ageDays - FRESH_DAYS) / (STALE_DAYS - FRESH_DAYS)
 }
 
-// Independent = distinct extractors, not the same file re-fetched
+// Independent = distinct extractors, not the same file re-fetched. A cdp-* source is an
+// identity lookup of the same value, not an independent attestation, so it never corroborates.
 function corroborationScore(provenance: ProvenanceEntry[]): number {
-  const sources = new Set(provenance.map((p) => p.source))
+  const sources = new Set(provenance.filter((p) => !isCdpSource(p)).map((p) => p.source))
   if (sources.size >= 3) return 1.0
   if (sources.size === 2) return 0.5
   return 0
 }
 
+function isCdpSource(p: ProvenanceEntry): boolean {
+  return p.source.startsWith('cdp-')
+}
+
+// Unverified only when every source is a cdp-unverified lookup. Any cdp-verified resolution or
+// real-extractor source (e.g. a shared email merged from another owner) lifts the penalty.
+function isCdpUnverified(provenance: ProvenanceEntry[]): boolean {
+  return provenance.length > 0 && provenance.every((p) => p.source === 'cdp-unverified')
+}
+
 export function scoreContact(
   contact: RawContact,
   now: Date = new Date(),
 ): { score: number; confidence: SecurityContactConfidence } {
   const raw =
     WEIGHTS.tier * TIER_SCORE[contact.tier] +
-    WEIGHTS.channel * channelQuality(contact.channel, contact.value) +
+    WEIGHTS.channel * channelQuality(contact) +
     WEIGHTS.freshness * freshnessScore(contact.provenance, now) +
     WEIGHTS.corroboration * corroborationScore(contact.provenance) -
+    (isCdpUnverified(contact.provenance) ? CDP_UNVERIFIED_PENALTY : 0) -
     (contact.channel === 'github-handle' ? HANDLE_ONLY_PENALTY : 0)
 
   const score = Math.round(Math.min(1, Math.max(0, raw)) * 1000) / 1000

services/libs/data-access-layer/src/members/identities.ts

@@ -712,6 +712,45 @@ export async function findMembersByGithubHandles(
   )
 }
 
+export async function findResolvableEmailsForMembers(
+  qx: QueryExecutor,
+  memberIds: string[],
+): Promise<{ memberId: string; email: string; verified: boolean }[]> {
+  if (memberIds.length === 0) return []
+  return qx.select(
+    `
+      WITH emails AS (
+        SELECT "memberId", lower(value) AS email, bool_or(verified) AS verified
+        FROM "memberIdentities"
+        WHERE "memberId" IN ($(memberIds:csv))
+          AND type = $(emailType)
+          AND value NOT ILIKE '%@users.noreply.github.com'
+          AND "deletedAt" IS NULL
+        GROUP BY "memberId", lower(value)
+      ),
+      "usernameTwins" AS (
+        SELECT "memberId", lower(value) AS email
+        FROM "memberIdentities"
+        WHERE "memberId" IN ($(memberIds:csv))
+          AND type = $(usernameType)
+          AND verified = true
+          AND "deletedAt" IS NULL
+      )
+      SELECT
+        e."memberId",
+        e.email,
+        (e.verified OR t.email IS NOT NULL) AS verified
+      FROM emails e
+      LEFT JOIN "usernameTwins" t ON t."memberId" = e."memberId" AND t.email = e.email
+    `,
+    {
+      memberIds,
+      emailType: MemberIdentityType.EMAIL,
+      usernameType: MemberIdentityType.USERNAME,
+    },
+  )
+}
+
 export async function findVerifiedEmailsByMemberIds(
   qx: QueryExecutor,
   memberIds: string[],