fix: code review fixes

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

services/apps/packages_worker/src/security-contacts/extractors/http.ts

@@ -7,6 +7,10 @@ export function registryHeaders(userAgent: string): Record<string, string> {
   return { 'User-Agent': userAgent }
 }
 
+// Genuinely-absent → null body; every other non-200 throws so transient failures (429/5xx/...)
+// are treated as failures and the pipeline preserves existing data instead of wiping it.
+const ABSENT_STATUSES = new Set([404, 410])
+
 export interface FetchTextResult {
   status: number
   text: string | null
@@ -21,8 +25,9 @@ export async function fetchText(
   const timeoutId = setTimeout(() => controller.abort(), timeoutMs)
   try {
     const res = await fetch(url, { headers, signal: controller.signal })
-    if (res.status !== 200) return { status: res.status, text: null }
-    return { status: 200, text: await res.text() }
+    if (res.status === 200) return { status: 200, text: await res.text() }
+    if (ABSENT_STATUSES.has(res.status)) return { status: res.status, text: null }
+    throw new Error(`fetchText ${url} failed: HTTP ${res.status}`)
   } finally {
     clearTimeout(timeoutId)
   }
@@ -42,8 +47,9 @@ export async function fetchJson(
   const timeoutId = setTimeout(() => controller.abort(), timeoutMs)
   try {
     const res = await fetch(url, { headers, signal: controller.signal })
-    if (res.status !== 200) return { status: res.status, json: null }
-    return { status: 200, json: await res.json() }
+    if (res.status === 200) return { status: 200, json: await res.json() }
+    if (ABSENT_STATUSES.has(res.status)) return { status: res.status, json: null }
+    throw new Error(`fetchJson ${url} failed: HTTP ${res.status}`)
   } finally {
     clearTimeout(timeoutId)
   }

services/apps/packages_worker/src/security-contacts/extractors/securityContactsFile.ts

@@ -73,11 +73,11 @@ export const extractSecurityContactsFile: Extractor = async (target, deps) => {
     const email =
       entry.email ?? (await resolvePublicEmail(entry.handle, deps.fetchTimeoutMs, token))
     if (email) {
-      // name = handle so the reconciler can identity-link this to a bare handle from another source
+      // handle = the username this email was resolved from (used for identity-linking).
       contacts.push({
         channel: 'email',
         value: email,
-        name: entry.handle,
+        handle: entry.handle,
         role: 'security-team',
         tier: 'A',
         provenance: prov(),

services/apps/packages_worker/src/security-contacts/extractors/securityTxt.ts

@@ -9,6 +9,12 @@ const PLATFORM_HOSTS = new Set(['github.com', 'www.github.com', 'gitlab.com', 'b
 
 const FIELD_RE = /^([A-Za-z-]+):\s*(.+?)\s*$/
 
+// target.homepage is externally-sourced. Requiring https already blocks the classic SSRF target
+// (cloud-metadata IMDS is http-only); we also reject obvious loopback/localhost.
+function isBlockedHost(h: string): boolean {
+  return h === 'localhost' || h === '::1' || h === '0.0.0.0' || h.startsWith('127.')
+}
+
 export function parseSecurityTxt(
   text: string,
   sourceUrl: string,
@@ -65,12 +71,13 @@ export const extractSecurityTxt: Extractor = async (target, deps) => {
   let host: string
   try {
     const u = new URL(target.homepage)
+    if (u.protocol !== 'https:') return { contacts: [], policies: {} }
     origin = u.origin
     host = u.hostname.toLowerCase()
   } catch {
     return { contacts: [], policies: {} }
   }
-  if (PLATFORM_HOSTS.has(host)) return { contacts: [], policies: {} }
+  if (PLATFORM_HOSTS.has(host) || isBlockedHost(host)) return { contacts: [], policies: {} }
 
   const url = `${origin}/.well-known/security.txt`
   const { text } = await fetchText(url, deps.fetchTimeoutMs)

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

@@ -2,6 +2,7 @@ import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 import { getServiceChildLogger } from '@crowd/logging'
 
 import { getSecurityContactsConfig } from '../config'
+import { mapWithConcurrency } from '../utils/concurrency'
 
 import { extractPvr } from './extractors/pvr'
 import { extractManifest } from './extractors/registry'
@@ -19,7 +20,7 @@ import {
   RepoPolicies,
   RepoTarget,
 } from './types'
-import { writeContacts } from './writeContacts'
+import { markRepoAttempted, writeContacts } from './writeContacts'
 
 const log = getServiceChildLogger('security-contacts')
 
@@ -101,22 +102,6 @@ function toTarget(row: SweepRow): RepoTarget {
   return { repoId: row.id, url: row.url, homepage: row.homepage, packages: row.packages ?? [] }
 }
 
-async function runWithConcurrency<T>(
-  items: T[],
-  concurrency: number,
-  task: (item: T) => Promise<void>,
-): Promise<void> {
-  let idx = 0
-  await Promise.all(
-    Array.from({ length: Math.min(concurrency, items.length) }, async () => {
-      while (idx < items.length) {
-        const item = items[idx++]
-        await task(item)
-      }
-    }),
-  )
-}
-
 async function processRepo(
   target: RepoTarget,
   deps: ExtractorDeps,
@@ -125,10 +110,10 @@ async function processRepo(
   // One failing extractor must not sink the repo.
   const results = await Promise.allSettled(EXTRACTORS.map((extract) => extract(target, deps)))
 
-  // If every extractor failed (e.g. a transient network outage), skip the write entirely so we
-  // don't delete previously good contacts/policies; leaving contacts_last_refreshed retries next sweep.
+  // Every extractor failed (transient outage) — preserve existing data, just record the attempt.
   if (results.every((r) => r.status === 'rejected')) {
-    log.warn({ repoId: target.repoId }, 'All extractors failed — skipping write to preserve data')
+    log.warn({ repoId: target.repoId }, 'All extractors failed — preserving existing data')
+    await markRepoAttempted(qx, target.repoId)
     return
   }
 
@@ -168,7 +153,7 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise<B
   }
 
   const targets = batch.map(toTarget)
-  await runWithConcurrency(targets, CONCURRENCY, (target) => processRepo(target, deps, qx))
+  await mapWithConcurrency(targets, CONCURRENCY, (target) => processRepo(target, deps, qx))
 
   log.info({ processed: targets.length }, 'Security contacts batch complete')
   return { processed: targets.length }

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

@@ -50,6 +50,7 @@ function mergeInto(target: RawContact, src: RawContact): void {
   target.role = higherRole(target.role, src.role)
   target.tier = higherTier(target.tier, src.tier)
   if (!target.name && src.name) target.name = src.name
+  if (!target.handle && src.handle) target.handle = src.handle
 }
 
 function exactMatchMerge(contacts: RawContact[]): RawContact[] {
@@ -66,13 +67,12 @@ function exactMatchMerge(contacts: RawContact[]): RawContact[] {
   return [...byKey.values()]
 }
 
-// Collapse a resolved handle into its email: extractors set the email contact's
-// `name` to the originating handle, so a github-handle whose value matches becomes
-// redundant once the email is known. The email (higher-quality) channel is kept.
+// Collapse a bare github-handle into the email A3 resolved it from, matched via the explicit
+// `handle` field (never the display name, to avoid merging unrelated people who share a name).
 function identityLinkMerge(contacts: RawContact[]): RawContact[] {
   const emailByHandle = new Map<string, RawContact>()
   for (const c of contacts) {
-    if (c.channel === 'email' && c.name) emailByHandle.set(c.name.toLowerCase(), c)
+    if (c.channel === 'email' && c.handle) emailByHandle.set(c.handle.toLowerCase(), c)
   }
 
   const out: RawContact[] = []

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

@@ -25,6 +25,8 @@ export interface RawContact {
   value: string
   role: ContactRole
   name?: string
+  /** Username an A3 email was resolved from — used only to identity-link a bare github-handle. */
+  handle?: string
   tier: SourceTier
   provenance: ProvenanceEntry[]
 }

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

@@ -7,6 +7,14 @@ import { RepoPolicies, ScoredContact } from './types'
  * the policy columns in one transaction. Policy columns use COALESCE so a run that doesn't
  * rediscover a field (partial/failed extractor pass) never clears a previously known value.
  */
+// Records the attempt without touching contacts/policies — preserves existing data on total
+// failure while advancing contacts_last_refreshed so the repo isn't reprocessed this sweep.
+export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Promise<void> {
+  await qx.result('UPDATE repos SET contacts_last_refreshed = NOW() WHERE id = $(repoId)', {
+    repoId,
+  })
+}
+
 export async function writeContacts(
   qx: QueryExecutor,
   repoId: string,