#!/usr/bin/env python3

"""The firewall, answered as "what can another machine reach?"

Listing zones and services is what firewall-cmd already does. The question it
does not answer is the one that matters, because it needs both halves at once: a
port is reachable only when something is LISTENING on a network address AND the
firewall permits it. Either alone tells you nothing -- which is how a tidy set of
rules coexists with an exposed database, as it does on this machine, where
Fedora Workstation's zone opens every port above 1024 and rootless containers
publish on all interfaces.

Changes go through firewall-cmd, which is polkit-aware, so they prompt.

    panama-firewall snapshot
    panama-firewall add-service NAME | remove-service NAME
    panama-firewall add-port PORT/PROTO | remove-port PORT/PROTO
    panama-firewall set-zone INTERFACE ZONE
    panama-firewall set-default-zone ZONE
"""

from __future__ import annotations

import json
import re
import os
import shutil
import subprocess
import sys
from pathlib import Path

ZONE = re.compile(r"^[A-Za-z0-9_-]{1,32}$")
SERVICE = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
INTERFACE = re.compile(r"^[A-Za-z0-9_.:-]{1,32}$")
PORT_SPEC = re.compile(r"^(\d{1,5})(?:-(\d{1,5}))?/(tcp|udp)$")

# Things whose exposure is worth saying out loud. Not a judgement about the
# software -- a database on the network is simply a different risk from a
# printer, and someone should know which they have.
DATA_STORES = {
    5432: "PostgreSQL", 3306: "MySQL", 3307: "MySQL", 6379: "Redis",
    27017: "MongoDB", 5984: "CouchDB", 9200: "Elasticsearch", 11211: "memcached",
    5433: "PostgreSQL", 1433: "SQL Server", 8086: "InfluxDB", 7000: "Cassandra",
}

# Ports whose purpose is worth naming when nothing else identifies them.
WELL_KNOWN = {
    22: "SSH", 3389: "Remote desktop", 5353: "mDNS", 5355: "LLMNR",
    1716: "KDE Connect", 631: "Printing", 139: "Samba", 445: "Samba",
    3000: "Development server", 8080: "HTTP alternate",
}


class BoundaryError(RuntimeError):
    """A user-visible validation or firewall failure."""


def run(command: list[str], timeout: float = 30.0) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(command, capture_output=True, text=True,
                              timeout=timeout, check=False)
    except (OSError, subprocess.TimeoutExpired) as error:
        raise BoundaryError(f"{command[0]} did not answer.") from error


def firewall(*arguments: str, timeout: float = 30.0) -> str:
    if not shutil.which("firewall-cmd"):
        raise BoundaryError("firewalld is not installed.")
    result = run(["firewall-cmd", *arguments], timeout=timeout)
    if result.returncode != 0:
        message = (result.stderr or result.stdout).strip().splitlines()
        text = message[-1] if message else "The firewall could not be read."
        if "not authorized" in text.lower() or "dismissed" in text.lower():
            raise BoundaryError("That firewall change was not authorized.")
        raise BoundaryError(text[:200])
    return result.stdout.strip()


def zone_detail(name: str) -> dict:
    try:
        raw = firewall(f"--zone={name}", "--list-all")
    except BoundaryError:
        return {}
    detail = {"name": name, "interfaces": [], "services": [], "ports": [],
              "richRules": [], "target": ""}
    key = ""
    for line in raw.splitlines()[1:]:
        if ":" not in line:
            continue
        key, _, value = line.strip().partition(":")
        value = value.strip()
        if key == "interfaces":
            detail["interfaces"] = value.split()
        elif key == "services":
            detail["services"] = value.split()
        elif key == "ports":
            detail["ports"] = value.split()
        elif key == "target":
            detail["target"] = value
        elif key == "rich rules":
            detail["richRules"] = [rule for rule in value.splitlines() if rule.strip()]
    return detail


def service_ports(name: str) -> list[str]:
    """The ports a named service stands for, from firewalld's own definition."""
    recorded = fixture()
    if recorded is not None:
        return list(recorded.get("servicePorts", {}).get(name, []))
    try:
        return firewall("--permanent", f"--service={name}", "--get-ports").split()
    except BoundaryError:
        return []


