78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""Safety guardrails for mutating firewall operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from opnsense_mcp.audit import audit_log
|
|
from opnsense_mcp.config import settings
|
|
|
|
|
|
class GuardError(Exception):
|
|
"""Raised when a safety guard blocks an operation."""
|
|
|
|
|
|
def require_credentials() -> None:
|
|
if not settings.api_key or not settings.api_secret:
|
|
raise GuardError(
|
|
"OPNSENSE_API_KEY and OPNSENSE_API_SECRET must be set. "
|
|
"Create an API key in OPNsense: System → Access → Users."
|
|
)
|
|
|
|
|
|
def require_writes(action: str, payload: dict[str, Any] | None = None) -> None:
|
|
require_credentials()
|
|
if settings.dry_run:
|
|
audit_log(action, {"payload": payload or {}, "blocked": "dry_run"}, dry_run=True)
|
|
return
|
|
if not settings.allow_writes:
|
|
raise GuardError(
|
|
f"Write operation '{action}' blocked. Set OPNSENSE_ALLOW_WRITES=true to enable."
|
|
)
|
|
|
|
|
|
def is_dry_run_response(action: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
audit_log(action, {"payload": payload or {}}, dry_run=True)
|
|
return {
|
|
"dry_run": True,
|
|
"action": action,
|
|
"payload": payload,
|
|
"message": "Dry run — no changes applied. Set OPNSENSE_DRY_RUN=false to execute.",
|
|
}
|
|
|
|
|
|
def validate_firewall_rule(rule: dict[str, Any], *, is_new: bool = False) -> None:
|
|
"""Reject dangerous or mis-tagged firewall rules."""
|
|
description = str(rule.get("description", ""))
|
|
if is_new and settings.require_description_prefix:
|
|
prefix = settings.require_description_prefix
|
|
if prefix and not description.lower().startswith(prefix.lower()):
|
|
raise GuardError(
|
|
f"New rules must have description starting with '{prefix}'. "
|
|
f"Got: '{description}'"
|
|
)
|
|
|
|
if not settings.block_wan_inbound_any:
|
|
return
|
|
|
|
interface = str(rule.get("interface", rule.get("interface_name", ""))).lower()
|
|
action = str(rule.get("action", "pass")).lower()
|
|
source = str(rule.get("source_net", rule.get("source", "any"))).lower()
|
|
dest = str(rule.get("destination_net", rule.get("destination", "any"))).lower()
|
|
|
|
wan_markers = ("wan", "opt", "pppoe", "dhcp")
|
|
is_wan = any(m in interface for m in wan_markers) or interface in ("", "any")
|
|
|
|
if action in ("pass", "allow") and is_wan:
|
|
if source in ("any", "", "0.0.0.0/0", "::/0") and dest not in ("any", "", "this firewall"):
|
|
raise GuardError(
|
|
"Blocked: inbound WAN allow rule with source 'any' is not permitted. "
|
|
"Use a specific source CIDR or disable OPNSENSE_BLOCK_WAN_INBOUND_ANY."
|
|
)
|
|
|
|
|
|
def validate_alias_mutation(name: str) -> None:
|
|
protected = {"bogons", "bogonsv6", "virusprot", "sshlockout", "__wan_network"}
|
|
if name.lower() in protected:
|
|
raise GuardError(f"Alias '{name}' is protected and cannot be modified via MCP.")
|