Class: Homebrew::Vulns::Match Private

Inherits:
Object
  • Object
show all
Includes:
Utils::Output::Mixin
Defined in:
vulns/match.rb

Overview

This class is part of a private API. This class may only be used in the Homebrew/brew repository. Third parties should avoid using this class if possible, as it may be removed or changed without warning.

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), issues versionless queries against each, resolves distro advisories to their upstream CVEs, and evaluates each hit's affected range against the version we ship.

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 is traded for recall and every candidate carries a strategy/confidence label for the reviewer.

Defined Under Namespace

Classes: Evidence, Hit, Identity

Constant Summary collapse

STRATEGY_PRECISION =

This constant is part of a private API. This constant may only be used in the Homebrew/brew repository. Third parties should avoid using this constant if possible, as it may be removed or changed without warning.

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.

T.let(
  { git: 4, registry: 3, cpansa: 2, distro: 1 }.freeze,
  T::Hash[Symbol, Integer],
)
CONFIDENCE =

This constant is part of a private API. This constant may only be used in the Homebrew/brew repository. Third parties should avoid using this constant if possible, as it may be removed or changed without warning.

Recorded in database_specific.confidence for the reviewer.

T.let(
  { git: "high", registry: "high", cpansa: "medium", distro: "low" }.freeze,
  T::Hash[Symbol, String],
)

Instance Method Summary collapse

Methods included from Utils::Output::Mixin

#issue_reporting_message, #odebug, #odeprecated, #odie, #odisabled, #ofail, #oh1, #oh1_title, #ohai, #ohai_title, #onoe, #opoo, #opoo_outside_github_actions, #opoo_without_github_actions_annotation, #pretty_deprecated, #pretty_disabled, #pretty_duration, #pretty_install_status, #pretty_installed, #pretty_uninstalled, #pretty_unmarked, #pretty_upgradable, #pretty_warning

Constructor Details

#initialize(repology: nil, cpan_sec: nil, overrides: nil, bulk: false) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

  • repology (Repology, nil) (defaults to: nil)
  • cpan_sec (CPANSec, nil) (defaults to: nil)
  • overrides (AdvisoryOverrides, nil) (defaults to: nil)
  • bulk (Boolean) (defaults to: false) 


126
127
128
129
130
131
132
133
 
# File 'vulns/match.rb', line 126

def initialize(repology: nil, cpan_sec: nil, overrides: nil, bulk: false)
  @repology = repology
  @cpan_sec = cpan_sec
  @overrides = overrides
  @bulk = bulk
  @vuln_cache = T.let({}, T::Hash[String, T.nilable(Vulnerability)])
  @history = T.let(History.new, History)
end

Instance Method Details

#advisories_for(formula) ⇒ Array<Hit>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Returns one Hit per distinct vulnerability (grouped by CVE alias) reached by any strategy. Distro-ecosystem records are resolved to their upstream CVE(s) so multi-CVE advisories split into per-CVE hits and collapse onto the same CVE reached via GIT/registry. All queries are versionless so historic bump-fixed advisories are returned; #range_status evaluates each hit against the shipped version.

Parameters:

Returns:



168
169
170
171
172
 
# File 'vulns/match.rb', line 168

def advisories_for(formula)
  result = T.let([], T::Array[Hit])
  each_advisory_batch([formula]) { |_, hits| result = hits }
  result
end

#affected_entry(formula, hit, events, fixed, status, status_evidence) ⇒ Hash{Symbol => T.untyped}

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
 
# File 'vulns/match.rb', line 592

def affected_entry(formula, hit, events, fixed, status, status_evidence)
  eco = T.let({ fix: fixed ? "bump" : nil }, T::Hash[Symbol, T.nilable(String)])
  eco[:range_state] = status.state.to_s if status
  eco[:upstream_fixed_in] = status.fixed_in if status&.fixed_in
  # Attribute the resource whose subject decided the state, falling back
  # to the highest-precision evidence when nothing was comparable.
  if (resource = status_evidence&.resource || hit.resource)
    eco[:resource] = resource
    eco[:resource_purl] = (status_evidence if status_evidence&.resource)&.key ||
                          hit.evidence.find { |e| e.resource == resource }&.key
  end
  {
    package:            {
      ecosystem: OsvExport::ECOSYSTEM,
      name:      formula.name,
      purl:      OsvExport.purl(formula.name),
    },
    ranges:             [{ type: "ECOSYSTEM", events: }],
    ecosystem_specific: eco,
  }
