admin / Synapse-Cortex

"""
Phase 5 verification for the Human-in-the-Loop Vulnerability Remediation playbook.

The playbook's Phase 1 and Phase 5 nodes run the EXACT SAME predefined discovery
command (see playbooks/vuln-remediation-hitl-linux.json — node data.command). Its
output is one package name per line: the set of packages with a pending upgrade.

A CVE is confirmed remediated when the package that carried it drops OUT of that
set between the baseline scan (Phase 1) and the verification scan (Phase 5).

This module is intentionally dependency-free and deterministic so the assistant
can call it verbatim and record the structured result on the ticket.
"""

from __future__ import annotations

from dataclasses import dataclass, field, asdict


def parse_scan(raw: str) -> set[str]:
    """Turn raw discovery output into a set of still-upgradable package names.

    The command already emits `sort -u`'d one-per-line names, so parsing is just
    a normalise-and-collect. Kept tolerant of blank lines / stray whitespace so a
    noisy SSH capture can't silently skew the diff."""
    return {line.strip() for line in (raw or "").splitlines() if line.strip()}


@dataclass
class VerificationResult:
    resolved: list[str] = field(default_factory=list)      # target pkgs now patched
    still_vulnerable: list[str] = field(default_factory=list)  # target pkgs still pending
    newly_upgradable: list[str] = field(default_factory=list)  # regressions introduced
    success: bool = False
    summary: str = ""


def verify_remediation(baseline_raw: str, verify_raw: str,
                       target_packages: list[str]) -> VerificationResult:
    """Diff the two identical-command scans and decide, per target package,
    whether the vulnerability was resolved.

    Args:
        baseline_raw:   raw stdout of the Phase 1 scan.
        verify_raw:     raw stdout of the Phase 5 scan (same command).
        target_packages: the packages the approved remediation was meant to fix
                         (derived from the ticket's vulnerabilities[].package).

    Success == every targeted package left the "upgradable" set AND the scan did
    not introduce a new upgradable package (no unexpected regression)."""
    before = parse_scan(baseline_raw)
    after = parse_scan(verify_raw)
    targets = {p.strip() for p in target_packages if p and p.strip()}

    resolved = sorted(p for p in targets if p in before and p not in after)
    still_vulnerable = sorted(p for p in targets if p in after)
    # A package that appears in the verify scan but was NOT pending before is a
    # regression the patch pulled in — surfaced, but not counted as our success.
    newly_upgradable = sorted((after - before) - targets)

    success = bool(targets) and not still_vulnerable

    if success:
        summary = (f"Verified: {', '.join(resolved) or 'no change needed'} no longer "
                   f"reports a pending security upgrade.")
    else:
        summary = ("Verification FAILED — still pending: "
                   f"{', '.join(still_vulnerable) or '(no targets supplied)'}.")
    if newly_upgradable:
        summary += f" Note: new upgradable packages appeared: {', '.join(newly_upgradable)}."

    return VerificationResult(
        resolved=resolved,
        still_vulnerable=still_vulnerable,
        newly_upgradable=newly_upgradable,
        success=success,
        summary=summary,
    )


if __name__ == "__main__":
    # Illustrative self-check: samba was patched, libssl was not.
    baseline = "libssl3\nsamba\nsudo\n"
    verify = "libssl3\nsudo\n"
    res = verify_remediation(baseline, verify, target_packages=["samba"])
    assert res.success and res.resolved == ["samba"], res
    res2 = verify_remediation(baseline, verify, target_packages=["samba", "libssl3"])
    assert not res2.success and res2.still_vulnerable == ["libssl3"], res2
    import json
    print(json.dumps(asdict(res), indent=2))
    print(json.dumps(asdict(res2), indent=2))