PyPI Advisory Processor — Design Document

1. Overview

Purpose: Process the PyPA advisory database’s OSV YAML files for PyPI advisories (PYSEC-*.yaml) into CVEMetadata and related tables.

Data source: PyPA Advisory Database — https://github.com/pypa/advisory-database.git (git repository cloned to /data/advisory-database; the vulns/ tree). Not the GitHub advisory-database repo, which has advisories/ rather than vulns/ — pointing --repo at it is a known footgun and produces data directory not found.

Schedule: Daily (cron 0 6 * * *)

Timeout: 45 minutes

Resources: 256 CPU units, 512 MB memory

What it reads:

  • Git repository at /data/advisory-database/vulns/ — walks files matching PYSEC-*.yaml
  • Read replica: existing sourceFileHash values for per-file resume (CVEMetadata.sourceFileHash via LoadProcessedHashes)
  • Read replica: BulkDataDumpTracker row for pypi_advisory (HEAD SHA dedup)

What it writes:

  • CVEMetadata (upsert, source=pypi)
  • CVEDescription, CVEMetadataReferences, CVEMetric, CVEProblemType
  • CVEAffected, CVEAffectedVersion
  • Dependency, DependencyRegistry, PackageVersion (ecosystem pypi), PackageVersionCVE, GitHubRepoDependency — via db.EnrichAffectedWithDependency for every affected entry
  • CVEAlias — always, via db.InsertAliases, even when the advisory carries no alias list, so the same-cveId cross-source backfill runs on every store
  • A derived Vulnetix CVSS v4 metric (containerType="vulnetix") from the advisory descriptions
  • BulkDataDumpTracker (tracker update)
  • When AI enrichment is enabled (PIX_INFERENCE_ENABLED + gateway token, i.e. ECS only): CVEProblemType (derivedBy="vulnetix"), CVEAttackTechnique / CVEAttackMitigation / CVEAttackDetection / CVEAttackD3fend, CVETreeSitter, and one PixLog row per pass

What it deliberately does not write: malicious-package advisories. MAL-* records and anything flagged by the six-criteria osv.IsMaliciousPackage check are skipped, so the isMaliciousPackage=true rows that exist under source='pypi' come from other writers (the OSV and oss-malware processors), not from here.

Environment variables:

  • DATABASE_URL — RDS Write Proxy connection string (required)
  • DATABASE_URL_READ — RDS Read Replica connection string (optional; falls back to write)

2. Business Logic

Git Pull Strategy

Uses processor.PullOrClone which attempts git pull --ff-only on the existing clone at /data/advisory-database, cloning it if absent. In ECS the fetch depth is processor.IncrementalDepth; locally it is 1. On failure the process exits 1 with an Errored notification.

Delta Window

processor.DetectChangedFiles narrows the walk to recently changed files (skipped on the first run and under --force), so a steady-state run touches a handful of files rather than the whole vulns/ tree.

SHA-based Freshness Check

After pull, compares HEAD SHA against BulkDataDumpTracker.sha256 for source pypi_advisory. If unchanged and --force is not set, exits immediately (idempotent no-op).

Per-File Resume

Before walking files, LoadProcessedHashes(source="pypi") queries the read replica for all (cveId, sourceFileHash) pairs. This becomes the skipSet. For each file: compute SHA1, derive cveId, check skipSet[cveId] == sha1 → skip if match.

Malicious Package Detection

MapAdvisory returns nil for any advisory matching:

  • database_specific.malicious == true
  • database_specific.{type,category,severity} contains “MALICIOUS”
  • database_specific.source is ossf-malicious-packages or openssf-malicious-packages
  • References containing ossf/malicious-packages URLs
  • Credits containing “OpenSSF”, “Package Analysis”, or “malicious-packages”
  • ID prefix MAL-

CVE ID Derivation

Scans aliases[] for the first CVE-YYYY-NNNN+ pattern. If found, uses it as the cveId; otherwise uses the advisory id field (PYSEC-2024-123).

Batch Processing

Files are processed in transactions of 100 files each. Each file uses a SAVEPOINT so a single parse/store failure rolls back only that file, leaving the rest of the batch healthy.

Tracker Update

Only updates BulkDataDumpTracker if filesErrored == 0 (clean run) and the run context was not cancelled (main.go:149-166). Stores HEAD SHA in sha256.

Known defect — the intent is that a run cut short by the soft deadline also leaves the SHA un-advanced, because advancing it silently drops the unprocessed files. The guard at main.go:159 only tests ctx.Err(), and the soft-deadline break in internal/processor/pipeline.go:161-164 returns normally without cancelling the context, so the SHA is written. The next run then exits NoWork on the unchanged SHA. Tracked in .repo/efficacy/pypi-git-processor.yaml.

Exit Semantics

File errors do not by themselves fail the task: notify.Finalize decides, and the process exits 0 unless the failure ratio is systemic (main.go:140-155). A pipeline-level error or a failed tracker write exits 1.

3. Architecture Diagram

graph TD subgraph "cmd/pypi-git-processor/" MAIN[main.go] end subgraph "internal/processor/" GIT[git.go — PullOrClone] PIPE[pipeline.go — Run, processBatch, storeAdvisory] WALK[walker.go — WalkAdvisories] end subgraph "internal/osv/" PARSE[parser.go — ParseYAML] MAP[mapper.go — MapAdvisory, IsMaliciousPackage] TYPES[types.go — Advisory, CVESourceData] end subgraph "internal/db/" POOL[pool.go — Pool] TRACK[tracker.go — GetTracker, UpsertTracker] RESUME[resume.go — LoadProcessedHashes, ComputeFileSHA1] BATCH[batch.go — WithTx] CVEMETA[cvemetadata.go — UpsertCVEMetadata] CHILDREN[cvedescription, cvereference, cvemetric, cveaffected, cvealias, cveproblemtype] end MAIN --> GIT MAIN --> TRACK MAIN --> POOL MAIN --> PIPE PIPE --> WALK PIPE --> RESUME PIPE --> PARSE PIPE --> MAP PIPE --> BATCH PIPE --> CVEMETA PIPE --> CHILDREN

