Skip to content

klyrek-monitor

Status: live on PyPI — pip install klyrek-monitor

Snapshots a ScanResult to disk and diffs two snapshots to surface what changed between scans — new/removed endpoints, new/removed/changed technologies, and new/resolved findings.

from klyrek_monitor.snapshot import save_snapshot, load_snapshot
from klyrek_monitor.diff import compute_diff

save_snapshot(result, Path("scan-2026-07-05.json"))
old = load_snapshot(Path("scan-2026-07-05.json"))
new = load_snapshot(Path("scan-2026-07-06.json"))
diff = compute_diff(old, new)
diff.has_changes   # bool
diff.new_endpoints, diff.removed_endpoints
diff.new_technologies, diff.removed_technologies, diff.changed_technologies
diff.new_findings, diff.resolved_findings

Real diff computed between two actual sandbox scans run a few hours apart: has_changes: False — nothing changed on the sandbox in that window, which is exactly the correct, honest result. The diff isn't supposed to invent changes that didn't happen.

load_snapshot fully reconstructs the original ScanResult — including Target, Endpoint, Technology, and Finding dataclasses and their datetime fields — from the JSON on disk, so a diff compares real, complete scan state, not a lossy summary.

API reference

klyrek_monitor.snapshot

Persist and reload a ScanResult snapshot, for later diffing against a fresh scan.

load_snapshot 

load_snapshot(path: Path) -> ScanResult

Reload a ScanResult previously written by save_snapshot().

Source code in klyrek_monitor\snapshot.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def load_snapshot(path: Path) -> ScanResult:
    """Reload a ScanResult previously written by save_snapshot()."""
    data = json.loads(path.read_text(encoding="utf-8"))
    target_data = data["target"]
    target = Target(
        base_url=target_data["base_url"],
        id=target_data["id"],
        hosts=target_data.get("hosts", []),
        notes=target_data.get("notes"),
        created_at=datetime.fromisoformat(target_data["created_at"]),
    )
    finished_at_raw = data.get("finished_at")
    result = ScanResult(
        target=target,
        id=data["id"],
        started_at=datetime.fromisoformat(data["started_at"]),
        finished_at=datetime.fromisoformat(finished_at_raw) if finished_at_raw else None,
        metadata=data.get("metadata", {}),
    )
    for endpoint_data in data.get("endpoints", []):
        result.add_endpoint(_endpoint_from_dict(endpoint_data))
    for tech_data in data.get("technologies", []):
        result.add_technology(_technology_from_dict(tech_data))
    for finding_data in data.get("findings", []):
        result.add_finding(_finding_from_dict(finding_data))
    return result

save_snapshot 

save_snapshot(result: ScanResult, path: Path) -> None

Write a ScanResult to disk as JSON, for later comparison via load_snapshot().

Source code in klyrek_monitor\snapshot.py 
25
26
27
def save_snapshot(result: ScanResult, path: Path) -> None:
    """Write a ScanResult to disk as JSON, for later comparison via load_snapshot()."""
    path.write_text(json.dumps(result, default=_default, indent=2), encoding="utf-8")

klyrek_monitor.diff

Compute the difference between two ScanResults for the same target.

compute_diff 

compute_diff(old: ScanResult, new: ScanResult) -> ScanDiff

Compare two scans of the same target and summarize what changed.

Source code in klyrek_monitor\diff.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def compute_diff(old: ScanResult, new: ScanResult) -> ScanDiff:
    """Compare two scans of the same target and summarize what changed."""
    old_endpoints = {_endpoint_key(e): e for e in old.endpoints}
    new_endpoints = {_endpoint_key(e): e for e in new.endpoints}

    old_tech_by_name = {t.name: t for t in old.technologies}
    new_tech_by_name = {t.name: t for t in new.technologies}

    old_findings = {_finding_key(f): f for f in old.findings}
    new_findings = {_finding_key(f): f for f in new.findings}

    changed_technologies = [
        TechnologyChange(
            name=name,
            category=new_tech_by_name[name].category,
            old_version=old_tech_by_name[name].version,
            new_version=new_tech_by_name[name].version,
        )
        for name in old_tech_by_name.keys() & new_tech_by_name.keys()
        if old_tech_by_name[name].version != new_tech_by_name[name].version
    ]

    return ScanDiff(
        new_endpoints=[e for key, e in new_endpoints.items() if key not in old_endpoints],
        removed_endpoints=[e for key, e in old_endpoints.items() if key not in new_endpoints],
        new_technologies=[
            t for name, t in new_tech_by_name.items() if name not in old_tech_by_name
        ],
        removed_technologies=[
            t for name, t in old_tech_by_name.items() if name not in new_tech_by_name
        ],
        changed_technologies=changed_technologies,
        new_findings=[f for key, f in new_findings.items() if key not in old_findings],
        resolved_findings=[f for key, f in old_findings.items() if key not in new_findings],
    )

klyrek_monitor.report

Render a ScanDiff as a human-readable Markdown change summary.

render_diff_markdown 

render_diff_markdown(diff: ScanDiff) -> str

Render a ScanDiff as Markdown; a one-line message if nothing changed.

Source code in klyrek_monitor\report.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def render_diff_markdown(diff: ScanDiff) -> str:
    """Render a ScanDiff as Markdown; a one-line message if nothing changed."""
    if not diff.has_changes:
        return "No changes detected since the last scan."

    lines: list[str] = ["# Klyrek Monitor: Changes Detected", ""]

    if diff.new_endpoints:
        lines.append(f"## New Endpoints ({len(diff.new_endpoints)})")
        lines.extend(f"- `{e.method} {e.url}`" for e in diff.new_endpoints)
        lines.append("")

    if diff.removed_endpoints:
        lines.append(f"## Removed Endpoints ({len(diff.removed_endpoints)})")
        lines.extend(f"- `{e.method} {e.url}`" for e in diff.removed_endpoints)
        lines.append("")

    if diff.new_technologies:
        lines.append(f"## New Technologies ({len(diff.new_technologies)})")
        lines.extend(
            f"- {t.category}: {t.name} {t.version or ''}".rstrip() for t in diff.new_technologies
        )
        lines.append("")

    if diff.removed_technologies:
        lines.append(f"## Removed Technologies ({len(diff.removed_technologies)})")
        lines.extend(f"- {t.category}: {t.name}" for t in diff.removed_technologies)
        lines.append("")

    if diff.changed_technologies:
        lines.append(f"## Technology Version Changes ({len(diff.changed_technologies)})")
        lines.extend(
            f"- {c.name}: {c.old_version or 'unknown'} -> {c.new_version or 'unknown'}"
            for c in diff.changed_technologies
        )
        lines.append("")

    if diff.new_findings:
        lines.append(f"## New Findings ({len(diff.new_findings)})")
        lines.extend(f"- **{f.severity.value.upper()}**: {f.title}" for f in diff.new_findings)
        lines.append("")

    if diff.resolved_findings:
        lines.append(f"## Resolved Findings ({len(diff.resolved_findings)})")
        lines.extend(f"- {f.title}" for f in diff.resolved_findings)
        lines.append("")

    return "\n".join(lines)