Findings.VulnCorrelation — analytics ingestion & vulnerability correlation

Architecture dossier for src/Findings/StellaOps.Findings.VulnCorrelation.WebService — the worker host that owns the analytics.* PostgreSQL schema and derives it from the Scanner, Attestor, and Concelier event streams.

Part of the consolidated Findings module. See architecture.mdfor the module map (Ledger · Security · VulnCorrelation · RiskEngine · merged VulnExplorer).

1. Overview

Findings.VulnCorrelation is a worker host: it has no gateway route, no read-model endpoints, and no identity envelope — only worker health probes and /buildinfo.json (WebService/Program.cs:24-25,67-71). Everything it does happens on three background stream consumers, and everything it produces lands in the analytics.* schema of the shared platform database.

It was carved out of platform-web by SPRINT_20260703_001 PAC-5 (platform-agnostic cleanup). The ingestion/correlation BackgroundServices and the analytics.* schema migrations were relocated verbatim — including the stream/checkpoint/CAS resolution and the Platform:AnalyticsIngestion configuration section names — so the worker resumes from exactly the checkpoint platform-web left behind (WebService/Program.cs:1-25, ServiceCollectionExtensions.cs:1-9).

Binding law. The worker references only the domain-agnostic StellaOps.Findings.Disposition.Contracts plus SBOM/messaging/CAS libraries — no VexLens, Signals, or vuln-derivation domain types (Program.cs:21-22, verified against StellaOps.Findings.VulnCorrelation.WebService.csproj:14-32: Npgsql, NuGet.Versioning, Concelier.SbomIntegration, Scanner.Surface.FS, Router/StellaOps.Messaging, Findings.Disposition.Contracts, Worker.Health, Hosting).

2. Design principles

1. Single writer of analytics.*. The worker owns the DDL and every ingestion/derivation write. No other service inserts into these tables.
2. Worker-only surface. No HTTP API means no gateway registration (Router__Enabled: "false" in the compose overlay) and no envelope authentication to get wrong.
3. Resumable, at-least-once ingestion. The scanner lane persists a file checkpoint after every handled event; all writes are upserts/dedupes so a replayed event is a no-op.
4. Idempotent, embedded, forward-only schema (AGENTS.md §2.7) — auto-applied at startup.
5. Domain-agnostic. Vulnerability facts are read from Concelier’s vuln.* schema; the worker adds no vuln-derivation logic of its own beyond version-range matching.

3. Components

src/Findings/
├── StellaOps.Findings.VulnCorrelation.WebService/
│   ├── Program.cs                       # worker host: migrations + messaging + 3 hosted services
│   ├── ServiceCollectionExtensions.cs   # AddAnalyticsIngestion(...)
│   ├── Options/AnalyticsIngestionOptions.cs   # section "Platform:AnalyticsIngestion"
│   ├── Models/                          # ScannerJobEngineEvents, RekorEvents, AdvisoryEvents
│   ├── Services/
│   │   ├── AnalyticsIngestionService.cs        # scanner/SBOM lane  (BackgroundService)
│   │   ├── AttestationIngestionService.cs      # attestor lane      (BackgroundService)
│   │   ├── VulnerabilityCorrelationService.cs  # Concelier lane     (BackgroundService)
│   │   ├── AnalyticsIngestionDataSource.cs     # Npgsql data source
│   │   ├── CasContentReader.cs                 # FileCasContentReader (root-jailed)
│   │   └── FindingDispositionCache.cs          # in-memory disposition cache lane
│   └── Utilities/                       # PurlParser, VersionRuleEvaluator, TenantNormalizer,
│                                        # VulnerabilityCorrelationRules, LicenseExpressionRenderer, Sha256Hasher
└── __Libraries/StellaOps.Findings.VulnCorrelation.Persistence/
    ├── Extensions/VulnCorrelationMigrationExtensions.cs   # AddVulnCorrelationMigrations
    └── Migrations/001_v1_analytics_baseline.sql           # embedded; the analytics.* schema

Tests: src/Findings/__Tests/StellaOps.Findings.VulnCorrelation.Tests/ (ingestion fixture, edge-case, real-dataset, schema-integration, attestation-payload parsing, PURL/version-rule, tenant-normalizer, disposition-lane).