end

#aggregate_state_at(formula, hit) ⇒ Symbol?

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
 
# File 'vulns/match.rb', line 700

def aggregate_state_at(formula, hit)
  results = hit.evidence.filter_map do |ev|
    # Evidence built without a subject_version (distro queries, own-
    # identity rows for a formula with no derivable tag) is deliberately
    # uncheckable and must stay that way at historical revisions too;
    # substituting the historical formula version would compare it
    # against the distro record's distro-versioned range.
    next if ev.subject_version.nil?

    present, subject = subject_version_at(formula, ev)
    # Absence means this formula revision did not ship the vulnerable
    # package. Treat it as fixed for boundary walking so a temporary
    # removal can be the fix boundary while still allowing the walk to
    # find an older affected revision.
    next :fixed unless present
    next :unknown if subject.nil?

    evidence_range_status(ev, subject)&.state || :unknown
  end
  return if results.empty?
  return :affected if results.include?(:affected)
  return if results.include?(:unknown)
  return :fixed if results.include?(:fixed)

  :not_applicable
end

#build_osv_queries(identity, formula_version) ⇒ Array<Array<(OSV::Package, Evidence)>>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
 
# File 'vulns/match.rb', line 264

def build_osv_queries(identity, formula_version)
  queries = T.let([], T::Array[[OSV::Package, Evidence]])

  if (repo = identity.git_repo)
    queries << [{ ecosystem: "GIT", name: repo, version: nil },
                Evidence.new(strategy: :git, ecosystem: "GIT", name: repo,
                             subject_version: identity.git_tag || formula_version,
                             key: repo).freeze]
  end

  if (pkg = identity.primary_package) && pkg.ecosystem != "CPAN"
    queries << [{ ecosystem: pkg.ecosystem, name: pkg.name, version: nil },
                Evidence.new(strategy: :registry, ecosystem: pkg.ecosystem, name: pkg.name,
                             subject_version: pkg.version, 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: nil },
                Evidence.new(strategy: :registry, ecosystem: pkg.ecosystem, name: pkg.name,
                             subject_version: pkg.version, 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, ecosystem:, name: srcname,
                               key: "#{ecosystem}/#{srcname}").freeze]
    end
  end

  queries
end

#confidence_for(hit, status) ⇒ String

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



578
579
580
581
582
583
584
 
# File 'vulns/match.rb', line 578

def confidence_for(hit, status)
  base = CONFIDENCE.fetch(hit.strategy)
  return base if status

  # No comparable range: the reviewer must set the boundary by hand.
  (base == "high") ? "medium" : "low"
end

#cpan_evidence(identity) ⇒ Array<Evidence>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
 
# File 'vulns/match.rb', line 300

def cpan_evidence(identity)
  result = T.let([], T::Array[Evidence])
  primary = identity.primary_package
  if primary&.ecosystem == "CPAN"
    result << Evidence.new(strategy: :cpansa, ecosystem: "CPAN", name: primary.name,
                           subject_version: primary.version, key: primary.purl)
  end
  identity.resource_packages.each do |resource, pkg|
    next if pkg.ecosystem != "CPAN"

    result << Evidence.new(strategy: :cpansa, ecosystem: "CPAN", name: pkg.name,
                           subject_version: pkg.version, key: pkg.purl, resource:)
  end
  result
end

#cpan_sec ⇒ CPANSec

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Returns:



141
142
143
 
# File 'vulns/match.rb', line 141

def cpan_sec
  @cpan_sec ||= CPANSec.load
end

