diff --git a/backend/.env.dist.local b/backend/.env.dist.local index adca4976c3..3fdede3c01 100755 --- a/backend/.env.dist.local +++ b/backend/.env.dist.local @@ -151,6 +151,7 @@ CROWD_TEMPORAL_ENCRYPTION_KEY_ID=local CROWD_TEMPORAL_ENCRYPTION_KEY=FweBMRnGCLshER8FlSvNusQA6G3MRUKt # Temporal — packages namespace +CROWD_PACKAGES_TEMPORAL_SERVER_URL=localhost:7233 CROWD_PACKAGES_TEMPORAL_NAMESPACE=default # Seach sync api diff --git a/backend/config/custom-environment-variables.json b/backend/config/custom-environment-variables.json index 4d1d01b6f5..93de9c67da 100644 --- a/backend/config/custom-environment-variables.json +++ b/backend/config/custom-environment-variables.json @@ -180,6 +180,12 @@ "certificate": "CROWD_TEMPORAL_CERTIFICATE", "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY" }, + "packagesTemporal": { + "serverUrl": "CROWD_PACKAGES_TEMPORAL_SERVER_URL", + "namespace": "CROWD_PACKAGES_TEMPORAL_NAMESPACE", + "certificate": "CROWD_TEMPORAL_CERTIFICATE", + "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY" + }, "searchSyncApi": { "baseUrl": "CROWD_SEARCH_SYNC_API_URL" }, diff --git a/backend/config/default.json b/backend/config/default.json index e60af044e2..de0a7a9e53 100644 --- a/backend/config/default.json +++ b/backend/config/default.json @@ -43,6 +43,7 @@ "isEnabled": "false" }, "temporal": {}, + "packagesTemporal": {}, "searchSyncApi": {}, "encryption": {}, "openStatusApi": {}, diff --git a/backend/src/api/public/v1/packages/getPackage.ts b/backend/src/api/public/v1/packages/getPackage.ts index 44c61ca50f..a8b435d121 100644 --- a/backend/src/api/public/v1/packages/getPackage.ts +++ b/backend/src/api/public/v1/packages/getPackage.ts @@ -1,3 +1,4 @@ +import { createHash } from 'crypto' import type { Request, Response } from 'express' import { NotFoundError } from '@crowd/common' @@ -8,8 +9,10 @@ import { getStewardshipSummary, securityContactConfidenceBand, } from '@crowd/data-access-layer' +import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal' import { getPackagesQx } from '@/db/packagesDb' +import { getPackagesTemporalClient } from '@/db/packagesTemporal' import { ok } from '@/utils/api' import { validateOrThrow } from '@/utils/validation' @@ -32,16 +35,43 @@ function repoMappingLabel(confidence: number | null): 'High' | 'Medium' | 'Low' return 'Low' } +// Deterministic per-purl id: concurrent requests for the same purl attach to the +// same workflow run (WorkflowIdConflictPolicy.USE_EXISTING below) instead of each +// kicking off its own ingest. +function ondemandWorkflowId(purl: string): string { + return `security-contacts-ondemand/${createHash('sha1').update(purl).digest('hex')}` +} + export async function getPackage(req: Request, res: Response): Promise { const { purl } = validateOrThrow(purlQuerySchema, req.query) const qx = await getPackagesQx() - const pkg = await getPackageDetailByPurl(qx, purl) + let pkg = await getPackageDetailByPurl(qx, purl) if (!pkg) { throw new NotFoundError() } + if (pkg.contactsLastRefreshed == null) { + try { + const packagesTemporal = await getPackagesTemporalClient() + await packagesTemporal.workflow.execute('ingestSecurityContactsForPurlWorkflow', { + taskQueue: 'packages-worker', + workflowId: ondemandWorkflowId(purl), + workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, + workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING, + // Bounds the request's wait if packages-worker is down or its task queue is + // backed up — the activity's startToCloseTimeout only starts counting once a + // worker picks the task up, so without this the request could hang indefinitely. + workflowExecutionTimeout: '60 seconds', + args: [purl], + }) + pkg = (await getPackageDetailByPurl(qx, purl)) ?? pkg + } catch (err) { + req.log.warn(err, 'On-demand security contacts ingest failed — serving cached detail') + } + } + const [{ rows: advisories }, stewardshipSummary] = await Promise.all([ getAdvisoriesByPackageId(qx, pkg.id), pkg.stewardshipId ? getStewardshipSummary(qx, Number(pkg.stewardshipId)) : null, diff --git a/backend/src/conf/index.ts b/backend/src/conf/index.ts index 553167cde5..938ffb66c2 100644 --- a/backend/src/conf/index.ts +++ b/backend/src/conf/index.ts @@ -86,6 +86,14 @@ export const PACKAGES_DB_CONFIG: IDatabaseConfig | undefined = config.has('packa ? config.get('packagesDb') : undefined +// packages_worker (npm/maven/pypi/osv/security-contacts/...) runs in its own Temporal +// namespace, separate from the API's default namespace — see CROWD_PACKAGES_TEMPORAL_NAMESPACE. +export const PACKAGES_TEMPORAL_CONFIG: ITemporalConfig | undefined = config.has( + 'packagesTemporal.namespace', +) + ? config.get('packagesTemporal') + : undefined + export const SEGMENT_CONFIG: SegmentConfiguration = config.get('segment') export const COMPREHEND_CONFIG: ComprehendConfiguration = diff --git a/backend/src/db/packagesTemporal.ts b/backend/src/db/packagesTemporal.ts new file mode 100644 index 0000000000..4e5a981d1c --- /dev/null +++ b/backend/src/db/packagesTemporal.ts @@ -0,0 +1,45 @@ +import { IS_DEV_ENV, SERVICE } from '@crowd/common' +import { Client, Connection, getDataConverter } from '@crowd/temporal' + +import { PACKAGES_TEMPORAL_CONFIG } from '@/conf' + +let _init: Promise | undefined + +// Separate connection from the API's default req.temporal client — packages_worker +// (npm/maven/pypi/osv/security-contacts/...) polls task queues in its own Temporal +// namespace (CROWD_PACKAGES_TEMPORAL_NAMESPACE), not the API's default namespace. +export function getPackagesTemporalClient(): Promise { + if (!_init) { + if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) { + throw new Error('Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE') + } + + const cfg = PACKAGES_TEMPORAL_CONFIG + _init = Connection.connect({ + address: cfg.serverUrl, + tls: + cfg.certificate && cfg.privateKey + ? { + clientCertPair: { + crt: Buffer.from(cfg.certificate, 'base64'), + key: Buffer.from(cfg.privateKey, 'base64'), + }, + } + : undefined, + }) + .then( + async (connection) => + new Client({ + connection, + namespace: cfg.namespace, + identity: SERVICE, + dataConverter: IS_DEV_ENV ? undefined : await getDataConverter(), + }), + ) + .catch((err) => { + _init = undefined + throw err + }) + } + return _init +} diff --git a/scripts/services/security-contacts-worker.yaml b/scripts/services/security-contacts-worker.yaml index 934afe2c69..4706d39425 100644 --- a/scripts/services/security-contacts-worker.yaml +++ b/scripts/services/security-contacts-worker.yaml @@ -7,6 +7,7 @@ x-env-args: &env-args SHELL: /bin/sh SUPPRESS_NO_CONFIG_WARNING: 'true' CROWD_TEMPORAL_TASKQUEUE: packages-worker + CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE} services: security-contacts-worker: diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index b747a3bd3d..5b035b52cb 100644 --- a/services/apps/packages_worker/src/activities.ts +++ b/services/apps/packages_worker/src/activities.ts @@ -32,4 +32,7 @@ export { } from './pypi/activities' export { getCriticalPypiCount } from './pypi/downloads/getCriticalPypiCount' export { processNuGetBatch } from './nuget/activities' -export { processSecurityContactsBatch } from './security-contacts/activities' +export { + processSecurityContactsBatch, + ingestSecurityContactsForPurlActivity, +} from './security-contacts/activities' diff --git a/services/apps/packages_worker/src/security-contacts/activities.ts b/services/apps/packages_worker/src/security-contacts/activities.ts index e6be0fecde..454d10e32f 100644 --- a/services/apps/packages_worker/src/security-contacts/activities.ts +++ b/services/apps/packages_worker/src/security-contacts/activities.ts @@ -3,6 +3,7 @@ import { getServiceChildLogger } from '@crowd/logging' import { getSecurityContactsConfig } from '../config' import { getPackagesDb } from '../db' +import { IngestSingleResult, ingestSecurityContactsForPurl } from './ingestSingle' import { BatchResult, processBatch } from './processBatch' const log = getServiceChildLogger('security-contacts-activity') @@ -15,3 +16,14 @@ export async function processSecurityContactsBatch(): Promise { log.info({ ...result }, 'Security contacts batch activity complete') return result } + +export async function ingestSecurityContactsForPurlActivity( + purl: string, +): Promise { + const config = getSecurityContactsConfig() + const qx = await getPackagesDb() + + const result = await ingestSecurityContactsForPurl(qx, config, purl) + log.info({ purl, ...result }, 'On-demand security contacts ingest activity complete') + return result +} diff --git a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts new file mode 100644 index 0000000000..61e162f0a2 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts @@ -0,0 +1,102 @@ +import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' +import { getServiceChildLogger } from '@crowd/logging' + +import { getSecurityContactsConfig } from '../config' + +import { buildBaseDeps, processRepo } from './processBatch' +import { RepoPackage, RepoTarget } from './types' +import { markRepoAttempted } from './writeContacts' + +const log = getServiceChildLogger('security-contacts-ondemand') + +type Config = ReturnType + +export interface IngestSingleResult { + found: boolean + repoId?: string +} + +interface SingleRepoRow { + repoId: string + url: string + homepage: string | null + archived: boolean | null + packages: RepoPackage[] | null +} + +// Mirrors the best-repo LATERAL selection in getPackageDetailByPurl +// (services/libs/data-access-layer/src/osspckgs/api.ts) so the repo we ingest is the +// one the read side surfaces. No `is_critical` filter — non-critical purls are exactly +// what this on-demand path exists for. `packages` aggregates every package linked to +// the repo (not just the requested purl) so extractors see every ecosystem, matching +// the shape the batch sweep builds. +// +// No host filter: processRepo already degrades gracefully for non-github repos (the +// github-specific extractors no-op, security.txt/registry-manifest extractors still run) — +// filtering here would leave non-github repos permanently NULL, re-triggering this +// on-demand path on every single request. +async function findBestRepoForPurl(qx: QueryExecutor, purl: string): Promise { + return qx.selectOneOrNone( + ` + SELECT r.id::text AS "repoId", + r.url, + r.homepage, + r.archived, + ( + SELECT json_agg(json_build_object('purl', p2.purl, 'ecosystem', p2.ecosystem)) + FROM package_repos pr2 + JOIN packages p2 ON p2.id = pr2.package_id + WHERE pr2.repo_id = r.id + ) AS packages + FROM packages p + JOIN LATERAL ( + SELECT pr.repo_id + FROM package_repos pr + WHERE pr.package_id = p.id + ORDER BY pr.confidence DESC, (pr.source = 'declared') DESC + LIMIT 1 + ) best ON true + JOIN repos r ON r.id = best.repo_id + WHERE p.purl = $(purl) + `, + { purl }, + ) +} + +function toTarget(row: SingleRepoRow): RepoTarget { + return { + repoId: row.repoId, + url: row.url, + homepage: row.homepage, + archived: row.archived, + packages: row.packages ?? [], + } +} + +/** + * On-demand ingest for a single purl, bypassing the daily critical-only sweep. + * Used when the akrites API hits a repo that has never been evaluated + * (repos.contacts_last_refreshed IS NULL). + */ +export async function ingestSecurityContactsForPurl( + qx: QueryExecutor, + config: Config, + purl: string, +): Promise { + const row = await findBestRepoForPurl(qx, purl) + if (!row) { + log.info({ purl }, 'On-demand security contacts: no linked repo found — skipping') + return { found: false } + } + + const target = toTarget(row) + const baseDeps = buildBaseDeps(config) + try { + await processRepo(target, baseDeps, qx) + } catch (err) { + log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed') + await markRepoAttempted(qx, target.repoId).catch(() => undefined) + } + + return { found: true, repoId: target.repoId } +} diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index 9033a8dfa5..3a178777c1 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/services/apps/packages_worker/src/security-contacts/processBatch.ts @@ -72,8 +72,7 @@ async function fetchBatch(qx: QueryExecutor): Promise { FROM repos r JOIN package_repos pr ON pr.repo_id = r.id JOIN packages p ON p.id = pr.package_id AND p.is_critical - WHERE r.host = 'github' - AND ( + WHERE ( -- never evaluated → always eligible r.contacts_last_refreshed IS NULL -- evaluated but no contacts found yet → retry on the daily cadence @@ -109,7 +108,15 @@ function toTarget(row: SweepRow): RepoTarget { } } -async function processRepo( +export function buildBaseDeps(config: Config): Omit { + return { + fetchTimeoutMs: FETCH_TIMEOUT_MS, + userAgent: config.userAgent, + githubGet: (path, opts) => githubApiGet(path, FETCH_TIMEOUT_MS, opts), + } +} + +export async function processRepo( target: RepoTarget, baseDeps: Omit, qx: QueryExecutor, @@ -162,11 +169,7 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise = { - fetchTimeoutMs: FETCH_TIMEOUT_MS, - userAgent: config.userAgent, - githubGet: (path, opts) => githubApiGet(path, FETCH_TIMEOUT_MS, opts), - } + const deps = buildBaseDeps(config) const targets = batch.map(toTarget) // Fixed-cadence heartbeat: a slow repo can outlast the 2-minute heartbeatTimeout even while diff --git a/services/apps/packages_worker/src/security-contacts/workflows.ts b/services/apps/packages_worker/src/security-contacts/workflows.ts index fa9c64b072..2bfff86514 100644 --- a/services/apps/packages_worker/src/security-contacts/workflows.ts +++ b/services/apps/packages_worker/src/security-contacts/workflows.ts @@ -1,6 +1,7 @@ import { continueAsNew, log, proxyActivities } from '@temporalio/workflow' import type * as activities from './activities' +import type { IngestSingleResult } from './ingestSingle' const acts = proxyActivities({ startToCloseTimeout: '30 minutes', @@ -14,6 +15,17 @@ const acts = proxyActivities({ }, }) +// Single-repo, synchronous, on-demand path: short bound (no continueAsNew/heartbeat — +// one repo's extractors, not a whole sweep), so a caller awaiting the result (e.g. the +// akrites API on a cache miss) doesn't hang. +const singleActs = proxyActivities({ + startToCloseTimeout: '45 seconds', + retry: { + initialInterval: '5 seconds', + maximumAttempts: 2, + }, +}) + export async function ingestSecurityContacts(): Promise { const result = await acts.processSecurityContactsBatch() if (result.processed === 0) { @@ -22,3 +34,9 @@ export async function ingestSecurityContacts(): Promise { } await continueAsNew() } + +export async function ingestSecurityContactsForPurlWorkflow( + purl: string, +): Promise { + return singleActs.ingestSecurityContactsForPurlActivity(purl) +} diff --git a/services/apps/packages_worker/src/workflows/index.ts b/services/apps/packages_worker/src/workflows/index.ts index 91cf431947..9b762282b4 100644 --- a/services/apps/packages_worker/src/workflows/index.ts +++ b/services/apps/packages_worker/src/workflows/index.ts @@ -25,4 +25,7 @@ export { ingestPypiDownloadsDaily, } from '../pypi/downloads/ingestPypiDownloads' export { ingestNuGetPackages } from '../nuget/workflows' -export { ingestSecurityContacts } from '../security-contacts/workflows' +export { + ingestSecurityContacts, + ingestSecurityContactsForPurlWorkflow, +} from '../security-contacts/workflows'