Merge pull request #1 from Cleverconnection/codex/extend-ctfcli-to-support-docker-challenge-fields

Support Docker plugin challenge fields and write-ups

Author: Cleverconnection

ctfcli/core/challenge.py

@@ -263,6 +263,64 @@ def _validate_files(self):
  if not (self.challenge_directory / challenge_file).exists():
  raise InvalidChallengeFile(f"File {challenge_file} could not be loaded")
 
+ @staticmethod
+ def _coerce_int(value: Any) -> Any:
+ if isinstance(value, int):
+ return value
+
+ if isinstance(value, str):
+ stripped_value = value.strip()
+ if stripped_value.isdigit() or (
+ stripped_value.startswith("-") and stripped_value[1:].isdigit()
+ ):
+ try:
+ return int(stripped_value)
+ except ValueError:
+ return value
+
+ return value
+
+ def _extract_container_fields(self, ignore: Tuple[str, ...] = ()) -> Dict[str, Any]:
+ container_keys = {
+ "image",
+ "port",
+ "command",
+ "volumes",
+ "ctype",
+ "initial",
+ "decay",
+ "minimum",
+ }
+
+ numeric_keys = {"port", "initial", "decay", "minimum"}
+
+ container_payload: Dict[str, Any] = {}
+ sources: List[Dict[str, Any]] = []
+
+ type_data = self.get("type_data")
+ if isinstance(type_data, dict):
+ sources.append(type_data)
+
+ if "extra" not in ignore:
+ extra = self.get("extra")
+ if isinstance(extra, dict):
+ sources.append(extra)
+
+ sources.append(self)
+
+ for source in sources:
+ for key in container_keys:
+ if key not in source or source[key] in (None, ""):
+ continue
+
+ value = source[key]
+ if key in numeric_keys:
+ value = self._coerce_int(value)
+
+ container_payload[key] = value
+
+ return container_payload
+
  def _get_initial_challenge_payload(self, ignore: Tuple[str] = ()) -> Dict:
  challenge = self
  challenge_payload = {
@@ -275,28 +333,26 @@ def _get_initial_challenge_payload(self, ignore: Tuple[str] = ()) -> Dict:
  "state": "hidden",
  }
 
- # Some challenge types (e.g., dynamic) override value.
- # We can't send it to CTFd because we don't know the current value
- if challenge.get("value", None) is not None:
- # if value is an int as string, cast it
- if type(challenge["value"]) == str and challenge["value"].isdigit():
- challenge_payload["value"] = int(challenge["value"])
-
- if type(challenge["value"] == int):
- challenge_payload["value"] = challenge["value"]
+ value = challenge.get("value", None)
+ if value is not None:
+ challenge_payload["value"] = self._coerce_int(value)
 
  if "attempts" not in ignore:
  challenge_payload["max_attempts"] = challenge.get("attempts", 0)
 
  if "connection_info" not in ignore:
  challenge_payload["connection_info"] = challenge.get("connection_info", None)
 
- if "logic" not in ignore:
- if challenge.get("logic"):
- challenge_payload["logic"] = challenge.get("logic") or "any"
+ if "logic" not in ignore and challenge.get("logic"):
+ challenge_payload["logic"] = challenge.get("logic") or "any"
 
  if "extra" not in ignore:
- challenge_payload = {**challenge_payload, **challenge.get("extra", {})}
+ extra = challenge.get("extra", {})
+ if isinstance(extra, dict):
+ challenge_payload = {**challenge_payload, **extra}
+
+ container_fields = self._extract_container_fields(ignore=ignore)
+ challenge_payload.update(container_fields)
 
  return challenge_payload
 
@@ -354,6 +410,52 @@ def _create_tags(self):
  )
  r.raise_for_status()
 
