138 lines
8.0 KiB
Python
138 lines
8.0 KiB
Python
"""Firewall filter rule tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
from pfsense_mcp.audit import audit_log
|
|
from pfsense_mcp.client import get_client
|
|
from pfsense_mcp.config import settings
|
|
from pfsense_mcp.guards import (
|
|
is_dry_run_response,
|
|
require_writes,
|
|
validate_firewall_rule,
|
|
)
|
|
|
|
|
|
def register(mcp: FastMCP) -> None:
|
|
@mcp.tool()
|
|
def pfsense_list_firewall_rules(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""List firewall filter rules. query: comma-separated filters, e.g. 'interface=wan' or 'descr__contains=mcp'."""
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.get("firewall/rules", params=client.list_params(limit, offset, query))
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_get_firewall_rule(rule_id: int, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Get a single firewall rule by numeric id."""
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.get("firewall/rule", params={"id": rule_id})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_add_firewall_rule(rule: dict[str, Any], apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Add a firewall rule. Fields use pfSense REST v2 names (type, interface, ipprotocol, protocol, source, destination, descr, ...). descr must start with the configured prefix (default 'mcp:'). Set apply=true to apply immediately."""
|
|
validate_firewall_rule(rule, is_new=True)
|
|
payload = dict(rule)
|
|
if settings.dry_run:
|
|
return json.dumps(is_dry_run_response("add_firewall_rule", payload), indent=2)
|
|
require_writes("add_firewall_rule", payload)
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.post("firewall/rule", data=payload, params={"apply": apply})
|
|
audit_log("add_firewall_rule", {"rule": rule, "apply": apply, "result": result})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_update_firewall_rule(rule_id: int, rule: dict[str, Any], apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Update an existing firewall rule by id (PATCH semantics — only send changed fields)."""
|
|
validate_firewall_rule(rule, is_new=False)
|
|
payload = dict(rule)
|
|
payload["id"] = rule_id
|
|
if settings.dry_run:
|
|
return json.dumps(is_dry_run_response("update_firewall_rule", payload), indent=2)
|
|
require_writes("update_firewall_rule", payload)
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.patch("firewall/rule", data=payload, params={"apply": apply})
|
|
audit_log("update_firewall_rule", {"id": rule_id, "rule": rule, "result": result})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_delete_firewall_rule(rule_id: int, apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Delete a firewall rule by id."""
|
|
payload = {"id": rule_id}
|
|
if settings.dry_run:
|
|
return json.dumps(is_dry_run_response("delete_firewall_rule", payload), indent=2)
|
|
require_writes("delete_firewall_rule", payload)
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.delete("firewall/rule", params={"id": rule_id, "apply": apply})
|
|
audit_log("delete_firewall_rule", payload | {"result": result})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_toggle_firewall_rule(rule_id: int, disabled: bool, apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Enable or disable a firewall rule. disabled=true disables the rule."""
|
|
payload = {"id": rule_id, "disabled": disabled}
|
|
if settings.dry_run:
|
|
return json.dumps(is_dry_run_response("toggle_firewall_rule", payload), indent=2)
|
|
require_writes("toggle_firewall_rule", payload)
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.patch("firewall/rule", data=payload, params={"apply": apply})
|
|
audit_log("toggle_firewall_rule", payload | {"result": result})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_apply_firewall(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Apply pending firewall changes."""
|
|
if settings.dry_run:
|
|
return json.dumps(is_dry_run_response("apply_firewall", {}), indent=2)
|
|
require_writes("apply_firewall", {})
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.post("firewall/apply")
|
|
audit_log("apply_firewall", {"result": result})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_get_firewall_apply_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Check whether firewall changes are pending application."""
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.get("firewall/apply")
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_list_firewall_states(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""List active firewall states. query example: 'source__contains=10.77'."""
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.get("firewall/states", params=client.list_params(limit, offset, query))
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_kill_firewall_states(source: str, protocol: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""Kill firewall states originating from a source IP/CIDR."""
|
|
payload: dict[str, Any] = {"source": source}
|
|
if protocol:
|
|
payload["protocol"] = protocol
|
|
if settings.dry_run:
|
|
return json.dumps(is_dry_run_response("kill_states", payload), indent=2)
|
|
require_writes("kill_states", payload)
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.delete("firewall/states", params=payload)
|
|
audit_log("kill_states", payload | {"result": result})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_list_firewall_schedules(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""List firewall schedules."""
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.get("firewall/schedules", params={"limit": limit, "offset": offset})
|
|
return json.dumps(result, indent=2, default=str)
|
|
|
|
@mcp.tool()
|
|
def pfsense_list_virtual_ips(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str:
|
|
"""List virtual IPs (CARP, IP alias, proxy ARP)."""
|
|
client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl)
|
|
result = client.get("firewall/virtual_ips", params={"limit": limit, "offset": offset})
|
|
return json.dumps(result, indent=2, default=str)
|