Skip to content

Commit 7009346

Browse files
committed
feat: add refresh security contats api
Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>
1 parent 04214f7 commit 7009346

6 files changed

Lines changed: 295 additions & 18 deletions

File tree

backend/src/api/public/v1/akrites-external/index.ts

Lines changed: 39 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -12,26 +12,37 @@ import { getAkritesExternalContactDetailBatch } from '../packages/getAkritesExte
1212
import { getAkritesExternalPackageDetail } from '../packages/getAkritesExternalPackageDetail'
1313
import { getAkritesExternalPackageDetailBatch } from '../packages/getAkritesExternalPackageDetailBatch'
1414
import { getBlastRadiusJob } from '../packages/getBlastRadiusJob'
15+
import { refreshAkritesExternalContactDetail } from '../packages/refreshAkritesExternalContactDetail'
1516
import { submitBlastRadiusJob } from '../packages/submitBlastRadiusJob'
1617

1718
const rateLimiter = createRateLimiter({ max: 60, windowMs: 60 * 1000 })
1819

19-
// Blast-radius jobs kick off a Temporal workflow per request, so they get their own,
20-
// much stricter limiter — configurable via env so it can be tuned without a redeploy.
21-
// Defaults to 5 requests/hour.
22-
const blastRadiusRateLimitMax = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_MAX)
23-
const blastRadiusRateLimitWindowMs = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_WINDOW_MS)
20+
// Shared by every endpoint below that kicks off a Temporal workflow per request — those
21+
// get their own, much stricter limiter than plain reads, configurable via env so it can
22+
// be tuned without a redeploy.
23+
function envTunableRateLimiter(envPrefix: string, defaultMax: number, defaultWindowMs: number) {
24+
const max = Number(process.env[`${envPrefix}_MAX`])
25+
const windowMs = Number(process.env[`${envPrefix}_WINDOW_MS`])
26+
return createRateLimiter({
27+
max: Number.isSafeInteger(max) && max > 0 ? max : defaultMax,
28+
windowMs: Number.isSafeInteger(windowMs) && windowMs > 0 ? windowMs : defaultWindowMs,
29+
})
30+
}
31+
32+
// Blast-radius jobs default to 5 requests/hour.
33+
const blastRadiusRateLimiter = envTunableRateLimiter(
34+
'AKRITES_BLAST_RADIUS_RATE_LIMIT',
35+
5,
36+
60 * 60 * 1000,
37+
)
2438

25-
const blastRadiusRateLimiter = createRateLimiter({
26-
max:
27-
Number.isSafeInteger(blastRadiusRateLimitMax) && blastRadiusRateLimitMax > 0
28-
? blastRadiusRateLimitMax
29-
: 5,
30-
windowMs:
31-
Number.isSafeInteger(blastRadiusRateLimitWindowMs) && blastRadiusRateLimitWindowMs > 0
32-
? blastRadiusRateLimitWindowMs
33-
: 60 * 60 * 1000,
34-
})
39+
// /contacts/refresh blocks ~10-20s per request (vs. the read-only /contacts/detail
40+
// endpoints), so it gets its own limiter. Defaults to 20 requests/hour.
41+
const contactRefreshRateLimiter = envTunableRateLimiter(
42+
'AKRITES_CONTACT_REFRESH_RATE_LIMIT',
43+
20,
44+
60 * 60 * 1000,
45+
)
3546

