From d89ebcaf64c7b3b9885630f219af6454eef19eab Mon Sep 17 00:00:00 2001 From: umbra2728 Date: Wed, 4 Feb 2026 19:34:44 +0300 Subject: [PATCH] test: migrate to pytest Replace unittest-based coverage with pytest and update CI/docs to run the new suite. --- .github/workflows/publish.yml | 4 +- .gitignore | 3 + README.md | 5 +- pyproject.toml | 5 + tests/test_config.py | 174 ++++++-- tests/test_ctfd_client.py | 812 +++++++++++++++++++++------------- uv.lock | 54 +++ 7 files changed, 705 insertions(+), 352 deletions(-) diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 0b9782f..0c33f82 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -32,7 +32,7 @@ jobs: run: uv sync --frozen --no-editable - name: Run tests - run: uv run python -m unittest discover -s tests + run: uv run pytest - name: Build distributions run: uv build --no-sources @@ -63,4 +63,4 @@ jobs: - name: Publish to PyPI uses: pypa/gh-action-pypi-publish@release/v1 with: - packages_dir: dist + packages-dir: dist diff --git a/.gitignore b/.gitignore index 6b9fee8..c3d372d 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,6 @@ htmlcov/ # OS / editor .DS_Store + +# Claude +.claude/ diff --git a/README.md b/README.md index a2cfd11..38e2115 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,6 @@ # CTFd MCP server (user scope) +![](https://badge.mcpx.dev?type=server 'MCP Server') [![GitHub Release](https://img.shields.io/github/v/release/umbra2728/ctfd-mcp?sort=semver)](https://github.com/umbra2728/ctfd-mcp/releases) [![License](https://img.shields.io/github/license/umbra2728/ctfd-mcp)](LICENSE) [![Python](https://img.shields.io/badge/Python-3.13%2B-blue)](https://www.python.org/downloads/) @@ -112,14 +113,14 @@ If something breaks or you have questions, reach out: ## Testing -- Run `uv run python -m tests.test_ctfd_client` (requires a real `CTFD_URL` plus token or username/password) to exercise challenge fetching/submission flows. +- Run `uv run pytest`. - Timeouts are configurable via env: `CTFD_TIMEOUT` (total), `CTFD_CONNECT_TIMEOUT`, `CTFD_READ_TIMEOUT` (seconds). Defaults are 20s total / 10s connect / 15s read. ## Development - Dev dependencies: `uv sync --group dev` - Lint/format: `uv run ruff check .` and `uv run ruff format .` -- Tests: `uv run python -m unittest discover -s tests` +- Tests: `uv run pytest` - Pre-commit: `uv run pre-commit install` (see `CONTRIBUTING.md`) ## License diff --git a/pyproject.toml b/pyproject.toml index 4522713..12a3968 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,9 +27,14 @@ ctfd-mcp = "ctfd_mcp.main:main" [dependency-groups] dev = [ "pre-commit>=3.6.0", + "pytest>=8.0.0", "ruff>=0.5.0", ] +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src", "."] + [tool.ruff] line-length = 88 target-version = "py313" diff --git a/tests/test_config.py b/tests/test_config.py index adeecba..4362425 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,69 +1,147 @@ -"""Config loading tests (ruff: ignore E402 for sys.path adjustment).""" +import pytest -# ruff: noqa: E402 - -import os -import sys -import unittest -from pathlib import Path -from unittest.mock import patch - -ROOT = Path(__file__).resolve().parents[1] -SRC = ROOT / "src" -for path in (SRC, ROOT): - if str(path) not in sys.path: - sys.path.insert(0, str(path)) - -from ctfd_mcp.config import load_config +from ctfd_mcp.config import DEFAULT_USER_AGENT, ConfigError, _parse_timeout, load_config -def _load_with_env(env: dict[str, str]): - """Load config with isolated env and no .env side effects.""" - with patch("ctfd_mcp.config.load_dotenv", return_value=None): - with patch.dict(os.environ, env, clear=True): - return load_config() +def _load_with_env(monkeypatch: pytest.MonkeyPatch, env: dict[str, str]): + monkeypatch.setattr("ctfd_mcp.config.load_dotenv", lambda: None) + for key in ( + "CTFD_URL", + "CTFD_TOKEN", + "CTFD_SESSION", + "CTFD_CSRF_TOKEN", + "CTFD_USERNAME", + "CTFD_PASSWORD", + "CTFD_TIMEOUT", + "CTFD_CONNECT_TIMEOUT", + "CTFD_READ_TIMEOUT", + "CTFD_USER_AGENT", + ): + monkeypatch.delenv(key, raising=False) + for key, value in env.items(): + monkeypatch.setenv(key, value) + return load_config() -class ConfigPrecedenceTests(unittest.TestCase): - def test_username_password_take_priority(self): - env = { +def test_username_password_take_priority(monkeypatch: pytest.MonkeyPatch): + cfg = _load_with_env( + monkeypatch, + { "CTFD_URL": "https://ctfd.example.com", "CTFD_USERNAME": "user1", "CTFD_PASSWORD": "pw1", "CTFD_TOKEN": "token-should-be-ignored", "CTFD_SESSION": "session-should-be-ignored", "CTFD_CSRF_TOKEN": "csrf-should-be-ignored", - } - cfg = _load_with_env(env) - self.assertEqual(cfg.username, "user1") - self.assertEqual(cfg.password, "pw1") - self.assertIsNone(cfg.token) - self.assertIsNone(cfg.session_cookie) - self.assertIsNone(cfg.csrf_token) + }, + ) + assert cfg.username == "user1" + assert cfg.password == "pw1" + assert cfg.token is None + assert cfg.session_cookie is None + assert cfg.csrf_token is None - def test_token_beats_session_cookie(self): - env = { + +def test_token_beats_session_cookie(monkeypatch: pytest.MonkeyPatch): + cfg = _load_with_env( + monkeypatch, + { "CTFD_URL": "https://ctfd.example.com", "CTFD_TOKEN": "use-this-token", "CTFD_SESSION": "drop-this-session", "CTFD_CSRF_TOKEN": "csrf-should-be-ignored", - } - cfg = _load_with_env(env) - self.assertEqual(cfg.token, "use-this-token") - self.assertIsNone(cfg.session_cookie) - self.assertIsNone(cfg.csrf_token) + }, + ) + assert cfg.token == "use-this-token" + assert cfg.session_cookie is None + assert cfg.csrf_token is None - def test_session_cookie_when_no_other_creds(self): - env = { + +def test_session_cookie_when_no_other_creds(monkeypatch: pytest.MonkeyPatch): + cfg = _load_with_env( + monkeypatch, + { "CTFD_URL": "https://ctfd.example.com", "CTFD_SESSION": "session-only", - } - cfg = _load_with_env(env) - self.assertEqual(cfg.session_cookie, "session-only") - self.assertIsNone(cfg.token) - self.assertIsNone(cfg.username) - self.assertIsNone(cfg.password) + }, + ) + assert cfg.session_cookie == "session-only" + assert cfg.token is None + assert cfg.username is None + assert cfg.password is None -if __name__ == "__main__": - unittest.main() +def test_missing_base_url(monkeypatch: pytest.MonkeyPatch): + with pytest.raises(ConfigError, match="CTFD_URL is not set"): + _load_with_env( + monkeypatch, + { + "CTFD_TOKEN": "token", + }, + ) + + +def test_missing_creds(monkeypatch: pytest.MonkeyPatch): + with pytest.raises(ConfigError, match="Set CTFD_TOKEN"): + _load_with_env( + monkeypatch, + { + "CTFD_URL": "https://ctfd.example.com", + }, + ) + + +def test_invalid_url(monkeypatch: pytest.MonkeyPatch): + with pytest.raises(ConfigError, match="CTFD_URL must be a full URL"): + _load_with_env( + monkeypatch, + { + "CTFD_URL": "ctfd.example.com", + "CTFD_TOKEN": "token", + }, + ) + + +def test_parse_timeout_valid(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("CTFD_TIMEOUT", "12.5") + assert _parse_timeout("CTFD_TIMEOUT") == 12.5 + + +def test_parse_timeout_invalid(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("CTFD_TIMEOUT", "nope") + with pytest.raises(ConfigError, match="CTFD_TIMEOUT must be a number"): + _parse_timeout("CTFD_TIMEOUT") + + +def test_auth_header(monkeypatch: pytest.MonkeyPatch): + cfg = _load_with_env( + monkeypatch, + { + "CTFD_URL": "https://ctfd.example.com", + "CTFD_TOKEN": "token", + }, + ) + assert cfg.auth_header == {"Authorization": "Token token"} + + +def test_user_agent_default(monkeypatch: pytest.MonkeyPatch): + cfg = _load_with_env( + monkeypatch, + { + "CTFD_URL": "https://ctfd.example.com", + "CTFD_TOKEN": "token", + }, + ) + assert cfg.user_agent == DEFAULT_USER_AGENT + + +def test_user_agent_override(monkeypatch: pytest.MonkeyPatch): + cfg = _load_with_env( + monkeypatch, + { + "CTFD_URL": "https://ctfd.example.com", + "CTFD_TOKEN": "token", + "CTFD_USER_AGENT": " custom-agent/1.0 ", + }, + ) + assert cfg.user_agent == "custom-agent/1.0" diff --git a/tests/test_ctfd_client.py b/tests/test_ctfd_client.py index 54854f8..dbd31d5 100644 --- a/tests/test_ctfd_client.py +++ b/tests/test_ctfd_client.py @@ -1,332 +1,544 @@ -"""CTFd client tests (ruff: ignore E402 for sys.path adjustment).""" - -# ruff: noqa: E402 - import asyncio import json -import os -import sys -import types -import unittest -from pathlib import Path - import httpx +import pytest -ROOT = Path(__file__).resolve().parents[1] -SRC = ROOT / "src" -for path in (SRC, ROOT): - if str(path) not in sys.path: - sys.path.insert(0, str(path)) - -# Avoid optional dependency issues in the test runner (python-dotenv). -if "dotenv" not in sys.modules: - sys.modules["dotenv"] = types.SimpleNamespace(load_dotenv=lambda *_, **__: None) - -from ctfd_mcp.config import Config # type: ignore -from ctfd_mcp.ctfd_client import AuthError, CTFdClient, CTFdClientError # type: ignore - -CTFD_URL = os.getenv("CTFD_URL") -CTFD_USERNAME = os.getenv("CTFD_USERNAME") -CTFD_PASSWORD = os.getenv("CTFD_PASSWORD") -CTFD_TOKEN = os.getenv("CTFD_TOKEN") - -CTFD_LIVE = os.getenv("CTFD_LIVE", "") -CTFD_LIVE_CHALLENGE_ID = os.getenv("CTFD_LIVE_CHALLENGE_ID") -CTFD_LIVE_FLAG = os.getenv("CTFD_LIVE_FLAG") +from ctfd_mcp.config import Config, ConfigError +from ctfd_mcp.ctfd_client import ( + AuthError, + CTFdClient, + CTFdClientError, + NotFoundError, + RateLimitError, + _html_to_text, +) +from ctfd_mcp.server import ( + _challenge_markdown, + _format_error, + challenge_details, + list_challenges, + start_container, + stop_container, + submit_flag, +) -def _has_creds() -> bool: - # Accept either token or username/password for real CTFd instance. - return bool(CTFD_URL and (CTFD_TOKEN or (CTFD_USERNAME and CTFD_PASSWORD))) - -def _truthy(value: str | None) -> bool: - if not value: - return False - return value.strip().lower() in {"1", "true", "yes", "y", "on"} +def run(coro): + return asyncio.run(coro) -class TestCTFdClientLive(unittest.TestCase): - """Lightweight live tests against the provided CTFd instance.""" +@pytest.fixture +def base_config(): + return Config(base_url="https://ctfd.example.com", token="test-token") - @unittest.skipUnless( - _truthy(CTFD_LIVE) and _has_creds(), - "Live tests disabled (set CTFD_LIVE=1 plus CTFD_URL/credentials to enable)", + +@pytest.fixture +def make_client(base_config): + def _make(transport: httpx.MockTransport, config: Config | None = None): + cfg = config or base_config + client = CTFdClient(cfg, timeout=None) + run(client._client.aclose()) + client._client = httpx.AsyncClient( + base_url=cfg.base_url, + headers=client._client.headers, + cookies=client._client.cookies, + transport=transport, + follow_redirects=True, + http2=False, + ) + return client + + return _make + + +def test_user_agent_header_custom(make_client): + calls: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(request.url.path) + assert request.headers.get("User-Agent") == "custom-agent/1.0" + if request.url.path == "/api/v1/challenges/123": + return httpx.Response( + 200, + json={ + "success": True, + "data": { + "id": 123, + "name": "Example", + "category": "misc", + "value": 100, + "description": "

