Add Advisory History

Signed-off-by: Sampurna Pyne <sampurnapyne1710@gmail.com>

Author: Samk1710

vulnerabilities/templates/advisory_detail.html

@@ -13,6 +13,13 @@
 
 {% if advisoryv2 %}
     <section class="section pt-4">
+        {% if is_snapshot %}
+        <div class="notification is-warning mb-3 has-text-centered" style="max-width: fit-content; margin: 0 auto;">
+            <i class="fa fa-history mr-1"></i> Historical snapshot collected on <strong>{{ advisoryv2.date_collected|date:"Y-m-d H:i" }} UTC</strong>.
+            <a href="{% url 'advisory_package_details' avid=advisoryv2.avid %}" class="ml-3"><strong>View current version</strong> <i class="fa fa-arrow-right is-size-7"></i></a>
+        </div>
+        {% endif %}
+
         <div class="details-container">
             <article class="panel is-info panel-header-only">
                 <div class="panel-heading py-2 is-size-6">

vulnerabilities/templates/advisory_package_details.html

@@ -0,0 +1,118 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from django import template
+from packageurl import PackageURL
+
+register = template.Library()
+
+
+@register.filter
+def format_diff_for_ui(changes) -> dict:
+    """
+    Example:
+        {
+            'affected_packages': {
+                'added': [{'package': {'type': 'pypi', 'name': 'requests', 'version': '2.25.0'}, 'affected_version_range': '==2.25.0', 'fixed_version_range': '==2.25.1'}],
+                'removed': [{'package': {'type': 'pypi', 'name': 'requests', 'version': '2.24.0'}, 'affected_version_range': '==2.24.0', 'fixed_version_range': '==2.24.1'}]
+            }
+        }
+    should result in:
+            {
+                'Affected Packages': {
+                    'added': [{'header': 'Affected package', 'attributes': [('PURL', 'pkg:pypi/requests@2.25.0'), ('Affected version', '==2.25.0'), ('Fixed Version', '==2.25.1')] }],
+                    'removed': [{'header': 'Affected package', 'attributes': [('PURL', 'pkg:pypi/requests@2.24.0'), ('Affected version', '==2.24.0'), ('Fixed Version', '==2.24.1')] }]
+                }
+            }
+    """
+    formatted = {}
+
+    for field, change in changes.items():
+        label = field.replace("_", " ").title()
+
+        if "old" in change or "new" in change:
+            formatted[label] = change
+            continue
+
+        formatted[label] = {"added": [], "removed": []}
+
+        for change_type in ["added", "removed"]:
+            for item in change.get(change_type, []):
+                if field == "affected_packages":
+                    attributes = []
+
+                    if package := item.get("package"):
+                        package_string = (
+                            str(PackageURL(**package))
+                            if isinstance(package, dict)
+                            else str(package)
+                        )
+                        attributes.append(("PURL", package_string))
+
+                    if affected_version_range := item.get("affected_version_range"):
+                        attributes.append(("Affected version", affected_version_range))
+
+                    if fixed_version_range := item.get("fixed_version_range"):
+                        attributes.append(("Fixed Version", fixed_version_range))
+
+                    formatted[label][change_type].append(
+                        {"header": "Affected package", "attributes": attributes}
+                    )
+
+                elif field == "references":
+                    attributes = []
+
+                    if reference_url := item.get("url"):
+                        attributes.append(("URL", reference_url))
+
+                    if reference_id := item.get("reference_id"):
+                        attributes.append(("ID", reference_id))
+
+                    formatted[label][change_type].append(
+                        {"header": "Reference", "attributes": attributes}
+                    )
+
+                elif field == "severities":
+                    attributes = []
+
+                    if scoring_system := item.get("system") or item.get("scoring_system"):
+                        scoring_system_string = str(scoring_system)
+                        if "cvss" in scoring_system_string.lower():
+                            scoring_system_string = scoring_system_string.upper()
+                        else:
+                            scoring_system_string = scoring_system_string.replace("_", " ").title()
+                        attributes.append(("System", scoring_system_string))
+
+                    if severity_value := item.get("value"):
+                        attributes.append(("Value", severity_value))
+
+                    if scoring_elements := item.get("scoring_elements"):
+                        attributes.append(("Elements", scoring_elements))
+
+                    formatted[label][change_type].append(
+                        {"header": "Severity", "attributes": attributes}
+                    )
+
+                elif field == "patches":
+                    attributes = []
+
+                    if patch_url := (item.get("url") or item.get("repository")):
+                        attributes.append(("URL", patch_url))
+
+                    if patch_commit := item.get("commit"):
+                        attributes.append(("Commit", patch_commit))
+
+                    formatted[label][change_type].append(
+                        {"header": "Patch", "attributes": attributes}
+                    )
+
+                else:
+                    formatted[label][change_type].append(item)
+
+    return formatted

