From 366de3cd0c11da06b8f53e4f6b348037b87bd900 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 28 Jul 2026 16:18:35 +0100 Subject: [PATCH 01/13] feat: add reporting protocol tables Signed-off-by: Mouad BANI --- .../V1785283200__reporting_protocol.sql | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) create mode 100644 backend/src/osspckgs/migrations/V1785283200__reporting_protocol.sql 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() +); From 4dd407d9422015225677f3b9257fca7ce597cc46 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 28 Jul 2026 16:31:51 +0100 Subject: [PATCH 02/13] feat: add deterministic security policy classifier Signed-off-by: Mouad BANI --- .../security-contacts/protocol/classify.ts | 117 ++++++++++++++++++ .../src/security-contacts/protocol/types.ts | 72 +++++++++++ 2 files changed, 189 insertions(+) create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/classify.ts create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/types.ts 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..738c641182 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/classify.ts @@ -0,0 +1,117 @@ +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 + +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 + } + + return { + clean, + isTemplate, + methods: hits.map((h) => h.method), + linkedUrls, + } +} 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..b20a92421d --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/types.ts @@ -0,0 +1,72 @@ +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 + 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 +} + +export interface AssembleStageResult { + reposAssembled: number + declaredCount: number +} From 4f0a7898827f567b87f15b56b5419bc2fb37a9ff Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 28 Jul 2026 16:40:10 +0100 Subject: [PATCH 03/13] feat: add parsed protocol validator Signed-off-by: Mouad BANI --- .../security-contacts/protocol/validate.ts | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/validate.ts 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..e93dee33ae --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/validate.ts @@ -0,0 +1,44 @@ +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']) +const SENTINEL_ENDPOINTS = new Set(['github-pvr', '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, ' ') +} + +export function validateParsedProtocol( + parsed: ParsedProtocol, + sourceText: string, +): { ok: boolean; reasons: string[] } { + const reasons: string[] = [] + const lowerSource = sourceText.toLowerCase() + const normalizedSource = normalize(sourceText) + + let preferredCount = 0 + for (const m of parsed.methods) { + 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.status === 'preferred') preferredCount++ + if (SENTINEL_ENDPOINTS.has(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 } +} From 826a217c460afd4633d4818580042ea2c8cb5125 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 28 Jul 2026 16:54:47 +0100 Subject: [PATCH 04/13] feat: add blob and linked-page fetchers for reporting protocol Signed-off-by: Mouad BANI --- .../protocol/fetchContent.ts | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts 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..34c0e6b75c --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts @@ -0,0 +1,67 @@ +import { createHash } from 'node:crypto' + +import { parseGithubUrl } from '../../enricher/fetchLightRepo' +import type { githubApiGet } from '../githubToken' + +const MAX_PAGE_BYTES = 500_000 + +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(/>/g, '>') + .replace(/&#?\w+;/g, ' ') + .replace(/[ \t]+/g, ' ') + .replace(/\n{3,}/g, '\n\n') + .trim() +} + +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 fetch(url, { + signal: controller.signal, + headers: { 'User-Agent': 'crowd.dev-reporting-protocol' }, + redirect: 'follow', + }) + if (!res.ok) return null + const raw = await res.text() + const text = htmlToText(raw.slice(0, MAX_PAGE_BYTES)) + if (!text) return null + return { text, hash: sha256Hex(text) } + } catch { + return null + } finally { + clearTimeout(timeoutHandle) + } +} From e4f71574fa06825e1915d28b3d1fca83aaf5f87c Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 10:52:31 +0100 Subject: [PATCH 05/13] feat: add bedrock LLM protocol extraction with strict schema Signed-off-by: Mouad BANI --- pnpm-lock.yaml | 31 ++-- services/apps/packages_worker/package.json | 1 + services/apps/packages_worker/src/config.ts | 16 ++ .../security-contacts/protocol/llmExtract.ts | 139 ++++++++++++++++++ 4 files changed, 173 insertions(+), 14 deletions(-) create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts 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/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/protocol/llmExtract.ts b/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts new file mode 100644 index 0000000000..35d0168ac1 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts @@ -0,0 +1,139 @@ +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 + +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 +} + +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 +} + +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 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 answer: string | undefined = body?.content?.[0]?.text + if (!answer) return null + return parseLlmJson(answer) + } catch (err) { + log.warn( + { errMsg: (err as Error).message, modelId: cfg.modelId }, + 'LLM protocol extraction failed', + ) + return null + } finally { + clearTimeout(timeoutHandle) + } +} From 0474d820c09e43919e17819510fa2c5a4cb1e309 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 11:03:30 +0100 Subject: [PATCH 06/13] feat: add reporting protocol parse stage Signed-off-by: Mouad BANI --- .../security-contacts/protocol/parseStage.ts | 188 ++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts 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..edbd3f7020 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts @@ -0,0 +1,188 @@ +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 } 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 +} + +async function selectUnparsedBlobs(qx: QueryExecutor, limit: number): Promise { + return qx.select( + `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 + 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[] + }, +): Promise { + await qx.result( + `INSERT INTO security_policy_parses + (blob_oid, source_kind, url, parser, parser_version, status, parsed, linked_urls) + VALUES ($(blobOid), $(sourceKind), $(url), $(parser), $(parserVersion), $(status), $(parsed), $(linkedUrls)) + 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, + parsed_at = NOW()`, + { ...row, parsed: JSON.stringify(row.parsed), parserVersion: PARSER_VERSION }, + ) +} + +async function parseText( + text: string, + deps: ParseStageDeps, + cfg: Cfg, +): Promise<{ + parser: 'deterministic' | 'llm' + status: ParseRowStatus + parsed: ParsedProtocol + linkedUrls: string[] +}> { + const verdict = classifySecurityPolicy(text) + if (verdict.clean) { + return { + parser: 'deterministic', + status: verdict.isTemplate ? 'template' : 'ok', + parsed: { methods: verdict.methods, guidelines: null }, + linkedUrls: verdict.linkedUrls, + } + } + const llmParsed = 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 } + } + return { + parser: 'llm', + status: 'degraded', + parsed: { methods: verdict.methods, guidelines: null }, + linkedUrls: verdict.linkedUrls, + } +} + +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 +} + +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, + } + + 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, deps, cfg) + 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.parsed.methods.length === 0 && fileParse.linkedUrls.length > 0) { + for (const url of fileParse.linkedUrls) { + const page = await deps.fetchPage(url, cfg.fetchTimeoutMs) + if (!page) continue + if (await alreadyParsed(qx, page.hash)) continue + const pageParse = await parseText(page.text, deps, cfg) + await insertParse(qx, { + blobOid: page.hash, + sourceKind: 'linked-page', + url, + parser: pageParse.parser, + status: pageParse.status, + parsed: pageParse.parsed, + linkedUrls: [], + }) + result.linkedPages++ + } + } + } 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 +} From 51f9e23f36fd3501039e963e6f3eadf9f620a9f9 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 11:22:18 +0100 Subject: [PATCH 07/13] feat: add reporting protocol assemble stage Signed-off-by: Mouad BANI --- .../security-contacts/protocol/assemble.ts | 151 ++++++++++++++++++ .../protocol/assembleStage.ts | 109 +++++++++++++ 2 files changed, 260 insertions(+) create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/assemble.ts create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/assembleStage.ts 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..3253a92215 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts @@ -0,0 +1,151 @@ +import { parseGithubUrl } from '../../enricher/fetchLightRepo' + +import { + AssembledProtocol, + ParseRowStatus, + ParsedProtocol, + ProtocolMethod, + ProtocolMethodType, +} from './types' + +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) + } + + 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) { + push({ + ...m, + endpoint: m.type === 'github-pvr' ? pvrAdvisoryUrl(input.repoUrl) : 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) { + push({ + ...m, + endpoint: m.type === 'github-pvr' ? pvrAdvisoryUrl(input.repoUrl) : 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 +} From 22321886516757e73dd8f6979516179605e00714 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 11:31:38 +0100 Subject: [PATCH 08/13] feat: wire reporting protocol workflow and schedule Signed-off-by: Mouad BANI --- .../apps/packages_worker/src/activities.ts | 2 + .../src/bin/security-contacts-worker.ts | 2 + .../src/security-contacts/activities.ts | 49 +++++++++++++++- .../security-contacts/protocol/schedule.ts | 57 +++++++++++++++++++ .../src/security-contacts/workflows.ts | 12 ++++ .../packages_worker/src/workflows/index.ts | 1 + 6 files changed, 122 insertions(+), 1 deletion(-) create mode 100644 services/apps/packages_worker/src/security-contacts/protocol/schedule.ts diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index e470b91054..e44a0c6a22 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/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/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/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/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' From bfef882f6f4929a457b8bdedfe0365dfca6f6df2 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 12:17:47 +0100 Subject: [PATCH 09/13] docs: record reporting protocol decisions in ADR-0010 Signed-off-by: Mouad BANI --- docs/adr/0010-security-contacts-worker.md | 100 ++++++++++++++++++ .../src/tmp/validate-reporting-protocol.sql | 72 +++++++++++++ 2 files changed, 172 insertions(+) create mode 100644 services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql diff --git a/docs/adr/0010-security-contacts-worker.md b/docs/adr/0010-security-contacts-worker.md index 351c225db6..3c0766b824 100644 --- a/docs/adr/0010-security-contacts-worker.md +++ b/docs/adr/0010-security-contacts-worker.md @@ -281,3 +281,103 @@ 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 extracted text for linked pages), `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 (one hop, 500 KB cap, text-extracted) and parsed under the same +content-keyed cache as `linked-page` rows. + +### 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/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..b67a5e6622 --- /dev/null +++ b/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql @@ -0,0 +1,72 @@ +-- 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 +-- Healthy: zero rows. +SELECT 'bad_type' AS problem, COUNT(*) +FROM repo_reporting_protocols rp, + jsonb_array_elements(rp.methods) m +WHERE 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' 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); From 424a10ed72f0d8513b0ce9d1b4df065addc60219 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 13:49:42 +0100 Subject: [PATCH 10/13] fix(review): address PR #4413 review feedback Signed-off-by: Mouad BANI --- docs/adr/0010-security-contacts-worker.md | 11 +- .../protocol/fetchContent.ts | 105 +++++++++++++-- .../security-contacts/protocol/parseStage.ts | 122 +++++++++++++----- .../security-contacts/protocol/validate.ts | 40 +++++- .../src/tmp/validate-reporting-protocol.sql | 8 +- 5 files changed, 236 insertions(+), 50 deletions(-) diff --git a/docs/adr/0010-security-contacts-worker.md b/docs/adr/0010-security-contacts-worker.md index 3c0766b824..affb7c030c 100644 --- a/docs/adr/0010-security-contacts-worker.md +++ b/docs/adr/0010-security-contacts-worker.md @@ -307,7 +307,8 @@ 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 extracted text for linked pages), `source_kind` + 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, @@ -332,8 +333,12 @@ requires every emitted endpoint to appear in the source (URLs verbatim; emails a 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 (one hop, 500 KB cap, text-extracted) and parsed under the same -content-keyed cache as `linked-page` rows. +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 diff --git a/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts b/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts index 34c0e6b75c..de3e1414b0 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts @@ -4,6 +4,11 @@ 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') @@ -30,18 +35,98 @@ export async function fetchBlob( export function htmlToText(html: string): string { return html - .replace(//gi, ' ') - .replace(//gi, ' ') + .replace(//gi, ' ') + .replace(//gi, ' ') .replace(/<[^>]+>/g, ' ') - .replace(/&/g, '&') .replace(/</g, '<') .replace(/>/g, '>') - .replace(/&#?\w+;/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) + ) +} + +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 +} + +async function fetchGuarded(url: string, signal: AbortSignal): Promise { + let current = url + for (let hop = 0; hop <= MAX_REDIRECTS; hop++) { + if (isBlockedUrl(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, @@ -49,14 +134,10 @@ export async function fetchLinkedPage( const controller = new AbortController() const timeoutHandle = setTimeout(() => controller.abort(), timeoutMs) try { - const res = await fetch(url, { - signal: controller.signal, - headers: { 'User-Agent': 'crowd.dev-reporting-protocol' }, - redirect: 'follow', - }) - if (!res.ok) return null - const raw = await res.text() - const text = htmlToText(raw.slice(0, MAX_PAGE_BYTES)) + 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 { diff --git a/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts index edbd3f7020..fad13ca568 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts @@ -5,7 +5,7 @@ import { getReportingProtocolConfig } from '../../config' import type { githubApiGet } from '../githubToken' import { PARSER_VERSION, classifySecurityPolicy } from './classify' -import { fetchBlob, fetchLinkedPage } from './fetchContent' +import { fetchBlob, fetchLinkedPage, sha256Hex } from './fetchContent' import { llmExtractProtocol } from './llmExtract' import { ParseRowStatus, ParseStageResult, ParsedProtocol } from './types' import { validateParsedProtocol } from './validate' @@ -25,17 +25,22 @@ interface BlobJob { 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 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 + `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 }, ) @@ -70,6 +75,24 @@ async function insertParse( ) } +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, @@ -114,6 +137,39 @@ async function alreadyParsed(qx: QueryExecutor, oid: string): Promise { 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: [], + }) + result.linkedPages++ + } + return { anyFailure } +} + export async function runParseStage( qx: QueryExecutor, deps: ParseStageDeps, @@ -130,6 +186,12 @@ export async function runParseStage( failed: 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) { @@ -145,7 +207,26 @@ export async function runParseStage( result.failed++ continue } - const fileParse = await parseText(text, deps, cfg) + 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', @@ -156,25 +237,6 @@ export async function runParseStage( result[fileParse.parser === 'deterministic' ? 'deterministic' : 'llm']++ if (fileParse.status === 'degraded') result.degraded++ if (fileParse.status === 'template') result.template++ - - if (fileParse.parsed.methods.length === 0 && fileParse.linkedUrls.length > 0) { - for (const url of fileParse.linkedUrls) { - const page = await deps.fetchPage(url, cfg.fetchTimeoutMs) - if (!page) continue - if (await alreadyParsed(qx, page.hash)) continue - const pageParse = await parseText(page.text, deps, cfg) - await insertParse(qx, { - blobOid: page.hash, - sourceKind: 'linked-page', - url, - parser: pageParse.parser, - status: pageParse.status, - parsed: pageParse.parsed, - linkedUrls: [], - }) - result.linkedPages++ - } - } } catch (err) { result.failed++ log.warn({ blobOid: job.blob_oid, errMsg: (err as Error).message }, 'Blob parse failed') diff --git a/services/apps/packages_worker/src/security-contacts/protocol/validate.ts b/services/apps/packages_worker/src/security-contacts/protocol/validate.ts index e93dee33ae..5da91dd2c0 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/validate.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/validate.ts @@ -9,7 +9,10 @@ const METHOD_TYPES = new Set([ 'mailing-list', ]) const METHOD_STATUSES = new Set(['preferred', 'accepted', 'fallback', 'prohibited']) -const SENTINEL_ENDPOINTS = new Set(['github-pvr', 'security.txt']) +const SENTINEL_ENDPOINT_BY_TYPE: Record = { + 'github-pvr': 'github-pvr', + 'security-txt': 'security.txt', +} function normalize(text: string): string { return text @@ -19,20 +22,53 @@ function normalize(text: string): string { .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 (SENTINEL_ENDPOINTS.has(m.endpoint)) continue + 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}`) diff --git a/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql b/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql index b67a5e6622..65963905d7 100644 --- a/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql +++ b/services/apps/packages_worker/src/tmp/validate-reporting-protocol.sql @@ -27,16 +27,18 @@ FROM repo_reporting_protocols rp JOIN critical c ON c.id = rp.repo_id; -- 3. Method sanity --- Healthy: zero rows. +-- 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' NOT IN ('github-pvr','email','web-form','bounty-platform','security-txt','mailing-list') +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' NOT IN ('preferred','accepted','fallback','prohibited') +WHERE m->>'status' IS NULL + OR m->>'status' NOT IN ('preferred','accepted','fallback','prohibited') UNION ALL SELECT 'multiple_preferred', COUNT(*) FROM ( From be30d281fcb85fd9518e7798a0e74993bbae1eb3 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 14:26:21 +0100 Subject: [PATCH 11/13] fix(review): address review feedback Signed-off-by: Mouad BANI --- .../security-contacts/protocol/assemble.ts | 21 ++++++++- .../protocol/fetchContent.ts | 43 +++++++++++++++++-- .../security-contacts/protocol/validate.ts | 2 +- 3 files changed, 60 insertions(+), 6 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts b/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts index 3253a92215..0deb9bffff 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/assemble.ts @@ -7,6 +7,7 @@ import { ProtocolMethod, ProtocolMethodType, } from './types' +import { SENTINEL_ENDPOINT_BY_TYPE } from './validate' export interface AssembleInput { repoUrl: string @@ -58,13 +59,27 @@ export function assembleProtocol(input: AssembleInput): AssembledProtocol { 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: m.type === 'github-pvr' ? pvrAdvisoryUrl(input.repoUrl) : m.endpoint, + endpoint, confidence: 'declared', provenance: { path: fp.path, blobOid: fp.blobOid, parser: fp.parser }, }) @@ -74,9 +89,11 @@ export function assembleProtocol(input: AssembleInput): AssembledProtocol { 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: m.type === 'github-pvr' ? pvrAdvisoryUrl(input.repoUrl) : m.endpoint, + endpoint, confidence: 'declared', provenance: { url: pp.url, blobOid: pp.hash, parser: pp.parser }, }) diff --git a/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts b/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts index de3e1414b0..bb8a603506 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/fetchContent.ts @@ -1,4 +1,5 @@ import { createHash } from 'node:crypto' +import { lookup } from 'node:dns/promises' import { parseGithubUrl } from '../../enricher/fetchLightRepo' import type { githubApiGet } from '../githubToken' @@ -35,8 +36,8 @@ export async function fetchBlob( export function htmlToText(html: string): string { return html - .replace(//gi, ' ') - .replace(//gi, ' ') + .replace(/]*>/gi, ' ') + .replace(/]*>/gi, ' ') .replace(/<[^>]+>/g, ' ') .replace(/</g, '<') .replace(/>/g, '>') @@ -62,6 +63,16 @@ function isPrivateIpv4(host: string): boolean { ) } +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 { @@ -77,10 +88,36 @@ export function isBlockedUrl(raw: string): boolean { 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 (isBlockedUrl(current)) return null + if (await isRequestBlocked(current)) return null const res = await fetch(current, { signal, headers: { 'User-Agent': 'crowd.dev-reporting-protocol' }, diff --git a/services/apps/packages_worker/src/security-contacts/protocol/validate.ts b/services/apps/packages_worker/src/security-contacts/protocol/validate.ts index 5da91dd2c0..54781c34ba 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/validate.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/validate.ts @@ -9,7 +9,7 @@ const METHOD_TYPES = new Set([ 'mailing-list', ]) const METHOD_STATUSES = new Set(['preferred', 'accepted', 'fallback', 'prohibited']) -const SENTINEL_ENDPOINT_BY_TYPE: Record = { +export const SENTINEL_ENDPOINT_BY_TYPE: Record = { 'github-pvr': 'github-pvr', 'security-txt': 'security.txt', } From bf09cea578d741c96c7901eb717a54d88b66ce30 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 15:35:09 +0100 Subject: [PATCH 12/13] fix: skip LLM for pointer-only security files in reporting protocol Signed-off-by: Mouad BANI --- .../src/security-contacts/protocol/classify.ts | 7 +++++++ .../src/security-contacts/protocol/parseStage.ts | 8 ++++++++ .../src/security-contacts/protocol/types.ts | 1 + 3 files changed, 16 insertions(+) diff --git a/services/apps/packages_worker/src/security-contacts/protocol/classify.ts b/services/apps/packages_worker/src/security-contacts/protocol/classify.ts index 738c641182..028d072a20 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/classify.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/classify.ts @@ -25,6 +25,8 @@ const DEFAULT_TEMPLATE_RE = 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 @@ -108,9 +110,14 @@ export function classifySecurityPolicy(text: string): ClassifierVerdict { 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/parseStage.ts b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts index fad13ca568..07dbd75122 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts @@ -112,6 +112,14 @@ async function parseText( linkedUrls: verdict.linkedUrls, } } + if (verdict.pointerOnly) { + return { + parser: 'deterministic', + status: 'ok', + parsed: { methods: verdict.methods, guidelines: null }, + linkedUrls: verdict.linkedUrls, + } + } const llmParsed = await deps.llmExtract(text, { modelId: cfg.llmModelId, timeoutMs: cfg.llmTimeoutMs, diff --git a/services/apps/packages_worker/src/security-contacts/protocol/types.ts b/services/apps/packages_worker/src/security-contacts/protocol/types.ts index b20a92421d..3d256bba8d 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/types.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/types.ts @@ -33,6 +33,7 @@ export type ParseRowStatus = 'ok' | 'template' | 'degraded' export interface ClassifierVerdict { clean: boolean isTemplate: boolean + pointerOnly: boolean methods: ParsedMethod[] linkedUrls: string[] } From 75df7807d79079220dd8b0304603a25f74fee3fd Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 29 Jul 2026 16:36:14 +0100 Subject: [PATCH 13/13] feat: persist per-file LLM cost for reporting protocol parses Signed-off-by: Mouad BANI --- ...5369600__reporting_protocol_llm_tokens.sql | 5 +++ .../security-contacts/protocol/llmExtract.ts | 34 ++++++++++++++++--- .../security-contacts/protocol/parseStage.ts | 24 ++++++++++--- .../src/security-contacts/protocol/types.ts | 1 + 4 files changed, 55 insertions(+), 9 deletions(-) create mode 100644 backend/src/osspckgs/migrations/V1785369600__reporting_protocol_llm_tokens.sql 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/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts b/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts index 35d0168ac1..f36a875214 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/llmExtract.ts @@ -12,6 +12,10 @@ 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, @@ -78,6 +82,11 @@ export interface LlmExtractConfig { secretAccessKey: string | undefined } +export interface LlmExtractResult { + parsed: ParsedProtocol | null + costUsd: number | null +} + const clients = new Map() function clientFor(cfg: LlmExtractConfig, region: string): BedrockRuntimeClient { @@ -96,13 +105,19 @@ function clientFor(cfg: LlmExtractConfig, region: string): BedrockRuntimeClient 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 { +): Promise { if (!cfg.accessKeyId || !cfg.secretAccessKey) { log.warn({ modelId: cfg.modelId }, 'Missing Bedrock credentials — skipping LLM extraction') - return null + return { parsed: null, costUsd: null } } const region = LLM_MODEL_REGION_MAP[cfg.modelId as LlmModelType] ?? DEFAULT_REGION @@ -124,15 +139,24 @@ export async function llmExtractProtocol( }) 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 null - return parseLlmJson(answer) + 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 null + 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 index 07dbd75122..3ec475391b 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/parseStage.ts @@ -56,12 +56,13 @@ async function insertParse( 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) - VALUES ($(blobOid), $(sourceKind), $(url), $(parser), $(parserVersion), $(status), $(parsed), $(linkedUrls)) + (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, @@ -70,6 +71,7 @@ async function insertParse( 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 }, ) @@ -102,6 +104,7 @@ async function parseText( status: ParseRowStatus parsed: ParsedProtocol linkedUrls: string[] + costUsd: number | null }> { const verdict = classifySecurityPolicy(text) if (verdict.clean) { @@ -110,6 +113,7 @@ async function parseText( status: verdict.isTemplate ? 'template' : 'ok', parsed: { methods: verdict.methods, guidelines: null }, linkedUrls: verdict.linkedUrls, + costUsd: null, } } if (verdict.pointerOnly) { @@ -118,22 +122,30 @@ async function parseText( status: 'ok', parsed: { methods: verdict.methods, guidelines: null }, linkedUrls: verdict.linkedUrls, + costUsd: null, } } - const llmParsed = await deps.llmExtract(text, { + 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 } + 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, } } @@ -172,8 +184,10 @@ async function processLinkedPages( status: pageParse.status, parsed: pageParse.parsed, linkedUrls: [], + costUsd: pageParse.costUsd, }) result.linkedPages++ + if (pageParse.costUsd !== null) result.llmCostUsd += pageParse.costUsd } return { anyFailure } } @@ -192,6 +206,7 @@ export async function runParseStage( template: 0, linkedPages: 0, failed: 0, + llmCostUsd: 0, } const limitLlm = createLimiter(Math.max(1, cfg.llmConcurrency)) @@ -245,6 +260,7 @@ export async function runParseStage( 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') diff --git a/services/apps/packages_worker/src/security-contacts/protocol/types.ts b/services/apps/packages_worker/src/security-contacts/protocol/types.ts index 3d256bba8d..971bdcf364 100644 --- a/services/apps/packages_worker/src/security-contacts/protocol/types.ts +++ b/services/apps/packages_worker/src/security-contacts/protocol/types.ts @@ -65,6 +65,7 @@ export interface ParseStageResult { template: number linkedPages: number failed: number + llmCostUsd: number } export interface AssembleStageResult {