nearxos/pfsense-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial pfsense-mcp: REST API v2 based MCP server (72 tools)

Browse Source
This commit is contained in:
root
2026-06-12 18:59:15 +00:00
commit ee6c626a80
22 changed files with 1438 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/pfsense_mcp/tools/__init__.py Normal file
View File

@@ -0,0 +1,29 @@
"""Tool package registration."""
from mcp.server.fastmcp import FastMCP
from pfsense_mcp.tools import (
aliases,
dhcp,
diagnostics,
dns,
firewall,
interfaces,
nat,
routes,
system,
vpn,
)
def register_all(mcp: FastMCP) -> None:
system.register(mcp)
interfaces.register(mcp)
firewall.register(mcp)
aliases.register(mcp)
nat.register(mcp)
dhcp.register(mcp)
diagnostics.register(mcp)
routes.register(mcp)
vpn.register(mcp)
dns.register(mcp)

80
src/pfsense_mcp/tools/aliases.py Normal file
View File

@@ -0,0 +1,80 @@
"""Firewall alias 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_alias_mutation
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_list_aliases(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""List firewall aliases. query example: 'name__contains=block'."""
client = get_client()
result = client.get("firewall/aliases", params=client.list_params(limit, offset, query))
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_alias(alias_id: int) -> str:
"""Get an alias by numeric id."""
client = get_client()
result = client.get("firewall/alias", params={"id": alias_id})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_add_alias(
name: str,
alias_type: str,
addresses: list[str],
description: str = "",
apply: bool = False,
) -> str:
"""Add a firewall alias. alias_type: host, network, or port. addresses: list of IPs/CIDRs/ports."""
validate_alias_mutation(name)
payload: dict[str, Any] = {
"name": name,
"type": alias_type,
"address": addresses,
"descr": description,
}
if settings.dry_run:
return json.dumps(is_dry_run_response("add_alias", payload), indent=2)
require_writes("add_alias", payload)
client = get_client()
result = client.post("firewall/alias", data=payload, params={"apply": apply})
audit_log("add_alias", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_update_alias(alias_id: int, fields: dict[str, Any], apply: bool = False) -> str:
"""Update an alias by id (PATCH — send only changed fields, e.g. {'address': [...]})."""
if "name" in fields:
validate_alias_mutation(str(fields["name"]))
payload = dict(fields)
payload["id"] = alias_id
if settings.dry_run:
return json.dumps(is_dry_run_response("update_alias", payload), indent=2)
require_writes("update_alias", payload)
client = get_client()
result = client.patch("firewall/alias", data=payload, params={"apply": apply})
audit_log("update_alias", {"id": alias_id, "fields": fields, "result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_delete_alias(alias_id: int, apply: bool = False) -> str:
"""Delete an alias by id."""
payload = {"id": alias_id}
if settings.dry_run:
return json.dumps(is_dry_run_response("delete_alias", payload), indent=2)
require_writes("delete_alias", payload)
client = get_client()
result = client.delete("firewall/alias", params={"id": alias_id, "apply": apply})
audit_log("delete_alias", payload | {"result": result})
return json.dumps(result, indent=2, default=str)

79
src/pfsense_mcp/tools/dhcp.py Normal file
View File

@@ -0,0 +1,79 @@
"""DHCP server tools: leases, server config, static mappings."""
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
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_list_dhcp_leases(limit: int = 200, offset: int = 0, query: str = "") -> str:
"""List DHCP leases. query example: 'hostname__contains=nas' or 'ip__contains=10.77'."""
client = get_client()
result = client.get(
"status/dhcp_server/leases", params=client.list_params(limit, offset, query)
)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_dhcp_servers(limit: int = 50, offset: int = 0) -> str:
"""List DHCP server configurations per interface (ranges, options, enabled state)."""
client = get_client()
result = client.get("services/dhcp_servers", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_dhcp_static_mappings(
interface_id: str, limit: int = 200, offset: int = 0
) -> str:
"""List DHCP static mappings (reservations) for a DHCP server. interface_id: parent interface (e.g. 'lan')."""
client = get_client()
result = client.get(
"services/dhcp_server/static_mappings",
params={"parent_id": interface_id, "limit": limit, "offset": offset},
)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_add_dhcp_static_mapping(
interface_id: str,
mac: str,
ip: str,
hostname: str = "",
description: str = "",
) -> str:
"""Add a DHCP static mapping (reservation) on an interface's DHCP server."""
payload: dict[str, Any] = {
"parent_id": interface_id,
"mac": mac,
"ipaddr": ip,
"hostname": hostname,
"descr": description,
}
if settings.dry_run:
return json.dumps(is_dry_run_response("add_dhcp_static_mapping", payload), indent=2)
require_writes("add_dhcp_static_mapping", payload)
client = get_client()
result = client.post("services/dhcp_server/static_mapping", data=payload)
audit_log("add_dhcp_static_mapping", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_delete_dhcp_static_mapping(interface_id: str, mapping_id: int) -> str:
"""Delete a DHCP static mapping by id."""
payload = {"parent_id": interface_id, "id": mapping_id}
if settings.dry_run:
return json.dumps(is_dry_run_response("delete_dhcp_static_mapping", payload), indent=2)
require_writes("delete_dhcp_static_mapping", payload)
client = get_client()
result = client.delete("services/dhcp_server/static_mapping", params=payload)
audit_log("delete_dhcp_static_mapping", payload | {"result": result})
return json.dumps(result, indent=2, default=str)

68
src/pfsense_mcp/tools/diagnostics.py Normal file
View File

@@ -0,0 +1,68 @@
"""Diagnostics: ARP table, logs, system activity."""
from __future__ import annotations
import json
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
def _get_log(scope: str, limit: int, offset: int, query: str) -> str:
client = get_client()
result = client.get(f"status/logs/{scope}", params=client.list_params(limit, offset, query))
return json.dumps(result, indent=2, default=str)
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_get_arp_table(limit: int = 200, offset: int = 0, query: str = "") -> str:
"""Get the ARP table. query example: 'ip__contains=10.77' or 'hostname__contains=nas'."""
client = get_client()
result = client.get(
"diagnostics/arp_table", params=client.list_params(limit, offset, query)
)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_delete_arp_entry(entry_id: int) -> str:
"""Delete a single ARP table entry by id (from pfsense_get_arp_table)."""
payload = {"id": entry_id}
if settings.dry_run:
return json.dumps(is_dry_run_response("delete_arp_entry", payload), indent=2)
require_writes("delete_arp_entry", payload)
client = get_client()
result = client.delete("diagnostics/arp_table/entry", params=payload)
audit_log("delete_arp_entry", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_firewall_log(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""Get firewall log entries (blocked/passed traffic)."""
return _get_log("firewall", limit, offset, query)
@mcp.tool()
def pfsense_get_system_log(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""Get system log entries."""
return _get_log("system", limit, offset, query)
@mcp.tool()
def pfsense_get_dhcp_log(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""Get DHCP daemon log entries."""
return _get_log("dhcp", limit, offset, query)
@mcp.tool()
def pfsense_get_auth_log(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""Get authentication log entries (logins, failures)."""
return _get_log("authentication", limit, offset, query)
@mcp.tool()
def pfsense_get_system_activity() -> str:
"""Get system activity (top-style process list)."""
client = get_client()
result = client.get("diagnostics/system_activity")
return json.dumps(result, indent=2, default=str)

75
src/pfsense_mcp/tools/dns.py Normal file
View File

@@ -0,0 +1,75 @@
"""DNS resolver (Unbound) tools: host overrides, apply."""
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
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_list_dns_host_overrides(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""List DNS resolver host overrides (local DNS records)."""
client = get_client()
result = client.get(
"services/dns_resolver/host_overrides",
params=client.list_params(limit, offset, query),
)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_add_dns_host_override(
host: str,
domain: str,
ip: list[str],
description: str = "",
apply: bool = False,
) -> str:
"""Add a DNS resolver host override. ip: list of addresses for the record."""
payload: dict[str, Any] = {
"host": host,
"domain": domain,
"ip": ip,
"descr": description,
}
if settings.dry_run:
return json.dumps(is_dry_run_response("add_dns_host_override", payload), indent=2)
require_writes("add_dns_host_override", payload)
client = get_client()
result = client.post(
"services/dns_resolver/host_override", data=payload, params={"apply": apply}
)
audit_log("add_dns_host_override", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_delete_dns_host_override(override_id: int, apply: bool = False) -> str:
"""Delete a DNS host override by id."""
payload = {"id": override_id}
if settings.dry_run:
return json.dumps(is_dry_run_response("delete_dns_host_override", payload), indent=2)
require_writes("delete_dns_host_override", payload)
client = get_client()
result = client.delete(
"services/dns_resolver/host_override", params={"id": override_id, "apply": apply}
)
audit_log("delete_dns_host_override", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_apply_dns_resolver() -> str:
"""Apply pending DNS resolver changes."""
if settings.dry_run:
return json.dumps(is_dry_run_response("apply_dns_resolver", {}), indent=2)
require_writes("apply_dns_resolver", {})
client = get_client()
result = client.post("services/dns_resolver/apply")
audit_log("apply_dns_resolver", {"result": result})
return json.dumps(result, indent=2, default=str)

137
src/pfsense_mcp/tools/firewall.py Normal file
View File

@@ -0,0 +1,137 @@
"""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 = "") -> str:
"""List firewall filter rules. query: comma-separated filters, e.g. 'interface=wan' or 'descr__contains=mcp'."""
client = get_client()
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) -> str:
"""Get a single firewall rule by numeric id."""
client = get_client()
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) -> 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()
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) -> 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()
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) -> 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()
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) -> 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()
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() -> 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()
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() -> str:
"""Check whether firewall changes are pending application."""
client = get_client()
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 = "") -> str:
"""List active firewall states. query example: 'source__contains=10.77'."""
client = get_client()
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 = "") -> 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()
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) -> str:
"""List firewall schedules."""
client = get_client()
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) -> str:
"""List virtual IPs (CARP, IP alias, proxy ARP)."""
client = get_client()
result = client.get("firewall/virtual_ips", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)

