POC: Add malware advisories sync from private GCP bucket

Summary

POC for syncing malware advisories into a GitLab instance via the PMDB distribution service (PDS), following the v3 export format described in the SSCS add-on design doc.

The instance authenticates to PDS with a malware_advisories-scoped IJWT (audience gitlab-pmdb-distribution-service); PDS validates the token, consults the per-PURL manifest, and returns short-lived signed GCP URLs for the delta archives the instance is missing. Each archive is a .tar.zst containing NDJSON chunks which are parsed, upserted, and event-published for downstream scanning. Per-PURL checkpoints make sync resumable independently for each registry.

This MR delivers the sync/ingest plumbing only. The malware advisory + affected-package tables and downstream scanning live in the companion MR !226519 (closed).

High-level architecture

Component diagram showing how cron, the per-PURL sync service, PDS, the IJWT path, and ingestion fit together.

flowchart LR
    subgraph CDot["CustomersDot (SM/Dedicated)"]
        SAT[("service_access_tokens<br/>(IJWT, daily sync)")]
    end

    subgraph GitLab["GitLab Instance"]
        Worker["MalwareAdvisoriesSyncWorker<br/>(every 5 minutes)"]
        AddOn{"SSCS add-on<br/>available?"}
        Service["MalwareAdvisorySyncService"]
        Configs["MalwareAdvisorySyncConfiguration<br/>(one config per permitted PURL)"]
        Pds["Connector::Pds<br/>(per PURL)"]
        Tokens["CloudConnector::Tokens.get<br/>SaaS self-sign<br/>SM/Dedicated load from DB"]
        Reader["Archive::TarZstReader"]
        Fabricator["MalwareAdvisoryDataObjectFabricator"]
        Ingest["MalwareAdvisoryIngestionService"]
        Checkpoint[("pm_checkpoints<br/>data_type: malware_advisories")]
        Event["IngestedMalwareAdvisoryEvent<br/>(per advisory)"]
    end

    subgraph PDS["PMDB Distribution Service (Runway)"]
        PdsAPI["delta and all endpoints"]
        ManifestCache[("Redis<br/>manifest.json cache")]
    end

    Bucket[("Private GCP bucket<br/>v3 full_dataset.tar.zst<br/>v3 deltas tar.zst")]

    Worker --> AddOn
    AddOn -- yes --> Service
    Service --> Configs
    Service --> Pds
    Pds --> Tokens
    SAT -. read .-> Tokens
    Pds -- "Bearer IJWT<br/>+ X-Gitlab headers" --> PdsAPI
    PdsAPI --> ManifestCache
    PdsAPI -- "signed URL list" --> Pds
    Pds -- "GET signed URL" --> Bucket
    Bucket -- "tar.zst bytes" --> Reader
    Reader -- "NDJSON chunks" --> Fabricator
    Fabricator -- "MalwareAdvisoryDataObject" --> Ingest
    Ingest -- "upsert advisories and packages" --> Event
    Service -- "advance per-PURL sequence" --> Checkpoint

Sync flow (per worker run)

Sequence diagram of one cron tick: per-PURL config iteration, /all vs /delta dispatch, per-archive download and ingest, checkpoint advance.

sequenceDiagram
    participant Cron
    participant Worker as MalwareAdvisoriesSyncWorker
    participant Service as MalwareAdvisorySyncService
    participant Cfg as SyncConfiguration
    participant Pds as Pds connector
    participant Reader as TarZstReader
    participant DB as PostgreSQL
    participant PDS as PMDB Distribution Service

    Cron->>Worker: every 5 minutes
    Worker->>Worker: license and SSCS add-on check
    Worker->>Service: execute with lease
    Service->>Cfg: configs_for malware_advisories
    Cfg-->>Service: one config per permitted PURL

    loop For each PURL config
        Service->>DB: find_or_initialize checkpoint by data_type version_format purl_type
        DB-->>Service: checkpoint
        Service->>Pds: data_after checkpoint
        alt checkpoint blank
            Pds->>PDS: GET all with purl_type query
            PDS-->>Pds: one entry delta full_dataset and signed_url
        else checkpoint present
            Pds->>PDS: GET delta with purl_type and timestamp query
            PDS-->>Pds: array of entries each with delta timestamp and signed_url, or 204
        end

        Pds->>Pds: sort entries by sequence ascending
        loop For each archive
            Pds->>Pds: GET signed URL no auth headers
            Pds->>Reader: TarZstReader new bytes
            Reader-->>Pds: lazy stream of NDJSON chunks
            loop For each chunk
                Pds-->>Service: data_file with io sequence chunk
                Service->>Service: MalwareAdvisoryIngestionService execute slice
                Service->>DB: upsert pm_malware_advisories and pm_malware_affected_packages
                Service->>DB: publish IngestedMalwareAdvisoryEvent
                Service->>DB: checkpoint update sequence chunk
            end
        end
    end

sequence is the archive timestamp (e.g. 1770301601 from 1770301601.tar.zst). Once a PURL's checkpoint is at sequence N, the next /delta call carries timestamp=N and PDS only returns archives newer than that.

First-time sync (/all)

Step-by-step walkthrough of the very first sync for a PURL: when /all is chosen, how the full_dataset response is consumed, and how subsequent ticks cut over to /delta.

