diff --git a/README.md b/README.md index 6b424e2..d665e01 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,8 @@ Model Context Protocol server for [pfSense](https://www.pfsense.org/) firewall m ## Features -- **55+ tools** covering system status, interfaces, firewall rules, aliases, NAT, DHCP, diagnostics, routing, VPN (OpenVPN/IPsec/WireGuard), and DNS resolver +- **77+ tools** covering system status, interfaces, firewall rules, aliases, NAT, DHCP, diagnostics, routing, VPN (OpenVPN/IPsec/WireGuard), and DNS resolver +- **Multi-firewall**: connect to many pfSense boxes dynamically via `pfsense_connect(url, api_key)` — no static firewall in config required - **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 @@ -53,6 +54,8 @@ cp .env.example .env ### 3. Register with Hermes +No static firewall URL/key required — credentials are provided at runtime via `pfsense_connect`. + ```yaml mcp_servers: pfsense: @@ -60,13 +63,43 @@ mcp_servers: 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`. +Then `hermes gateway restart`. Connect to a firewall in chat: + +```text +pfsense_connect(url="https://10.77.50.1", api_key="...", name="branch-office") +pfsense_test_connection() +pfsense_list_firewall_rules() +``` + +### Multi-firewall workflow + +```text +# Register firewalls (name optional — defaults from IP/hostname) +pfsense_connect("https://10.77.50.1", "key-a", name="hq") +pfsense_connect("https://10.77.60.1", "key-b", name="branch") + +# List and switch +pfsense_list_targets() +pfsense_use_target("branch") + +# Or pass target / inline credentials on any tool +pfsense_get_system_status(target="hq") +pfsense_get_gateway_status(url="https://10.77.60.1", api_key="key-b") +``` + +| Tool | Purpose | +|------|---------| +| `pfsense_connect` | Register + activate a firewall (url, api_key, name) | +| `pfsense_use_target` | Switch active firewall | +| `pfsense_list_targets` | List registered firewalls | +| `pfsense_get_active_target` | Show current active target | +| `pfsense_disconnect_target` | Remove a registered target | + +Optional env defaults (`PFSENSE_URL`, `PFSENSE_API_KEY`) still work as a fallback when no target is connected. ### 4. Register with Cursor (`~/.cursor/mcp.json`) diff --git a/pyproject.toml b/pyproject.toml index b549156..15da2f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "pfsense-mcp" -version = "1.0.0" +version = "1.1.0" description = "Model Context Protocol server for pfSense firewall management via REST API v2" readme = "README.md" requires-python = ">=3.10" diff --git a/src/pfsense_mcp/__init__.py b/src/pfsense_mcp/__init__.py index d29aae2..ec53322 100644 --- a/src/pfsense_mcp/__init__.py +++ b/src/pfsense_mcp/__init__.py @@ -1,3 +1,3 @@ """pfSense MCP server — Model Context Protocol integration for pfSense firewalls.""" -__version__ = "1.0.0" +__version__ = "1.1.0" diff --git a/src/pfsense_mcp/client.py b/src/pfsense_mcp/client.py index ff10c59..89043e0 100644 --- a/src/pfsense_mcp/client.py +++ b/src/pfsense_mcp/client.py @@ -8,7 +8,7 @@ from typing import Any import httpx from pfsense_mcp.config import settings -from pfsense_mcp.guards import require_credentials +from pfsense_mcp.guards import GuardError class PfsenseAPIError(Exception): @@ -28,14 +28,25 @@ class PfsenseClient: url: str | None = None, api_key: str | None = None, *, + username: str | None = None, + password: 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.username = username if username is not None else settings.username + self.password = password if password is not None else settings.password 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 _require_credentials(self) -> None: + if not self.api_key and not (self.username and self.password): + raise GuardError( + "No API credentials for this target. Call pfsense_connect(url, api_key) or pass " + "url/api_key on the tool." + ) + def _request( self, method: str, @@ -44,14 +55,14 @@ class PfsenseClient: params: dict[str, Any] | None = None, data: dict[str, Any] | None = None, ) -> Any: - require_credentials() + self._require_credentials() url = f"{self.base_url}/api/v2/{endpoint.lstrip('/')}" - headers = {} + headers: dict[str, str] = {} auth = None if self.api_key: headers["X-API-Key"] = self.api_key else: - auth = (settings.username, settings.password) + auth = (self.username, self.password) clean_params = None if params: @@ -111,10 +122,7 @@ class PfsenseClient: 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". - """ + """Build standard v2 collection query params.""" params: dict[str, Any] = {"limit": limit, "offset": offset} if query: for pair in query.split(","): @@ -124,11 +132,18 @@ class PfsenseClient: return params -_client: PfsenseClient | None = None +def get_client( + target: str = "", + url: str = "", + api_key: str = "", + verify_ssl: bool | None = None, +) -> PfsenseClient: + """Resolve and return a client for the given target or inline connection.""" + from pfsense_mcp.targets import resolve_client - -def get_client() -> PfsenseClient: - global _client - if _client is None: - _client = PfsenseClient() - return _client + return resolve_client( + target=target, + url=url, + api_key=api_key, + verify_ssl=verify_ssl, + ) diff --git a/src/pfsense_mcp/guards.py b/src/pfsense_mcp/guards.py index 334d343..c3221f4 100644 --- a/src/pfsense_mcp/guards.py +++ b/src/pfsense_mcp/guards.py @@ -13,10 +13,11 @@ class GuardError(Exception): def require_credentials() -> None: + """Legacy env-level check; prefer per-client credentials via pfsense_connect.""" 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." + "No pfSense target connected. Call pfsense_connect(url, api_key) first, " + "or pass target/url/api_key on the tool." ) diff --git a/src/pfsense_mcp/server.py b/src/pfsense_mcp/server.py index 63d95e7..205af97 100644 --- a/src/pfsense_mcp/server.py +++ b/src/pfsense_mcp/server.py @@ -13,27 +13,30 @@ 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." + "pfSense firewall MCP for multiple firewalls (REST API v2 package required). " + "Always call pfsense_connect(url, api_key, name=...) before other tools, or pass " + "target/url/api_key on each call. Use pfsense_list_targets and pfsense_use_target to " + "switch between firewalls. Write tools require PFSENSE_ALLOW_WRITES=true. " + f"New firewall rules must use descr prefix '{settings.require_description_prefix}'." ), ) @mcp.tool() def pfsense_get_config() -> str: - """Get current MCP server configuration (no secrets exposed).""" + """Get MCP server configuration and registered targets (no secrets).""" + from pfsense_mcp.targets import registry + 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), + "default_url": settings.url, + "env_api_key_configured": bool(settings.api_key and settings.api_key != "CHANGE-ME"), + "active_target": registry.active().name if registry.active() else None, + "registered_targets": registry.list(), "version": __version__, }, indent=2, @@ -41,30 +44,46 @@ def pfsense_get_config() -> str: @mcp.tool() -def pfsense_test_connection() -> str: - """Test connectivity and authentication to the pfSense REST API.""" +def pfsense_test_connection( + target: str = "", + url: str = "", + api_key: str = "", + verify_ssl: bool | None = None, +) -> str: + """Test connectivity to a pfSense REST API. Uses inline url/api_key, named target, or active target.""" from pfsense_mcp.client import PfsenseAPIError, get_client + from pfsense_mcp.guards import GuardError + from pfsense_mcp.targets import registry - client = get_client() try: + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) + cfg = registry.resolve(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) version = client.get("system/version") status = client.get("status/system") return json.dumps( - {"status": "ok", "url": settings.url, "version": version, "system": status}, + { + "status": "ok", + "target": cfg.name if cfg.name != "__inline__" else "inline", + "url": client.base_url, + "version": version, + "system": status, + }, indent=2, default=str, ) + except GuardError as exc: + return json.dumps({"status": "error", "message": str(exc)}, indent=2) except PfsenseAPIError as exc: return json.dumps( { "status": "error", - "url": settings.url, + "url": url or (registry.active().normalized_url() if registry.active() else 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." + "Install pfSense-pkg-RESTAPI and create an API key under " + "System → REST API → Keys, then pfsense_connect(url, api_key)." ), }, indent=2, diff --git a/src/pfsense_mcp/targets.py b/src/pfsense_mcp/targets.py new file mode 100644 index 0000000..024908f --- /dev/null +++ b/src/pfsense_mcp/targets.py @@ -0,0 +1,213 @@ +"""Multi-firewall target registry and connection resolution.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +from pfsense_mcp.client import PfsenseClient +from pfsense_mcp.config import settings +from pfsense_mcp.guards import GuardError + + +@dataclass +class TargetConfig: + name: str + url: str + api_key: str = "" + username: str = "" + password: str = "" + verify_ssl: bool = False + description: str = "" + registered_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + def normalized_url(self) -> str: + url = self.url.strip().rstrip("/") + if not url: + raise GuardError("Target URL is empty.") + if not url.startswith(("http://", "https://")): + url = f"https://{url}" + return url + + def has_credentials(self) -> bool: + return bool(self.api_key or (self.username and self.password)) + + def public_view(self) -> dict[str, Any]: + return { + "name": self.name, + "url": self.normalized_url(), + "verify_ssl": self.verify_ssl, + "description": self.description, + "api_key_configured": bool(self.api_key), + "basic_auth_configured": bool(self.username and self.password), + "registered_at": self.registered_at, + } + + +class TargetRegistry: + def __init__(self) -> None: + self._targets: dict[str, TargetConfig] = {} + self._active: str | None = None + + @staticmethod + def _normalize_name(name: str) -> str: + name = name.strip().lower() + name = re.sub(r"[^a-z0-9_-]+", "-", name) + name = re.sub(r"-+", "-", name).strip("-") + if not name: + raise GuardError("Target name must contain at least one alphanumeric character.") + return name + + @staticmethod + def _derive_name(url: str) -> str: + host = url.strip().rstrip("/") + host = re.sub(r"^https?://", "", host) + host = host.split("/")[0].split(":")[0] + return TargetRegistry._normalize_name(host or "pfsense") + + def register( + self, + *, + name: str = "", + url: str, + api_key: str = "", + username: str = "", + password: str = "", + verify_ssl: bool = False, + description: str = "", + set_active: bool = True, + ) -> TargetConfig: + if not url.strip(): + raise GuardError("url is required (e.g. 'https://10.77.50.1' or '10.77.50.1').") + target_name = self._normalize_name(name) if name.strip() else self._derive_name(url) + cfg = TargetConfig( + name=target_name, + url=url, + api_key=api_key, + username=username, + password=password, + verify_ssl=verify_ssl, + description=description, + ) + if not cfg.has_credentials(): + raise GuardError( + "api_key is required (pfSense REST API v2 key from System → REST API → Keys)." + ) + self._targets[target_name] = cfg + if set_active: + self._active = target_name + return cfg + + def unregister(self, name: str) -> None: + key = self._normalize_name(name) + if key not in self._targets: + raise GuardError(f"Unknown target '{name}'.") + del self._targets[key] + if self._active == key: + self._active = next(iter(self._targets), None) + + def set_active(self, name: str) -> TargetConfig: + key = self._normalize_name(name) + if key not in self._targets: + raise GuardError(f"Unknown target '{name}'. Call pfsense_connect first.") + self._active = key + return self._targets[key] + + def get(self, name: str) -> TargetConfig: + key = self._normalize_name(name) + if key not in self._targets: + raise GuardError(f"Unknown target '{name}'.") + return self._targets[key] + + def active(self) -> TargetConfig | None: + if self._active and self._active in self._targets: + return self._targets[self._active] + return None + + def list(self) -> list[dict[str, Any]]: + active = self._active + rows = [] + for cfg in self._targets.values(): + row = cfg.public_view() + row["active"] = cfg.name == active + rows.append(row) + return sorted(rows, key=lambda r: r["name"]) + + def resolve( + self, + *, + target: str = "", + url: str = "", + api_key: str = "", + username: str = "", + password: str = "", + verify_ssl: bool | None = None, + ) -> TargetConfig: + """Resolve connection details: inline args > named target > active target > env default.""" + if url.strip() or api_key.strip() or username.strip() or password.strip(): + if not url.strip(): + raise GuardError("url is required when passing inline credentials.") + cfg = TargetConfig( + name="__inline__", + url=url, + api_key=api_key, + username=username, + password=password, + verify_ssl=settings.verify_ssl if verify_ssl is None else verify_ssl, + ) + if not cfg.has_credentials(): + raise GuardError("Inline connection requires api_key (or username+password).") + return cfg + + if target.strip(): + return self.get(target) + + active = self.active() + if active: + return active + + if settings.url and settings.url not in ("https://192.168.1.1", "https://CHANGE-ME"): + if settings.api_key and settings.api_key not in ("", "CHANGE-ME"): + return TargetConfig( + name="__env__", + url=settings.url, + api_key=settings.api_key, + username=settings.username, + password=settings.password, + verify_ssl=settings.verify_ssl, + ) + + raise GuardError( + "No pfSense target connected. Call pfsense_connect(url, api_key, name=...) first, " + "or pass target/url/api_key on this tool." + ) + + +registry = TargetRegistry() + + +def resolve_client( + target: str = "", + url: str = "", + api_key: str = "", + username: str = "", + password: str = "", + verify_ssl: bool | None = None, +) -> PfsenseClient: + cfg = registry.resolve( + target=target, + url=url, + api_key=api_key, + username=username, + password=password, + verify_ssl=verify_ssl, + ) + return PfsenseClient( + url=cfg.normalized_url(), + api_key=cfg.api_key, + username=cfg.username, + password=cfg.password, + verify_ssl=cfg.verify_ssl, + ) diff --git a/src/pfsense_mcp/tools/__init__.py b/src/pfsense_mcp/tools/__init__.py index 80135af..ac2b6d6 100644 --- a/src/pfsense_mcp/tools/__init__.py +++ b/src/pfsense_mcp/tools/__init__.py @@ -12,11 +12,13 @@ from pfsense_mcp.tools import ( nat, routes, system, + targets, vpn, ) def register_all(mcp: FastMCP) -> None: + targets.register(mcp) system.register(mcp) interfaces.register(mcp) firewall.register(mcp) diff --git a/src/pfsense_mcp/tools/aliases.py b/src/pfsense_mcp/tools/aliases.py index d72f072..36cb85c 100644 --- a/src/pfsense_mcp/tools/aliases.py +++ b/src/pfsense_mcp/tools/aliases.py @@ -15,16 +15,16 @@ from pfsense_mcp.guards import is_dry_run_response, require_writes, validate_ali def register(mcp: FastMCP) -> None: @mcp.tool() - def pfsense_list_aliases(limit: int = 100, offset: int = 0, query: str = "") -> str: + def pfsense_list_aliases(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List firewall aliases. query example: 'name__contains=block'.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_get_alias(alias_id: int, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get an alias by numeric id.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get("firewall/alias", params={"id": alias_id}) return json.dumps(result, indent=2, default=str) @@ -35,6 +35,10 @@ def register(mcp: FastMCP) -> None: addresses: list[str], description: str = "", apply: bool = False, + target: str = "", + url: str = "", + api_key: str = "", + verify_ssl: bool | None = None, ) -> str: """Add a firewall alias. alias_type: host, network, or port. addresses: list of IPs/CIDRs/ports.""" validate_alias_mutation(name) @@ -47,13 +51,13 @@ def register(mcp: FastMCP) -> None: if settings.dry_run: return json.dumps(is_dry_run_response("add_alias", payload), indent=2) require_writes("add_alias", payload) - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_update_alias(alias_id: int, fields: dict[str, Any], apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Update an alias by id (PATCH — send only changed fields, e.g. {'address': [...]}).""" if "name" in fields: validate_alias_mutation(str(fields["name"])) @@ -62,19 +66,19 @@ def register(mcp: FastMCP) -> None: if settings.dry_run: return json.dumps(is_dry_run_response("update_alias", payload), indent=2) require_writes("update_alias", payload) - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_delete_alias(alias_id: int, apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 index 59dae8b..935572b 100644 --- a/src/pfsense_mcp/tools/dhcp.py +++ b/src/pfsense_mcp/tools/dhcp.py @@ -15,27 +15,28 @@ 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: + def pfsense_list_dhcp_leases(limit: int = 200, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List DHCP leases. query example: 'hostname__contains=nas' or 'ip__contains=10.77'.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_dhcp_servers(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List DHCP server configurations per interface (ranges, options, enabled state).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 + interface_id: str, limit: int = 200, offset: int = 0, + target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None, ) -> str: """List DHCP static mappings (reservations) for a DHCP server. interface_id: parent interface (e.g. 'lan').""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get( "services/dhcp_server/static_mappings", params={"parent_id": interface_id, "limit": limit, "offset": offset}, @@ -49,6 +50,10 @@ def register(mcp: FastMCP) -> None: ip: str, hostname: str = "", description: str = "", + target: str = "", + url: str = "", + api_key: str = "", + verify_ssl: bool | None = None, ) -> str: """Add a DHCP static mapping (reservation) on an interface's DHCP server.""" payload: dict[str, Any] = { @@ -61,19 +66,19 @@ def register(mcp: FastMCP) -> None: 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_delete_dhcp_static_mapping(interface_id: str, mapping_id: int, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 index e58b5d7..6917a3f 100644 --- a/src/pfsense_mcp/tools/diagnostics.py +++ b/src/pfsense_mcp/tools/diagnostics.py @@ -13,56 +13,56 @@ 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_get_arp_table(limit: int = 200, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get the ARP table. query example: 'ip__contains=10.77' or 'hostname__contains=nas'.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_delete_arp_entry(entry_id: int, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_get_firewall_log(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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: + def pfsense_get_system_log(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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: + def pfsense_get_dhcp_log(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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: + def pfsense_get_auth_log(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get authentication log entries (logins, failures).""" return _get_log("authentication", limit, offset, query) @mcp.tool() - def pfsense_get_system_activity() -> str: + def pfsense_get_system_activity(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get system activity (top-style process list).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 index 2591059..f333a84 100644 --- a/src/pfsense_mcp/tools/dns.py +++ b/src/pfsense_mcp/tools/dns.py @@ -15,9 +15,9 @@ 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: + def pfsense_list_dns_host_overrides(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List DNS resolver host overrides (local DNS records).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get( "services/dns_resolver/host_overrides", params=client.list_params(limit, offset, query), @@ -31,6 +31,10 @@ def register(mcp: FastMCP) -> None: ip: list[str], description: str = "", apply: bool = False, + target: str = "", + url: str = "", + api_key: str = "", + verify_ssl: bool | None = None, ) -> str: """Add a DNS resolver host override. ip: list of addresses for the record.""" payload: dict[str, Any] = { @@ -42,7 +46,7 @@ def register(mcp: FastMCP) -> None: 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.post( "services/dns_resolver/host_override", data=payload, params={"apply": apply} ) @@ -50,13 +54,13 @@ def register(mcp: FastMCP) -> None: return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_delete_dns_host_override(override_id: int, apply: bool = False) -> str: + def pfsense_delete_dns_host_override(override_id: int, apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.delete( "services/dns_resolver/host_override", params={"id": override_id, "apply": apply} ) @@ -64,12 +68,12 @@ def register(mcp: FastMCP) -> None: return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_apply_dns_resolver() -> str: + def pfsense_apply_dns_resolver(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 index 0793a53..b2357f4 100644 --- a/src/pfsense_mcp/tools/firewall.py +++ b/src/pfsense_mcp/tools/firewall.py @@ -19,34 +19,34 @@ from pfsense_mcp.guards import ( def register(mcp: FastMCP) -> None: @mcp.tool() - def pfsense_list_firewall_rules(limit: int = 100, offset: int = 0, query: str = "") -> str: + 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() + 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) -> str: + 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() + 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) -> str: + 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() + 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) -> str: + 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) @@ -54,62 +54,62 @@ def register(mcp: FastMCP) -> None: 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() + 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) -> str: + 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() + 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) -> str: + 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() + 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() -> str: + 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() + 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() -> str: + 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() + 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 = "") -> str: + 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() + 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 = "") -> str: + 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: @@ -117,21 +117,21 @@ def register(mcp: FastMCP) -> None: if settings.dry_run: return json.dumps(is_dry_run_response("kill_states", payload), indent=2) require_writes("kill_states", payload) - client = get_client() + 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) -> str: + 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() + 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) -> str: + 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() + 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) diff --git a/src/pfsense_mcp/tools/interfaces.py b/src/pfsense_mcp/tools/interfaces.py index 47aa70e..56290ec 100644 --- a/src/pfsense_mcp/tools/interfaces.py +++ b/src/pfsense_mcp/tools/interfaces.py @@ -14,40 +14,40 @@ from pfsense_mcp.guards import is_dry_run_response, require_writes def register(mcp: FastMCP) -> None: @mcp.tool() - def pfsense_get_interface_status() -> str: + def pfsense_get_interface_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get live status of all interfaces (up/down, IPs, media, in/out bytes).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_interfaces(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List configured interfaces (assignments and settings).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_vlans(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List VLAN interface configurations.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_get_carp_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get CARP (high availability) status and virtual IPs.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get("status/carp") return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_apply_interfaces() -> str: + def pfsense_apply_interfaces(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 index 1c65317..e8db718 100644 --- a/src/pfsense_mcp/tools/nat.py +++ b/src/pfsense_mcp/tools/nat.py @@ -16,63 +16,63 @@ 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: + def pfsense_list_port_forwards(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List port forward (destination NAT) rules.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_add_port_forward(rule: dict[str, Any], apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_update_port_forward(rule_id: int, fields: dict[str, Any], apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_delete_port_forward(rule_id: int, apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_get_outbound_nat_mode(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get the outbound NAT mode (automatic, hybrid, manual, disabled).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_outbound_nat_mappings(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List outbound NAT mappings.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get( "firewall/nat/outbound/mappings", params={"limit": limit, "offset": offset} ) @@ -80,9 +80,9 @@ def register(mcp: FastMCP) -> None: # ── 1:1 NAT ───────────────────────────────────────────────────────── @mcp.tool() - def pfsense_list_one_to_one_mappings(limit: int = 100, offset: int = 0) -> str: + def pfsense_list_one_to_one_mappings(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List 1:1 NAT mappings.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get( "firewall/nat/one_to_one/mappings", params={"limit": limit, "offset": offset} ) diff --git a/src/pfsense_mcp/tools/routes.py b/src/pfsense_mcp/tools/routes.py index 56f0ad9..153692b 100644 --- a/src/pfsense_mcp/tools/routes.py +++ b/src/pfsense_mcp/tools/routes.py @@ -15,29 +15,30 @@ from pfsense_mcp.guards import is_dry_run_response, require_writes def register(mcp: FastMCP) -> None: @mcp.tool() - def pfsense_get_gateway_status() -> str: + def pfsense_get_gateway_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get gateway status: latency, packet loss, online/offline.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_gateways(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List configured gateways.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_static_routes(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List static routes.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 + network: str, gateway: str, description: str = "", apply: bool = False, + target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None, ) -> str: """Add a static route. network: CIDR, gateway: gateway name.""" payload: dict[str, Any] = { @@ -48,30 +49,30 @@ def register(mcp: FastMCP) -> None: 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_delete_static_route(route_id: int, apply: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_apply_routing(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> 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() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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 index 5dd6796..6dedff2 100644 --- a/src/pfsense_mcp/tools/system.py +++ b/src/pfsense_mcp/tools/system.py @@ -14,28 +14,28 @@ 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: + def pfsense_get_system_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get system status: CPU, memory, disk, temperature, uptime.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get("status/system") return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_get_version() -> str: + def pfsense_get_version(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get pfSense version and patch information.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_services(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List system services with running/stopped status.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_service_action(name: str, action: str, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Control a service. action: start, stop, restart.""" allowed = {"start", "stop", "restart"} if action not in allowed: @@ -44,31 +44,31 @@ def register(mcp: FastMCP) -> None: if settings.dry_run: return json.dumps(is_dry_run_response("service_action", payload), indent=2) require_writes("service_action", payload) - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_packages(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List installed pfSense packages.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_available_packages(limit: int = 100, offset: int = 0, query: str = "", target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List packages available to install. query example: 'name__contains=wireguard'.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_certificates(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List certificates (no private keys returned by this tool).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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"]: @@ -77,9 +77,9 @@ def register(mcp: FastMCP) -> None: return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_get_config_history(limit: int = 20, offset: int = 0) -> str: + def pfsense_get_config_history(limit: int = 20, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List configuration change history revisions (who/when/what).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get( "diagnostics/config_history/revisions", params={"limit": limit, "offset": offset}, @@ -87,21 +87,21 @@ def register(mcp: FastMCP) -> None: return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_get_dns_settings() -> str: + def pfsense_get_dns_settings(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get system DNS server settings.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get("system/dns") return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_get_hostname() -> str: + def pfsense_get_hostname(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get system hostname and domain.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get("system/hostname") return json.dumps(result, indent=2, default=str) @mcp.tool() - def pfsense_reboot(confirm: bool = False) -> str: + def pfsense_reboot(confirm: bool = False, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Reboot the firewall. Requires confirm=true AND PFSENSE_ALLOW_WRITES=true.""" if not confirm: return json.dumps( @@ -114,7 +114,7 @@ def register(mcp: FastMCP) -> None: if settings.dry_run: return json.dumps(is_dry_run_response("reboot", {}), indent=2) require_writes("reboot", {}) - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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/targets.py b/src/pfsense_mcp/tools/targets.py new file mode 100644 index 0000000..a675628 --- /dev/null +++ b/src/pfsense_mcp/tools/targets.py @@ -0,0 +1,93 @@ +"""Multi-firewall connection management tools.""" + +from __future__ import annotations + +import json + +from mcp.server.fastmcp import FastMCP + +from pfsense_mcp.client import PfsenseAPIError +from pfsense_mcp.config import settings +from pfsense_mcp.guards import GuardError +from pfsense_mcp.targets import registry, resolve_client + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def pfsense_connect( + url: str, + api_key: str, + name: str = "", + verify_ssl: bool = False, + description: str = "", + set_active: bool = True, + ) -> str: + """Register and optionally activate a pfSense firewall target. + + url: base URL or IP (e.g. 'https://10.77.50.1' or '10.77.50.1'). + api_key: REST API v2 key from System → REST API → Keys. + name: optional short name (defaults from hostname/IP). Re-registering updates credentials. + """ + cfg = registry.register( + name=name, + url=url, + api_key=api_key, + verify_ssl=verify_ssl, + description=description, + set_active=set_active, + ) + return json.dumps( + { + "status": "connected", + "target": cfg.public_view(), + "active": registry.active().name if registry.active() else None, + "hint": f"Use target='{cfg.name}' on tools, or omit target while '{cfg.name}' is active.", + }, + indent=2, + ) + + @mcp.tool() + def pfsense_use_target(name: str) -> str: + """Switch the active pfSense target for subsequent tool calls.""" + cfg = registry.set_active(name) + return json.dumps({"status": "ok", "active": cfg.public_view()}, indent=2) + + @mcp.tool() + def pfsense_list_targets() -> str: + """List registered pfSense targets (no secrets). Shows which target is active.""" + active = registry.active() + return json.dumps( + { + "active": active.name if active else None, + "targets": registry.list(), + "count": len(registry.list()), + }, + indent=2, + ) + + @mcp.tool() + def pfsense_disconnect_target(name: str) -> str: + """Remove a registered pfSense target from the session.""" + registry.unregister(name) + return json.dumps( + { + "status": "removed", + "name": name, + "active": registry.active().name if registry.active() else None, + }, + indent=2, + ) + + @mcp.tool() + def pfsense_get_active_target() -> str: + """Get the currently active pfSense target (no secrets).""" + active = registry.active() + if not active: + return json.dumps( + { + "status": "none", + "message": "No active target. Call pfsense_connect(url, api_key) first.", + }, + indent=2, + ) + return json.dumps({"status": "ok", "active": active.public_view()}, indent=2) diff --git a/src/pfsense_mcp/tools/vpn.py b/src/pfsense_mcp/tools/vpn.py index 7de5d57..a191d9e 100644 --- a/src/pfsense_mcp/tools/vpn.py +++ b/src/pfsense_mcp/tools/vpn.py @@ -11,57 +11,57 @@ from pfsense_mcp.client import get_client def register(mcp: FastMCP) -> None: @mcp.tool() - def pfsense_get_openvpn_status() -> str: + def pfsense_get_openvpn_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get OpenVPN server/client status with connected clients.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_openvpn_servers(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List OpenVPN server configurations.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_openvpn_clients(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List OpenVPN client configurations.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_get_ipsec_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get IPsec tunnel status (phase 1/2 SAs).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_ipsec_phase1(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List configured IPsec phase 1 entries.""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_get_wireguard_status(target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """Get WireGuard status (requires the WireGuard package to be installed).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_wireguard_tunnels(limit: int = 50, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List WireGuard tunnels (requires the WireGuard package).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) 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: + def pfsense_list_wireguard_peers(limit: int = 100, offset: int = 0, target: str = "", url: str = "", api_key: str = "", verify_ssl: bool | None = None) -> str: """List WireGuard peers (requires the WireGuard package).""" - client = get_client() + client = get_client(target=target, url=url, api_key=api_key, verify_ssl=verify_ssl) result = client.get("vpn/wireguard/peers", params={"limit": limit, "offset": offset}) return json.dumps(result, indent=2, default=str)