#cpansa_vulnerability(adv, id:) ⇒ Vulnerability

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Synthesise a Vulnerability for a CPANSA advisory when OSV has no record. id is scoped to the single CVE (or CPANSA id) being handled so a multi-CVE advisory whose CVEs are absent from OSV yields distinct records instead of collapsing under the lowest CVE in dedup.

Parameters:

Returns:



251
252
253
254
255
256
257
258
259
 
# File 'vulns/match.rb', line 251

def cpansa_vulnerability(adv, id:)
  summary = adv.description.to_s.lines.first&.strip
  Vulnerability.new({
    "id"         => id,
    "summary"    => summary,
    "details"    => adv.description,
    "references" => adv.references.map { |u| { "type" => "WEB", "url" => u } },
  }.compact)
end

#dedup_by_cve(hits) ⇒ Array<Hit>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



447
448
449
450
451
452
453
454
455
456
457
 
# File 'vulns/match.rb', line 447

def dedup_by_cve(hits)
  hits.group_by(&:canonical_id).map do |_, group|
    next group.fetch(0) if group.one?

    primary = group.max_by { |h| STRATEGY_PRECISION.fetch(h.strategy) }
    raise ArgumentError, "Cannot pick a primary hit from an empty group" if primary.nil?

    Hit.new(vulnerability: primary.vulnerability,
            evidence:      group.flat_map(&:evidence).uniq)
  end
end

#distro_packages_for(name) ⇒ Repology::DistroMap

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Bulk mode (the --all sweep) trusts the published index; only a single-formula run (the PR bot, or an explicit named check) may hit the live Repology API for a formula the index doesn't yet cover.

Parameters:

Returns:



408
409
410
411
412
413
414
415
416
 
# File 'vulns/match.rb', line 408

def distro_packages_for(name)
  indexed = repology.distro_packages_for(name)
  return indexed if indexed.any? || @bulk

  Repology.lookup(name)
rescue CachedFeed::Error => e
  odebug "Repology lookup for #{name} failed: #{e.message}"
  {}
end

#each_advisory_batch(formulae, &_blk) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

This method returns an undefined value.

Bulk form of #advisories_for: builds the labelled queries for a chunk of formulae at once, sends them through a single OSV.query_batch (which itself slices at BATCH_SIZE), then yields (formula, hits) in input order. Per-formula query counts vary widely (one distro entry per ecosystem×srcname), so chunking bounds memory without accumulating the whole tap's queries or records; the @vuln_cache still spans chunks.

Parameters:



187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'vulns/match.rb', line 187

def each_advisory_batch(formulae, &_blk)
  formulae.each_slice(BULK_CHUNK) do |chunk|
    identities = T.let({}, T::Hash[Formula, Identity])
    chunk.each do |formula|
      next if @overrides&.skip_formula?(formula.name)

      identities[formula] = identify(formula)
    end
    labelled = T.let([], T::Array[[OSV::Package, [Formula, Evidence]]])
    identities.each do |f, identity|
      next unless identity.identifiable?

      build_osv_queries(identity, f.version.to_s).each do |query, evidence|
        labelled << [query, [f, evidence]]
      end
    end

    by_formula = T.let({}, T::Hash[Formula, T::Hash[String, T::Array[Evidence]]])
    if labelled.any?
      OSV.query_batch(labelled.map(&:first)).each_with_index do |stubs, i|
        formula, evidence = labelled.fetch(i).last
        id_evidence = by_formula[formula] ||= {}
        stubs.each { |stub| (id_evidence[stub.fetch("id")] ||= []) << evidence }
      end
    end

    prefetch_vulnerabilities(by_formula.each_value.flat_map(&:keys))

    chunk.each do |f|
      identity = identities[f]
      next yield f, [] unless identity&.identifiable?

      yield f, hits_from(by_formula[f] || {}, identity)
    end
  end
end

#evidence_range_status(evidence, subject_version) ⇒ Vulnerability::RangeStatus?

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



502
503
504
505
506
507
508
509
510
511
 
# File 'vulns/match.rb', line 502

