diff --git a/backend/src/osspckgs/migrations/V1785283200__reporting_protocol.sql b/backend/src/osspckgs/migrations/V1785283200__reporting_protocol.sql new file mode 100644 index 0000000000..a8e4b7909c --- /dev/null +++ b/backend/src/osspckgs/migrations/V1785283200__reporting_protocol.sql @@ -0,0 +1,26 @@ +-- Content-keyed parse cache for declared security policies (files + linked pages), +-- and the assembled per-repo reporting protocol. See ADR-0010 addendum. +CREATE TABLE IF NOT EXISTS security_policy_parses ( + blob_oid TEXT PRIMARY KEY, -- git blob oid for files; sha256 of extracted text for linked pages + source_kind TEXT NOT NULL, -- 'security-file' | 'linked-page' + url TEXT, -- linked-page rows only: the fetched URL (join key from linked_urls) + parser TEXT NOT NULL, -- 'deterministic' | 'llm' + parser_version INT NOT NULL, + status TEXT NOT NULL, -- 'ok' | 'template' | 'degraded' + parsed JSONB NOT NULL DEFAULT '{}', + linked_urls TEXT[] NOT NULL DEFAULT '{}', + parsed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS security_policy_parses_linked_page_url_idx + ON security_policy_parses (url) + WHERE source_kind = 'linked-page'; + +CREATE TABLE IF NOT EXISTS repo_reporting_protocols ( + repo_id BIGINT PRIMARY KEY REFERENCES repos(id) ON DELETE CASCADE, + declared BOOLEAN NOT NULL, + methods JSONB NOT NULL DEFAULT '[]', + guidelines JSONB, + sources JSONB NOT NULL DEFAULT '[]', + assembled_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/backend/src/osspckgs/migrations/V1785369600__reporting_protocol_llm_tokens.sql b/backend/src/osspckgs/migrations/V1785369600__reporting_protocol_llm_tokens.sql new file mode 100644 index 0000000000..0685949587 --- /dev/null +++ b/backend/src/osspckgs/migrations/V1785369600__reporting_protocol_llm_tokens.sql @@ -0,0 +1,5 @@ +-- Persist Bedrock LLM cost (USD) per parse so spend is measurable per file. +-- Null for deterministic rows (no LLM call); set for every parser='llm' row, +-- including degraded ones (a failed extraction still bills tokens). +ALTER TABLE security_policy_parses + ADD COLUMN IF NOT EXISTS llm_cost_usd NUMERIC(12, 6); diff --git a/docs/adr/0010-security-contacts-worker.md b/docs/adr/0010-security-contacts-worker.md index 351c225db6..affb7c030c 100644 --- a/docs/adr/0010-security-contacts-worker.md +++ b/docs/adr/0010-security-contacts-worker.md @@ -281,3 +281,108 @@ derived at read time as the band of the highest contact score, not stored. - **Source-format drift** — SECURITY-INSIGHTS schema versions, registry API shapes, and GitHub's `stats/contributors` 202-polling behavior all change over time. Extractor isolation limits blast radius to one source; fixture-based tests catch parser regressions. + + +## Addendum (2026-07-29): Vulnerability reporting protocol + +Adds a sister data model answering "**how** does this project expect external vulnerability +reporting?" per repo — distinct from security contacts, which answer *who*. The source of truth +is what the project itself declared: security files from the enricher's `repo_well_known_files` +inventory, the pages they link to, and the authoritative `pvr_enabled` flag. Inferred contacts +from `security_contacts` never blend in as if declared; they appear only as clearly-labeled +fallback when nothing was declared. + +### Volume and parser split (prod analysis, 2026-07-28) + +Of 114,045 critical GitHub repos, 10,349 (9.1%) have a security file: 10,495 files collapsing +to 6,125 distinct blobs (top-20 shared blobs cover ~1,900 repos of boilerplate). A probe over +all 6,120 reachable blobs showed **69.2% deterministically resolvable** (a single declared +method, or several with exactly one preference-cued), **14% pointer-only** (the file is just a +link to an external policy page), **21.6% with conditional routing** ("only email if a GHSA is +not possible"), **53% with negation language** ("do NOT open a public issue"), 2.3% GitHub +default template. Volume is not the constraint; precision is — hence **hybrid, +deterministic-first**: the classifier fully settles clean blobs, an LLM handles the residue and +prose fields, and a deterministic validator gates every LLM write. + +### Data model + +- **`security_policy_parses`** — content-keyed parse cache. PK `blob_oid` (git blob oid for + files; sha256 of the URL for linked pages, so two URLs with identical content stay + independently joinable from `linked_urls`), `source_kind` + (`security-file`/`linked-page`), `url` (linked-page rows), `parser` + (`deterministic`/`llm`), `parser_version`, `status` (`ok`/`template`/`degraded`), `parsed` + JSONB (methods + guidelines), `linked_urls`. Identical content across repos is parsed once, + ever; a `parser_version` bump is a targeted re-parse, not a migration. +- **`repo_reporting_protocols`** — assembled per-repo answer. PK `repo_id`, `declared`, + `methods` JSONB (ordered array of `{type, status, endpoint, condition, confidence, + provenance}`), `guidelines` JSONB, `sources` JSONB, `assembled_at`. Method `type` ∈ + github-pvr | email | web-form | bounty-platform | security-txt | mailing-list; `status` ∈ + preferred | accepted | fallback | prohibited (`prohibited` captures negation language); + `confidence` ∈ declared | inferred. Plain upsert — fully derived and recomputable, no + soft delete. + +### Parse stage (blob-driven) + +`repo_well_known_files` is the work queue (live `security` rows for critical GitHub repos whose +`blob_oid` lacks a parse at the current version); this pipeline never probes repos for files. +Blobs are fetched once by oid through the shared GitHub gateway. The classifier (same +windowing family as the B1 extractor) emits a `clean` verdict — single usable method, or +exactly one preference-cued among several, no conditional language, negation on a method's own +line marks it `prohibited` — which is stored as-is. Residue goes to the LLM; the validator +requires every emitted endpoint to appear in the source (URLs verbatim; emails also via +deobfuscation normalization — "security at python dot org"), valid enums, and at most one +`preferred` — failures are stored `status='degraded'` (classifier partials, no guidelines). +The LLM can never invent a channel. Pointer-only parses record up to 3 linked URLs; each +linked page is fetched once per URL (SSRF-guarded: http(s) only, private/loopback/link-local +and metadata hosts blocked, redirects revalidated per hop, body capped at 500 KB while +streaming) and parsed as a `linked-page` row. For a pointer-only blob the file row is written +only after every linked page has a parse row, so a transient page failure leaves the blob +unmarked and the next daily sweep retries the whole unit. Batches are drawn in random order so +permanently failing blobs cannot starve the queue. + +### Assembly + +Repos are re-assembled when inputs change (no protocol row, `contacts_last_refreshed` newer +than `assembled_at`, or a newer parse for one of their blobs). Merge rules: `ok`/`template` +parses contribute methods and guidelines with provenance — **`degraded` parses contribute +nothing**; `pvr_enabled = true` adds a `github-pvr` method when the files are silent, and +`pvr_enabled = false` **vetoes** a declared github-pvr method (the A2-vetoes-B1 rule applied +to the protocol); github-pvr sentinel endpoints are rewritten per repo to +`…/security/advisories/new`; dedup on type+endpoint; at most one `preferred`; sort preferred > +accepted > fallback > prohibited. Only when nothing is declared: up to 3 `inferred`/`fallback` +methods derived from live `security_contacts` (email, github-pvr, web-form channels, by score). +Every repo in the population gets a row — `declared=false` with an empty `methods` array for +the ~89 no-signal repos. + +### LLM contract + +Direct AWS Bedrock calls (`@aws-sdk/client-bedrock-runtime`, module-local in `llmExtract.ts`) +— deliberately **not** the legacy class-based `LlmService` in `common_services` (class pattern ++ prompt-history DB coupling) and **not** a shared provider-agnostic lib speaking to a LiteLLM +proxy (built during implementation, then dropped: no LiteLLM infra today; revisit if CDP +standardizes multi-provider LLM infrastructure — schema and prompt carry over unchanged). +Existing `CROWD_AWS_BEDROCK_ACCESS_KEY_ID`/`CROWD_AWS_BEDROCK_SECRET_ACCESS_KEY` credentials; +default `LlmModelType.CLAUDE_HAIKU_4_5` with region from `LLM_MODEL_REGION_MAP`. The JSON +schema is embedded in the system prompt (Bedrock InvokeModel has no structured-output mode); +`parseLlmJson` parses the answer. Missing credentials or any failure → `degraded` parse, never +a thrown error. No prompt-history persistence. + +### Scheduling + +Own Temporal schedule `reporting-protocol-ingestion` (daily 07:00, `SKIP` overlap, 24 h +execution timeout) inside the security-contacts worker, independent of the contacts schedule so +a slow LLM pass never stalls contact ingestion. The workflow drains parsing first +(`continueAsNew` while a batch parsed anything; an all-failed batch falls through to assembly +instead of recursing — failed blobs get no row and retry on the next daily tick), then drains +assembly. Batch sizes: 200 blobs (parse), 2,000 repos (assemble). Both activities ride the +shared 30-minute proxy and heartbeat on a fixed 30 s cadence under its 2-minute +`heartbeatTimeout`. + +### Deferred + +Other interaction-profile domains (contribution intake, governance, maintainer roster, +communication channels, code of conduct — the content-keyed cache and section pattern extend +to them); org-level `.github` default files (GitHub serves them for repos without their own +SECURITY.md; the inventory doesn't capture them — measure the gap first); non-GitHub declared +parsing (no file inventory; such repos assemble as `declared=false` + inferred fallback); API +exposure on the akrites endpoints. diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f8e9565781..7247b9fafe 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1300,6 +1300,9 @@ importers: '@anthropic-ai/claude-agent-sdk': specifier: ^0.3.216 version: 0.3.217(@anthropic-ai/sdk@0.112.5(zod@4.3.6))(@modelcontextprotocol/sdk@1.29.0(zod@4.3.6))(zod@4.3.6) + '@aws-sdk/client-bedrock-runtime': + specifier: ^3.572.0 + version: 3.572.0 '@crowd/archetype-standard': specifier: workspace:* version: link:../../archetypes/standard @@ -10940,8 +10943,8 @@ snapshots: dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sso-oidc': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11135,11 +11138,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sso-oidc@3.572.0(@aws-sdk/client-sts@3.572.0)': + '@aws-sdk/client-sso-oidc@3.572.0': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11178,7 +11181,6 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: - - '@aws-sdk/client-sts' - aws-crt '@aws-sdk/client-sso@3.556.0': @@ -11354,11 +11356,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sts@3.572.0': + '@aws-sdk/client-sts@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11397,6 +11399,7 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: + - '@aws-sdk/client-sso-oidc' - aws-crt '@aws-sdk/client-sts@3.985.0': @@ -11562,7 +11565,7 @@ snapshots: '@aws-sdk/credential-provider-ini@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/credential-provider-env': 3.568.0 '@aws-sdk/credential-provider-process': 3.572.0 '@aws-sdk/credential-provider-sso': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) @@ -11739,7 +11742,7 @@ snapshots: '@aws-sdk/credential-provider-web-identity@3.568.0(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/types': 2.12.0 @@ -12051,7 +12054,7 @@ snapshots: '@aws-sdk/token-providers@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/shared-ini-file-loader': 2.4.0 @@ -16873,19 +16876,19 @@ snapshots: etag: 1.8.1 finalhandler: 2.1.1 fresh: 2.0.0 - http-errors: 2.0.0 + http-errors: 2.0.1 merge-descriptors: 2.0.0 mime-types: 3.0.2 on-finished: 2.4.1 once: 1.4.0 parseurl: 1.3.3 proxy-addr: 2.0.7 - qs: 6.14.0 + qs: 6.15.3 range-parser: 1.2.1 router: 2.2.0 send: 1.2.1 serve-static: 2.2.1 - statuses: 2.0.1 + statuses: 2.0.2 type-is: 2.1.0 vary: 1.1.2 transitivePeerDependencies: @@ -17024,7 +17027,7 @@ snapshots: escape-html: 1.0.3 on-finished: 2.4.1 parseurl: 1.3.3 - statuses: 2.0.1 + statuses: 2.0.2 transitivePeerDependencies: - supports-color diff --git a/services/apps/packages_worker/package.json b/services/apps/packages_worker/package.json index dac0d8dca7..abad411c0a 100644 --- a/services/apps/packages_worker/package.json +++ b/services/apps/packages_worker/package.json @@ -91,6 +91,7 @@ "@crowd/slack": "workspace:*", "@crowd/types": "workspace:*", "@anthropic-ai/claude-agent-sdk": "^0.3.216", + "@aws-sdk/client-bedrock-runtime": "^3.572.0", "@dsnp/parquetjs": "^1.7.0", "@google-cloud/bigquery": "^8.3.1", "@google-cloud/storage": "7.19.0", diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index 4ea68d0caf..d0bf0a4d24 100644 --- a/services/apps/packages_worker/src/activities.ts +++ b/services/apps/packages_worker/src/activities.ts @@ -52,6 +52,8 @@ export { processRubyGemsCoreBatch, processRubyGemsCriticalBatch } from './rubyge export { processSecurityContactsBatch, ingestSecurityContactsForPurlActivity, + runProtocolParseBatch, + runProtocolAssembleBatch, } from './security-contacts/activities' export { blastRadiusStart, diff --git a/services/apps/packages_worker/src/bin/security-contacts-worker.ts b/services/apps/packages_worker/src/bin/security-contacts-worker.ts index 387b0a7c1d..a86f3bae19 100644 --- a/services/apps/packages_worker/src/bin/security-contacts-worker.ts +++ b/services/apps/packages_worker/src/bin/security-contacts-worker.ts @@ -1,8 +1,10 @@ +import { scheduleReportingProtocolIngestion } from '../security-contacts/protocol/schedule' import { scheduleSecurityContactsIngestion } from '../security-contacts/schedule' import { svc } from '../service' setImmediate(async () => { await svc.init() await scheduleSecurityContactsIngestion() + await scheduleReportingProtocolIngestion() await svc.start() }) diff --git a/services/apps/packages_worker/src/config.ts b/services/apps/packages_worker/src/config.ts index 2306a0d7d4..94854d105d 100644 --- a/services/apps/packages_worker/src/config.ts +++ b/services/apps/packages_worker/src/config.ts @@ -1,3 +1,5 @@ +import { LlmModelType } from '@crowd/types' + function requireEnv(name: string): string { const val = process.env[name] if (!val) throw new Error(`Missing required environment variable: ${name}`) @@ -119,3 +121,17 @@ export function getDockerhubConfig() { idleSleepSec: requireEnvInt('DOCKERHUB_IDLE_SLEEP_SEC'), } } + +export function getReportingProtocolConfig() { + return { + parseBatchSize: parseInt(process.env.REPORTING_PROTOCOL_PARSE_BATCH_SIZE ?? '200', 10), + assembleBatchSize: parseInt(process.env.REPORTING_PROTOCOL_ASSEMBLE_BATCH_SIZE ?? '2000', 10), + concurrency: parseInt(process.env.REPORTING_PROTOCOL_CONCURRENCY ?? '10', 10), + fetchTimeoutMs: parseInt(process.env.REPORTING_PROTOCOL_FETCH_TIMEOUT_MS ?? '15000', 10), + llmModelId: process.env.REPORTING_PROTOCOL_LLM_MODEL_ID ?? LlmModelType.CLAUDE_HAIKU_4_5, + llmTimeoutMs: parseInt(process.env.REPORTING_PROTOCOL_LLM_TIMEOUT_MS ?? '60000', 10), + llmConcurrency: parseInt(process.env.REPORTING_PROTOCOL_LLM_CONCURRENCY ?? '4', 10), + llmAccessKeyId: process.env.CROWD_AWS_BEDROCK_ACCESS_KEY_ID, + llmSecretAccessKey: process.env.CROWD_AWS_BEDROCK_SECRET_ACCESS_KEY, + } +} diff --git a/services/apps/packages_worker/src/security-contacts/activities.ts b/services/apps/packages_worker/src/security-contacts/activities.ts index 95b1a21d0b..dcbf379ede 100644 --- a/services/apps/packages_worker/src/security-contacts/activities.ts +++ b/services/apps/packages_worker/src/security-contacts/activities.ts @@ -1,10 +1,18 @@ +import { heartbeat } from '@temporalio/activity' + import { getServiceChildLogger } from '@crowd/logging' -import { getSecurityContactsConfig } from '../config' +import { getReportingProtocolConfig, getSecurityContactsConfig } from '../config' import { getCdpDb, getPackagesDb } from '../db' +import { githubApiGet } from './githubToken' import { IngestSingleResult, ingestSecurityContactsForPurl } from './ingestSingle' import { BatchResult, processBatch } from './processBatch' +import { runAssembleStage } from './protocol/assembleStage' +import { fetchLinkedPage } from './protocol/fetchContent' +import { llmExtractProtocol } from './protocol/llmExtract' +import { runParseStage } from './protocol/parseStage' +import { AssembleStageResult, ParseStageResult } from './protocol/types' const log = getServiceChildLogger('security-contacts-activity') @@ -29,3 +37,42 @@ export async function ingestSecurityContactsForPurlActivity( log.info({ purl, ...result }, 'On-demand security contacts ingest activity complete') return result } + +// Fixed-cadence heartbeat, same rationale as processBatch.ts: a slow blob (LLM call) can +// outlast the 2-minute heartbeatTimeout on the shared activity proxy. +async function withHeartbeat(fn: () => Promise): Promise { + const heartbeatTimer = setInterval(() => { + try { + heartbeat() + } catch (err) { + log.warn({ errMsg: (err as Error).message }, 'Heartbeat failed') + } + }, 30_000) + try { + return await fn() + } finally { + clearInterval(heartbeatTimer) + } +} + +export async function runProtocolParseBatch(): Promise { + const cfg = getReportingProtocolConfig() + const qx = await getPackagesDb() + const result = await withHeartbeat(() => + runParseStage( + qx, + { githubGet: githubApiGet, fetchPage: fetchLinkedPage, llmExtract: llmExtractProtocol }, + cfg, + ), + ) + log.info({ ...result }, 'Reporting protocol parse batch activity complete') + return result +} + +export async function runProtocolAssembleBatch(): Promise { + const cfg = getReportingProtocolConfig() + const qx = await getPackagesDb() + const result = await withHeartbeat(() => runAssembleStage(qx, cfg)) + log.info({ ...result }, 'Reporting protocol assemble batch activity complete') + return result +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts b/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts new file mode 100644 index 0000000000..0deb9bffff --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts @@ -0,0 +1,168 @@ +import { parseGithubUrl } from '../../enricher/fetchLightRepo' + +import { + AssembledProtocol, + ParseRowStatus, + ParsedProtocol, + ProtocolMethod, + ProtocolMethodType, +} from './types' +import { SENTINEL_ENDPOINT_BY_TYPE } from './validate' + +export interface AssembleInput { + repoUrl: string + pvrEnabled: boolean | null + securityTxtUrl: string | null + fileParses: Array<{ + blobOid: string + path: string + parser: 'deterministic' | 'llm' + status: ParseRowStatus + parsed: ParsedProtocol + }> + pageParses: Array<{ + hash: string + url: string + parser: 'deterministic' | 'llm' + status: ParseRowStatus + parsed: ParsedProtocol + }> + fallbackContacts: Array<{ channel: string; value: string; score: number }> +} + +const STATUS_ORDER: Record = { + preferred: 0, + accepted: 1, + fallback: 2, + prohibited: 3, +} +const FALLBACK_CHANNEL_MAP: Record = { + email: 'email', + 'github-pvr': 'github-pvr', + 'web-form': 'web-form', +} + +function pvrAdvisoryUrl(repoUrl: string): string { + const { owner, name } = parseGithubUrl(repoUrl) + return `https://github.com/${owner}/${name}/security/advisories/new` +} + +export function assembleProtocol(input: AssembleInput): AssembledProtocol { + const methods: ProtocolMethod[] = [] + const seen = new Set() + const sources: Array> = [] + + const push = (m: ProtocolMethod) => { + const key = `${m.type}:${m.endpoint.toLowerCase()}` + if (seen.has(key)) return + seen.add(key) + methods.push(m) + } + + // Rewrite sentinel endpoints to actionable URLs. github-pvr always resolves + // from the repo URL; security-txt resolves from repos.security_txt_url when + // known — a bare "security.txt" mention with no discoverable URL is dropped + // rather than emitted as a non-actionable endpoint. + const resolveEndpoint = (m: { type: ProtocolMethodType; endpoint: string }): string | null => { + if (m.type === 'github-pvr') return pvrAdvisoryUrl(input.repoUrl) + if (m.type === 'security-txt' && m.endpoint === SENTINEL_ENDPOINT_BY_TYPE['security-txt']) { + return input.securityTxtUrl + } + return m.endpoint + } + + for (const fp of input.fileParses) { + if (fp.status === 'degraded') continue + sources.push({ path: fp.path, blobOid: fp.blobOid, parser: fp.parser }) + for (const m of fp.parsed.methods) { + const endpoint = resolveEndpoint(m) + if (endpoint === null) continue + push({ + ...m, + endpoint, + confidence: 'declared', + provenance: { path: fp.path, blobOid: fp.blobOid, parser: fp.parser }, + }) + } + } + for (const pp of input.pageParses) { + if (pp.status === 'degraded') continue + sources.push({ url: pp.url, blobOid: pp.hash, parser: pp.parser }) + for (const m of pp.parsed.methods) { + const endpoint = resolveEndpoint(m) + if (endpoint === null) continue + push({ + ...m, + endpoint, + confidence: 'declared', + provenance: { url: pp.url, blobOid: pp.hash, parser: pp.parser }, + }) + } + } + + let declaredMethods = methods + if (input.pvrEnabled === false) { + declaredMethods = declaredMethods.filter((m) => m.type !== 'github-pvr') + } + if (input.pvrEnabled === true && !declaredMethods.some((m) => m.type === 'github-pvr')) { + const hasPreferred = declaredMethods.some((m) => m.status === 'preferred') + declaredMethods.push({ + type: 'github-pvr', + status: hasPreferred ? 'accepted' : 'preferred', + endpoint: pvrAdvisoryUrl(input.repoUrl), + condition: null, + confidence: 'declared', + provenance: { api: 'pvr-flag' }, + }) + sources.push({ api: 'pvr-flag' }) + } + if (input.securityTxtUrl && !declaredMethods.some((m) => m.type === 'security-txt')) { + declaredMethods.push({ + type: 'security-txt', + status: 'accepted', + endpoint: input.securityTxtUrl, + condition: null, + confidence: 'declared', + provenance: { api: 'security-txt-url' }, + }) + sources.push({ api: 'security-txt-url' }) + } + + const declared = declaredMethods.length > 0 + + let finalMethods = declaredMethods + if (!declared) { + finalMethods = input.fallbackContacts + .filter((c) => FALLBACK_CHANNEL_MAP[c.channel]) + .sort((a, b) => b.score - a.score) + .slice(0, 3) + .map((c) => ({ + type: FALLBACK_CHANNEL_MAP[c.channel], + status: 'fallback' as const, + endpoint: c.value, + condition: null, + confidence: 'inferred' as const, + provenance: { channel: c.channel }, + })) + if (finalMethods.length > 0) sources.push({ table: 'security_contacts' }) + } + + let preferredSeen = false + for (const m of finalMethods) { + if (m.status === 'preferred') { + if (preferredSeen) m.status = 'accepted' + preferredSeen = true + } + } + finalMethods = finalMethods + .map((m, i) => ({ m, i })) + .sort((a, b) => STATUS_ORDER[a.m.status] - STATUS_ORDER[b.m.status] || a.i - b.i) + .map((x) => x.m) + + const guidelines = + [...input.fileParses, ...input.pageParses].find( + (p) => p.status !== 'degraded' && p.parsed.guidelines, + )?.parsed.guidelines ?? null + + return { declared, methods: finalMethods, guidelines, sources } +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/assembleStage.ts b/services/apps/packages_worker/src/security-contacts/protocol/assembleStage.ts new file mode 100644 index 0000000000..68347067ce --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/assembleStage.ts @@ -0,0 +1,109 @@ +import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' +import { prepareBulkInsert } from '@crowd/data-access-layer/src/utils' +import { getServiceChildLogger } from '@crowd/logging' + +import { getReportingProtocolConfig } from '../../config' + +import { AssembleInput, assembleProtocol } from './assemble' +import { AssembleStageResult } from './types' + +const log = getServiceChildLogger('reporting-protocol:assemble') + +type Cfg = ReturnType + +const PROTOCOL_COLUMNS = ['repo_id', 'declared', 'methods', 'guidelines', 'sources'] +const PROTOCOL_UPSERT_SET = `(repo_id) DO UPDATE SET + declared = EXCLUDED.declared, + methods = EXCLUDED.methods, + guidelines = EXCLUDED.guidelines, + sources = EXCLUDED.sources, + assembled_at = NOW()` + +interface RepoRow { + id: string + url: string + pvr_enabled: boolean | null + security_txt_url: string | null + file_parses: AssembleInput['fileParses'] | null + page_parses: AssembleInput['pageParses'] | null + fallback_contacts: AssembleInput['fallbackContacts'] | null +} + +async function selectReposToAssemble(qx: QueryExecutor, limit: number): Promise { + return qx.select( + `WITH stale AS ( + SELECT DISTINCT r.id + 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 + LEFT JOIN repo_reporting_protocols rp ON rp.repo_id = r.id + WHERE rp.repo_id IS NULL + OR r.contacts_last_refreshed > rp.assembled_at + OR EXISTS ( + SELECT 1 FROM repo_well_known_files w + JOIN security_policy_parses sp ON sp.blob_oid = w.blob_oid + WHERE w.repo_id = r.id AND w.file_type = 'security' + AND w.deleted_at IS NULL AND sp.parsed_at > rp.assembled_at) + LIMIT $(limit) + ) + SELECT r.id::text AS id, r.url, r.pvr_enabled, r.security_txt_url, + (SELECT json_agg(json_build_object( + 'blobOid', sp.blob_oid, 'path', w.path, 'parser', sp.parser, + 'status', sp.status, 'parsed', sp.parsed)) + FROM repo_well_known_files w + JOIN security_policy_parses sp ON sp.blob_oid = w.blob_oid + WHERE w.repo_id = r.id AND w.file_type = 'security' AND w.deleted_at IS NULL + ) AS file_parses, + (SELECT json_agg(json_build_object( + 'hash', lp.blob_oid, 'url', lp.url, 'parser', lp.parser, + 'status', lp.status, 'parsed', lp.parsed)) + FROM repo_well_known_files w + JOIN security_policy_parses sp ON sp.blob_oid = w.blob_oid + JOIN security_policy_parses lp + ON lp.source_kind = 'linked-page' AND lp.url = ANY(sp.linked_urls) + WHERE w.repo_id = r.id AND w.file_type = 'security' AND w.deleted_at IS NULL + ) AS page_parses, + (SELECT json_agg(json_build_object( + 'channel', sc.channel, 'value', sc.value, 'score', sc.score)) + FROM security_contacts sc + WHERE sc.repo_id = r.id AND sc.deleted_at IS NULL + ) AS fallback_contacts + FROM repos r + JOIN stale s ON s.id = r.id`, + { limit }, + ) +} + +export async function runAssembleStage(qx: QueryExecutor, cfg: Cfg): Promise { + const repos = await selectReposToAssemble(qx, cfg.assembleBatchSize) + const rows = repos.map((r) => { + const assembled = assembleProtocol({ + repoUrl: r.url, + pvrEnabled: r.pvr_enabled, + securityTxtUrl: r.security_txt_url, + fileParses: r.file_parses ?? [], + pageParses: r.page_parses ?? [], + fallbackContacts: (r.fallback_contacts ?? []).map((c) => ({ ...c, score: Number(c.score) })), + }) + return { + repo_id: r.id, + declared: assembled.declared, + methods: JSON.stringify(assembled.methods), + guidelines: assembled.guidelines ? JSON.stringify(assembled.guidelines) : null, + sources: JSON.stringify(assembled.sources), + } + }) + + if (rows.length > 0) { + await qx.result( + prepareBulkInsert('repo_reporting_protocols', PROTOCOL_COLUMNS, rows, PROTOCOL_UPSERT_SET), + ) + } + + const result: AssembleStageResult = { + reposAssembled: rows.length, + declaredCount: rows.filter((r) => r.declared).length, + } + log.info({ ...result }, 'Reporting protocol assemble stage complete') + return result +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/classify.ts b/services/apps/packages_worker/src/security-contacts/protocol/classify.ts new file mode 100644 index 0000000000..028d072a20 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/classify.ts @@ -0,0 +1,124 @@ +import { ClassifierVerdict, ParsedMethod, ProtocolMethodType } from './types' + +export const PARSER_VERSION = 1 + +const EMAIL_RE = /[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}/g +const URL_RE = /https?:\/\/[^\s)<>\]"']+/g +const HEADING_RE = /^#{1,6}\s+(.*)$/ +const KEYWORD_RE = + /\b(report|security|vulnerabilit(?:y|ies)|disclosure|disclose|contact|advisor(?:y|ies))\b/i +const NEGATIVE_SECTION_RE = /acknowledg|hall of fame|thanks|credits|honou?r|researchers/i +const PVR_RE = + /private vulnerability reporting|\/security\/advisories\/new|draft(?:ing)? a (?:new )?security advisory|\bGHSA\b|github security advisor/i +const BOUNTY_RE = + /hackerone\.com|bugcrowd\.com|huntr\.(?:dev|com)|intigriti\.com|yeswehack\.com|hackenproof\.com|tidelift\.com\/security/i +const MAILING_LIST_RE = /mailing list|lists\.[a-z0-9.-]+\.[a-z]{2,}|groups\.google\.com/i +const FORM_URL_RE = /\/(report|security\/report|vulnerability|disclosure|bounty)[a-z/-]*(\?|$|#)/i +const SECURITY_TXT_RE = /security\.txt/i +const PREFERENCE_RE = + /\bpreferred\b|\bpreferably\b|\bprimary\b|please use|we (?:strongly )?(?:recommend|encourage|prefer)|first choice/i +const CONDITIONAL_RE = + /if you (?:cannot|can't|are unable|are unsure)|only (?:if|when)|unless\b|is not possible/i +const NEGATION_RE = /do not|don't|never\b|avoid\b/i +const DEFAULT_TEMPLATE_RE = + /use this section to tell people about which versions|tell them where to go, how often they can expect/i +const GITHUB_PROFILE_RE = /^https?:\/\/(?:www\.)?github\.com\/[^/]+\/?$/i +const ISSUE_TRACKER_RE = /github\.com\/[^/]+\/[^/]+\/issues/i + +const POINTER_TEXT_MAX_CHARS = 200 + +interface Hit { + method: ParsedMethod + line: string +} + +function cleanUrl(url: string): string { + return url.replace(/[.,;:]+$/, '') +} + +function urlType(url: string): ProtocolMethodType | null { + if (BOUNTY_RE.test(url)) return 'bounty-platform' + if (ISSUE_TRACKER_RE.test(url)) return 'web-form' + if (FORM_URL_RE.test(url)) return 'web-form' + return null +} + +export function classifySecurityPolicy(text: string): ClassifierVerdict { + const lines = text.split('\n') + const hits: Hit[] = [] + const linkedUrls: string[] = [] + const seen = new Set() + let heading = '' + + const add = (type: ProtocolMethodType, endpoint: string, line: string) => { + const key = `${type}:${endpoint.toLowerCase()}` + if (seen.has(key)) return + seen.add(key) + hits.push({ method: { type, status: 'accepted', endpoint, condition: null }, line }) + } + + for (let i = 0; i < lines.length; i++) { + const line = lines[i] + const headingMatch = HEADING_RE.exec(line.trim()) + if (headingMatch) { + heading = headingMatch[1] + continue + } + if (NEGATIVE_SECTION_RE.test(heading)) continue + + if (PVR_RE.test(line)) add('github-pvr', 'github-pvr', line) + if (SECURITY_TXT_RE.test(line)) add('security-txt', 'security.txt', line) + + const window = lines.slice(Math.max(0, i - 2), i + 3).join('\n') + if (!KEYWORD_RE.test(window)) continue + + for (const email of line.match(EMAIL_RE) ?? []) add('email', email, line) + for (const raw of line.match(URL_RE) ?? []) { + const url = cleanUrl(raw) + if (/\/security\/advisories/i.test(url)) continue + if (GITHUB_PROFILE_RE.test(url)) continue + const type = urlType(url) + if (type) add(type, url, line) + else if (linkedUrls.length < 3 && !linkedUrls.includes(url)) linkedUrls.push(url) + } + if (MAILING_LIST_RE.test(line)) { + const listUrl = (line.match(URL_RE) ?? []).map(cleanUrl).find((u) => MAILING_LIST_RE.test(u)) + const listEmail = (line.match(EMAIL_RE) ?? [])[0] + if (listUrl || listEmail) add('mailing-list', listUrl ?? listEmail, line) + } + } + + for (const hit of hits) { + if (NEGATION_RE.test(hit.line) && hit.method.type !== 'github-pvr') { + hit.method.status = 'prohibited' + } + } + + const usable = hits.filter((h) => h.method.status !== 'prohibited') + const cued = usable.filter((h) => PREFERENCE_RE.test(h.line)) + const hasConditional = CONDITIONAL_RE.test(text) + const isTemplate = DEFAULT_TEMPLATE_RE.test(text) + + let clean = false + if (isTemplate) { + clean = true + } else if (!hasConditional && usable.length === 1) { + usable[0].method.status = 'preferred' + clean = true + } else if (!hasConditional && usable.length > 1 && cued.length === 1) { + cued[0].method.status = 'preferred' + clean = true + } + + const residual = text.replace(URL_RE, ' ').replace(/\s+/g, ' ').trim() + const pointerOnly = + hits.length === 0 && linkedUrls.length > 0 && residual.length <= POINTER_TEXT_MAX_CHARS + + return { + clean, + isTemplate, + pointerOnly, + methods: hits.map((h) => h.method), + linkedUrls, + } +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts b/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts new file mode 100644 index 0000000000..bb8a603506 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts @@ -0,0 +1,185 @@ +import { createHash } from 'node:crypto' +import { lookup } from 'node:dns/promises' + +import { parseGithubUrl } from '../../enricher/fetchLightRepo' +import type { githubApiGet } from '../githubToken' + +const MAX_PAGE_BYTES = 500_000 +const MAX_REDIRECTS = 3 +const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]) + +const BLOCKED_HOST_RE = /^(localhost|metadata\.google\.internal|.+\.(local|internal|localdomain))$/i +const IPV4_RE = /^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$/ + +export function sha256Hex(text: string): string { + return createHash('sha256').update(text).digest('hex') +} + +export async function fetchBlob( + deps: { githubGet: typeof githubApiGet }, + repoUrl: string, + blobOid: string, + timeoutMs: number, +): Promise { + let owner: string + let name: string + try { + ;({ owner, name } = parseGithubUrl(repoUrl)) + } catch { + return null + } + const { text } = await deps.githubGet(`/repos/${owner}/${name}/git/blobs/${blobOid}`, timeoutMs, { + raw: true, + }) + return text +} + +export function htmlToText(html: string): string { + return html + .replace(/]*>/gi, ' ') + .replace(/]*>/gi, ' ') + .replace(/<[^>]+>/g, ' ') + .replace(/</g, '<') + .replace(/>/g, '>') + .replace(/&#?(?!amp;)\w+;/g, ' ') + .replace(/&/g, '&') + .replace(/[ \t]+/g, ' ') + .replace(/\n{3,}/g, '\n\n') + .trim() +} + +function isPrivateIpv4(host: string): boolean { + const octets = host.split('.').map(Number) + if (octets.some((o) => o > 255)) return true + const [a, b] = octets + return ( + a === 0 || + a === 10 || + a === 127 || + (a === 100 && b >= 64 && b <= 127) || + (a === 169 && b === 254) || + (a === 172 && b >= 16 && b <= 31) || + (a === 192 && b === 168) + ) +} + +function isBlockedIpv6(addr: string): boolean { + const a = addr.toLowerCase() + if (a === '::1' || a === '::') return true + const mapped = a.match(/^::ffff:(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$/) + if (mapped) return isPrivateIpv4(mapped[1]) + if (/^f[cd][0-9a-f]{2}:/.test(a)) return true // fc00::/7 unique-local + if (/^fe[89ab][0-9a-f]:/.test(a)) return true // fe80::/10 link-local + return false +} + +export function isBlockedUrl(raw: string): boolean { + let url: URL + try { + url = new URL(raw) + } catch { + return true + } + if (url.protocol !== 'http:' && url.protocol !== 'https:') return true + const host = url.hostname + if (BLOCKED_HOST_RE.test(host)) return true + if (host.includes(':') || host.startsWith('[')) return true + if (IPV4_RE.test(host)) return isPrivateIpv4(host) + return false +} + +// The static isBlockedUrl check only rejects literal private IPs; a public +// hostname whose DNS record points at 127.0.0.1, RFC1918, or a metadata +// address slips through. Resolve every hop and validate the answers before +// connecting. This narrows but does not fully close the DNS-rebinding window +// (fetch resolves again internally); acceptable for a daily internal batch +// worker fetching public policy pages. +async function resolvesToBlockedAddress(host: string): Promise { + let addrs: Array<{ address: string; family: number }> + try { + addrs = await lookup(host, { all: true }) + } catch { + return true + } + if (addrs.length === 0) return true + return addrs.some(({ address, family }) => + family === 6 ? isBlockedIpv6(address) : isPrivateIpv4(address), + ) +} + +async function isRequestBlocked(raw: string): Promise { + if (isBlockedUrl(raw)) return true + const host = new URL(raw).hostname + if (IPV4_RE.test(host)) return false // public IPv4 literal already validated + return resolvesToBlockedAddress(host) +} + +async function fetchGuarded(url: string, signal: AbortSignal): Promise { + let current = url + for (let hop = 0; hop <= MAX_REDIRECTS; hop++) { + if (await isRequestBlocked(current)) return null + const res = await fetch(current, { + signal, + headers: { 'User-Agent': 'crowd.dev-reporting-protocol' }, + redirect: 'manual', + }) + if (REDIRECT_STATUSES.has(res.status)) { + const location = res.headers.get('location') + if (!location) return null + current = new URL(location, current).toString() + continue + } + return res + } + return null +} + +async function readBodyCapped(res: Response, maxBytes: number): Promise { + const reader = res.body?.getReader() + if (!reader) { + return (await res.text()).slice(0, maxBytes) + } + const chunks: Uint8Array[] = [] + let received = 0 + for (;;) { + const { done, value } = await reader.read() + if (done) break + if (value) { + chunks.push(value) + received += value.byteLength + if (received >= maxBytes) { + await reader.cancel() + break + } + } + } + const merged = new Uint8Array(Math.min(received, maxBytes)) + let offset = 0 + for (const chunk of chunks) { + const take = Math.min(chunk.byteLength, merged.byteLength - offset) + if (take <= 0) break + merged.set(chunk.subarray(0, take), offset) + offset += take + } + return new TextDecoder().decode(merged) +} + +export async function fetchLinkedPage( + url: string, + timeoutMs: number, +): Promise<{ text: string; hash: string } | null> { + const controller = new AbortController() + const timeoutHandle = setTimeout(() => controller.abort(), timeoutMs) + try { + const res = await fetchGuarded(url, controller.signal) + if (!res || !res.ok) return null + const raw = await readBodyCapped(res, MAX_PAGE_BYTES) + const text = htmlToText(raw) + if (!text) return null + return { text, hash: sha256Hex(text) } + } catch { + return null + } finally { + clearTimeout(timeoutHandle) + } +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts b/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts new file mode 100644 index 0000000000..f36a875214 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts @@ -0,0 +1,163 @@ +import { BedrockRuntimeClient, InvokeModelCommand } from '@aws-sdk/client-bedrock-runtime' + +import { parseLlmJson } from '@crowd/common' +import { getServiceChildLogger } from '@crowd/logging' +import { LLM_MODEL_REGION_MAP, LlmModelType } from '@crowd/types' + +import { ParsedProtocol } from './types' + +const log = getServiceChildLogger('reporting-protocol:llm') + +const MAX_TOKENS = 4096 +const DEFAULT_REGION = 'us-east-1' +const MAX_INPUT_CHARS = 100_000 + +// Claude Haiku 4.5 rates ($/token). Adjust if the model or Bedrock pricing changes. +const USD_PER_INPUT_TOKEN = 1 / 1_000_000 +const USD_PER_OUTPUT_TOKEN = 5 / 1_000_000 + +const PROTOCOL_SCHEMA = { + type: 'object', + additionalProperties: false, + required: ['methods', 'guidelines'], + properties: { + methods: { + type: 'array', + items: { + type: 'object', + additionalProperties: false, + required: ['type', 'status', 'endpoint', 'condition'], + properties: { + type: { + type: 'string', + enum: [ + 'github-pvr', + 'email', + 'web-form', + 'bounty-platform', + 'security-txt', + 'mailing-list', + ], + }, + status: { type: 'string', enum: ['preferred', 'accepted', 'fallback', 'prohibited'] }, + endpoint: { type: 'string' }, + condition: { type: ['string', 'null'] }, + }, + }, + }, + guidelines: { + type: ['object', 'null'], + additionalProperties: false, + required: ['generalPrinciples', 'avoid', 'recommend'], + properties: { + generalPrinciples: { type: 'array', items: { type: 'string' } }, + avoid: { type: 'array', items: { type: 'string' } }, + recommend: { + type: 'array', + items: { + type: 'object', + additionalProperties: false, + required: ['scenario', 'action'], + properties: { scenario: { type: 'string' }, action: { type: 'string' } }, + }, + }, + }, + }, + }, +} + +const SYSTEM_PROMPT = `You extract a project's declared vulnerability-reporting protocol from its security policy text. +Rules: +- Only emit methods the text actually declares. Never invent endpoints; copy them exactly as written (deobfuscate "name at host dot tld" into a plain email address). +- For github-pvr and security-txt methods, use the literal endpoint sentinels "github-pvr" and "security.txt". +- status: "preferred" for the method the project asks for first (at most one), "accepted" for other welcomed methods, "fallback" for methods offered only when others fail, "prohibited" for channels the text tells reporters NOT to use. +- condition: the verbatim-ish clause gating a method (e.g. "only if opening a GHSA is not possible"), else null. +- guidelines: distill reporting expectations into generalPrinciples / avoid / recommend. Use null when the text has none. +- Output nothing but the JSON.` + +export interface LlmExtractConfig { + modelId: string + timeoutMs: number + accessKeyId: string | undefined + secretAccessKey: string | undefined +} + +export interface LlmExtractResult { + parsed: ParsedProtocol | null + costUsd: number | null +} + +const clients = new Map() + +function clientFor(cfg: LlmExtractConfig, region: string): BedrockRuntimeClient { + const key = `${region}:${cfg.accessKeyId}` + let client = clients.get(key) + if (!client) { + client = new BedrockRuntimeClient({ + region, + credentials: { + accessKeyId: cfg.accessKeyId as string, + secretAccessKey: cfg.secretAccessKey as string, + }, + }) + clients.set(key, client) + } + return client +} + +function usageCostUsd(body: unknown): number | null { + const usage = (body as { usage?: { input_tokens?: number; output_tokens?: number } })?.usage + if (!usage) return null + return (usage.input_tokens ?? 0) * USD_PER_INPUT_TOKEN + (usage.output_tokens ?? 0) * USD_PER_OUTPUT_TOKEN +} + +export async function llmExtractProtocol( + text: string, + cfg: LlmExtractConfig, +): Promise { + if (!cfg.accessKeyId || !cfg.secretAccessKey) { + log.warn({ modelId: cfg.modelId }, 'Missing Bedrock credentials — skipping LLM extraction') + return { parsed: null, costUsd: null } + } + + const region = LLM_MODEL_REGION_MAP[cfg.modelId as LlmModelType] ?? DEFAULT_REGION + const controller = new AbortController() + const timeoutHandle = setTimeout(() => controller.abort(), cfg.timeoutMs) + try { + const command = new InvokeModelCommand({ + modelId: cfg.modelId, + accept: 'application/json', + contentType: 'application/json', + body: JSON.stringify({ + anthropic_version: 'bedrock-2023-05-31', + max_tokens: MAX_TOKENS, + system: `${SYSTEM_PROMPT}\n\nRespond ONLY with a single JSON object matching this JSON schema, no prose:\n${JSON.stringify(PROTOCOL_SCHEMA)}`, + messages: [ + { role: 'user', content: [{ type: 'text', text: text.slice(0, MAX_INPUT_CHARS) }] }, + ], + }), + }) + const res = await clientFor(cfg, region).send(command, { abortSignal: controller.signal }) + const body = JSON.parse(res.body.transformToString()) + const costUsd = usageCostUsd(body) + const answer: string | undefined = body?.content?.[0]?.text + if (!answer) return { parsed: null, costUsd } + try { + return { parsed: parseLlmJson(answer), costUsd } + } catch (err) { + log.warn( + { errMsg: (err as Error).message, modelId: cfg.modelId }, + 'LLM protocol extraction failed', + ) + return { parsed: null, costUsd } + } + } catch (err) { + log.warn( + { errMsg: (err as Error).message, modelId: cfg.modelId }, + 'LLM protocol extraction failed', + ) + return { parsed: null, costUsd: null } + } finally { + clearTimeout(timeoutHandle) + } +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts new file mode 100644 index 0000000000..3ec475391b --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts @@ -0,0 +1,274 @@ +import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' +import { getServiceChildLogger } from '@crowd/logging' + +import { getReportingProtocolConfig } from '../../config' +import type { githubApiGet } from '../githubToken' + +import { PARSER_VERSION, classifySecurityPolicy } from './classify' +import { fetchBlob, fetchLinkedPage, sha256Hex } from './fetchContent' +import { llmExtractProtocol } from './llmExtract' +import { ParseRowStatus, ParseStageResult, ParsedProtocol } from './types' +import { validateParsedProtocol } from './validate' + +const log = getServiceChildLogger('reporting-protocol:parse') + +export interface ParseStageDeps { + githubGet: typeof githubApiGet + fetchPage: typeof fetchLinkedPage + llmExtract: typeof llmExtractProtocol +} + +type Cfg = ReturnType + +interface BlobJob { + blob_oid: string + url: string +} + +// Random batch order: failed blobs get no parse row, so a deterministic +// ORDER BY window full of permanent failures would starve everything behind it. +async function selectUnparsedBlobs(qx: QueryExecutor, limit: number): Promise { + return qx.select( + `SELECT * FROM ( + SELECT DISTINCT ON (w.blob_oid) w.blob_oid, r.url + FROM repo_well_known_files w + JOIN repos r ON r.id = w.repo_id AND r.host = 'github' + JOIN package_repos pr ON pr.repo_id = w.repo_id + JOIN packages p ON p.id = pr.package_id AND p.is_critical + LEFT JOIN security_policy_parses sp + ON sp.blob_oid = w.blob_oid AND sp.parser_version = $(parserVersion) + WHERE w.file_type = 'security' AND w.deleted_at IS NULL AND sp.blob_oid IS NULL + ORDER BY w.blob_oid, w.repo_id + ) unparsed + ORDER BY random() + LIMIT $(limit)`, + { limit, parserVersion: PARSER_VERSION }, + ) +} + +async function insertParse( + qx: QueryExecutor, + row: { + blobOid: string + sourceKind: 'security-file' | 'linked-page' + url: string | null + parser: 'deterministic' | 'llm' + status: ParseRowStatus + parsed: ParsedProtocol + linkedUrls: string[] + costUsd: number | null + }, +): Promise { + await qx.result( + `INSERT INTO security_policy_parses + (blob_oid, source_kind, url, parser, parser_version, status, parsed, linked_urls, llm_cost_usd) + VALUES ($(blobOid), $(sourceKind), $(url), $(parser), $(parserVersion), $(status), $(parsed), $(linkedUrls), $(costUsd)) + ON CONFLICT (blob_oid) DO UPDATE SET + source_kind = EXCLUDED.source_kind, + url = EXCLUDED.url, + parser = EXCLUDED.parser, + parser_version = EXCLUDED.parser_version, + status = EXCLUDED.status, + parsed = EXCLUDED.parsed, + linked_urls = EXCLUDED.linked_urls, + llm_cost_usd = EXCLUDED.llm_cost_usd, + parsed_at = NOW()`, + { ...row, parsed: JSON.stringify(row.parsed), parserVersion: PARSER_VERSION }, + ) +} + +function createLimiter(max: number): (fn: () => Promise) => Promise { + let active = 0 + const waiters: Array<() => void> = [] + return async (fn: () => Promise): Promise => { + if (active >= max) { + await new Promise((resolve) => waiters.push(resolve)) + } + active++ + try { + return await fn() + } finally { + active-- + const next = waiters.shift() + if (next) next() + } + } +} + +async function parseText( + text: string, + deps: ParseStageDeps, + cfg: Cfg, +): Promise<{ + parser: 'deterministic' | 'llm' + status: ParseRowStatus + parsed: ParsedProtocol + linkedUrls: string[] + costUsd: number | null +}> { + const verdict = classifySecurityPolicy(text) + if (verdict.clean) { + return { + parser: 'deterministic', + status: verdict.isTemplate ? 'template' : 'ok', + parsed: { methods: verdict.methods, guidelines: null }, + linkedUrls: verdict.linkedUrls, + costUsd: null, + } + } + if (verdict.pointerOnly) { + return { + parser: 'deterministic', + status: 'ok', + parsed: { methods: verdict.methods, guidelines: null }, + linkedUrls: verdict.linkedUrls, + costUsd: null, + } + } + const { parsed: llmParsed, costUsd } = await deps.llmExtract(text, { + modelId: cfg.llmModelId, + timeoutMs: cfg.llmTimeoutMs, + accessKeyId: cfg.llmAccessKeyId, + secretAccessKey: cfg.llmSecretAccessKey, + }) + if (llmParsed && validateParsedProtocol(llmParsed, text).ok) { + return { + parser: 'llm', + status: 'ok', + parsed: llmParsed, + linkedUrls: verdict.linkedUrls, + costUsd, + } + } + return { + parser: 'llm', + status: 'degraded', + parsed: { methods: verdict.methods, guidelines: null }, + linkedUrls: verdict.linkedUrls, + costUsd, + } +} + +async function alreadyParsed(qx: QueryExecutor, oid: string): Promise { + const row = await qx.selectOneOrNone( + 'SELECT 1 FROM security_policy_parses WHERE blob_oid = $(oid) AND parser_version = $(parserVersion)', + { oid, parserVersion: PARSER_VERSION }, + ) + return row !== null +} + +// Linked-page rows are keyed by sha256(url), not content hash: two URLs with +// identical content must both stay joinable from linked_urls at assembly. +async function processLinkedPages( + qx: QueryExecutor, + deps: ParseStageDeps, + cfg: Cfg, + linkedUrls: string[], + result: ParseStageResult, +): Promise<{ anyFailure: boolean }> { + let anyFailure = false + for (const url of linkedUrls) { + const pageKey = sha256Hex(url) + if (await alreadyParsed(qx, pageKey)) continue + const page = await deps.fetchPage(url, cfg.fetchTimeoutMs) + if (!page) { + anyFailure = true + continue + } + const pageParse = await parseText(page.text, deps, cfg) + await insertParse(qx, { + blobOid: pageKey, + sourceKind: 'linked-page', + url, + parser: pageParse.parser, + status: pageParse.status, + parsed: pageParse.parsed, + linkedUrls: [], + costUsd: pageParse.costUsd, + }) + result.linkedPages++ + if (pageParse.costUsd !== null) result.llmCostUsd += pageParse.costUsd + } + return { anyFailure } +} + +export async function runParseStage( + qx: QueryExecutor, + deps: ParseStageDeps, + cfg: Cfg, +): Promise { + const jobs = await selectUnparsedBlobs(qx, cfg.parseBatchSize) + const result: ParseStageResult = { + blobsParsed: 0, + deterministic: 0, + llm: 0, + degraded: 0, + template: 0, + linkedPages: 0, + failed: 0, + llmCostUsd: 0, + } + + const limitLlm = createLimiter(Math.max(1, cfg.llmConcurrency)) + const limitedDeps: ParseStageDeps = { + ...deps, + llmExtract: (text, llmCfg) => limitLlm(() => deps.llmExtract(text, llmCfg)), + } + + let cursor = 0 + const workers = Array.from({ length: Math.min(cfg.concurrency, jobs.length) }, async () => { + while (cursor < jobs.length) { + const job = jobs[cursor++] + try { + const text = await fetchBlob( + { githubGet: deps.githubGet }, + job.url, + job.blob_oid, + cfg.fetchTimeoutMs, + ) + if (text === null) { + result.failed++ + continue + } + const fileParse = await parseText(text, limitedDeps, cfg) + const pointerOnly = fileParse.parsed.methods.length === 0 && fileParse.linkedUrls.length > 0 + + if (pointerOnly) { + // The file row marks the blob done, so it is written only once every + // linked page has a parse row — a transient page failure leaves the + // blob unmarked and the next sweep retries the whole unit. + const { anyFailure } = await processLinkedPages( + qx, + limitedDeps, + cfg, + fileParse.linkedUrls, + result, + ) + if (anyFailure) { + result.failed++ + continue + } + } + + await insertParse(qx, { + blobOid: job.blob_oid, + sourceKind: 'security-file', + url: null, + ...fileParse, + }) + result.blobsParsed++ + result[fileParse.parser === 'deterministic' ? 'deterministic' : 'llm']++ + if (fileParse.status === 'degraded') result.degraded++ + if (fileParse.status === 'template') result.template++ + if (fileParse.costUsd !== null) result.llmCostUsd += fileParse.costUsd + } catch (err) { + result.failed++ + log.warn({ blobOid: job.blob_oid, errMsg: (err as Error).message }, 'Blob parse failed') + } + } + }) + await Promise.all(workers) + + log.info({ ...result }, 'Reporting protocol parse stage complete') + return result +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/schedule.ts b/services/apps/packages_worker/src/security-contacts/protocol/schedule.ts new file mode 100644 index 0000000000..1fdcd69791 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/schedule.ts @@ -0,0 +1,57 @@ +import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client' + +import { svc } from '../../service' +import { ingestReportingProtocols } from '../../workflows' + +const SCHEDULE_ID = 'reporting-protocol-ingestion' +const WORKFLOW_EXECUTION_TIMEOUT = '24 hours' + +function scheduleAction() { + return { + type: 'startWorkflow' as const, + workflowType: ingestReportingProtocols, + workflowId: 'reporting-protocol-daily', + taskQueue: 'security-contacts-worker', + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + retry: { + initialInterval: '30 seconds', + backoffCoefficient: 2, + maximumAttempts: 3, + }, + args: [] as [], + } +} + +export async function scheduleReportingProtocolIngestion(): Promise { + const { temporal } = svc + if (!temporal) throw new Error('Temporal client not initialized') + + try { + await temporal.schedule.create({ + scheduleId: SCHEDULE_ID, + spec: { + cronExpressions: ['0 7 * * *'], + }, + policies: { + overlap: ScheduleOverlapPolicy.SKIP, + catchupWindow: '1 hour', + }, + action: scheduleAction(), + }) + } catch (err) { + if (err instanceof ScheduleAlreadyRunning) { + svc.log.info('Schedule reporting-protocol-ingestion already exists, reconciling action.') + const handle = temporal.schedule.getHandle(SCHEDULE_ID) + await handle.update((prev) => ({ + ...prev, + policies: { + ...prev.policies, + overlap: ScheduleOverlapPolicy.SKIP, + }, + action: scheduleAction(), + })) + } else { + throw err + } + } +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/types.ts b/services/apps/packages_worker/src/security-contacts/protocol/types.ts new file mode 100644 index 0000000000..971bdcf364 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/types.ts @@ -0,0 +1,74 @@ +export type ProtocolMethodType = + | 'github-pvr' + | 'email' + | 'web-form' + | 'bounty-platform' + | 'security-txt' + | 'mailing-list' + +export type ProtocolMethodStatus = 'preferred' | 'accepted' | 'fallback' | 'prohibited' + +export type MethodConfidence = 'declared' | 'inferred' + +export interface ParsedMethod { + type: ProtocolMethodType + status: ProtocolMethodStatus + endpoint: string + condition: string | null +} + +export interface ProtocolGuidelines { + generalPrinciples: string[] + avoid: string[] + recommend: Array<{ scenario: string; action: string }> +} + +export interface ParsedProtocol { + methods: ParsedMethod[] + guidelines: ProtocolGuidelines | null +} + +export type ParseRowStatus = 'ok' | 'template' | 'degraded' + +export interface ClassifierVerdict { + clean: boolean + isTemplate: boolean + pointerOnly: boolean + methods: ParsedMethod[] + linkedUrls: string[] +} + +export interface ProtocolMethod extends ParsedMethod { + confidence: MethodConfidence + provenance: { + path?: string + blobOid?: string + url?: string + parser?: 'deterministic' | 'llm' + api?: string + channel?: string + } +} + +export interface AssembledProtocol { + declared: boolean + methods: ProtocolMethod[] + guidelines: ProtocolGuidelines | null + sources: Array> +} + +export interface ParseStageResult { + blobsParsed: number + deterministic: number + llm: number + degraded: number + template: number + linkedPages: number + failed: number + llmCostUsd: number +} + +export interface AssembleStageResult { + reposAssembled: number + declaredCount: number +} diff --git a/services/apps/packages_worker/src/security-contacts/protocol/validate.ts b/services/apps/packages_worker/src/security-contacts/protocol/validate.ts new file mode 100644 index 0000000000..54781c34ba --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/validate.ts @@ -0,0 +1,80 @@ +import { ParsedProtocol } from './types' + +const METHOD_TYPES = new Set([ + 'github-pvr', + 'email', + 'web-form', + 'bounty-platform', + 'security-txt', + 'mailing-list', +]) +const METHOD_STATUSES = new Set(['preferred', 'accepted', 'fallback', 'prohibited']) +export const SENTINEL_ENDPOINT_BY_TYPE: Record = { + 'github-pvr': 'github-pvr', + 'security-txt': 'security.txt', +} + +function normalize(text: string): string { + return text + .toLowerCase() + .replace(/\s*(?:\[at\]|\(at\)| at )\s*/g, '@') + .replace(/\s*(?:\[dot\]|\(dot\)| dot )\s*/g, '.') + .replace(/\s+/g, ' ') +} + +function isStringArray(value: unknown): value is string[] { + return Array.isArray(value) && value.every((v) => typeof v === 'string') +} + +function hasValidGuidelinesShape(parsed: ParsedProtocol): boolean { + const g = parsed.guidelines + if (g === null || g === undefined) return true + return ( + typeof g === 'object' && + isStringArray(g.generalPrinciples) && + isStringArray(g.avoid) && + Array.isArray(g.recommend) && + g.recommend.every((r) => r && typeof r.scenario === 'string' && typeof r.action === 'string') + ) +} + +export function validateParsedProtocol( + parsed: ParsedProtocol, + sourceText: string, +): { ok: boolean; reasons: string[] } { + if (!parsed || !Array.isArray(parsed.methods)) { + return { ok: false, reasons: ['methods is not an array'] } + } + + const reasons: string[] = [] + if (!hasValidGuidelinesShape(parsed)) reasons.push('malformed guidelines') + + const lowerSource = sourceText.toLowerCase() + const normalizedSource = normalize(sourceText) + + let preferredCount = 0 + for (const m of parsed.methods) { + if (!m || typeof m !== 'object') { + reasons.push('malformed method entry') + continue + } + if (!METHOD_TYPES.has(m.type)) reasons.push(`invalid type: ${m.type}`) + if (!METHOD_STATUSES.has(m.status)) reasons.push(`invalid status: ${m.status}`) + if (m.condition !== null && m.condition !== undefined && typeof m.condition !== 'string') { + reasons.push(`malformed condition on: ${m.type}`) + } + if (m.status === 'preferred') preferredCount++ + if (typeof m.endpoint !== 'string' || m.endpoint.trim() === '') { + reasons.push(`empty endpoint on: ${m.type}`) + continue + } + if (SENTINEL_ENDPOINT_BY_TYPE[m.type] === m.endpoint) continue + const needle = m.endpoint.toLowerCase() + if (!lowerSource.includes(needle) && !normalizedSource.includes(needle)) { + reasons.push(`endpoint not in source: ${m.endpoint}`) + } + } + if (preferredCount > 1) reasons.push(`multiple preferred methods: ${preferredCount}`) + + return { ok: reasons.length === 0, reasons } +} diff --git a/services/apps/packages_worker/src/security-contacts/workflows.ts b/services/apps/packages_worker/src/security-contacts/workflows.ts index 2bfff86514..9e094aec0d 100644 --- a/services/apps/packages_worker/src/security-contacts/workflows.ts +++ b/services/apps/packages_worker/src/security-contacts/workflows.ts @@ -40,3 +40,15 @@ export async function ingestSecurityContactsForPurlWorkflow( ): Promise { return singleActs.ingestSecurityContactsForPurlActivity(purl) } + +export async function ingestReportingProtocols(): Promise { + const parsed = await acts.runProtocolParseBatch() + if (parsed.blobsParsed > 0) { + await continueAsNew() + } + const assembled = await acts.runProtocolAssembleBatch() + if (assembled.reposAssembled > 0) { + await continueAsNew() + } + log.info('Reporting protocol ingestion complete — no more work, exiting.') +} diff --git a/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql b/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql new file mode 100644 index 0000000000..65963905d7 --- /dev/null +++ b/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql @@ -0,0 +1,74 @@ +-- Prod validation for the reporting-protocol pipeline (all read-only). +-- Run section by section; each header says what healthy looks like. + +-- 1. Parse progress and split +-- Healthy: rows climb toward ~6.1k distinct security blobs; deterministic ≈ 65-75%, +-- degraded well under 10%. +SELECT source_kind, parser, status, COUNT(*) +FROM security_policy_parses +GROUP BY source_kind, parser, status +ORDER BY source_kind, parser, status; + +-- 2. Assembly coverage +-- Healthy: repos_with_protocol approaches the critical population; declared_pct ≈ 15-20% +-- (files + PVR flag overlap). +WITH critical AS ( + SELECT DISTINCT r.id + 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 +) +SELECT + (SELECT COUNT(*) FROM critical) AS critical_repos, + COUNT(*) AS repos_with_protocol, + COUNT(*) FILTER (WHERE rp.declared) AS declared, + ROUND(COUNT(*) FILTER (WHERE rp.declared)::numeric / NULLIF(COUNT(*), 0) * 100, 1) AS declared_pct +FROM repo_reporting_protocols rp +JOIN critical c ON c.id = rp.repo_id; + +-- 3. Method sanity +-- Always returns four rows; healthy: every count is zero. +SELECT 'bad_type' AS problem, COUNT(*) +FROM repo_reporting_protocols rp, + jsonb_array_elements(rp.methods) m +WHERE m->>'type' IS NULL + OR m->>'type' NOT IN ('github-pvr','email','web-form','bounty-platform','security-txt','mailing-list') +UNION ALL +SELECT 'bad_status', COUNT(*) +FROM repo_reporting_protocols rp, + jsonb_array_elements(rp.methods) m +WHERE m->>'status' IS NULL + OR m->>'status' NOT IN ('preferred','accepted','fallback','prohibited') +UNION ALL +SELECT 'multiple_preferred', COUNT(*) +FROM ( + SELECT rp.repo_id + FROM repo_reporting_protocols rp, + jsonb_array_elements(rp.methods) m + WHERE m->>'status' = 'preferred' + GROUP BY rp.repo_id + HAVING COUNT(*) > 1 +) x +UNION ALL +SELECT 'inferred_marked_declared', COUNT(*) +FROM repo_reporting_protocols rp, + jsonb_array_elements(rp.methods) m +WHERE rp.declared = false AND m->>'confidence' = 'declared'; + +-- 4. PVR flag consistency +-- Healthy: zero — pvr_enabled=false repos must carry no github-pvr method. +SELECT COUNT(*) +FROM repo_reporting_protocols rp +JOIN repos r ON r.id = rp.repo_id AND r.pvr_enabled = false +WHERE EXISTS ( + SELECT 1 FROM jsonb_array_elements(rp.methods) m WHERE m->>'type' = 'github-pvr'); + +-- 5. Ground-truth sample: 15 declared + 15 inferred for manual spot-check +(SELECT r.url, rp.declared, rp.methods + FROM repo_reporting_protocols rp JOIN repos r ON r.id = rp.repo_id + WHERE rp.declared ORDER BY rp.assembled_at DESC LIMIT 15) +UNION ALL +(SELECT r.url, rp.declared, rp.methods + FROM repo_reporting_protocols rp JOIN repos r ON r.id = rp.repo_id + WHERE NOT rp.declared AND jsonb_array_length(rp.methods) > 0 + ORDER BY rp.assembled_at DESC LIMIT 15); diff --git a/services/apps/packages_worker/src/workflows/index.ts b/services/apps/packages_worker/src/workflows/index.ts index 4b9712187b..2ce5b58e60 100644 --- a/services/apps/packages_worker/src/workflows/index.ts +++ b/services/apps/packages_worker/src/workflows/index.ts @@ -35,5 +35,6 @@ export { ingestRubyGemsCriticalDetails, ingestRubyGemsPackages } from '../rubyge export { ingestSecurityContacts, ingestSecurityContactsForPurlWorkflow, + ingestReportingProtocols, } from '../security-contacts/workflows' export { analyzeBlastRadius } from '../blast-radius/workflows'