+ def _upload_writeup(self) -> None:
+ writeup_path = self.challenge_directory / "WRITEUP.md"
+
+ if not writeup_path.exists() or not writeup_path.is_file():
+ return
+
+ try:
+ content = writeup_path.read_text(encoding="utf-8")
+ except Exception as exc: # pragma: no cover - unexpected IO error
+ log.warning("Failed to read write-up for challenge %s: %s", self.get("name"), exc)
+ return
+
+ payload = {"challenge": self.challenge_id, "content": content, "publish": True}
+
+ try:
+ response = self.api.post("/api/v1/writeups", json=payload)
+ except Exception as exc: # pragma: no cover - network error
+ log.warning(
+ "Failed to upload write-up for challenge %s: %s", self.get("name", self.challenge_id), exc
+ )
+ return
+
+ if not response.ok:
+ log.warning(
+ "Failed to upload write-up for challenge %s: (%s) %s",
+ self.get("name", self.challenge_id),
+ response.status_code,
+ response.text,
+ )
+ return
+
+ try:
+ data = response.json()
+ except ValueError: # pragma: no cover - unexpected response body
+ log.warning(
+ "Failed to parse write-up upload response for challenge %s", self.get("name", self.challenge_id)
+ )
+ return
+
+ if not data.get("success", True):
+ log.warning(
+ "CTFd rejected write-up upload for challenge %s: %s",
+ self.get("name", self.challenge_id),
+ data,
+ )
+
  def _delete_file(self, remote_location: str):
  remote_files = self.api.get("/api/v1/files?type=challenge").json()["data"]
 
@@ -557,6 +659,14 @@ def _normalize_challenge(self, challenge_data: Dict[str, Any]):
  "state",
  "connection_info",
  "logic",
+ "image",
+ "port",
+ "command",
+ "volumes",
+ "ctype",
+ "initial",
+ "decay",
+ "minimum",
  ]
  for key in copy_keys:
  if key in challenge_data:
@@ -568,13 +678,6 @@ def _normalize_challenge(self, challenge_data: Dict[str, Any]):
  challenge["attribution"] = challenge["attribution"].strip().replace("\r\n", "\n").replace("\t", "")
  challenge["attempts"] = challenge_data["max_attempts"]
 
- for key in ["initial", "decay", "minimum"]:
- if key in challenge_data:
- if "extra" not in challenge:
- challenge["extra"] = {}
-
- challenge["extra"][key] = challenge_data[key]
-
  # Add flags
  r = self.api.get(f"/api/v1/challenges/{self.challenge_id}/flags")
  r.raise_for_status()
@@ -696,6 +799,8 @@ def sync(self, ignore: Tuple[str] = ()) -> None:
  click.secho(f"Failed to sync challenge: ({r.status_code}) {r.text}", fg="red")
  r.raise_for_status()
 
+ self._upload_writeup()
+
  # Update flags
  if "flags" not in ignore:
  self._delete_existing_flags()
@@ -824,6 +929,8 @@ def create(self, ignore: Tuple[str] = ()) -> None:
 
  self.challenge_id = r.json()["data"]["id"]
 
+ self._upload_writeup()
+
  # Create flags
  if challenge.get("flags") and "flags" not in ignore:
  self._create_flags()

ctfcli/spec/challenge-example.yml

@@ -12,12 +12,11 @@ type: standard
 
 # The extra field provides additional fields for data during the install/sync commands/
 # Fields in extra can be used to supply additional information for other challenge types
-# For example the follow extra field is for dynamic challenges. To use these following
-# extra fields, set the type to "dynamic" and uncomment the "extra" section below
-# extra:
-# initial: 500
-# decay: 100
-# minimum: 50
+# Many deployments (including the CTFd Docker plugin) accept dynamic scoring fields
+# like the following at the top level or under "extra"/"type_data":
+# initial: 500
+# decay: 100
+# minimum: 50
 
 # Settings used for Dockerfile deployment
 # If not used, remove or set to null
@@ -26,6 +25,14 @@ type: standard
 # Follow Docker best practices and assign a tag
 image: null
 
+# Additional container configuration fields supported by the CTFd Docker plugin
+# can optionally be provided here or within the extra/type_data sections.
+# port: 8080
+# command: ./start.sh
+# volumes:
+# - /host/path:/container/path
+# ctype: docker
+
 # Specify a protocol that should be used to connect to the running image
 # For example if the image is a website you can specify http or https
 # Otherwise you can specify tcp

tests/core/test_challenge.py

@@ -1360,9 +1360,90 @@ def mock_post(*args, **kwargs):
  [call("/api/v1/challenges/3", json={"state": "visible"}), call().raise_for_status()]
  )
  mock_api.post.assert_called_once_with("/api/v1/challenges", json=expected_challenge_payload)