def evidence_range_status(evidence, subject_version)
  return if subject_version.nil?

  if evidence.strategy == :cpansa
    adv = evidence.advisory
    CPANSec.range_status(adv, subject_version) if adv
  else
    evidence.source_record&.range_status(evidence.ecosystem, evidence.name, subject_version)
  end
end

#fetch_vulnerability(id) ⇒ Vulnerability?

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



434
435
436
 
# File 'vulns/match.rb', line 434

def fetch_vulnerability(id)
  @vuln_cache.fetch(id) { @vuln_cache[id] = load_vulnerability(id) }
end

#first_fixed_version(formula, hit) ⇒ String, ...

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Walk homebrew-core git history (newest first) via History and return the pkg_version at the oldest revision where the aggregate of every checkable subject is still :fixed. Re-running the full per-evidence range check with each revision's subject versions keeps last_affected and exclusive-bound semantics intact and stops as soon as any subject (primary or a resource) drops back into :affected, so a primary fixed at 2.0 with a resource fixed at 3.0 yields 3.0.

Returns:

  • nil when the current aggregate is not :fixed.
  • :never_affected when the walk reaches :not_applicable (or the start of the formula's history) without ever seeing :affected, i.e. Homebrew jumped from a version below introduced straight past fixed and never shipped an affected build. The caller drops the candidate rather than emitting {introduced: "0", fixed: <first>}.
  • :history_unavailable when the tap is a shallow clone, the formula has no git history or a revision cannot be loaded or compared, so the caller can skip the candidate rather than inventing a boundary.
  • a pkg_version String when the walk hits :affected.

Parameters:

Returns:



634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
 
# File 'vulns/match.rb', line 634

def first_fixed_version(formula, hit)
  return unless range_status(hit, formula_name: formula.name)&.first&.fixed?

  last_fixed = T.let(formula.pkg_version.to_s, String)
  result = @history.walk(formula) do |old|
    aggregate = aggregate_state_at(old, hit)
    case aggregate
    when :fixed
      last_fixed = old.pkg_version.to_s
      nil
    when :affected then last_fixed
    when :not_applicable then :never_affected
    when nil then :history_unavailable
    else raise TypeError, "unexpected historical aggregate: #{aggregate.inspect}"
    end
  end
  result || :never_affected
end

#first_reintroduced_version(formula, hit) ⇒ String, ...

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Return the lowest representable formula pkg_version in the newest contiguous affected run. This is the introduced boundary when a reviewed fixed range becomes affected again after a formula or resource regression, including a run whose version strings move backwards after a version_scheme change. After finding the transition, the remaining history is checked so the new interval cannot cover a known non-affected formula version. :not_reintroduced means no prior non-affected revision was verified, either because all loadable history remained affected, the tap is a shallow clone, the formula has no git history or a revision could not be loaded or compared safely.

Parameters:

Returns:



665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
 
# File 'vulns/match.rb', line 665

def first_reintroduced_version(formula, hit)
  return unless range_status(hit, formula_name: formula.name)&.first&.affected?

  first_affected = T.let(formula.pkg_version.to_s, String)
  transition_found = T.let(false, T::Boolean)
  result = @history.walk(formula) do |old|
    aggregate = aggregate_state_at(old, hit)
    next :not_reintroduced if aggregate.nil?

    pkg_version = old.pkg_version.to_s
    begin
      historical = PkgVersion.parse(pkg_version)
      boundary = PkgVersion.parse(first_affected)
      if transition_found
        next :not_reintroduced if aggregate != :affected && historical >= boundary
      elsif aggregate == :affected
        first_affected = pkg_version if historical < boundary
      else
        next :not_reintroduced if historical >= boundary

        transition_found = true
      end
    rescue ArgumentError
      next :not_reintroduced
    end
    nil
  end
  # Every early result is fail-closed: untrusted history, an uncomparable
  # revision or a known non-affected version above the boundary.
  return :not_reintroduced unless result.nil?

  transition_found ? first_affected : :not_reintroduced
end

#hits_from(id_evidence, identity) ⇒ Array<Hit>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
 
# File 'vulns/match.rb', line 227

def hits_from(id_evidence, identity)
  hits = resolve_upstream(id_evidence, identity)
  cpan_evidence(identity).each do |ev|
    cpan_sec.advisories_for(ev.name).each do |adv|
      annotated = Evidence.new(**ev.to_h, advisory: adv).freeze
      if adv.cves.any?
        adv.cves.each do |cve|
          record = fetch_vulnerability(cve) || cpansa_vulnerability(adv, id: cve)
          hits << Hit.new(vulnerability: record, evidence: [annotated])
        end
      else
        hits << Hit.new(vulnerability: cpansa_vulnerability(adv, id: adv.id.to_s),
                        evidence:      [annotated])
      end
    end
  end
  dedup_by_cve(hits)
end

#identify(formula) ⇒ Identity

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'vulns/match.rb', line 146

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

#load_vulnerability(id) ⇒ Vulnerability?

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



439
440
441
442
443
444
 
# File 'vulns/match.rb', line 439

def load_vulnerability(id)
  Vulnerability.new(OSV.vulnerability(id))
rescue OSV::Error => e
  odebug "OSV.vulnerability(#{id}) failed: #{e.message}"
  nil
end

#own_evidence(identity) ⇒ Array<Evidence>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Evidence rows pointing at our own identity keys (git repo, primary registry package) with the formula/package version as subject. Attached to distro-resolved upstream hits so #range_status can evaluate the upstream CVE record's affected[] against something comparable.

Parameters:

Returns:



391
392
393
394
395
396
397
398
399
400
401
402
 
# File 'vulns/match.rb', line 391

def own_evidence(identity)
  result = T.let([], T::Array[Evidence])
  if (repo = identity.git_repo)
    result << Evidence.new(strategy: :distro, ecosystem: "GIT", name: repo,
                           subject_version: identity.git_tag, key: "upstream:#{repo}").freeze
  end
  if (pkg = identity.primary_package)
    result << Evidence.new(strategy: :distro, ecosystem: pkg.ecosystem, name: pkg.name,
                           subject_version: pkg.version, key: "upstream:#{pkg.purl}").freeze
  end
  result
end

#prefetch_vulnerabilities(ids) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

This method returns an undefined value.

OSV querybatch returns id/modified stubs. Warm @vuln_cache with the full records for a chunk's stub ids before per-formula processing so #resolve_upstream reads mostly from cache.

Parameters:



425
426
427
428
429
430
431
 
# File 'vulns/match.rb', line 425

def prefetch_vulnerabilities(ids)
  missing = ids.uniq.reject { |id| @vuln_cache.key?(id) }
  missing.each_slice(MAX_VULN_FETCH_THREADS) do |slice|
    slice.map { |id| [id, Thread.new { load_vulnerability(id) }] }
         .each { |id, t| @vuln_cache[id] = t.value }
  end
end

#range_status(hit, formula_name: nil) ⇒ Array<(Vulnerability::RangeStatus, Evidence)>?

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Evaluate hit against every evidence's subject, each against the record that evidence was matched against, and aggregate: :affected if any subject is affected (a fixed primary must not hide an affected resource, or vice versa), else unknown if any subject is unresolved, else :fixed if any is fixed, else :not_applicable only when every comparable subject says so. Returns [status, evidence] where evidence is the one whose result was chosen (used by #first_fixed_version and for the emitted record's resource attribution), or nil if no evidence produced a checkable answer.

