Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 54 additions & 19 deletions backend/src/api/public/v1/akrites-external/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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),
Comment thread
ulemons marked this conversation as resolved.
)
router.use('/contacts', contactsSubRouter)

// TODO: the contract gates blast-radius behind a dedicated read:advisories scope
Expand Down
91 changes: 91 additions & 0 deletions backend/src/api/public/v1/akrites-external/openapi.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Comment thread
ulemons marked this conversation as resolved.
securityContacts: [
{
channel: 'email',
Expand Down
Original file line number Diff line number Diff line change
@@ -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> = {},
): 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' }),
Comment thread
ulemons marked this conversation as resolved.
])

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()
})
})
Loading
Loading