Add Homebrew::Vulns::Vulnerability OSV record wrapper

This commit is contained in:
Andrew Nesbitt committed 2026-07-12 19:18:53 +01:00
1 parent 876abd4213
commit a9a0392667
2 files changed
+591

No files matched your search

@@ -0,0 +1,363 @@
# typed: false
# frozen_string_literal: true
require "vulns/vulnerability"
RSpec.describe Homebrew::Vulns::Vulnerability do
def vuln(data)
described_class.new(data)
end
def semver_range(*events)
{ "affected" => [{ "ranges" => [{ "type" => "SEMVER", "events" => events }] }] }
end
describe "basic accessors" do
it "parses id, summary, details, aliases, references" do
v = vuln(
"id" => "CVE-2024-1234",
"summary" => "Something bad",
"details" => "Longer description",
"aliases" => ["GHSA-xxxx-yyyy-zzzz"],
"references" => [{ "type" => "WEB", "url" => "https://example.com" }],
)
expect(v.id).to eq "CVE-2024-1234"
expect(v.summary).to eq "Something bad"
expect(v.details).to eq "Longer description"
expect(v.aliases).to eq ["GHSA-xxxx-yyyy-zzzz"]
expect(v.references).to eq [{ "type" => "WEB", "url" => "https://example.com" }]
end
it "defaults missing arrays to empty" do
v = vuln("id" => "CVE-2024-1234")
expect(v.aliases).to eq []
expect(v.references).to eq []
expect(v.affected).to eq []
end
end
describe "#severity" do
it "extracts critical from a CVSS v3.1 vector" do
v = vuln("id" => "CVE-2024-0001",
"severity" => [{ "type" => "CVSS_V3",
"score" => "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H" }])
expect(v.severity).to eq :critical
expect(v.severity_display).to eq "CRITICAL"
expect(v.severity_level).to eq 4
end
it "extracts high from a CVSS v3.1 vector" do
v = vuln("id" => "CVE-2024-0002",
"severity" => [{ "type" => "CVSS_V3",
"score" => "CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:C/C:H/I:H/A:H" }])
expect(v.severity).to eq :high
expect(v.severity_level).to eq 3
end
it "extracts medium from a CVSS v3.1 vector" do
v = vuln("id" => "CVE-2024-0003",
"severity" => [{ "type" => "CVSS_V3",
"score" => "CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N" }])
expect(v.severity).to eq :medium
expect(v.severity_level).to eq 2
end
it "extracts low from a CVSS v3.1 vector" do
v = vuln("id" => "CVE-2024-0004",
"severity" => [{ "type" => "CVSS_V3",
"score" => "CVSS:3.1/AV:L/AC:H/PR:H/UI:R/S:U/C:L/I:N/A:N" }])
expect(v.severity).to eq :low
expect(v.severity_level).to eq 1
end
it "extracts critical from a CVSS v3.0 vector" do
v = vuln("id" => "CVE-2024-0005",
"severity" => [{ "type" => "CVSS_V3",
"score" => "CVSS:3.0/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H" }])
expect(v.severity).to eq :critical
end
it "returns nil severity for a zero-impact CVSS vector" do
v = vuln("id" => "CVE-2026-1234",
"severity" => [{ "type" => "CVSS_V4",
"score" => "CVSS:4.0/AV:N/AC:L/AT:N/PR:N/UI:N/VC:N/VI:N/VA:N/SC:N/SI:N/SA:N" }])
expect(v.severity).to be_nil
expect(v.severity_display).to eq "UNKNOWN"
expect(v.severity_level).to eq 0
end
it "returns nil severity for an invalid CVSS vector" do
v = vuln("id" => "CVE-2024-0003", "severity" => [{ "type" => "CVSS_V3", "score" => "INVALID-CVSS" }])
expect(v.severity).to be_nil
expect(v.severity_display).to eq "UNKNOWN"
end
it "returns nil severity for an empty CVSS vector" do
v = vuln("id" => "CVE-2024-0003", "severity" => [{ "type" => "CVSS_V3", "score" => "" }])
expect(v.severity).to be_nil
end
it "falls back to database_specific when the CVSS vector cannot be scored" do
v = vuln("id" => "CVE-2024-0003",
"severity" => [{ "type" => "CVSS_V3", "score" => "INVALID-CVSS" }],
"database_specific" => { "severity" => "HIGH" })
expect(v.severity).to eq :high
end
# v4 is not scored locally so priority falls through to the v3 entry.
it "falls through unsupported CVSS versions in priority order" do
v = vuln("id" => "CVE-2026-1234",
"severity" => [
{ "type" => "CVSS_V2", "score" => "AV:L/AC:H/Au:S/C:P/I:P/A:N" },
{ "type" => "CVSS_V4",
"score" => "CVSS:4.0/AV:N/AC:L/AT:N/PR:H/UI:N/VC:L/VI:L/VA:N/SC:N/SI:N/SA:N" },
{ "type" => "CVSS_V3", "score" => "CVSS:3.0/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H" },
])
expect(v.severity).to eq :critical
end
it "extracts severity from top-level database_specific" do
v = vuln("id" => "CVE-2024-1234", "database_specific" => { "severity" => "HIGH" })
expect(v.severity).to eq :high
end
it "extracts severity from affected database_specific" do
v = vuln("id" => "CVE-2024-1234", "affected" => [{ "database_specific" => { "severity" => "MEDIUM" } }])
expect(v.severity).to eq :medium
end
it "extracts severity from affected ecosystem_specific" do
v = vuln("id" => "OSV-2023-1234", "affected" => [{ "ecosystem_specific" => { "severity" => "HIGH" } }])
expect(v.severity).to eq :high
end
it "prefers ecosystem_specific over database_specific within an affected entry" do
v = vuln("id" => "OSV-2023-1234",
"affected" => [{ "ecosystem_specific" => { "severity" => "HIGH" },
"database_specific" => { "severity" => "LOW" } }])
expect(v.severity).to eq :high
end
it "normalises MODERATE to medium" do
v = vuln("id" => "CVE-2024-1234", "database_specific" => { "severity" => "MODERATE" })
expect(v.severity).to eq :medium
expect(v.severity_display).to eq "MEDIUM"
end
it "returns UNKNOWN when no severity source is present" do
v = vuln("id" => "CVE-2024-1234")
expect(v.severity).to be_nil
expect(v.severity_display).to eq "UNKNOWN"
expect(v.severity_level).to eq 0
end
end
describe "#identifiers and #cve_ids" do
it "returns id followed by aliases" do
v = vuln("id" => "GHSA-xxxx-yyyy-zzzz", "aliases" => ["CVE-2024-1234", "OSV-2024-1"])
expect(v.identifiers).to eq ["GHSA-xxxx-yyyy-zzzz", "CVE-2024-1234", "OSV-2024-1"]
end
it "handles missing aliases" do
expect(vuln("id" => "CVE-2024-1234").identifiers).to eq ["CVE-2024-1234"]
end
it "extracts CVE ids from id and aliases" do
v = vuln("id" => "CVE-2024-1234", "aliases" => ["GHSA-xxxx-yyyy-zzzz", "CVE-2024-5678"])
expect(v.cve_ids).to contain_exactly("CVE-2024-1234", "CVE-2024-5678")
end
end
describe "#advisory_url and #fix_urls" do
it "returns the first ADVISORY reference url" do
v = vuln("id" => "CVE-2024-1234",
"references" => [
{ "type" => "WEB", "url" => "https://example.com" },
{ "type" => "ADVISORY", "url" => "https://security.example.com/advisory" },
])
expect(v.advisory_url).to eq "https://security.example.com/advisory"
end
it "returns nil when no ADVISORY reference exists" do
expect(vuln("id" => "CVE-2024-1234").advisory_url).to be_nil
end
it "returns all FIX reference urls" do
v = vuln("id" => "CVE-2024-1234",
"references" => [
{ "type" => "FIX", "url" => "https://github.com/example/commit/abc123" },
{ "type" => "FIX", "url" => "https://github.com/example/commit/def456" },
{ "type" => "WEB", "url" => "https://example.com" },
])
expect(v.fix_urls).to contain_exactly(
"https://github.com/example/commit/abc123",
"https://github.com/example/commit/def456",
)
end
end
describe "#fixed_versions" do
it "extracts fixed events from affected ranges" do
v = vuln("id" => "CVE-2024-1234",
"affected" => [{ "ranges" => [{ "type" => "SEMVER",
"events" => [{ "introduced" => "0" },
{ "fixed" => "v1.2.3" }] }] }])
expect(v.fixed_versions).to eq ["v1.2.3"]
end
it "deduplicates" do
v = vuln("id" => "CVE-2024-1234",
"affected" => [{ "ranges" => [{ "events" => [{ "fixed" => "v1.0.0" }] },
{ "events" => [{ "fixed" => "v1.0.0" }] }] }])
expect(v.fixed_versions).to eq ["v1.0.0"]
end
end
describe "#affects_version?" do
it "returns true when there is no affected data" do
expect(vuln("id" => "CVE-2024-1234").affects_version?("1.0.0")).to be true
end
it "matches explicit version lists, tolerating a v prefix" do
v = vuln("id" => "CVE-2024-1234", "affected" => [{ "versions" => ["v1.0.0", "1.0.2"] }])
expect(v.affects_version?("1.0.0")).to be true
expect(v.affects_version?("v1.0.0")).to be true
expect(v.affects_version?("1.0.2")).to be true
expect(v.affects_version?("2.0.0")).to be false
end
it "checks introduced/fixed SEMVER ranges" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "1.0.0" },
{ "fixed" => "1.5.0" })))
expect(v.affects_version?("1.2.0")).to be true
expect(v.affects_version?("1.4.9")).to be true
expect(v.affects_version?("1.5.0")).to be false
expect(v.affects_version?("2.0.0")).to be false
expect(v.affects_version?("0.9.0")).to be false
end
it "matches an open-ended range introduced at 0" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "0" })))
expect(v.affects_version?("0.1.0")).to be true
expect(v.affects_version?("1.2.3")).to be true
end
it "matches an open-ended range introduced at v0" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "v0" })))
expect(v.affects_version?("1.2.3")).to be true
end
it "handles limit events" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "0" },
{ "limit" => "2.0.0" })))
expect(v.affects_version?("1.9.9")).to be true
expect(v.affects_version?("2.0.0")).to be false
end
it "treats a wildcard limit as unbounded above" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "1.0.0" },
{ "limit" => "*" })))
expect(v.affects_version?("1.0.0")).to be true
expect(v.affects_version?("9.9.9")).to be true
expect(v.affects_version?("0.9.9")).to be false
end
it "handles multiple intervals in one range" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "0" },
{ "limit" => "2.0.0" },
{ "introduced" => "3.0.0" })))
expect(v.affects_version?("1.9.9")).to be true
expect(v.affects_version?("2.0.0")).to be false
expect(v.affects_version?("2.9.9")).to be false
expect(v.affects_version?("3.0.0")).to be true
expect(v.affects_version?("4.0.0")).to be true
end
it "handles last_affected events inclusively" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "1.0.0" },
{ "last_affected" => "1.5.0" })))
expect(v.affects_version?("1.5.0")).to be true
expect(v.affects_version?("1.5.1")).to be false
end
it "checks SEMVER ranges regardless of package purl ecosystem" do
v = vuln("id" => "CVE-2024-1234",
"affected" => [{ "package" => { "purl" => "pkg:npm/lodash" },
"ranges" => [{ "type" => "SEMVER",
"events" => [{ "introduced" => "0" },
{ "fixed" => "4.17.21" }] }] }])
expect(v.affects_version?("4.17.20")).to be true
expect(v.affects_version?("4.17.21")).to be false
end
# GIT and ECOSYSTEM ranges cannot be evaluated locally; defer to OSV's
# server-side match rather than dropping the finding.
it "keeps a match when only non-SEMVER ranges are present" do
v = vuln("id" => "CVE-2024-1234",
"affected" => [{ "ranges" => [{ "type" => "GIT",
"events" => [{ "introduced" => "abc123" },
{ "fixed" => "def456" }] }] }])
expect(v.affects_version?("1.0.0")).to be true
end
it "fails closed when SEMVER comparison raises" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "1.0.0" },
{ "fixed" => "1.5.0" })))
allow(Homebrew::Vulns::Semver).to receive(:compare).and_raise(StandardError, "boom")
expect(v.affects_version?("1.2.0")).to be true
end
it "fails closed when the target version is not valid SemVer" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "0" },
{ "fixed" => "3.0.1" })))
expect(v.affects_version?("openssl-3.0.0")).to be true
end
it "fails closed when a range boundary is not valid SemVer" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "0" },
{ "fixed" => "not.a.version" })))
expect(v.affects_version?("1.0.0")).to be true
end
it "fails closed when the lower bound is unparseable and the upper bound excludes the target" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "bad" },
{ "fixed" => "2.0.0" })))
expect(v.affects_version?("3.0.0")).to be true
end
it "fails closed when the upper bound is unparseable and the lower bound excludes the target" do
v = vuln({ "id" => "CVE-2024-1234" }.merge(semver_range({ "introduced" => "2.0.0" },
{ "fixed" => "bad" })))
expect(v.affects_version?("1.0.0")).to be true
end
it "keeps a match when a non-matching explicit list coexists with an uncheckable range" do
v = vuln("id" => "CVE-2024-1234",
"affected" => [{ "versions" => ["1.0.0", "1.0.1"],
"ranges" => [{ "type" => "GIT",
"events" => [{ "introduced" => "abc123" },
{ "fixed" => "def456" }] }] }])
expect(v.affects_version?("2.0.0")).to be true
end
it "keeps a match when a non-matching SEMVER range coexists with an uncheckable range" do
v = vuln("id" => "CVE-2024-1234",
"affected" => [
{ "ranges" => [{ "type" => "SEMVER",
"events" => [{ "introduced" => "1.0.0" }, { "fixed" => "1.5.0" }] }] },
{ "ranges" => [{ "type" => "ECOSYSTEM",
"events" => [{ "introduced" => "0" }, { "fixed" => "2.0.0" }] }] },
])
expect(v.affects_version?("3.0.0")).to be true
end
end
describe ".from_osv_list" do
it "wraps each hash in a Vulnerability" do
list = described_class.from_osv_list([{ "id" => "CVE-2024-1111" }, { "id" => "CVE-2024-2222" }])
expect(list.size).to eq 2
expect(list.first).to be_a described_class
expect(list.first.id).to eq "CVE-2024-1111"
end
end
end
+228
View File
@@ -0,0 +1,228 @@
# typed: strict
# frozen_string_literal: true
require "vulns/cvss"
require "vulns/semver"
module Homebrew
module Vulns
# Wraps a single OSV vulnerability record.
# See https://ossf.github.io/osv-schema/ for the field definitions.
class Vulnerability
include Utils::Output::Mixin
CVSS_TYPE_PRIORITY = T.let(
{ "CVSS_V4" => 4, "CVSS_V3" => 3, "CVSS_V2" => 2 }.freeze,
T::Hash[String, Integer],
)
private_constant :CVSS_TYPE_PRIORITY
SEVERITY_LEVEL = T.let(
{ critical: 4, high: 3, medium: 2, low: 1 }.freeze,
T::Hash[Symbol, Integer],
)
private_constant :SEVERITY_LEVEL
sig { returns(String) }
attr_reader :id
sig { returns(T.nilable(String)) }
attr_reader :summary, :details
sig { returns(T.nilable(Symbol)) }
attr_reader :severity
sig { returns(T::Array[String]) }
attr_reader :aliases
sig { returns(T::Array[T::Hash[String, T.untyped]]) }
attr_reader :references, :affected
sig { params(data: T::Hash[String, T.untyped]).void }
def initialize(data)
@id = T.let(data.fetch("id"), String)
@summary = T.let(data["summary"], T.nilable(String))
@details = T.let(data["details"], T.nilable(String))
@aliases = T.let(Array(data["aliases"]), T::Array[String])
@references = T.let(Array(data["references"]), T::Array[T::Hash[String, T.untyped]])
@affected = T.let(Array(data["affected"]), T::Array[T::Hash[String, T.untyped]])
@severity = T.let(extract_severity(data), T.nilable(Symbol))
end
sig { params(list: T::Array[T::Hash[String, T.untyped]]).returns(T::Array[Vulnerability]) }
def self.from_osv_list(list)
list.map { |data| new(data) }
end
sig { returns(String) }
def severity_display
severity&.to_s&.upcase || "UNKNOWN"
end
sig { returns(Integer) }
def severity_level
sev = severity
return 0 if sev.nil?
SEVERITY_LEVEL.fetch(sev, 0)
end
sig { returns(T::Array[String]) }
def identifiers
[id, *aliases].compact
end
sig { returns(T::Array[String]) }
def cve_ids
identifiers.select { |i| i.start_with?("CVE-") }
end
sig { returns(T.nilable(String)) }
def advisory_url
references.find { |r| r["type"] == "ADVISORY" }&.dig("url")
end
sig { returns(T::Array[String]) }
def fix_urls
references.select { |r| r["type"] == "FIX" }.filter_map { |r| r["url"] }
end
sig { returns(T::Array[String]) }
def fixed_versions
affected.flat_map do |aff|
Array(aff["ranges"]).flat_map do |range|
Array(range["events"]).filter_map { |e| e["fixed"] }
end
end.uniq
end
# OSV has already matched this record against the queried version. This
# method only overrides that with `false` when every affected entry can
# be evaluated locally (explicit `versions` list or `SEMVER` range) and
# none of them contains the target. Any uncheckable range or comparison
# failure keeps the match.
sig { params(version: String).returns(T::Boolean) }
def affects_version?(version)
return true if affected.empty?
target = normalize_version(version)
checkable = T.let(false, T::Boolean)
affected.each do |aff|
versions = Array(aff["versions"])
if versions.any?
checkable = true
return true if versions.any? { |v| normalize_version(v.to_s) == target }
end
Array(aff["ranges"]).each do |range|
return true if range["type"] != "SEMVER"
checkable = true
return true if in_semver_range?(target, Array(range["events"]))
end
end
!checkable
end
sig { params(data: T::Hash[String, T.untyped]).returns(T.nilable(Symbol)) }
def extract_severity(data)
Array(data["severity"])
.select { |s| CVSS_TYPE_PRIORITY.key?(s["type"]) }
.sort_by { |s| -CVSS_TYPE_PRIORITY.fetch(s["type"]) }
.each do |s|
sev = CVSS.severity(s["score"].to_s)
return sev if sev
end
top = normalize_severity(data.dig("database_specific", "severity"))
return top if top
affected.each do |aff|
eco = normalize_severity(aff.dig("ecosystem_specific", "severity"))
return eco if eco
db = normalize_severity(aff.dig("database_specific", "severity"))
return db if db
end
nil
end
sig { params(value: T.nilable(String)).returns(T.nilable(Symbol)) }
def normalize_severity(value)
case value&.downcase
when "critical" then :critical
when "high" then :high
when "medium", "moderate" then :medium
when "low" then :low
end
end
sig { params(version: String).returns(String) }
def normalize_version(version)
version.sub(/\Av/i, "")
end
class Uncomparable < StandardError
end
private_constant :Uncomparable
sig { params(target: String, events: T::Array[T::Hash[String, T.untyped]]).returns(T::Boolean) }
def in_semver_range?(target, events)
return false if events.empty?
intervals(events).any? do |lower, upper, upper_inclusive|
above = lower.nil? || semver_compare!(target, lower) >= 0
below = if upper.nil?
true
elsif upper_inclusive
semver_compare!(target, upper) <= 0
else
semver_compare!(target, upper).negative?
end
above && below
end
rescue => e
odebug "Failed to check #{target} against SEMVER range for #{id}: #{e}"
true
end
sig { params(left: String, right: String).returns(Integer) }
def semver_compare!(left, right)
Semver.compare(left, right) || raise(Uncomparable, "cannot compare #{left.inspect} with #{right.inspect}")
end
sig {
params(events: T::Array[T::Hash[String, T.untyped]])
.returns(T::Array[[T.nilable(String), T.nilable(String), T::Boolean]])
}
def intervals(events)
result = []
lower = T.let(nil, T.nilable(String))
events.each do |event|
if (intro = event["introduced"])
result << [lower, nil, false] if lower
intro = normalize_version(intro.to_s)
lower = (intro == "0") ? nil : intro
elsif (fixed = event["fixed"])
result << [lower, normalize_version(fixed.to_s), false]
lower = nil
elsif (last = event["last_affected"])
result << [lower, normalize_version(last.to_s), true]
lower = nil
elsif (limit = event["limit"])
limit = limit.to_s
upper = (limit == "*") ? nil : normalize_version(limit)
result << [lower, upper, false]
lower = nil
end
end
result << [lower, nil, false] if lower || result.empty?
result
end
end
end
end