Parameters:

  • hit (Hit)
  • formula_name (String, nil) (defaults to: nil)

Returns:



473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
 
# File 'vulns/match.rb', line 473

def range_status(hit, formula_name: nil)
  subjects = hit.evidence.reject { |ev| ev.strategy == :distro && ev.subject_version.nil? }
                .group_by(&:resource).map do |_, evidence|
    results = evidence.filter_map do |ev|
      status = evidence_range_status(ev, ev.subject_version)
      [status, ev] if status
    end
    results.find { |s, _| s.affected? } || results.find { |s, _| s.fixed? } || results.first
  end
  results = subjects.compact
  selected = results.find { |s, _| s.affected? }
  selected ||= results.find { |s, _| s.fixed? } || results.first unless subjects.include?(nil)
  override = @overrides&.advisory_override(formula_name, hit.identifiers) if formula_name
  return selected unless override

  status, evidence = selected ||
                     (results.find { |candidate, _| candidate.state == override.state } if override.state) ||
                     [nil, hit.primary_evidence]
  state = override.state || status&.state
  return selected unless state

  fixed_in = override.fixed_in_overridden ? override.fixed_in : status&.fixed_in
  [Vulnerability::RangeStatus.new(state:, fixed_in:).freeze, evidence]
end

#record_id(formula, hit) ⇒ String

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



