vulns: add Match with identify and advisories_for

Match#identify composes Identify (git repo/tag, registry package for the
primary URL and each resource) with the Repology index (distro source
package names, live lookup fallback) into a single Identity per formula.

Match#advisories_for builds one OSV querybatch across GIT, language-
registry and versionless distro-ecosystem queries, routes CPAN packages
through CPANSec, fetches full records once per id (cached across calls),
and collapses hits sharing a CVE alias into one Hit whose strategy is the
highest-precision path that reached it, keeping every path as evidence.

Repology and CPANSec feeds are loaded once per Match instance and are
injectable for tests. This is authoring-time code for advisory-database
CI and the homebrew-core PR bot; it never runs on a user's machine.
This commit is contained in:
Andrew Nesbitt committed 2026-07-28 10:29:30 +01:00
1 parent e3a61648bb
commit 6dff421708
2 files changed
+493

No files matched your search

+257
View File
@@ -0,0 +1,257 @@
# typed: false
# frozen_string_literal: true
require "vulns/match"
RSpec.describe Homebrew::Vulns::Match do
let(:repology) do
Homebrew::Vulns::Repology.new({ "meta" => {}, "formulae" => {
"requests" => { "Debian" => ["requests"], "Alpine" => ["py3-requests"] },
} })
end
let(:cpan_sec) do
Homebrew::Vulns::CPANSec.new({ "meta" => {}, "dists" => {
"Image-ExifTool" => { "advisories" => [
{ "id" => "CPANSA-Image-ExifTool-2021-22204", "cves" => ["CVE-2021-22204"],
"affected_versions" => ["<12.24"], "fixed_versions" => ["12.24"] },
] },
} })
end
let(:matcher) { described_class.new(repology:, cpan_sec:) }
def stub_repology_lookup(result = {})
allow(Homebrew::Vulns::Repology).to receive(:lookup).and_return(result)
end
describe "#identify" do
it "derives git repo/tag, primary registry package, resources and distro packages" do
f = formula("requests") do
T.bind(self, T.class_of(Formula))
homepage "https://requests.readthedocs.io"
url "https://files.pythonhosted.org/packages/aa/bb/cc/requests-2.31.0.tar.gz"
head "https://github.com/psf/requests.git"
resource "certifi" do
url "https://files.pythonhosted.org/packages/11/22/33/certifi-2024.2.2.tar.gz"
end
resource "vendored-c" do
url "https://example.com/blob-1.0.tar.gz"
end
end
identity = matcher.identify(f)
expect(identity.git_repo).to eq "https://github.com/psf/requests"
expect(identity.git_tag).to eq "2.31.0"
expect(identity.primary_package.ecosystem).to eq "PyPI"
expect(identity.primary_package.name).to eq "requests"
expect(identity.primary_package.version).to eq "2.31.0"
expect(identity.resource_packages.keys).to eq ["certifi"]
expect(identity.resource_packages["certifi"].purl).to eq "pkg:pypi/certifi@2024.2.2"
expect(identity.distro_packages)
.to eq("Debian" => ["requests"], "Alpine" => ["py3-requests"])
expect(identity.any?).to be true
end
it "falls back to Repology.lookup when the index has no entry" do
f = formula("newthing") do
T.bind(self, T.class_of(Formula))
url "https://example.com/newthing-1.0.tar.gz"
end
stub_repology_lookup({ "Debian" => ["newthing"] })
expect(matcher.identify(f).distro_packages).to eq("Debian" => ["newthing"])
end
it "swallows a Repology lookup error to an empty distro map" do
f = formula("newthing") do
T.bind(self, T.class_of(Formula))
url "https://example.com/newthing-1.0.tar.gz"
end
allow(Homebrew::Vulns::Repology).to receive(:lookup)
.and_raise(Homebrew::Vulns::CachedFeed::Error, "boom")
expect(matcher.identify(f).distro_packages).to eq({})
end
it "reports any? false when nothing is derivable" do
f = formula("mystery") do
T.bind(self, T.class_of(Formula))
url "https://example.com/mystery-1.0.tar.gz"
end
stub_repology_lookup
expect(matcher.identify(f).any?).to be false
end
end
describe "#build_osv_queries" do
def pkg(ecosystem:, name:, version:, purl:)
Homebrew::Vulns::Identify::RegistryPackage.new(ecosystem:, name:, version:, purl:)
end
it "emits GIT, registry (primary + resource) and distro queries with matching evidence" do
identity = described_class::Identity.new(
git_repo: "https://github.com/psf/requests",
git_tag: "v2.31.0",
primary_package: pkg(ecosystem: "PyPI", name: "requests", version: "2.31.0",
purl: "pkg:pypi/requests@2.31.0"),
resource_packages: { "certifi" => pkg(ecosystem: "PyPI", name: "certifi", version: "2024.2.2",
purl: "pkg:pypi/certifi@2024.2.2") },
distro_packages: { "Debian" => ["requests"], "Alpine" => ["py3-requests"] },
)
queries = matcher.build_osv_queries(identity)
expect(queries.map(&:first)).to eq [
{ ecosystem: "GIT", name: "https://github.com/psf/requests", version: "v2.31.0" },
{ ecosystem: "PyPI", name: "requests", version: "2.31.0" },
{ ecosystem: "PyPI", name: "certifi", version: "2024.2.2" },
{ ecosystem: "Debian", name: "requests", version: nil },
{ ecosystem: "Alpine", name: "py3-requests", version: nil },
]
expect(queries.map { |_, e| [e.strategy, e.key, e.resource] }).to eq [
[:git, "https://github.com/psf/requests", nil],
[:registry, "pkg:pypi/requests@2.31.0", nil],
[:registry, "pkg:pypi/certifi@2024.2.2", "certifi"],
[:distro, "Debian/requests", nil],
[:distro, "Alpine/py3-requests", nil],
]
end
it "excludes CPAN packages from OSV queries and omits GIT when no repo derived" do
identity = described_class::Identity.new(
git_repo: nil,
git_tag: "13.55",
primary_package: pkg(ecosystem: "CPAN", name: "Image-ExifTool", version: "13.55",
purl: "pkg:cpan/EXIFTOOL/Image-ExifTool@13.55"),
resource_packages: { "extra" => pkg(ecosystem: "CPAN", name: "Try-Tiny", version: "0.31",
purl: "pkg:cpan/ETHER/Try-Tiny@0.31") },
distro_packages: {},
)
expect(matcher.build_osv_queries(identity)).to eq []
end
end
describe "#cpan_advisory_ids" do
it "returns CVE ids for CPAN primary and resource packages via CPANSec" do
identity = described_class::Identity.new(
git_repo: nil, git_tag: nil,
primary_package: Homebrew::Vulns::Identify::RegistryPackage.new(
ecosystem: "CPAN", name: "Image-ExifTool", version: "12.00",
purl: "pkg:cpan/EXIFTOOL/Image-ExifTool@12.00"
),
resource_packages: {}, distro_packages: {}
)
ids = matcher.cpan_advisory_ids(identity)
expect(ids.map(&:first)).to eq ["CVE-2021-22204"]
expect(ids.first.last.strategy).to eq :cpansa
end
end
describe "#advisories_for" do
let(:exiftool) do
formula("exiftool") do
T.bind(self, T.class_of(Formula))
url "https://cpan.metacpan.org/authors/id/E/EX/EXIFTOOL/Image-ExifTool-12.00.tar.gz"
head "https://github.com/exiftool/exiftool.git"
end
end
before { stub_repology_lookup({ "Debian" => ["libimage-exiftool-perl"] }) }
it "queries every strategy in one batch, fetches full records, and dedups by CVE alias" do
expect(Homebrew::Vulns::OSV).to receive(:query_batch).with(
[
{ ecosystem: "GIT", name: "https://github.com/exiftool/exiftool", version: "12.00" },
{ ecosystem: "Debian", name: "libimage-exiftool-perl", version: nil },
],
).and_return(
[
[{ "id" => "CVE-2021-22204" }],
[{ "id" => "DSA-4910-1" }, { "id" => "DSA-0000-0" }],
],
)
allow(Homebrew::Vulns::OSV).to receive(:vulnerability).with("CVE-2021-22204").and_return(
{ "id" => "CVE-2021-22204", "aliases" => ["GHSA-xxxx-yyyy-zzzz"] },
)
allow(Homebrew::Vulns::OSV).to receive(:vulnerability).with("DSA-4910-1").and_return(
{ "id" => "DSA-4910-1", "aliases" => ["CVE-2021-22204"] },
)
allow(Homebrew::Vulns::OSV).to receive(:vulnerability).with("DSA-0000-0").and_return(
{ "id" => "DSA-0000-0", "aliases" => [] },
)
hits = matcher.advisories_for(exiftool)
expect(hits.map(&:canonical_id).sort).to eq ["CVE-2021-22204", "DSA-0000-0"]
merged = hits.find { |h| h.canonical_id == "CVE-2021-22204" }
expect(merged.strategy).to eq :git
expect(merged.evidence.map(&:strategy)).to eq [:git, :cpansa, :distro]
expect(merged.vulnerability.id).to eq "CVE-2021-22204"
expect(hits.find { |h| h.canonical_id == "DSA-0000-0" }.strategy).to eq :distro
end
it "drops ids whose full record cannot be fetched" do
allow(Homebrew::Vulns::OSV).to receive(:query_batch).and_return(
[[{ "id" => "CVE-9999-0000" }], []],
)
allow(Homebrew::Vulns::OSV).to receive(:vulnerability)
.and_raise(Homebrew::Vulns::OSV::ApiError, "404")
expect(matcher.advisories_for(exiftool)).to eq []
end
it "returns [] without hitting OSV when nothing is identifiable" do
f = formula("mystery") do
T.bind(self, T.class_of(Formula))
url "https://example.com/mystery-1.0.tar.gz"
end
stub_repology_lookup
expect(Homebrew::Vulns::OSV).not_to receive(:query_batch)
expect(matcher.advisories_for(f)).to eq []
end
it "caches OSV.vulnerability lookups across calls" do
allow(Homebrew::Vulns::OSV).to receive(:query_batch)
.and_return([[{ "id" => "CVE-2021-22204" }], []])
expect(Homebrew::Vulns::OSV).to receive(:vulnerability).once
.and_return({ "id" => "CVE-2021-22204" })
matcher.advisories_for(exiftool)
matcher.advisories_for(exiftool)
end
end
describe described_class::Hit do
def vuln(id, aliases: [])
Homebrew::Vulns::Vulnerability.new({ "id" => id, "aliases" => aliases })
end
def ev(strategy, key: "k")
Homebrew::Vulns::Match::Evidence.new(strategy:, key:)
end
it "sorts evidence by descending strategy precision and reports the highest as #strategy" do
hit = described_class.new(vulnerability: vuln("CVE-1"),
evidence: [ev(:distro), ev(:git), ev(:registry)])
expect(hit.evidence.map(&:strategy)).to eq [:git, :registry, :distro]
expect(hit.strategy).to eq :git
end
it "uses the lowest CVE alias as canonical_id, or the record id when there is none" do
expect(described_class.new(vulnerability: vuln("GHSA-x", aliases: ["CVE-2024-2", "CVE-2024-1"]),
evidence: [ev(:git)]).canonical_id).to eq "CVE-2024-1"
expect(described_class.new(vulnerability: vuln("GHSA-y"),
evidence: [ev(:git)]).canonical_id).to eq "GHSA-y"
end
it "rejects empty evidence" do
expect { described_class.new(vulnerability: vuln("CVE-1"), evidence: []) }
.to raise_error(ArgumentError)
end
end
end
+236
View File
@@ -0,0 +1,236 @@
# typed: strict
# frozen_string_literal: true
require "vulns/cpan_sec"
require "vulns/identify"
require "vulns/osv"
require "vulns/repology"
require "vulns/vulnerability"
module Homebrew
module Vulns
# Authoring-time advisory matcher. For a given {Formula} it derives every
# OSV.dev query key it can (forge repository, language-registry package for
# the primary URL and each `resource`, distro source packages via
# {Repology}, CPAN distribution via {CPANSec}) and returns the deduplicated
# set of vulnerabilities any of them hit, tagged with the strategy that
# reached each one.
#
# Runs in `Homebrew/advisory-database` CI and the homebrew-core PR bot to
# produce candidate `BREW-*` records for human review; never on a user's
# machine, so request volume and false-positive rate are traded for recall.
class Match
include Utils::Output::Mixin
# Descending precision. When several strategies reach the same CVE the
# highest is reported as the hit's primary strategy; the rest are kept as
# supporting evidence.
STRATEGY_PRECISION = T.let(
{ git: 4, registry: 3, cpansa: 2, distro: 1 }.freeze,
T::Hash[Symbol, Integer],
)
Identity = Struct.new(
:git_repo, :git_tag, :primary_package, :resource_packages, :distro_packages,
keyword_init: true
) do
extend T::Sig
sig { returns(T::Boolean) }
def any?
!git_repo.nil? || !primary_package.nil? || resource_packages.any? || distro_packages.any?
end
end
Evidence = Struct.new(:strategy, :key, :resource, keyword_init: true)
class Hit
sig { returns(Vulnerability) }
attr_reader :vulnerability
sig { returns(T::Array[Evidence]) }
attr_reader :evidence
sig { params(vulnerability: Vulnerability, evidence: T::Array[Evidence]).void }
def initialize(vulnerability:, evidence:)
raise ArgumentError, "Hit requires at least one Evidence" if evidence.empty?
@vulnerability = T.let(vulnerability, Vulnerability)
@evidence = T.let(
evidence.sort_by { |e| -STRATEGY_PRECISION.fetch(e.strategy) }.freeze,
T::Array[Evidence],
)
end
sig { returns(Symbol) }
def strategy
evidence.fetch(0).strategy
end
sig { returns(T.nilable(String)) }
def resource
evidence.fetch(0).resource
end
sig { returns(String) }
def canonical_id
vulnerability.cve_ids.min || vulnerability.id
end
end
sig { params(repology: T.nilable(Repology), cpan_sec: T.nilable(CPANSec)).void }
def initialize(repology: nil, cpan_sec: nil)
@repology = T.let(repology, T.nilable(Repology))
@cpan_sec = T.let(cpan_sec, T.nilable(CPANSec))
@vuln_cache = T.let({}, T::Hash[String, T.nilable(T::Hash[String, T.untyped])])
end
sig { returns(Repology) }
def repology
@repology ||= Repology.load
end
sig { returns(CPANSec) }
def cpan_sec
@cpan_sec ||= CPANSec.load
end
sig { params(formula: Formula).returns(Identity) }
def identify(formula)
stable = formula.stable
stable_url = stable&.url
Identity.new(
git_repo: Identify.repo_url(stable_url, formula.head&.url, formula.homepage),
git_tag: Identify.tag(stable_url) || stable&.specs&.dig(:tag) || stable&.version&.to_s,
primary_package: Identify.registry_package(stable_url),
resource_packages: formula.resources.filter_map do |r|
pkg = Identify.registry_package(r.url)
[r.name, pkg] if pkg
end.to_h.freeze,
distro_packages: distro_packages_for(formula.name),
).freeze
end
# Returns one {Hit} per distinct vulnerability (grouped by CVE alias)
# reached by any strategy. Each hit's `evidence` lists every path that
# reached it, highest-precision first.
sig { params(formula: Formula).returns(T::Array[Hit]) }
def advisories_for(formula)
identity = identify(formula)
return [] unless identity.any?
labelled = build_osv_queries(identity)
id_evidence = T.let({}, T::Hash[String, T::Array[Evidence]])
if labelled.any?
OSV.query_batch(labelled.map(&:first)).each_with_index do |stubs, i|
evidence = labelled.fetch(i).last
stubs.each { |stub| (id_evidence[stub.fetch("id")] ||= []) << evidence }
end
end
cpan_advisory_ids(identity).each { |id, evidence| (id_evidence[id] ||= []) << evidence }
hits = id_evidence.filter_map do |id, evidence|
record = fetch_vulnerability(id)
Hit.new(vulnerability: Vulnerability.new(record), evidence:) if record
end
dedup_by_cve(hits)
end
sig { params(identity: Identity).returns(T::Array[[OSV::Package, Evidence]]) }
def build_osv_queries(identity)
queries = T.let([], T::Array[[OSV::Package, Evidence]])
if (repo = identity.git_repo) && (tag = identity.git_tag)
queries << [{ ecosystem: "GIT", name: repo, version: tag },
Evidence.new(strategy: :git, key: repo).freeze]
end
if (pkg = identity.primary_package) && pkg.ecosystem != "CPAN"
queries << [{ ecosystem: pkg.ecosystem, name: pkg.name, version: pkg.version },
Evidence.new(strategy: :registry, key: pkg.purl).freeze]
end
identity.resource_packages.each do |resource, pkg|
next if pkg.ecosystem == "CPAN"
queries << [{ ecosystem: pkg.ecosystem, name: pkg.name, version: pkg.version },
Evidence.new(strategy: :registry, key: pkg.purl, resource:).freeze]
end
identity.distro_packages.each do |ecosystem, srcnames|
srcnames.each do |srcname|
queries << [{ ecosystem:, name: srcname, version: nil },
Evidence.new(strategy: :distro, key: "#{ecosystem}/#{srcname}").freeze]
end
end
queries
end
sig { params(identity: Identity).returns(T::Array[[String, Evidence]]) }
def cpan_advisory_ids(identity)
result = T.let([], T::Array[[String, Evidence]])
cpan_packages(identity).each do |pkg, resource|
evidence = Evidence.new(strategy: :cpansa, key: pkg.purl, resource:).freeze
cpan_sec.advisories_for(pkg.name).each do |adv|
ids = adv.cves.presence || [adv.id.to_s]
ids.each { |id| result << [id, evidence] }
end
end
result
end
sig {
params(identity: Identity).returns(T::Array[[Identify::RegistryPackage, T.nilable(String)]])
}
def cpan_packages(identity)
result = T.let([], T::Array[[Identify::RegistryPackage, T.nilable(String)]])
primary = identity.primary_package
result << [primary, nil] if primary&.ecosystem == "CPAN"
identity.resource_packages.each do |resource, pkg|
result << [pkg, resource] if pkg.ecosystem == "CPAN"
end
result
end
sig { params(name: String).returns(Repology::DistroMap) }
def distro_packages_for(name)
indexed = repology.distro_packages_for(name)
return indexed if indexed.any?
Repology.lookup(name)
rescue CachedFeed::Error => e
odebug "Repology lookup for #{name} failed: #{e.message}"
{}
end
# OSV `querybatch` returns id/modified stubs; the full record is fetched
# once per id and cached across formulae.
sig { params(id: String).returns(T.nilable(T::Hash[String, T.untyped])) }
def fetch_vulnerability(id)
@vuln_cache.fetch(id) do
@vuln_cache[id] = begin
OSV.vulnerability(id)
rescue OSV::Error => e
odebug "OSV.vulnerability(#{id}) failed: #{e.message}"
nil
end
end
end
sig { params(hits: T::Array[Hit]).returns(T::Array[Hit]) }
def dedup_by_cve(hits)
hits.group_by(&:canonical_id).map do |_, group|
next group.fetch(0) if group.one?
primary = T.must(group.max_by { |h| STRATEGY_PRECISION.fetch(h.strategy) })
Hit.new(vulnerability: primary.vulnerability,
evidence: group.flat_map(&:evidence).uniq)
end
end
end
end
end