4. Deployment Diagram

flowchart TD HOOK[Local ECR hook
post-push-ecr.sh] -->|push ARM64 image| ECR[ECR: go-processors
tag: go-pypi-git-processor-sha-xxx] ECR --> TASKDEF[ECS Task Definition
go-pypi-git-processor] TASKDEF --> EB[EventBridge Schedule
go-pypi-git-processor
cron 0 6 * * ? *] EB -->|trigger| FARGATE[ECS Fargate Task
vdb-scheduler cluster
ARM64 ap-southeast-2] FARGATE --> CW[CloudWatch Logs
/ecs/vdb-scheduler/go-pypi-git-processor] FARGATE --> WRITE[RDS Write Proxy
cf-hyperdrive.proxy-...] FARGATE --> READ[RDS Read Replica
hyperdrive-saas-read...] FARGATE -->|baked into image
pypi-git-data stage| REPO[/data/advisory-database]

5. Processing Flow

flowchart TD START([Start]) --> ENV{DATABASE_URL set?} ENV -->|no| FAIL([Exit 1]) ENV -->|yes| CONNECT[Connect to DB pool] CONNECT --> PULL[PullOrClone /data/advisory-database] PULL -->|error| FAIL PULL -->|ok — headSHA| TRACKER{Tracker sha256
== headSHA?} TRACKER -->|yes, no --force| EXIT0([Exit 0 — unchanged]) TRACKER -->|no or --force| LOAD[LoadProcessedHashes source=pypi] LOAD --> WALK[WalkAdvisories vulns/ PYSEC-*.yaml] WALK --> BATCH[For each batch of 100 files] BATCH --> TX[Begin transaction] TX --> FILE[For each file] FILE --> SP[SAVEPOINT sp_file] SP --> READ2[ReadFile + SHA1] READ2 --> PARSE[ParseYAML] PARSE -->|error| ROLLBACK[ROLLBACK TO SAVEPOINT
count error] PARSE -->|ok| CVEID[Derive cveId from aliases] CVEID --> SKIP{skipSet match?} SKIP -->|yes, no force| RELEASE[RELEASE SAVEPOINT
count skipped] SKIP -->|no or force| MALICIOUS{IsMaliciousPackage?} MALICIOUS -->|yes| RELEASE MALICIOUS -->|no| STORE[storeAdvisory
UpsertCVEMetadata + children] STORE -->|error| ROLLBACK STORE -->|ok| RELEASE ROLLBACK --> FILE RELEASE --> FILE FILE -->|done| COMMIT[Commit transaction] COMMIT --> BATCH BATCH -->|done| ERRORS{filesErrored > 0?} ERRORS -->|yes| FAIL2([Exit 1 — skip tracker]) ERRORS -->|no| UPSERT_TRACKER[UpsertTracker pypi_advisory
sha256=headSHA] UPSERT_TRACKER --> DONE([Exit 0])

6. Data Mapping

erDiagram BulkDataDumpTracker { string source PK "pypi_advisory" bigint lastProcessedAt int frequency string sha256 "git HEAD SHA" int totalCVEs } CVEMetadata { string cveId PK "CVE-YYYY-NNNN or PYSEC-..." string source PK "pypi" string dataVersion string state int datePublished "Unix seconds" int dateUpdated string vectorString string title string sourceAdvisoryRef "PYSEC advisory ref" string affectedVendor string affectedProduct boolean isMaliciousPackage string sourceFileHash "SHA1 of file bytes" bigint lastFetchedAt } CVEDescription { string cveId FK string source FK string lang string value } CVEMetadataReferences { string uuid PK string cveId FK string source FK string url string type string referenceSource "PyPI" } CVEMetric { string cveId FK string source FK string metricType string vectorString } CVEAffected { string uuid PK string cveId FK string source FK string packageName string collectionURL string affectedHash "MD5 dedup key" } CVEAffectedVersion { string uuid PK string affectedUuid FK string version string status string lessThan } CVEAlias { string primaryCveId FK string aliasCveId string aliasSource } CVEMetadata ||--o{ CVEDescription : has CVEMetadata ||--o{ CVEMetadataReferences : has CVEMetadata ||--o{ CVEMetric : has CVEMetadata ||--o{ CVEAffected : has CVEAffected ||--o{ CVEAffectedVersion : has CVEMetadata ||--o{ CVEAlias : aliases

S3 Persistence

  • Archive path: pypi/files/{sha256}/{filename} ✓
  • Quarantine path: failed-feeds/pypi-git-processor/{YYYY-MM-DD}/{reason}/{filename} ✓
  • Failure reasons emitted: parse-error, store-error

Uses s3client.Uploader from internal/s3client/uploader.go. Skipped when S3_BUCKET_NAME is unset (local dev).

flowchart LR SRC[Source feed] --> PROC[pypi-git-processor] PROC -->|success| ARCHIVE[("S3: pypi/files/{sha256}/{filename}")] PROC -->|failure| Q[("S3: failed-feeds/pypi-git-processor/{date}/{reason}/{filename}")] PROC --> DB[(PostgreSQL)]

See the S3 Persistence Contract for the full reason taxonomy.