- mock_api.get.assert_not_called()
- mock_api.delete.assert_not_called()
+ mock_api.get.assert_not_called()
+ mock_api.delete.assert_not_called()
+
+
+class TestContainerChallengePayload(unittest.TestCase):
+ minimal_challenge = BASE_DIR / "fixtures" / "challenges" / "test-challenge-minimal" / "challenge.yml"
+
+ def test_initial_payload_includes_container_fields_from_root(self):
+ overrides = {
+ "image": "library/test-challenge:latest",
+ "port": "8080",
+ "command": "/run.sh",
+ "volumes": ["/tmp:/app"],
+ "ctype": "container",
+ "initial": "100",
+ "decay": "5",
+ "minimum": "10",
+ }
+
+ challenge = Challenge(self.minimal_challenge, overrides)
+ payload = challenge._get_initial_challenge_payload()
+
+ self.assertEqual(payload["image"], "library/test-challenge:latest")
+ self.assertEqual(payload["port"], 8080)
+ self.assertEqual(payload["command"], "/run.sh")
+ self.assertEqual(payload["volumes"], ["/tmp:/app"])
+ self.assertEqual(payload["ctype"], "container")
+ self.assertEqual(payload["initial"], 100)
+ self.assertEqual(payload["decay"], 5)
+ self.assertEqual(payload["minimum"], 10)
+
+ def test_initial_payload_merges_container_fields_from_extra_and_type_data(self):
+ overrides = {
+ "extra": {"port": "9000", "command": "start"},
+ "type_data": {"volumes": ["/var:/srv"], "ctype": "docker"},
+ }
+
+ challenge = Challenge(self.minimal_challenge, overrides)
+ payload = challenge._get_initial_challenge_payload()
+
+ self.assertEqual(payload["port"], 9000)
+ self.assertEqual(payload["command"], "start")
+ self.assertEqual(payload["volumes"], ["/var:/srv"])
+ self.assertEqual(payload["ctype"], "docker")
+
+ def test_root_container_fields_override_nested_values(self):
+ overrides = {
+ "port": 7000,
+ "extra": {"port": "6000"},
+ "type_data": {"port": "5000"},
+ }
 
+ challenge = Challenge(self.minimal_challenge, overrides)
+ payload = challenge._get_initial_challenge_payload()
+
+ self.assertEqual(payload["port"], 7000)
+
+
+class TestWriteupUpload(unittest.TestCase):
+ minimal_challenge = BASE_DIR / "fixtures" / "challenges" / "test-challenge-minimal" / "challenge.yml"
+
+ @mock.patch("ctfcli.core.challenge.API")
+ def test_upload_writeup_posts_content(self, mock_api_constructor: MagicMock):
+ challenge = Challenge(self.minimal_challenge)
+ challenge.challenge_id = 123
+
+ writeup_path = challenge.challenge_directory / "WRITEUP.md"
+ writeup_path.write_text("# Sample Write-up", encoding="utf-8")
+
+ mock_api = mock_api_constructor.return_value
+ mock_response = MagicMock()
+ mock_response.ok = True
+ mock_response.json.return_value = {"success": True}
+ mock_api.post.return_value = mock_response
+
+ try:
+ challenge._upload_writeup()
+ finally:
+ writeup_path.unlink(missing_ok=True)
+
+ mock_api.post.assert_called_once_with(
+ "/api/v1/writeups",
+ json={"challenge": 123, "content": "# Sample Write-up", "publish": True},
+ )
 
 class TestLintChallenge(unittest.TestCase):
  minimal_challenge = BASE_DIR / "fixtures" / "challenges" / "test-challenge-minimal" / "challenge.yml"
@@ -1737,6 +1818,9 @@ def test_normalize_fetches_and_normalizes_challenge(self, mock_api_constructor:
  "description": "Test Description",
  "attribution": "Test Attribution",
  "attempts": 5,
+ "initial": 100,
+ "decay": 10,
+ "minimum": 10,
  "flags": [
  "flag{test-flag}",
  {"content": "flag{test-static}", "type": "static", "data": "case_insensitive"},
@@ -1747,11 +1831,6 @@ def test_normalize_fetches_and_normalizes_challenge(self, mock_api_constructor:
  "topics": ["topic-1", "topic-2"],
  "next": None,
  "requirements": {"prerequisites": ["First Test Challenge", "Other Test Challenge"], "anonymize": False},
- "extra": {
- "initial": 100,
- "decay": 10,
- "minimum": 10,
- },
  },
  normalized_data,
  )