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, HistoricalObservation, Hit, Identity, Interval, ReconciledHistory, ResourceRecord

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_once, #opoo_outside_github_actions, #opoo_without_github_actions_annotation, #pretty_cannot_install, #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, strict_upstream: 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)
  • strict_upstream (Boolean) (defaults to: false) 


225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'vulns/match.rb', line 225

def initialize(repology: nil, cpan_sec: nil, overrides: nil, bulk: false, strict_upstream: false)
  @repology = repology
  @cpan_sec = cpan_sec
  @overrides = overrides
  @bulk = bulk
  @strict_upstream = strict_upstream
  @vuln_cache = T.let({}, T::Hash[String, T.nilable(Vulnerability)])
  @vuln_errors = T.let({}, T::Hash[String, OSV::Error])
  @history = T.let(History.new, History)
  @reconciliation_observations = T.let(
    {}.compare_by_identity,
    T::Hash[Formula, T::Hash[[T::Array[String], T::Array[Evidence]], HistoricalObservation]],
  )
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 reached by any strategy, grouped by its connected id/alias family. 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:



303
304
305
306
307
 
# File 'vulns/match.rb', line 303

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:



870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
 
# File 'vulns/match.rb', line 870

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:



1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
 
# File 'vulns/match.rb', line 1150

def aggregate_state_at(formula, hit)
  # 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.
  subjects = hit.evidence.reject { |ev| ev.subject_version.nil? }.group_by(&:resource)
  results = subjects.flat_map do |_, evidence|
    states = evidence.map do |ev|
      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 :absent unless present
      next :unknown if subject.nil?
      next :prerelease_boundary if evidence_prerelease_boundary?(ev, subject)

      evidence_range_status(ev, subject)&.state || :unknown
    end
    # As in `range_status`, a source that cannot be compared (such as a
    # commit-only GIT range) does not hide a comparison of the same
    # subject from another source. Other subjects still fail closed.
    states -= [:unknown] if states.intersect?([:affected, :fixed, :not_applicable])
    states.map { |state| (state == :absent) ? :fixed : state }
  end
  return if results.empty? || results.include?(:prerelease_boundary)
  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:



422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
 
# File 'vulns/match.rb', line 422

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:



856
857
858
859
860
861
862
 
# File 'vulns/match.rb', line 856

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:



494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
 
# File 'vulns/match.rb', line 494

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:



249
250
251
 
# File 'vulns/match.rb', line 249

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:



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

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

#current_prerelease_boundary?(hit) ⇒ Boolean

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.

Reuse the same ambiguity rule as historical observations for every current subject, including evidence retained from other aliases.

Parameters:

Returns:

  • (Boolean) 


749
750
751
 
# File 'vulns/match.rb', line 749

def current_prerelease_boundary?(hit)
  hit.evidence.any? { |ev| evidence_prerelease_boundary?(ev, ev.subject_version) }
end

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



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
698
699
700
701
702
703
 
# File 'vulns/match.rb', line 666

