From 3ccd12d9232150133f8c2c9198e2f152079417f6 Mon Sep 17 00:00:00 2001 From: nearxos Date: Fri, 12 Jun 2026 22:54:07 +0300 Subject: [PATCH] Add persisted target registry for Cursor golden profiles. Targets save to PFSENSE_TARGETS_PATH so golden-pfsense and other firewalls reload across MCP sessions. --- .env.example | 1 + src/pfsense_mcp/config.py | 1 + src/pfsense_mcp/server.py | 7 ++- src/pfsense_mcp/targets.py | 42 +++++++++++--- src/pfsense_mcp/targets_store.py | 95 ++++++++++++++++++++++++++++++++ 5 files changed, 134 insertions(+), 12 deletions(-) create mode 100644 src/pfsense_mcp/targets_store.py diff --git a/.env.example b/.env.example index 9456319..8852ca2 100644 --- a/.env.example +++ b/.env.example @@ -14,3 +14,4 @@ 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 +PFSENSE_TARGETS_PATH=~/.local/share/pfsense-mcp/targets.json diff --git a/src/pfsense_mcp/config.py b/src/pfsense_mcp/config.py index 6649a43..3d812ae 100644 --- a/src/pfsense_mcp/config.py +++ b/src/pfsense_mcp/config.py @@ -23,6 +23,7 @@ class Settings(BaseSettings): require_description_prefix: str = "mcp:" block_wan_inbound_any: bool = True audit_log_path: str = "/root/.hermes/logs/pfsense-mcp-audit.jsonl" + targets_path: str = "~/.local/share/pfsense-mcp/targets.json" settings = Settings() diff --git a/src/pfsense_mcp/server.py b/src/pfsense_mcp/server.py index 205af97..2bdfb86 100644 --- a/src/pfsense_mcp/server.py +++ b/src/pfsense_mcp/server.py @@ -14,9 +14,10 @@ mcp = FastMCP( "pfsense", instructions=( "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. " + "Saved targets load from PFSENSE_TARGETS_PATH on startup. Use pfsense_list_targets, " + "pfsense_use_target('name'), or pfsense_connect(url, api_key, name=...) to register more. " + "Pass target='name' on tools, or rely on the active saved target. " + "Write tools require PFSENSE_ALLOW_WRITES=true. " f"New firewall rules must use descr prefix '{settings.require_description_prefix}'." ), ) diff --git a/src/pfsense_mcp/targets.py b/src/pfsense_mcp/targets.py index 024908f..633ca0c 100644 --- a/src/pfsense_mcp/targets.py +++ b/src/pfsense_mcp/targets.py @@ -50,6 +50,7 @@ class TargetRegistry: def __init__(self) -> None: self._targets: dict[str, TargetConfig] = {} self._active: str | None = None + self._loading: bool = False @staticmethod def _normalize_name(name: str) -> str: @@ -77,20 +78,24 @@ class TargetRegistry: password: str = "", verify_ssl: bool = False, description: str = "", + registered_at: 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, - ) + cfg_kwargs: dict[str, Any] = { + "name": target_name, + "url": url, + "api_key": api_key, + "username": username, + "password": password, + "verify_ssl": verify_ssl, + "description": description, + } + if registered_at.strip(): + cfg_kwargs["registered_at"] = registered_at.strip() + cfg = TargetConfig(**cfg_kwargs) if not cfg.has_credentials(): raise GuardError( "api_key is required (pfSense REST API v2 key from System → REST API → Keys)." @@ -98,6 +103,7 @@ class TargetRegistry: self._targets[target_name] = cfg if set_active: self._active = target_name + self._persist() return cfg def unregister(self, name: str) -> None: @@ -107,14 +113,23 @@ class TargetRegistry: del self._targets[key] if self._active == key: self._active = next(iter(self._targets), None) + self._persist() 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 + self._persist() return self._targets[key] + def _persist(self) -> None: + if self._loading: + return + from pfsense_mcp.targets_store import save_registry + + save_registry(self) + def get(self, name: str) -> TargetConfig: key = self._normalize_name(name) if key not in self._targets: @@ -188,6 +203,15 @@ class TargetRegistry: registry = TargetRegistry() +def _load_persisted_targets() -> None: + from pfsense_mcp.targets_store import load_registry + + load_registry(registry) + + +_load_persisted_targets() + + def resolve_client( target: str = "", url: str = "", diff --git a/src/pfsense_mcp/targets_store.py b/src/pfsense_mcp/targets_store.py new file mode 100644 index 0000000..cacc11c --- /dev/null +++ b/src/pfsense_mcp/targets_store.py @@ -0,0 +1,95 @@ +"""Persist registered pfSense targets to disk for reuse across MCP sessions.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from pfsense_mcp.targets import TargetConfig, TargetRegistry + + +def targets_path() -> Path: + from pfsense_mcp.config import settings + + return Path(os.path.expanduser(settings.targets_path)) + + +def load_registry(registry: TargetRegistry) -> None: + path = targets_path() + if not path.exists(): + return + try: + data = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + return + + registry._loading = True + loaded_active: str | None = None + for entry in data.get("targets", []): + if not isinstance(entry, dict): + continue + name = str(entry.get("name", "")).strip() + url = str(entry.get("url", "")).strip() + if not name or not url: + continue + try: + registry.register( + name=name, + url=url, + api_key=str(entry.get("api_key", "")), + username=str(entry.get("username", "")), + password=str(entry.get("password", "")), + verify_ssl=bool(entry.get("verify_ssl", False)), + description=str(entry.get("description", "")), + registered_at=str(entry.get("registered_at", "")), + set_active=False, + ) + except Exception: + continue + + active = str(data.get("active", "")).strip() + if active: + try: + registry.set_active(active) + loaded_active = active + except Exception: + pass + + if not loaded_active and registry.list(): + first = registry.list()[0]["name"] + registry.set_active(first) + + registry._loading = False + + +def save_registry(registry: TargetRegistry) -> None: + path = targets_path() + path.parent.mkdir(parents=True, exist_ok=True) + + targets: list[dict[str, Any]] = [] + for cfg in registry._targets.values(): + targets.append( + { + "name": cfg.name, + "url": cfg.normalized_url(), + "api_key": cfg.api_key, + "username": cfg.username, + "password": cfg.password, + "verify_ssl": cfg.verify_ssl, + "description": cfg.description, + "registered_at": cfg.registered_at, + } + ) + + payload = { + "active": registry._active, + "targets": sorted(targets, key=lambda row: row["name"]), + } + path.write_text(json.dumps(payload, indent=2) + "\n") + try: + path.chmod(0o600) + except OSError: + pass