generate_openvex.py

#!/usr/bin/env python3
import xml.etree.ElementTree as ET
import json
from datetime import datetime, timezone

NAMESPACES = {
    'b': 'http://cyclonedx.org/schema/bom/1.6'
}


def _find_element(parent: ET.Element, tag: str) -> ET.Element | None:
    return parent.find(tag, NAMESPACES)


def _find_stripped_text(parent: ET.Element, tag: str) -> str | None:
    el = _find_element(parent, tag)
    return el.text.strip() if el is not None else None


def _add_optional_date(parent: ET.Element, tag: str, target: dict, key: str) -> None:
    el = _find_element(parent, tag)
    if el is not None and el.text:
        try:
            dt = datetime.fromisoformat(el.text.strip()).astimezone(timezone.utc)
            target[key] = dt.isoformat().replace('+00:00', 'Z')
        except ValueError as e:
            raise ValueError(f"Invalid ISO date format in <{tag}>: {el.text}") from e


def load_cyclonedx(path: str = 'VEX.cyclonedx.xml') -> ET.Element:
    return ET.parse(path).getroot()


def to_openvex(root: ET.Element) -> dict:
    serial_number = root.get('serialNumber')
    if not serial_number:
        raise ValueError("CycloneDX BOM must have a 'serialNumber' attribute")

    version = int(root.get('version', '1'))

    result = {
        '@context': 'https://openvex.dev/ns/v0.2.0',
        '@id': f"https://commons.apache.org/security/vex/{serial_number}",
        'author': 'Apache Commons Security Team <security@commons.apache.org>',
        'role': 'Security Team',
        'version': version,
        'tooling': (
            "This document was automatically converted from the `VEX.cyclonedx.xml` file.\n"
            "Do not edit this file directly, run `generate_openvex.py` to regenerate it."
        )
    }

    _add_optional_date(root, 'b:metadata/b:timestamp', result, 'timestamp')

    component = _find_element(root, 'b:metadata/b:component')
    if component is None:
        raise ValueError("Missing <component> in <metadata>")

    product = to_openvex_product(component)

    result['statements'] = [
        to_openvex_statement(vuln, product)
        for vuln in root.findall('.//b:vulnerability', NAMESPACES)
    ]

    return result


def to_openvex_product(component: ET.Element) -> dict:
    purl = _find_element(component, 'b:purl')
    if purl is None or not purl.text:
        raise ValueError("Component must include a non-empty <purl> element")

    return {
        '@id': purl.text,
        'identifiers': {
            'purl': purl.text
        }
    }


def to_openvex_vulnerability(vuln: ET.Element) -> dict:
    cdx_id = _find_stripped_text(vuln, 'b:id')
    if not cdx_id:
        raise ValueError("Vulnerability must have an <id>")

    entry = {'name': cdx_id}

    source = _find_element(vuln, 'b:source')
    if source is not None:
        entry['@id'] = _find_stripped_text(source, 'b:url')

    entry['aliases'] = [
        _find_stripped_text(ref, 'b:id')
        for ref in vuln.findall('b:references/b:reference', NAMESPACES)
    ]

    return entry


def to_openvex_statement(vuln: ET.Element, product: dict) -> dict:
    analysis = _find_element(vuln, 'b:analysis')
    if analysis is None:
        raise ValueError("Missing <analysis> in vulnerability")

    state = _find_stripped_text(analysis, 'b:state')
    if not state:
        raise ValueError("Missing <state> in vulnerability analysis")

    statement = {
        'products': [product],
        'vulnerability': to_openvex_vulnerability(vuln),
        'status': to_openvex_status(state)
    }

    justification = _find_stripped_text(analysis, 'b:justification')
    if justification:
        statement['justification'] = to_openvex_justification(justification)

    detail = _find_stripped_text(analysis, 'b:detail')
    if detail:
        statement['status_notes'] = detail

    remediation = _find_stripped_text(vuln, 'b:recommendation')
    if remediation:
        statement['action_statement'] = remediation
    else:
        if statement['status'] == 'affected':
            raise ValueError("Affected vulnerabilities must have a <recommendation> element")

    _add_optional_date(analysis, 'b:firstIssued', statement, 'timestamp')
    _add_optional_date(analysis, 'b:lastUpdated', statement, 'last_updated')

    return statement


def to_openvex_status(cdx_status: str) -> str:
    mapping = {
        "resolved": "fixed",
        "exploitable": "affected",
        "in_triage": "under_investigation",
        "false_positive": "not_affected",
        "not_affected": "not_affected"
    }
    status = mapping.get(cdx_status.strip().lower())
    if not status:
        raise ValueError(f"Unknown CycloneDX status: '{cdx_status}'")
    return status


def to_openvex_justification(cdx_justification: str) -> str:
    mapping = {
        "code_not_present": "vulnerable_code_not_present",
        "code_not_reachable": "vulnerable_code_not_in_execute_path",
        "requires_configuration": "vulnerable_code_cannot_be_controlled_by_adversary",
        "requires_dependency": "component_not_present",
        "requires_environment": "vulnerable_code_cannot_be_controlled_by_adversary",
        "protected_by_compiler": "inline_mitigations_already_exist",
        "protected_at_runtime": "inline_mitigations_already_exist",
        "protected_by_mitigating_control": "inline_mitigations_already_exist"
    }
    result = mapping.get(cdx_justification.strip().lower())
    if not result:
        raise ValueError(f"Unknown CycloneDX justification: '{cdx_justification}'")
    return result


def main():
    cyclonedx_root = load_cyclonedx()
    openvex_doc = to_openvex(cyclonedx_root)
    with open('openvex.json', 'w') as f:
        json.dump(openvex_doc, f, indent=2)
    print("OpenVEX document written to 'openvex.json'")


if __name__ == "__main__":
    main()