def allowed_ports(zone: dict) -> list[tuple[int, int, str, str]]:
    """Every port the zone permits, as (low, high, protocol, what allowed it)."""
    allowed = []
    for spec in zone.get("ports", []):
        found = PORT_SPEC.match(spec)
        if found:
            low = int(found.group(1))
            high = int(found.group(2) or found.group(1))
            allowed.append((low, high, found.group(3), "the open port range"))
    for service in zone.get("services", []):
        for spec in service_ports(service):
            found = PORT_SPEC.match(spec)
            if found:
                low = int(found.group(1))
                high = int(found.group(2) or found.group(1))
                allowed.append((low, high, found.group(3), f"the {service} service"))
    return allowed


# Above this, a UDP socket is almost certainly the local end of an outbound
# conversation -- a browser talking to the internet -- rather than a service
# waiting to be contacted. Listing those as "exposed" buries the two rows that
# matter under twenty that do not.
EPHEMERAL_FLOOR = 32768


def fixture() -> dict | None:
    """A recorded firewall and set of listeners, for testing the crossing.

    The rule this page exists for -- exposed means listening AND permitted --
    cannot be tested against a machine whose firewall permits everything, and
    the blocked case needs a listener below port 1024, which needs root to
    create. So the inputs can be supplied instead. Live behaviour is unchanged
    when the variable is unset.
    """
    path = os.environ.get("PANAMA_FIREWALL_FIXTURE")
    if not path:
        return None
    try:
        return json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise BoundaryError("The recorded firewall state could not be read.") from error


def listeners() -> list[dict]:
    """Sockets a machine on the network could actually connect TO.

    TCP listeners are unambiguous: they are in LISTEN state because something
    intends to accept connections. UDP has no such state, so a browser's
    outbound socket looks identical to a service -- which is why an ephemeral
    UDP port with no well-known meaning is left out rather than reported as an
    exposure someone should worry about.
    """
    recorded = fixture()
    if recorded is not None:
        return list(recorded.get("listeners", []))

    result = run(["ss", "-tulpnH"])
    if result.returncode != 0:
        return []
    found = {}
    for line in result.stdout.splitlines():
        parts = line.split()
        if len(parts) < 5:
            continue
        protocol = parts[0]
        address = parts[4]
        # Loopback is not reachable from anywhere else, so it is not exposure.
        if address.startswith(("127.", "[::1]")):
            continue
        port_text = address.rsplit(":", 1)[-1]
        if not port_text.isdigit():
            continue
        process = ""
        owner = re.search(r'users:\(\("([^"]+)"', line)
        if owner:
            process = owner.group(1)
        port = int(port_text)
        if (protocol.startswith("udp") and port >= EPHEMERAL_FLOOR
                and port not in WELL_KNOWN and port not in DATA_STORES):
            continue
        key = (port, protocol)
        # A port bound on several addresses is one exposure, not four.
        if key not in found or (process and not found[key]["process"]):
            found[key] = {"port": port, "protocol": protocol, "process": process,
                          "address": address}
    return sorted(found.values(), key=lambda entry: entry["port"])


def describe(entry: dict) -> tuple[str, str]:
    """A name for what is listening, and how much it matters."""
    port = entry["port"]
    if port in DATA_STORES:
        return DATA_STORES[port], "data"
    if port in WELL_KNOWN:
        return WELL_KNOWN[port], "known"
    process = entry.get("process") or ""
    return (process or f"port {port}"), "other"


