commit ee6c626a80f202709cab146140fa500b09fcb926 Author: root Date: Fri Jun 12 18:59:15 2026 +0000 Initial pfsense-mcp: REST API v2 based MCP server (72 tools) diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..9456319 --- /dev/null +++ b/.env.example @@ -0,0 +1,16 @@ +# pfSense REST API v2 connection +PFSENSE_URL=https://192.168.1.1 +# Preferred: API key from System -> REST API -> Keys +PFSENSE_API_KEY= +# Fallback: basic auth (requires BasicAuth enabled in REST API settings) +PFSENSE_USERNAME= +PFSENSE_PASSWORD= +PFSENSE_VERIFY_SSL=false +PFSENSE_TIMEOUT=60 + +# Safety controls +PFSENSE_ALLOW_WRITES=false +PFSENSE_DRY_RUN=false +PFSENSE_REQUIRE_DESCRIPTION_PREFIX=mcp: +PFSENSE_BLOCK_WAN_INBOUND_ANY=true +PFSENSE_AUDIT_LOG_PATH=/root/.hermes/logs/pfsense-mcp-audit.jsonl diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c068ad8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.env +.venv/ +dist/ +build/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..6b424e2 --- /dev/null +++ b/README.md @@ -0,0 +1,157 @@ +# pfSense MCP + +Model Context Protocol server for [pfSense](https://www.pfsense.org/) firewall management via the [REST API v2 package](https://pfrest.org/) (`pfSense-pkg-RESTAPI`). Sibling project to `opnsense-mcp` with the same safety model. + +## Features + +- **55+ tools** covering system status, interfaces, firewall rules, aliases, NAT, DHCP, diagnostics, routing, VPN (OpenVPN/IPsec/WireGuard), and DNS resolver +- **Safety guardrails**: writes disabled by default, dry-run mode, WAN any/any rule blocking, required description prefix +- **Audit logging**: all mutations logged to JSONL +- **`apply` control**: write tools accept `apply=true` to apply immediately, or batch changes and call the apply tool + +## Prerequisites + +pfSense does not ship a REST API. Install the free **REST API v2 package** on the firewall first: + +```sh +# On the pfSense shell (adjust version to your pfSense release — see pfrest.org) +pkg-static add https://github.com/jaredhendrickson13/pfsense-api/releases/latest/download/pfSense-2.8.0-pkg-RESTAPI.pkg +``` + +Then in the web UI: **System → REST API → Settings** (enable), and **System → REST API → Keys** to generate an API key. + +## Quick Start + +### 1. Install + +```bash +pip install -e /path/to/pfsense-mcp +``` + +Or with the Hermes venv: + +```bash +/usr/local/lib/hermes-agent/venv/bin/pip install -e /root/projects/pfsense-mcp +``` + +### 2. Configure environment + +```bash +cp .env.example .env +``` + +| Variable | Default | Description | +|----------|---------|-------------| +| `PFSENSE_URL` | `https://192.168.1.1` | Firewall base URL | +| `PFSENSE_API_KEY` | — | REST API key (preferred) | +| `PFSENSE_USERNAME` / `PFSENSE_PASSWORD` | — | Basic auth fallback | +| `PFSENSE_VERIFY_SSL` | `false` | Verify TLS certificate | +| `PFSENSE_ALLOW_WRITES` | `false` | Enable mutating operations | +| `PFSENSE_DRY_RUN` | `false` | Log writes without executing | +| `PFSENSE_REQUIRE_DESCRIPTION_PREFIX` | `mcp:` | Prefix for new firewall rules (`descr`) | +| `PFSENSE_BLOCK_WAN_INBOUND_ANY` | `true` | Block WAN any→any pass rules | + +### 3. Register with Hermes + +```yaml +mcp_servers: + pfsense: + command: /usr/local/lib/hermes-agent/venv/bin/pfsense-mcp + timeout: 120 + enabled: true + env: + PFSENSE_URL: "https://your-pfsense" + PFSENSE_API_KEY: "your-api-key" + PFSENSE_VERIFY_SSL: "false" + PFSENSE_ALLOW_WRITES: "false" +``` + +Then `hermes mcp test pfsense` and `hermes gateway restart`. + +### 4. Register with Cursor (`~/.cursor/mcp.json`) + +```json +{ + "mcpServers": { + "pfsense": { + "command": "pfsense-mcp", + "env": { + "PFSENSE_URL": "https://your-pfsense", + "PFSENSE_API_KEY": "your-api-key" + } + } + } +} +``` + +## Tool Categories + +### Read-only (always available with valid credentials) + +| Tool | Description | +|------|-------------| +| `pfsense_test_connection` | Verify API auth and connectivity | +| `pfsense_get_system_status` | CPU, memory, disk, uptime | +| `pfsense_get_interface_status` | Live interface status | +| `pfsense_list_firewall_rules` | Filter rules (with query filters) | +| `pfsense_list_firewall_states` | Active state table | +| `pfsense_list_aliases` | Firewall aliases | +| `pfsense_list_port_forwards` | Destination NAT rules | +| `pfsense_list_dhcp_leases` | DHCP lease table | +| `pfsense_get_arp_table` | ARP / IP→MAC | +| `pfsense_get_gateway_status` | Gateway health | +| `pfsense_get_firewall_log` | Firewall log | +| `pfsense_get_openvpn_status` | OpenVPN connected clients | +| `pfsense_get_ipsec_status` | IPsec SAs | +| `pfsense_list_wireguard_tunnels` | WireGuard (plugin) | +| `pfsense_list_dns_host_overrides` | Local DNS records | +| `pfsense_get_config_history` | Config change audit trail | + +### Write operations (require `PFSENSE_ALLOW_WRITES=true`) + +| Tool | Description | +|------|-------------| +| `pfsense_add_firewall_rule` | Add filter rule (descr prefix required) | +| `pfsense_toggle_firewall_rule` | Enable/disable rule | +| `pfsense_apply_firewall` | Apply pending changes | +| `pfsense_add_alias` / `pfsense_update_alias` | Manage aliases | +| `pfsense_add_port_forward` | Add destination NAT | +| `pfsense_add_dhcp_static_mapping` | DHCP reservation | +| `pfsense_add_dns_host_override` | Local DNS record | +| `pfsense_add_static_route` | Static route | +| `pfsense_service_action` | Start/stop/restart services | +| `pfsense_kill_firewall_states` | Kill states by source | +| `pfsense_reboot` | Reboot (double confirm) | + +### Query filtering + +List tools accept a `query` parameter with pfSense REST v2 query syntax: + +```text +query="interface=wan" exact match +query="descr__contains=mcp" substring +query="ip__startswith=10.77" prefix +query="interface=wan,protocol=tcp" multiple conditions +``` + +## Differences vs opnsense-mcp + +| Aspect | OPNsense | pfSense | +|--------|----------|---------| +| API | Built-in (`/api/module/controller/command`) | Package required (`/api/v2/...`) | +| Auth | key+secret basic auth | `X-API-Key` header or basic auth | +| IDs | UUIDs | Numeric ids (positional) | +| Apply | Savepoint + rollback workflow | `apply=true` param or apply endpoints | +| Rule description field | `description` | `descr` | + +## Security Notes + +- Never commit API keys; use environment variables or Hermes secrets. +- Start read-only (`PFSENSE_ALLOW_WRITES=false`). +- Create a dedicated pfSense user + API key with minimal privileges. +- Review the audit log at `PFSENSE_AUDIT_LOG_PATH`. +- pfSense numeric ids shift when objects are deleted — always list immediately before mutating by id. + +## License + +MIT diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b549156 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,25 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "pfsense-mcp" +version = "1.0.0" +description = "Model Context Protocol server for pfSense firewall management via REST API v2" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +authors = [{ name = "pfSense MCP" }] +keywords = ["mcp", "pfsense", "firewall", "hermes"] +dependencies = [ + "mcp>=1.0.0", + "httpx>=0.27.0", + "pydantic>=2.0.0", + "pydantic-settings>=2.0.0", +] + +[project.scripts] +pfsense-mcp = "pfsense_mcp.server:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/pfsense_mcp"] diff --git a/src/pfsense_mcp/__init__.py b/src/pfsense_mcp/__init__.py new file mode 100644 index 0000000..d29aae2 --- /dev/null +++ b/src/pfsense_mcp/__init__.py @@ -0,0 +1,3 @@ +"""pfSense MCP server — Model Context Protocol integration for pfSense firewalls.""" + +__version__ = "1.0.0" diff --git a/src/pfsense_mcp/__main__.py b/src/pfsense_mcp/__main__.py new file mode 100644 index 0000000..51657e1 --- /dev/null +++ b/src/pfsense_mcp/__main__.py @@ -0,0 +1,3 @@ +from pfsense_mcp.server import main + +main() diff --git a/src/pfsense_mcp/audit.py b/src/pfsense_mcp/audit.py new file mode 100644 index 0000000..381bc5e --- /dev/null +++ b/src/pfsense_mcp/audit.py @@ -0,0 +1,23 @@ +"""Append-only audit log for mutating operations.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from pfsense_mcp.config import settings + + +def audit_log(action: str, details: dict[str, Any], *, dry_run: bool = False) -> None: + entry = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "action": action, + "dry_run": dry_run, + **details, + } + path = Path(settings.audit_log_path) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(entry, default=str) + "\n") diff --git a/src/pfsense_mcp/client.py b/src/pfsense_mcp/client.py new file mode 100644 index 0000000..ff10c59 --- /dev/null +++ b/src/pfsense_mcp/client.py @@ -0,0 +1,134 @@ +"""HTTP client for the pfSense REST API v2 (pfSense-pkg-RESTAPI).""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx + +from pfsense_mcp.config import settings +from pfsense_mcp.guards import require_credentials + + +class PfsenseAPIError(Exception): + """Raised when the pfSense REST API returns an error.""" + + def __init__(self, message: str, status_code: int | None = None, body: Any = None): + super().__init__(message) + self.status_code = status_code + self.body = body + + +class PfsenseClient: + """Wrapper around pfSense REST API v2 endpoints (/api/v2/...).""" + + def __init__( + self, + url: str | None = None, + api_key: str | None = None, + *, + verify_ssl: bool | None = None, + timeout: float | None = None, + ): + self.base_url = (url or settings.url).rstrip("/") + self.api_key = api_key if api_key is not None else settings.api_key + self.verify_ssl = verify_ssl if verify_ssl is not None else settings.verify_ssl + self.timeout = timeout if timeout is not None else settings.timeout + + def _request( + self, + method: str, + endpoint: str, + *, + params: dict[str, Any] | None = None, + data: dict[str, Any] | None = None, + ) -> Any: + require_credentials() + url = f"{self.base_url}/api/v2/{endpoint.lstrip('/')}" + headers = {} + auth = None + if self.api_key: + headers["X-API-Key"] = self.api_key + else: + auth = (settings.username, settings.password) + + clean_params = None + if params: + clean_params = {k: v for k, v in params.items() if v is not None and v != ""} + + with httpx.Client(verify=self.verify_ssl, timeout=self.timeout) as client: + response = client.request( + method, + url, + headers=headers, + auth=auth, + params=clean_params, + json=data, + ) + if response.status_code >= 400: + try: + body = response.json() + except json.JSONDecodeError: + body = response.text + raise PfsenseAPIError( + f"pfSense API error {response.status_code} for {method} {url}", + status_code=response.status_code, + body=body, + ) + if not response.content: + return {} + try: + return response.json() + except json.JSONDecodeError: + return {"raw": response.text} + + def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any: + return self._request("GET", endpoint, params=params) + + def post( + self, + endpoint: str, + data: dict[str, Any] | None = None, + params: dict[str, Any] | None = None, + ) -> Any: + return self._request("POST", endpoint, params=params, data=data) + + def patch( + self, + endpoint: str, + data: dict[str, Any] | None = None, + params: dict[str, Any] | None = None, + ) -> Any: + return self._request("PATCH", endpoint, params=params, data=data) + + def delete(self, endpoint: str, params: dict[str, Any] | None = None) -> Any: + return self._request("DELETE", endpoint, params=params) + + def list_params( + self, + limit: int = 100, + offset: int = 0, + query: str = "", + ) -> dict[str, Any]: + """Build standard v2 collection query params. + + query: comma-separated field filters, e.g. "descr__contains=mcp,interface=wan". + """ + params: dict[str, Any] = {"limit": limit, "offset": offset} + if query: + for pair in query.split(","): + if "=" in pair: + key, value = pair.split("=", 1) + params[key.strip()] = value.strip() + return params + + +_client: PfsenseClient | None = None + + +def get_client() -> PfsenseClient: + global _client + if _client is None: + _client = PfsenseClient() + return _client diff --git a/src/pfsense_mcp/config.py b/src/pfsense_mcp/config.py new file mode 100644 index 0000000..6649a43 --- /dev/null +++ b/src/pfsense_mcp/config.py @@ -0,0 +1,28 @@ +"""Configuration loaded from environment variables.""" + +from __future__ import annotations + +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_prefix="PFSENSE_", extra="ignore") + + url: str = "https://192.168.1.1" + # REST API v2 key (preferred). Generate in pfSense: System → REST API → Keys. + api_key: str = "" + # Fallback: local user basic auth (requires auth_methods to include BasicAuth). + username: str = "" + password: str = "" + verify_ssl: bool = False + timeout: float = 60.0 + + # Safety controls + allow_writes: bool = False + dry_run: bool = False + require_description_prefix: str = "mcp:" + block_wan_inbound_any: bool = True + audit_log_path: str = "/root/.hermes/logs/pfsense-mcp-audit.jsonl" + + +settings = Settings() diff --git a/src/pfsense_mcp/guards.py b/src/pfsense_mcp/guards.py new file mode 100644 index 0000000..334d343 --- /dev/null +++ b/src/pfsense_mcp/guards.py @@ -0,0 +1,85 @@ +"""Safety guardrails for mutating firewall operations.""" + +from __future__ import annotations + +from typing import Any + +from pfsense_mcp.audit import audit_log +from pfsense_mcp.config import settings + + +class GuardError(Exception): + """Raised when a safety guard blocks an operation.""" + + +def require_credentials() -> None: + if not settings.api_key and not (settings.username and settings.password): + raise GuardError( + "Set PFSENSE_API_KEY (System → REST API → Keys) or " + "PFSENSE_USERNAME + PFSENSE_PASSWORD for basic auth." + ) + + +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 PFSENSE_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 PFSENSE_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("descr", 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 descr starting with '{prefix}'. Got: '{description}'" + ) + + if not settings.block_wan_inbound_any: + return + + interface = rule.get("interface", "") + if isinstance(interface, list): + interface = ",".join(str(i) for i in interface) + interface = str(interface).lower() + rule_type = str(rule.get("type", "pass")).lower() + source = rule.get("source", "any") + if isinstance(source, dict): + source = source.get("address", source.get("network", "any")) + source = str(source).lower() + dest = rule.get("destination", "any") + if isinstance(dest, dict): + dest = dest.get("address", dest.get("network", "any")) + dest = str(dest).lower() + + wan_markers = ("wan", "opt") + is_wan = any(m in interface for m in wan_markers) or interface in ("", "any") + + if rule_type == "pass" and is_wan: + if source in ("any", "", "0.0.0.0/0", "::/0") and dest in ("any", "", "0.0.0.0/0"): + raise GuardError( + "Blocked: WAN pass rule with source 'any' and destination 'any' is not " + "permitted. Use specific networks or disable PFSENSE_BLOCK_WAN_INBOUND_ANY." + ) + + +def validate_alias_mutation(name: str) -> None: + protected = {"bogons", "bogonsv6", "virusprot", "sshguard"} + if name.lower() in protected: + raise GuardError(f"Alias '{name}' is protected and cannot be modified via MCP.") diff --git a/src/pfsense_mcp/server.py b/src/pfsense_mcp/server.py new file mode 100644 index 0000000..63d95e7 --- /dev/null +++ b/src/pfsense_mcp/server.py @@ -0,0 +1,83 @@ +"""pfSense MCP server entrypoint.""" + +from __future__ import annotations + +import json + +from mcp.server.fastmcp import FastMCP + +from pfsense_mcp import __version__ +from pfsense_mcp.config import settings +from pfsense_mcp.tools import register_all + +mcp = FastMCP( + "pfsense", + instructions=( + "pfSense firewall management MCP (REST API v2 package required on the firewall). " + "Use read tools freely for diagnostics. Write tools require PFSENSE_ALLOW_WRITES=true. " + f"New firewall rules must use descr prefix '{settings.require_description_prefix}'. " + "Most write tools accept apply=true to apply immediately, or call the apply tool after." + ), +) + + +@mcp.tool() +def pfsense_get_config() -> str: + """Get current MCP server configuration (no secrets exposed).""" + return json.dumps( + { + "url": settings.url, + "verify_ssl": settings.verify_ssl, + "allow_writes": settings.allow_writes, + "dry_run": settings.dry_run, + "require_description_prefix": settings.require_description_prefix, + "block_wan_inbound_any": settings.block_wan_inbound_any, + "api_key_configured": bool(settings.api_key), + "basic_auth_configured": bool(settings.username and settings.password), + "version": __version__, + }, + indent=2, + ) + + +@mcp.tool() +def pfsense_test_connection() -> str: + """Test connectivity and authentication to the pfSense REST API.""" + from pfsense_mcp.client import PfsenseAPIError, get_client + + client = get_client() + try: + version = client.get("system/version") + status = client.get("status/system") + return json.dumps( + {"status": "ok", "url": settings.url, "version": version, "system": status}, + indent=2, + default=str, + ) + except PfsenseAPIError as exc: + return json.dumps( + { + "status": "error", + "url": settings.url, + "message": str(exc), + "status_code": exc.status_code, + "body": exc.body, + "hint": ( + "Ensure the pfSense REST API v2 package (pfSense-pkg-RESTAPI) is " + "installed and an API key is configured under System → REST API → Keys." + ), + }, + indent=2, + default=str, + ) + + +register_all(mcp) + + +def main() -> None: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/src/pfsense_mcp/tools/__init__.py b/src/pfsense_mcp/tools/__init__.py new file mode 100644 index 0000000..80135af --- /dev/null +++ b/src/pfsense_mcp/tools/__init__.py @@ -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) diff --git a/src/pfsense_mcp/tools/aliases.py b/src/pfsense_mcp/tools/aliases.py new file mode 100644 index 0000000..d72f072 --- /dev/null +++ b/src/pfsense_mcp/tools/aliases.py @@ -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) diff --git a/src/pfsense_mcp/tools/dhcp.py b/src/pfsense_mcp/tools/dhcp.py new file mode 100644 index 0000000..59dae8b --- /dev/null +++ b/src/pfsense_mcp/tools/dhcp.py @@ -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) diff --git a/src/pfsense_mcp/tools/diagnostics.py b/src/pfsense_mcp/tools/diagnostics.py new file mode 100644 index 0000000..e58b5d7 --- /dev/null +++ b/src/pfsense_mcp/tools/diagnostics.py @@ -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) diff --git a/src/pfsense_mcp/tools/dns.py b/src/pfsense_mcp/tools/dns.py new file mode 100644 index 0000000..2591059 --- /dev/null +++ b/src/pfsense_mcp/tools/dns.py @@ -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) diff --git a/src/pfsense_mcp/tools/firewall.py b/src/pfsense_mcp/tools/firewall.py new file mode 100644 index 0000000..0793a53 --- /dev/null +++ b/src/pfsense_mcp/tools/firewall.py @@ -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) diff --git a/src/pfsense_mcp/tools/interfaces.py b/src/pfsense_mcp/tools/interfaces.py new file mode 100644 index 0000000..47aa70e --- /dev/null +++ b/src/pfsense_mcp/tools/interfaces.py @@ -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) diff --git a/src/pfsense_mcp/tools/nat.py b/src/pfsense_mcp/tools/nat.py new file mode 100644 index 0000000..1c65317 --- /dev/null +++ b/src/pfsense_mcp/tools/nat.py @@ -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) diff --git a/src/pfsense_mcp/tools/routes.py b/src/pfsense_mcp/tools/routes.py new file mode 100644 index 0000000..56f0ad9 --- /dev/null +++ b/src/pfsense_mcp/tools/routes.py @@ -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) diff --git a/src/pfsense_mcp/tools/system.py b/src/pfsense_mcp/tools/system.py new file mode 100644 index 0000000..5dd6796 --- /dev/null +++ b/src/pfsense_mcp/tools/system.py @@ -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) diff --git a/src/pfsense_mcp/tools/vpn.py b/src/pfsense_mcp/tools/vpn.py new file mode 100644 index 0000000..7de5d57 --- /dev/null +++ b/src/pfsense_mcp/tools/vpn.py @@ -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)