Publish an event when a malware advisory is ingested

Bala Kumarrequested to merge
606613-publish-malware-advisory-ingested-event into master

What does this MR do and why?

Continuous vulnerability scanning learns about new advisories by subscribing to an ingestion event. The public advisory path publishes one; the malware advisory path published nothing, so a malware advisory could never reach CVS no matter what the scanner did.

This publishes PackageMetadata::IngestedMalwareAdvisoryEvent for each malware advisory ingested inside the publication-age window.

Resolves #606613

This is the first of the Wave 2 CVS issues on &21156 and unblocks the scanner in #612094 (closed), which subscribes to this event.

Implementation notes

The publication-age window is referenced, not copied. #612096 (closed) decided to reuse the existing 14-day window unchanged, so the constant points at the public path's value rather than duplicating 14.days. The two cannot drift.

Publication happens after the transaction commits. The malware tables are on the sec database and the upsert runs inside a SecApplicationRecord.transaction. Publishing inside it would let a subscriber read an advisory that is not yet visible.

Ids are deduped. import_data can carry the same advisory_xid more than once when a multi-package advisory is split across NDJSON lines. The ingestion task already dedupes for the upsert conflict key; without deduping here too, one advisory would produce several events.

Advisories with no published date publish nothing rather than raising, matching how the ingestion task treats them.

On the event base class

The sibling PackageMetadata::IngestedAdvisoryEvent inherits Gitlab::EventStore::Event, but that base is no longer permitted for new events — Gitlab/EventStoreCloudEventInheritance requires Gitlab::EventStore::CloudEvent. The sibling is grandfathered as one of 102 entries in .rubocop_todo/gitlab/event_store_cloud_event_inheritance.yml.

So this event is CloudEvents compliant rather than a literal mirror. Ingestion is an instance-wide background process, so there is no acting user and no organization to attribute; source is instance and subject identifies the advisory. build_cloud_event accepts both as optional, so nothing is invented to satisfy the format.

One consequence for the subscriber in #612094 (closed): the payload is read with event.event_data[:malware_advisory_id], not event.data[...], because CloudEvent nests the payload inside the envelope.

Test coverage

File Cases
spec/events/package_metadata/ingested_malware_advisory_event_spec.rb Builds with the advisory id, attributes to the instance, plus the shared a cloud event with schema contract
ee/spec/services/package_metadata/malware_advisory_ingestion_service_spec.rb Publishes only inside the window, not outside it, one event for a duplicated advisory, nothing for an undated advisory, and publication only after commit

The commit-ordering test asserts the advisory is already readable when the event fires, which is the failure the ordering exists to prevent.

Spec and RuboCop output 
$ bundle exec rspec spec/events/package_metadata/ingested_malware_advisory_event_spec.rb
5 examples, 0 failures

$ bundle exec rspec ee/spec/services/package_metadata/malware_advisory_ingestion_service_spec.rb
11 examples, 0 failures

$ bundle exec rubocop <the six changed files>
no offenses detected

Local reproduction

Run on a GDK with ingest_malware_advisories enabled. Two advisories are ingested in one batch: one published 3 days ago (inside the 14-day window) and one 400 days ago (outside it).

Script, and before / after observations 
# /tmp/repro.rb — counts IngestedMalwareAdvisoryEvent publications during one ingestion
Gitlab::EventStore.singleton_class.prepend(Module.new do
  def publish(event)
    (Thread.current[:observed_events] ||= []) << event
    super
  end
end)
Thread.current[:observed_events] = []

def build(xid, days_ago)
  PackageMetadata::MalwareAdvisoryDataObject.new(
    'malware_advisory' => {
      'id' => xid, 'source' => 'glam', 'title' => "t #{xid}", 'description' => 'd',
      'published_date' => days_ago.days.ago.to_date.iso8601, 'withdrawn' => nil,
      'identifiers' => [{ 'type' => 'glam', 'name' => xid, 'value' => xid }], 'urls' => []
    },
    'package' => { 'name' => "pkg-#{xid.downcase}", 'purl_type' => 'npm', 'affected_range' => '>=0.0.0' }
  )
end

recent = build('GLAM-2026-08-90001', 3)    # inside the window
old    = build('GLAM-2024-01-90002', 400)  # outside it

before = PackageMetadata::MalwareAdvisory.count
PackageMetadata::MalwareAdvisoryIngestionService.execute([recent, old])
after = PackageMetadata::MalwareAdvisory.count

published = Thread.current[:observed_events]
  .select { |e| e.is_a?(PackageMetadata::IngestedMalwareAdvisoryEvent) }

puts "advisories ingested: #{after - before}"
puts "events published   : #{published.size}"
published.each do |e|
  id = e.event_data[:malware_advisory_id]
  puts "  -> advisory_id=#{id} xid=#{PackageMetadata::MalwareAdvisory.find_by_id(id)&.advisory_xid}"
end

Before — on master

advisories ingested        : 2
IngestedMalwareAdvisoryEvent published: 0
in-window advisory persisted : true
out-of-window persisted too  : true

Both advisories land in the database and nothing is published, so CVS never hears about either.

After — on this branch

advisories ingested        : 2
IngestedMalwareAdvisoryEvent published: 1
   -> advisory_id=473289 xid=GLAM-2026-08-90001 subject=package_metadata/malware_advisories/473289
in-window advisory persisted : true
out-of-window persisted too  : true

One event, for the 3-day-old advisory only. The 400-day-old one is still ingested — it is simply out of CVS scope by policy, and remains covered by the SBOM/CI path, which applies no age filter.

Transaction semantics: before and after the after_commit change

The review raised that publishing straight after a transaction do ... end block only guarantees after-commit semantics when that transaction is the outermost one. Publication now registers on the transaction itself, so the guarantee holds regardless of the caller.

Measured in a Rails runner rather than a spec, because RSpec's transactional fixtures hold an outer transaction open for the whole example and make these cases unreachable.

Case Before (publish after the block) After (after_commit)
Normal call, no wrapping transaction 1 event 1 event
Wrapped in an outer transaction 1 event published while the outer transaction was still open 0 while open, 1 after it commits
Outer transaction rolls back 1 event published, advisory not persisted 0 events

The rollback row is the more serious of the two. The old code announced an advisory that does not exist: a subscriber would resolve malware_advisory_id and find nothing. Both rows are now correct.

=== CASE 2: wrapped in an OUTER transaction ===
  events published while outer txn still open: 0
  events published after outer txn committed  : 1
  -> deferred correctly: true

=== CASE 3: outer transaction ROLLS BACK ===
  events published: 0
  advisory persisted: false
  -> no event for rolled-back data: true

A note on the xid format. The first attempt used GLAM-REPRO-RECENT, which silently ingested nothing: MalwareAdvisory validates the xid format, the record was invalid, and bulk_upsert! logged and skipped it. Use a realistic GLAM-YYYY-MM-NNNNN id or the script will appear to do nothing.

Related

Edited by Bala Kumar

Merge request reports