def snapshot() -> dict:
    recorded = fixture()
    if recorded is not None:
        zones = list(recorded.get("zones", []))
        permitted = []
        for zone in zones:
            permitted.extend(allowed_ports(zone))
        exposed = []
        for entry in listeners():
            allowed_by = ""
            for low, high, protocol, reason in permitted:
                if str(entry.get("protocol", "")).startswith(protocol) and low <= int(entry["port"]) <= high:
                    if not allowed_by or reason != "the open port range":
                        allowed_by = reason
            if not allowed_by:
                continue
            name, kind = describe(entry)
            exposed.append({"name": name, "port": int(entry["port"]),
                            "protocol": entry.get("protocol", ""),
                            "process": entry.get("process", ""),
                            "allowedBy": allowed_by, "kind": kind})
        return {"running": True, "enabledAtBoot": True, "available": True,
                "defaultZone": zones[0]["name"] if zones else "", "allZones": [],
                "activeZones": {}, "zones": zones, "exposed": exposed,
                "exposedDataStores": [e for e in exposed if e["kind"] == "data"],
                "sshSessions": 0, "error": ""}

    running = run(["systemctl", "is-active", "firewalld"]).stdout.strip() == "active"
    enabled = run(["systemctl", "is-enabled", "firewalld"]).stdout.strip() == "enabled"
    if not shutil.which("firewall-cmd") or not running:
        return {"running": running, "enabledAtBoot": enabled, "available": False,
                "zones": [], "activeZones": {}, "defaultZone": "", "exposed": [],
                "allZones": [], "error": ""}

    default_zone = firewall("--get-default-zone")
    all_zones = firewall("--get-zones").split()

    active = {}
    current = ""
    for line in firewall("--get-active-zones").splitlines():
        if not line.startswith(" "):
            current = line.split()[0] if line.split() else ""
            continue
        if "interfaces:" in line and current:
            active[current] = line.split(":", 1)[1].split()

    zones = [zone_detail(name) for name in active] or [zone_detail(default_zone)]
    zones = [zone for zone in zones if zone]

    # The cross-reference: listening AND permitted.
    permitted = []
    for zone in zones:
        permitted.extend(allowed_ports(zone))

    exposed = []
    for entry in listeners():
        allowed_by = ""
        for low, high, protocol, reason in permitted:
            if entry["protocol"].startswith(protocol) and low <= entry["port"] <= high:
                # A named service is a better explanation than a range.
                if not allowed_by or reason != "the open port range":
                    allowed_by = reason
        if not allowed_by:
            continue
        name, kind = describe(entry)
        exposed.append({
            "name": name,
            "port": entry["port"],
            "protocol": entry["protocol"],
            "process": entry["process"],
            "allowedBy": allowed_by,
            "kind": kind,
        })

    return {
        "running": running,
        "enabledAtBoot": enabled,
        "available": True,
        "defaultZone": default_zone,
        "allZones": all_zones,
        "activeZones": active,
        "zones": zones,
        "exposed": exposed,
        # Counted separately so the page can lead with it.
        "exposedDataStores": [entry for entry in exposed if entry["kind"] == "data"],
        "sshSessions": len([line for line in run(
            ["ss", "-tnH", "state", "established", "( sport = :22 )"]).stdout.splitlines()
            if line.strip()]),
        "error": "",
    }


def require(pattern: re.Pattern, value: str, message: str) -> str:
    if not pattern.fullmatch(value or ""):
        raise BoundaryError(message)
    return value


def change(zone: str, *arguments: str) -> None:
    """Apply to the running firewall and to the stored configuration.

    Both, because a change that survives a reboot but is not in effect -- or the
    reverse -- is a firewall nobody can reason about.
    """
    firewall(f"--zone={zone}", *arguments, timeout=120)
    firewall("--permanent", f"--zone={zone}", *arguments, timeout=120)


def active_zone() -> str:
    state = snapshot()
    zones = state.get("zones") or []
    return zones[0]["name"] if zones else state.get("defaultZone", "")


def main(arguments: list[str]) -> int:
    try:
        if arguments == ["snapshot"]:
            print(json.dumps(snapshot(), separators=(",", ":")))
            return 0

        if len(arguments) == 2 and arguments[0] in ("add-service", "remove-service"):
            name = require(SERVICE, arguments[1], "That is not a service name.")
            verb = "--add-service" if arguments[0] == "add-service" else "--remove-service"
            change(active_zone(), f"{verb}={name}")
        elif len(arguments) == 2 and arguments[0] in ("add-port", "remove-port"):
            spec = require(PORT_SPEC, arguments[1], "That is not a port.")
            verb = "--add-port" if arguments[0] == "add-port" else "--remove-port"
            change(active_zone(), f"{verb}={spec}")
        elif len(arguments) == 3 and arguments[0] == "set-zone":
            interface = require(INTERFACE, arguments[1], "That is not a network interface.")
            zone = require(ZONE, arguments[2], "That is not a zone.")
            firewall(f"--zone={zone}", f"--change-interface={interface}", timeout=120)
            firewall("--permanent", f"--zone={zone}", f"--change-interface={interface}", timeout=120)
        elif len(arguments) == 2 and arguments[0] == "set-default-zone":
            zone = require(ZONE, arguments[1], "That is not a zone.")
            firewall(f"--set-default-zone={zone}", timeout=120)
        else:
            raise BoundaryError(
                "Usage: panama-firewall snapshot | add-service NAME | remove-service NAME | "
                "add-port PORT/PROTO | remove-port PORT/PROTO | set-zone INTERFACE ZONE | "
                "set-default-zone ZONE")
    except BoundaryError as error:
        try:
            state = snapshot()
        except BoundaryError:
            state = {"running": False, "available": False, "zones": [], "exposed": []}
        state["error"] = str(error)
        print(json.dumps(state, separators=(",", ":")))
        return 0

    print(json.dumps(snapshot(), separators=(",", ":")))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))