diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index 67165b5d..c98198da 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -91,7 +91,9 @@ class CommandGroup(BaseModel): # === Services === "list_services": CommandGroup( commands={ - "default": CommandSpec(args=("systemctl", "list-units", "--type=service", "--all", "--no-pager")), + "default": CommandSpec( + args=("systemctl", "list-units", "--type=service", "--all", "--no-pager", "--output=json") + ), } ), "running_services": CommandGroup( @@ -108,7 +110,9 @@ class CommandGroup(BaseModel): ), "service_logs": CommandGroup( commands={ - "default": CommandSpec(args=("journalctl", "-u", "{service_name}", "-n", "{lines}", "--no-pager")), + "default": CommandSpec( + args=("journalctl", "-u", "{service_name}", "-n", "{lines}", "--no-pager"), + ), } ), # === Network === diff --git a/src/linux_mcp_server/formatters.py b/src/linux_mcp_server/formatters.py index 2cf49090..f330b253 100644 --- a/src/linux_mcp_server/formatters.py +++ b/src/linux_mcp_server/formatters.py @@ -189,21 +189,6 @@ def format_services_list(stdout: str, running_count: int | None = None) -> str: return "\n".join(lines) -def format_service_status(stdout: str, service_name: str) -> str: - """Format service status output. - - Args: - stdout: Raw output from systemctl status. - service_name: Name of the service. - - Returns: - Formatted string representation. - """ - lines = [f"=== Status of {service_name} ===\n"] - lines.append(stdout) - return "\n".join(lines) - - def format_service_logs(stdout: str, service_name: str, lines_count: int) -> str: """Format service logs output. diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index 50b4d11a..ae9560d9 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -1,16 +1,15 @@ """Service management tools.""" +import json import typing as t +from fastmcp.exceptions import ToolError from mcp.types import ToolAnnotations from pydantic import Field from linux_mcp_server.audit import log_tool_call from linux_mcp_server.commands import get_command -from linux_mcp_server.formatters import format_service_logs -from linux_mcp_server.formatters import format_service_status -from linux_mcp_server.formatters import format_services_list -from linux_mcp_server.parsers import parse_service_count +from linux_mcp_server.models import LogEntries from linux_mcp_server.server import mcp from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers from linux_mcp_server.utils.types import Host @@ -27,27 +26,25 @@ @disallow_local_execution_in_containers async def list_services( host: Host = None, -) -> str: +) -> list[dict[str, str]]: """List all systemd services. Retrieves all systemd service units with their load state, active state, - sub-state, and description. Also includes a count of currently running services. + sub-state, and description. + + Returns: + list[dict[str, str]]: A list of dictionaries containing information about each service. + + Raises: + ToolError: If an error occurs while listing services. """ cmd = get_command("list_services") returncode, stdout, stderr = await cmd.run(host=host) if returncode != 0: - return f"Error listing services: {stderr}" + raise ToolError(f"Error listing services: {stderr}") - # Get running services count - running_cmd = get_command("running_services") - returncode_summary, stdout_summary, _ = await running_cmd.run(host=host) - - running_count = None - if returncode_summary == 0: - running_count = parse_service_count(stdout_summary) - - return format_services_list(stdout, running_count) + return t.cast(list[dict[str, str]], json.loads(stdout)) @mcp.tool( @@ -71,23 +68,34 @@ async def get_service_status( """Get status of a specific systemd service. Retrieves detailed service information including active/enabled state, - main PID, memory usage, CPU time, and recent log entries from the journal. + main PID, memory usage, etc. + + Returns: + A string containing the output of the systemctl status command. + + Raises: + ToolError: If there was an error getting the service status. """ # Ensure service name has .service suffix if not present if not service_name.endswith(".service") and "." not in service_name: service_name = f"{service_name}.service" cmd = get_command("service_status") - _, stdout, stderr = await cmd.run(host=host, service_name=service_name) + returncode, stdout, stderr = await cmd.run(host=host, service_name=service_name) + + # systemctl status uses exit status 4 when the unit does not exist (see systemctl(1)). + # This is stable across locales; stderr text alone may be translated. + if returncode == 4: + raise ToolError(f"Service '{service_name}' not found on this system.") # Note: systemctl status returns non-zero for inactive services, but that's expected if not stdout and stderr: # Service not found if "not found" in stderr.lower() or "could not be found" in stderr.lower(): - return f"Service '{service_name}' not found on this system." - return f"Error getting service status: {stderr}" + raise ToolError(f"Service '{service_name}' not found on this system.") + raise ToolError(f"Error getting service status: {stderr}") - return format_service_status(stdout, service_name) + return stdout @mcp.tool( @@ -108,11 +116,17 @@ async def get_service_logs( ], lines: t.Annotated[int, Field(description="Number of log lines to retrieve.", ge=1, le=10_000)] = 50, host: Host = None, -) -> str: +) -> LogEntries: """Get recent logs for a specific systemd service. - Retrieves journal entries for the specified service unit, including - timestamps, priority levels, and log messages. + Retrieves journal lines for the specified service unit (default journalctl + short format, same style as get_journal_logs). + + Raises: + ToolError: If journal output is empty, or journalctl fails. + + Returns: + LogEntries: Structured log lines, unit name, and line count. """ # Ensure service name has .service suffix if not present if not service_name.endswith(".service") and "." not in service_name: @@ -122,11 +136,11 @@ async def get_service_logs( returncode, stdout, stderr = await cmd.run(host=host, service_name=service_name, lines=lines) if returncode != 0: - if "not found" in stderr.lower() or "no entries" in stderr.lower(): - return f"No logs found for service '{service_name}'. The service may not exist or has no log entries." - return f"Error getting service logs: {stderr}" + raise ToolError(f"Error getting service logs: {stderr}") if is_empty_output(stdout): - return f"No log entries found for service '{service_name}'." + raise ToolError(f"No log entries found for service '{service_name}'.") + + entries = [line for line in stdout.strip().splitlines() if line] - return format_service_logs(stdout, service_name, lines) + return LogEntries(entries=entries, unit=service_name) diff --git a/tests/test_formatters.py b/tests/test_formatters.py index 37de51b0..f6f3a996 100644 --- a/tests/test_formatters.py +++ b/tests/test_formatters.py @@ -8,7 +8,6 @@ from linux_mcp_server.formatters import format_process_detail from linux_mcp_server.formatters import format_process_list from linux_mcp_server.formatters import format_service_logs -from linux_mcp_server.formatters import format_service_status from linux_mcp_server.formatters import format_services_list from linux_mcp_server.models import ListeningPort from linux_mcp_server.models import NetworkConnection @@ -221,17 +220,6 @@ def test_format_with_count(self): assert "Summary: 5 services currently running" in result -class TestFormatServiceStatus: - """Tests for format_service_status function.""" - - def test_format_service_status(self): - """Test formatting service status.""" - stdout = "● ssh.service - OpenBSD Secure Shell server\n Active: active (running)" - result = format_service_status(stdout, "ssh.service") - assert "=== Status of ssh.service ===" in result - assert "Active: active (running)" in result - - class TestFormatServiceLogs: """Tests for format_service_logs function.""" diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index f9d49710..c93a9a0e 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -4,6 +4,8 @@ import pytest +from fastmcp.exceptions import ToolError + @pytest.mark.skipif(sys.platform != "linux", reason="Only passes on Linux") class TestServices: @@ -17,54 +19,72 @@ async def test_list_services(self, mcp_client): assert all(any(n in result_text for n in case) for case in expected), "Did not find all expected values" - @pytest.mark.parametrize( - "service_name, expected", - ( - ("sshd.service", ("active", "inactive", "loaded", "not found")), - ("nonexistent-service-xyz123", ("not found", "could not", "error")), - ), - ) - async def test_get_service_status(self, mcp_client, service_name, expected): - result = await mcp_client.call_tool("get_service_status", arguments={"service_name": service_name}) + async def test_get_service_status(self, mock_execute_with_fallback, mcp_client): + """Test getting service status with mocked systemctl output (no real sshd unit required).""" + mock_output = "● sshd.service - OpenSSH server\n Loaded: loaded\n Active: active (running)" + mock_execute_with_fallback.return_value = (0, mock_output, "") + + result = await mcp_client.call_tool("get_service_status", arguments={"service_name": "sshd.service"}) result_text = result.content[0].text.casefold() - assert any(n in result_text for n in expected), "Did not find any expected values" + assert "sshd.service" in result_text + assert "loaded" in result_text + assert "active" in result_text + mock_execute_with_fallback.assert_called() async def test_get_service_status_with_nonexistent_service(self, mcp_client): - result = await mcp_client.call_tool( - "get_service_status", arguments={"service_name": "nonexistent-service-xyz123"} - ) - result_text = result.content[0].text.casefold() - expected = ( - "not found", - "could not", - "error", + with pytest.raises(ToolError, match="Service 'nonexistent-service-xyz123.service' not found on this system."): + await mcp_client.call_tool("get_service_status", arguments={"service_name": "nonexistent-service-xyz123"}) + + async def test_get_service_status_error(self, mock_execute_with_fallback, mcp_client): + """Test that get_service_status raises ToolError when systemctl fails.""" + mock_execute_with_fallback.return_value = (1, "", "Failed to get unit file state: Connection refused") + + with pytest.raises(ToolError, match="Error getting service status: Failed to get unit file state"): + await mcp_client.call_tool("get_service_status", arguments={"service_name": "sshd"}) + + async def test_get_service_logs(self, mock_execute_with_fallback, mcp_client): + """Test getting service logs with mocked output.""" + mock_output = ( + "Jan 01 12:00:00 host sshd[1234]: session opened for user test\n" + "Jan 01 12:00:01 host sshd[1234]: session closed for user test" ) - assert any(n in result_text for n in expected), "Did not find any expected values" + mock_execute_with_fallback.return_value = (0, mock_output, "") - async def test_get_service_logs(self, mcp_client): result = await mcp_client.call_tool("get_service_logs", arguments={"service_name": "sshd.service", "lines": 5}) - # Filter out empty lines, header lines (=), and journalctl boot markers (--) - result_lines = [ - line - for line in result.content[0].text.split("\n") - if line and not line.startswith("=") and not line.startswith("--") - ] - assert len(result_lines) <= 5, "Got more lines than expected" + assert result.structured_content is not None, "Expected structured data" + content = result.structured_content + assert content["lines_count"] == 2 + assert len(content["entries"]) == 2 + assert "session opened for user test" in content["entries"][0] + assert content["unit"] == "sshd.service" + mock_execute_with_fallback.assert_called() - async def test_get_service_logs_with_nonexistent_service(self, mcp_client): - result = await mcp_client.call_tool( - "get_service_logs", arguments={"service_name": "nonexistent-service-xyz123", "lines": 10} - ) - result_text = result.content[0].text.casefold() - expected = ( - "not found", - "no entries", - "error", - ) + async def test_get_service_logs_with_nonexistent_service(self, mock_execute_with_fallback, mcp_client): + """Empty journal (e.g. unknown unit with no lines) raises; real journalctl output is not deterministic.""" + mock_execute_with_fallback.return_value = (0, "", "") + + with pytest.raises(ToolError, match="No log entries found for service 'nonexistent-service-xyz123.service'."): + await mcp_client.call_tool( + "get_service_logs", arguments={"service_name": "nonexistent-service-xyz123", "lines": 10} + ) + + mock_execute_with_fallback.assert_called() - assert any(n in result_text for n in expected), "Did not find any expected values" + async def test_get_service_logs_error(self, mock_execute_with_fallback, mcp_client): + """Test that get_service_logs raises ToolError when journalctl fails.""" + mock_execute_with_fallback.return_value = (1, "", "Failed to access journal: Permission denied") + + with pytest.raises(ToolError, match="Error getting service logs: Failed to access journal"): + await mcp_client.call_tool("get_service_logs", arguments={"service_name": "sshd", "lines": 10}) + + async def test_list_services_error(self, mock_execute_with_fallback, mcp_client): + """Test that list_services raises ToolError when systemctl fails.""" + mock_execute_with_fallback.return_value = (1, "", "Failed to connect to bus: No such file or directory") + + with pytest.raises(ToolError, match="Error listing services: Failed to connect to bus"): + await mcp_client.call_tool("list_services") class TestRemoteServices: @@ -72,14 +92,19 @@ class TestRemoteServices: async def test_list_services_remote(self, mock_execute_with_fallback, mcp_client): """Test listing services on a remote host.""" - mock_output = "UNIT LOAD ACTIVE SUB DESCRIPTION\nnginx.service loaded active running Nginx server\n" + mock_output = '[{"unit":"nginx.service","load":"loaded","active":"active","sub":"running","description":"Nginx HTTP Server"}]' mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool("list_services", arguments={"host": "remote.example.com"}) - result_text = result.content[0].text.casefold() - assert "nginx.service" in result_text - assert "system services" in result_text + assert result.structured_content is not None + services = result.structured_content["result"] + assert isinstance(services, list) + assert services[0]["unit"] == "nginx.service" + assert services[0]["load"] == "loaded" + assert services[0]["active"] == "active" + assert services[0]["sub"] == "running" + assert services[0]["description"] == "Nginx HTTP Server" mock_execute_with_fallback.assert_called() async def test_get_service_status_remote(self, mock_execute_with_fallback, mcp_client): @@ -98,14 +123,19 @@ async def test_get_service_status_remote(self, mock_execute_with_fallback, mcp_c async def test_get_service_logs_remote(self, mock_execute_with_fallback, mcp_client): """Test getting service logs on a remote host.""" - mock_output = "Jan 01 12:00:00 host nginx[1234]: Starting Nginx\nJan 01 12:00:01 host nginx[1234]: Started" + mock_output = "Jan 01 12:00:00 remote nginx[1]: Starting Nginx\nJan 01 12:00:01 remote nginx[1]: Started" mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool( "get_service_logs", arguments={"service_name": "nginx", "host": "remote.example.com", "lines": 50} ) - result_text = result.content[0].text.casefold() - assert "nginx" in result_text - assert "starting" in result_text + assert result.structured_content is not None + content = result.structured_content + + assert content["lines_count"] == 2 + assert len(content["entries"]) == 2 + assert "Starting Nginx" in content["entries"][0] + assert "Started" in content["entries"][1] + assert content["unit"] == "nginx.service" mock_execute_with_fallback.assert_called()