53
src/pfsense_mcp/tools/interfaces.py Normal file
View File

@@ -0,0 +1,53 @@
"""Network interface tools."""
from __future__ import annotations
import json
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
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_get_interface_status() -> str:
"""Get live status of all interfaces (up/down, IPs, media, in/out bytes)."""
client = get_client()
result = client.get("status/interfaces")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_interfaces(limit: int = 100, offset: int = 0) -> str:
"""List configured interfaces (assignments and settings)."""
client = get_client()
result = client.get("interfaces", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_vlans(limit: int = 100, offset: int = 0) -> str:
"""List VLAN interface configurations."""
client = get_client()
result = client.get("interface/vlans", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_carp_status() -> str:
"""Get CARP (high availability) status and virtual IPs."""
client = get_client()
result = client.get("status/carp")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_apply_interfaces() -> str:
"""Apply pending interface configuration changes."""
if settings.dry_run:
return json.dumps(is_dry_run_response("apply_interfaces", {}), indent=2)
require_writes("apply_interfaces", {})
client = get_client()
result = client.post("interface/apply")
audit_log("apply_interfaces", {"result": result})
return json.dumps(result, indent=2, default=str)

89
src/pfsense_mcp/tools/nat.py Normal file
View File

@@ -0,0 +1,89 @@
"""NAT tools: port forwards, outbound NAT, 1:1 NAT."""
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
def register(mcp: FastMCP) -> None:
# ── Port forwards (destination NAT) ─────────────────────────────────
@mcp.tool()
def pfsense_list_port_forwards(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""List port forward (destination NAT) rules."""
client = get_client()
result = client.get(
"firewall/nat/port_forwards", params=client.list_params(limit, offset, query)
)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_add_port_forward(rule: dict[str, Any], apply: bool = False) -> str:
"""Add a port forward. Fields: interface, protocol, source, destination, target, local_port, descr, ..."""
payload = dict(rule)
if settings.dry_run:
return json.dumps(is_dry_run_response("add_port_forward", payload), indent=2)
require_writes("add_port_forward", payload)
client = get_client()
result = client.post("firewall/nat/port_forward", data=payload, params={"apply": apply})
audit_log("add_port_forward", {"rule": rule, "result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_update_port_forward(rule_id: int, fields: dict[str, Any], apply: bool = False) -> str:
"""Update a port forward by id (PATCH — only changed fields)."""
payload = dict(fields)
payload["id"] = rule_id
if settings.dry_run:
return json.dumps(is_dry_run_response("update_port_forward", payload), indent=2)
require_writes("update_port_forward", payload)
client = get_client()
result = client.patch("firewall/nat/port_forward", data=payload, params={"apply": apply})
audit_log("update_port_forward", {"id": rule_id, "fields": fields, "result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_delete_port_forward(rule_id: int, apply: bool = False) -> str:
"""Delete a port forward by id."""
payload = {"id": rule_id}
if settings.dry_run:
return json.dumps(is_dry_run_response("delete_port_forward", payload), indent=2)
require_writes("delete_port_forward", payload)
client = get_client()
result = client.delete("firewall/nat/port_forward", params={"id": rule_id, "apply": apply})
audit_log("delete_port_forward", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
# ── Outbound NAT ────────────────────────────────────────────────────
@mcp.tool()
def pfsense_get_outbound_nat_mode() -> str:
"""Get the outbound NAT mode (automatic, hybrid, manual, disabled)."""
client = get_client()
result = client.get("firewall/nat/outbound/mode")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_outbound_nat_mappings(limit: int = 100, offset: int = 0) -> str:
"""List outbound NAT mappings."""
client = get_client()
result = client.get(
"firewall/nat/outbound/mappings", params={"limit": limit, "offset": offset}
)
return json.dumps(result, indent=2, default=str)
# ── 1:1 NAT ─────────────────────────────────────────────────────────
@mcp.tool()
def pfsense_list_one_to_one_mappings(limit: int = 100, offset: int = 0) -> str:
"""List 1:1 NAT mappings."""
client = get_client()
result = client.get(
"firewall/nat/one_to_one/mappings", params={"limit": limit, "offset": offset}
)
return json.dumps(result, indent=2, default=str)

77
src/pfsense_mcp/tools/routes.py Normal file
View File

@@ -0,0 +1,77 @@
"""Routing and gateway 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
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_get_gateway_status() -> str:
"""Get gateway status: latency, packet loss, online/offline."""
client = get_client()
result = client.get("status/gateways")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_gateways(limit: int = 50, offset: int = 0) -> str:
"""List configured gateways."""
client = get_client()
result = client.get("routing/gateways", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_static_routes(limit: int = 100, offset: int = 0) -> str:
"""List static routes."""
client = get_client()
result = client.get("routing/static_routes", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_add_static_route(
network: str, gateway: str, description: str = "", apply: bool = False
) -> str:
"""Add a static route. network: CIDR, gateway: gateway name."""
payload: dict[str, Any] = {
"network": network,
"gateway": gateway,
"descr": description,
}
if settings.dry_run:
return json.dumps(is_dry_run_response("add_static_route", payload), indent=2)
require_writes("add_static_route", payload)
client = get_client()
result = client.post("routing/static_route", data=payload, params={"apply": apply})
audit_log("add_static_route", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_delete_static_route(route_id: int, apply: bool = False) -> str:
"""Delete a static route by id."""
payload = {"id": route_id}
if settings.dry_run:
return json.dumps(is_dry_run_response("delete_static_route", payload), indent=2)
require_writes("delete_static_route", payload)
client = get_client()
result = client.delete("routing/static_route", params={"id": route_id, "apply": apply})
audit_log("delete_static_route", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_apply_routing() -> str:
"""Apply pending routing changes."""
if settings.dry_run:
return json.dumps(is_dry_run_response("apply_routing", {}), indent=2)
require_writes("apply_routing", {})
client = get_client()
result = client.post("routing/apply")
audit_log("apply_routing", {"result": result})
return json.dumps(result, indent=2, default=str)

120
src/pfsense_mcp/tools/system.py Normal file
View File

@@ -0,0 +1,120 @@
"""System status, services, packages, and certificates."""
from __future__ import annotations
import json
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 GuardError, is_dry_run_response, require_writes
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_get_system_status() -> str:
"""Get system status: CPU, memory, disk, temperature, uptime."""
client = get_client()
result = client.get("status/system")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_version() -> str:
"""Get pfSense version and patch information."""
client = get_client()
result = client.get("system/version")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_services(limit: int = 100, offset: int = 0) -> str:
"""List system services with running/stopped status."""
client = get_client()
result = client.get("status/services", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_service_action(name: str, action: str) -> str:
"""Control a service. action: start, stop, restart."""
allowed = {"start", "stop", "restart"}
if action not in allowed:
raise GuardError(f"action must be one of: {', '.join(sorted(allowed))}")
payload = {"name": name, "action": action}
if settings.dry_run:
return json.dumps(is_dry_run_response("service_action", payload), indent=2)
require_writes("service_action", payload)
client = get_client()
result = client.post("status/service", data=payload)
audit_log("service_action", payload | {"result": result})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_packages(limit: int = 100, offset: int = 0) -> str:
"""List installed pfSense packages."""
client = get_client()
result = client.get("system/packages", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_available_packages(limit: int = 100, offset: int = 0, query: str = "") -> str:
"""List packages available to install. query example: 'name__contains=wireguard'."""
client = get_client()
result = client.get(
"system/package/available", params=client.list_params(limit, offset, query)
)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_certificates(limit: int = 100, offset: int = 0) -> str:
"""List certificates (no private keys returned by this tool)."""
client = get_client()
result = client.get("system/certificates", params={"limit": limit, "offset": offset})
if isinstance(result, dict) and isinstance(result.get("data"), list):
for cert in result["data"]:
if isinstance(cert, dict):
cert.pop("prv", None)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_config_history(limit: int = 20, offset: int = 0) -> str:
"""List configuration change history revisions (who/when/what)."""
client = get_client()
result = client.get(
"diagnostics/config_history/revisions",
params={"limit": limit, "offset": offset},
)
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_dns_settings() -> str:
"""Get system DNS server settings."""
client = get_client()
result = client.get("system/dns")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_hostname() -> str:
"""Get system hostname and domain."""
client = get_client()
result = client.get("system/hostname")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_reboot(confirm: bool = False) -> str:
"""Reboot the firewall. Requires confirm=true AND PFSENSE_ALLOW_WRITES=true."""
if not confirm:
return json.dumps(
{
"status": "confirmation_required",
"message": "This reboots the firewall. Call again with confirm=true.",
},
indent=2,
)
if settings.dry_run:
return json.dumps(is_dry_run_response("reboot", {}), indent=2)
require_writes("reboot", {})
client = get_client()
result = client.post("diagnostics/reboot")
audit_log("reboot", {"result": result})
return json.dumps(result, indent=2, default=str)

67
src/pfsense_mcp/tools/vpn.py Normal file
View File

@@ -0,0 +1,67 @@
"""VPN status tools: OpenVPN, IPsec, WireGuard (plugin)."""
from __future__ import annotations
import json
from mcp.server.fastmcp import FastMCP
from pfsense_mcp.client import get_client
def register(mcp: FastMCP) -> None:
@mcp.tool()
def pfsense_get_openvpn_status() -> str:
"""Get OpenVPN server/client status with connected clients."""
client = get_client()
result = client.get("status/openvpn/servers")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_openvpn_servers(limit: int = 50, offset: int = 0) -> str:
"""List OpenVPN server configurations."""
client = get_client()
result = client.get("vpn/openvpn/servers", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_openvpn_clients(limit: int = 50, offset: int = 0) -> str:
"""List OpenVPN client configurations."""
client = get_client()
result = client.get("vpn/openvpn/clients", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_ipsec_status() -> str:
"""Get IPsec tunnel status (phase 1/2 SAs)."""
client = get_client()
result = client.get("status/ipsec/sas")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_ipsec_phase1(limit: int = 50, offset: int = 0) -> str:
"""List configured IPsec phase 1 entries."""
client = get_client()
result = client.get("vpn/ipsec/phase1s", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_get_wireguard_status() -> str:
"""Get WireGuard status (requires the WireGuard package to be installed)."""
client = get_client()
result = client.get("vpn/wireguard/settings")
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_wireguard_tunnels(limit: int = 50, offset: int = 0) -> str:
"""List WireGuard tunnels (requires the WireGuard package)."""
client = get_client()
result = client.get("vpn/wireguard/tunnels", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)
@mcp.tool()
def pfsense_list_wireguard_peers(limit: int = 100, offset: int = 0) -> str:
"""List WireGuard peers (requires the WireGuard package)."""
client = get_client()
result = client.get("vpn/wireguard/peers", params={"limit": limit, "offset": offset})
return json.dumps(result, indent=2, default=str)