secubox-deb/packages/secubox-cve-triage/tests/test_emit.py

# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
import json
import os
import stat

import pytest

from api.wafgen.emit import CATEGORY_KEY, build_category, write_category
from api.wafgen.extract import Candidate


def cand(cve="CVE-2024-3400", vendor="paloaltonetworks", product="pan-os",
         path="/api/v1/totp/user-backup"):
    return Candidate(cve=cve, vendor=vendor, product=product,
                     cpe=f"cpe:2.3:a:{vendor}:{product}:*", path=path, severity="critical")


def _rules_file(tmp_path):
    p = tmp_path / "waf-rules.json"
    p.write_text(json.dumps({
        "_meta": {"version": "1.2.0"},
        "categories": {
            "sqli": {"name": "SQLi", "severity": "critical",
                     "patterns": [{"id": "sqli-001", "pattern": "union select", "desc": "x"}]},
        },
    }, indent=2))
    return p


def test_category_is_always_detect_mode():
    c = build_category([cand()], now="2026-07-18T00:00:00Z")
    assert c["mode"] == "detect"
    assert c["generated"] is True


def test_pattern_path_is_regex_escaped():
    # A path with regex-special chars must be escaped so it matches literally.
    c = build_category([cand(path="/mgmt/tm/util/bash.php?x=1")], now="2026-07-18T00:00:00Z")
    pat = c["patterns"][0]["pattern"]
    assert "\\.php" in pat and "\\?" in pat


def test_write_preserves_the_other_categories(tmp_path):
    p = _rules_file(tmp_path)
    write_category(p, [cand()], now="2026-07-18T00:00:00Z")
    d = json.loads(p.read_text())
    # sqli untouched, product_absent_probes added.
    assert "sqli" in d["categories"]
    assert d["categories"]["sqli"]["patterns"][0]["id"] == "sqli-001"
    assert CATEGORY_KEY in d["categories"]
    assert d["categories"][CATEGORY_KEY]["mode"] == "detect"
    assert d["_meta"]["version"] == "1.2.0"


def test_regenerate_is_idempotent(tmp_path):
    p = _rules_file(tmp_path)
    write_category(p, [cand()], now="2026-07-18T00:00:00Z")
    first = p.read_text()
    write_category(p, [cand()], now="2026-07-18T00:00:00Z")
    assert p.read_text() == first


def test_regenerate_replaces_only_its_own_category(tmp_path):
    p = _rules_file(tmp_path)
    write_category(p, [cand(cve="CVE-2024-3400")], now="2026-07-18T00:00:00Z")
    write_category(p, [cand(cve="CVE-2023-46747", vendor="f5", product="big-ip",
                            path="/mgmt/tm/util/bash")], now="2026-07-18T00:00:00Z")
    d = json.loads(p.read_text())
    ids = [x["id"] for x in d["categories"][CATEGORY_KEY]["patterns"]]
    assert any("2023-46747" in i for i in ids)
    assert not any("2024-3400" in i for i in ids)   # replaced, not appended
    assert "sqli" in d["categories"]


def test_empty_candidates_writes_an_empty_category(tmp_path):
    p = _rules_file(tmp_path)
    write_category(p, [], now="2026-07-18T00:00:00Z")
    d = json.loads(p.read_text())
    assert d["categories"][CATEGORY_KEY]["patterns"] == []


def test_temp_file_path_is_unique_per_call(tmp_path, monkeypatch):
    # Concurrent writers (CLI cron regen + panel regen) must not share a
    # single fixed ".json.tmp" name: two writers racing on the same fixed
    # name means one process's os.replace() can promote the OTHER
    # process's in-flight content (a silent lost update). Spy on the
    # actual src path passed to os.replace() across two calls — with a
    # fixed name this would be identical every time; with mkstemp() it
    # must differ.
    import api.wafgen.emit as emit_mod

    p = _rules_file(tmp_path)
    real_replace = os.replace
    seen_src_paths = []

    def spy_replace(src, dst):
        seen_src_paths.append(str(src))
        return real_replace(src, dst)

    monkeypatch.setattr(emit_mod.os, "replace", spy_replace)

    write_category(p, [cand()], now="2026-07-18T00:00:00Z")
    write_category(p, [cand()], now="2026-07-18T00:00:00Z")

    assert len(seen_src_paths) == 2
    assert seen_src_paths[0] != seen_src_paths[1]


def test_no_leftover_tmp_file_after_successful_write(tmp_path):
    p = _rules_file(tmp_path)
    write_category(p, [cand()], now="2026-07-18T00:00:00Z")
    leftovers = [f.name for f in tmp_path.iterdir() if f.name != p.name]
    assert leftovers == []


def test_write_category_raises_on_missing_rules_file(tmp_path):
    missing = tmp_path / "waf-rules.json"
    with pytest.raises(FileNotFoundError):
        write_category(missing, [cand()], now="2026-07-18T00:00:00Z")


def test_write_preserves_original_file_permission_mode(tmp_path):
    # tempfile.mkstemp() always creates its file 0600 regardless of umask;
    # os.replace() would make waf-rules.json inherit that, silently
    # tightening it from its live mode (typically 0644) and breaking the
    # WAF reader if it runs as a different user (permission-drift outage).
    p = _rules_file(tmp_path)
    os.chmod(p, 0o644)
    write_category(p, [cand()], now="2026-07-18T00:00:00Z")
    assert stat.S_IMODE(os.stat(p).st_mode) == 0o644


def test_write_category_raises_on_corrupt_rules_file_and_leaves_it_unchanged(tmp_path):
    p = tmp_path / "waf-rules.json"
    corrupt = "{ not valid json"
    p.write_text(corrupt)
    with pytest.raises(json.JSONDecodeError):
        write_category(p, [cand()], now="2026-07-18T00:00:00Z")
    # A generator must never blank the WAF's live rules on failure.
    assert p.read_text() == corrupt
    leftovers = [f.name for f in tmp_path.iterdir() if f.name != p.name]
    assert leftovers == []