vulnerabilities/templatetags/diff_advisory_history.py

@@ -691,6 +691,36 @@ def compute_content_id(advisory_data):
     return content_id
 
 
+def get_normalized_advisory_v2(advisory_data):
+    """
+    Return a normalized dictionary of advisory data for diffing history.
+
+    :param advisory_data: An AdvisoryV2 object
+    :return: A normalized dictionary of advisory data
+    """
+    from vulnerabilities.models import AdvisoryV2
+
+    if not isinstance(advisory_data, AdvisoryV2):
+        raise ValueError("Unsupported advisory data type for history diffing")
+
+    advisory_data = advisory_data.to_advisory_data()
+
+    normalized_data = {
+        "advisory_id": advisory_data.advisory_id,
+        "aliases": advisory_data.aliases,
+        "summary": advisory_data.summary,
+        "affected_packages": [pkg.to_dict() for pkg in advisory_data.affected_packages if pkg],
+        "references": [ref.to_dict() for ref in advisory_data.references if ref],
+        "severities": [sev.to_dict() for sev in advisory_data.severities if sev],
+        "weaknesses": advisory_data.weaknesses,
+        "patches": [patch.to_dict() for patch in advisory_data.patches],
+        "url": advisory_data.url,
+        # "date_published": str(advisory_data.date_published) if advisory_data.date_published else None,
+    }
+
+    return normalized_data
+
+
 def compute_content_id_v2(advisory_data):
     """
     Compute a unique content_id for an advisory by normalizing its data and hashing it.
@@ -1129,3 +1159,72 @@ def safe_altcha_redirect(next_url: str) -> redirect:
         return redirect(next_url)
 
     return redirect("/")
+
+
+def exclude_epss_severities(severities: list) -> list:
+    """
+    Return a copy of the severities list with all EPSS entries removed.
+    EPSS scores are updated daily and can be tracked separately in EPSS tab.
+    """
+    return [
+        severity
+        for severity in (severities or [])
+        if str(severity.get("system", "")).lower() != "epss"
+    ]
+
+
+def diff_advisories_v2(old_data: dict, new_data: dict) -> dict:
+    """
+    Diff two normalised advisories and return the changes.
+    """
+    changes = {}
+    all_keys = set(old_data.keys()).union(new_data.keys())
+
+    for field in all_keys:
+        old_field_value = old_data.get(field)
+        new_field_value = new_data.get(field)
+
+        if isinstance(old_field_value, list) or isinstance(new_field_value, list):
+            old_items_list = old_field_value or []
+            new_items_list = new_field_value or []
+
+            if field == "severities":
+                old_items_list = exclude_epss_severities(old_items_list)
+                new_items_list = exclude_epss_severities(new_items_list)
+
+            if all(isinstance(item, str) for item in old_items_list + new_items_list):
+                # "aliases", "weaknesses"
+                old_items_set = set(old_items_list)
+                new_items_set = set(new_items_list)
+                if old_items_set != new_items_set:
+                    changes[field] = {
+                        "added": sorted(new_items_set - old_items_set),
+                        "removed": sorted(old_items_set - new_items_set),
+                    }
+            elif all(isinstance(item, dict) for item in old_items_list + new_items_list):
+                # "references", "affected_packages", "patches", "severities"
+                old_items_set = {
+                    json.dumps(item, sort_keys=True, default=str) for item in old_items_list
+                }
+                new_items_set = {
+                    json.dumps(item, sort_keys=True, default=str) for item in new_items_list
+                }
+                if old_items_set != new_items_set:
+                    changes[field] = {
+                        "added": [
+                            json.loads(item) for item in sorted(new_items_set - old_items_set)
+                        ],
+                        "removed": [
+                            json.loads(item) for item in sorted(old_items_set - new_items_set)
+                        ],
+                    }
+        elif (
+            isinstance(old_field_value, str)
+            or isinstance(new_field_value, str)
+            or (old_field_value is None and new_field_value is None)
+        ):
+            # "url", "date_published", "summary"
+            if old_field_value != new_field_value:
+                changes[field] = {"old": old_field_value, "new": new_field_value}
+
+    return changes