def dedup_by_aliases(hits)
  groups = T.let([], T::Array[T::Array[Hit]])
  pending = hits.dup
  until pending.empty?
    first = pending.shift
    raise ArgumentError, "Cannot start an empty alias group" if first.nil?

    # Evidence provenance can legitimately lead to several vulnerabilities,
    # so only the vulnerability's own identifiers connect hits.
    identifiers = T.let({}, T::Hash[String, T::Boolean])
    first.vulnerability.identifiers.each { |identifier| identifiers[identifier] = true }
    group = T.let([first], T::Array[Hit])
    loop do
      connected, remaining = pending.partition do |hit|
        hit.vulnerability.identifiers.any? { |identifier| identifiers.key?(identifier) }
      end
      break if connected.empty?

      connected.each do |hit|
        hit.vulnerability.identifiers.each { |identifier| identifiers[identifier] = true }
      end
      group.concat(connected)
      pending = remaining
    end
    groups << group
  end

  groups.map do |group|
    next group.fetch(0) if group.one?

    # Members of one alias family usually share a `canonical_id`, so the
    # record id is what actually settles the order for most groups. Rank
    # once so the merged evidence order is fixed too, not just the primary.
    ranked = group.sort_by { |h| [-STRATEGY_PRECISION.fetch(h.strategy), h.canonical_id, h.vulnerability.id] }
    Hit.new(vulnerability: ranked.fetch(0).vulnerability,
            evidence:      ranked.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:



606
607
608
609
610
611
612
613
614
 
# File 'vulns/match.rb', line 606

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, on_error: nil, &_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:



323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
 
# File 'vulns/match.rb', line 323

def each_advisory_batch(formulae, on_error: nil, &_blk)
  e = T.let(nil, T.nilable(OSV::Error))
  formulae.each_slice(BULK_CHUNK) do |chunk|
    identities = T.let({}, T::Hash[Formula, Identity])
    chunk.each do |formula|
      next if 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?
      begin
        responses = OSV.query_batch(labelled.map(&:first))
      rescue OSV::Error => e
        raise unless on_error

        chunk.each { |formula| on_error.call(formula, e) }
        next
      end
      responses.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?

      begin
        hits = hits_from(by_formula[f] || {}, identity)
      rescue OSV::Error => e
        raise unless on_error

        on_error.call(f, e)
        next
      end
      yield f, hits
    end
  end
end

#evidence_prerelease_boundary?(evidence, version) ⇒ Boolean

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.

CPANSA uses its own constraints rather than the attached OSV ranges.

Parameters:

Returns:

  • (Boolean) 


755
756
757
758
759
 
# File 'vulns/match.rb', line 755

def evidence_prerelease_boundary?(evidence, version)
  return false if evidence.strategy == :cpansa || version.nil?

  evidence.source_record&.prerelease_boundary?(evidence.ecosystem, evidence.name, version) || false
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:



765
766
767
768
769
770
771
772
773
774
 
# File 'vulns/match.rb', line 765

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:



645
646
647
648
649
650
651
652
653
 
# File 'vulns/match.rb', line 645

def fetch_vulnerability(id)
  error = @vuln_errors[id]
  raise error if error

  @vuln_cache.fetch(id) { @vuln_cache[id] = load_vulnerability(id) }
rescue OSV::Error => e
  @vuln_errors[id] = e
  raise
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. This includes a version that moved backwards, leaving no later fixed version above the affected one.
  • :shared_fixed_version when the current pkg_version also has an affected build, so no released version is fixed yet.
  • a pkg_version String when the walk hits :affected. A fix that changed a resource without a revision bump shares its version with affected builds, which the version cannot distinguish, so the boundary is the next version with only fixed builds.

Parameters:

Returns:



919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
 
# File 'vulns/match.rb', line 919

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

  fixed_versions = T.let([formula.pkg_version], T::Array[PkgVersion])
  result = @history.walk(formula) do |old|
    aggregate = aggregate_state_at(old, hit)
    version = old.pkg_version
    case aggregate
    when :fixed
      fixed_versions << version if fixed_versions.last != version
      nil
    when :affected
      later = fixed_versions.take_while { |fixed| fixed != version }
      boundary = later.last
      if boundary.nil?
        :shared_fixed_version
      elsif boundary > version
        boundary.to_s
      else
        :history_unavailable
      end
    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_introduced_version(formula, hit, first_fixed: nil) ⇒ String, 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.

Return the earliest affected pkg_version for a new record. Check all history: one interval must cover every affected version and no version with only non-affected builds. A resource change without a revision bump leaves the shared version affected if any of its builds were affected. Unreadable history and unrepresentable intervals require manual review.

Parameters:

Returns:



1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
 
# File 'vulns/match.rb', line 1073

def first_introduced_version(formula, hit, first_fixed: nil)
  current_state = aggregate_state_at(formula, hit)
  return :history_unavailable unless [:affected, :fixed].include?(current_state)

  affected_versions = T.let([], T::Array[PkgVersion])
  unaffected_versions = T.let([], T::Array[PkgVersion])
  ((current_state == :affected) ? affected_versions : unaffected_versions) << formula.pkg_version
  result = @history.walk(formula) do |old|
    case aggregate_state_at(old, hit)
    when :affected then affected_versions << old.pkg_version
    when :fixed, :not_applicable then unaffected_versions << old.pkg_version
    else next :history_unavailable
    end
    nil
  end
  return :history_unavailable unless result.nil?

  introduced = affected_versions.min
  return :history_unavailable unless introduced

  fixed = PkgVersion.parse(first_fixed) if first_fixed
  return :history_unavailable if fixed && affected_versions.any? { |version| version >= fixed }

  unaffected_versions -= affected_versions
  if unaffected_versions.any? { |version| version >= introduced && (!fixed || version < fixed) }
    return :history_unavailable
  end

  introduced.to_s
rescue ArgumentError
  :history_unavailable
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. :history_unavailable means a revision could not be loaded or compared, or the formula's history is missing or shallow. :not_reintroduced means no representable transition was found in the available history.

Parameters:

Returns:



1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
 
# File 'vulns/match.rb', line 1117

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 :history_unavailable 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 :history_unavailable
    end
    nil
  end
  return result unless result.nil?

  transition_found ? first_affected : :not_reintroduced
end

#history_load_failures ⇒ Array<History::LoadFailure>

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:



241
 
# File 'vulns/match.rb', line 241

def history_load_failures = @history.load_failures

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



380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
 
# File 'vulns/match.rb', line 380

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 = begin
            fetch_vulnerability(cve)
          rescue OSV::NotFoundError
            nil
          end
          record ||= 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_aliases(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:



254
255
256
257
258
259
260
261
262
263
264
265
266
267
 
# File 'vulns/match.rb', line 254

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:   primary_registry_package(formula),
    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:



656
657
658
659
660
661
662
663
 
# File 'vulns/match.rb', line 656

def load_vulnerability(id)
  Vulnerability.new(OSV.vulnerability(id))
rescue OSV::Error => e
  raise if @strict_upstream

  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:



589
590
591
592
593
594
595
596
597
598
599
600
 
# File 'vulns/match.rb', line 589

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:



623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
 
# File 'vulns/match.rb', line 623

def prefetch_vulnerabilities(ids)
  missing = ids.uniq.reject { |id| @vuln_cache.key?(id) || @vuln_errors.key?(id) }
  missing.each_slice(MAX_VULN_FETCH_THREADS) do |slice|
    threads = slice.map do |id|
      [id, Thread.new do
        load_vulnerability(id)
      rescue OSV::Error => e
        e
      end]
    end
    threads.each do |id, thread|
      result = thread.value
      if result.is_a?(OSV::Error)
        @vuln_errors[id] = result
      else
        @vuln_cache[id] = result
      end
    end
  end
end

#preserve_homebrew_ranges?(formula_name, identifiers) ⇒ Boolean

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:

  • (Boolean) 


484
485
486
 
# File 'vulns/match.rb', line 484

def preserve_homebrew_ranges?(formula_name, identifiers)
  !!@overrides&.preserve_homebrew_ranges?(formula_name, identifiers)
end

#primary_registry_package(formula, strict: true) ⇒ Identify::RegistryPackage?

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 formula's source-derived registry package, unless a reviewed override supplies an identity for formulae built from non-registry URLs. Historical conflicts are uncheckable so one stale override cannot abort matching every formula.

Parameters:

  • formula (Formula)
  • strict (Boolean) (defaults to: true)

Returns:



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

def primary_registry_package(formula, strict: true)
  derived = Identify.registry_package(formula.stable&.url)
  override = @overrides&.registry_package_override(formula.name)
  return derived unless override

  if derived
    if derived.ecosystem != override.ecosystem || derived.name != override.name
      raise AdvisoryOverrides::Error, "#{formula.name}.registry_package conflicts with its source URL" if strict

      return
    end
    return derived
  end

  version = formula.stable&.version&.to_s
  return if version.nil?

  Identify.registry_package_for(ecosystem: override.ecosystem, name: override.name, version:)
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. A current prerelease ambiguity needs an explicit reviewed state.

Parameters:

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

Returns:



719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
 
# File 'vulns/match.rb', line 719

def range_status(hit, formula_name: nil)
  override = @overrides&.advisory_override(formula_name, hit.identifiers) if formula_name
  return if !override&.state && current_prerelease_boundary?(hit)

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

#reconcile_history(formula, hit, require_resource_presence: false, reuse_platforms: false) ⇒ ReconciledHistory

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.

Reconstruct every representable affected interval from complete history. Callers can emit new candidates or reconcile matched records; generated patch-fix records must retain their annotation-based ranges. Unlike first_fixed_version, this must visit builds before the first transition, including below-introduced versions and removed resources. An unresolved or protected result never supplies replacement boundaries.

Parameters:

  • formula (Formula)
  • hit (Hit)
  • require_resource_presence (Boolean) (defaults to: false)
  • reuse_platforms (Boolean) (defaults to: false)

Returns:



958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
 
# File 'vulns/match.rb', line 958

def reconcile_history(formula, hit, require_resource_presence: false, reuse_platforms: false)
  if preserve_homebrew_ranges?(formula.name, hit.identifiers)
    return ReconciledHistory.new(state: :preserved, reasons: [])
  end

  unseen = require_resource_presence ? hit.evidence.select(&:resource).dup : []
  # Matching metadata alone does not identify an observation: constraints
  # and the alias family (including patch annotations) must also agree.
  observation_key = [hit.identifiers, hit.evidence]
  observe = lambda do |build|
    unseen.reject! do |ev|
      present, version = subject_version_at(build, ev)
      present && !version.nil?
    end
    if reuse_platforms && @history.platform_independent?(build)
      cached = @reconciliation_observations[build] ||= {}
      cached[observation_key] ||= reconciliation_observation(build, hit)
    else
      reconciliation_observation(build, hit)
    end
  end
  observations = [observe.call(formula)]
  result = @history.walk(formula, complete: true, reuse_platforms:) do |old|
    observations << observe.call(old)
    nil
  end
  reasons = observations.flat_map(&:reasons)
  reasons << :resource_not_found if unseen.any?
  reasons << :history_unavailable unless result.nil?
  if @overrides&.advisory_override(formula.name, hit.identifiers)
    reasons << :upstream_override_requires_review
  end

  intervals = T.let([], T::Array[Interval])
  observations.sort_by(&:version).chunk_while { |left, right| left.version == right.version }.each do |builds|
    version = builds.fetch(0).version
    previous = intervals.last
    if builds.any? { |build| build.state == :affected }
      intervals << Interval.new(introduced: version.to_s) if previous.nil? || previous.fixed
    elsif previous && previous.fixed.nil?
      intervals[-1] = previous.with(fixed: version.to_s)
    end
  end
  if intervals.last&.fixed.nil? && intervals.any? &&
     observations.none? { |row| row.version == formula.pkg_version && row.state == :affected }
    reasons << :unrepresentable_interval
  end

  if observations.any? do |row|
    row.patched && (intervals.empty? || intervals.any? do |interval|
      fixed = interval.fixed
      row.version >= PkgVersion.parse(interval.introduced) && (!fixed || row.version <= PkgVersion.parse(fixed))
    end)
  end
    reasons << :advisory_patch_requires_review
  end

  return ReconciledHistory.new(state: :unresolved, reasons: reasons.uniq.sort) if reasons.any?
  return ReconciledHistory.new(state: :never_affected, reasons: []) if intervals.empty?

  ReconciledHistory.new(state: :range, intervals:, reasons: [])
end

#reconciliation_observation(formula, hit) ⇒ HistoricalObservation

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.

Preserve disagreements within an upstream alias family and unknown subjects even when another subject is affected. Both prevent a history rewrite, although ordinary matching can still report the affected hit.

Parameters:

Returns:



1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
 
# File 'vulns/match.rb', line 1025

def reconciliation_observation(formula, hit)
  reasons = T.let([], T::Array[Symbol])
  subjects = hit.evidence.reject { |ev| ev.strategy == :distro && ev.subject_version.nil? }
                .group_by { |ev| [ev.ecosystem, ev.name, ev.resource] }.map do |_, evidence|
    states = evidence.map do |ev|
      next :unknown if ev.subject_version.nil?

      present, version = subject_version_at(formula, ev)
      next(ev.resource ? :fixed : :subject_changed) unless present
      next :prerelease_boundary if evidence_prerelease_boundary?(ev, version)

      evidence_range_status(ev, version)&.state || :unknown
    end.uniq
    reasons << :uncomparable_source if states.include?(:unknown)
    reasons << :subject_changed if states.include?(:subject_changed)
    reasons << :prerelease_boundary if states.include?(:prerelease_boundary)
    if states.include?(:affected) && states.intersect?([:fixed, :not_applicable])
      reasons << :conflicting_upstream
    end
    states
  end.flatten
  reasons << :uncomparable_source if subjects.empty?
  state = if subjects.include?(:affected)
    :affected
  elsif subjects.intersect?([:unknown, :subject_changed, :prerelease_boundary]) || subjects.empty?
    :unknown
  elsif subjects.include?(:fixed)
    :fixed
  else
    :not_applicable
  end

  resolved = Scanner.resolved_ids(formula.serialized_patches)
  formula.resources.each do |resource|
    resource.patches.each do |patch|
      resolved.concat(patch.resolves.map(&:upcase)) if patch.is_a?(ExternalPatch) || patch.is_a?(LocalPatch)
    end
  end
  HistoricalObservation.new(version: formula.pkg_version, state:,
                            patched: resolved.intersect?(hit.identifiers.map(&:upcase)), reasons:)
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:



844
845
846
 
# File 'vulns/match.rb', line 844

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:



849
850
851
 
# File 'vulns/match.rb', line 849

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:



244
245
246
 
# File 'vulns/match.rb', line 244

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:



566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
 
# File 'vulns/match.rb', line 566

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) : []
  rescue OSV::NotFoundError
    # A dangling upstream link supplies no record. Keep the cached error
    # so fetching this ID directly from query results still fails closed.
    []
  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:



530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
 
# File 'vulns/match.rb', line 530

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

#resource_advisories_for(records) ⇒ 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.

Rediscover stored resource subjects using fresh versionless queries. Stored versions only reproduce the candidate's range basis. The history walk must establish every version and witness each resource's identity.

Parameters:

Returns:



461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
 
# File 'vulns/match.rb', line 461

def resource_advisories_for(records)
  evidence = records.flat_map(&:evidence).uniq
  return [] if evidence.empty?

  subjects = evidence.group_by { |ev| [ev.ecosystem, ev.name] }
  packages = subjects.keys.map { |ecosystem, name| { ecosystem:, name:, version: nil } }
  results = OSV.query_batch(packages)
  id_evidence = T.let({}, T::Hash[String, T::Array[Evidence]])
  subjects.values.zip(results).each do |rows, stubs|
    Array(stubs).each do |stub|
      id = stub["id"]
      next unless id.is_a?(String)

      (id_evidence[id] ||= []).concat(rows)
    end
  end
  prefetch_vulnerabilities(id_evidence.keys)
  identity = Identity.new(resource_packages: [], distro_packages: {})
  hits = dedup_by_aliases(resolve_upstream(id_evidence, identity))
  hits.select { |hit| records.any? { |record| record.upstream.intersect?(hit.identifiers) } }
end

#skip_formula?(formula_name) ⇒ Boolean

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:

  • (Boolean) 


489
490
491
 
# File 'vulns/match.rb', line 489

def skip_formula?(formula_name)
  !!@overrides&.skip_formula?(formula_name)
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:



1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
 
# File 'vulns/match.rb', line 1189

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 = primary_registry_package(formula, strict: false)
    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_introduced: nil, history: 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_introduced is the first affected Homebrew version (from #first_introduced_version or #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. With history, emit all reconstructed intervals and derive fix metadata from the last interval, including a final open interval.

Parameters:

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

Returns:



794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
 
# File 'vulns/match.rb', line 794

def to_brew_record(formula, hit, first_fixed: nil, first_introduced: nil, history: 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)

  if history && (history.state != :range || history.intervals.empty? ||
                 history.intervals.any? { |interval| !interval.valid? })
    raise ArgumentError, "Candidate history must contain valid affected intervals"
  end

  fixed = history ? history.fixed : first_fixed
  fixed ||= formula.pkg_version.to_s if !history && status&.fixed?
  events = T.let(history ? history.events : [{ introduced: first_introduced || "0" }],
                 T::Array[T::Hash[Symbol, String]])
  events << { fixed: } if !history && 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 }.uniq,
    },
  }, T::Hash[Symbol, T.untyped])

  if history
    record.fetch(:affected).fetch(0).fetch(:ecosystem_specific)[:range_state] = fixed ? "fixed" : "affected"
  end

  if status.nil? && current_prerelease_boundary?(hit)
    record[:database_specific][:review_reason] = "prerelease_boundary"
  end

  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