566
567
568
 
# File 'vulns/match.rb', line 566

def record_id(formula, hit)
  OsvExport.record_id(formula, hit.canonical_id)
end

#record_ids(formula, hit) ⇒ Array<String>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

Returns:



571
572
573
 
# File 'vulns/match.rb', line 571

def record_ids(formula, hit)
  ([record_id(formula, hit)] + hit.identifiers.map { |id| OsvExport.record_id(formula, id) }).uniq
end

#repology ⇒ Repology

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Returns:



136
137
138
 
# File 'vulns/match.rb', line 136

def repology
  @repology ||= Repology.load
end

#resolve_to_cves(record, seen, budget) ⇒ Array<Vulnerability>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Returns the set of CVE records record derives from. [record] if it is one already; [] if the walk exhausts without reaching a CVE (the caller then keeps record itself as a low-confidence hit).

Parameters:

Returns:



372
373
374
375
376
377
378
379
380
381
382
383
384
 
# File 'vulns/match.rb', line 372

def resolve_to_cves(record, seen, budget)
  return [record] if record.cve_ids.any?
  return [] if budget.zero?

  follow = record.upstream.presence
  follow ||= record.related.grep(CVE_ID) if record.id.start_with?(RELATED_AS_UPSTREAM_PREFIX)
  Array(follow).uniq.flat_map do |ref|
    next [] unless seen.add?(ref)

    upstream = fetch_vulnerability(ref)
    upstream ? resolve_to_cves(upstream, seen, budget - 1) : []
  end.uniq(&:id)
end

#resolve_upstream(id_evidence, identity) ⇒ Array<Hit>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Turn id => [Evidence, ...] into [Hit, ...], resolving each record to the CVE(s) it derives from. upstream is walked transitively with a per-walk visited set (chains like USN -> UBUNTU-CVE-* -> CVE-* occur in practice). related links to different vulnerabilities per the OSV schema and is only consulted for its bare CVE ids when upstream is empty (AlmaLinux ALSA records use it that way). A record that is already a CVE by id or alias, or that reaches no CVE within the hop budget, is kept as-is. Each resolved hit gains synthesised evidence pointing at our own identity so #range_status can check the CVE record's affected[] against our version.

Parameters:

Returns:



336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
 
# File 'vulns/match.rb', line 336

def resolve_upstream(id_evidence, identity)
  own = own_evidence(identity)
  hits = T.let([], T::Array[Hit])

  id_evidence.each do |id, evidence|
    record = fetch_vulnerability(id)
    next if record.nil?

    resolved = resolve_to_cves(record, Set[id], MAX_UPSTREAM_HOPS)
    if resolved.empty?
      hits << Hit.new(vulnerability: record, evidence:)
      next
    end

    resolved.each do |cve_record|
      ev = cve_record.equal?(record) ? evidence : evidence + own
      hits << Hit.new(vulnerability: cve_record, evidence: ev)
    end
  end

  hits
end

#subject_version_at(formula, evidence) ⇒ Array<(Boolean, [String, nil])>

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Resolve a historical subject by upstream package identity so resource label changes do not look like removals. The boolean distinguishes a package that was absent (not affected at that revision) from one that was present but had no comparable version (uncheckable).

Parameters:

Returns:



732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
 
# File 'vulns/match.rb', line 732