hi

", + "type": "standard", + "tags": [], + "files": [], + }, + }, + ) + return httpx.Response(404, json={"success": False, "message": "not found"}) + + transport = httpx.MockTransport(handler) + cfg = Config( + base_url="https://ctfd.example.com", + token="test-token", + user_agent="custom-agent/1.0", ) - def test_list_and_get_challenge(self): - """Ensure we can reach CTFd and fetch challenge details. + client = make_client(transport, config=cfg) + try: + detail = run(client.get_challenge(123)) + assert detail.get("id") == 123 + finally: + run(client.aclose()) - This is intentionally parameterized to avoid hard-coding a specific CTFd instance, - challenge id, or flag. For flag submission, provide CTFD_LIVE_CHALLENGE_ID and - CTFD_LIVE_FLAG. - """ - - async def _run(): - cfg = Config( - base_url=CTFD_URL, - token=CTFD_TOKEN, - username=CTFD_USERNAME, - password=CTFD_PASSWORD, - ) - client = CTFdClient(cfg, timeout=None) - try: - challenges = await client.list_challenges() - self.assertIsInstance(challenges, list) - self.assertGreater(len(challenges), 0, "Expected at least one challenge") - - if CTFD_LIVE_CHALLENGE_ID: - challenge_id = int(CTFD_LIVE_CHALLENGE_ID) - else: - # Default to the first challenge returned by the API. - challenge_id = int(challenges[0]["id"]) - - detail = await client.get_challenge(challenge_id) - self.assertEqual(detail.get("id"), challenge_id) - - # Optional end-to-end submission (requires knowing a valid flag). - if CTFD_LIVE_FLAG: - result = await client.submit_flag(challenge_id, CTFD_LIVE_FLAG) - self.assertIsInstance(result, dict) - self.assertIn("status", result) - status = result.get("status") - self.assertTrue( - status is None or isinstance(status, str), - f"Unexpected status type: {type(status)}", - ) - except AuthError as exc: - self.fail(f"Auth should not fail with provided creds: {exc}") - finally: - await client.aclose() - - asyncio.run(_run()) + assert calls == ["/api/v1/challenges/123"] -class TestCTFdClientHelpers(unittest.TestCase): - def test_k8s_type_detection(self): - cfg = Config(base_url="https://ctfd.example.com", token="placeholder") - client = CTFdClient(cfg, timeout=None) - self.assertTrue(client._is_k8s_type("k8s")) - self.assertTrue(client._is_k8s_type("dynamic_kubernetes")) - self.assertFalse(client._is_k8s_type("dynamic_docker")) +def test_token_auth_headers_and_no_csrf(make_client): + calls: list[tuple[str, str]] = [] + def handler(request: httpx.Request) -> httpx.Response: + calls.append((request.method, request.url.path)) + assert request.headers.get("Authorization") == "Token test-token" + assert request.headers.get("CSRF-Token") is None -class TestStopContainerValidation(unittest.TestCase): - def test_dynamic_docker_requires_container_id(self): - cfg = Config(base_url="https://ctfd.example.com", token="placeholder") - client = CTFdClient(cfg, timeout=None) - - async def fake_get_challenge(self, challenge_id: int): - return {"type": "dynamic_docker"} - - client.get_challenge = types.MethodType(fake_get_challenge, client) - - async def _run(): - with self.assertRaises(CTFdClientError): - await client.stop_container(challenge_id=123) - - asyncio.run(_run()) - - -class TestCsrfTokenEnsure(unittest.TestCase): - def test_submit_flag_ensures_csrf_for_session_cookie(self): - calls: list[tuple[str, str, str | None]] = [] - - def handler(request: httpx.Request) -> httpx.Response: - calls.append( - ( - request.method, - request.url.path, - request.headers.get("CSRF-Token"), - ) - ) - if request.url.path == "/api/v1/csrf_token": - return httpx.Response( - 200, json={"success": True, "data": {"csrf_token": "api-token"}} - ) - if request.url.path == "/challenges": - html = '' - return httpx.Response(200, text=html, headers={"Content-Type": "text/html"}) - if request.url.path == "/api/v1/challenges/attempt": - if request.headers.get("CSRF-Token") != "page-nonce": - return httpx.Response(403, json={"success": False, "message": "CSRF"}) - return httpx.Response( - 200, - json={ - "success": True, - "data": {"status": "correct", "message": "ok"}, + if request.url.path == "/api/v1/challenges/123": + return httpx.Response( + 200, + json={ + "success": True, + "data": { + "id": 123, + "name": "Example", + "category": "misc", + "value": 100, + "description": "