4. Data flow

orchestrator:events ────────► AnalyticsIngestionService ──► CAS (SBOM) ──► analytics.artifacts
  (scanner.event.report.ready │                                              analytics.raw_sboms
   scanner.scan.completed)    │                                              analytics.components
                              │                                              analytics.artifact_components
                              └─ finding.disposition.changed ──► in-memory FindingDispositionCache
                                                                 (not persisted)

attestor:events ────────────► AttestationIngestionService ─► CAS (DSSE bundle) ─► analytics.attestations
  (rekor.entry.logged                                                             analytics.raw_attestations
   rekor.inclusion.verified)                                                      analytics.vex_overrides
                                                                                  analytics.artifacts (provenance/SLSA)

concelier:advisory.observation.updated:v1 ┐
concelier:advisory.linkset.updated:v1     ┴► VulnerabilityCorrelationService ─► analytics.component_vulns
                                             (reads Concelier vuln.* in the same DB)  analytics.artifacts (counts)

4.1 Scanner / SBOM lane (AnalyticsIngestionService)

• Subscribes to Streams:ScannerStream (default orchestrator:events) as IEventStream<JobEngineEventEnvelope>; handles kinds scanner.event.report.ready and scanner.scan.completed (Models/ScannerJobEngineEvents.cs:160-164, AnalyticsIngestionService.cs:165-169).
• Resolves the SBOM from CAS (FileCasContentReader, root-jailed to Cas:RootPath — a URI resolving outside the root is refused, CasContentReader.cs:52-56), parses it with IParsedSbomParser (Concelier SBOM integration), then in one transaction upserts analytics.artifacts (ON CONFLICT (digest)), analytics.raw_sboms (ON CONFLICT (content_hash) DO NOTHING), analytics.components (ON CONFLICT (purl, hash_sha256)), and analytics.artifact_components (ON CONFLICT (artifact_id, component_id) DO NOTHING).
• After commit it calls CorrelateForPurlsAsync(purls) and UpdateArtifactCountsAsync(artifactId) on the correlation service (AnalyticsIngestionService.cs:753-768) — so a newly ingested SBOM is correlated immediately, without waiting for a Concelier advisory event.
• Checkpoint: after every handled event the stream entry id is written to <Cas:RootPath>/.state/platform-scanner-stream.checkpoint (overridable via Streams:ScannerCheckpointFilePath, AnalyticsIngestionService.cs:505-530). On start the subscription resumes from that entry unless StartFromBeginning=true.
• Disposition cache lane: finding.disposition.changed payloads ride the same orchestrator:events stream under the generic EMIT envelope kind and are discriminated on the payload eventType marker (AnalyticsIngestionService.cs:114-121,177-194). They populate an in-memory IFindingDispositionCache only — nothing is persisted, and the enforced disposition read-model is served by Findings.Security (PAC-4), not from here (ServiceCollectionExtensions.cs:54-59). This worker-local cache is not the Security freshness authority. Findings.Security independently subscribes to the same domain-agnostic event and repairs its durable cache from the Ledger every 60 seconds; see security-read-model.md.

4.2 Attestor lane (AttestationIngestionService)

• Subscribes to Streams:AttestorStream (default attestor:events) as IEventStream<RekorEntryEvent>; handles rekor.entry.logged and rekor.inclusion.verified (Models/RekorEvents.cs:73-79, AttestationIngestionService.cs:104-106).
• Reads the DSSE bundle from CAS through Attestations:BundleUriTemplate (default bundle:{digest}), parses the in-toto statement, and writes analytics.attestations (ON CONFLICT (dsse_payload_hash)) plus analytics.raw_attestations (ON CONFLICT (content_hash) DO NOTHING).
• Provenance predicates (predicate type containing slsa/provenance) set analytics.artifacts.provenance_attested = TRUE and raise slsa_level monotonically (GREATEST, AttestationIngestionService.cs:1110-1125).
• VEX predicates (OpenVEX statements or a CycloneDX vulnerabilities array) are projected into analytics.vex_overrides with a NOT EXISTS dedupe guard on (attestation, vuln, artifact, component purl) (:1131-1170, :692-716).

4.3 Concelier / correlation lane (VulnerabilityCorrelationService)

