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, 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)
  • bulk (Boolean) (defaults to: false) 


112
113
114
115
116
117
118
119
 
# File 'vulns/match.rb', line 112

def initialize(repology: nil, cpan_sec: nil, bulk: false)
  @repology = repology
  @cpan_sec = cpan_sec
  @bulk = bulk
  @vuln_cache = T.let({}, T::Hash[String, T.nilable(Vulnerability)])
  @formula_versions = T.let({}, T::Hash[String, FormulaVersions])
  @formula_rev_lists = T.let({}, T::Hash[String, T::Array[[String, String]]])
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:



154
155
156
157
158
 
# File 'vulns/match.rb', line 154

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:



540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
 
# File 'vulns/match.rb', line 540

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:



608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
 
# File 'vulns/match.rb', line 608

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?

    subject = subject_version(formula, ev.resource)&.to_s
    evidence_range_status(ev, subject)
  end
  return if results.empty?
  return :affected if results.any?(&:affected?)
  return :fixed if results.any?(&: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:



244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
 
# File 'vulns/match.rb', line 244

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:



526
527
528
529
530
531
532
 
# File 'vulns/match.rb', line 526

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:



280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'vulns/match.rb', line 280

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:



127
128
129
 
# File 'vulns/match.rb', line 127

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:



231
232
233
234
235
236
237
238
239
 
# File 'vulns/match.rb', line 231

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:



427
428
429
430
431
432
433
434
435
 
# File 'vulns/match.rb', line 427

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

#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:



388
389
390
391
392
393
394
395
396
 
# File 'vulns/match.rb', line 388

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:



173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
 
# File 'vulns/match.rb', line 173

def each_advisory_batch(formulae, &_blk)
  formulae.each_slice(BULK_CHUNK) do |chunk|
    identities = chunk.map { |f| [f, identify(f)] }
    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))

    identities.each do |f, identity|
      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:



463
464
465
466
467
468
469
470
471
472
 
# File 'vulns/match.rb', line 463

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:



414
415
416
 
# File 'vulns/match.rb', line 414

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 FormulaVersions 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>}.
  • a pkg_version String when the walk hits :affected, or when it stops at an unloadable revision (best-effort boundary; the reviewer can tighten).

The rev-list and per-revision loads are cached per formula.

Parameters:

Returns:



583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
 
# File 'vulns/match.rb', line 583

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

  fv = @formula_versions[formula.name] ||= FormulaVersions.new(formula)
  revs = @formula_rev_lists[formula.name] ||=
    [].tap { |a| fv.rev_list("HEAD") { |rev, entry| a << [rev, entry] } }

  last_fixed = T.let(formula.pkg_version.to_s, String)
  revs.each do |rev, entry|
    state = fv.formula_at_revision(rev, entry) do |old|
      [aggregate_state_at(old, hit), old.pkg_version.to_s]
    end
    # `nil` means the revision failed to load; can't verify further.
    return last_fixed if state.nil?

    aggregate, pkg_version = state
    return :never_affected if aggregate == :not_applicable
    return last_fixed if aggregate != :fixed

    last_fixed = pkg_version
  end
  :never_affected
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:



207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'vulns/match.rb', line 207

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:



132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'vulns/match.rb', line 132

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:



419
420
421
422
423
424
 
# File 'vulns/match.rb', line 419

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:



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

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:



405
406
407
408
409
410
411
 
# File 'vulns/match.rb', line 405

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) ⇒ 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 :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:

Returns:



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

def range_status(hit)
  results = hit.evidence.filter_map do |ev|
    status = evidence_range_status(ev, ev.subject_version)
    [status, ev] if status
  end
  return if results.empty?

  results.find { |s, _| s.affected? } ||
    results.find { |s, _| s.fixed? } ||
    results.first
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:



122
123
124
 
# File 'vulns/match.rb', line 122

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:



352
353
354
355
356
357
358
359
360
361
362
363
364
 
# File 'vulns/match.rb', line 352

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:



316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
 
# File 'vulns/match.rb', line 316

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(formula, resource) ⇒ Version?

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:



628
629
630
631
632
633
634
635
636
637
638
 
# File 'vulns/match.rb', line 628

def subject_version(formula, resource)
  if resource
    begin
      formula.resource(resource)&.version
    rescue ResourceMissingError
      nil
    end
  else
    formula.version
  end
end

#to_brew_record(formula, hit, first_fixed: 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). 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)
  • now (Time) (defaults to: Time.now.utc)

Returns:



488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
 
# File 'vulns/match.rb', line 488

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

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

  record = T.let({
    schema_version:    OsvExport::SCHEMA_VERSION,
    id:                "#{OsvExport::ID_PREFIX}-#{formula.name}-#{hit.canonical_id}",
    published:         timestamp,
    modified:          timestamp,
    upstream:          vuln.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