hi

", + "type": "standard", + "tags": [], + "files": [], }, - ) - return httpx.Response(404, json={"success": False, "message": "not found"}) + }, + ) - transport = httpx.MockTransport(handler) - cfg = Config( - base_url="https://ctfd.example.com", - session_cookie="session", + if request.url.path == "/api/v1/challenges/attempt": + body = json.loads(request.content.decode("utf-8")) + assert body["challenge_id"] == 123 + assert body["submission"] == "flag{test}" + return httpx.Response( + 200, + json={ + "success": True, + "data": {"status": "correct", "message": "ok"}, + }, + ) + + return httpx.Response(404, json={"success": False, "message": "not found"}) + + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + detail = run(client.get_challenge(123)) + assert detail.get("id") == 123 + result = run(client.submit_flag(123, "flag{test}")) + assert result.get("status") == "correct" + finally: + run(client.aclose()) + + paths = [p for _, p in calls] + assert paths == ["/api/v1/challenges/123", "/api/v1/challenges/attempt"] + + +def test_session_cookie_csrf_refresh_on_403(make_client): + calls: list[tuple[str, str, str | None]] = [] + state = {"refreshed": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls.append( + (request.method, request.url.path, request.headers.get("CSRF-Token")) ) - client = CTFdClient(cfg, timeout=None) - - async def _run(): - await client._client.aclose() - client._client = httpx.AsyncClient( - base_url=cfg.base_url, - headers=client._client.headers, - cookies=client._client.cookies, - transport=transport, - follow_redirects=True, - http2=False, + if request.url.path == "/api/v1/csrf_token": + state["refreshed"] += 1 + return httpx.Response( + 200, + json={ + "success": True, + "data": {"csrf_token": f"api-token-{state['refreshed']}"}, + }, ) - try: - result = await client.submit_flag(1, "flag{test}") - finally: - await client.aclose() - self.assertEqual(result.get("status"), "correct") - - asyncio.run(_run()) - - paths = [p for _, p, _ in calls] - self.assertIn("/api/v1/csrf_token", paths) - self.assertIn("/challenges", paths) - self.assertEqual(paths[-1], "/api/v1/challenges/attempt") - self.assertEqual(calls[-1][2], "page-nonce") - - def test_submit_flag_refreshes_csrf_on_403_for_session_cookie(self): - calls: list[tuple[str, str, str | None]] = [] - state = {"refreshed": 0} - - def handler(request: httpx.Request) -> httpx.Response: - calls.append( - ( - request.method, - request.url.path, - request.headers.get("CSRF-Token"), - ) + if request.url.path == "/challenges": + html = ( + '' # noqa: E501 ) - if request.url.path == "/api/v1/csrf_token": - state["refreshed"] += 1 - return httpx.Response( - 200, - json={ - "success": True, - "data": {"csrf_token": f"api-token-{state['refreshed']}"}, - }, - ) - if request.url.path == "/challenges": - html = ( - '' - ) - return httpx.Response(200, text=html, headers={"Content-Type": "text/html"}) - if request.url.path == "/api/v1/challenges/attempt": - if request.headers.get("CSRF-Token") != "page-nonce-1": - return httpx.Response(403, json={"success": False, "message": "CSRF"}) - return httpx.Response( - 200, - json={ - "success": True, - "data": {"status": "correct", "message": "ok"}, - }, - ) - return httpx.Response(404, json={"success": False, "message": "not found"}) - - transport = httpx.MockTransport(handler) - cfg = Config( - base_url="https://ctfd.example.com", - session_cookie="session", - ) - client = CTFdClient(cfg, timeout=None) - client._csrf_token = "stale" - - async def _run(): - await client._client.aclose() - client._client = httpx.AsyncClient( - base_url=cfg.base_url, - headers=client._client.headers, - cookies=client._client.cookies, - transport=transport, - follow_redirects=True, - http2=False, + return httpx.Response(200, text=html, headers={"Content-Type": "text/html"}) + if request.url.path == "/api/v1/challenges/attempt": + if request.headers.get("CSRF-Token") != "page-nonce-1": + return httpx.Response(403, json={"success": False, "message": "CSRF"}) + return httpx.Response( + 200, + json={ + "success": True, + "data": {"status": "correct", "message": "ok"}, + }, ) - try: - result = await client.submit_flag(1, "flag{test}") - finally: - await client.aclose() - self.assertEqual(result.get("status"), "correct") + return httpx.Response(404, json={"success": False, "message": "not found"}) - asyncio.run(_run()) + transport = httpx.MockTransport(handler) + cfg = Config(base_url="https://ctfd.example.com", session_cookie="session") + client = make_client(transport, config=cfg) + client._csrf_token = "stale" + try: + result = run(client.submit_flag(1, "flag{test}")) + assert result.get("status") == "correct" + finally: + run(client.aclose()) - # Two POSTs: first with stale token, second after refresh. - post_tokens = [t for m, p, t in calls if m == "POST" and p.endswith("/attempt")] - self.assertEqual(post_tokens[0], "stale") - self.assertEqual(post_tokens[-1], "page-nonce-1") + post_tokens = [t for m, p, t in calls if m == "POST" and p.endswith("/attempt")] + assert post_tokens[0] == "stale" + assert post_tokens[-1] == "page-nonce-1" -class TestCTFdClientTokenAuthOffline(unittest.TestCase): - def test_get_and_submit_do_not_require_csrf_for_token_auth(self): - calls: list[tuple[str, str]] = [] +def test_session_cookie_fetches_csrf_token(make_client): + calls: list[str] = [] - def handler(request: httpx.Request) -> httpx.Response: - calls.append((request.method, request.url.path)) - self.assertEqual(request.headers.get("Authorization"), "Token test-token") - self.assertIsNone(request.headers.get("CSRF-Token")) + def handler(request: httpx.Request) -> httpx.Response: + calls.append(request.url.path) + if request.url.path == "/api/v1/csrf_token": + return httpx.Response( + 200, + json={"success": True, "data": {"csrf_token": "api-token"}}, + ) + if request.url.path == "/challenges": + html = '' + return httpx.Response(200, text=html, headers={"Content-Type": "text/html"}) + if request.url.path == "/api/v1/challenges/attempt": + assert request.headers.get("CSRF-Token") == "page-nonce" + return httpx.Response( + 200, + json={ + "success": True, + "data": {"status": "correct", "message": "ok"}, + }, + ) + return httpx.Response(404, json={"success": False, "message": "not found"}) - if request.url.path == "/api/v1/challenges/123": - return httpx.Response( - 200, - json={ - "success": True, - "data": { - "id": 123, - "name": "Example", - "category": "misc", - "value": 100, - "description": "