3647
export function akritesExternalRouter(): Router {
3748
const router = Router()
@@ -63,10 +74,20 @@ export function akritesExternalRouter(): Router {
6374
// closest issued one — READ_MAINTAINER_ROLES (maintainer data) — NOT READ_PACKAGES.
6475
// TODO: swap for cdp:maintainers:read once issued.
6576
const contactsSubRouter = Router()
66-
contactsSubRouter.use(rateLimiter)
6777
contactsSubRouter.use(requireScopes([SCOPES.READ_MAINTAINER_ROLES]))
68-
contactsSubRouter.get('/detail', safeWrap(getAkritesExternalContactDetail))
69-
contactsSubRouter.post(/^\/detail:batch\/?$/, safeWrap(getAkritesExternalContactDetailBatch))
78+
contactsSubRouter.get('/detail', rateLimiter, safeWrap(getAkritesExternalContactDetail))
79+
contactsSubRouter.post(
80+
/^\/detail:batch\/?$/,
81+
rateLimiter,
82+
safeWrap(getAkritesExternalContactDetailBatch),
83+
)
84+
// Sync, single-purl on-demand ingest — starts a Temporal workflow and blocks ~10-20s,
85+
// so it gets the dedicated contactRefreshRateLimiter, not the shared rateLimiter above.
86+
contactsSubRouter.post(
87+
'/refresh',
88+
contactRefreshRateLimiter,
89+
safeWrap(refreshAkritesExternalContactDetail),
90+
)
7091
router.use('/contacts', contactsSubRouter)
7192

7293
// TODO: the contract gates blast-radius behind a dedicated read:advisories scope

backend/src/api/public/v1/akrites-external/openapi.yaml

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1246,3 +1246,77 @@ paths:
12461246
application/json:
12471247
schema:
12481248
$ref: '#/components/schemas/Error'
1249+
1250+
/akrites-external/contacts/refresh:
1251+
post:
1252+
operationId: refreshContactDetail
1253+
summary: On-demand security-contacts ingest for a single PURL
1254+
description: >
1255+
Synchronous, single-purl only — no batch variant, since fanning this out over
1256+
many purls would multiply concurrent Temporal workflow starts. Blocks ~10-20s
1257+
while a single-repo Temporal workflow (ingestSecurityContactsForPurlWorkflow)
1258+
ingests the linked repo's security contacts, then returns the same
1259+
ContactDetail payload as GET /contacts/detail. Concurrent requests for the
1260+
same purl attach to the same running workflow instead of starting duplicates.
1261+
1262+
1263+
Rate limited independently of the other /contacts endpoints — default
1264+
20 requests/hour, tunable via AKRITES_CONTACT_REFRESH_RATE_LIMIT_MAX /
1265+
AKRITES_CONTACT_REFRESH_RATE_LIMIT_WINDOW_MS.
1266+
1267+
1268+
Not yet implemented: this is a synchronous, blocking call. Future work should
1269+
make this async — an accept-and-poll job, similar to POST /blast-radius/jobs —
1270+
so callers aren't held open for 10-20s per request.
1271+
tags: [Contacts]
1272+
security:
1273+
- M2MBearer:
1274+
- read:maintainer-roles
1275+
requestBody:
1276+
required: true
1277+
content:
1278+
application/json:
1279+
schema:
1280+
type: object
1281+
required: [purl]
1282+
properties:
1283+
purl:
1284+
type: string
1285+
example: pkg:npm/%40angular/core
1286+
responses:
1287+
'200':
1288+
description: Security contact detail, freshly ingested.
1289+
content:
1290+
application/json:
1291+
schema:
1292+
$ref: '#/components/schemas/ContactDetail'
1293+
'400':
1294+
description: Malformed purl.
1295+
content:
1296+
application/json:
1297+
schema:
1298+
$ref: '#/components/schemas/Error'
1299+
'401':
1300+
description: Missing or invalid bearer token.
1301+
content:
1302+
application/json:
1303+
schema:
1304+
$ref: '#/components/schemas/Error'
1305+
'403':
1306+
description: Token missing read:maintainer-roles scope.
1307+
content:
1308+
application/json:
1309+
schema:
1310+
$ref: '#/components/schemas/Error'
1311+
'404':
1312+
description: Purl has no linked repo to ingest contacts from.
1313+
content:
1314+
application/json:
1315+
schema:
1316+
$ref: '#/components/schemas/Error'
1317+
'429':
1318+
description: Too many requests — rate-limited independently of the other endpoints.
1319+
content:
1320+
application/json:
1321+
schema:
1322+
$ref: '#/components/schemas/Error'

backend/src/api/public/v1/packages/purl.ts

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,10 @@ export const purlFieldSchema = z
3939

4040
export const purlQuerySchema = z.object({ purl: purlFieldSchema })
4141

42+
// Single-purl body (as opposed to purlsBodySchema's array) — normalizes like
43+
// purlFieldSchema/purlQuerySchema, for endpoints that take exactly one purl in the body.
44+
export const purlBodySchema = z.object({ purl: purlFieldSchema })
45+
4246
// Loose schema for search filters: normalizes without requiring the pkg: prefix,
4347
// so partial inputs (e.g. "@babel/core" or "lodash") are accepted.
4448
export const purlFilterSchema = z.string().trim().transform(normalizePurl).optional()
Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
import type { Request, Response } from 'express'
2+
import { describe, expect, it, vi } from 'vitest'
3+
4+
import type { AkritesExternalContactDetailRow } from '@crowd/data-access-layer'
5+
6+
import { refreshAkritesExternalContactDetail } from './refreshAkritesExternalContactDetail'
7+
8+
const { execute, getContactDetailsByPurls } = vi.hoisted(() => ({
9+
execute: vi.fn(),
10+
getContactDetailsByPurls: vi.fn(),
11+
}))
12+
13+
vi.mock('@/db/packagesTemporal', () => ({
14+
getPackagesTemporalClient: vi.fn().mockResolvedValue({ workflow: { execute } }),
15+
}))
16+
17+
vi.mock('@/db/packagesDb', () => ({
18+
getPackagesQx: vi.fn().mockResolvedValue({}),
19+
}))
20+
21+
vi.mock('@crowd/data-access-layer', () => ({
22+
getContactDetailsByPurls,
23+
}))
24+
25+
function baseRow(
26+
overrides: Partial<AkritesExternalContactDetailRow> = {},
27+
): AkritesExternalContactDetailRow {
28+
return {
29+
purl: 'pkg:npm/lodash',
30+
name: 'lodash',
31+
ecosystem: 'npm',
32+
securityPolicyUrl: null,
33+
vulnerabilityReportingUrl: null,
34+
bugBountyUrl: null,
35+
pvrEnabled: null,
36+
declaredRepositoryUrl: null,
37+
resolvedRepositoryUrl: null,
38+
repoMappingConfidence: null,
39+
securityContacts: [],
40+
...overrides,
41+
}
42+
}
43+
44+
function mockReqRes(body: unknown) {
45+
execute.mockClear()
46+
getContactDetailsByPurls.mockClear()
47+
48+
const req = { body } as unknown as Request
49+
50+
const json = vi.fn()
51+
const status = vi.fn().mockReturnValue({ json })
52+
const res = { status, json } as unknown as Response
53+
54+
return { req, res, status, json }
55+
}
56+
57+
describe('refreshAkritesExternalContactDetail', () => {
58+
it('executes ingestSecurityContactsForPurlWorkflow and returns the re-read contact detail', async () => {
59+
execute.mockResolvedValue({ found: true, repoId: 'repo-1' })
60+
getContactDetailsByPurls.mockResolvedValue([baseRow()])
61+
62+
const { req, res, json } = mockReqRes({ purl: 'pkg:npm/lodash' })
63+
64+
await refreshAkritesExternalContactDetail(req, res)
65+
66+
expect(execute).toHaveBeenCalledTimes(1)
67+
const [workflowType, options] = execute.mock.calls[0]
68+
expect(workflowType).toBe('ingestSecurityContactsForPurlWorkflow')
69+
expect(options.taskQueue).toBe('security-contacts-worker')
70+
expect(options.workflowId).toMatch(/^security-contacts-ondemand:[0-9a-f]{64}$/)
71+
expect(options.args).toEqual(['pkg:npm/lodash'])
72+
73+
expect(getContactDetailsByPurls).toHaveBeenCalledWith(expect.anything(), ['pkg:npm/lodash'])
74+
expect(json).toHaveBeenCalledWith(expect.objectContaining({ purl: 'pkg:npm/lodash' }))
75+
})
76+
77+
it('derives the same deterministic workflowId for the same purl', async () => {
78+
execute.mockResolvedValue({ found: true })
79+
getContactDetailsByPurls.mockResolvedValue([baseRow()])
80+
81+
const { req: req1, res: res1 } = mockReqRes({ purl: 'pkg:npm/lodash' })
82+
await refreshAkritesExternalContactDetail(req1, res1)
83+
const id1 = execute.mock.calls[0][1].workflowId
84+
85+
const { req: req2, res: res2 } = mockReqRes({ purl: 'pkg:npm/lodash' })
86+
await refreshAkritesExternalContactDetail(req2, res2)
87+
const id2 = execute.mock.calls[0][1].workflowId
88+
89+
expect(id1).toBe(id2)
90+
})
91+
92+
it('throws NotFoundError when the workflow reports no linked repo', async () => {
93+
execute.mockResolvedValue({ found: false })
94+
95+
const { req, res } = mockReqRes({ purl: 'pkg:npm/left-pad' })
96+
97+
await expect(refreshAkritesExternalContactDetail(req, res)).rejects.toThrow()
98+
expect(getContactDetailsByPurls).not.toHaveBeenCalled()
99+
})
100+
101+
it('throws NotFoundError when the re-read finds no row', async () => {
102+
execute.mockResolvedValue({ found: true })
103+
getContactDetailsByPurls.mockResolvedValue([])
104+
105+
const { req, res } = mockReqRes({ purl: 'pkg:npm/lodash' })
106+
107+
await expect(refreshAkritesExternalContactDetail(req, res)).rejects.toThrow()
108+
})
109+
110+
it('rejects a request missing purl without executing a workflow', async () => {
111+
const { req, res } = mockReqRes({})
112+
113+
await expect(refreshAkritesExternalContactDetail(req, res)).rejects.toThrow()
114+
expect(execute).not.toHaveBeenCalled()
115+
})
116+
})
Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
import { createHash } from 'crypto'
2+
import type { Request, Response } from 'express'
3+
4+
import { NotFoundError } from '@crowd/common'
5+
import { getContactDetailsByPurls } from '@crowd/data-access-layer'
6+
import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal'
7+
import { TemporalWorkflowId } from '@crowd/types'
8+
9+
import { getPackagesQx } from '@/db/packagesDb'
10+
import { getPackagesTemporalClient } from '@/db/packagesTemporal'
11+
import { ok } from '@/utils/api'
12+
import { validateOrThrow } from '@/utils/validation'
13+
14+
import { toAkritesExternalContactDetail } from './akritesExternalContactDetail'
15+
import { purlBodySchema } from './purl'
16+
17+
interface IngestSecurityContactsForPurlResult {
18+
found: boolean
19+
repoId?: string
20+
}
21+
22+
// Deterministic, purl-derived workflowId: concurrent callers hitting the same purl attach
23+
// to the same running workflow (USE_EXISTING) instead of each starting their own ingest —
24+
// same pattern as integrationService.ts's github-nango-sync workflow start.
25+
function refreshWorkflowId(purl: string): string {
26+
return `${TemporalWorkflowId.SECURITY_CONTACTS_ONDEMAND}:${createHash('sha256').update(purl).digest('hex')}`
27+
}
28+
29+
// Sync, single-purl on-demand ingest. Blocks ~10-20s (the worker's single-repo bound —
30+
// see security-contacts/workflows.ts's singleActs timeout) — no batch variant, since
31+
// fanning this out over many purls would multiply concurrent Temporal workflow starts.
32+
export async function refreshAkritesExternalContactDetail(
33+
req: Request,
34+
res: Response,
35+
): Promise<void> {
36+
const { purl } = validateOrThrow(purlBodySchema, req.body)
37+
38+
const packagesTemporal = await getPackagesTemporalClient()
39+
const result = await packagesTemporal.workflow.execute<
40+
(purl: string) => Promise<IngestSecurityContactsForPurlResult>
41+
>('ingestSecurityContactsForPurlWorkflow', {
42+
taskQueue: 'security-contacts-worker',
43+
workflowId: refreshWorkflowId(purl),
44+
workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
45+
workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
46+
args: [purl],
47+
})
48+
49+
if (!result.found) {
50+
throw new NotFoundError()
51+
}
52+
53+
const qx = await getPackagesQx()
54+
const [row] = await getContactDetailsByPurls(qx, [purl])
55+
56+
if (!row) {
57+
throw new NotFoundError()
58+
}
59+
60+
ok(res, toAkritesExternalContactDetail(row))
61+
}

services/libs/types/src/enums/temporal.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,4 +12,5 @@ export enum TemporalWorkflowId {
1212
DELETE_ORPHAN_MEMBER = 'delete-orphan-member',
1313

1414
BLAST_RADIUS_ANALYSIS = 'blast-radius-analysis',
15+
SECURITY_CONTACTS_ONDEMAND = 'security-contacts-ondemand',
1516
}

0 commit comments

Comments
 (0)