restart_service

Instructions

Implementation Reference

• src/tools/services.py:12-61 (handler)

  The restart_service function that executes the tool logic. It performs a safety check, gets service status, calls Webmin's init module restart_action, checks post-restart status, and returns a ToolResult.

  async def restart_service(
      client: WebminClient,
      service: str,
      safe_mode: bool = True,
  ) -> ToolResult:
      """Restart a system service.
  
      Args:
          client: Authenticated WebminClient instance.
          service: Name of the service to restart.
          safe_mode: Whether safe mode is enabled.
  
      Returns:
          ToolResult with restart status.
      """
      # Safety check
      blocked = safety_check_or_fail("restart", service, safe_mode)
      if blocked:
          return blocked
  
      try:
          # Get current status
          pre_status = await client.call("init", "status_action", service)
  
          # Restart the service
          result = await client.call("init", "restart_action", service)
  
          # Get new status
          post_status = await client.call("init", "status_action", service)
  
          # Parse result - typically [exit_code, output]
          exit_code = result[0] if isinstance(result, list) else result
          output = result[1] if isinstance(result, list) and len(result) > 1 else ""
  
          return ToolResult.ok({
              "service": service,
              "action": "restart",
              "success": post_status == 1,  # 1 = running
              "running": post_status == 1,
              "exit_code": exit_code,
              "output": output,
              "status_before": "running" if pre_status == 1 else "stopped",
              "status_after": "running" if post_status == 1 else "stopped",
          })
  
      except Exception as e:
          return ToolResult.fail(
              code="RESTART_SERVICE_ERROR",
              message=f"Failed to restart service '{service}': {e}",
          )

• src/server.py:225-243 (schema)

  Tool registration/schema definition for 'restart_service'. Defines the tool name, description, and input schema (requires 'service' parameter, with optional 'server').

  Tool(
      name="restart_service",
      description=(
          "Restart a system service. The service will be stopped and then "
          "started again. Some critical services (ssh, webmin) may be blocked "
          "in safe mode."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              **SERVER_PARAM,
              "service": {
                  "type": "string",
                  "description": "Name of the service to restart (e.g., 'nginx', 'cron')",
              },
          },
          "required": ["service"],
      },
  ),

• src/server.py:1391-1398 (registration)

  Dispatch/handler routing for 'restart_service' in call_tool(). Extracts the 'service' argument from the request and delegates to services.restart_service(), passing the safe_mode config setting.

  if name == "restart_service":
      service = arguments.get("service")
      if not service:
          return ToolResult.fail(
              code="MISSING_ARGUMENT",
              message="Missing required argument: service",
          )
      return await services.restart_service(client, service, config.safe_mode)

• src/safety.py:68-110 (helper)

  Safety check helper used by restart_service. The check_service_operation() function determines whether a service can be restarted - blocking critical services (ssh, webmin, etc.) in safe mode.

  def check_service_operation(
      service: str,
      operation: str,
      safe_mode: bool = True,
  ) -> SafetyCheckResult:
      """Check if a service operation is allowed.
  
      Args:
          service: Name of the service.
          operation: Type of operation (start, stop, restart, enable, disable).
          safe_mode: Whether safe mode is enabled.
  
      Returns:
          SafetyCheckResult indicating if the operation is allowed.
      """
      service_lower = service.lower()
  
      # Check if service is in blocked list
      if service_lower in BLOCKED_SERVICES:
          if operation in ("stop", "disable"):
              return SafetyCheckResult(
                  allowed=False,
                  reason=f"Cannot {operation} critical service '{service}'. "
                  f"This service is required for system operation or remote access.",
              )
          if operation == "restart" and safe_mode:
              return SafetyCheckResult(
                  allowed=False,
                  reason=f"Cannot restart critical service '{service}' in safe mode. "
                  f"Disable safe mode (WEBMIN_SAFE_MODE=false) to allow this operation.",
              )
  
      # Check restart-only services
      if service_lower in RESTART_ONLY_SERVICES:
          if operation == "stop" and safe_mode:
              return SafetyCheckResult(
                  allowed=False,
                  reason=f"Cannot stop service '{service}' in safe mode. "
                  f"This service provides important functionality. "
                  f"Use restart instead, or disable safe mode.",
              )
  
      return SafetyCheckResult(allowed=True)

• tests/test_tools_services.py:17-69 (helper)

  Test suite for restart_service covering success, blocked (sshd in safe mode), allowed without safe mode, and error scenarios.

  class TestRestartService:
      """Tests for restart_service tool."""
  
      async def test_restart_service_success(self, mock_client: AsyncMock) -> None:
          """Test successful service restart."""
          # Setup: service is running before and after
          mock_client.call.side_effect = [
              1,  # status_action (running)
              [0, ""],  # restart_action
              1,  # status_action (running)
          ]
  
          result = await services.restart_service(mock_client, "nginx", safe_mode=True)
  
          assert result.success
          assert result.data["service"] == "nginx"
          assert result.data["action"] == "restart"
          assert result.data["running"] is True
          assert result.data["status_before"] == "running"
          assert result.data["status_after"] == "running"
  
      async def test_restart_service_blocked_for_ssh(self, mock_client: AsyncMock) -> None:
          """Test that restarting sshd is blocked in safe mode."""
          result = await services.restart_service(mock_client, "sshd", safe_mode=True)
  
          assert not result.success
          assert result.error.code == "SAFETY_BLOCKED"
          # Ensure no API calls were made
          mock_client.call.assert_not_called()
  
      async def test_restart_service_allowed_without_safe_mode(
          self, mock_client: AsyncMock
      ) -> None:
          """Test that restarting sshd is allowed without safe mode."""
          mock_client.call.side_effect = [
              1,  # status_action (running)
              [0, ""],  # restart_action
              1,  # status_action (running)
          ]
  
          result = await services.restart_service(mock_client, "sshd", safe_mode=False)
  
          assert result.success
          assert result.data["service"] == "sshd"
  
      async def test_restart_service_error(self, mock_client: AsyncMock) -> None:
          """Test error handling during restart."""
          mock_client.call.side_effect = Exception("Connection failed")
  
          result = await services.restart_service(mock_client, "nginx", safe_mode=True)
  
          assert not result.success
          assert result.error.code == "RESTART_SERVICE_ERROR"