def subject_version_at(formula, evidence)
  unless evidence.resource
    stable = formula.stable
    stable_url = stable&.url
    if evidence.ecosystem == "GIT"
      repo = Identify.repo_url(stable_url, formula.head&.url, formula.homepage)
      return [true, nil] if repo.nil?
      return [true, nil] if repo != evidence.name

      version = Identify.tag(stable_url) || stable&.specs&.dig(:tag) || stable&.version&.to_s
      return [true, version]
    end

    primary_package = Identify.registry_package(stable_url)
    return [true, nil] if primary_package.nil?
    if primary_package.ecosystem != evidence.ecosystem || primary_package.name != evidence.name
      return [false, nil]
    end

    return [true, primary_package.version]
  end

  exact_resource = formula.resources.find { |resource| resource.name == evidence.resource }
  exact_identity_unknown = T.let(false, T::Boolean)
  if exact_resource
    exact_package = Identify.registry_package(exact_resource.url)
    if exact_package.nil?
      # A reused resource label is not enough to prove package identity.
      # Search the remaining resources in case the package was renamed,
      # then leave this revision uncheckable if no identity can be found.
      exact_identity_unknown = true
    elsif exact_package.ecosystem == evidence.ecosystem && exact_package.name == evidence.name
      return [true, exact_package.version]
    end
  end

  formula.resources.each do |resource|
    next if resource.equal?(exact_resource)

    package = Identify.registry_package(resource.url)
    next if package.nil?
    next if package.ecosystem != evidence.ecosystem || package.name != evidence.name

    return [true, package.version]
  end

  return [true, nil] if exact_identity_unknown

  [false, nil]
end

#to_brew_record(formula, hit, first_fixed: nil, first_reintroduced: nil, now: Time.now.utc) ⇒ Hash{Symbol => T.untyped}

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Emit a candidate BREW-* OSV record for hit against formula.

first_fixed is the PkgVersion at which Homebrew first shipped a fix (from #first_fixed_version or a hand-set value), and first_reintroduced is the first Homebrew version in a newer affected interval (from #first_reintroduced_version). Otherwise #range_status is consulted: affected? == false sets fixed: pkg_version and ecosystem_specific.fix: "bump"; affected? == true (or no comparable range) emits no fixed event and fix: null. As with OsvExport.record_for, OsvExport.merge_existing preserves on-disk ranges on rewrite so a hand-corrected boundary sticks.

Parameters:

  • formula (Formula)
  • hit (Hit)
  • first_fixed (String, nil) (defaults to: nil)
  • first_reintroduced (String, nil) (defaults to: nil)
  • now (Time) (defaults to: Time.now.utc)

Returns:



530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
 
# File 'vulns/match.rb', line 530

def to_brew_record(formula, hit, first_fixed: nil, first_reintroduced: nil, now: Time.now.utc)
  vuln = hit.vulnerability
  timestamp = now.strftime("%Y-%m-%dT%H:%M:%SZ")
  status, status_evidence = range_status(hit, formula_name: formula.name)

  fixed = first_fixed
  fixed ||= formula.pkg_version.to_s if status&.fixed?
  events = T.let([{ introduced: first_reintroduced || "0" }], T::Array[T::Hash[Symbol, String]])
  events << { fixed: } if fixed

  record = T.let({
    schema_version:    OsvExport::SCHEMA_VERSION,
    id:                record_id(formula, hit),
    published:         timestamp,
    modified:          timestamp,
    upstream:          hit.identifiers,
    affected:          [affected_entry(formula, hit, events, fixed, status, status_evidence)],
    database_specific: {
      source:            "matched",
      strategy:          hit.strategy.to_s,
      confidence:        confidence_for(hit, status),
      upstream_evidence: hit.evidence.map { |e| e.to_h.except(:advisory, :source_record).compact },
    },
  }, T::Hash[Symbol, T.untyped])

  record[:summary] = vuln.summary if vuln.summary
  record[:details] = vuln.details if vuln.details
  record[:severity] = vuln.severity_entries if vuln.severity_entries.any?
  if (refs = vuln.references).any?
    record[:references] = refs.uniq { |r| [r["type"], URI::RFC2396_PARSER.unescape(r["url"].to_s)] }
  end

  record
end