742 lines
24 KiB
Python
742 lines
24 KiB
Python
"""Firewall daemon handler.
|
|
|
|
Reads from the pre-computed state for status endpoints. Executes
|
|
firewall-cmd with sudo for mutations. Refers state after each mutation.
|
|
"""
|
|
|
|
import logging
|
|
from contextlib import suppress
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from daemon.iface import (
|
|
DELETE_FIREWALL_FORWARD_PORT_REMOVE,
|
|
DELETE_FIREWALL_RICH_RULES_REMOVE,
|
|
DELETE_FIREWALL_ZONES_DELETE,
|
|
GET_FIREWALL_CONFIG,
|
|
GET_FIREWALL_CONFIG_PENDING,
|
|
GET_FIREWALL_INTERFACES,
|
|
GET_FIREWALL_RICH_RULES,
|
|
GET_FIREWALL_SERVICES,
|
|
GET_FIREWALL_STATE,
|
|
GET_FIREWALL_ZONES,
|
|
GET_FIREWALL_ZONES_ALL,
|
|
GET_FIREWALL_ZONES_INFO,
|
|
PATCH_FIREWALL_CONFIG,
|
|
POST_FIREWALL_CONFIG,
|
|
POST_FIREWALL_CONFIG_APPLY,
|
|
POST_FIREWALL_FORWARD_PORT_ADD,
|
|
POST_FIREWALL_MASQUERADE,
|
|
POST_FIREWALL_RICH_RULES_ADD,
|
|
POST_FIREWALL_ZONES_CREATE,
|
|
POST_FIREWALL_ZONES_INTERFACES,
|
|
POST_FIREWALL_ZONES_SERVICES,
|
|
)
|
|
from daemon.server import NotFoundError, refresh_state, registry
|
|
from lib.common import load_json, run, save_json
|
|
from lib.firewall import (
|
|
_normalize_target,
|
|
_parse_active_zones,
|
|
_parse_zone_output,
|
|
)
|
|
from lib.firewall import (
|
|
save_backup as _save_backup,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROJECT_DIR = Path(__file__).resolve().parent.parent.parent
|
|
CONFIG_DIR = PROJECT_DIR / "config" / "firewall"
|
|
CONFIG_FILE = CONFIG_DIR / "config.json"
|
|
DEFAULT_CONFIG = {"zones": {}}
|
|
|
|
|
|
def _get_state() -> dict[str, Any] | None:
|
|
"""Return the current firewall state from the state store."""
|
|
from lib.state import state as state_store
|
|
|
|
return state_store.get("firewall")
|
|
|
|
|
|
def _ensure_config_file() -> None:
|
|
"""Initialize config file with defaults if missing."""
|
|
if not CONFIG_FILE.exists():
|
|
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
save_json(CONFIG_FILE, DEFAULT_CONFIG, indent=2)
|
|
|
|
|
|
def _get_config() -> dict[str, Any]:
|
|
"""Load the firewall config file."""
|
|
_ensure_config_file()
|
|
return load_json(CONFIG_FILE)
|
|
|
|
|
|
def _save_config(cfg: dict[str, Any]) -> None:
|
|
"""Persist firewall config to disk."""
|
|
_ensure_config_file()
|
|
save_json(CONFIG_FILE, cfg, indent=2)
|
|
|
|
|
|
def _reload() -> None:
|
|
"""Reload firewalld to apply permanent changes."""
|
|
run(["firewall-cmd", "--reload"], sudo=True)
|
|
|
|
|
|
def _fp_to_str(fp: dict[str, Any]) -> str:
|
|
"""Convert a forward-port dict to firewall-cmd CLI argument string."""
|
|
parts = [f"port={fp['port']}", f"proto={fp['proto']}"]
|
|
if "toaddr" in fp:
|
|
parts.append(f"toaddr={fp['toaddr']}")
|
|
if "toport" in fp:
|
|
parts.append(f"toport={fp['toport']}")
|
|
return "/".join(parts)
|
|
|
|
|
|
def _get_forward_ports(zone_name: str) -> list[str]:
|
|
"""Return forward-port entries for a zone as CLI-style strings."""
|
|
with suppress(Exception):
|
|
fps = _parse_zone_output(
|
|
zone_name,
|
|
run(["firewall-cmd", f"--zone={zone_name}", "--list-all"], sudo=True),
|
|
).get("forward-ports", [])
|
|
return [_fp_to_str(fp) for fp in fps if isinstance(fp, dict)]
|
|
return []
|
|
|
|
|
|
def _config_apply() -> dict[str, Any]:
|
|
"""Apply the declarative config to live firewalld."""
|
|
from lib.firewall import get_config as _get_lib_config
|
|
|
|
cfg = _get_lib_config()
|
|
cfg_zones = cfg.get("zones", {})
|
|
|
|
full_state: dict[str, Any] = {
|
|
"active_zones": {},
|
|
"interfaces": [],
|
|
"available_services": [],
|
|
"zones": {},
|
|
"rich_rules": {},
|
|
"timestamp": "",
|
|
}
|
|
_save_backup(full_state)
|
|
|
|
available = run(["firewall-cmd", "--get-zones"], sudo=True).split()
|
|
applied: list[str] = []
|
|
for zone_name, zone_cfg in cfg_zones.items():
|
|
need_create = zone_name not in available
|
|
|
|
if need_create:
|
|
target = _normalize_target(zone_cfg.get("target", "DEFAULT"))
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--set-target={target}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
_reload()
|
|
else:
|
|
desired_target = _normalize_target(zone_cfg.get("target", "DEFAULT"))
|
|
if desired_target != "default":
|
|
with suppress(RuntimeError):
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--set-target={desired_target}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
|
|
current_svcs: list[str] = []
|
|
with suppress(Exception):
|
|
current_svcs = _parse_zone_output(
|
|
zone_name,
|
|
run(["firewall-cmd", f"--zone={zone_name}", "--list-all"], sudo=True),
|
|
).get("services", [])
|
|
for svc in current_svcs:
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--remove-service={svc}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
for svc in zone_cfg.get("services", []):
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--add-service={svc}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
|
|
current_ifaces: list[str] = []
|
|
with suppress(Exception):
|
|
current_ifaces = _parse_zone_output(
|
|
zone_name,
|
|
run(["firewall-cmd", f"--zone={zone_name}", "--list-all"], sudo=True),
|
|
).get("interfaces", [])
|
|
for iface in current_ifaces:
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
"--remove-interface=" + iface,
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
for iface in zone_cfg.get("interfaces", []):
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
"--add-interface=" + iface,
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
|
|
mq = zone_cfg.get("masquerade", False)
|
|
if mq is not None:
|
|
action = "--add-masquerade" if mq else "--remove-masquerade"
|
|
run(
|
|
["firewall-cmd", f"--zone={zone_name}", action, "--permanent"],
|
|
sudo=True,
|
|
)
|
|
|
|
for rule_entry in zone_cfg.get("rich_rules", []):
|
|
rule_str = (
|
|
rule_entry.get("rule", "")
|
|
if isinstance(rule_entry, dict)
|
|
else str(rule_entry)
|
|
)
|
|
if rule_str:
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--add-rich-rule={rule_str}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
|
|
current_fps = _get_forward_ports(zone_name)
|
|
for fp_str in current_fps:
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--remove-forward-port={fp_str}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
for fp_entry in zone_cfg.get("forward_ports", []):
|
|
fp_str = fp_entry if isinstance(fp_entry, str) else _fp_to_str(fp_entry)
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--add-forward-port={fp_str}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
|
|
applied.append(zone_name)
|
|
|
|
_reload()
|
|
full_state = {
|
|
"active_zones": {},
|
|
"interfaces": [],
|
|
"available_services": [],
|
|
"zones": {},
|
|
"rich_rules": {},
|
|
"timestamp": "",
|
|
}
|
|
backup_path = _save_backup(full_state)
|
|
logger.info("Firewall config applied to %d zones", len(applied))
|
|
return {
|
|
"applied_zones": applied,
|
|
"backup": backup_path,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — GET endpoints read from state, mutations call refresh_state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _get_fw_state() -> dict[str, Any]:
|
|
"""Return firewall state from the state store, or empty dict if absent."""
|
|
fw = _get_state()
|
|
if fw is None:
|
|
return {}
|
|
return fw
|
|
|
|
|
|
@registry.register(GET_FIREWALL_INTERFACES)
|
|
def get_interfaces(_request: Any, _body: Any) -> list[dict[str, Any]]:
|
|
"""GET /firewall/interfaces — return active interfaces from state."""
|
|
fw = _get_fw_state()
|
|
return fw.get("interfaces", [])
|
|
|
|
|
|
@registry.register(GET_FIREWALL_ZONES)
|
|
def get_zones(_request: Any, _body: Any) -> dict[str, Any]:
|
|
fw = _get_fw_state()
|
|
active = fw.get("active_zones", {})
|
|
zones = fw.get("zones", {})
|
|
return {"active": active, "available": list(zones.keys())}
|
|
|
|
|
|
@registry.register(GET_FIREWALL_ZONES_INFO)
|
|
def get_zone_info(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body or "zone" not in body:
|
|
raise ValueError("'zone' is required")
|
|
zone = body["zone"]
|
|
fw = _get_fw_state()
|
|
zones = fw.get("zones", {})
|
|
if zone not in zones:
|
|
raise NotFoundError(f"Zone '{zone}' does not exist")
|
|
return zones[zone]
|
|
|
|
|
|
@registry.register(GET_FIREWALL_ZONES_ALL)
|
|
def get_all_zones_info(_request: Any, _body: Any) -> list[dict[str, Any]]:
|
|
fw = _get_fw_state()
|
|
active = fw.get("active_zones", {})
|
|
zones = fw.get("zones", {})
|
|
result: list[dict[str, Any]] = []
|
|
for zone_name in active:
|
|
if zone_name in zones:
|
|
result.append(zones[zone_name])
|
|
return result
|
|
|
|
|
|
@registry.register(GET_FIREWALL_SERVICES)
|
|
def get_services(_request: Any, _body: Any) -> list[str]:
|
|
fw = _get_fw_state()
|
|
return fw.get("available_services", [])
|
|
|
|
|
|
@registry.register(GET_FIREWALL_CONFIG)
|
|
def get_config(_request: Any, _body: Any) -> dict[str, Any]:
|
|
return _get_config()
|
|
|
|
|
|
@registry.register(POST_FIREWALL_CONFIG)
|
|
def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body or "zones" not in body:
|
|
raise ValueError("'zones' key is required")
|
|
if not isinstance(body["zones"], dict):
|
|
raise ValueError("'zones' must be a dict")
|
|
_save_config(body)
|
|
logger.info("Firewall config saved (%d zones)", len(body["zones"]))
|
|
refresh_state(["firewall"])
|
|
return {"config_saved": True}
|
|
|
|
|
|
@registry.register(PATCH_FIREWALL_CONFIG)
|
|
def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body must be a JSON object")
|
|
from lib.common import deep_merge
|
|
|
|
current = _get_config()
|
|
merged = deep_merge(current, body)
|
|
_save_config(merged)
|
|
logger.info("Firewall config patched: %s", sorted(body.keys()))
|
|
refresh_state(["firewall"])
|
|
return {"config_saved": True}
|
|
|
|
|
|
@registry.register(GET_FIREWALL_CONFIG_PENDING)
|
|
def config_pending_handler(_request: Any, _body: Any) -> dict[str, Any]:
|
|
fw = _get_fw_state()
|
|
return fw.get("pending", {})
|
|
|
|
|
|
@registry.register(POST_FIREWALL_CONFIG_APPLY)
|
|
def config_apply(_request: Any, _body: Any) -> dict[str, Any]:
|
|
result = _config_apply()
|
|
logger.info("Firewall config applied: %s", result.get("applied_zones", []))
|
|
refresh_state(["firewall"])
|
|
return result
|
|
|
|
|
|
@registry.register(POST_FIREWALL_ZONES_CREATE)
|
|
def create_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone_name = body.get("name", "").strip()
|
|
target = body.get("target", "default").strip() or "default"
|
|
if not zone_name:
|
|
raise ValueError("Zone name is required")
|
|
available = run(["firewall-cmd", "--get-zones"], sudo=True).split()
|
|
if zone_name in available:
|
|
raise ValueError(f"Zone '{zone_name}' already exists")
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone_name}",
|
|
f"--set-target={target}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
_reload()
|
|
logger.info("Zone '%s' created (target=%s)", zone_name, target)
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone_name}
|
|
|
|
|
|
@registry.register(DELETE_FIREWALL_ZONES_DELETE)
|
|
def delete_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body or "zone" not in body:
|
|
raise ValueError("'zone' is required")
|
|
zone = body["zone"]
|
|
available = run(["firewall-cmd", "--get-zones"], sudo=True).split()
|
|
if zone not in available:
|
|
raise NotFoundError(f"Zone '{zone}' does not exist")
|
|
run(["firewall-cmd", f"--zone={zone}", "--delete", "--permanent"], sudo=True)
|
|
_reload()
|
|
logger.info("Zone '%s' deleted", zone)
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone}
|
|
|
|
|
|
@registry.register(POST_FIREWALL_ZONES_INTERFACES)
|
|
def set_zone_interfaces(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone = body.get("zone", "").strip()
|
|
interfaces = body.get("interfaces", [])
|
|
if not zone:
|
|
raise ValueError("'zone' is required")
|
|
if zone not in run(["firewall-cmd", "--get-zones"], sudo=True).split():
|
|
raise NotFoundError(f"Zone '{zone}' does not exist")
|
|
|
|
# Determine old zone for each interface being reassigned
|
|
active_raw = run(["firewall-cmd", "--get-active-zones"], sudo=True)
|
|
active = _parse_active_zones(active_raw)
|
|
|
|
for iface in interfaces:
|
|
# Find which zone currently owns this interface
|
|
old_zone = None
|
|
for az, az_ifaces in active.items():
|
|
if iface in az_ifaces:
|
|
old_zone = az
|
|
break
|
|
# Remove from old zone (if different from target)
|
|
if old_zone and old_zone != zone:
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={old_zone}",
|
|
"--remove-interface=" + iface,
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
# Add to target zone
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone}",
|
|
"--add-interface=" + iface,
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
|
|
_reload()
|
|
|
|
# Update config
|
|
cfg = _get_config()
|
|
cfg.setdefault("zones", {})
|
|
cfg["zones"].setdefault(zone, {})
|
|
cfg["zones"][zone]["interfaces"] = list(interfaces)
|
|
# Remove interface from any old zone in config
|
|
for old_zone_name, old_zone_cfg in cfg["zones"].items():
|
|
if old_zone_name == zone:
|
|
continue
|
|
old_ifaces = old_zone_cfg.get("interfaces", [])
|
|
new_ifaces = [i for i in old_ifaces if i not in interfaces]
|
|
if len(new_ifaces) < len(old_ifaces):
|
|
if new_ifaces:
|
|
old_zone_cfg["interfaces"] = new_ifaces
|
|
elif "interfaces" in old_zone_cfg:
|
|
del old_zone_cfg["interfaces"]
|
|
_save_config(cfg)
|
|
|
|
logger.info("Zone '%s' interfaces set to %s", zone, interfaces)
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone, "interfaces": interfaces}
|
|
|
|
|
|
@registry.register(POST_FIREWALL_ZONES_SERVICES)
|
|
def set_zone_services(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone = body.get("zone", "").strip()
|
|
services = body.get("services", [])
|
|
if not zone:
|
|
raise ValueError("'zone' is required")
|
|
if zone not in run(["firewall-cmd", "--get-zones"], sudo=True).split():
|
|
raise NotFoundError(f"Zone '{zone}' does not exist")
|
|
current = _parse_zone_output(
|
|
zone, run(["firewall-cmd", f"--zone={zone}", "--list-all"], sudo=True)
|
|
).get("services", [])
|
|
for svc in current:
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone}",
|
|
f"--remove-service={svc}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
check=False,
|
|
)
|
|
for svc in services:
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone}",
|
|
f"--add-service={svc}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
_reload()
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone, "services": services}
|
|
|
|
|
|
@registry.register(POST_FIREWALL_RICH_RULES_ADD)
|
|
def add_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone = body.get("zone", "").strip()
|
|
rule = body.get("rule", "").strip()
|
|
if not zone or not rule:
|
|
raise ValueError("'zone' and 'rule' are required")
|
|
if zone not in run(["firewall-cmd", "--get-zones"], sudo=True).split():
|
|
raise NotFoundError(f"Zone '{zone}' does not exist")
|
|
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone}",
|
|
"--add-rich-rule=" + rule,
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
_reload()
|
|
cfg = _get_config()
|
|
cfg.setdefault("zones", {}).setdefault(zone, {}).setdefault("rich_rules", [])
|
|
rule_id = uuid4().hex[:8]
|
|
entry = {"id": rule_id, "rule": rule}
|
|
cfg["zones"][zone]["rich_rules"].append(entry)
|
|
_save_config(cfg)
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone, "id": rule_id, "rule": rule}
|
|
|
|
|
|
@registry.register(DELETE_FIREWALL_RICH_RULES_REMOVE)
|
|
def remove_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone = body.get("zone", "").strip()
|
|
rule_id = body.get("id", "").strip()
|
|
if not zone or not rule_id:
|
|
raise ValueError("'zone' and 'id' are required")
|
|
if zone not in run(["firewall-cmd", "--get-zones"], sudo=True).split():
|
|
raise NotFoundError(f"Zone '{zone}' does not exist")
|
|
cfg = _get_config()
|
|
zone_cfg = cfg.get("zones", {}).get(zone, {})
|
|
entry = None
|
|
for r in zone_cfg.get("rich_rules", []):
|
|
if r.get("id") == rule_id:
|
|
entry = r
|
|
break
|
|
if entry is None:
|
|
raise NotFoundError(f"Rich rule '{rule_id}' not found in zone '{zone}'")
|
|
rule = entry["rule"]
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone}",
|
|
"--remove-rich-rule=" + rule,
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
_reload()
|
|
zone_cfg["rich_rules"] = [
|
|
r for r in zone_cfg.get("rich_rules", []) if r.get("id") != rule_id
|
|
]
|
|
_save_config(cfg)
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone, "id": rule_id}
|
|
|
|
|
|
@registry.register(GET_FIREWALL_RICH_RULES)
|
|
def list_rich_rules(_request: Any, body: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
if not body or "zone" not in body:
|
|
raise ValueError("'zone' is required")
|
|
zone = body["zone"]
|
|
fw = _get_fw_state()
|
|
rich_rules = fw.get("rich_rules", {})
|
|
cfg = _get_config()
|
|
cfg_entries = cfg.get("zones", {}).get(zone, {}).get("rich_rules", [])
|
|
result: list[dict[str, Any]] = []
|
|
zone_rules = rich_rules.get(zone, [])
|
|
for rule_str in zone_rules:
|
|
matched = next((e for e in cfg_entries if e.get("rule") == rule_str), None)
|
|
if matched:
|
|
result.append({"id": matched["id"], "rule": rule_str})
|
|
else:
|
|
result.append({"rule": rule_str})
|
|
return result
|
|
|
|
|
|
@registry.register(POST_FIREWALL_MASQUERADE)
|
|
def set_masquerade(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone = body.get("zone", "").strip()
|
|
enable = body.get("enable")
|
|
if not zone or enable is None:
|
|
raise ValueError("'zone' and 'enable' (bool) are required")
|
|
action = "--add-masquerade" if enable else "--remove-masquerade"
|
|
run(["firewall-cmd", f"--zone={zone}", action, "--permanent"], sudo=True)
|
|
_reload()
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone, "masquerade": bool(enable)}
|
|
|
|
|
|
@registry.register(POST_FIREWALL_FORWARD_PORT_ADD)
|
|
def add_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone = body.get("zone", "").strip()
|
|
port = body.get("port")
|
|
proto = body.get("proto", "").strip()
|
|
toaddr = body.get("toaddr")
|
|
toport = body.get("toport")
|
|
if not zone or port is None or not proto:
|
|
raise ValueError("'zone', 'port', and 'proto' are required")
|
|
|
|
fwd = f"port={port}/proto={proto}"
|
|
if toaddr and toport:
|
|
fwd += f"/toaddr={toaddr}/toport={toport}"
|
|
elif toport:
|
|
fwd += f"/toport={toport}"
|
|
elif toaddr:
|
|
fwd += f"/toaddr={toaddr}"
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone}",
|
|
f"--add-forward-port={fwd}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
_reload()
|
|
fp_id = uuid4().hex[:8]
|
|
entry: dict[str, Any] = {"id": fp_id, "port": int(port), "proto": proto}
|
|
if toaddr:
|
|
entry["toaddr"] = toaddr
|
|
if toport:
|
|
entry["toport"] = int(toport)
|
|
cfg = _get_config()
|
|
cfg.setdefault("zones", {}).setdefault(zone, {}).setdefault("forward_ports", [])
|
|
cfg["zones"][zone]["forward_ports"].append(entry)
|
|
_save_config(cfg)
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone, "id": fp_id, "port": int(port), "proto": proto}
|
|
|
|
|
|
@registry.register(DELETE_FIREWALL_FORWARD_PORT_REMOVE)
|
|
def remove_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
zone = body.get("zone", "").strip()
|
|
port = body.get("port")
|
|
proto = body.get("proto", "").strip()
|
|
if not zone or port is None or not proto:
|
|
raise ValueError("'zone', 'port', and 'proto' are required")
|
|
available = run(["firewall-cmd", "--get-zones"], sudo=True).split()
|
|
if zone not in available:
|
|
raise NotFoundError(f"Zone '{zone}' does not exist")
|
|
fwd = f"port={port}/proto={proto}"
|
|
cfg = _get_config()
|
|
fps = cfg.get("zones", {}).get(zone, {}).get("forward_ports", [])
|
|
found = False
|
|
for fp in fps:
|
|
if fp.get("port") == port and fp.get("proto") == proto:
|
|
found = True
|
|
if fp.get("toaddr") and fp.get("toport"):
|
|
fwd += f"/toaddr={fp['toaddr']}/toport={fp['toport']}"
|
|
elif fp.get("toport"):
|
|
fwd += f"/toport={fp['toport']}"
|
|
elif fp.get("toaddr"):
|
|
fwd += f"/toaddr={fp['toaddr']}"
|
|
break
|
|
if not found:
|
|
raise NotFoundError(f"Forward port {port}/{proto} not found in zone '{zone}'")
|
|
run(
|
|
[
|
|
"firewall-cmd",
|
|
f"--zone={zone}",
|
|
f"--remove-forward-port={fwd}",
|
|
"--permanent",
|
|
],
|
|
sudo=True,
|
|
)
|
|
_reload()
|
|
cfg.setdefault("zones", {}).setdefault(zone, {})
|
|
cfg["zones"][zone]["forward_ports"] = [
|
|
fp for fp in fps if not (fp.get("port") == port and fp.get("proto") == proto)
|
|
]
|
|
_save_config(cfg)
|
|
refresh_state(["firewall"])
|
|
return {"zone": zone, "port": int(port), "proto": proto}
|
|
|
|
|
|
@registry.register(GET_FIREWALL_STATE)
|
|
def get_state(_request: Any, _body: Any) -> dict[str, Any]:
|
|
fw = _get_state()
|
|
if fw is None:
|
|
return {}
|
|
return {
|
|
"active_zones": fw.get("active_zones", {}),
|
|
"interfaces": fw.get("interfaces", []),
|
|
"available_services": fw.get("available_services", []),
|
|
"zones": fw.get("zones", {}),
|
|
"rich_rules": fw.get("rich_rules", {}),
|
|
"timestamp": fw.get("timestamp", ""),
|
|
}
|