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))