Add persisted target registry for Cursor golden profiles.
Targets save to PFSENSE_TARGETS_PATH so golden-pfsense and other firewalls reload across MCP sessions.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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}'."
|
||||
),
|
||||
)
|
||||
|
||||
@@ -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 = "",
|
||||
|
||||
95
src/pfsense_mcp/targets_store.py
Normal file
95
src/pfsense_mcp/targets_store.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user