vulncheck/tests/test_eol_sweep_stability.py

"""A repeated EOL sweep over an unchanged inventory must change nothing.

The audit log filled with EOL-MSEXCHANGE-2007 / -2010 / -2016 flipping
open→patched→open, several times per run, on hosts where nothing had changed —
including one running only Exchange Server Subscription Edition, which is fully
supported and should carry no Exchange finding at all.

Two causes, one symptom. Companion packages ("2007 Standard Anti-spam Filter
Updates", "Server 2016 Language Pack - Greek", "Server 2010 MAPI Client and
CDO") claimed a release from the year in their name; and because msexchange is
a single-release slug, each of those findings then superseded the others once
per package per sweep.

This replays the inventory through the real state machine — upsert, supersede,
reconcile, against SQLite — and asserts the fixed point: sweep 2 and 3 write no
status change at all. Run: python tests/test_eol_sweep_stability.py
"""
import os
import sys
import types
from datetime import datetime

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# Every status change goes through log_vulnerability_change. Standing in for
# app.routers.vulnerabilities (the whole FastAPI stack) makes each transition
# observable — including the ones inside a single sweep.
TRANSITIONS: list = []
_fake = types.ModuleType("app.routers.vulnerabilities")
_fake.log_vulnerability_change = (
    lambda db, uid, vid, old, new, reason=None, cve_id=None, source=None,
    hostname=None: TRANSITIONS.append(
        f"{cve_id}: {getattr(old, 'value', old)} → {getattr(new, 'value', new)}"
        f"  [{source}] {reason}"))
sys.modules["app.routers.vulnerabilities"] = _fake

from sqlalchemy import create_engine                        # noqa: E402
from sqlalchemy.orm import sessionmaker                     # noqa: E402

from app.models.base import Base                            # noqa: E402
import app.models.user, app.models.group                    # noqa: E402,F401
import app.models.audit_log, app.models.setting             # noqa: E402,F401
from app.models.asset import Asset                          # noqa: E402
from app.models.vulnerability import (                      # noqa: E402
    Vulnerability, VulnerabilityStatus)
from app.services import eol_service as E                   # noqa: E402
from app.services import ms_lifecycle_service as MSL        # noqa: E402

# endoflife.date's msexchange catalogue, trimmed to the releases in play.
# Offline and pinned: the assertions are about our state machine, not about
# what the feed says this week.
CATALOG = {"result": {"releases": [
    {"name": "subscription", "label": "Subscription Edition", "isMaintained": True,
     "latest": {"name": "15.2.2562.45"}},
    {"name": "2019", "label": "2019", "isMaintained": True,
     "eolFrom": "2025-10-14", "eoesFrom": "2026-11-01",
     "latest": {"name": "15.2.1544.36"}},
    {"name": "2016", "label": "2016", "isMaintained": True,
     "eolFrom": "2025-10-14", "eoesFrom": "2026-11-01",
     "latest": {"name": "15.1.2507.61"}},
    {"name": "2013", "label": "2013", "isMaintained": False,
     "eolFrom": "2023-04-11", "latest": {"name": "15.0.1497.48"}},
    {"name": "2010", "label": "2010", "isMaintained": False,
     "eolFrom": "2020-10-13", "latest": {"name": "14.3.513.0"}},
    {"name": "2007", "label": "2007", "isMaintained": False,
     "eolFrom": "2017-04-11", "latest": {"name": "8.3.517.0"}},
]}}

# What an Exchange host really lists. Only the last two entries per host are
# the product; the rest ship alongside it.
def _pkgs(*pairs):
    return [{"name": n, "version": v} for n, v in pairs]


ADDONS = _pkgs(
    ("Microsoft Exchange 2007 Standard Anti-spam Filter Updates", "8.3.517.0"),
    ("Microsoft Exchange 2007 Enterprise Rules Updates", "8.3.517.0"),
    ("Microsoft Exchange Server 2016 Language Pack - Greek", "15.1.2507.6"),
    ("Microsoft Exchange Server 2010 MAPI Client and CDO 1.2.1", "6.5.8353.0"),
    ("Microsoft Lync Server 2013, Bootstrapper Prerequisites Installer Package",
     "5.0.8308.0"),
    ("Microsoft Unified Communications Managed API 4.0 Core Runtime 64-bit",
     "5.0.8308.0"),
)
SE_HOST = ADDONS + _pkgs(
    ("Microsoft Exchange Server", "15.2.2562.45"),
    ("Microsoft Exchange Server Subscription Edition", "15.2.2562.45"),
)
E2016_HOST = ADDONS + _pkgs(
    ("Microsoft Exchange Server", "15.1.2507.6"),
    ("Microsoft Exchange Server 2016 Cumulative Update 23", "15.1.2507.6"),
)


