From 57d60764993eb4a0823fd83263f18965cd302268 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mi=C5=82osz=20Skaza?= Date: Mon, 1 Jun 2026 15:19:41 +0200 Subject: [PATCH 1/3] add scheduled_at field --- ctfcli/core/challenge.py | 80 ++++++++++++++++++ ctfcli/spec/challenge-example.yml | 7 ++ tests/core/test_challenge.py | 133 +++++++++++++++++++++++++++++- 3 files changed, 219 insertions(+), 1 deletion(-) diff --git a/ctfcli/core/challenge.py b/ctfcli/core/challenge.py index d87f807..2fdc0c4 100644 --- a/ctfcli/core/challenge.py +++ b/ctfcli/core/challenge.py @@ -1,6 +1,7 @@ import logging import re import subprocess +from datetime import datetime, timezone from os import PathLike from pathlib import Path from typing import Any @@ -67,6 +68,7 @@ class Challenge(dict): "hints", "requirements", "next", + "scheduled_at", "state", "version", ] @@ -81,6 +83,7 @@ class Challenge(dict): "files", "hints", "requirements", + "scheduled_at", "state", "version", ] @@ -133,8 +136,67 @@ def is_default_challenge_property(key: str, value: Any) -> bool: if key == "requirements" and value == {"prerequisites": [], "anonymize": False}: return True + if key == "scheduled_at" and value is None: + return True + return bool(key == "next" and value is None) + def _parse_scheduled_at(self, value: Any) -> "datetime | None": + # Never assume a timezone for scheduled_at: always expect an explicit offset + if value is None: + return None + + if isinstance(value, datetime): + # PyYAML parses unquoted ISO timestamps directly into datetime objects + parsed = value + elif isinstance(value, str): + if not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError as e: + raise InvalidChallengeFile( + f"Challenge file at {self.challenge_file_path} has an invalid 'scheduled_at' value " + f"'{value}': expected an ISO 8601 datetime" + ) from e + else: + raise InvalidChallengeFile( + f"Challenge file at {self.challenge_file_path} has an invalid 'scheduled_at' value: " + "expected an ISO 8601 datetime string" + ) + + if parsed.tzinfo is None: + raise InvalidChallengeFile( + f"Challenge file at {self.challenge_file_path} 'scheduled_at' value '{value}' is missing a " + "timezone. ctfcli does not assume timezones - specify an explicit offset " + "(e.g. 2026-06-15T12:00:00+00:00 for UTC)" + ) + + return parsed + + @staticmethod + def _normalize_scheduled_at(value: Any) -> "str | None": + # CTFd stores and returns scheduled_at as a naive UTC datetime. + # Make the timezone explicit (UTC) on the challenge + if not value: + return None + + parsed = value if isinstance(value, datetime) else datetime.fromisoformat(value) + parsed = parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) + + return parsed.isoformat() + + def _compare_scheduled_at(self, local: Any, remote: Any) -> bool: + # Compare two scheduled_at values by the instant they represent, so that + # equivalent times written with different offsets compare as equal. + local_parsed = self._parse_scheduled_at(local) + remote_parsed = self._parse_scheduled_at(remote) + + if local_parsed is None or remote_parsed is None: + return local_parsed == remote_parsed + + return local_parsed.astimezone(timezone.utc) == remote_parsed.astimezone(timezone.utc) + @staticmethod def clone(config, remote_challenge): name = remote_challenge["name"] @@ -318,6 +380,11 @@ def _get_initial_challenge_payload(self, ignore: tuple[str] = ()) -> dict: if "connection_info" not in ignore: challenge_payload["connection_info"] = challenge.get("connection_info", None) + if "scheduled_at" not in ignore: + # _parse_scheduled_at validates the timezone is explicit and raises otherwise + parsed_scheduled_at = self._parse_scheduled_at(challenge.get("scheduled_at")) + challenge_payload["scheduled_at"] = parsed_scheduled_at.isoformat() if parsed_scheduled_at else None + if "logic" not in ignore and challenge.get("logic"): challenge_payload["logic"] = challenge.get("logic") or "any" @@ -758,6 +825,9 @@ def _normalize_challenge(self, challenge_data: dict[str, Any]): if key in challenge_data: challenge[key] = challenge_data[key] + # CTFd returns scheduled_at as a naive UTC datetime - make the timezone explicit + challenge["scheduled_at"] = self._normalize_scheduled_at(challenge_data.get("scheduled_at")) + challenge["description"] = challenge_data["description"].strip().replace("\r\n", "\n").replace("\t", "") challenge["attribution"] = challenge_data.get("attribution", "") if challenge["attribution"]: @@ -1118,6 +1188,13 @@ def lint(self, skip_hadolint=False, flag_format="flag{") -> bool: if challenge.get(field) is None: issues["fields"].append(f"challenge.yml is missing required field: {field}") + # Check that scheduled_at, if present, carries an explicit timezone + if challenge.get("scheduled_at") is not None: + try: + self._parse_scheduled_at(challenge.get("scheduled_at")) + except InvalidChallengeFile as e: + issues["fields"].append(str(e)) + # Check that the image field and Dockerfile match if (self.challenge_directory / "Dockerfile").is_file() and challenge.get("image", "") != ".": issues["dockerfile"].append("Dockerfile exists but image field does not point to it") @@ -1295,6 +1372,9 @@ def verify(self, ignore: tuple[str] = ()) -> bool: if key == "next" and self._compare_challenge_next(challenge[key], normalized_challenge[key]): continue + if key == "scheduled_at" and self._compare_scheduled_at(challenge[key], normalized_challenge[key]): + continue + click.secho( f"{key} comparison failed.", fg="yellow", diff --git a/ctfcli/spec/challenge-example.yml b/ctfcli/spec/challenge-example.yml index d7d5d0f..f83d710 100644 --- a/ctfcli/spec/challenge-example.yml +++ b/ctfcli/spec/challenge-example.yml @@ -158,6 +158,13 @@ requirements: # if you want to remove or disable it. next: null +# scheduled_at schedules a timed release: a visible challenge stays hidden from +# players until this moment passes. Omit or set to null for no schedule. +# Must be an ISO 8601 datetime WITH an explicit timezone offset +# scheduled_at: "2026-06-15T12:00:00+00:00" # UTC +# scheduled_at: "2026-06-15T14:00:00+02:00" # CEST +# scheduled_at: null + # The state of the challenge. # If the field is omitted, the challenge is visible by default. # If provided, the field can take one of two values: hidden, visible. diff --git a/tests/core/test_challenge.py b/tests/core/test_challenge.py index de892db..89a1e70 100644 --- a/tests/core/test_challenge.py +++ b/tests/core/test_challenge.py @@ -1,5 +1,6 @@ import re import unittest +from datetime import datetime, timezone from pathlib import Path from unittest import mock from unittest.mock import ANY, MagicMock, call, mock_open @@ -412,6 +413,7 @@ def test_updates_simple_properties(self, mock_api_constructor: MagicMock, *args, "value": 150, "state": "hidden", "connection_info": "https://example.com", + "scheduled_at": None, "max_attempts": 0, } @@ -458,6 +460,7 @@ def test_updates_attempts(self, mock_api_constructor: MagicMock, *args, **kwargs "state": "hidden", "max_attempts": 5, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -506,6 +509,7 @@ def test_updates_extra_properties(self, mock_api_constructor: MagicMock, *args, "application_name": "application-name", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -563,6 +567,7 @@ def test_updates_flags(self, mock_api_constructor: MagicMock, *args, **kwargs): "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -652,6 +657,7 @@ def test_updates_topics(self, mock_api_constructor: MagicMock, *args, **kwargs): "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -713,6 +719,7 @@ def test_updates_tags(self, mock_api_constructor: MagicMock, *args, **kwargs): "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -777,6 +784,7 @@ def test_updates_files(self, mock_api_constructor: MagicMock, *args, **kwargs): "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } def mock_get(*args, **kwargs): @@ -932,6 +940,7 @@ def test_updates_hints_with_requirements(self, mock_api_constructor: MagicMock, "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -1039,6 +1048,7 @@ def test_updates_requirements(self, mock_api_constructor: MagicMock, *args, **kw "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -1084,6 +1094,7 @@ def test_challenge_cannot_require_itself( "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } def mock_get(*args, **kwargs): @@ -1145,6 +1156,7 @@ def test_defaults_to_standard_challenge_type(self, mock_api_constructor: MagicMo "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -1183,6 +1195,7 @@ def test_defaults_to_visible_state(self, mock_api_constructor: MagicMock, *args, "value": 150, "max_attempts": 0, "connection_info": None, + "scheduled_at": None, # initial patch should set the state to hidden for the duration of the update "state": "hidden", } @@ -1236,6 +1249,7 @@ def test_does_not_update_dynamic_value(self, mock_api_constructor: MagicMock, *a "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -1295,6 +1309,7 @@ def test_updates_multiple_attributes_at_once(self, mock_api_constructor: MagicMo "state": "hidden", "max_attempts": 5, "connection_info": "https://example.com", + "scheduled_at": None, } mock_api: MagicMock = mock_api_constructor.return_value @@ -1356,6 +1371,7 @@ def test_does_not_update_ignored_attributes(self): "value", "attempts", "connection_info", + "scheduled_at", "state", # complex types "extra", @@ -1379,6 +1395,7 @@ def test_does_not_update_ignored_attributes(self): "state": "visible", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } # This nightmare is necessary because on python 3.8 for whatever reason "with" with multiple context managers @@ -1409,6 +1426,7 @@ def test_does_not_update_ignored_attributes(self): "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } # expect the payload to modify values with new ones from challenge.yml @@ -1435,6 +1453,10 @@ def test_does_not_update_ignored_attributes(self): challenge["connection_info"] = "https://example.com" del expected_challenge_payload["connection_info"] + if p == "scheduled_at": + challenge["scheduled_at"] = "2026-06-15T12:00:00+00:00" + del expected_challenge_payload["scheduled_at"] + if p == "state": challenge[p] = "new-value" @@ -1511,6 +1533,7 @@ def test_creates_standard_challenge(self, mock_api_constructor: MagicMock, *args "max_attempts": 5, "type": "standard", "connection_info": "https://example.com", + "scheduled_at": None, "extra_property": "extra_property_value", "state": "hidden", } @@ -1638,7 +1661,7 @@ def test_exits_if_files_do_not_exist(self, mock_api_constructor: MagicMock, *arg def test_does_not_set_ignored_attributes(self): # fmt:off properties = [ - "value", "category", "description", "attribution", "attempts", "connection_info", "state", # simple types + "value", "category", "description", "attribution", "attempts", "connection_info", "scheduled_at", "state", # simple types # noqa: E501 "extra", "flags", "topics", "tags", "files", "hints", "requirements", "solution" # complex types ] # fmt:on @@ -1662,6 +1685,7 @@ def test_does_not_set_ignored_attributes(self): "state": "hidden", "max_attempts": 0, "connection_info": None, + "scheduled_at": None, } # add a property that should be defined but ignored @@ -1691,6 +1715,10 @@ def test_does_not_set_ignored_attributes(self): challenge["connection_info"] = "https://example.com" del expected_challenge_payload["connection_info"] + if p == "scheduled_at": + challenge["scheduled_at"] = "2026-06-15T12:00:00+00:00" + del expected_challenge_payload["scheduled_at"] + if p == "state": challenge[p] = "new-value" @@ -2145,6 +2173,7 @@ def test_normalize_fetches_and_normalizes_challenge(self, mock_api_constructor: "hints": ["free hint", {"content": "paid hint", "cost": 100}], "topics": ["topic-1", "topic-2"], "next": None, + "scheduled_at": None, "requirements": {"prerequisites": ["First Test Challenge", "Other Test Challenge"], "anonymize": False}, "extra": { "initial": 100, @@ -2278,3 +2307,105 @@ def test_additional_keys_are_appended(self): loaded_data = yaml.safe_load(dumped_data) self.assertDictEqual(challenge, loaded_data) + + +class TestChallengeScheduledAt(unittest.TestCase): + minimal_challenge = BASE_DIR / "fixtures" / "challenges" / "test-challenge-minimal" / "challenge.yml" + + def test_parse_accepts_timezone_aware_string(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T12:00:00+00:00"}) + parsed = challenge._parse_scheduled_at(challenge["scheduled_at"]) + self.assertEqual(parsed, datetime(2026, 6, 15, 12, 0, 0, tzinfo=timezone.utc)) + + def test_parse_accepts_z_suffix(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T12:00:00Z"}) + parsed = challenge._parse_scheduled_at(challenge["scheduled_at"]) + self.assertEqual(parsed.astimezone(timezone.utc), datetime(2026, 6, 15, 12, 0, 0, tzinfo=timezone.utc)) + + def test_parse_accepts_timezone_aware_datetime(self): + # PyYAML parses an unquoted ISO timestamp into a datetime object + challenge = Challenge( + self.minimal_challenge, + {"scheduled_at": datetime(2026, 6, 15, 14, 0, 0, tzinfo=timezone.utc)}, + ) + parsed = challenge._parse_scheduled_at(challenge["scheduled_at"]) + self.assertEqual(parsed, datetime(2026, 6, 15, 14, 0, 0, tzinfo=timezone.utc)) + + def test_parse_returns_none_for_none_or_empty(self): + challenge = Challenge(self.minimal_challenge) + self.assertIsNone(challenge._parse_scheduled_at(None)) + self.assertIsNone(challenge._parse_scheduled_at("")) + + def test_parse_rejects_naive_string(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T12:00:00"}) + with self.assertRaises(InvalidChallengeFile) as ctx: + challenge._parse_scheduled_at(challenge["scheduled_at"]) + self.assertIn("timezone", str(ctx.exception)) + + def test_parse_rejects_naive_datetime(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": datetime(2026, 6, 15, 12, 0, 0)}) # noqa: DTZ001 + with self.assertRaises(InvalidChallengeFile): + challenge._parse_scheduled_at(challenge["scheduled_at"]) + + def test_parse_rejects_invalid_string(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "not-a-date"}) + with self.assertRaises(InvalidChallengeFile): + challenge._parse_scheduled_at(challenge["scheduled_at"]) + + def test_payload_includes_scheduled_at_iso(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T14:00:00+02:00"}) + payload = challenge._get_initial_challenge_payload() + # The explicit offset is preserved when sent to CTFd (CTFd normalizes server-side) + self.assertEqual(payload["scheduled_at"], "2026-06-15T14:00:00+02:00") + + def test_payload_scheduled_at_none_when_absent(self): + challenge = Challenge(self.minimal_challenge) + payload = challenge._get_initial_challenge_payload() + self.assertIsNone(payload["scheduled_at"]) + + def test_payload_omits_scheduled_at_when_ignored(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T12:00:00+00:00"}) + payload = challenge._get_initial_challenge_payload(ignore=("scheduled_at",)) + self.assertNotIn("scheduled_at", payload) + + def test_payload_raises_on_naive_scheduled_at(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T12:00:00"}) + with self.assertRaises(InvalidChallengeFile): + challenge._get_initial_challenge_payload() + + def test_normalize_makes_utc_explicit(self): + # CTFd returns naive UTC; ctfcli should write it back with an explicit offset + self.assertEqual( + Challenge._normalize_scheduled_at("2026-06-15T12:00:00"), + "2026-06-15T12:00:00+00:00", + ) + + def test_normalize_returns_none_for_none(self): + self.assertIsNone(Challenge._normalize_scheduled_at(None)) + + def test_compare_equal_for_same_instant_different_offsets(self): + challenge = Challenge(self.minimal_challenge) + # 14:00+02:00 == 12:00+00:00 (same instant) + self.assertTrue(challenge._compare_scheduled_at("2026-06-15T14:00:00+02:00", "2026-06-15T12:00:00+00:00")) + + def test_compare_not_equal_for_different_instants(self): + challenge = Challenge(self.minimal_challenge) + self.assertFalse(challenge._compare_scheduled_at("2026-06-15T13:00:00+00:00", "2026-06-15T12:00:00+00:00")) + + def test_compare_handles_none(self): + challenge = Challenge(self.minimal_challenge) + self.assertTrue(challenge._compare_scheduled_at(None, None)) + self.assertFalse(challenge._compare_scheduled_at("2026-06-15T12:00:00+00:00", None)) + + def test_lint_flags_naive_scheduled_at(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T12:00:00"}) + with self.assertRaises(LintException) as ctx: + challenge.lint(skip_hadolint=True) + field_issues = " ".join(ctx.exception.issues["fields"]) + self.assertIn("scheduled_at", field_issues) + self.assertIn("timezone", field_issues) + + def test_lint_accepts_timezone_aware_scheduled_at(self): + challenge = Challenge(self.minimal_challenge, {"scheduled_at": "2026-06-15T12:00:00+00:00"}) + # Should not raise for scheduled_at (no other lint issues in the minimal fixture) + self.assertTrue(challenge.lint(skip_hadolint=True)) From 9bf02da8585e80d0ad092ed1fab7465b26c96b15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mi=C5=82osz=20Skaza?= Date: Fri, 17 Jul 2026 07:39:57 +0200 Subject: [PATCH 2/3] fix datetime parsing --- ctfcli/core/challenge.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/ctfcli/core/challenge.py b/ctfcli/core/challenge.py index 2fdc0c4..4d30149 100644 --- a/ctfcli/core/challenge.py +++ b/ctfcli/core/challenge.py @@ -141,6 +141,15 @@ def is_default_challenge_property(key: str, value: Any) -> bool: return bool(key == "next" and value is None) + @staticmethod + def _datetime_from_iso(value: str) -> datetime: + # datetime.fromisoformat only accepts a 'Z' suffix on Python 3.11+, + # so translate it to an explicit offset for the versions that don't + if value.endswith(("Z", "z")): + value = value[:-1] + "+00:00" + + return datetime.fromisoformat(value) + def _parse_scheduled_at(self, value: Any) -> "datetime | None": # Never assume a timezone for scheduled_at: always expect an explicit offset if value is None: @@ -153,7 +162,7 @@ def _parse_scheduled_at(self, value: Any) -> "datetime | None": if not value.strip(): return None try: - parsed = datetime.fromisoformat(value) + parsed = self._datetime_from_iso(value) except ValueError as e: raise InvalidChallengeFile( f"Challenge file at {self.challenge_file_path} has an invalid 'scheduled_at' value " @@ -174,14 +183,14 @@ def _parse_scheduled_at(self, value: Any) -> "datetime | None": return parsed - @staticmethod - def _normalize_scheduled_at(value: Any) -> "str | None": + @classmethod + def _normalize_scheduled_at(cls, value: Any) -> "str | None": # CTFd stores and returns scheduled_at as a naive UTC datetime. # Make the timezone explicit (UTC) on the challenge if not value: return None - parsed = value if isinstance(value, datetime) else datetime.fromisoformat(value) + parsed = value if isinstance(value, datetime) else cls._datetime_from_iso(value) parsed = parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) return parsed.isoformat() From 4a3afe58cc92c6e769179731523c188362807459 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mi=C5=82osz=20Skaza?= Date: Fri, 17 Jul 2026 07:46:42 +0200 Subject: [PATCH 3/3] merge main --- CHANGELOG.md | 14 ++ ctfcli/__init__.py | 2 +- ctfcli/__main__.py | 46 ++++- ctfcli/cli/challenges.py | 2 +- ctfcli/cli/pages.py | 6 +- ctfcli/core/challenge.py | 73 +++++++ ctfcli/core/page.py | 7 +- ctfcli/spec/challenge-example.yml | 13 +- ctfcli/utils/integrations/__init__.py | 8 + ctfcli/utils/integrations/github.py | 41 ++++ ctfcli/utils/integrations/gitlab.py | 20 ++ pyproject.toml | 2 +- tests/core/test_challenge.py | 266 +++++++++++++++++++++++++- tests/core/test_page.py | 43 +++++ 14 files changed, 530 insertions(+), 13 deletions(-) create mode 100644 ctfcli/utils/integrations/__init__.py create mode 100644 ctfcli/utils/integrations/github.py create mode 100644 ctfcli/utils/integrations/gitlab.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b8168e2..514b705 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog +# 0.1.8 / 2026-07-14 + +### Added + +- Add support for hint requirements through hint keys + - Hints in challenge.yml can now specify a `key` and list other hints' keys under `requirements` to gate them behind those hints +- Add repo integrations to create challenge repos with sync integrations during `ctf init`, including `ctf init --github --gitlab` +- Add `--force` to `ctf pages` + +### Fixed + +- Fix return value for sync verification to properly check for failed challenge verification counts +- Stream POST requests using a multipart encoder to handle large file uploads without reading the entire file into memory + # 0.1.7 / 2026-02-26 ### Added diff --git a/ctfcli/__init__.py b/ctfcli/__init__.py index b8acfd3..8c79cab 100644 --- a/ctfcli/__init__.py +++ b/ctfcli/__init__.py @@ -1,2 +1,2 @@ -__version__ = "0.1.7" +__version__ = "0.1.8" __name__ = "ctfcli" diff --git a/ctfcli/__main__.py b/ctfcli/__main__.py index 76211c1..72801a3 100644 --- a/ctfcli/__main__.py +++ b/ctfcli/__main__.py @@ -22,6 +22,7 @@ ) from ctfcli.core.plugins import load_plugins from ctfcli.utils.git import check_if_dir_is_inside_git_repo +from ctfcli.utils.integrations import INTEGRATIONS # Init logging logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO").upper()) @@ -35,8 +36,12 @@ def init( directory: str | os.PathLike | None = None, no_git: bool = False, no_commit: bool = False, + github: bool = False, + gitlab: bool = False, ): - log.debug(f"init: (directory={directory}, no_git={no_git}, no_commit={no_commit})") + log.debug( + f"init: (directory={directory}, no_git={no_git}, no_commit={no_commit}, github={github}, gitlab={gitlab})" + ) project_path = Path.cwd() # Create our project directory if requested @@ -62,7 +67,26 @@ def init( ctf_token = click.prompt("Please enter CTFd Admin Access Token", default="", show_default=False) - # Confirm information with user + # Collect CI integrations chosen via switches + selected = [] + if github: + selected.append("github") + if gitlab: + selected.append("gitlab") + + # If no integration switch was passed, ask once for a comma-separated list + if not selected: + available = ", ".join(INTEGRATIONS) + response = click.prompt( + f"Any integrations to add: (comma-separated, leave blank for none)\n" + f"Available: {available} \n" + "Integrations", + default="", + show_default=False, + ) + selected = [name.strip().lower() for name in response.split(",") if name.strip()] + + # Confirm information with user before creating anything if not click.confirm(f"Do you want to continue with {ctf_url} and {ctf_token}", default=True): click.echo("Aborted!") return @@ -74,6 +98,20 @@ def init( with (project_path / ".ctf" / "config").open(mode="a+") as config_file: config.write(config_file) + # Generate integration files, collecting the paths to stage in the initial commit + paths_to_add = [".ctf/config"] + for name in selected: + entry = INTEGRATIONS.get(name) + if entry is None: + click.secho( + f"Unknown integration '{name}'. Available: {', '.join(INTEGRATIONS)}", + fg="yellow", + ) + continue + + _label, creator = entry + paths_to_add.extend(creator(project_path)) + # if git init is to be skipped we can return if no_git: click.secho("Skipping git init.", fg="yellow") @@ -88,7 +126,7 @@ def init( click.secho("Skipping git commit.", fg="yellow") return - subprocess.call(["git", "add", ".ctf/config"], cwd=project_path) + subprocess.call(["git", "add", *paths_to_add], cwd=project_path) subprocess.call(["git", "commit", "-m", "init ctfcli project"], cwd=project_path) return @@ -100,7 +138,7 @@ def init( click.secho("Skipping git commit.", fg="yellow") return - subprocess.call(["git", "add", ".ctf/config"], cwd=project_path) + subprocess.call(["git", "add", *paths_to_add], cwd=project_path) subprocess.call(["git", "commit", "-m", "init ctfcli project"], cwd=project_path) def config(self): diff --git a/ctfcli/cli/challenges.py b/ctfcli/cli/challenges.py index e2333e6..4441259 100644 --- a/ctfcli/cli/challenges.py +++ b/ctfcli/cli/challenges.py @@ -795,7 +795,7 @@ def deploy( if existing_challenge: click.secho(f"Updating challenge '{challenge_name}'", fg="blue") challenge_instance.sync( - ignore=["flags", "topics", "tags", "files", "hints", "requirements", "state"] + ignore=["flags", "topics", "tags", "files", "hints", "requirements", "module", "state"] ) else: click.secho(f"Creating challenge '{challenge_name}'", fg="blue") diff --git a/ctfcli/cli/pages.py b/ctfcli/cli/pages.py index 5bb155e..1b1f2b9 100644 --- a/ctfcli/cli/pages.py +++ b/ctfcli/cli/pages.py @@ -28,7 +28,7 @@ def _page_operation(self, page: Page, operation: str, *args, **kwargs) -> int: click.secho(str(e), fg="red") return 1 - def push(self, page: str | None = None) -> int: + def push(self, page: str | None = None, force: bool = False) -> int: pages = Page.get_local_pages() if page: @@ -37,14 +37,14 @@ def push(self, page: str | None = None) -> int: None, ) if page_object: - return self._page_operation(page_object, "push") + return self._page_operation(page_object, "push", force=force) click.secho(f"Could not find page '{page}'", fg="red") return 1 return_code = 0 for page_object in pages: - status = self._page_operation(page_object, "push") + status = self._page_operation(page_object, "push", force=force) if status == 1: return_code = 1 diff --git a/ctfcli/core/challenge.py b/ctfcli/core/challenge.py index 4d30149..f1754eb 100644 --- a/ctfcli/core/challenge.py +++ b/ctfcli/core/challenge.py @@ -69,6 +69,7 @@ class Challenge(dict): "requirements", "next", "scheduled_at", + "module", "state", "version", ] @@ -139,6 +140,9 @@ def is_default_challenge_property(key: str, value: Any) -> bool: if key == "scheduled_at" and value is None: return True + if key == "module" and value is None: + return True + return bool(key == "next" and value is None) @staticmethod @@ -772,6 +776,42 @@ def _set_next(self, _next): r = self.api.patch(f"/api/v1/challenges/{self.challenge_id}", json=next_payload) r.raise_for_status() + def _set_module(self): + module = self.get("module", None) + + if module is None or module == "": + # explicit null (or empty) module - remove the challenge from its module + module_id = None + else: + # module is always treated as a name - coerce so a numeric name + # (e.g. "2024", which YAML loads as an int) is handled as a string + module = str(module) + + # find the module id from the modules installed on the remote + module_id = None + r = self.api.get("/api/v1/modules") + r.raise_for_status() + remote_modules = r.json()["data"] + for remote_module in remote_modules: + if remote_module["name"] == module: + module_id = remote_module["id"] + break + + # the module does not exist yet - create it + if module_id is None: + r = self.api.post("/api/v1/modules", json={"name": module}) + r.raise_for_status() + module_id = r.json()["data"]["id"] + click.secho( + f'Created module "{module}". ' + "Remember to assign audiences to it in the admin panel if you want to restrict access.", + fg="yellow", + ) + + module_payload = {"module_id": module_id} + r = self.api.patch(f"/api/v1/challenges/{self.challenge_id}", json=module_payload) + r.raise_for_status() + # Compare challenge requirements, will resolve all IDs to names def _compare_challenge_requirements(self, r1: list[str | int], r2: list[str | int]) -> bool: remote_challenges = self.load_installed_challenges() @@ -811,6 +851,14 @@ def normalize_next(r): return normalize_next(r1) == normalize_next(r2) + # Compare module assignments - modules are always referenced by name, so a + # numeric name (loaded from YAML as an int) is coerced to a string to compare + def _compare_challenge_module(self, m1: str | int | None, m2: str | int | None) -> bool: + def normalize_module(m): + return None if m is None else str(m) + + return normalize_module(m1) == normalize_module(m2) + # Normalize challenge data from the API response to match challenge.yml # It will remove any extra fields from the remote, as well as expand external references # that have to be fetched separately (e.g., files, flags, hints, etc.) @@ -929,6 +977,7 @@ def _normalize_challenge(self, challenge_data: dict[str, Any]): r2.raise_for_status() challenges = r2.json()["data"] challenge["requirements"]["prerequisites"] = [c["name"] for c in challenges if c["id"] in requirements] + # Add anonymize flag challenge["requirements"]["anonymize"] = (r.json().get("data") or {}).get("anonymize", False) @@ -942,6 +991,16 @@ def _normalize_challenge(self, challenge_data: dict[str, Any]): else: challenge["next"] = None + # Add module + module_id = challenge_data.get("module_id") + if module_id: + # Prefer the module name over the ID + r = self.api.get(f"/api/v1/modules/{module_id}") + r.raise_for_status() + challenge["module"] = (r.json().get("data") or {}).get("name", None) + else: + challenge["module"] = None + return challenge # Create a dictionary of remote files in { basename: {"url": "", "location": ""} } format @@ -1076,6 +1135,13 @@ def sync(self, ignore: tuple[str] = ()) -> None: if "next" not in ignore: self._set_next(_next) + # Update module + # Only touch the module assignment if the key is present in challenge.yml - + # an explicit "module: null" removes the challenge from its module, + # while an absent key leaves the remote assignment untouched + if "module" in challenge and "module" not in ignore: + self._set_module() + if "solution" not in ignore: resolved_solution = self._resolve_solution_path() if not resolved_solution: @@ -1166,6 +1232,10 @@ def create(self, ignore: tuple[str] = ()) -> None: if "next" not in ignore: self._set_next(_next) + # Assign module + if challenge.get("module") and "module" not in ignore: + self._set_module() + # Add solution if "solution" not in ignore: self._create_solution() @@ -1384,6 +1454,9 @@ def verify(self, ignore: tuple[str] = ()) -> bool: if key == "scheduled_at" and self._compare_scheduled_at(challenge[key], normalized_challenge[key]): continue + if key == "module" and self._compare_challenge_module(challenge[key], normalized_challenge[key]): + continue + click.secho( f"{key} comparison failed.", fg="yellow", diff --git a/ctfcli/core/page.py b/ctfcli/core/page.py index a3eb484..a7fbc83 100644 --- a/ctfcli/core/page.py +++ b/ctfcli/core/page.py @@ -152,9 +152,14 @@ def pull(self, overwrite=False): with open(page_path, "wb+") as page_file: frontmatter.dump(self.as_frontmatter_post(), page_file) - def push(self): + def push(self, force=False): # upload / create remote copy of a local page if getattr(self, "page_id", None): + if force: + click.secho("Syncing existing page instead (because of --force)", fg="yellow") + self.sync() + return + raise IllegalPageOperation( f"Cannot push page '{self.page_file_path}' - remote version exists. Use sync instead." ) diff --git a/ctfcli/spec/challenge-example.yml b/ctfcli/spec/challenge-example.yml index f83d710..14ec1ae 100644 --- a/ctfcli/spec/challenge-example.yml +++ b/ctfcli/spec/challenge-example.yml @@ -151,7 +151,7 @@ requirements: # - "Are you alive" # anonymize: true -# The next is used to display a next recommended challenge to a user when +# The next is used to display a next recommended challenge to a user when # the user correctly answers the current challenge. # Can be removed if unused # Accepts a challenge name as a string, a challenge ID as an integer, or null @@ -165,6 +165,17 @@ next: null # scheduled_at: "2026-06-15T14:00:00+02:00" # CEST # scheduled_at: null +# The module is used to assign this challenge to a CTFd module. +# Modules group challenges together, and access to a module can be restricted +# to specific audiences (groups of users or teams) in the CTFd admin panel. +# If the module does not exist on the CTFd instance yet, it will be created +# during install/sync - audiences still have to be assigned to it manually. +# Accepts a module name as a string, or null if you want to remove the +# challenge from its module. +# If the field is omitted, the remote module assignment is left untouched. +# Can be removed if unused +module: null + # The state of the challenge. # If the field is omitted, the challenge is visible by default. # If provided, the field can take one of two values: hidden, visible. diff --git a/ctfcli/utils/integrations/__init__.py b/ctfcli/utils/integrations/__init__.py new file mode 100644 index 0000000..a73ca1c --- /dev/null +++ b/ctfcli/utils/integrations/__init__.py @@ -0,0 +1,8 @@ +from ctfcli.utils.integrations.github import create_github_integration +from ctfcli.utils.integrations.gitlab import create_gitlab_integration + +# name -> (human label, creator). Add a future provider here + one switch in init(). +INTEGRATIONS = { + "github": ("GitHub Actions", create_github_integration), + "gitlab": ("GitLab CI/CD", create_gitlab_integration), +} diff --git a/ctfcli/utils/integrations/github.py b/ctfcli/utils/integrations/github.py new file mode 100644 index 0000000..cc60ffc --- /dev/null +++ b/ctfcli/utils/integrations/github.py @@ -0,0 +1,41 @@ +from pathlib import Path + +import click + +GITHUB_ACTIONS_SYNC_WORKFLOW = """\ +name: Sync challenges to CTFd + +on: + push: + branches: + - main + - master + +jobs: + sync: + # Only run on the repository's default branch + if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.x" + + - name: Install ctfcli + run: pip install ctfcli + + - name: Install challenges to CTFd + run: ctf challenge install --force +""" + + +def create_github_integration(project_path: Path) -> list[str]: + workflows_dir = project_path / ".github" / "workflows" + workflows_dir.mkdir(parents=True, exist_ok=True) + (workflows_dir / "sync.yml").write_text(GITHUB_ACTIONS_SYNC_WORKFLOW) + click.secho("Created .github/workflows/sync.yml", fg="green") + return [".github/workflows/sync.yml"] diff --git a/ctfcli/utils/integrations/gitlab.py b/ctfcli/utils/integrations/gitlab.py new file mode 100644 index 0000000..910e78d --- /dev/null +++ b/ctfcli/utils/integrations/gitlab.py @@ -0,0 +1,20 @@ +from pathlib import Path + +import click + +GITLAB_CI_SYNC_PIPELINE = """\ +sync-challenges: + image: python:3 + # Only run on the repository's default branch + rules: + - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH + script: + - pip install ctfcli + - ctf challenge install --force +""" + + +def create_gitlab_integration(project_path: Path) -> list[str]: + (project_path / ".gitlab-ci.yml").write_text(GITLAB_CI_SYNC_PIPELINE) + click.secho("Created .gitlab-ci.yml", fg="green") + return [".gitlab-ci.yml"] diff --git a/pyproject.toml b/pyproject.toml index 61ff86b..a21a106 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "ctfcli" -version = "0.1.7" +version = "0.1.8" description = "ctfcli is a tool to manage Capture The Flag events and challenges" authors = [ { name = "Kevin Chung", email = "kchung@ctfd.io" }, diff --git a/tests/core/test_challenge.py b/tests/core/test_challenge.py index 89a1e70..5554a4d 100644 --- a/tests/core/test_challenge.py +++ b/tests/core/test_challenge.py @@ -1076,6 +1076,162 @@ def test_updates_requirements(self, mock_api_constructor: MagicMock, *args, **kw mock_api.post.assert_not_called() mock_api.delete.assert_not_called() + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) + @mock.patch("ctfcli.core.challenge.API") + def test_updates_module_by_name(self, mock_api_constructor: MagicMock, *args, **kwargs): + challenge = Challenge(self.minimal_challenge, {"module": "Test Module"}) + + def mock_get(*args, **kwargs): + path = args[0] + + if path == "/api/v1/modules": + mock_response = MagicMock() + mock_response.json.return_value = { + "success": True, + "data": [ + {"id": 1, "name": "Other Module", "description": None}, + {"id": 2, "name": "Test Module", "description": None}, + ], + } + return mock_response + + return MagicMock() + + mock_api: MagicMock = mock_api_constructor.return_value + mock_api.get.side_effect = mock_get + + challenge.sync(ignore=["files"]) + + mock_api.get.assert_has_calls([call("/api/v1/modules")]) + mock_api.patch.assert_has_calls( + [ + call("/api/v1/challenges/1", json={"module_id": 2}), + call().raise_for_status(), + ] + ) + + # the module already exists, so it should not be created + mock_api.post.assert_not_called() + + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) + @mock.patch("ctfcli.core.challenge.click.secho") + @mock.patch("ctfcli.core.challenge.API") + def test_creates_module_if_missing(self, mock_api_constructor: MagicMock, *args, **kwargs): + challenge = Challenge(self.minimal_challenge, {"module": "New Module"}) + + def mock_get(*args, **kwargs): + path = args[0] + + if path == "/api/v1/modules": + mock_response = MagicMock() + mock_response.json.return_value = {"success": True, "data": []} + return mock_response + + return MagicMock() + + def mock_post(*args, **kwargs): + path = args[0] + + if path == "/api/v1/modules": + mock_response = MagicMock() + mock_response.json.return_value = { + "success": True, + "data": {"id": 7, "name": "New Module", "description": None}, + } + return mock_response + + return MagicMock() + + mock_api: MagicMock = mock_api_constructor.return_value + mock_api.get.side_effect = mock_get + mock_api.post.side_effect = mock_post + + challenge.sync(ignore=["files"]) + + mock_api.get.assert_has_calls([call("/api/v1/modules")]) + mock_api.post.assert_has_calls([call("/api/v1/modules", json={"name": "New Module"})]) + mock_api.patch.assert_has_calls( + [ + call("/api/v1/challenges/1", json={"module_id": 7}), + call().raise_for_status(), + ] + ) + + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) + @mock.patch("ctfcli.core.challenge.API") + def test_numeric_module_treated_as_name(self, mock_api_constructor: MagicMock, *args, **kwargs): + # a numeric module (YAML loads "42" as an int) is always treated as a name, + # not a module id - it is resolved (and created if missing) by name + challenge = Challenge(self.minimal_challenge, {"module": 42}) + + def mock_get(*args, **kwargs): + path = args[0] + + if path == "/api/v1/modules": + mock_response = MagicMock() + mock_response.json.return_value = {"success": True, "data": []} + return mock_response + + return MagicMock() + + def mock_post(*args, **kwargs): + path = args[0] + + if path == "/api/v1/modules": + mock_response = MagicMock() + mock_response.json.return_value = { + "success": True, + "data": {"id": 7, "name": "42", "description": None}, + } + return mock_response + + return MagicMock() + + mock_api: MagicMock = mock_api_constructor.return_value + mock_api.get.side_effect = mock_get + mock_api.post.side_effect = mock_post + + challenge.sync(ignore=["files"]) + + # the numeric name is resolved against the remote and created as a string name + mock_api.get.assert_has_calls([call("/api/v1/modules")]) + mock_api.post.assert_has_calls([call("/api/v1/modules", json={"name": "42"})]) + mock_api.patch.assert_has_calls( + [ + call("/api/v1/challenges/1", json={"module_id": 7}), + call().raise_for_status(), + ] + ) + + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) + @mock.patch("ctfcli.core.challenge.API") + def test_removes_module_with_explicit_null(self, mock_api_constructor: MagicMock, *args, **kwargs): + challenge = Challenge(self.minimal_challenge, {"module": None}) + + mock_api: MagicMock = mock_api_constructor.return_value + challenge.sync(ignore=["files"]) + + mock_api.patch.assert_has_calls( + [ + call("/api/v1/challenges/1", json={"module_id": None}), + call().raise_for_status(), + ] + ) + mock_api.post.assert_not_called() + + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) + @mock.patch("ctfcli.core.challenge.API") + def test_does_not_touch_module_if_absent(self, mock_api_constructor: MagicMock, *args, **kwargs): + challenge = Challenge(self.minimal_challenge) + + mock_api: MagicMock = mock_api_constructor.return_value + challenge.sync(ignore=["files"]) + + # without a module key in challenge.yml the remote module assignment should be left untouched + self.assertNotIn(call("/api/v1/modules"), mock_api.get.call_args_list) + for patch_call in mock_api.patch.call_args_list: + self.assertNotIn("module_id", patch_call.kwargs.get("json", {})) + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) @mock.patch("ctfcli.core.challenge.click.secho") @mock.patch("ctfcli.core.challenge.API") @@ -1381,6 +1537,7 @@ def test_does_not_update_ignored_attributes(self): "files", "hints", "requirements", + "module", "solution", # fmt: on ] @@ -1466,6 +1623,9 @@ def test_does_not_update_ignored_attributes(self): if p in ["flags", "topics", "tags", "files", "hints", "requirements"]: challenge[p] = ["new-value"] + if p == "module": + challenge[p] = "new-value" + if p == "solution": challenge[p] = "challenge.yml" @@ -1643,6 +1803,59 @@ def mock_post(*args, **kwargs): any_order=True, ) + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) + @mock.patch("ctfcli.core.challenge.click.secho") + @mock.patch("ctfcli.core.challenge.API") + def test_creates_challenge_with_module(self, mock_api_constructor: MagicMock, *args, **kwargs): + challenge = Challenge(self.minimal_challenge, {"module": "New Module"}) + + def mock_get(*args, **kwargs): + path = args[0] + + if path == "/api/v1/modules": + mock_response = MagicMock() + mock_response.json.return_value = {"success": True, "data": []} + return mock_response + + return MagicMock() + + def mock_post(*args, **kwargs): + path = args[0] + + if path == "/api/v1/challenges": + mock_response = MagicMock() + mock_response.json.return_value = {"success": True, "data": {"id": 3}} + return mock_response + + if path == "/api/v1/modules": + mock_response = MagicMock() + mock_response.json.return_value = { + "success": True, + "data": {"id": 7, "name": "New Module", "description": None}, + } + return mock_response + + return MagicMock() + + mock_api: MagicMock = mock_api_constructor.return_value + mock_api.get.side_effect = mock_get + mock_api.post.side_effect = mock_post + + challenge.create() + + mock_api.post.assert_has_calls( + [ + call("/api/v1/challenges", json=ANY), + call("/api/v1/modules", json={"name": "New Module"}), + ] + ) + mock_api.patch.assert_has_calls( + [ + call("/api/v1/challenges/3", json={"module_id": 7}), + call().raise_for_status(), + ] + ) + @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) @mock.patch("ctfcli.core.challenge.API") def test_exits_if_files_do_not_exist(self, mock_api_constructor: MagicMock, *args, **kwargs): @@ -1662,7 +1875,7 @@ def test_does_not_set_ignored_attributes(self): # fmt:off properties = [ "value", "category", "description", "attribution", "attempts", "connection_info", "scheduled_at", "state", # simple types # noqa: E501 - "extra", "flags", "topics", "tags", "files", "hints", "requirements", "solution" # complex types + "extra", "flags", "topics", "tags", "files", "hints", "requirements", "module", "solution" # complex types ] # fmt:on @@ -1728,6 +1941,9 @@ def test_does_not_set_ignored_attributes(self): if p in ["flags", "topics", "tags", "files", "hints", "requirements"]: challenge[p] = ["new-value"] + if p == "module": + challenge[p] = "new-value" + if p == "solution": challenge[p] = "challenge.yml" @@ -2174,6 +2390,7 @@ def test_normalize_fetches_and_normalizes_challenge(self, mock_api_constructor: "topics": ["topic-1", "topic-2"], "next": None, "scheduled_at": None, + "module": None, "requirements": {"prerequisites": ["First Test Challenge", "Other Test Challenge"], "anonymize": False}, "extra": { "initial": 100, @@ -2184,6 +2401,53 @@ def test_normalize_fetches_and_normalizes_challenge(self, mock_api_constructor: normalized_data, ) + @mock.patch("ctfcli.core.challenge.API") + def test_normalize_resolves_module_name(self, mock_api_constructor: MagicMock): + mock_api: MagicMock = mock_api_constructor.return_value + + def mock_get(*args, **kwargs): + path = args[0] + + if path == "/api/v1/modules/5": + mock_response = MagicMock() + mock_response.json.return_value = { + "success": True, + "data": {"id": 5, "name": "Test Module", "description": None}, + } + return mock_response + + return self.mock_get(*args, **kwargs) + + mock_api.get.side_effect = mock_get + + challenge = Challenge(self.full_challenge) + challenge.challenge_id = 3 + + normalized_data = challenge._normalize_challenge( + { + "name": "Test Challenge", + "description": "Test Description", + "max_attempts": 5, + "module_id": 5, + } + ) + + self.assertEqual("Test Module", normalized_data["module"]) + + @mock.patch("ctfcli.core.challenge.API") + def test_compare_challenge_module(self, mock_api_constructor: MagicMock): + challenge = Challenge(self.full_challenge) + challenge.challenge_id = 3 + + # modules are compared by name + self.assertTrue(challenge._compare_challenge_module("Test Module", "Test Module")) + self.assertTrue(challenge._compare_challenge_module(None, None)) + self.assertFalse(challenge._compare_challenge_module("Other Module", "Test Module")) + + # a numeric name (loaded from YAML as an int) is coerced to a string + self.assertTrue(challenge._compare_challenge_module(42, "42")) + self.assertFalse(challenge._compare_challenge_module(42, "Test Module")) + @mock.patch("ctfcli.core.challenge.API") def test_verify_checks_if_challenge_is_the_same(self, mock_api_constructor: MagicMock): mock_api: MagicMock = mock_api_constructor.return_value diff --git a/tests/core/test_page.py b/tests/core/test_page.py index e522f69..f3be143 100644 --- a/tests/core/test_page.py +++ b/tests/core/test_page.py @@ -437,6 +437,49 @@ def test_cannot_upload_local_page_if_remote_exists(self, mock_api_constructor: M "Cannot push page 'html-page.html' - remote version exists. Use sync instead.", str(e.exception) ) + @mock.patch("ctfcli.core.config.Path.cwd", return_value=minimal_challenge_cwd) + @mock.patch("ctfcli.core.page.click.secho") + @mock.patch("ctfcli.core.page.API") + def test_force_push_syncs_local_page_if_remote_exists( + self, mock_api_constructor: MagicMock, mock_secho: MagicMock, *args, **kwargs + ): + mock_api = mock_api_constructor.return_value + + mock_page_data = { + "format": "html", + "files": [], + "draft": False, + "title": "HTML Page", + "id": 1, + "content": "

Hello World!

", + "auth_required": False, + "hidden": False, + "route": "html-page", + } + + mock_api.get.return_value.json.return_value = { + "success": True, + "data": [mock_page_data], + } + + page_path = BASE_DIR / "fixtures" / "challenges" / "pages" / "html-page.html" + page = Page(page_path=page_path) + page.push(force=True) + + expected_page_payload = { + "route": "html-page", + "title": "HTML Page", + "content": "

Hello World!

", + "draft": False, + "hidden": False, + "auth_required": False, + "format": "html", + } + + mock_secho.assert_called_once_with("Syncing existing page instead (because of --force)", fg="yellow") + mock_api.post.assert_not_called() + mock_api.patch.assert_called_once_with("/api/v1/pages/1", json=expected_page_payload) + def test_get_format(self): formats = { ".md": "markdown",