resf/distro-tools
1
0
Fork 0
mirror of https://github.com/resf/distro-tools.git synced 2024-11-21 20:51:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Properly report all arches and RPMs

Browse Source
This commit is contained in:
Mustafa Gezen 2023-02-05 02:14:58 +01:00
parent 47179f62f5
commit ba9c8c2f05
Signed by untrusted user who does not match committer: mustafa
GPG Key ID: DCDF010D946438C1
1 changed files with 52 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
apollo/server/routes/api_osv.py
View File

@ -9,7 +9,7 @@ from slugify import slugify
from apollo.db import Advisory from apollo.db import Advisory
from apollo.db.advisory import fetch_advisories from apollo.db.advisory import fetch_advisories
from apollo.rpmworker.repomd import EPOCH_RE, NVRA_RE from apollo.rpmworker.repomd import EPOCH_RE, NEVRA_RE
from apollo.server.settings import UI_URL, get_setting from apollo.server.settings import UI_URL, get_setting
from common.fastapi import Params, to_rfc3339_date from common.fastapi import Params, to_rfc3339_date
@ -104,95 +104,75 @@ class OSVAdvisory(BaseModel):
def to_osv_advisory(ui_url: str, advisory: Advisory) -> OSVAdvisory: def to_osv_advisory(ui_url: str, advisory: Advisory) -> OSVAdvisory:
affected_pkgs = [] affected_pkgs = []
vendors = []
pkg_name_map = {} pkg_name_map = {}
for pkg in advisory.packages: for pkg in advisory.packages:
if pkg.supported_product.vendor not in vendors:
vendors.append(pkg.supported_product.vendor)
nevra = NEVRA_RE.search(pkg.nevra)
name = nevra.group(1)
arch = nevra.group(5).lower()
product_name = slugify(pkg.product_name) product_name = slugify(pkg.product_name)
if pkg.supported_products_rh_mirror: if pkg.supported_products_rh_mirror:
product_name = f"{slugify(pkg.supported_product.variant)}:{pkg.supported_products_rh_mirror.match_major_version}" product_name = f"{slugify(pkg.supported_product.variant)}:{pkg.supported_products_rh_mirror.match_major_version}"
if product_name not in pkg_name_map: if product_name not in pkg_name_map:
pkg_name_map[product_name] = {} pkg_name_map[product_name] = {}
if pkg.package_name not in pkg_name_map[product_name]: if arch not in pkg_name_map[product_name]:
pkg_name_map[product_name][pkg.package_name] = [] pkg_name_map[product_name][arch] = {}
if name not in pkg_name_map[product_name][arch]:
pkg_name_map[product_name][arch][name] = []
pkg_name_map[product_name][pkg.package_name].append(pkg) pkg_name_map[product_name][arch][name].append((pkg, nevra))
vendors = [] for product_name, arches in pkg_name_map.items():
for product_name, pkgs in pkg_name_map.items(): for arch, affected_arches in arches.items():
for pkg_name, affected_packages in pkgs.items(): if not affected_arches:
if not affected_packages:
continue continue
first_pkg = None for pkg_name, affected_packages in affected_arches.items():
noarch_pkg = None for pkg in affected_packages:
arch = None x = pkg[0]
nvra = None nevra = pkg[1]
ver_rel = None
for x in affected_packages:
if x.supported_product.vendor not in vendors:
vendors.append(x.supported_product.vendor)
nvra = NVRA_RE.search(EPOCH_RE.sub("", x.nevra))
if not nvra:
continue
ver_rel = f"{nvra.group(2)}-{nvra.group(3)}"
if x.supported_products_rh_mirror:
first_pkg = x
arch = x.supported_products_rh_mirror.match_arch
break
arch = nvra.group(4).lower()
if arch == "src": ver_rel = f"{nevra.group(3)}-{nevra.group(4)}"
continue slugified = slugify(x.supported_product.variant)
slugified_distro = slugify(x.product_name)
for arch_, _ in arches.items():
slugified_arch = f"-{slugify(arch_)}"
slugified_distro = slugified_distro.replace(
slugified_arch,
"",
)
epoch = nevra.group(2)
if arch == "noarch": purl = f"pkg:rpm/{slugified}/{pkg_name}@{ver_rel}?arch={arch}&distro={slugified_distro}&epoch={epoch}"
noarch_pkg = x
continue
first_pkg = x affected = OSVAffected(
break package=OSVPackage(
ecosystem=product_name,
if not first_pkg and noarch_pkg: name=pkg_name,
first_pkg = noarch_pkg purl=purl,
),
if not ver_rel: ranges=[
continue OSVRange(
type="ECOSYSTEM",
purl = None repo=x.repo_name,
if first_pkg: events=[
slugified = slugify(first_pkg.supported_product.variant) OSVEvent(introduced="0"),
slugified_distro = slugify(first_pkg.product_name) OSVEvent(fixed=ver_rel),
slugified_distro = slugified_distro.replace( ],
f"-{slugify(arch)}", database_specific=OSVRangeDatabaseSpecific(),
"", )
)
purl = f"pkg:rpm/{slugified}/{pkg_name}@{ver_rel}?arch={arch}&distro={slugified_distro}"
affected = OSVAffected(
package=OSVPackage(
ecosystem=product_name,
name=pkg_name,
purl=purl,
),
ranges=[],
versions=[],
ecosystem_specific=OSVEcosystemSpecific(),
database_specific=OSVAffectedDatabaseSpecific(),
)
for x in affected_packages:
ranges = [
OSVRange(
type="ECOSYSTEM",
repo=x.repo_name,
events=[
OSVEvent(introduced="0"),
OSVEvent(fixed=ver_rel),
], ],
database_specific=OSVRangeDatabaseSpecific(), versions=[],
ecosystem_specific=OSVEcosystemSpecific(),
database_specific=OSVAffectedDatabaseSpecific(),
) )
]
affected.ranges.extend(ranges)
affected_pkgs.append(affected) affected_pkgs.append(affected)
references = [ references = [
OSVReference(type="ADVISORY", url=f"{ui_url}/{advisory.name}"), OSVReference(type="ADVISORY", url=f"{ui_url}/{advisory.name}"),