• Subscribes to Streams:ConcelierObservationStream (concelier:advisory.observation.updated:v1) and Streams:ConcelierLinksetStream (concelier:advisory.linkset.updated:v1).
• For an advisory event it resolves the affected PURLs from Concelier’s schema (SELECT DISTINCT aff.package_purl FROM vuln.advisory_affected aff JOIN vuln.advisories adv …) and correlates them.
• CorrelateForPurlsAsync loads the matching analytics.components, then joins Concelier’s vuln.advisory_affected / vuln.advisories / vuln.sources / vuln.advisory_cvss / vuln.kev_flags / vuln.advisory_canonical (state = 'active', best CVSS by is_primary DESC, base_score DESC) and upserts analytics.component_vulns (ON CONFLICT (component_id, vuln_id)). affects is decided by VersionRuleEvaluator against the advisory’s normalized_versions; fixed_version / fix_available come from the same rule set (VulnerabilityCorrelationService.cs:105-178,343-400).
• UpdateArtifactCountsAsync recomputes analytics.artifacts.{vulnerability,critical,high,medium,low}_count from artifact_components ⋈ component_vulns WHERE affects = TRUE (:193-218).

Cross-schema read dependency. The correlation lane reads Concelier’s vuln.* tables directly from the same database. Concelier owns that schema; VulnCorrelation only reads it. A change to vuln.advisory_affected / vuln.advisories column names breaks correlation.

5. Ownership boundary with Platform (PAC-5) — read this before editing either side

The relocation comment in Program.cs:12-13 (“Platform keeps ONLY the read-only analytics QUERY surface — it no longer DERIVES”) is accurate for ingestion and correlation, but it is not the whole picture: the maintenance lane still runs in platform-weband writes analytics.daily_vulnerability_counts / analytics.daily_component_counts (via compute_daily_rollups) and refreshes the MVs. Consequence for operators: if platform-web’s analytics maintenance is disabled, the dashboards’ trend series and materialized views go stale even though the VulnCorrelation worker is healthy. Both services must point at the same database.

6. Database schema (analytics, owned here)

Auto-migration (AGENTS.md §2.7): Program.cs:44-52 calls AddVulnCorrelationMigrations(...), which wires AddStartupMigrations<PostgresOptions>(schemaName: "analytics", moduleName: "Findings.VulnCorrelation", assembly). The SQL is an embedded resource (Migrations\**\*.sql, with Migrations\_archived\** excluded so archived files can never collide on leaf name) and is idempotent (CREATE … IF NOT EXISTS / CREATE OR REPLACE / ADD COLUMN IF NOT EXISTS) — a no-op against the pre-PAC-5 database that already carried these tables, and a full create on a fresh one.

Materialized views: mv_supplier_concentration, mv_license_distribution, mv_vuln_exposure, mv_attestation_coverage (each with a unique index so REFRESH … CONCURRENTLY works).

Functions: sp_top_suppliers, sp_license_heatmap, sp_vuln_exposure, sp_fixable_backlog, sp_attestation_gaps, sp_mttr_by_severity, compute_daily_rollups, refresh_all_views, plus the helpers normalize_supplier, categorize_license, parse_purl, update_updated_at_column.

Reading the baseline. 001_v1_analytics_baseline.sql was assembled by concatenating the historical Platform migrations without collapsing superseded DDL: several functions (e.g. sp_vuln_exposure, compute_daily_rollups, sp_top_suppliers) are CREATE OR REPLACE-d more than once, and mv_vuln_exposure is created twice. The file is idempotent and the last definition in the file wins — do not read the first occurrence and assume it is authoritative. Later ALTER TABLE … ADD COLUMN IF NOT EXISTS blocks (e.g. artifacts.medium_count, artifacts.low_count at :2360-2362) are part of the same baseline.

7. Configuration

All keys keep the Platform:AnalyticsIngestionsection name from before the relocation (AnalyticsIngestionOptions.SectionName), so the pre-PAC-5 env keys resolve unchanged.

Messaging is registered with RequireTransport = false (Program.cs:57): with no transport plugin the worker starts, logs “no event stream configured”, and idles — it does not crash. That is the intended posture but it means a missing transport shows up as silent non-ingestion, not as a failed container.

