Add a Firewall page, led by what is actually reachable
Listing zones and services is what firewall-cmd already does. The question it does not answer needs both halves at once: a port is reachable only when something is LISTENING on a network address AND the firewall permits it. On this machine that crossing is the whole story. The rules look unremarkable -- one zone, three services, a port range -- and what they mean is that PostgreSQL and Redis, published by rootless containers on every interface, are reachable by anyone on the network. Neither half says that alone, which is exactly how a tidy rules list coexists with an open database. Nothing was misconfigured: Fedora's default zone met podman's default publish behaviour. Ephemeral client sockets are excluded. A browser's outbound UDP port is indistinguishable from a service in ss, and listing twenty of them buried the two rows that mattered. Closing the port range names what it would cut off, by service, before doing it, and removing ssh says so when someone is connected over it. Rich rules are shown and never edited: a syntax is not a setting, but hiding it would misrepresent the configuration. The contract needed a recorded firewall, and the reason is worth keeping. The rule this page exists for cannot be tested against this machine -- its zone permits everything above 1024, so "listening" and "listening and permitted" give identical answers, and a blocked listener needs a port below 1024, which needs root. With the crossing deleted, the contract passed. It now runs against a fixture where two listeners are blocked, and catches it. Also here: polkit response files are written 0600 rather than at the default mask, the agent sweeps requests left by an instance that did not exit cleanly, and the write sweep waits for its harness to be ready instead of reporting the startup race as settings that failed. Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
This commit is contained in:
Gabriel Brown
+371
@@ -0,0 +1,371 @@
|
||||
#!/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:]))
|
||||
Reference in New Issue
Block a user