x1pro/immortalwrt-patches/tools/audit_ui_packages.py

#!/usr/bin/env python3
"""Static safety gates for the v46.1 UI-only LuCI packages."""

from __future__ import annotations

import json
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent
EXPECTED_PACKAGES = {
    "luci-theme-argon",
    "luci-app-argon-config",
    "luci-app-wgtunnel",
    "luci-app-tr3000-status",
}
FORBIDDEN_PACKAGES = {
    "luci-app-turboacc-mtk",
    "luci-app-wrtbwmon",
    "kmod-mediatek_hnat",
    "kmod-warp",
    "kmod-mt_wifi",
    "kmod-tcp-bbr",
}
GLOBAL_FORBIDDEN = {
    "ifup lan": "must never bounce the management LAN",
    "wifi restart": "must never reset Wi-Fi from a UI package",
    "/etc/init.d/network restart": "must never restart the global network",
    "/etc/init.d/firewall restart": "must never restart fw4",
}
FRONTEND_FORBIDDEN = {
    "handleSaveApply(": "WG UI must not invoke LuCI global apply",
    "fs.exec": "browser code must not execute programs",
    "uci.load('network')": "WG UI must not receive generic network UCI access",
    'uci.load("network")': "WG UI must not receive generic network UCI access",
    "new form.Map('network'": "WG UI must use JSONMap instead of network UCI",
    'new form.Map("network"': "WG UI must use JSONMap instead of network UCI",
}
KEY_LIKE = re.compile(r"(?<![A-Za-z0-9+/])[A-Za-z0-9+/]{43}=(?![A-Za-z0-9+/=])")


def fail(message: str) -> None:
    print(f"FAIL: {message}")
    raise SystemExit(1)


def text_files(package: Path):
    for path in sorted(package.rglob("*")):
        if not path.is_file():
            continue
        # Repo-only documentation (not shipped in the IPK) may legitimately
        # name forbidden patterns when describing them; only scan shipped/code
        # files for the forbidden-content gates.
        if path.suffix == ".md" or path.name.upper().startswith("README"):
            continue
        # Test harnesses (not shipped) legitimately reference forbidden package
        # names when asserting they are absent; skip them too.
        if "tests" in path.relative_to(package).parts:
            continue
        try:
            yield path, path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            continue


def load_json(path: Path) -> dict:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        fail(f"invalid JSON {path.relative_to(ROOT)}: {exc}")


def sole_acl(path: Path, name: str) -> dict:
    data = load_json(path)
    if set(data) != {name}:
        fail(f"{path.relative_to(ROOT)} must contain only ACL {name}")
    return data[name]


def assert_ubus_only(block: dict, obj: str, methods: set[str]) -> None:
    if set(block) != {"ubus"}:
        fail(f"ACL block for {obj} must grant ubus only, got {sorted(block)}")
    ubus = block["ubus"]
    if set(ubus) != {obj} or set(ubus[obj]) != methods:
        fail(f"ACL for {obj} expected methods {sorted(methods)}, got {ubus}")


def check_wgtunnel() -> None:
    package = ROOT / "luci-app-wgtunnel"
    acl = sole_acl(
        package / "root/usr/share/rpcd/acl.d/luci-app-wgtunnel.json",
        "luci-app-wgtunnel",
    )
    assert_ubus_only(acl.get("read", {}), "luci.wgtunnel", {"get", "status"})
    assert_ubus_only(
        acl.get("write", {}),
        "luci.wgtunnel",
        {"prepare", "apply", "rollback", "reconnect"},
    )

    frontend = package / "htdocs/luci-static/resources/view/wgtunnel.js"
    source = frontend.read_text(encoding="utf-8")
    for needle, reason in FRONTEND_FORBIDDEN.items():
        if needle in source:
            fail(f"{frontend.relative_to(ROOT)} contains {needle!r}: {reason}")
    for required in ("form.JSONMap", "luci.wgtunnel", "prepare", "apply"):
        if required not in source:
            fail(f"{frontend.relative_to(ROOT)} lacks {required!r}")

    backend = package / "root/usr/share/rpcd/ucode/luci.wgtunnel"
    if not backend.is_file():
        fail(f"missing {backend.relative_to(ROOT)}")
    source = backend.read_text(encoding="utf-8")
    for method in ("get", "status", "prepare", "apply", "rollback", "reconnect"):
        if re.search(rf"\b{re.escape(method)}\s*:", source) is None:
            fail(f"WG backend lacks method {method}")
    for required in ("10.99.0.2/32", "10.99.0.1/32", "wg0", "vxlan0"):
        if required not in source:
            fail(f"WG backend lacks topology fuse {required}")


def check_status() -> None:
    package = ROOT / "luci-app-tr3000-status"
    acl = sole_acl(
        package / "root/usr/share/rpcd/acl.d/luci-app-tr3000-status.json",
        "luci-app-tr3000-status",
    )
    assert_ubus_only(acl.get("read", {}), "luci.tr3000_status", {"get"})
    if "write" in acl:
        fail("Link Health ACL must not contain a write block")

    backend = package / "root/usr/share/rpcd/ucode/luci.tr3000_status"
    frontend = package / "htdocs/luci-static/resources/view/status/tr3000.js"
    if not backend.is_file() or not frontend.is_file():
        fail("Link Health backend/frontend is missing")
    for path in (backend, frontend):
        source = path.read_text(encoding="utf-8")
        for needle in ("private_key", "preshared_key", "wifi-key", "wireless key"):
            if needle in source.lower():
                fail(f"{path.relative_to(ROOT)} references secret field {needle!r}")


def check_argon_acl() -> None:
    path = ROOT / "luci-app-argon-config/root/usr/share/rpcd/acl.d/luci-app-argon-config.json"
    acl = sole_acl(path, "luci-app-argon-config")
    read_ubus = acl.get("read", {}).get("ubus", {})
    write_ubus = acl.get("write", {}).get("ubus", {})
    mutators = {"remove", "rename"}
    if mutators & set(read_ubus.get("luci.argon", [])):
        fail("Argon remove/rename methods must not be granted in read ACL")
    if not mutators <= set(write_ubus.get("luci.argon", [])):
        fail("Argon write ACL must grant remove and rename")


def check_all_sources() -> None:
    packages = {path.name for path in ROOT.iterdir() if path.is_dir() and path.name.startswith("luci-")}
    missing = EXPECTED_PACKAGES - packages
    if missing:
        fail(f"missing package directories: {sorted(missing)}")

    for package_name in sorted(EXPECTED_PACKAGES):
        package = ROOT / package_name
        for path, source in text_files(package):
            rel = path.relative_to(ROOT)
            for needle, reason in GLOBAL_FORBIDDEN.items():
                if needle in source:
                    fail(f"{rel} contains {needle!r}: {reason}")
            for forbidden in FORBIDDEN_PACKAGES:
                if forbidden in source:
                    fail(f"{rel} pulls forbidden package {forbidden}")
            if KEY_LIKE.search(source):
                fail(f"{rel} contains a possible WireGuard key literal")
            if path.suffix == ".json":
                load_json(path)


def main() -> int:
    check_all_sources()
    check_argon_acl()
    check_wgtunnel()
    check_status()
    print("PASS: v46.1 UI-only static safety gates")
    return 0


if __name__ == "__main__":
    sys.exit(main())