stop_container

Instructions

Implementation Reference

• src/ctfd_mcp/ctfd_client.py:581-627 (handler)

  Main stop_container handler: dispatches to stop_dynamic_container, stop_k8s_container, or stop_owl_container based on challenge type. Requires container_id for dynamic_docker, challenge_id for k8s or ctfd-owl.

  async def stop_container(
      self, *, container_id: int | None = None, challenge_id: int | None = None
  ) -> dict[str, Any]:
      """
      Unified stop:
      - dynamic_docker: requires container_id
      - dynamic_check_docker (owl): uses challenge_id
      - k8s-backed: uses challenge_id
      If no challenge_id is given, assumes dynamic_docker and stops by container_id.
      """
      if container_id is None and challenge_id is None:
          raise CTFdClientError(
              "Provide container_id or challenge_id to stop a container."
          )
  
      ctype: str | None = None
      if challenge_id is not None:
          details = await self.get_challenge(challenge_id)
          ctype = (details.get("type") or "").lower()
  
      if self._is_k8s_type(ctype):
          if challenge_id is None:
              raise CTFdClientError("k8s stop requires challenge_id.")
          return await self.stop_k8s_container(challenge_id)
      if ctype == "dynamic_docker":
          if container_id is None:
              raise CTFdClientError("dynamic_docker stop requires container_id.")
          try:
              return await self.stop_dynamic_container(container_id)
          except NotFoundError:
              if challenge_id is not None:
                  return await self.stop_k8s_container(challenge_id)
              raise
      if ctype == "dynamic_check_docker":
          if challenge_id is None:
              raise CTFdClientError("ctfd-owl stop requires challenge_id.")
          return await self.stop_owl_container(challenge_id)
  
      if ctype:
          raise CTFdClientError(
              f"Unsupported challenge type '{ctype}' for container stop."
          )
  
      # No challenge_id was provided; default to whale-style stop by container_id.
      if container_id is None:
          raise CTFdClientError("dynamic_docker stop requires container_id.")
      return await self.stop_dynamic_container(container_id)

• src/ctfd_mcp/ctfd_client.py:260-276 (handler)

  Helper: stop_dynamic_container - sends DELETE /api/v1/containers/{container_id}

  async def stop_dynamic_container(self, container_id: int) -> dict[str, Any]:
      """Stop a dynamic_docker instance."""
      payload = await self._request(
          "DELETE",
          f"/api/v1/containers/{container_id}",
      )
      data = payload.get("data") or payload or {}
      message = data.get("message") or data or "stopped"
      if isinstance(message, dict):
          message = message.get("message") or str(message)
      return self._trim_none(
          {
              "status": data.get("status") or payload.get("success"),
              "message": message,
              "raw": data,
          }
      )

• src/ctfd_mcp/ctfd_client.py:294-309 (handler)

  Helper: stop_k8s_container - sends POST /api/v1/k8s/delete with CSRF nonce

  async def stop_k8s_container(self, challenge_id: int) -> dict[str, Any]:
      """
      Stop a Kubernetes-backed instance exposed via /api/v1/k8s endpoints.
      """
      files = {"challenge_id": (None, str(challenge_id))}
      if self._csrf_token:
          files["nonce"] = (None, self._csrf_token)
  
      await self._k8s_request(
          "POST", "/api/v1/k8s/delete", files=files, action="stop"
      )
      status = await self._get_k8s_container(challenge_id)
      status["status"] = (
          "stopped" if not status.get("instance_running") else "running"
      )
      return self._trim_none(status)

• src/ctfd_mcp/ctfd_client.py:534-555 (handler)

  Helper: stop_owl_container - sends DELETE /plugins/ctfd-owl/container with challenge_id

  async def stop_owl_container(self, challenge_id: int) -> dict[str, Any]:
      """
      Stop a dynamic_check_docker (ctfd-owl) instance.
      Uses the same endpoint with DELETE and challenge_id.
      """
      headers = self._owl_headers()
      payload = await self._request(
          "DELETE",
          "/plugins/ctfd-owl/container",
          params={"challenge_id": challenge_id},
          headers=headers,
      )
      data = payload.get("data") or payload
      message = data.get("message") or data or "stopped"
      if isinstance(message, dict):
          message = message.get("message") or str(message)
      return self._trim_none(
          {
              "status": data.get("status") or payload.get("success"),
              "message": message,
              "raw": data,
          }

• src/ctfd_mcp/server.py:113-125 (registration)

  MCP tool registration: @mcp.tool decorator registers stop_container as an MCP tool with description

  @mcp.tool(
      description="Unified stop: whale requires container_id; ctfd-owl/k8s require challenge_id."
  )
  async def stop_container(
      container_id: int | None = None, challenge_id: int | None = None
  ):
      client = await _get_client()
      try:
          return await client.stop_container(
              container_id=container_id, challenge_id=challenge_id
          )
      except Exception as exc:  # noqa: BLE001
          raise _format_error(exc)