def _sweep(db, asset, packages):
    """The production sweep, exactly as the eol-check button and the nightly
    job call it — not a re-implementation of it."""
    E.run_eol_for_packages(db, asset, packages, reconcile=True)
    db.commit()


def _findings(db, asset_id):
    return {v.cve_id: getattr(v.status, "value", v.status)
            for v in db.query(Vulnerability)
                       .filter(Vulnerability.asset_id == asset_id).all()}


def demo():
    E.fetch_product = lambda db, slug: CATALOG if slug == "msexchange" else {}
    MSL.resolve_ms_lifecycle_eol = lambda db, name, version: None   # own source

    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine)
    db = sessionmaker(bind=engine)()

    se = Asset(hostname="mailhyb", ip_address="10.0.0.1")
    ex = Asset(hostname="exch2016", ip_address="10.0.0.2")
    db.add_all([se, ex])
    db.commit()

    # The false positive earlier runs left behind on the SE host.
    db.add(Vulnerability(
        cve_id="EOL-MSEXCHANGE-2007", asset_id=se.id, cvss_score=9.8,
        severity="critical", status=VulnerabilityStatus.open,
        title="stale false positive", detected_at=datetime.now(),
        package_name="Microsoft Exchange 2007 Standard Anti-spam Filter Updates",
        sources='["endoflife.date"]', first_detected_by="eol_check"))
    db.commit()

    for run in (1, 2, 3):
        TRANSITIONS.clear()
        _sweep(db, se, SE_HOST)
        _sweep(db, ex, E2016_HOST)

        se_f, ex_f = _findings(db, se.id), _findings(db, ex.id)
        # Subscription Edition is supported — no Exchange finding may stand,
        # and the stale 2007 row is retracted, once.
        assert se_f.get("EOL-MSEXCHANGE-2007") == "patched", se_f
        assert not [k for k, v in se_f.items()
                    if k.startswith("EOL-MSEXCHANGE") and v == "open"], se_f
        # The 2016 host keeps its one true finding and invents no others.
        assert ex_f.get("EOL-MSEXCHANGE-2016") == "open", ex_f
        assert [k for k in ex_f if k.startswith("EOL-MSEXCHANGE-")] == \
            ["EOL-MSEXCHANGE-2016"], ex_f
        # Nothing changed on either host, so nothing may change in the data.
        if run > 1:
            assert not TRANSITIONS, f"sweep {run} flapped:\n  " + "\n  ".join(TRANSITIONS)

    # The OS-level finding is produced by the caller, from asset fields the
    # package list knows nothing about. Seeded through seen_ids it survives;
    # forgotten, reconcile retracts it every night and the next run reopens it.
    os_row = Vulnerability(
        cve_id="EOL-WINDOWS-SERVER-2016", asset_id=ex.id, cvss_score=9.0,
        severity="high", status=VulnerabilityStatus.open,
        title="Windows Server 2016 — EOL", detected_at=datetime.now(),
        package_name="Microsoft Windows Server 2016", sources='["endoflife.date"]',
        first_detected_by="eol_check")
    db.add(os_row)
    db.commit()

    E.run_eol_for_packages(db, ex, E2016_HOST, reconcile=True,
                           seen_ids={os_row.id})
    db.commit()
    assert os_row.status == VulnerabilityStatus.open, "seeded OS finding was retracted"

    E.run_eol_for_packages(db, ex, E2016_HOST, reconcile=True)
    db.commit()
    assert os_row.status == VulnerabilityStatus.patched, (
        "unseeded OS finding must be retracted — otherwise seen_ids is untested")

    print("ok  repeated EOL sweeps over an unchanged inventory change nothing")


if __name__ == "__main__":
    demo()