hi

", - "type": "standard", - "tags": [], - "files": [], + transport = httpx.MockTransport(handler) + cfg = Config(base_url="https://ctfd.example.com", session_cookie="session") + client = make_client(transport, config=cfg) + try: + result = run(client.submit_flag(1, "flag{test}")) + assert result.get("status") == "correct" + finally: + run(client.aclose()) + + assert "/api/v1/csrf_token" in calls + assert "/challenges" in calls + assert calls[-1] == "/api/v1/challenges/attempt" + + +def test_html_to_text_and_nonce_parsing(base_config): + assert _html_to_text("

Hello

World
") == "Hello\nWorld" + client = CTFdClient(base_config, timeout=None) + try: + assert client._extract_nonce('') == "abc" + assert client._extract_nonce("var config = {'csrfNonce': \"def\"};") == "def" + finally: + run(client.aclose()) + + +def test_full_url(base_config): + client = CTFdClient(base_config, timeout=None) + try: + assert client._full_url("https://example.com/x") == "https://example.com/x" + assert client._full_url("/files/x") == "https://ctfd.example.com/files/x" + finally: + run(client.aclose()) + + +def test_list_challenges_filters(make_client): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/challenges": + return httpx.Response( + 200, + json={ + "success": True, + "data": [ + {"id": 1, "name": "A", "category": "web", "solved": True}, + { + "id": 2, + "name": "B", + "category": "pwn", + "solved_by_me": False, }, - }, - ) - - if request.url.path == "/api/v1/challenges/attempt": - body = json.loads(request.content.decode("utf-8")) - self.assertEqual(body["challenge_id"], 123) - self.assertEqual(body["submission"], "flag{test}") - return httpx.Response( - 200, - json={ - "success": True, - "data": {"status": "correct", "message": "ok"}, - }, - ) - - return httpx.Response(404, json={"success": False, "message": "not found"}) - - transport = httpx.MockTransport(handler) - cfg = Config(base_url="https://ctfd.example.com", token="test-token") - client = CTFdClient(cfg, timeout=None) - - async def _run(): - await client._client.aclose() - client._client = httpx.AsyncClient( - base_url=cfg.base_url, - headers=client._client.headers, - cookies=client._client.cookies, - transport=transport, - follow_redirects=True, - http2=False, + ], + }, ) - try: - detail = await client.get_challenge(123) - self.assertEqual(detail.get("id"), 123) - result = await client.submit_flag(123, "flag{test}") - self.assertEqual(result.get("status"), "correct") - finally: - await client.aclose() + return httpx.Response(404, json={"success": False}) - asyncio.run(_run()) - - paths = [p for _, p in calls] - self.assertEqual(paths, ["/api/v1/challenges/123", "/api/v1/challenges/attempt"]) + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + results = run(client.list_challenges(category="pwn", only_unsolved=True)) + assert len(results) == 1 + assert results[0]["id"] == 2 + finally: + run(client.aclose()) -if __name__ == "__main__": - unittest.main() +def test_error_status_mapping(make_client): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/challenges/404": + return httpx.Response(404, json={"success": False}) + if request.url.path == "/api/v1/challenges/401": + return httpx.Response(401, json={"success": False}) + if request.url.path == "/api/v1/challenges/429": + return httpx.Response( + 429, + json={"success": False}, + headers={"Retry-After": "120"}, + ) + return httpx.Response(500, json={"success": False, "message": "boom"}) + + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + with pytest.raises(NotFoundError): + run(client.get_challenge(404)) + with pytest.raises(AuthError): + run(client.get_challenge(401)) + with pytest.raises(RateLimitError) as exc: + run(client.get_challenge(429)) + assert exc.value.retry_after == "120" + with pytest.raises(CTFdClientError, match="CTFd error 500"): + run(client.get_challenge(500)) + finally: + run(client.aclose()) + + +def test_redirect_error(make_client): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(302) + + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + with pytest.raises(AuthError, match="Unexpected redirect"): + run(client.get_challenge(1)) + finally: + run(client.aclose()) + + +def test_non_json_response(make_client): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/challenges/123": + return httpx.Response(200, text="nope", headers={"Content-Type": "text/html"}) + return httpx.Response(404, json={"success": False}) + + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + with pytest.raises(CTFdClientError, match="non-JSON response"): + run(client.get_challenge(123)) + finally: + run(client.aclose()) + + +def test_dynamic_container_flow(make_client): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/challenges/1": + return httpx.Response( + 200, + json={"success": True, "data": {"id": 1, "type": "dynamic_docker"}}, + ) + if request.url.path == "/api/v1/containers": + return httpx.Response( + 200, + json={ + "success": True, + "data": {"id": 2, "challenge_id": 1, "host": "h", "port": 1337}, + }, + ) + if request.url.path == "/api/v1/containers/2": + return httpx.Response( + 200, + json={"success": True, "data": {"status": "stopped", "message": "ok"}}, + ) + return httpx.Response(404, json={"success": False}) + + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + result = run(client.start_container(1)) + assert result["connection_info"] == "h:1337" + stopped = run(client.stop_container(container_id=2, challenge_id=1)) + assert stopped["message"] == "ok" + finally: + run(client.aclose()) + + +def test_owl_container_flow(make_client): + calls: list[tuple[str, str]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append((request.method, request.url.path)) + if request.url.path == "/api/v1/challenges/1": + return httpx.Response( + 200, + json={"success": True, "data": {"id": 1, "type": "dynamic_check_docker"}}, + ) + if request.url.path == "/plugins/ctfd-owl/container" and request.method == "POST": + return httpx.Response(200, json={"success": True, "data": {"success": True}}) + if request.url.path == "/plugins/ctfd-owl/container" and request.method == "GET": + return httpx.Response( + 200, + json={ + "success": True, + "data": { + "containers_data": [ + {"host": "owl", "port": 9000, "container_id": "c1"} + ], + "ip": "1.2.3.4", + }, + }, + ) + if request.url.path == "/plugins/ctfd-owl/container" and request.method == "DELETE": + return httpx.Response( + 200, + json={"success": True, "data": {"status": "stopped", "message": "ok"}}, + ) + return httpx.Response(404, json={"success": False}) + + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + result = run(client.start_container(1)) + assert result["connection_info"] == "1.2.3.4:9000" + stopped = run(client.stop_container(challenge_id=1)) + assert stopped["message"] == "ok" + finally: + run(client.aclose()) + + assert ("POST", "/plugins/ctfd-owl/container") in calls + assert ("GET", "/plugins/ctfd-owl/container") in calls + assert ("DELETE", "/plugins/ctfd-owl/container") in calls + + +def test_k8s_container_flow(make_client): + calls: list[tuple[str, str, dict[str, str]]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/challenges/1": + return httpx.Response( + 200, + json={"success": True, "data": {"id": 1, "type": "k8s"}}, + ) + if request.url.path == "/api/v1/k8s/create": + calls.append( + ( + request.method, + request.url.path, + {k.lower(): v for k, v in request.headers.items()}, + ) + ) + return httpx.Response(302, headers={"Location": "/challenges"}) + if request.url.path == "/api/v1/k8s/delete": + calls.append( + ( + request.method, + request.url.path, + {k.lower(): v for k, v in request.headers.items()}, + ) + ) + return httpx.Response(302, headers={"Location": "/challenges"}) + if request.url.path == "/api/v1/k8s/get": + return httpx.Response( + 200, + json={ + "ConnectionHost": "k8s", + "ConnectionPort": 31337, + "InstanceRunning": True, + "ThisChallengeInstance": True, + }, + ) + return httpx.Response(404, json={"success": False}) + + transport = httpx.MockTransport(handler) + client = make_client(transport) + try: + client._csrf_token = "csrf-token" + result = run(client.start_container(1)) + assert result["connection_info"] == "k8s:31337" + stopped = run(client.stop_container(challenge_id=1)) + assert stopped["status"] in {"stopped", "running"} + finally: + run(client.aclose()) + + for _, path, headers in calls: + assert headers.get("origin") == "https://ctfd.example.com" + assert headers.get("referer") == "https://ctfd.example.com/challenges" + + +def test_format_error_mapping(): + assert str(_format_error(ConfigError("bad"))) == "Configuration error: bad" + assert str(_format_error(AuthError("no"))) == "Auth failed: no" + assert str(_format_error(NotFoundError("no"))) == "Not found: no" + err = _format_error(RateLimitError("no", retry_after="30")) + assert str(err) == "Rate limited. Retry-After=30." + assert str(_format_error(CTFdClientError("no"))) == "CTFd API error: no" + assert str(_format_error(ValueError("nope"))) == "Unexpected error: nope" + + +def test_challenge_markdown_formatting(): + details = { + "id": 1, + "name": "Challenge", + "category": "web", + "value": 100, + "solved": True, + "description_text": "Hello", + "connection_info": "host:1", + "files": ["https://ctfd.example.com/files/a"], + } + md = _challenge_markdown(details) + assert "# Challenge" in md + assert "ID: 1 / Category: web / Points: 100 / Solved" in md + assert "## Description" in md + assert "Hello" in md + assert "## Connection" in md + assert "host:1" in md + assert "## Files" in md + assert "- https://ctfd.example.com/files/a" in md + + +def test_server_tools_map_errors(monkeypatch: pytest.MonkeyPatch): + class StubClient: + async def list_challenges(self, **_): + raise AuthError("bad") + + async def get_challenge(self, *_): + raise NotFoundError("missing") + + async def submit_flag(self, *_): + raise CTFdClientError("boom") + + async def start_container(self, *_): + raise RateLimitError("slow", retry_after="5") + + async def stop_container(self, *_, **__): + raise CTFdClientError("stop") + + async def fake_get_client(): + return StubClient() + + monkeypatch.setattr("ctfd_mcp.server._get_client", fake_get_client) + + with pytest.raises(RuntimeError, match="Auth failed"): + run(list_challenges()) + with pytest.raises(RuntimeError, match="Not found"): + run(challenge_details(1)) + with pytest.raises(RuntimeError, match="CTFd API error: boom"): + run(submit_flag(1, "flag")) + with pytest.raises(RuntimeError, match="Rate limited"): + run(start_container(1)) + with pytest.raises(RuntimeError, match="CTFd API error: stop"): + run(stop_container(container_id=1)) + with pytest.raises(RuntimeError, match="CTFd API error: boom"): + run(submit_flag(1, "flag")) diff --git a/uv.lock b/uv.lock index 959d025..822f66b 100644 --- a/uv.lock +++ b/uv.lock @@ -186,6 +186,7 @@ dependencies = [ [package.dev-dependencies] dev = [ { name = "pre-commit" }, + { name = "pytest" }, { name = "ruff" }, ] @@ -200,6 +201,7 @@ requires-dist = [ [package.metadata.requires-dev] dev = [ { name = "pre-commit", specifier = ">=3.6.0" }, + { name = "pytest", specifier = ">=8.0.0" }, { name = "ruff", specifier = ">=0.5.0" }, ] @@ -285,6 +287,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/61/66938bbb5fc52dbdf84594873d5b51fb1f7c7794e9c0f5bd885f30bc507b/idna-3.11-py3-none-any.whl", hash = "sha256:771a87f49d9defaf64091e6e6fe9c18d4833f140bd19464795bc32d966ca37ea", size = 71008, upload-time = "2025-10-12T14:55:18.883Z" }, ] +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + [[package]] name = "jsonschema" version = "4.25.1" @@ -346,6 +357,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/88/b2/d0896bdcdc8d28a7fc5717c305f1a861c26e18c05047949fb371034d98bd/nodeenv-1.10.0-py2.py3-none-any.whl", hash = "sha256:5bb13e3eed2923615535339b3c620e76779af4cb4c6a90deccc9e36b274d3827", size = 23438, upload-time = "2025-12-20T14:08:52.782Z" }, ] +[[package]] +name = "packaging" +version = "26.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/65/ee/299d360cdc32edc7d2cf530f3accf79c4fca01e96ffc950d8a52213bd8e4/packaging-26.0.tar.gz", hash = "sha256:00243ae351a257117b6a241061796684b084ed1c516a08c48a3f7e147a9d80b4", size = 143416, upload-time = "2026-01-21T20:50:39.064Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" }, +] + [[package]] name = "platformdirs" version = "4.5.1" @@ -355,6 +375,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/28/3bfe2fa5a7b9c46fe7e13c97bda14c895fb10fa2ebf1d0abb90e0cea7ee1/platformdirs-4.5.1-py3-none-any.whl", hash = "sha256:d03afa3963c806a9bed9d5125c8f4cb2fdaf74a55ab60e5d59b3fde758104d31", size = 18731, upload-time = "2025-12-05T13:52:56.823Z" }, ] +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + [[package]] name = "pre-commit" version = "4.5.1" @@ -462,6 +491,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c1/60/5d4751ba3f4a40a6891f24eec885f51afd78d208498268c734e256fb13c4/pydantic_settings-2.12.0-py3-none-any.whl", hash = "sha256:fddb9fd99a5b18da837b29710391e945b1e30c135477f484084ee513adb93809", size = 51880, upload-time = "2025-11-10T14:25:45.546Z" }, ] +[[package]] +name = "pygments" +version = "2.19.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631, upload-time = "2025-06-21T13:39:12.283Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" }, +] + [[package]] name = "pyjwt" version = "2.10.1" @@ -476,6 +514,22 @@ crypto = [ { name = "cryptography" }, ] +[[package]] +name = "pytest" +version = "9.0.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.1"