From ad935bed347c2a7096934af514252a942bc4d7e9 Mon Sep 17 00:00:00 2001 From: xTITUSMAXIMUSX Date: Wed, 1 Jul 2026 09:05:04 -0500 Subject: [PATCH] Add hosted http transport with per-request auth (multi-tenant) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit VyMCP can now run as one shared server (VYMCP_TRANSPORT=http) that many engineers connect to, each authenticating with their own vym_ token per request — instead of everyone running a local stdio instance. - ServerConfig splits server settings from the per-request token; api_token only required for stdio - VyManagerTokenVerifier validates each request's bearer token against VyManager (cached); tools resolve a per-token client via current_client() - plans are owner-scoped so one engineer can't apply another's pending change - main() runs stdio or streamable-http per config; stdio path unchanged Verified live against VyManager: http client connects with a bearer token and acts as that user; unauthenticated/invalid tokens rejected; a second token cannot apply the first's plan. --- .env.example | 23 +++++++--- src/vymcp/auth.py | 46 +++++++++++++++++++ src/vymcp/changes.py | 11 ++++- src/vymcp/client.py | 86 ++++++++++++++++++++++++++++-------- src/vymcp/config.py | 80 ++++++++++++++++++++++++--------- src/vymcp/discovery.py | 6 +-- src/vymcp/server.py | 29 +++++++++--- src/vymcp/writes.py | 19 +++++--- tests/test_changes.py | 20 ++++++--- tests/test_config.py | 66 ++++++++++++++++++--------- tests/test_generic_driver.py | 6 +-- tests/test_write_tools.py | 2 +- 12 files changed, 299 insertions(+), 95 deletions(-) create mode 100644 src/vymcp/auth.py diff --git a/.env.example b/.env.example index 3609985..6de7ba0 100644 --- a/.env.example +++ b/.env.example @@ -1,16 +1,27 @@ # Base URL of the VyManager API (no trailing slash). VYMANAGER_BASE_URL=http://localhost:8000 -# A VyManager personal access token (Sites -> API Tokens). Read-only recommended. -VYMANAGER_API_TOKEN=vym_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx - # Verify the VyManager TLS certificate (set to false only for trusted dev setups). VYMANAGER_VERIFY_SSL=true # Request timeout in seconds. VYMANAGER_TIMEOUT=30 -# Enable configuration-changing tools (off by default). This is a kill-switch -# independent of the token: leave unset/false for a strictly read-only server. -# Even when true, a read-only token still cannot write (VyManager enforces it). +# Enable configuration-changing tools (off by default). Kill-switch independent of +# the token: even a read-write token cannot change config unless this is true. VYMANAGER_ENABLE_WRITES=false + +# --- Transport -------------------------------------------------------------- +# "stdio" (default): the MCP client launches VyMCP locally, one process per user. +# "http": VyMCP runs as a shared server; each client authenticates per request. +VYMCP_TRANSPORT=stdio + +# stdio only: the personal access token this local server uses (Sites -> API +# Tokens; read-only recommended). Not used in http mode — clients present their +# own bearer token per request. +VYMANAGER_API_TOKEN=vym_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx + +# http only: bind address, port, and the externally reachable URL. +VYMCP_HOST=127.0.0.1 +VYMCP_PORT=8080 +# VYMCP_PUBLIC_URL=https://vymcp.example.com diff --git a/src/vymcp/auth.py b/src/vymcp/auth.py new file mode 100644 index 0000000..9e325e9 --- /dev/null +++ b/src/vymcp/auth.py @@ -0,0 +1,46 @@ +"""Bearer-token verification for the hosted (http) transport. + +Each request carries a ``vym_`` token; we validate it against VyManager and hand +back an AccessToken whose subject is a stable, token-derived id used to scope the +caller's pending plans. The raw token is carried through so tools can act as that +caller (VyManager still enforces the token's RBAC and records the real user). +""" + +from __future__ import annotations + +import hashlib +import time + +from mcp.server.auth.provider import AccessToken, TokenVerifier + +from .client import VyManagerClient +from .config import ServerConfig + +# A cheap authenticated endpoint: 200 = valid token, 401 = not. +_VALIDATE_PATH = "/session/sites" +_CACHE_TTL_SECONDS = 60.0 + + +class VyManagerTokenVerifier(TokenVerifier): + def __init__(self, server_config: ServerConfig) -> None: + self._server_config = server_config + self._cache: dict[str, tuple[float, AccessToken]] = {} + + async def verify_token(self, token: str) -> AccessToken | None: + now = time.monotonic() + cached = self._cache.get(token) + if cached and cached[0] > now: + return cached[1] + + client = VyManagerClient(self._server_config.client_config(token)) + try: + await client.get(_VALIDATE_PATH) + except Exception: + return None + finally: + await client.aclose() + + subject = "vym_" + hashlib.sha256(token.encode()).hexdigest()[:16] + access = AccessToken(token=token, client_id=subject, scopes=[], subject=subject) + self._cache[token] = (now + _CACHE_TTL_SECONDS, access) + return access diff --git a/src/vymcp/changes.py b/src/vymcp/changes.py index f9f42d1..6a96a4d 100644 --- a/src/vymcp/changes.py +++ b/src/vymcp/changes.py @@ -25,6 +25,7 @@ @dataclass class Plan: plan_id: str + owner: str # the caller who proposed it; only they may apply it instance_id: str feature: str path: str # VyManager endpoint the apply will POST to @@ -50,6 +51,7 @@ def _prune(self) -> None: def create( self, *, + owner: str, instance_id: str, feature: str, path: str, @@ -60,6 +62,7 @@ def create( self._prune() plan = Plan( plan_id="plan_" + secrets.token_hex(8), + owner=owner, instance_id=instance_id, feature=feature, path=path, @@ -70,9 +73,13 @@ def create( self._plans[plan.plan_id] = plan return plan - def get(self, plan_id: str) -> Plan | None: + def get(self, plan_id: str, owner: str) -> Plan | None: + """Return the plan only if it exists and belongs to ``owner``.""" self._prune() - return self._plans.get(plan_id) + plan = self._plans.get(plan_id) + if plan is None or plan.owner != owner: + return None + return plan def consume(self, plan_id: str) -> None: """Remove a plan after it has been successfully applied (single-use).""" diff --git a/src/vymcp/client.py b/src/vymcp/client.py index 6d9f25b..b0b4253 100644 --- a/src/vymcp/client.py +++ b/src/vymcp/client.py @@ -11,7 +11,7 @@ import httpx -from .config import Config +from .config import Config, ServerConfig logger = logging.getLogger("vymcp") @@ -122,28 +122,78 @@ def _explain_error(response: httpx.Response, instance_id: str | None) -> str: return f"VyManager request failed ({status})." + (f" {detail}" if detail else "") -_client: VyManagerClient | None = None +_server_config: ServerConfig | None = None +_clients: dict[str, VyManagerClient] = {} +_override: VyManagerClient | None = None # test hook -def get_client() -> VyManagerClient: - """Lazily build a shared VyManager client from the environment on first use.""" - global _client - if _client is None: - config = Config.from_env() - logger.info("Connecting to VyManager at %s", config.base_url) - _client = VyManagerClient(config) - return _client +def server_config() -> ServerConfig: + global _server_config + if _server_config is None: + _server_config = ServerConfig.from_env() + logger.info( + "VyManager %s (%s transport)", _server_config.base_url, _server_config.transport + ) + return _server_config + + +def _current_token() -> str: + """The token for the current request: the caller's bearer token in http mode, + or the server's env token in stdio mode.""" + try: + from mcp.server.auth.middleware.auth_context import get_access_token + + access = get_access_token() + except Exception: + access = None + if access is not None: + return access.token + + token = server_config().api_token + if not token: + raise VyManagerError("No API token available for this request.") + return token + + +def current_client() -> VyManagerClient: + """The VyManager client for the current request's identity. + + In http mode each caller's token gets its own (cached) client; in stdio mode + there is a single env-token client. Tests can override via set_client(). + """ + if _override is not None: + return _override + token = _current_token() + if token not in _clients: + _clients[token] = VyManagerClient(server_config().client_config(token)) + return _clients[token] + + +def current_owner() -> str: + """A stable identity for the current caller, used to scope pending plans.""" + try: + from mcp.server.auth.middleware.auth_context import get_access_token + + access = get_access_token() + except Exception: + access = None + if access is not None and access.subject: + return access.subject + return "local" def set_client(client: VyManagerClient | None) -> None: - """Override the shared client (used by tests).""" - global _client - _client = client + """Override the client for all requests (used by tests).""" + global _override + _override = client async def close_client() -> None: - """Close the shared client on shutdown.""" - global _client - if _client is not None: - await _client.aclose() - _client = None + """Close all cached clients on shutdown.""" + global _override + for client in list(_clients.values()): + await client.aclose() + _clients.clear() + if _override is not None: + await _override.aclose() + _override = None diff --git a/src/vymcp/config.py b/src/vymcp/config.py index 4bc5bd1..16590f7 100644 --- a/src/vymcp/config.py +++ b/src/vymcp/config.py @@ -15,47 +15,85 @@ def writes_enabled() -> bool: This is the operator kill-switch, independent of the token's own scope: even a read-write token cannot change config unless writes are explicitly enabled. """ - import os - return os.environ.get("VYMANAGER_ENABLE_WRITES", "false").strip().lower() in _TRUTHY +def _verify_ssl() -> bool: + return os.environ.get("VYMANAGER_VERIFY_SSL", "true").lower() not in _FALSEY + + +def _timeout() -> float: + try: + return float(os.environ.get("VYMANAGER_TIMEOUT", "30")) + except ValueError: + return 30.0 + + @dataclass(frozen=True) class Config: + """Per-client config: everything needed to talk to VyManager as one token.""" + base_url: str api_token: str verify_ssl: bool = True timeout: float = 30.0 + +@dataclass(frozen=True) +class ServerConfig: + """Server-level config. In http mode the token arrives per request, so + ``api_token`` is only required for stdio (single-tenant, local) mode.""" + + base_url: str + verify_ssl: bool = True + timeout: float = 30.0 + transport: str = "stdio" # "stdio" or "http" + host: str = "127.0.0.1" + port: int = 8080 + public_url: str = "http://127.0.0.1:8080" # externally reachable URL (http mode) + api_token: str | None = None + @classmethod - def from_env(cls) -> Config: + def from_env(cls) -> ServerConfig: base_url = os.environ.get("VYMANAGER_BASE_URL") - api_token = os.environ.get("VYMANAGER_API_TOKEN") + if not base_url: + raise RuntimeError("Missing required environment variable: VYMANAGER_BASE_URL.") - missing = [ - name - for name, value in ( - ("VYMANAGER_BASE_URL", base_url), - ("VYMANAGER_API_TOKEN", api_token), - ) - if not value - ] - if missing or base_url is None or api_token is None: + transport = os.environ.get("VYMCP_TRANSPORT", "stdio").strip().lower() + if transport not in {"stdio", "http"}: + raise RuntimeError(f"VYMCP_TRANSPORT must be 'stdio' or 'http', got '{transport}'.") + + api_token = os.environ.get("VYMANAGER_API_TOKEN") + if transport == "stdio" and not api_token: raise RuntimeError( - "Missing required environment variable(s): " - + ", ".join(missing) - + ". See .env.example." + "VYMANAGER_API_TOKEN is required for stdio transport. " + "(In http mode, clients present their own token per request.)" ) - verify_ssl = os.environ.get("VYMANAGER_VERIFY_SSL", "true").lower() not in _FALSEY try: - timeout = float(os.environ.get("VYMANAGER_TIMEOUT", "30")) + port = int(os.environ.get("VYMCP_PORT", "8080")) except ValueError: - timeout = 30.0 + port = 8080 + + host = os.environ.get("VYMCP_HOST", "127.0.0.1") + public_url = os.environ.get("VYMCP_PUBLIC_URL") or f"http://{host}:{port}" return cls( base_url=base_url.rstrip("/"), + verify_ssl=_verify_ssl(), + timeout=_timeout(), + transport=transport, + host=host, + port=port, + public_url=public_url.rstrip("/"), api_token=api_token, - verify_ssl=verify_ssl, - timeout=timeout, + ) + + def client_config(self, token: str) -> Config: + """Build a per-client Config for a specific token.""" + return Config( + base_url=self.base_url, + api_token=token, + verify_ssl=self.verify_ssl, + timeout=self.timeout, ) diff --git a/src/vymcp/discovery.py b/src/vymcp/discovery.py index 5e453cf..d1d603f 100644 --- a/src/vymcp/discovery.py +++ b/src/vymcp/discovery.py @@ -15,7 +15,7 @@ from typing import Any -from .client import get_client +from .client import current_client _feature_cache: dict[str, dict[str, Any]] = {} _openapi_spec: dict[str, Any] | None = None @@ -30,7 +30,7 @@ def clear_cache() -> None: async def _get_feature(feature: str) -> dict[str, Any]: if feature not in _feature_cache: - _feature_cache[feature] = await get_client().get(f"/vyos/operations/{feature}") + _feature_cache[feature] = await current_client().get(f"/vyos/operations/{feature}") return _feature_cache[feature] @@ -64,7 +64,7 @@ async def get_top_level_fields(feature: str) -> dict[str, dict[str, Any]]: global _openapi_spec if _openapi_spec is None: try: - _openapi_spec = await get_client().get("/openapi.json") + _openapi_spec = await current_client().get("/openapi.json") except Exception: return {} diff --git a/src/vymcp/server.py b/src/vymcp/server.py index 0fd8838..f0ad677 100644 --- a/src/vymcp/server.py +++ b/src/vymcp/server.py @@ -11,7 +11,7 @@ from mcp.server.fastmcp import FastMCP -from .client import close_client, get_client +from .client import close_client, current_client from .features import FEATURES, resolve_feature from .writes import register_write_tools @@ -35,7 +35,7 @@ async def list_instances() -> list[dict[str, Any]]: Returns each instance's ``instance_id`` (pass it to other tools), name, site, host, VyOS version, and whether it is active. """ - client = get_client() + client = current_client() sites = await client.get("/session/sites") instances: list[dict[str, Any]] = [] @@ -73,7 +73,7 @@ async def get_capabilities(feature: str, instance_id: str) -> dict[str, Any]: instance_id: An instance_id from list_instances. """ resolved = resolve_feature(feature) - return await get_client().get( + return await current_client().get( f"/vyos/{resolved.slug}/capabilities", instance_id=instance_id ) @@ -87,7 +87,7 @@ async def get_config(feature: str, instance_id: str) -> dict[str, Any]: instance_id: An instance_id from list_instances. """ resolved = resolve_feature(feature) - return await get_client().get( + return await current_client().get( f"/vyos/{resolved.slug}/config", instance_id=instance_id ) @@ -98,8 +98,25 @@ async def get_config(feature: str, instance_id: str) -> dict[str, Any]: def main() -> None: - """Console-script entry point. Runs the server over stdio.""" - mcp.run() + """Console-script entry point. Runs stdio or a hosted http server per config.""" + from .client import server_config + + config = server_config() + if config.transport == "http": + from mcp.server.auth.settings import AuthSettings + + from .auth import VyManagerTokenVerifier + + mcp.settings.host = config.host + mcp.settings.port = config.port + mcp.settings.auth = AuthSettings( + issuer_url=config.base_url, # type: ignore[arg-type] # pydantic coerces str->AnyHttpUrl + resource_server_url=config.public_url, # type: ignore[arg-type] + ) + mcp._token_verifier = VyManagerTokenVerifier(config) + mcp.run(transport="streamable-http") + else: + mcp.run() if __name__ == "__main__": diff --git a/src/vymcp/writes.py b/src/vymcp/writes.py index baf4cc6..bf1d2c6 100644 --- a/src/vymcp/writes.py +++ b/src/vymcp/writes.py @@ -12,7 +12,7 @@ from . import discovery from .changes import Plan, plan_store -from .client import VyManagerError, get_client +from .client import VyManagerError, current_client, current_owner from .config import writes_enabled from .validation import validate_identifier, validate_values @@ -41,7 +41,7 @@ def _plan_response(plan: Plan) -> dict[str, Any]: async def _safe_get(path: str, instance_id: str) -> Any | None: """Best-effort read used to enrich an apply response; never fails the apply.""" try: - return await get_client().get(path, instance_id=instance_id) + return await current_client().get(path, instance_id=instance_id) except VyManagerError: return None @@ -83,6 +83,7 @@ def propose_create_address_group( + (f"; description {description!r}" if description else "") ) plan = plan_store.create( + owner=current_owner(), instance_id=instance_id, feature="firewall/groups", path=FIREWALL_GROUPS_PATH, @@ -105,6 +106,7 @@ def propose_add_address_group_members( f"{instance_id}: {', '.join(members)}" ) plan = plan_store.create( + owner=current_owner(), instance_id=instance_id, feature="firewall/groups", path=FIREWALL_GROUPS_PATH, @@ -127,6 +129,7 @@ def propose_remove_address_group_members( f"{instance_id}: {', '.join(members)}" ) plan = plan_store.create( + owner=current_owner(), instance_id=instance_id, feature="firewall/groups", path=FIREWALL_GROUPS_PATH, @@ -149,6 +152,7 @@ def propose_delete_address_group(instance_id: str, name: str) -> dict[str, Any]: "Any firewall rule referencing it may be affected." ) plan = plan_store.create( + owner=current_owner(), instance_id=instance_id, feature="firewall/groups", path=FIREWALL_GROUPS_PATH, @@ -243,6 +247,7 @@ async def propose_operations( + (f"; fields {fields}" if fields else "") ) plan = plan_store.create( + owner=current_owner(), instance_id=instance_id, feature=feature, path=f"/vyos/{feature}/batch", @@ -271,14 +276,14 @@ async def apply_change(plan_id: str, confirm: bool = False) -> dict[str, Any]: "Refusing to apply: set confirm=true only after the user has reviewed " "and approved this exact plan." ) - plan = plan_store.get(plan_id) + plan = plan_store.get(plan_id, current_owner()) if plan is None: raise ValueError( "Unknown or expired plan_id. Re-run the propose tool, show the user the " "new plan, then apply it." ) - client = get_client() + client = current_client() # Prime VyManager's diff baseline before changing anything. Its snapshot is # initialized on first read, so without this the post-change diff/discard @@ -333,7 +338,7 @@ async def apply_change(plan_id: str, confirm: bool = False) -> dict[str, Any]: @mcp.tool() async def confirm_change(instance_id: str) -> dict[str, Any]: """Confirm an active commit-confirm, making the change permanent and saving it.""" - result = await get_client().post(_CC_CONFIRM, instance_id=instance_id) + result = await current_client().post(_CC_CONFIRM, instance_id=instance_id) return { "confirmed": bool(result.get("success", True)) if isinstance(result, dict) else True, "message": result.get("message") if isinstance(result, dict) else None, @@ -355,7 +360,7 @@ async def discard_changes(instance_id: str) -> dict[str, Any]: "If the change locked you out, wait for the auto-revert. Otherwise call " "confirm_change(instance_id), then apply the inverse change to undo it." ) - result = await get_client().post(_DISCARD, instance_id=instance_id) + result = await current_client().post(_DISCARD, instance_id=instance_id) return { "discarded": bool(result.get("success", True)) if isinstance(result, dict) else True, "message": result.get("message") if isinstance(result, dict) else None, @@ -365,7 +370,7 @@ async def discard_changes(instance_id: str) -> dict[str, Any]: @mcp.tool() async def get_pending_changes(instance_id: str) -> dict[str, Any]: """Show unsaved changes and any active commit-confirm rollback timer for an instance.""" - client = get_client() + client = current_client() diff = await client.get(_CONFIG_DIFF, instance_id=instance_id) status = await client.get(_CC_STATUS, instance_id=instance_id) return { diff --git a/tests/test_changes.py b/tests/test_changes.py index 8d39168..f26e425 100644 --- a/tests/test_changes.py +++ b/tests/test_changes.py @@ -3,8 +3,9 @@ from vymcp.changes import PlanStore -def _make(store, instance_id="inst1"): +def _make(store, owner="u1", instance_id="inst1"): return store.create( + owner=owner, instance_id=instance_id, feature="firewall/groups", path="/vyos/firewall/groups/batch", @@ -18,25 +19,30 @@ def test_create_and_get(): store = PlanStore() plan = _make(store) assert plan.plan_id.startswith("plan_") - assert store.get(plan.plan_id) is plan + assert store.get(plan.plan_id, "u1") is plan assert plan.instance_id == "inst1" +def test_get_enforces_owner(): + store = PlanStore() + plan = _make(store, owner="alice") + assert store.get(plan.plan_id, "bob") is None # not yours + assert store.get(plan.plan_id, "alice") is plan + + def test_consume_is_single_use(): store = PlanStore() plan = _make(store) store.consume(plan.plan_id) - assert store.get(plan.plan_id) is None - # consuming again is a no-op, not an error - store.consume(plan.plan_id) + assert store.get(plan.plan_id, "u1") is None + store.consume(plan.plan_id) # no-op, not an error def test_expired_plan_is_pruned(): store = PlanStore(ttl_seconds=60) plan = _make(store) - # Backdate creation beyond the TTL. store._plans[plan.plan_id].created_at = time.monotonic() - 600 - assert store.get(plan.plan_id) is None + assert store.get(plan.plan_id, "u1") is None def test_unique_ids(): diff --git a/tests/test_config.py b/tests/test_config.py index 9b85a7a..cad419f 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,39 +1,63 @@ import pytest -from vymcp.config import Config, writes_enabled +from vymcp.config import ServerConfig, writes_enabled -def test_from_env_requires_url_and_token(monkeypatch): +def test_requires_base_url(monkeypatch): monkeypatch.delenv("VYMANAGER_BASE_URL", raising=False) + with pytest.raises(RuntimeError, match="VYMANAGER_BASE_URL"): + ServerConfig.from_env() + + +def test_stdio_requires_token(monkeypatch): + monkeypatch.setenv("VYMANAGER_BASE_URL", "http://h") monkeypatch.delenv("VYMANAGER_API_TOKEN", raising=False) - with pytest.raises(RuntimeError) as exc: - Config.from_env() - assert "VYMANAGER_BASE_URL" in str(exc.value) - assert "VYMANAGER_API_TOKEN" in str(exc.value) + monkeypatch.delenv("VYMCP_TRANSPORT", raising=False) # default stdio + with pytest.raises(RuntimeError, match="VYMANAGER_API_TOKEN"): + ServerConfig.from_env() -def test_from_env_parses_values(monkeypatch): - monkeypatch.setenv("VYMANAGER_BASE_URL", "http://host:8000/") - monkeypatch.setenv("VYMANAGER_API_TOKEN", "vym_abc") - monkeypatch.setenv("VYMANAGER_VERIFY_SSL", "false") - monkeypatch.setenv("VYMANAGER_TIMEOUT", "12") - cfg = Config.from_env() - assert cfg.base_url == "http://host:8000" # trailing slash stripped - assert cfg.api_token == "vym_abc" - assert cfg.verify_ssl is False - assert cfg.timeout == 12.0 +def test_http_does_not_require_token(monkeypatch): + monkeypatch.setenv("VYMANAGER_BASE_URL", "http://h:8000/") + monkeypatch.setenv("VYMCP_TRANSPORT", "http") + monkeypatch.delenv("VYMANAGER_API_TOKEN", raising=False) + cfg = ServerConfig.from_env() + assert cfg.transport == "http" + assert cfg.base_url == "http://h:8000" # trailing slash stripped + assert cfg.api_token is None + + +def test_http_public_url_default(monkeypatch): + monkeypatch.setenv("VYMANAGER_BASE_URL", "http://h") + monkeypatch.setenv("VYMCP_TRANSPORT", "http") + monkeypatch.setenv("VYMCP_HOST", "0.0.0.0") + monkeypatch.setenv("VYMCP_PORT", "9000") + monkeypatch.delenv("VYMCP_PUBLIC_URL", raising=False) + cfg = ServerConfig.from_env() + assert cfg.public_url == "http://0.0.0.0:9000" -def test_verify_ssl_defaults_true(monkeypatch): +def test_client_config_carries_token(monkeypatch): monkeypatch.setenv("VYMANAGER_BASE_URL", "http://h") monkeypatch.setenv("VYMANAGER_API_TOKEN", "vym_x") - monkeypatch.delenv("VYMANAGER_VERIFY_SSL", raising=False) - assert Config.from_env().verify_ssl is True + monkeypatch.setenv("VYMANAGER_VERIFY_SSL", "false") + cfg = ServerConfig.from_env() + cc = cfg.client_config("vym_abc") + assert cc.api_token == "vym_abc" + assert cc.base_url == "http://h" + assert cc.verify_ssl is False + + +def test_bad_transport(monkeypatch): + monkeypatch.setenv("VYMANAGER_BASE_URL", "http://h") + monkeypatch.setenv("VYMCP_TRANSPORT", "carrier-pigeon") + with pytest.raises(RuntimeError, match="stdio.*http|http.*stdio"): + ServerConfig.from_env() @pytest.mark.parametrize("value,expected", [ - ("true", True), ("1", True), ("yes", True), ("on", True), - ("false", False), ("0", False), ("", False), ("nope", False), + ("true", True), ("1", True), ("on", True), + ("false", False), ("", False), ("nope", False), ]) def test_writes_enabled(monkeypatch, value, expected): monkeypatch.setenv("VYMANAGER_ENABLE_WRITES", value) diff --git a/tests/test_generic_driver.py b/tests/test_generic_driver.py index b9a03ea..604a5c2 100644 --- a/tests/test_generic_driver.py +++ b/tests/test_generic_driver.py @@ -72,7 +72,7 @@ async def test_propose_builds_body_with_fields(write_tools, install_client): ) assert plan["applied"] is False from vymcp.changes import plan_store - body = plan_store.get(plan["plan_id"]).body + body = plan_store.get(plan["plan_id"], "local").body assert body == {"nat_type": "source", "operations": [{"op": "set_source_rule", "value": "100"}]} @@ -92,7 +92,7 @@ async def test_propose_zero_arg_op_needs_no_value(write_tools, install_client): install_client(_discovery_handler()) plan = await write_tools["propose_operations"]("nat", "i1", [{"op": "delete_all"}]) from vymcp.changes import plan_store - assert plan_store.get(plan["plan_id"]).body["operations"] == [{"op": "delete_all"}] + assert plan_store.get(plan["plan_id"], "local").body["operations"] == [{"op": "delete_all"}] async def test_propose_rejects_empty(write_tools, install_client): @@ -135,7 +135,7 @@ async def test_subject_feature_zero_value_op_ok(write_tools, install_client): "bonding", "i1", [{"op": "set_interface_disable"}], fields={"interface_name": "bond0"} ) from vymcp.changes import plan_store - body = plan_store.get(plan["plan_id"]).body + body = plan_store.get(plan["plan_id"], "local").body assert body == {"interface_name": "bond0", "operations": [{"op": "set_interface_disable"}]} diff --git a/tests/test_write_tools.py b/tests/test_write_tools.py index f837d31..532ea99 100644 --- a/tests/test_write_tools.py +++ b/tests/test_write_tools.py @@ -71,7 +71,7 @@ async def test_apply_logical_failure_not_applied(write_tools, install_client): assert "rejected" in res["error"] # a failed apply leaves the plan usable for a retry from vymcp.changes import plan_store - assert plan_store.get(plan["plan_id"]) is not None + assert plan_store.get(plan["plan_id"], "local") is not None async def test_apply_read_only_token_blocked(write_tools, install_client):