Function import_advisory

vulnerabilities/pipes/advisory.py:423–530

Create initial Vulnerability Package relationships for the advisory, including references and severity scores. Package relationships are established only for resolved (concrete) versions.

(
    advisory: Advisory,
    pipeline_id: str,
    confidence: int = MAX_CONFIDENCE,
    logger: Callable = None,
)

@transaction.atomic
def import_advisory(
    advisory: Advisory,
    pipeline_id: str,
    confidence: int = MAX_CONFIDENCE,
    logger: Callable = None,
):
    """
    Create initial Vulnerability Package relationships for the advisory,
    including references and severity scores.

    Package relationships are established only for resolved (concrete) versions.
    """
    from vulnerabilities import import_runner
    from vulnerabilities.improvers import default

    advisory_data: AdvisoryData = advisory.to_advisory_data()
    if logger:
        logger(f"Importing advisory id: {advisory.id}", level=logging.DEBUG)

    affected_purls = []
    fixed_purls = []
    for affected_package in advisory_data.affected_packages:
        package_affected_purls, package_fixed_purls = default.get_exact_purls(
            affected_package=affected_package
        )
        affected_purls.extend(package_affected_purls)
        fixed_purls.extend(package_fixed_purls)

    aliases = get_or_create_aliases(advisory_data.aliases)
    vulnerability = import_runner.get_or_create_vulnerability_and_aliases(
        vulnerability_id=None,
        aliases=aliases,
        summary=advisory_data.summary,
        advisory=advisory,
    )

    if not vulnerability:
        if logger:
            logger(f"Unable to get vulnerability for advisory: {advisory!r}", level=logging.ERROR)
        return

    for ref in advisory_data.references:
        reference = VulnerabilityReference.objects.get_or_none(
            reference_id=ref.reference_id,
            url=ref.url,
        )
        if not reference:
            reference = import_runner.create_valid_vulnerability_reference(
                reference_id=ref.reference_id,
                url=ref.url,
            )
        if reference:
            VulnerabilityRelatedReference.objects.update_or_create(
                reference=reference,
                vulnerability=vulnerability,
            )
        for severity in ref.severities:
            try: