albs-oval-errata-diff/albs_oval_errata_diff/comparer.py

299 lines
12 KiB
Python

"""
package comparer.py implemets difference checking logic
"""
import bz2
import datetime
from pathlib import Path
import re
from typing import Tuple, List, Dict, Any
import logging
import json
import xml.etree.ElementTree as ET
import requests
from .config import Config
from .package import Package
from .advisory import Advisory
def download_oval(url: str, download_dir: Path) -> str:
"""
download_oval downloads, decompreses oval file
and returns filepath of saved file
"""
response = requests.get(url, stream=True, timeout=30)
decompressor = bz2.BZ2Decompressor()
fname = url.split('/')[-1].replace('.bz2', '')
fpath = download_dir / fname
with open(fpath, 'wb') as flw:
for chunk in response.iter_content(chunk_size=128):
flw.write(decompressor.decompress(chunk))
return fpath
def download_errata(url: str, release_version: int, download_dir: Path) -> str:
"""
downloads errata_full.json file end returns file path
"""
response = requests.get(url, stream=True, timeout=30)
fname = f'alma-{release_version}.json'
fpath = download_dir / fname
with open(fpath, 'wb') as errata_file:
for chunk in response.iter_content(chunk_size=128):
errata_file.write(chunk)
return fpath
def extract_id_and_type(string: str) -> Tuple[str, str]:
"""
Extracts advisory id and advisory type from OVAL title or errata id
Example:
oval: "RHSA-2022:5749: .NET 6.0 bugfix update (Moderate)" -> (2022:5749, SA)
errata: ALSA-2022:6165 -> (id=2022:6165, SA)
"""
regexp = r'((RH|AL)(SA|BA|EA))-(\d{4}:\d+)'
res = re.search(regexp, string)
return res.group(4), res.group(3)
def parse_oval(fpath: str, not_before: datetime.datetime) -> Dict[str, Advisory]:
"""
Converts OVAL XML file to Dict of Advisories
"""
def extract_package(title: str) -> Package:
regexp = r'(.*) is earlier than \d+:(.+?(?=-))'
res = re.search(regexp, title)
name = res.group(1)
version = res.group(2)
return Package(name=name, version=version)
tree = ET.parse(fpath)
root = tree.getroot()
namespase = {
'n': 'http://oval.mitre.org/XMLSchema/oval-definitions-5',
}
res = {}
for definition in root.findall('n:definitions/', namespase):
title = definition.find('n:metadata/n:title', namespase).text
issued = definition.find(
'n:metadata/n:advisory/n:issued', namespase).attrib['date']
issued_dt = datetime.datetime.strptime(issued, "%Y-%m-%d")
# we are only interested in RHEL/OVAL SA/BA/EA
# released after RHEL 8.3
if not re.match(r'((RH|AL)(SA|BA|EA))', title) or issued_dt < not_before:
continue
# we are only interested in security based advisories
severity = definition.find(
'n:metadata/n:advisory/n:severity', namespase)
if severity is None:
continue
if severity.text.lower() not in ['low', 'moderate', 'important', 'critical']:
continue
advisory_id, advisory_type = extract_id_and_type(title)
packages = [extract_package(i.attrib['comment']) for
i in definition.findall(".//n:criterion", namespase)
if 'is earlier than' in i.attrib['comment']]
res[advisory_id] = Advisory(title=title, id=advisory_id,
advisory_type=advisory_type,
packages=packages)
return res
def parse_errata(fpath: str) -> Dict[str, Advisory]:
"""
Parses Alma Errata file and converts it to dict of Advisory instances
"""
with open(fpath, 'r', encoding='utf-8') as file_to_load:
erratas = json.load(file_to_load)
res = {}
for errata in erratas['data']:
title = errata['title']
advisory_id, advisory_type = extract_id_and_type(errata['id'])
packages = []
for package in errata['packages']:
full_name = f"{package['name']}-{package['version']}"
if full_name not in packages:
packages.append(full_name)
packages.sort()
res[advisory_id] = Advisory(title=title,
id=advisory_id,
advisory_type=advisory_type,
packages=packages)
return res
def compare(rhel_oval: Dict[str, Advisory],
alma_oval: Dict[str, Advisory],
alma_errata: Dict[str, Advisory],
advisory_exclude: List[str],
packages_exclude: List[str]) -> Tuple[dict, list]:
"""
compares rhel oval with alma oval and alma errata
"""
diff = []
report = {
# total amount of security advisories
'total_advisory_count': 0,
# amount of ALMA advisory that match with RHEL
'good_advisory_count': 0,
# total amount of differencies
'diff_count': 0,
# list of advisories excluded from diff check
'excluded_adv': [],
# list of packages excluded from diff check
'excluded_pkg': [],
# amount of oval advisories that dont exists in oval file
'oval_missing_advisory_count': 0,
# amount of oval advisories that have missing packages
'oval_missing_pkg_advisory_count': 0,
# list of missing oval advisories
'oval_missing_advisory': [],
# list of oval advisories that have missing packages
'oval_missing_pkg_advisory': [],
# amount of advisories that dont exists in errata file
'errata_missing_advisory_count': 0,
# amount of errata advisories that have missing packages
'errata_missing_pkg_advisory_count': 0,
# list of advisories that are missing in errata file
'errata_missing_advisory': [],
# list of errata advisories with missing packages
'errata_missing_pkg_advisory': [],
# total amount of unique missing packages across all alma SA
'missing_packages_unique_count': 0,
# list of unique packages that missing across all alma SA
'missing_packages_unique': []
}
for rhel_advisory_id, rhel_advisory in rhel_oval.items():
report['total_advisory_count'] += 1
advisory_name = f'AL{rhel_advisory.advisory_type}-{rhel_advisory_id}'
# filtering out advisories
if advisory_name in advisory_exclude:
report['excluded_advisory'].append(advisory_name)
continue
# filtefing out packages
packages_to_check: List[Package] = []
for package in rhel_advisory.packages:
if any(package.name == i for i in packages_exclude):
if str(package) not in report['excluded_pkg']:
report['excluded_pkg'].append(str(package))
else:
packages_to_check.append(package)
# check oval
try:
alma_oval_advisory = alma_oval[rhel_advisory_id]
except KeyError:
report['diff_count'] += 1
diff.append({'advisory_name': advisory_name,
'diff': 'advisory is missing in oval'})
report['oval_missing_advisory'].append(advisory_name)
report['oval_missing_advisory_count'] += 1
else:
# check if some packages are missing from OVAL advisories
alma_oval_packages = alma_oval_advisory.packages
alma_oval_missing_packages = [str(r) for r in packages_to_check
if str(r) not in [str(i) for i in alma_oval_packages]]
if alma_oval_missing_packages:
report['diff_count'] += 1
diff_str = f"OVAL advisory has missing packages: {','.join(alma_oval_missing_packages)}"
diff.append({'advisory_name': advisory_name,
'diff': diff_str})
report['oval_missing_pkg_advisory'].append(advisory_name)
report['oval_missing_pkg_advisory_count'] += 1
for missing_package in alma_oval_missing_packages:
if missing_package not in report['missing_packages_unique']:
report['missing_packages_unique'].append(
missing_package)
report['missing_packages_unique_count'] += 1
# check errata
try:
alma_errata_sa = alma_errata[rhel_advisory_id]
except KeyError:
report['errata_missing_advisory'].append(advisory_name)
report['errata_missing_advisory_count'] += 1
report['diff_count'] += 1
diff.append(
{'advisory_name': advisory_name, 'diff': 'advisory is missing in errata'})
continue
# check if some packages are missing from errata SA
alma_errata_packages = alma_errata_sa.packages
alma_errata_missing_packages = \
[str(r) for r in packages_to_check
if str(r) not in [str(i) for i in alma_errata_packages]]
if alma_errata_missing_packages:
report['diff_count'] += 1
diff_str = f"Errata advisory has missing packages: {','.join(alma_errata_missing_packages)}"
diff.append({'advisory_name': advisory_name,
'diff': diff_str})
report['errata_missing_pkg_advisory'].append(advisory_name)
report['errata_missing_pkg_advisory_count'] += 1
for missing_package in alma_errata_missing_packages:
if missing_package not in report['missing_packages_unique']:
report['missing_packages_unique'].append(missing_package)
report['missing_packages_unique_count'] += 1
else:
# if we here, all checks were passed
report['good_advisory_count'] += 1
for item in report.values():
if isinstance(item, list):
item.sort()
return report, diff
# starting point
def comparer_run(config: Config) -> Dict[str, Any]:
"""
comperer_run is the starting point of comparer component
"""
result = {}
for release, urls in config.releases.items():
logging.info('Processing release %i', release)
logging.info('Downloading rhel oval')
rhel_file = download_oval(urls.rhel_oval_url, config.download_dir)
logging.info('Parsing rhel oval')
rhel_oval_dict = parse_oval(rhel_file, config.not_before)
logging.info('Downloading alma oval')
alma_oval_file = download_oval(
urls.alma_oval_url, download_dir=config.download_dir)
logging.info('Parsing alma oval')
alma_oval_dict = parse_oval(alma_oval_file, config.not_before)
logging.info('Downloading alma errata')
alma_errata_file = download_errata(urls.alma_errata_url,
release, config.download_dir)
logging.info('Parsing alma errata')
alma_errata_dict = parse_errata(alma_errata_file)
logging.info('Comparing rhel and alma')
report_release, diff_release = \
compare(rhel_oval_dict,
alma_oval_dict,
alma_errata_dict,
config.advisory_exclude,
config.packages_exclude)
result[release] = {'report': report_release,
'diff': diff_release,
'rhel_oval_url': urls.rhel_oval_url,
'alma_oval_url': urls.alma_oval_url,
'alma_errata_url': urls.alma_errata_url}
result['report_generated'] = datetime.datetime.now().timestamp() * 1000
result['advisory_not_before'] = config.not_before.timestamp() * 1000
return result