diff --git a/README.md b/README.md index cfc7320..2205345 100644 --- a/README.md +++ b/README.md @@ -96,8 +96,17 @@ cforge prompts execute # MCP Servers cforge mcp-servers list cforge mcp-servers update [file.json] + +# Plugins (read-only admin API) +cforge plugins list [--search text] [--mode MODE] [--hook HOOK] [--tag TAG] [--json] +cforge plugins get +cforge plugins stats ``` +Plugin commands call `/admin/plugins` endpoints and require: +- `MCPGATEWAY_ADMIN_API_ENABLED=true` on the gateway +- A token with `admin.plugins` permission + ### Server Operations ```bash diff --git a/cforge/commands/resources/plugins.py b/cforge/commands/resources/plugins.py new file mode 100644 index 0000000..afd2483 --- /dev/null +++ b/cforge/commands/resources/plugins.py @@ -0,0 +1,133 @@ +# -*- coding: utf-8 -*- +"""Location: ./cforge/commands/resources/plugins.py +Copyright 2025 +SPDX-License-Identifier: Apache-2.0 +Authors: Matthew Grigsby + +CLI command group: plugins + +Note: + The CLI currently exposes read-only operations (list/get/stats) for plugins. + This matches the current capabilities of the gateway admin API: plugin + configuration is loaded from a YAML file at gateway startup, and the gateway + does not yet provide write endpoints for plugin CRUD/management. When + mcp-context-forge adds server-side write operations, this CLI can be extended + to support them. +""" + +# Standard +from enum import Enum +from typing import Any, Dict, Optional + +# Third-Party +import typer + +# First-Party +from cforge.common import ( + AuthenticationError, + CLIError, + get_console, + handle_exception, + make_authenticated_request, + print_json, + print_table, +) + + +class _CaseInsensitiveEnum(str, Enum): + """Enum that supports case-insensitive parsing for CLI options.""" + + @classmethod + def _missing_(cls, value: object) -> Optional["_CaseInsensitiveEnum"]: + """Resolve unknown values by matching enum values case-insensitively. + + Typer converts CLI strings into Enum members. Implementing `_missing_` + allows `--mode EnFoRcE` to resolve to `PluginMode.ENFORCE`, while still + rejecting unknown values. + """ + if not isinstance(value, str): + return None + value_folded = value.casefold() + for member in cls: + if member.value.casefold() == value_folded: + return member + return None + + +class PluginMode(_CaseInsensitiveEnum): + """Valid plugin mode filters supported by the gateway admin API.""" + + ENFORCE = "enforce" + PERMISSIVE = "permissive" + DISABLED = "disabled" + + +def _handle_plugins_exception(exception: Exception) -> None: + """Provide plugin-specific hints and raise a CLI error.""" + console = get_console() + + if isinstance(exception, AuthenticationError): + console.print("[yellow]Access denied. Requires admin.plugins permission.[/yellow]") + elif isinstance(exception, CLIError) and "(404)" in str(exception): + console.print("[yellow]Admin plugin API unavailable. Ensure MCPGATEWAY_ADMIN_API_ENABLED=true and gateway version supports /admin/plugins.[/yellow]") + + handle_exception(exception) + + +def plugins_list( + search: Optional[str] = typer.Option(None, "--search", help="Search by plugin name, description, or author"), + mode: Optional[PluginMode] = typer.Option(None, "--mode", help="Filter by mode"), + hook: Optional[str] = typer.Option(None, "--hook", help="Filter by hook type"), + tag: Optional[str] = typer.Option(None, "--tag", help="Filter by plugin tag"), + json_output: bool = typer.Option(False, "--json", help="Output as JSON"), +) -> None: + """List all plugins with optional filtering.""" + console = get_console() + + try: + params: Dict[str, Any] = {} + if search: + params["search"] = search + if mode: + params["mode"] = mode.value + if hook: + params["hook"] = hook + if tag: + params["tag"] = tag + + result = make_authenticated_request("GET", "/admin/plugins", params=params if params else None) + + if json_output: + print_json(result, "Plugins") + else: + plugins: list[dict[str, Any]] = result["plugins"] + + if plugins: + print_table(plugins, "Plugins", ["name", "version", "author", "mode", "status", "priority", "hooks", "tags"]) + else: + console.print("[yellow]No plugins found[/yellow]") + + except Exception as e: + _handle_plugins_exception(e) + + +def plugins_get( + name: str = typer.Argument(..., help="Plugin name"), +) -> None: + """Get details for a specific plugin.""" + try: + result = make_authenticated_request("GET", f"/admin/plugins/{name}") + print_json(result, f"Plugin {name}") + + except Exception as e: + _handle_plugins_exception(e) + + +def plugins_stats() -> None: + """Get plugin statistics.""" + try: + result = make_authenticated_request("GET", "/admin/plugins/stats") + print_json(result, "Plugin Statistics") + + except Exception as e: + _handle_plugins_exception(e) diff --git a/cforge/main.py b/cforge/main.py index 0e61b56..0d7bcaa 100644 --- a/cforge/main.py +++ b/cforge/main.py @@ -97,6 +97,11 @@ a2a_toggle, a2a_invoke, ) +from cforge.commands.resources.plugins import ( + plugins_get, + plugins_list, + plugins_stats, +) # Get the main app singleton app = get_app() @@ -232,6 +237,17 @@ a2a_app.command("toggle")(a2a_toggle) a2a_app.command("invoke")(a2a_invoke) +# --------------------------------------------------------------------------- +# Plugins command group +# --------------------------------------------------------------------------- + +plugins_app = typer.Typer(help="Manage gateway plugins (read-only)") +app.add_typer(plugins_app, name="plugins", rich_help_panel="Resources") + +plugins_app.command("list")(plugins_list) +plugins_app.command("get")(plugins_get) +plugins_app.command("stats")(plugins_stats) + # --------------------------------------------------------------------------- # Metrics command group # --------------------------------------------------------------------------- diff --git a/tests/commands/resources/test_plugins.py b/tests/commands/resources/test_plugins.py new file mode 100644 index 0000000..dd8e1fc --- /dev/null +++ b/tests/commands/resources/test_plugins.py @@ -0,0 +1,154 @@ +# -*- coding: utf-8 -*- +"""Location: ./tests/commands/resources/test_plugins.py +Copyright 2025 +SPDX-License-Identifier: Apache-2.0 +Authors: Matthew Grigsby + +Tests for the plugins commands. +""" + +# Third-Party +import pytest +import typer + +# First-Party +from cforge.commands.resources.plugins import PluginMode, plugins_get, plugins_list, plugins_stats +from cforge.common import AuthenticationError, CLIError +from tests.conftest import invoke_typer_command, patch_functions + + +class TestPluginCommands: + """Tests for plugins commands.""" + + def test_plugin_mode_enum_is_case_insensitive(self) -> None: + """Typer Enum choices should accept case-insensitive values.""" + assert PluginMode("EnFoRcE") == PluginMode.ENFORCE + + def test_plugin_mode_enum_missing_non_string(self) -> None: + """Non-string values should not be coerced into Enum members.""" + assert PluginMode._missing_(123) is None + + def test_plugin_mode_enum_missing_unknown_value(self) -> None: + """Unknown strings should not be coerced into Enum members.""" + assert PluginMode._missing_("nope") is None + + def test_plugins_list_success(self, mock_console) -> None: + """Test plugins list command with table output.""" + mock_response = { + "plugins": [ + {"name": "pii_filter", "version": "1.0.0", "author": "ContextForge", "mode": "enforce", "status": "enabled", "priority": 10, "hooks": ["tool_pre_invoke"], "tags": ["security"]} + ], + "total": 1, + "enabled_count": 1, + "disabled_count": 0, + } + + with patch_functions( + "cforge.commands.resources.plugins", + get_console=mock_console, + make_authenticated_request={"return_value": mock_response}, + print_table=None, + ) as mocks: + invoke_typer_command(plugins_list) + mocks.print_table.assert_called_once() + + def test_plugins_list_json_output(self, mock_console) -> None: + """Test plugins list with JSON output.""" + mock_response = {"plugins": [], "total": 0, "enabled_count": 0, "disabled_count": 0} + with patch_functions( + "cforge.commands.resources.plugins", + get_console=mock_console, + make_authenticated_request={"return_value": mock_response}, + print_json=None, + ) as mocks: + invoke_typer_command(plugins_list, json_output=True) + mocks.print_json.assert_called_once() + + def test_plugins_list_no_results(self, mock_console) -> None: + """Test plugins list with no results.""" + mock_response = {"plugins": [], "total": 0, "enabled_count": 0, "disabled_count": 0} + with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"return_value": mock_response}): + invoke_typer_command(plugins_list) + + assert any("No plugins found" in str(call) for call in mock_console.print.call_args_list) + + def test_plugins_list_with_filters(self, mock_console) -> None: + """Test plugins list with all filters.""" + with patch_functions( + "cforge.commands.resources.plugins", + get_console=mock_console, + make_authenticated_request={"return_value": {"plugins": [], "total": 0, "enabled_count": 0, "disabled_count": 0}}, + print_table=None, + ) as mocks: + invoke_typer_command(plugins_list, search="pii", mode=PluginMode.ENFORCE, hook="tool_pre_invoke", tag="security") + + call_args = mocks.make_authenticated_request.call_args + assert call_args[0][0] == "GET" + assert call_args[0][1] == "/admin/plugins" + assert call_args[1]["params"] == {"search": "pii", "mode": "enforce", "hook": "tool_pre_invoke", "tag": "security"} + + def test_plugins_list_error(self, mock_console) -> None: + """Test plugins list error handling.""" + with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"side_effect": Exception("API error")}): + with pytest.raises(typer.Exit): + invoke_typer_command(plugins_list) + + def test_plugins_get_success(self, mock_console) -> None: + """Test plugins get command.""" + mock_plugin = {"name": "pii_filter", "version": "1.0.0"} + with patch_functions( + "cforge.commands.resources.plugins", + get_console=mock_console, + make_authenticated_request={"return_value": mock_plugin}, + print_json=None, + ) as mocks: + invoke_typer_command(plugins_get, name="pii_filter") + mocks.print_json.assert_called_once() + + def test_plugins_get_error(self, mock_console) -> None: + """Test plugins get error handling.""" + with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"side_effect": Exception("API error")}): + with pytest.raises(typer.Exit): + invoke_typer_command(plugins_get, name="pii_filter") + + def test_plugins_stats_success(self, mock_console) -> None: + """Test plugins stats command.""" + mock_stats = {"total_plugins": 4, "enabled_plugins": 3, "disabled_plugins": 1, "plugins_by_hook": {"tool_pre_invoke": 3}, "plugins_by_mode": {"enforce": 3, "disabled": 1}} + with patch_functions( + "cforge.commands.resources.plugins", + get_console=mock_console, + make_authenticated_request={"return_value": mock_stats}, + print_json=None, + ) as mocks: + invoke_typer_command(plugins_stats) + mocks.print_json.assert_called_once() + + def test_plugins_stats_error(self, mock_console) -> None: + """Test plugins stats error handling.""" + with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"side_effect": Exception("API error")}): + with pytest.raises(typer.Exit): + invoke_typer_command(plugins_stats) + + def test_plugins_list_forbidden_shows_permission_hint(self, mock_console) -> None: + """Test plugins list shows a targeted hint on forbidden/admin failures.""" + with patch_functions( + "cforge.commands.resources.plugins", + get_console=mock_console, + make_authenticated_request={"side_effect": AuthenticationError("Authentication required but not configured")}, + ): + with pytest.raises(typer.Exit): + invoke_typer_command(plugins_list) + + assert any("Requires admin.plugins permission" in str(call) for call in mock_console.print.call_args_list) + + def test_plugins_list_not_found_shows_admin_api_hint(self, mock_console) -> None: + """Test plugins list shows an admin-api hint on 404.""" + with patch_functions( + "cforge.commands.resources.plugins", + get_console=mock_console, + make_authenticated_request={"side_effect": CLIError("API request failed (404): Not Found")}, + ): + with pytest.raises(typer.Exit): + invoke_typer_command(plugins_list) + + assert any("Admin plugin API unavailable" in str(call) for call in mock_console.print.call_args_list)