diff --git a/src/mcp_synology_container/modules/networks.py b/src/mcp_synology_container/modules/networks.py new file mode 100644 index 0000000..da012e8 --- /dev/null +++ b/src/mcp_synology_container/modules/networks.py @@ -0,0 +1,155 @@ +"""MCP tools for SYNO.Docker.Network: list, create, delete.""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from mcp.server.fastmcp import FastMCP + + from mcp_synology_container.config import AppConfig + from mcp_synology_container.dsm_client import DsmClient + +logger = logging.getLogger(__name__) + + +def register_networks(mcp: FastMCP, config: AppConfig, client: DsmClient) -> None: + """Register all network management tools with the MCP server.""" + + @mcp.tool() + async def list_networks() -> str: + """List all Docker networks with driver, subnet, gateway, and attached containers.""" + try: + data = await client.request("SYNO.Docker.Network", "list") + except Exception as e: + return f"Error listing networks: {e}" + + networks: list[dict[str, Any]] = data.get("network", []) + if not networks: + return "No networks found." + + lines = [f"Networks ({len(networks)} total):", ""] + for net in sorted(networks, key=lambda n: n.get("name", "")): + name = net.get("name", "?") + driver = net.get("driver", "?") + subnet = net.get("subnet") or "—" + gateway = net.get("gateway") or "—" + ipv6 = "IPv6" if net.get("enable_ipv6") else "" + containers: list[str] = net.get("containers") or [] + + lines.append(f" {name}") + lines.append(f" Driver: {driver}{(' ' + ipv6) if ipv6 else ''}") + lines.append(f" Subnet: {subnet}") + lines.append(f" Gateway: {gateway}") + if containers: + lines.append(f" Containers ({len(containers)}): {', '.join(containers)}") + else: + lines.append(" Containers: none") + lines.append("") + + return "\n".join(lines).rstrip() + + @mcp.tool() + async def create_network( + name: str, + driver: str = "bridge", + subnet: str | None = None, + gateway: str | None = None, + ip_range: str | None = None, + enable_ipv6: bool = False, + confirmed: bool = False, + ) -> str: + """Create a new Docker network. + + Args: + name: Name for the new network. + driver: Network driver. Defaults to "bridge". + subnet: Subnet in CIDR notation (e.g. "172.28.0.0/16"). + gateway: Gateway IP address. + ip_range: Allocatable IP range in CIDR notation. + enable_ipv6: Enable IPv6 support. Defaults to False. + confirmed: Must be True to actually create. Default False shows a preview. + """ + details = [f" Name: {name}", f" Driver: {driver}"] + if subnet: + details.append(f" Subnet: {subnet}") + if gateway: + details.append(f" Gateway: {gateway}") + if ip_range: + details.append(f" IP range:{ip_range}") + if enable_ipv6: + details.append(" IPv6: enabled") + + if not confirmed: + return ( + "Preview: would create network:\n" + + "\n".join(details) + + f"\n\nCall create_network(name={name!r}, confirmed=True) to confirm." + ) + + params: dict[str, Any] = { + "name": name, + "driver": driver, + "enable_ipv6": json.dumps(enable_ipv6), + } + if subnet is not None: + params["subnet"] = subnet + if gateway is not None: + params["gateway"] = gateway + if ip_range is not None: + params["iprange"] = ip_range + + try: + result = await client.request("SYNO.Docker.Network", "create", params=params) + except Exception as e: + return f"Error creating network '{name}': {e}" + + net_id = result.get("id", "") + id_str = f" (ID: {net_id[:12]})" if net_id else "" + return f"Network '{name}' created{id_str}." + + @mcp.tool() + async def delete_network(name: str, confirmed: bool = False) -> str: + """Delete a Docker network by name. + + Refuses deletion if any container is still attached to the network. + + Args: + name: Name of the network to delete. + confirmed: Must be True to actually delete. Default False shows a preview. + """ + # Fetch network list to validate existence and check for attached containers + try: + data = await client.request("SYNO.Docker.Network", "list") + except Exception as e: + return f"Error fetching network list: {e}" + + networks: list[dict[str, Any]] = data.get("network", []) + target = next((n for n in networks if n.get("name") == name), None) + + if target is None: + return f"Network '{name}' not found." + + attached: list[str] = target.get("containers") or [] + if attached: + return ( + f"Cannot delete network '{name}': " + f"{len(attached)} container(s) attached: {', '.join(attached)}" + ) + + if not confirmed: + driver = target.get("driver", "?") + subnet = target.get("subnet") or "—" + return ( + f"Preview: would delete network '{name}' (driver={driver}, subnet={subnet}).\n" + f"Call delete_network(name={name!r}, confirmed=True) to confirm." + ) + + try: + await client.request("SYNO.Docker.Network", "delete", params={"name": name}) + except Exception as e: + return f"Error deleting network '{name}': {e}" + + return f"Network '{name}' deleted." diff --git a/src/mcp_synology_container/server.py b/src/mcp_synology_container/server.py index 43dab7c..4caccd6 100644 --- a/src/mcp_synology_container/server.py +++ b/src/mcp_synology_container/server.py @@ -29,6 +29,7 @@ def create_server(config: AppConfig, client: DsmClient) -> FastMCP: from mcp_synology_container.modules.compose import register_compose from mcp_synology_container.modules.containers import register_containers from mcp_synology_container.modules.images import register_images + from mcp_synology_container.modules.networks import register_networks from mcp_synology_container.modules.projects import register_projects from mcp_synology_container.modules.system import register_system @@ -37,6 +38,7 @@ def create_server(config: AppConfig, client: DsmClient) -> FastMCP: register_compose(mcp, config, client) register_images(mcp, config, client) register_system(mcp, config, client) + register_networks(mcp, config, client) logger.info("MCP server configured with all tool modules") return mcp diff --git a/tests/test_modules/test_networks.py b/tests/test_modules/test_networks.py new file mode 100644 index 0000000..dcf3127 --- /dev/null +++ b/tests/test_modules/test_networks.py @@ -0,0 +1,344 @@ +"""Tests for modules/networks.py.""" + +import json +from unittest.mock import AsyncMock + +import pytest + + +def make_mock_mcp(): + tools: dict = {} + + class MockMCP: + def tool(self): + def decorator(fn): + tools[fn.__name__] = fn + return fn + + return decorator + + return MockMCP(), tools + + +def make_config(): + from mcp_synology_container.config import AppConfig, ConnectionConfig + + return AppConfig( + schema_version=1, + connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True), + ) + + +SAMPLE_NETWORKS = { + "network": [ + { + "id": "b741915823aa", + "name": "vault_default", + "driver": "bridge", + "subnet": "172.22.0.0/16", + "gateway": "172.22.0.1", + "iprange": "", + "enable_ipv6": False, + "containers": ["vault"], + }, + { + "id": "2fc7ebae3901", + "name": "host", + "driver": "host", + "subnet": "", + "gateway": "", + "iprange": "", + "enable_ipv6": False, + "containers": [], + }, + { + "id": "aabbcc112233", + "name": "my_bridge", + "driver": "bridge", + "subnet": "10.10.0.0/24", + "gateway": "10.10.0.1", + "iprange": "", + "enable_ipv6": False, + "containers": [], + }, + ] +} + + +# ────────────────────────────────────────────────────────────────────────────── +# list_networks +# ────────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_list_networks_shows_all(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = SAMPLE_NETWORKS + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["list_networks"]() + assert "vault_default" in result + assert "host" in result + assert "my_bridge" in result + assert "3 total" in result + + +@pytest.mark.asyncio +async def test_list_networks_shows_subnet_and_gateway(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = SAMPLE_NETWORKS + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["list_networks"]() + assert "172.22.0.0/16" in result + assert "172.22.0.1" in result + + +@pytest.mark.asyncio +async def test_list_networks_shows_attached_containers(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = SAMPLE_NETWORKS + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["list_networks"]() + assert "vault" in result + + +@pytest.mark.asyncio +async def test_list_networks_empty(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = {"network": []} + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["list_networks"]() + assert "No networks found" in result + + +@pytest.mark.asyncio +async def test_list_networks_api_error(): + from mcp_synology_container.dsm_client import SynologyError + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.side_effect = SynologyError("API error", code=102) + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["list_networks"]() + assert "Error" in result + + +# ────────────────────────────────────────────────────────────────────────────── +# create_network +# ────────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_create_network_preview(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["create_network"](name="mynet", driver="bridge") + assert "Preview" in result + assert "mynet" in result + assert "confirmed=True" in result + client.request.assert_not_called() + + +@pytest.mark.asyncio +async def test_create_network_confirmed(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = {"id": "deadbeef1234567890"} + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["create_network"]( + name="mynet", driver="bridge", subnet="192.168.100.0/24", confirmed=True + ) + assert "created" in result + assert "mynet" in result + + call = client.request.call_args + params = call.kwargs.get("params") or {} + assert params["name"] == "mynet" + assert params["driver"] == "bridge" + assert params["subnet"] == "192.168.100.0/24" + assert json.loads(params["enable_ipv6"]) is False + + +@pytest.mark.asyncio +async def test_create_network_with_ipv6(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = {"id": "abc123"} + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + await tools["create_network"](name="ipv6net", enable_ipv6=True, confirmed=True) + + call = client.request.call_args + params = call.kwargs.get("params") or {} + assert json.loads(params["enable_ipv6"]) is True + + +@pytest.mark.asyncio +async def test_create_network_optional_params_not_sent(): + """subnet/gateway/ip_range must not appear in params when not provided.""" + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = {} + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + await tools["create_network"](name="bare", confirmed=True) + + call = client.request.call_args + params = call.kwargs.get("params") or {} + assert "subnet" not in params + assert "gateway" not in params + assert "iprange" not in params + + +@pytest.mark.asyncio +async def test_create_network_api_error(): + from mcp_synology_container.dsm_client import SynologyError + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.side_effect = SynologyError("create failed", code=100) + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["create_network"](name="mynet", confirmed=True) + assert "Error" in result + + +# ────────────────────────────────────────────────────────────────────────────── +# delete_network +# ────────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_delete_network_preview(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = SAMPLE_NETWORKS + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["delete_network"](name="my_bridge") + assert "Preview" in result + assert "my_bridge" in result + assert "confirmed=True" in result + # Only the list call should have been made, not delete + assert client.request.call_count == 1 + assert client.request.call_args.args[1] == "list" + + +@pytest.mark.asyncio +async def test_delete_network_confirmed(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + + async def mock_request(api, method, **kwargs): + if method == "list": + return SAMPLE_NETWORKS + if method == "delete": + return {} + return {} + + client.request.side_effect = mock_request + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["delete_network"](name="my_bridge", confirmed=True) + assert "deleted" in result + assert "my_bridge" in result + + +@pytest.mark.asyncio +async def test_delete_network_not_found(): + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = SAMPLE_NETWORKS + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["delete_network"](name="nonexistent", confirmed=True) + assert "not found" in result + + +@pytest.mark.asyncio +async def test_delete_network_blocked_by_containers(): + """Deletion must be refused when containers are attached.""" + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + client.request.return_value = SAMPLE_NETWORKS + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + # vault_default has ["vault"] attached + result = await tools["delete_network"](name="vault_default", confirmed=True) + assert "Cannot delete" in result + assert "vault" in result + # delete API must not be called + assert client.request.call_count == 1 + assert client.request.call_args.args[1] == "list" + + +@pytest.mark.asyncio +async def test_delete_network_api_error(): + from mcp_synology_container.dsm_client import SynologyError + from mcp_synology_container.modules.networks import register_networks + + client = AsyncMock() + + async def mock_request(api, method, **kwargs): + if method == "list": + return SAMPLE_NETWORKS + raise SynologyError("delete failed", code=100) + + client.request.side_effect = mock_request + + mcp, tools = make_mock_mcp() + register_networks(mcp, make_config(), client) + + result = await tools["delete_network"](name="my_bridge", confirmed=True) + assert "Error" in result