When a PURL has never been synced its pm_checkpoints row (data_type malware_advisories) either doesn't exist yet (find_or_initialize_by returns an unsaved record) or has sequence = 0. Checkpoint#first_sync? covers both cases, and the connector uses it to pick the endpoint:

1. Worker tick — the cron worker enters MalwareAdvisorySyncService.execute. For each permitted PURL it loads the (possibly unsaved) checkpoint.
2. first_sync? is true — Connector::Pds#endpoint_path returns 'all' and request_query omits the timestamp parameter, so the call goes out as GET {pds}/all?purl_type=<purl>.
3. PDS response — a single JSON object {"delta": "full_dataset", "signed_url": "..."} pointing at the PURL's complete snapshot (v3/<purl>/full_dataset.tar.zst).
4. Sequence resolution — the /all response carries delta: "full_dataset". The connector reads the snapshot cutoff from the archive's bundled checkpoint.json ({"until": <epoch>}) via TarZstReader#checkpoint_until and uses that until epoch as the chunk sequence (falling back to Time.now.to_i only if checkpoint.json is absent).
5. Archive consumption — the snapshot is fetched from the signed URL (no auth headers), passed to Archive::TarZstReader, and each NDJSON chunk inside it is upserted through MalwareAdvisoryIngestionService.
6. Checkpoint persisted (atomic per archive) — the checkpoint advances per archive, not per chunk: since all 256 chunks of the snapshot share sequence = until, the row is committed only once the whole snapshot is ingested (sequence = until, chunk = last). A run interrupted mid-snapshot leaves the checkpoint untouched (first_sync? stays true).
7. Next tick — after a clean run first_sync? is false, so subsequent runs go through /delta?purl_type=<purl>&timestamp=<until> and only fetch archives published after the snapshot cutoff. After an interrupted run first_sync? is still true, so the next run re-runs /all and completes the snapshot — no partial bootstrap is mistaken for complete. Delta sequences advance monotonically because the connector sorts the response array by sequence ascending before iterating.

Failure modes worth knowing:

• An interrupted full sync does not advance the checkpoint (it is committed only after every chunk lands), so the next run re-runs /all and finishes the snapshot. Upserts are idempotent, so re-ingesting the already-applied chunks is harmless. The same per-archive guarantee covers a multi-chunk delta archive.
• If a PURL is later unchecked in the admin UI and re-checked, its checkpoint row persists across the gap, so first_sync? is false and the resync uses /delta from the last-known sequence (avoiding a redundant full snapshot).

IJWT handling

How CloudConnector::Tokens.get auto-dispatches SaaS self-signing vs Self-Managed CustomersDot tokens, plus the headers sent to PDS.

CloudConnector::Tokens.get(unit_primitive: :malware_advisories, resource: :instance) dispatches the two deployment types automatically:

Path	What happens	Where
GitLab.com	TokenIssuer self-signs a JWT per request with active_add_ons from GitlabSubscriptions::AddOnPurchase, 1-hour TTL	ee/lib/cloud_connector/tokens.rb
Self-Managed / Dedicated	TokenLoader returns CloudConnector::ServiceAccessToken.active.last&.token — the daily-synced CustomersDot IJWT stored encrypted in service_access_tokens	ee/lib/cloud_connector/tokens/token_loader.rb

See gitlab-com/content-sites/handbook!18229 (comment 3159494508) for the full handbook explanation; the same link is embedded as a code comment at Pds#ijwt_token.

Both paths feed the only header PDS's auth middleware reads:

• Authorization: Bearer {IJWT}