8. Tenancy and security invariants

1. The analytics.* schema has no tenant column. Verified: grep -i tenant over 001_v1_analytics_baseline.sql returns nothing. Components, artifacts, vulns, attestations and rollups are stored un-namespaced.
2. Tenancy is enforced only at the ingestion boundary, from the event envelope’s tenant (envelope.Tenant / payload TenantId) — never from a caller-supplied header — against the AllowedTenants allowlist (TenantNormalizer.IsAllowed, which strips a urn:tenant: prefix and compares case-insensitively).
3. An empty AllowedTenants list allows every tenant (Utilities/TenantNormalizer.cs, IsAllowed: if (allowedTenants is null || allowedTenants.Count == 0) return true;). Any deployment serving more than one tenant must set it — the live overlay sets Platform__AnalyticsIngestion__AllowedTenants__0=default.
4. The single-tenant invariant is now enforced in code, fail-closed (SPRINT_20260712_009 CLO-4). AnalyticsIngestionOptions.Validate() throws on startup if AllowedTenants lists more than one tenant — because the schema cannot separate them, a 2-tenant allowlist would silently blend both tenants’ inventory/licence/vuln/attestation data into one shared read model. That Validate() is now actually invoked: ServiceCollectionExtensions registers it through the options Validate(...) delegate feeding ValidateOnStart() (previously ValidateOnStart() was wired with no validator, so the fail-fast on a missing connection string was dead too). Normalize() canonicalizes with TenantNormalizer.Normalize (strips urn:tenant:, de-duplicates case-insensitively) so multiple spellings of one tenant collapse to one entry and do not trip the guard. The empty-allowlist “accept every tenant” default is deliberately left intact (flipping it to deny-all would silently stop ingestion on any deployment that omits the key); flipping it to deny-all is staged separately.
5. Consequently analytics.* is a single-tenant read model per deployment today. Platform’s /api/analytics/* endpoints require the analytics.read scope and RequireTenant(), but the tenant only participates in the cache key (PlatformAnalyticsService.cs:55-168) — the SQL cannot filter by tenant because the column does not exist. True per-tenant isolation would require a real tenant dimension end-to-end (schema column + ingestion write + read-path filter); that is a follow-up, not a current guarantee. Recorded in SPRINT_20260712_006 and SPRINT_20260712_009 (CLO-4) Decisions & Risks.
6. The worker has no gateway route and no identity envelope — correct for a worker (Router__Enabled: "false" in the overlay). Do not add HTTP endpoints here; the read surfaces belong to Platform (analytics) and Findings.Security (/api/v2/security*).

9. Deployment & operations

• Image stellaops/findings-vulncorrelation:dev, container stellaops-findings-vulncorrelation, built from src/Findings/StellaOps.Findings.VulnCorrelation.WebService (devops/docker/services-matrix.env:62).
• Deployed by the opt-in overlay devops/compose/docker-compose.findings-vulncorrelation.override.yml — it is not a service in docker-compose.stella-services.yml. The overlay mounts the shared scanner-poe-cas-data volume at /var/lib/stellaops/poe-cas so the CAS root and the scanner checkpoint file resolve to the same paths platform-web used, and runs the router/messaging transport in mounted-plugin mode purely so IEventStreamFactory can subscribe.
• Health: GET /health/liveness, GET /health/readiness (AddWorkerHealthChecks / MapWorkerHealthEndpoints), plus GET /buildinfo.json for image-drift detection.
• Cutover safety: exactly one consumer of the scanner stream exists — the ingestion code was removed from platform-web, so there is nothing to disable at runtime.

Failure modes worth knowing

10. Determinism & replay

Ingestion is at-least-once and every write is an upsert or a dedupe-guarded insert, so replaying a stream range converges to the same rows. The scanner checkpoint is advanced after the event is handled, so a crash between handling and checkpointing replays the last event (idempotently).

11. References

• Findings module map and sibling hosts: architecture.md.
• Platform read surface + maintenance lane: ../platform/platform-service.md.
• Relocation sprint (PAC-5): SPRINT_20260703_001_Platform_full_agnostic_cleanup.md.
• Auto-migration contract: AGENTS.md §2.7.