diff --git a/backend/src/api/public/v1/akrites-external/index.ts b/backend/src/api/public/v1/akrites-external/index.ts index 07bed2dd1b..20b8709150 100644 --- a/backend/src/api/public/v1/akrites-external/index.ts +++ b/backend/src/api/public/v1/akrites-external/index.ts @@ -12,26 +12,39 @@ import { getAkritesExternalContactDetailBatch } from '../packages/getAkritesExte import { getAkritesExternalPackageDetail } from '../packages/getAkritesExternalPackageDetail' import { getAkritesExternalPackageDetailBatch } from '../packages/getAkritesExternalPackageDetailBatch' import { getBlastRadiusJob } from '../packages/getBlastRadiusJob' +import { ingestAkritesExternalContactDetail } from '../packages/ingestAkritesExternalContactDetail' import { submitBlastRadiusJob } from '../packages/submitBlastRadiusJob' const rateLimiter = createRateLimiter({ max: 60, windowMs: 60 * 1000 }) -// Blast-radius jobs kick off a Temporal workflow per request, so they get their own, -// much stricter limiter — configurable via env so it can be tuned without a redeploy. -// Defaults to 5 requests/hour. -const blastRadiusRateLimitMax = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_MAX) -const blastRadiusRateLimitWindowMs = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_WINDOW_MS) +// Shared by every endpoint below that kicks off a Temporal workflow per request — those +// get their own, much stricter limiter than plain reads, configurable via env so it can +// be tuned without a redeploy. +function envTunableRateLimiter(envPrefix: string, defaultMax: number, defaultWindowMs: number) { + const max = Number(process.env[`${envPrefix}_MAX`]) + const windowMs = Number(process.env[`${envPrefix}_WINDOW_MS`]) + return createRateLimiter({ + max: Number.isSafeInteger(max) && max > 0 ? max : defaultMax, + windowMs: Number.isSafeInteger(windowMs) && windowMs > 0 ? windowMs : defaultWindowMs, + }) +} + +// Blast-radius jobs default to 5 requests/hour. +const blastRadiusRateLimiter = envTunableRateLimiter( + 'AKRITES_BLAST_RADIUS_RATE_LIMIT', + 5, + 60 * 60 * 1000, +) -const blastRadiusRateLimiter = createRateLimiter({ - max: - Number.isSafeInteger(blastRadiusRateLimitMax) && blastRadiusRateLimitMax > 0 - ? blastRadiusRateLimitMax - : 5, - windowMs: - Number.isSafeInteger(blastRadiusRateLimitWindowMs) && blastRadiusRateLimitWindowMs > 0 - ? blastRadiusRateLimitWindowMs - : 60 * 60 * 1000, -}) +// /contacts/ingest starts a Temporal workflow and blocks for it (worst case ~95s per +// attempt cycle, plus unbounded time waiting for a free worker slot — see +// security-contacts/workflows.ts's singleActs config), vs. the read-only /contacts/detail +// endpoints, so it gets its own limiter. Defaults to 20 requests/hour. +const contactIngestRateLimiter = envTunableRateLimiter( + 'AKRITES_CONTACT_INGEST_RATE_LIMIT', + 20, + 60 * 60 * 1000, +) export function akritesExternalRouter(): Router { const router = Router() @@ -62,11 +75,33 @@ export function akritesExternalRouter(): Router { // them via the packages scope. That scope isn't issued by Auth0 yet, so reuse the // closest issued one — READ_MAINTAINER_ROLES (maintainer data) — NOT READ_PACKAGES. // TODO: swap for cdp:maintainers:read once issued. + // requireScopes is applied per-route (not router-level) so each route can put its own + // rate limiter *before* the scope check — failed-auth requests still count against that + // route's quota — without forcing every route in this subrouter onto the same limiter + // instance. /ingest gets its own dedicated contactIngestRateLimiter instead of sharing + // the read endpoints' quota, matching the blast-radius jobs endpoint below. + const contactsScopes = [SCOPES.READ_MAINTAINER_ROLES] const contactsSubRouter = Router() - contactsSubRouter.use(rateLimiter) - contactsSubRouter.use(requireScopes([SCOPES.READ_MAINTAINER_ROLES])) - contactsSubRouter.get('/detail', safeWrap(getAkritesExternalContactDetail)) - contactsSubRouter.post(/^\/detail:batch\/?$/, safeWrap(getAkritesExternalContactDetailBatch)) + contactsSubRouter.get( + '/detail', + rateLimiter, + requireScopes(contactsScopes), + safeWrap(getAkritesExternalContactDetail), + ) + contactsSubRouter.post( + /^\/detail:batch\/?$/, + rateLimiter, + requireScopes(contactsScopes), + safeWrap(getAkritesExternalContactDetailBatch), + ) + // Sync, single-purl on-demand ingest — starts a Temporal workflow and blocks a while, + // so it gets the dedicated contactIngestRateLimiter, not the shared rateLimiter above. + contactsSubRouter.post( + '/ingest', + contactIngestRateLimiter, + requireScopes(contactsScopes), + safeWrap(ingestAkritesExternalContactDetail), + ) router.use('/contacts', contactsSubRouter) // TODO: the contract gates blast-radius behind a dedicated read:advisories scope diff --git a/backend/src/api/public/v1/akrites-external/openapi.yaml b/backend/src/api/public/v1/akrites-external/openapi.yaml index 03a89875fd..b41e96a6b0 100644 --- a/backend/src/api/public/v1/akrites-external/openapi.yaml +++ b/backend/src/api/public/v1/akrites-external/openapi.yaml @@ -1246,3 +1246,94 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + + /akrites-external/contacts/ingest: + post: + operationId: ingestContactDetail + summary: On-demand security-contacts ingest for a single PURL + description: > + Synchronous, single-purl only — no batch variant, since fanning this out over + many purls would multiply concurrent Temporal workflow starts. + + If the linked repo's security contacts have already been ingested at least + once, returns the existing ContactDetail immediately without triggering the + workflow. If the purl is unknown, or the package has no linked repo at all, + returns 404 immediately without triggering the workflow either. Otherwise + blocks while a single-repo Temporal workflow (ingestSecurityContactsForPurlWorkflow) + ingests the linked repo's security contacts — worst case ~95s per attempt + cycle, plus unbounded time waiting for a free worker slot — then returns the + same ContactDetail payload as GET /contacts/detail. Concurrent requests for + the same purl attach to the same running workflow instead of starting + duplicates. + + + Note: the underlying workflow itself has no "already ingested" gate — it + reprocesses and can replace a repo's contacts on every invocation. This + endpoint is what prevents repeat calls from re-triggering it for a purl + that's already been ingested. + + + Rate limited independently of the other /contacts endpoints — default + 20 requests/hour, tunable via AKRITES_CONTACT_INGEST_RATE_LIMIT_MAX / + AKRITES_CONTACT_INGEST_RATE_LIMIT_WINDOW_MS. + + + Not yet implemented: this is a synchronous, blocking call. Future work should + make this async — an accept-and-poll job, similar to POST /blast-radius/jobs — + so callers aren't held open for the duration of the ingest. + tags: [Contacts] + security: + - M2MBearer: + - read:maintainer-roles + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [purl] + properties: + purl: + type: string + example: pkg:npm/%40angular/core + responses: + '200': + description: >- + Security contact detail — either just ingested by this call, or + returned from a prior ingest without re-triggering the workflow. + content: + application/json: + schema: + $ref: '#/components/schemas/ContactDetail' + '400': + description: Malformed purl. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '401': + description: Missing or invalid bearer token. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '403': + description: Token missing read:maintainer-roles scope. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: >- + Purl is unknown, or the package has no linked repo to ingest + contacts from — returned without triggering the workflow. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '429': + description: Too many requests — rate-limited independently of the other endpoints. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' diff --git a/backend/src/api/public/v1/packages/akritesExternalContactDetail.test.ts b/backend/src/api/public/v1/packages/akritesExternalContactDetail.test.ts index 286bf71cfe..7927a8941c 100644 --- a/backend/src/api/public/v1/packages/akritesExternalContactDetail.test.ts +++ b/backend/src/api/public/v1/packages/akritesExternalContactDetail.test.ts @@ -18,6 +18,7 @@ function baseRow( declaredRepositoryUrl: 'https://github.com/lodash/lodash.git', resolvedRepositoryUrl: 'https://github.com/lodash/lodash', repoMappingConfidence: 0.9, + contactsLastRefreshed: '2024-01-01T00:00:00.000Z', securityContacts: [ { channel: 'email', diff --git a/backend/src/api/public/v1/packages/ingestAkritesExternalContactDetail.test.ts b/backend/src/api/public/v1/packages/ingestAkritesExternalContactDetail.test.ts new file mode 100644 index 0000000000..e21a082382 --- /dev/null +++ b/backend/src/api/public/v1/packages/ingestAkritesExternalContactDetail.test.ts @@ -0,0 +1,182 @@ +import type { Request, Response } from 'express' +import { describe, expect, it, vi } from 'vitest' + +import type { AkritesExternalContactDetailRow } from '@crowd/data-access-layer' +import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal' + +import { ingestAkritesExternalContactDetail } from './ingestAkritesExternalContactDetail' + +const { execute, getContactDetailsByPurls } = vi.hoisted(() => ({ + execute: vi.fn(), + getContactDetailsByPurls: vi.fn(), +})) + +vi.mock('@/db/packagesTemporal', () => ({ + getPackagesTemporalClient: vi.fn().mockResolvedValue({ workflow: { execute } }), +})) + +vi.mock('@/db/packagesDb', () => ({ + getPackagesQx: vi.fn().mockResolvedValue({}), +})) + +vi.mock('@crowd/data-access-layer', () => ({ + getContactDetailsByPurls, +})) + +function baseRow( + overrides: Partial = {}, +): AkritesExternalContactDetailRow { + return { + purl: 'pkg:npm/lodash', + name: 'lodash', + ecosystem: 'npm', + securityPolicyUrl: null, + vulnerabilityReportingUrl: null, + bugBountyUrl: null, + pvrEnabled: null, + declaredRepositoryUrl: null, + resolvedRepositoryUrl: null, + repoMappingConfidence: null, + contactsLastRefreshed: null, + securityContacts: [], + ...overrides, + } +} + +function mockReqRes(body: unknown) { + execute.mockClear() + getContactDetailsByPurls.mockClear() + + const req = { body } as unknown as Request + + const json = vi.fn() + const status = vi.fn().mockReturnValue({ json }) + const res = { status, json } as unknown as Response + + return { req, res, status, json } +} + +describe('ingestAkritesExternalContactDetail', () => { + it('returns the existing contact detail without triggering the workflow when already ingested', async () => { + getContactDetailsByPurls.mockResolvedValue([ + baseRow({ contactsLastRefreshed: '2024-01-01T00:00:00.000Z' }), + ]) + + const { req, res, json } = mockReqRes({ purl: 'pkg:npm/lodash' }) + + await ingestAkritesExternalContactDetail(req, res) + + expect(execute).not.toHaveBeenCalled() + expect(getContactDetailsByPurls).toHaveBeenCalledTimes(1) + expect(json).toHaveBeenCalledWith(expect.objectContaining({ purl: 'pkg:npm/lodash' })) + }) + + it('executes ingestSecurityContactsForPurlWorkflow and returns the re-read contact detail when never ingested', async () => { + execute.mockResolvedValue({ found: true, repoId: 'repo-1' }) + getContactDetailsByPurls + .mockResolvedValueOnce([ + baseRow({ + resolvedRepositoryUrl: 'https://github.com/lodash/lodash', + contactsLastRefreshed: null, + }), + ]) + .mockResolvedValueOnce([baseRow({ contactsLastRefreshed: '2024-01-01T00:00:00.000Z' })]) + + const { req, res, json } = mockReqRes({ purl: 'pkg:npm/lodash' }) + + await ingestAkritesExternalContactDetail(req, res) + + expect(execute).toHaveBeenCalledTimes(1) + const [workflowType, options] = execute.mock.calls[0] + expect(workflowType).toBe('ingestSecurityContactsForPurlWorkflow') + expect(options.taskQueue).toBe('security-contacts-worker') + expect(options.workflowId).toMatch(/^security-contacts-ondemand:[0-9a-f]{64}$/) + expect(options.workflowIdConflictPolicy).toBe(WorkflowIdConflictPolicy.USE_EXISTING) + expect(options.workflowIdReusePolicy).toBe(WorkflowIdReusePolicy.ALLOW_DUPLICATE) + expect(options.args).toEqual(['pkg:npm/lodash']) + + expect(getContactDetailsByPurls).toHaveBeenCalledTimes(2) + expect(getContactDetailsByPurls).toHaveBeenCalledWith(expect.anything(), ['pkg:npm/lodash']) + expect(json).toHaveBeenCalledWith(expect.objectContaining({ purl: 'pkg:npm/lodash' })) + }) + + it('derives the same deterministic workflowId for the same purl', async () => { + execute.mockResolvedValue({ found: true }) + getContactDetailsByPurls.mockResolvedValue([ + baseRow({ + resolvedRepositoryUrl: 'https://github.com/lodash/lodash', + contactsLastRefreshed: null, + }), + ]) + + const { req: req1, res: res1 } = mockReqRes({ purl: 'pkg:npm/lodash' }) + await ingestAkritesExternalContactDetail(req1, res1) + const id1 = execute.mock.calls[0][1].workflowId + + const { req: req2, res: res2 } = mockReqRes({ purl: 'pkg:npm/lodash' }) + await ingestAkritesExternalContactDetail(req2, res2) + const id2 = execute.mock.calls[0][1].workflowId + + expect(id1).toBe(id2) + }) + + it('throws NotFoundError without executing the workflow when the purl is unknown', async () => { + getContactDetailsByPurls.mockResolvedValue([]) + + const { req, res } = mockReqRes({ purl: 'pkg:npm/left-pad' }) + + await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow() + expect(execute).not.toHaveBeenCalled() + expect(getContactDetailsByPurls).toHaveBeenCalledTimes(1) + }) + + it('throws NotFoundError without executing the workflow when the package has no linked repo', async () => { + getContactDetailsByPurls.mockResolvedValue([ + baseRow({ resolvedRepositoryUrl: null, contactsLastRefreshed: null }), + ]) + + const { req, res } = mockReqRes({ purl: 'pkg:npm/left-pad' }) + + await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow() + expect(execute).not.toHaveBeenCalled() + expect(getContactDetailsByPurls).toHaveBeenCalledTimes(1) + }) + + it('throws NotFoundError when the workflow reports no linked repo', async () => { + execute.mockResolvedValue({ found: false }) + getContactDetailsByPurls.mockResolvedValue([ + baseRow({ + resolvedRepositoryUrl: 'https://github.com/example/left-pad', + contactsLastRefreshed: null, + }), + ]) + + const { req, res } = mockReqRes({ purl: 'pkg:npm/left-pad' }) + + await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow() + expect(getContactDetailsByPurls).toHaveBeenCalledTimes(1) + }) + + it('throws NotFoundError when the re-read finds no row', async () => { + execute.mockResolvedValue({ found: true }) + getContactDetailsByPurls + .mockResolvedValueOnce([ + baseRow({ + resolvedRepositoryUrl: 'https://github.com/lodash/lodash', + contactsLastRefreshed: null, + }), + ]) + .mockResolvedValueOnce([]) + + const { req, res } = mockReqRes({ purl: 'pkg:npm/lodash' }) + + await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow() + }) + + it('rejects a request missing purl without executing a workflow', async () => { + const { req, res } = mockReqRes({}) + + await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow() + expect(execute).not.toHaveBeenCalled() + }) +}) diff --git a/backend/src/api/public/v1/packages/ingestAkritesExternalContactDetail.ts b/backend/src/api/public/v1/packages/ingestAkritesExternalContactDetail.ts new file mode 100644 index 0000000000..06a2fb15de --- /dev/null +++ b/backend/src/api/public/v1/packages/ingestAkritesExternalContactDetail.ts @@ -0,0 +1,79 @@ +import { createHash } from 'crypto' +import type { Request, Response } from 'express' + +import { NotFoundError } from '@crowd/common' +import { getContactDetailsByPurls } from '@crowd/data-access-layer' +import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal' +import { TemporalWorkflowId } from '@crowd/types' + +import { getPackagesQx } from '@/db/packagesDb' +import { getPackagesTemporalClient } from '@/db/packagesTemporal' +import { ok } from '@/utils/api' +import { validateOrThrow } from '@/utils/validation' + +import { toAkritesExternalContactDetail } from './akritesExternalContactDetail' +import { purlBodySchema } from './purl' + +interface IngestSecurityContactsForPurlResult { + found: boolean + repoId?: string +} + +// Deterministic, purl-derived workflowId: concurrent callers hitting the same purl attach +// to the same running workflow (USE_EXISTING) instead of each starting their own ingest — +// same pattern as integrationService.ts's github-nango-sync workflow start. +function ingestWorkflowId(purl: string): string { + return `${TemporalWorkflowId.SECURITY_CONTACTS_ONDEMAND}:${createHash('sha256').update(purl).digest('hex')}` +} + +// Sync, single-purl on-demand ingest. Blocks on the worker's single-repo activity +// (security-contacts/workflows.ts's singleActs: 45s startToCloseTimeout x 2 attempts, +// worst case ~95s, plus unbounded time waiting for a free worker slot) — no batch +// variant, since fanning this out over many purls would multiply concurrent Temporal +// workflow starts. +// +// The underlying workflow itself has no "already processed" gate — it reprocesses and +// replaces a repo's contacts on every invocation — so this handler is the one place +// that must avoid re-triggering it for a purl that's already been ingested. Gate on +// repos.contacts_last_refreshed: null means never processed (trigger the workflow), +// non-null means a pass already ran (return what's there, even if it found no contacts). +export async function ingestAkritesExternalContactDetail( + req: Request, + res: Response, +): Promise { + const { purl } = validateOrThrow(purlBodySchema, req.body) + + const qx = await getPackagesQx() + const [existing] = await getContactDetailsByPurls(qx, [purl]) + if (existing?.contactsLastRefreshed) { + ok(res, toAkritesExternalContactDetail(existing)) + return + } + + if (!existing?.resolvedRepositoryUrl) { + throw new NotFoundError('Purl has no linked repository to ingest security contacts from') + } + + const packagesTemporal = await getPackagesTemporalClient() + const result = await packagesTemporal.workflow.execute< + (purl: string) => Promise + >('ingestSecurityContactsForPurlWorkflow', { + taskQueue: 'security-contacts-worker', + workflowId: ingestWorkflowId(purl), + workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING, + workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, + args: [purl], + }) + + if (!result.found) { + throw new NotFoundError('Purl has no linked repository to ingest security contacts from') + } + + const [row] = await getContactDetailsByPurls(qx, [purl]) + + if (!row) { + throw new NotFoundError('Contact detail not found after ingest') + } + + ok(res, toAkritesExternalContactDetail(row)) +} diff --git a/backend/src/api/public/v1/packages/purl.ts b/backend/src/api/public/v1/packages/purl.ts index 3606428b6a..b737d411c7 100644 --- a/backend/src/api/public/v1/packages/purl.ts +++ b/backend/src/api/public/v1/packages/purl.ts @@ -39,6 +39,10 @@ export const purlFieldSchema = z export const purlQuerySchema = z.object({ purl: purlFieldSchema }) +// Single-purl body (as opposed to purlsBodySchema's array) — normalizes like +// purlFieldSchema/purlQuerySchema, for endpoints that take exactly one purl in the body. +export const purlBodySchema = z.object({ purl: purlFieldSchema }) + // Loose schema for search filters: normalizes without requiring the pkg: prefix, // so partial inputs (e.g. "@babel/core" or "lodash") are accepted. export const purlFilterSchema = z.string().trim().transform(normalizePurl).optional() diff --git a/services/libs/data-access-layer/src/osspckgs/api.ts b/services/libs/data-access-layer/src/osspckgs/api.ts index 20261c432a..26368b87ce 100644 --- a/services/libs/data-access-layer/src/osspckgs/api.ts +++ b/services/libs/data-access-layer/src/osspckgs/api.ts @@ -878,7 +878,11 @@ export interface AkritesExternalContactDetailRow extends Pick, Pick< RepoDbRow, - 'securityPolicyUrl' | 'vulnerabilityReportingUrl' | 'bugBountyUrl' | 'pvrEnabled' + | 'securityPolicyUrl' + | 'vulnerabilityReportingUrl' + | 'bugBountyUrl' + | 'pvrEnabled' + | 'contactsLastRefreshed' > { securityContacts: SecurityContactRow[] | null // --- joined from repos via package_repos, same BEST_REPO_LINK_JOIN as @@ -905,6 +909,7 @@ export async function getContactDetailsByPurls( r.vulnerability_reporting_url AS "vulnerabilityReportingUrl", r.bug_bounty_url AS "bugBountyUrl", r.pvr_enabled AS "pvrEnabled", + r.contacts_last_refreshed AS "contactsLastRefreshed", ${SECURITY_CONTACTS_SUBQUERY} AS "securityContacts" FROM packages p ${BEST_REPO_LINK_JOIN} diff --git a/services/libs/types/src/enums/temporal.ts b/services/libs/types/src/enums/temporal.ts index dee45ec571..2d37fd9298 100644 --- a/services/libs/types/src/enums/temporal.ts +++ b/services/libs/types/src/enums/temporal.ts @@ -12,4 +12,5 @@ export enum TemporalWorkflowId { DELETE_ORPHAN_MEMBER = 'delete-orphan-member', BLAST_RADIUS_ANALYSIS = 'blast-radius-analysis', + SECURITY_CONTACTS_ONDEMAND = 'security-contacts-ondemand', }