PDS validates the JWT signature against the OIDC JWKS, checks that gitlab-pmdb-distribution-service is in aud, and that malware_advisories is in scopes (per Ahmad's design note). No other GitLab-specific headers are required.

The signed URLs PDS hands back are self-authenticating, so the subsequent archive download to GCS sends no headers either.

Archive format (.tar.zst)

Layout of a delta archive and how TarZstReader yields each NDJSON chunk for the ingestion loop.

Each signed URL points at a .tar.zst archive:

1770301601.tar.zst
  |- 00000001.ndjson
  |- 00000002.ndjson
  `- 00000003.ndjson

Connector::Archive::TarZstReader calls Zstd.decompress from the zstd-ruby gem (added in this MR -- see "External dependencies" below) and then uses Gem::Package::TarReader (stdlib) to yield each .ndjson chunk as a StringIO. Each chunk's filename (00000001.ndjson → chunk 1) gates checkpoint progress, so a partial sync mid-archive resumes from the last completed chunk. The gem ships libzstd as vendored C source, so the runtime has no system-level dependency on a zstd binary.

Admin controls

Dedicated PURL-type admin setting and the air-gapped (vendor-path) flow.

Dedicated PURL-type selection

A new admin setting package_metadata_malware_advisories_purl_types (smallint[] on application_settings) drives the per-PURL sync fan-out. It is independent of the existing package_metadata_purl_types setting (which still gates public advisory + license syncs), so admins can opt particular registries into malware sync without affecting any other sync flow. The Admin → Security & Compliance page renders a separate Malware Advisories block with its own checkbox grid for this control.

Concretely, on each worker run:

1. MalwareAdvisorySyncConfiguration.configs_for("malware_advisories") reads the selected PURL list via application_settings.package_metadata_malware_advisories_purl_types_names.
2. For each selected PURL it emits one MalwareAdvisorySyncConfiguration instance (purl_type: set).
3. MalwareAdvisorySyncService.execute iterates those configs; for each one it looks up the matching pm_checkpoints row (data_type malware_advisories) by (data_type, version_format, purl_type), instantiates a Connector::Pds scoped to that PURL, and runs the /delta (or /all) + ingest + checkpoint-advance loop in isolation.

Unchecking a PURL in the admin UI stops new syncs for that registry without touching its checkpoint row, so re-enabling it later resumes from the same sequence. Checking a new PURL the next worker tick creates its checkpoint row on first ingest and pulls /all because the row is blank.

POC choice -- subject to final production direction: introducing a dedicated package_metadata_malware_advisories_purl_types setting lets us toggle malware sync independently of the existing public PM advisories sync. If product decides malware sync should always cover the same PURL set as the public flow, the dedicated column, admin block, helper, and migration can all be dropped and MalwareAdvisorySyncConfiguration.permitted_purl_types re-pointed at package_metadata_purl_types_names. See discussion at !226660 (closed) (note 3382791662).

Air-gapped sync

Truly air-gapped instances cannot reach PDS (or CustomersDot) at all, so there is no live PDS call and no IJWT in play. The flow mirrors the existing offline path used by public PM advisories:

1. The customer runs an out-of-band download process on a network-connected host to fetch the malware advisory .tar.zst archives. That tooling is not in this MR -- it is being designed in #594758.
2. The downloaded archives are hand-carried (offline media, restricted-network bridge, etc.) onto the air-gapped instance and extracted into vendor/package_metadata/malware_advisories/v3/<purl_type>/<sequence>/<chunk>.ndjson.
3. MalwareAdvisorySyncConfiguration::Location.for_malware_advisories detects the vendor/... directory, returns :offline as the storage type, and MalwareAdvisorySyncService routes to Connector::Offline instead of Connector::Pds. The same fabricator + ingestion + per-PURL checkpoint loop applies; no network egress and no IJWT are required at runtime.

This MR provides the offline-storage code path (Connector::Offline is already wired into MalwareAdvisorySyncService) but explicitly does not provide the download tooling that produces the archives -- see #594758 for that.

Per-PURL checkpointing

How malware sync reuses the shared pm_checkpoints table, and its failure-isolation properties.

Malware advisories reuse the existing pm_checkpoints table rather than a dedicated one, distinguished by data_type = malware_advisories on the existing (purl_type, data_type, version_format) unique key — one row per PURL the instance syncs. A production DB-Lab clone showed ~35 rows total since 2023, so there is no scale/contention concern that would justify a second table. Each row stores:

• sequence (bigint) — latest archive timestamp ingested (the delta timestamp, or a full snapshot's until cutoff)
• chunk (smallint) — last NDJSON chunk consumed inside that archive

One PURL falling behind (e.g. PDS returns 500 for maven while npm succeeds) doesn't gate the others.

IngestedMalwareAdvisoryEvent -- bridge to scanning

Why ingestion publishes an event instead of triggering scans inline, and which subscriber in !226519 (closed) consumes it.

PackageMetadata::IngestedMalwareAdvisoryEvent is a Gitlab::EventStore::Event with a single-field payload { advisory_id: <pm_malware_advisories row id> }. MalwareAdvisoryIngestionService#publish! fires one event per advisory that was just upserted and was published within the last 14 days (PUBLISHED_ADVISORY_INTERVAL = 14.days). Older advisories are skipped on the assumption that any matching SBOM occurrence would already have been flagged when the advisory was first scanned.

Intended consumer (this MR's sibling): !226519 (closed) will register PackageMetadata::GlobalMalwareAdvisoryScanWorker as a subscriber via Gitlab::EventStore::Subscriptions::PackageMetadataSubscriptions, mirroring the existing wiring for the public PM advisories pipeline (ee/lib/gitlab/event_store/subscriptions/package_metadata_subscriptions.rb):

# Already in master (public PM advisories)
store.subscribe ::PackageMetadata::GlobalAdvisoryScanWorker,
                to: ::PackageMetadata::IngestedAdvisoryEvent

# What !226519 adds (malware)
store.subscribe ::PackageMetadata::GlobalMalwareAdvisoryScanWorker,
                to: ::PackageMetadata::IngestedMalwareAdvisoryEvent

The worker dequeues the event, loads the pm_malware_advisories row by advisory_id, walks the sbom_occurrences corpus via PossiblyAffectedOccurrencesFinder to find any installed package version that falls inside the advisory's affected_range, and creates a malware-type Vulnerabilities::Finding + Vulnerability per match through CreateVulnerabilityService.

Why we need this event. Customers expect Continuous Vulnerability Scanning (CVS) coverage for malware advisories on par with what already exists for regular (CVE) advisories. CVS in GitLab works by fanning out a per-advisory event from the ingestion pipeline to a scanning worker; this MR is the publishing side of that contract for the malware stream. From the design doc:

"Continuous Vulnerability Scanning (CVS) is currently triggered for every new or updated advisory published within the last 14 days. For each advisory, an event triggers a sidekiq job that is queued to create vulnerabilities for every affected package in the GitLab instance."

The subscriber, scanner, and the resulting Vulnerabilities::Finding creation are in !226519 (closed).

Comparison: PM advisories vs malware advisories

Side-by-side table across storage, format, auth, connector, checkpoint, PURL selection, add-on gating, cron, and air-gapped path.

Aspect	PM advisories (existing)	Malware advisories (this MR)
Storage	Public GCP bucket (anonymous)	Private GCP bucket via PDS (IJWT-authed signed URLs)
Format	NDJSON files, listObjects-based	.tar.zst archives, manifest-driven delta + full
Auth	None	IJWT via CloudConnector::Tokens.get + custom headers
Connector	Connector::Gcp (Google::Cloud::Storage.anonymous)	Connector::Pds (Gitlab::HTTP + signed URLs)
Checkpoint	pm_checkpoints (data_type: advisories/licenses/cve_enrichment)	pm_checkpoints (data_type: malware_advisories) -- same table
PURL selection	package_metadata_purl_types	package_metadata_malware_advisories_purl_types (independent)
Add-on gate	License-based only	SSCS add-on (malware-advisories-sscs)
Cron	every 5 minutes	every 5 minutes
Air-gapped path	Manual download to vendor/package_metadata/advisories	Manual download to vendor/package_metadata/malware_advisories (tooling tracked in #594758)

Why a separate sync mechanism

Duplication audit + 7-point case for keeping the malware pipeline independent of the public PM advisories pipeline.

A reasonable first instinct is "we already have a working PM advisories sync -- bolt malware onto it." The audit below shows where the new pipeline actually overlaps with the existing one, and why folding malware into the existing pipeline would push domain coupling into the wrong layers.

What's duplicated (shell only) vs genuinely new (substance)

Two terms used in the table below:

• shell -- the orchestration wrapper around a sync: worker, service, configuration class, fabricator, event class, and admin checkbox grid. These follow a single pattern that both pipelines reuse mechanically.
• substance -- the parts that actually decide what gets synced and how: the wire contract (connector), the on-disk format (decompressor), the advisory schema (data object), the persistence target (ingestion service + tables), and the access controls (add-on gate + PURL setting).

The shell can look identical between pipelines because it just iterates configs, drives a connector, and forwards parsed objects to an ingestion service. The substance can't, because each pipeline has a different upstream contract, different data shape, and different downstream consumers.

Layer	Duplicated from public sync?	What's actually different
Worker (MalwareAdvisoriesSyncWorker <-> AdvisoriesSyncWorker)	Shell only. Same cron / lease / license shell.	Adds the malware-advisories-sscs add-on gate.
Service (MalwareAdvisorySyncService <-> SyncService)	Shell only. Same orchestration: iterate configs, run connector, ingest slices, advance checkpoint.	Routes to the PDS connector and the malware ingestion service.
Configuration (MalwareAdvisorySyncConfiguration <-> SyncConfiguration)	Shell only. Same configs_for pattern + per-PURL fan-out.	Different admin setting, different storage type (:pds vs :gcp), v3 only.
Checkpoint model (Checkpoint)	Reused as-is.	Same pm_checkpoints table + Checkpoint model, scoped by data_type = malware_advisories; adds first_sync? + for_malware_advisories.
Fabricator (MalwareAdvisoryDataObjectFabricator <-> DataObjectFabricator)	Shell only. Iterates data_file.each, yields domain objects.	Wraps MalwareAdvisoryDataObject.
DataObject (MalwareAdvisoryDataObject <-> AdvisoryDataObject)	Substance -- schemas differ.	GLAM identifiers, urls, withdrawn -- no CVSS / CVE.
Ingestion service + tasks	Substance -- different target tables.	Writes pm_malware_advisories + pm_malware_affected_packages (added in !226519 (closed)).
Connector (Connector::Pds vs Connector::Gcp)	Substance -- different wire contract.	IJWT auth, signed URLs, tar.zst archives, manifest-driven /delta and /all.
Decompressor (Archive::TarZstReader)	Substance -- doesn't exist in public sync.	v2 NDJSON is uncompressed; v3 needs zstd + tar.
Event (IngestedMalwareAdvisoryEvent <-> IngestedAdvisoryEvent)	Shell only. Same pattern.	Different event class -- different subscribers in !226519 (closed).
Admin setting	Shell only. Same UI pattern (checkbox grid).	Separate column + helper so admins can opt in independently.

Why the substance forces a dedicated pipeline

1. The wire contract is incompatible. Public sync uses Google::Cloud::Storage.anonymous over a public bucket and walks files with listObjects -- no auth, no archives. Malware sync goes through PDS with IJWT bearer tokens (Authorization: Bearer <JWT> validated via OIDC against aud=gitlab-pmdb-distribution-service + scopes containing malware_advisories), signed-URL responses, and per-PURL .tar.zst archives. Trying to overload Connector::Gcp for both would hide the auth-and-compression branch behind a giant conditional.
2. The advisory schemas don't overlap. Public advisories carry CVE identifiers, CVSS v2/v3/v4, source glad/trivy-db. Malware advisories carry GLAM identifiers, urls, withdrawn dates, source glam. Reusing AdvisoryDataObject would either drop fields or accumulate nullable columns that mean different things per source.
3. Target tables are separate. pm_malware_advisories and pm_malware_affected_packages (added in !226519 (closed)) are not pm_advisories / pm_affected_packages. Different retention, different access patterns from the SBOM scanner.
4. Access is gated by an add-on. Malware sync only runs for instances with the malware-advisories-sscs purchase. Folding this gate into the public sync would mean the public worker has to know about a malware-specific add-on and route data based on it -- leaking domain coupling upward.
5. Failure isolation. A break in PDS (auth issue, bucket outage, tar.zst glitch) must not stop public advisory + license sync. Separate worker, queue, and service keep the blast radius narrow; Sidekiq retries and defer_on_database_health_signal are independently configurable per pipeline.
6. Independent observability and admin controls. Separate cron entry, separate cronjob: queue name, separate add-on-aware admin setting (package_metadata_malware_advisories_purl_types), separate dashboards -- all easier to wire when the pipeline is its own unit.
7. V3 is the new direction; V2 is legacy. The design doc explicitly calls v3 (manifest + tar.zst + signed URLs) a separate evolution path that will eventually subsume v2. Keeping malware on a v3-native pipeline avoids retrofitting v2 code with v3-only concerns it may never need.

Honest acknowledgement

The orchestration shell (worker, service, fabricator) could be unified later behind a generic interface parameterised by data type -- a refactor opportunity, not a blocker. The duplication is small (~200 lines total), located in clearly-bounded files, and lives only at the wiring layer; the substance that justifies the new pipeline is in the connector, decompressor, schema, ingestion, and admin controls.

What this MR contains

Bullet list of the sync + connector, ingestion, DB, and admin pieces added by this MR.

Sync + connector

• MalwareAdvisorySyncConfiguration — one config per admin-permitted PURL type (reads the new dedicated setting)
• MalwareAdvisoriesSyncWorker — cron worker, license + add-on gates, every 5 minutes (config/initializers/1_settings.rb)
• MalwareAdvisorySyncService — orchestrator with lease + stop signal, iterates per-PURL configs
• Connector::Pds — /delta + /all dispatch, multi-PURL response parser, IJWT + headers
• Connector::Archive::TarZstReader — zstd → tar → NDJSON chunk stream

Ingestion

• MalwareAdvisoryDataObject — captures advisory + packages + urls + withdrawn
• MalwareAdvisoryDataObjectFabricator — line iteration over BaseDataFile#each
• MalwareAdvisoryIngestionService + MalwareAdvisoryIngestionTask + MalwareAffectedPackageIngestionTask
• IngestedMalwareAdvisoryEvent — published per recent advisory

DB

• Reuse the existing pm_checkpoints table for malware checkpoints (new data_type: malware_advisories); add first_sync? + for_malware_advisories to PackageMetadata::Checkpoint
• Migration add_package_metadata_malware_advisories_purl_types_to_application_settings
• Extend Enums::PackageMetadata with malware_advisories data type and v3 version format

Admin

• New application_settings.package_metadata_malware_advisories_purl_types column + model validation, accessor, controller permitted params, helper, view block

External dependencies

Why a zstd decompressor is required, the zstd-ruby gem choice + license analysis, and the remaining gem-side gap (malware_advisories unit primitive).

Why we need a zstd decompressor

The existing public PM advisories sync (PackageMetadata::SyncService + Connector::Gcp) is uncompressed end-to-end -- each file is fetched whole and read line-by-line:

Format	Compression	Reader
v1	None -- plain CSV	Connector::CsvDataFile.parse runs CSV.parse(line) per line
v2	None -- plain NDJSON	Connector::NdjsonDataFile.parse runs Gitlab::Json.parse(line) per line

Connector::Gcp calls @gcp_file.download(skip_decompress: true) (ee/lib/gitlab/package_metadata/connector/gcp.rb:223), but that flag only controls GCS's transport-level gzip handling; the payload itself never gets compressed and there is no archive extraction anywhere in the pipeline.

The v3 format introduced for malware advisories changes that. From the design doc:

"All data objects are compressed with tar.zstd." "Inside the delta archives, data is split into one or more NDJSON chunk files named by sequence number."

So this MR has to do something the existing sync never did: take bytes off the wire, decompress the zstd stream, untar it, and yield the NDJSON chunks inside. Archive::TarZstReader is that step. A Gemfile audit at the start of this work confirmed there was no existing zstd-capable gem in the project, so something had to be brought in.

Choice in this MR: zstd-ruby gem

The MR adds one new Gemfile entry:

gem 'zstd-ruby', '~> 1.5', feature_category: :software_composition_analysis

Property	Value
Gem	zstd-ruby
Gem license (LICENSE file)	BSD-3-Clause -- permitted as New BSD in config/dependency_decisions.yml
Gem license (gemspec metadata)	MIT -- also permitted; the gemspec disagrees with the LICENSE file, which is a known maintainer oversight upstream. license_finder reads the gemspec, so it reports MIT; either reading clears policy.
Vendored upstream	Facebook's libzstd as C source
Upstream license	Dual BSD-3-Clause OR GPL-2.0; we select BSD-3-Clause. Not affected by the 2017 restriction on the now-discontinued Facebook BSD+PATENTS license.
Runtime requirement	None -- the gem's native extension links statically to the vendored libzstd, so no system zstd binary or libzstd.so lookup at runtime.
Install-time requirement	C toolchain (already present for the other native-extension gems like pg, nokogiri, grpc).

Archive::TarZstReader#decompress is a thin wrapper over Zstd.decompress; the gem raises a bare RuntimeError on bad input, which we rescue and re-raise as our typed Archive::TarZstReader::DecompressionError so the connector sees a consistent failure surface.

Other external dependency

Dependency	Kind	Status
gitlab-cloud-connector gem -- malware_advisories.yml unit primitive + pmdb_distribution_service.yml backend service	Upstream gem catalog change	CloudConnector::Tokens.get(unit_primitive: :malware_advisories, ...) will not produce a token (and so aud / scopes will not be populated) until the gem ships config/unit_primitives/malware_advisories.yml (license_types: [ultimate]) and config/backend_services/pmdb_distribution_service.yml. After that the Gemfile pin is bumped. Tracked at #599784 (comment).

Stdlib pieces (Gem::Package::TarReader, StringIO, URI) and existing project libraries (Gitlab::HTTP, Gitlab::Json, CloudConnector::Tokens, Gitlab::CurrentSettings) cover the rest.

What's NOT in this MR

Out of scope	Where it lands
pm_malware_advisories + pm_malware_affected_packages schema	!226519 (closed)
PackageMetadata::MalwareAdvisory + MalwareAffectedPackage AR models	!226519 (closed)
GlobalMalwareAdvisoryScanWorker + scanner integration	!226519 (closed)
Vulnerability creation for malware advisories	!226519 (closed)
Specs / factories	Removed -- this is a POC, tests follow the schema MR
malware_advisories unit primitive + pmdb_distribution_service backend YAMLs in gitlab-cloud-connector gem	Gem-side change; until added the issued IJWTs won't carry the scope/aud. See #599784 (comment).

How this MR and !226519 (closed) work together

Diagram showing this MR's sync output feeding the schema and scanning workers landing in !226519 (closed).

This MR (!226660)                          !226519
+------------------------------------+      +------------------------------+
| Sync + auth + admin                |      | Schema + scanning            |
|                                    |      |                              |
| 1. Pull IJWT from CloudConnector   |      | 1. pm_malware_advisories +   |
| 2. Call PDS /delta or /all         |      |    pm_malware_affected_packages |
| 3. Fetch + extract tar.zst         |      | 2. Subscribe to event        |
| 4. Upsert ----------> (writes to schema in !226519)                      |
| 5. Publish IngestedMalwareAdvisoryEvent --> 3. Scan SBOM, create vulns  |
|                                    |      |                              |
+------------------------------------+      +------------------------------+

The two MRs are mutually dependent at runtime; either can be reviewed in isolation but neither runs end-to-end without the other.

Local testing -- offline vendor path

Drive the sync end-to-end against GLAM v3 archives via the offline path. Two flows are covered: (i) full-dataset sync and (ii) delta sync. Verified on a local GDK -- full dataset ingests all 212,988 advisories, delta ingests 211,316.

Prereqs

Both MRs landed against master (or both branches checked out / lifted locally):

• !226660 (closed) -- sync + connector path (this MR).
• !226519 (closed) -- pm_malware_advisories / pm_malware_affected_packages schema and AR models.

Setup -- example archives

Drop the GLAM v3 archives into a local directory and export it. Any directory works; the test only reads from it.

export ARCHIVE_DIR=$HOME/npm_malware_advisory_example
ls "$ARCHIVE_DIR"
# Expected files:
#   v3_npm_full_dataset.tar.zst    <- full snapshot (hash-sharded NN/000000000.ndjson + checkpoint.json)
#   1774942712_npm_delta.tar.zst   <- one delta archive (flat 000000000.ndjson ... NNN.ndjson)

All other paths below are relative to the GitLab repo root, so run the shell snippets from there (cd path/to/gitlab).

Step 0 -- bring in the schema files (only if !226519 (closed) isn't merged)

If !226519 (closed) has merged to master, skip this step. Otherwise lift the schema-side files into the working tree (no scanner integration), then run migrations. Nothing here gets committed.

git fetch origin poc/malware-advisories-schema-589555:poc/malware-advisories-schema-589555

git checkout poc/malware-advisories-schema-589555 -- \
  db/migrate/20260309000001_create_pm_malware_advisories.rb \
  db/migrate/20260309000002_create_pm_malware_affected_packages.rb \
  ee/app/models/package_metadata/malware_advisory.rb \
  ee/app/models/package_metadata/malware_affected_package.rb \
  db/docs/pm_malware_advisories.yml \
  db/docs/pm_malware_affected_packages.yml \
  ee/spec/factories/package_metadata/pm_malware_advisories.rb \
  ee/spec/factories/package_metadata/pm_malware_affected_packages.rb \
  ee/app/validators/json_schemas/pm_malware_advisory_identifiers.json

bundle exec rake db:migrate

If the 20260309000001 / 20260309000002 migration timestamps clash with anything already on your master, rename the two files to fresh timestamps before running db:migrate.

Heads up: remove these lifted files from the working tree before pushing this branch (git checkout HEAD -- <paths> or rm) -- they belong to !226519 (closed), and leaving them in the tree will trip the unused-methods-linter pre-push hook on to_advisory_data_object / withdrawn?.

Confirm the schema has urls:

bundle exec rails runner '
result = ActiveRecord::Base.connection.execute(
  "SELECT column_name FROM information_schema.columns " \
  "WHERE table_name = '\''pm_malware_advisories'\'' ORDER BY ordinal_position"
)
puts result.to_a.map { |r| r["column_name"] }.inspect
'
# Should include "urls"

Confirm the sync config resolves to :offline

Once any archive is extracted under vendor/package_metadata/malware_advisories/ (Steps below), the config flips from :pds to :offline:

# rails console
config = PackageMetadata::MalwareAdvisorySyncConfiguration.configs_for('malware_advisories').find { |c| c.purl_type == 'npm' }
config.storage_type      # => :offline
config.base_uri.to_s     # => "<repo-root>/vendor/package_metadata/malware_advisories"
config.version_format    # => "v3"

PackageMetadata::MalwareAdvisorySyncConfiguration.configs_for('malware_advisories').map(&:purl_type)
# => ["gem", "golang", "maven", "npm", "nuget", "pypi", "cargo"]   (default admin set)

A reusable runner helper used by both flows below (skips the dev throttle and the malware-advisories-sscs add-on gate, releases any stale lease, runs the service directly, prints counts):

run_offline_sync() {
  bundle exec rails runner 'Gitlab::Redis::SharedState.with { |r| r.del("gitlab:exclusive_lease:mw_offline_test") }'
  PM_SYNC_IN_DEV=true bundle exec rails runner '
    PackageMetadata::Checkpoint.for_malware_advisories.delete_all
    PackageMetadata::MalwareAdvisory.delete_all
    PackageMetadata::MalwareAffectedPackage.delete_all
    lease = Gitlab::ExclusiveLease.new("mw_offline_test", timeout: 10.minutes)
    lease.try_obtain
    t = Time.now
    PackageMetadata::MalwareAdvisorySyncService.execute(lease: lease)
    puts "sync done in #{(Time.now - t).round(2)}s"
    puts "advisories=#{PackageMetadata::MalwareAdvisory.count} affected_packages=#{PackageMetadata::MalwareAffectedPackage.count}"
  '
}

PM_SYNC_IN_DEV=true is required: the dev path otherwise throttles each 200-row slice (THROTTLE_RATE = 0.75s) and bails at MAX_SYNC_DURATION = 4.minutes, which trips before ~210k rows finish.

---

i) Full dataset sync (/all equivalent)

The full-dataset archive is internally hash-sharded: 256 files at NN/000000000.ndjson (00..ff) plus a root checkpoint.json ({"until": <epoch>}). Extract it directly under v3/npm/ so the MalwareOffline glob (*/*.ndjson) picks up every shard; the checkpoint.json sits alongside at the PURL root.

rm -rf vendor/package_metadata/malware_advisories
mkdir -p vendor/package_metadata/malware_advisories/v3/npm
tar --zstd -C vendor/package_metadata/malware_advisories/v3/npm \
  -xf "$ARCHIVE_DIR/v3_npm_full_dataset.tar.zst"

# 256 ndjson files + the snapshot-cutoff marker
find vendor/package_metadata/malware_advisories/v3/npm -name '*.ndjson' | wc -l   # => 256
cat vendor/package_metadata/malware_advisories/v3/npm/checkpoint.json             # => {"until":1780656046}

Run and verify counts:

run_offline_sync
# advisories=212988 affected_packages=212988   (~40s)

Cross-check full parity against the raw source, and confirm the checkpoint is stamped with the snapshot cutoff:

# distinct advisory ids in the source files
find vendor/package_metadata/malware_advisories/v3/npm -name '*.ndjson' -exec cat {} + \
  | python3 -c "import sys,json; s={json.loads(l)['malware_advisory']['id'] for l in sys.stdin if l.strip()}; print(len(s))"
# => 212988  (matches DB advisory count; 0 missing)

bundle exec rails runner '
  cp = PackageMetadata::Checkpoint.for_malware_advisories.find_by(purl_type: "npm")
  puts "checkpoint npm: seq=#{cp.sequence} chunk=#{cp.chunk} first_sync?=#{cp.first_sync?}"
'
# => checkpoint npm: seq=1780656046 chunk=255 first_sync?=false

The checkpoint sequence is the until epoch from the archive's checkpoint.json -- the producer's "data current up to this point" cutoff -- so a subsequent run resumes via /delta?timestamp=1780656046 rather than re-ingesting the snapshot. The PDS /all path reads the same until from the archive's bundled checkpoint.json (via TarZstReader#checkpoint_until); the offline path reads it from the unpacked checkpoint.json at the PURL root.

Why the first shard matters here: a fresh PackageMetadata::Checkpoint defaults to sequence=0, chunk=0. The 00/... shard parses to sequence 0, so the base Offline#pending_files would treat it as "already synced" and drop all 802 of its records. MalwareOffline#pending_files overrides this to process every file on a first_sync? checkpoint. Without that fix the count comes up 802 short (212,186).

ii) Delta dataset sync

The delta archive is flat: 000000000.ndjson ... 000000021.ndjson in one directory. Extract it under a timestamp-named sequence dir (the timestamp in the filename):

rm -rf vendor/package_metadata/malware_advisories
mkdir -p vendor/package_metadata/malware_advisories/v3/npm/1774942712
tar --zstd -C vendor/package_metadata/malware_advisories/v3/npm/1774942712 \
  -xf "$ARCHIVE_DIR/1774942712_npm_delta.tar.zst"

find vendor/package_metadata/malware_advisories/v3/npm -name '*.ndjson' | sort | head
# => .../v3/npm/1774942712/000000000.ndjson ... 000000021.ndjson

Run and verify counts + checkpoint advance:

run_offline_sync
# advisories=211316 affected_packages=211316   (~30s)

bundle exec rails runner '
  cp = PackageMetadata::Checkpoint.for_malware_advisories.find_by(purl_type: "npm")
  puts "checkpoint npm: seq=#{cp.sequence} chunk=#{cp.chunk}"
'
# => checkpoint npm: seq=1774942712 chunk=21   (last consumed chunk)

Shared spot-checks (after either flow)

# rails console
PackageMetadata::MalwareAdvisory.first.attributes
  .slice("advisory_xid", "source_xid", "title", "urls", "withdrawn_date", "identifiers")
PackageMetadata::MalwareAffectedPackage.first.attributes
  .slice("pm_malware_advisory_id", "purl_type", "package_name", "affected_range")

PackageMetadata::MalwareAdvisory.where("array_length(urls, 1) > 0").count   # urls captured
PackageMetadata::MalwareAdvisory.where.not(withdrawn_date: nil).count       # withdrawn rows
PackageMetadata::MalwareAffectedPackage.group(:purl_type).count             # purl_type breakdown

Verify the event published

IngestedMalwareAdvisoryEvent only fires for advisories whose published_date is within PUBLISHED_ADVISORY_INTERVAL = 14.days. If your test data is older, build a synthetic recent advisory and spy on the publish call:

# rails console
events_seen = []
original_publish = Gitlab::EventStore.method(:publish)
Gitlab::EventStore.define_singleton_method(:publish) do |event|
  events_seen << event.data[:advisory_id] if event.is_a?(PackageMetadata::IngestedMalwareAdvisoryEvent)
  original_publish.call(event)
end

synthetic = PackageMetadata::MalwareAdvisoryDataObject.new(
  "malware_advisory" => {
    "id" => "GLAM-2026-05-99999", "source" => "glam",
    "title" => "TEST -- synthetic for event verification",
    "description" => "created by local test plan",
    "published_date" => Time.zone.today.iso8601,
    "identifiers" => [{ "type" => "glam", "name" => "GLAM-2026-05-99999", "value" => "2026-05-99999" }],
    "urls" => []
  },
  "package" => { "name" => "test-package", "purl_type" => "npm", "affected_range" => ">=0" }
)

PackageMetadata::MalwareAdvisoryIngestionService.execute([synthetic])
inserted = PackageMetadata::MalwareAdvisory.find_by(advisory_xid: "GLAM-2026-05-99999")
puts "Events captured: #{events_seen.inspect}; match? #{events_seen.first == inserted.id ? "YES" : "NO"}"

Re-run for idempotency

Re-running either flow (without clearing data) must not change row counts or advance the checkpoint:

bundle exec rails runner 'Gitlab::Redis::SharedState.with { |r| r.del("gitlab:exclusive_lease:mw_offline_test") }'
PM_SYNC_IN_DEV=true bundle exec rails runner '
  before = [PackageMetadata::MalwareAdvisory.count, PackageMetadata::MalwareAffectedPackage.count]
  lease = Gitlab::ExclusiveLease.new("mw_offline_test", timeout: 10.minutes); lease.try_obtain
  PackageMetadata::MalwareAdvisorySyncService.execute(lease: lease)
  after = [PackageMetadata::MalwareAdvisory.count, PackageMetadata::MalwareAffectedPackage.count]
  puts(before == after ? "IDEMPOTENT" : "DRIFTED #{before} -> #{after}")
'

Cleanup

rm -rf vendor/package_metadata/malware_advisories
bundle exec rails runner '
  PackageMetadata::Checkpoint.for_malware_advisories.delete_all
  PackageMetadata::MalwareAdvisory.delete_all
  PackageMetadata::MalwareAffectedPackage.delete_all
'

If you brought the schema files in via Step 0, undo them (and roll back the two migrations):

bundle exec rake db:rollback STEP=2
git checkout HEAD -- \
  db/migrate/20260309000001_create_pm_malware_advisories.rb \
  db/migrate/20260309000002_create_pm_malware_affected_packages.rb \
  ee/app/models/package_metadata/malware_advisory.rb \
  ee/app/models/package_metadata/malware_affected_package.rb \
  db/docs/pm_malware_advisories.yml \
  db/docs/pm_malware_affected_packages.yml \
  ee/spec/factories/package_metadata/pm_malware_advisories.rb \
  ee/spec/factories/package_metadata/pm_malware_affected_packages.rb \
  ee/app/validators/json_schemas/pm_malware_advisory_identifiers.json

Test plan

• Manual: enable SSCS add-on, configure a PURL list in the new admin block, observe MalwareAdvisoriesSyncWorker calls and pm_checkpoints rows (data_type malware_advisories) advance
• Manual: validate Pds#data_after against a fixture .tar.zst archive built from the example NDJSON file
• Air-gapped: drop a .tar.zst under vendor/package_metadata/malware_advisories/v3/<purl_type>/... and verify the offline connector picks it up

🤖 Generated with Claude Code