diff --git a/Makefile b/Makefile index cd23b00..d9e1dc4 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ -.PHONY: all -.IGNORE: lint format +.PHONY: lint format test clean +.IGNORE: format lint: ruff format --check . diff --git a/ctfcli/cli/media.py b/ctfcli/cli/media.py index e142f0b..7dc14ec 100644 --- a/ctfcli/cli/media.py +++ b/ctfcli/cli/media.py @@ -16,21 +16,18 @@ def add(self, path): api = API() filename = os.path.basename(path) - new_file = (filename, open(path, mode="rb")) location = f"media/{filename}" file_payload = { "type": "page", "location": location, } - # Specifically use data= here to send multipart/form-data - r = api.post("/api/v1/files", files={"file": new_file}, data=file_payload) - r.raise_for_status() - resp = r.json() - server_location = resp["data"][0]["location"] - - # Close the file handle - new_file[1].close() + with open(path, mode="rb") as file_handle: + # Specifically use data= here to send multipart/form-data + r = api.post("/api/v1/files", files={"file": (filename, file_handle)}, data=file_payload) + r.raise_for_status() + resp = r.json() + server_location = resp["data"][0]["location"] config.config.set("media", location, f"/files/{server_location}") diff --git a/ctfcli/core/api.py b/ctfcli/core/api.py index a7c19a4..3bb21c1 100644 --- a/ctfcli/core/api.py +++ b/ctfcli/core/api.py @@ -1,4 +1,4 @@ -from typing import Mapping +from collections.abc import Mapping from urllib.parse import urljoin from requests import Session @@ -76,12 +76,12 @@ def request(self, method, url, data=None, files=None, *args, **kwargs): headers = dict(headers) headers["Content-Type"] = multipart.content_type - return super(API, self).request( + return super().request( method, url, + *args, data=multipart, headers=headers, - *args, **kwargs, ) @@ -89,15 +89,15 @@ def request(self, method, url, data=None, files=None, *args, **kwargs): # modify the headers here instead of using self.headers because we don't want to # override the multipart/form-data case above if data is None and files is None: - if kwargs.get("headers", None) is None: + if kwargs.get("headers") is None: kwargs["headers"] = {} kwargs["headers"]["Content-Type"] = "application/json" - return super(API, self).request( + return super().request( method, url, + *args, data=data, files=files, - *args, **kwargs, ) diff --git a/ctfcli/core/challenge.py b/ctfcli/core/challenge.py index f1754eb..0226c51 100644 --- a/ctfcli/core/challenge.py +++ b/ctfcli/core/challenge.py @@ -1,7 +1,5 @@ import logging import re -import subprocess -from datetime import datetime, timezone from os import PathLike from pathlib import Path from typing import Any @@ -16,12 +14,17 @@ ChallengeException, InvalidChallengeDefinition, InvalidChallengeFile, - LintException, RemoteChallengeNotFound, ) -from ctfcli.core.image import Image -from ctfcli.utils.hashing import hash_file -from ctfcli.utils.tools import strings +from ctfcli.core.lint import lint_challenge +from ctfcli.core.properties import ( + NOT_PULLED, + PROPERTIES, + PropertyContext, + get_property, + has_property, + operation_order, +) log = logging.getLogger("ctfcli.core.challenge") @@ -43,51 +46,10 @@ def str_presenter(dumper, data): class Challenge(dict): - key_order = [ - "name", - "author", - "category", - "description", - "attribution", - "value", - "type", - "extra", - "image", - "protocol", - "host", - "connection_info", - "healthcheck", - "solution", - "attempts", - "logic", - "flags", - "files", - "topics", - "tags", - "files", - "hints", - "requirements", - "next", - "scheduled_at", - "module", - "state", - "version", - ] - - keys_with_newline = [ - "extra", - "image", - "attempts", - "flags", - "topics", - "tags", - "files", - "hints", - "requirements", - "scheduled_at", - "state", - "version", - ] + # The order of the keys in challenge.yml, as well as the create/sync/mirror/verify + # behavior of each of them, is defined by the property registry (ctfcli.core.properties) + key_order = [p.key for p in PROPERTIES] + keys_with_newline = [p.key for p in PROPERTIES if p.newline_before] @staticmethod def load_installed_challenge(challenge_id) -> dict: @@ -119,96 +81,7 @@ def load_installed_challenges() -> list: @staticmethod def is_default_challenge_property(key: str, value: Any) -> bool: - if key == "connection_info" and value is None: - return True - - if key == "attempts" and value == 0: - return True - - if key == "state" and value == "visible": - return True - - if key == "type" and value == "standard": - return True - - if key in ["tags", "hints", "topics", "requirements", "files"] and value == []: - return True - - if key == "requirements" and value == {"prerequisites": [], "anonymize": False}: - return True - - 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 - 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: - 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 = 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 " - 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 - - @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 cls._datetime_from_iso(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) + return has_property(key) and get_property(key).is_default(value) @staticmethod def clone(config, remote_challenge): @@ -291,59 +164,11 @@ def __init__(self, challenge_yml: str | PathLike, overrides=None): self._api = None # Assign an image if the challenge provides one, otherwise this will be set to None - self.image = self._process_challenge_image(self.get("image")) + self.image = get_property("image").resolve(PropertyContext(self)) def __str__(self): return self["name"] - def _process_challenge_image(self, challenge_image: str | None) -> Image | None: - if not challenge_image: - return None - - # Check if challenge_image is explicitly marked with registry:// prefix - if challenge_image.startswith("registry://"): - challenge_image = challenge_image.replace("registry://", "") - return Image(challenge_image) - - # Check if it's a library image - if challenge_image.startswith("library/"): - return Image(f"docker.io/{challenge_image}") - - # Check if it defines a known registry - known_registries = [ - "docker.io", - "gcr.io", - "ecr.aws", - "ghcr.io", - "azurecr.io", - "registry.digitalocean.com", - "registry.gitlab.com", - "registry.ctfd.io", - ] - for registry in known_registries: - if registry in challenge_image: - return Image(challenge_image) - - # Check if it's a path to dockerfile to be built - if (self.challenge_directory / challenge_image / "Dockerfile").exists(): - return Image(slugify(self["name"]), self.challenge_directory / self["image"]) - - # Check if it's a local pre-built image - if ( - subprocess.call( - ["docker", "inspect", challenge_image], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - == 0 - ): - return Image(challenge_image) - - # If the image is set, but we fail to determine whether it's local / remote - raise an exception - raise InvalidChallengeFile( - f"Challenge file at {self.challenge_file_path} defines an image, but it couldn't be resolved" - ) - def _load_challenge_id(self): remote_challenges = self.load_installed_challenges() if not remote_challenges: @@ -359,671 +184,24 @@ def _load_challenge_id(self): if self.challenge_id is None: raise RemoteChallengeNotFound(f"Could not load remote challenge with name '{self['name']}'") - def _validate_files(self): - files = self.get("files") or [] - for challenge_file in files: - if not (self.challenge_directory / challenge_file).exists(): - raise InvalidChallengeFile(f"File {challenge_file} could not be loaded") - - def _get_initial_challenge_payload(self, ignore: tuple[str] = ()) -> dict: - challenge = self - challenge_payload = { - "name": self["name"], - "category": self.get("category", ""), - "description": self.get("description", ""), - "attribution": self.get("attribution", ""), - "type": self.get("type", "standard"), - # Hide the challenge for the duration of the sync / creation - "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"] - - 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 "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" - - if "extra" not in ignore: - challenge_payload = {**challenge_payload, **challenge.get("extra", {})} - - return challenge_payload - - def _delete_existing_flags(self): - remote_flags = self.api.get("/api/v1/flags").json()["data"] - for flag in remote_flags: - if flag["challenge_id"] == self.challenge_id: - r = self.api.delete(f"/api/v1/flags/{flag['id']}") - r.raise_for_status() - - def _create_flags(self): - for flag in self["flags"]: - if type(flag) == str: - flag_payload = { - "content": flag, - "type": "static", - "challenge_id": self.challenge_id, - } - else: - flag_payload = {**flag, "challenge_id": self.challenge_id} - - r = self.api.post("/api/v1/flags", json=flag_payload) - r.raise_for_status() - - def _delete_existing_topics(self): - remote_topics = self.api.get(f"/api/v1/challenges/{self.challenge_id}/topics").json()["data"] - for topic in remote_topics: - r = self.api.delete(f"/api/v1/topics?type=challenge&target_id={topic['id']}") - r.raise_for_status() - - def _create_topics(self): - for topic in self["topics"]: - r = self.api.post( - "/api/v1/topics", - json={ - "value": topic, - "type": "challenge", - "challenge_id": self.challenge_id, - }, - ) - r.raise_for_status() - - def _delete_existing_tags(self): - remote_tags = self.api.get("/api/v1/tags").json()["data"] - for tag in remote_tags: - if tag["challenge_id"] == self.challenge_id: - r = self.api.delete(f"/api/v1/tags/{tag['id']}") - r.raise_for_status() - - def _create_tags(self): - for tag in self["tags"]: - r = self.api.post( - "/api/v1/tags", - json={"challenge_id": self.challenge_id, "value": tag}, - ) - r.raise_for_status() - - def _delete_file(self, remote_location: str): - remote_files = self.api.get("/api/v1/files?type=challenge").json()["data"] - - for remote_file in remote_files: - if remote_file["location"] == remote_location: - r = self.api.delete(f"/api/v1/files/{remote_file['id']}") - r.raise_for_status() - - def _create_file(self, local_path: Path): - new_file = (local_path.name, open(local_path, mode="rb")) - file_payload = {"challenge_id": self.challenge_id, "type": "challenge"} - - # Specifically use data= here to send multipart/form-data - r = self.api.post("/api/v1/files", files={"file": new_file}, data=file_payload) - r.raise_for_status() - - # Close the file handle - new_file[1].close() - - def _create_all_files(self): - new_files = [] - for challenge_file in self["files"]: - file_path = self.challenge_directory / challenge_file - new_files.append(("file", (file_path.name, file_path.open("rb")))) - - files_payload = {"challenge_id": self.challenge_id, "type": "challenge"} - - # Specifically use data= here to send multipart/form-data - r = self.api.post("/api/v1/files", files=new_files, data=files_payload) - r.raise_for_status() - - # Close the file handles - for file_payload in new_files: - file_payload[1][1].close() - - def _delete_existing_hints(self): - remote_hints = self.api.get("/api/v1/hints").json()["data"] - for hint in remote_hints: - if hint["challenge_id"] == self.challenge_id: - r = self.api.delete(f"/api/v1/hints/{hint['id']}") - r.raise_for_status() - - def _create_hints(self): - key_to_id = {} - target_hints = {} - - # Pass 1: create all hints; hints with requirements get blank content initially - # to prevent content from being exposed before prerequisites are enforced - for idx, hint in enumerate(self["hints"]): - if type(hint) == str: - hint_payload = { - "content": hint, - "title": "", - "cost": 0, - "challenge_id": self.challenge_id, - } - key = None - else: - has_requirements = bool(hint.get("requirements")) - hint_payload = { - "content": "" if has_requirements else hint["content"], - "title": hint.get("title", ""), - "cost": hint.get("cost", 0), - "challenge_id": self.challenge_id, - } - key = hint.get("key") - - r = self.api.post("/api/v1/hints", json=hint_payload) - r.raise_for_status() - - # Store IDs for processing later - target_hints[idx] = r.json()["data"]["id"] - if key is not None: - key_to_id[key] = r.json()["data"]["id"] - - # Pass 2: set requirements - for idx, hint in enumerate(self["hints"]): - if type(hint) == str: - continue - requirements = hint.get("requirements", []) - if not requirements: - continue - - prerequisite_ids = [] - for req_key in requirements: - if req_key in key_to_id: - preq_hint_id = key_to_id[req_key] - prerequisite_ids.append(preq_hint_id) - else: - click.secho( - f'Hint key "{req_key}" not found. Skipping invalid hint requirement.', - fg="yellow", - ) - - hint_id = target_hints[idx] - - # Pass 3: fill in real content - if prerequisite_ids: - r = self.api.patch( - f"/api/v1/hints/{hint_id}", - json={"requirements": {"prerequisites": prerequisite_ids}}, - ) - r.raise_for_status() - - # Now safe to set the real content - r = self.api.patch( - f"/api/v1/hints/{hint_id}", - json={"content": hint["content"]}, - ) - r.raise_for_status() - - def _parse_solution_definition(self) -> tuple[str, str] | None: - solution = self.get("solution", None) - if not solution: - return None - - if type(solution) == str: - return solution, "hidden" - - if type(solution) != dict: - click.secho( - "The solution field must be a string path or an object with path and state", - fg="red", - ) - return None - - solution_path = solution.get("path") - if type(solution_path) != str or not solution_path: - click.secho("The solution object must define a non-empty string path field", fg="red") - return None - - solution_state = solution.get("state", "hidden") - if type(solution_state) != str or solution_state not in ["hidden", "visible", "solved"]: - click.secho("The solution state must be one of: hidden, visible, solved", fg="red") - return None - - return solution_path, solution_state - - def _resolve_solution_path(self) -> tuple[Path, str] | None: - parsed_solution = self._parse_solution_definition() - if not parsed_solution: - return None - - solution_path_string, solution_state = parsed_solution - solution_path = self.challenge_directory / solution_path_string - if not solution_path.is_file(): - click.secho( - f"Solution file '{solution_path_string}' specified, but not found at {solution_path}", - fg="red", - ) - return None - - return solution_path, solution_state - - def _delete_existing_solution(self): - remote_solutions = self.api.get("/api/v1/solutions").json()["data"] - for solution in remote_solutions: - if solution["challenge_id"] == self.challenge_id: - r = self.api.delete(f"/api/v1/solutions/{solution['id']}") - r.raise_for_status() - - def _get_existing_solution_id(self) -> int | None: - r = self.api.get("/api/v1/solutions") - r.raise_for_status() - remote_solutions = r.json().get("data") or [] - for solution in remote_solutions: - if solution["challenge_id"] == self.challenge_id: - return solution["id"] - return None - - def _create_solution(self): - resolved_solution = self._resolve_solution_path() - if not resolved_solution: - return - solution_path, solution_state = resolved_solution - - solution_id = self._get_existing_solution_id() - if solution_id is None: - solution_payload_create = {"challenge_id": self.challenge_id, "state": solution_state, "content": ""} - - r = self.api.post("/api/v1/solutions", json=solution_payload_create) - r.raise_for_status() - solution_id = r.json()["data"]["id"] - else: - # Keep solution state in sync and clear stale content before rebuilding references. - r = self.api.patch( - f"/api/v1/solutions/{solution_id}", - json={"state": solution_state, "content": ""}, - ) - r.raise_for_status() - - with solution_path.open("r") as solution_file: - content = solution_file.read() - - # Find all images in the content (markdown format; ignore html format) - # Markdown format: ![alt text](image_url) - # Returns tuples: (full_match, alt_text, image_path) - markdown_images = re.findall(r"(!\[([^\]]*)\]\(([^\)]+)\))", content) - - # Find all snippet includes (MkDocs style: --8<-- "filename") - # Returns tuples: (full_match, filename) - snippet_includes = re.findall(r'(--8<--\s+["\']([^"\']+)["\'])', content) - - for mdx, alt, path in markdown_images: - local_path = solution_path.parent / path - new_file = (local_path.name, open(solution_path.parent / path, mode="rb")) - file_payload = { - "type": "solution", - "solution_id": solution_id, - } - - # Specifically use data= here to send multipart/form-data - r = self.api.post("/api/v1/files", files={"file": new_file}, data=file_payload) - r.raise_for_status() - resp = r.json() - server_location = resp["data"][0]["location"] - content = content.replace(mdx, f"![{alt}](/files/{server_location})") - - # Process snippet includes (--8<-- "filename") - for full_match, filename in snippet_includes: - snippet_file_path = solution_path.parent / filename - if snippet_file_path.exists(): - with snippet_file_path.open("r") as snippet_file: - snippet_content = snippet_file.read() - # Replace the --8<-- directive with the actual file content - content = content.replace(full_match, snippet_content) - else: - log.warning(f"Snippet file not found: {filename}") - - solution_payload_patch = {"content": content} - r = self.api.patch(f"/api/v1/solutions/{solution_id}", json=solution_payload_patch) - r.raise_for_status() - - def _set_required_challenges(self): - remote_challenges = self.load_installed_challenges() - required_challenges = [] - anonymize = False - if type(self["requirements"]) == dict: - rc = self["requirements"].get("prerequisites", []) - anonymize = self["requirements"].get("anonymize", False) - else: - rc = self["requirements"] - - for required_challenge in rc: - if type(required_challenge) == str: - # requirement by name - # find the challenge id from installed challenges - found = False - for remote_challenge in remote_challenges: - if remote_challenge["name"] == required_challenge: - required_challenges.append(remote_challenge["id"]) - found = True - break - if found is False: - click.secho( - f'Challenge id cannot be found. Skipping invalid requirement name "{required_challenge}".', - fg="yellow", - ) - - elif type(required_challenge) == int: - # requirement by challenge id - # trust it and use it directly - required_challenges.append(required_challenge) - - required_challenge_ids = list(set(required_challenges)) - - if self.challenge_id in required_challenge_ids: - click.secho( - "Challenge cannot require itself. Skipping invalid requirement.", - fg="yellow", - ) - required_challenges.remove(self.challenge_id) - required_challenges.sort() - - requirements_payload = { - "requirements": { - "prerequisites": required_challenges, - "anonymize": anonymize, - } - } - r = self.api.patch(f"/api/v1/challenges/{self.challenge_id}", json=requirements_payload) - r.raise_for_status() - - def _set_next(self, _next): - if type(_next) == str: - # nid by name - # find the challenge id from installed challenges - remote_challenges = self.load_installed_challenges() - for remote_challenge in remote_challenges: - if remote_challenge["name"] == _next: - _next = remote_challenge["id"] - break - if type(_next) == str: - click.secho( - "Challenge cannot find next challenge. Maybe it is invalid name or id. It will be cleared.", - fg="yellow", - ) - _next = None - elif type(_next) == int and _next > 0: - # nid by challenge id - # trust it and use it directly - _next = remote_challenge["id"] - else: - _next = None - - if self.challenge_id == _next: - click.secho( - "Challenge cannot set next challenge itself. Skipping invalid next challenge.", - fg="yellow", - ) - _next = None - - next_payload = {"next_id": _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() - - def normalize_requirements(requirements): - normalized = [] - for r in requirements: - if type(r) == int: - for remote_challenge in remote_challenges: - if remote_challenge["id"] == r: - normalized.append(remote_challenge["name"]) - break - else: - normalized.append(r) - - return normalized - - nr1 = normalize_requirements(r1) - nr1.sort() - nr2 = normalize_requirements(r2) - nr2.sort() - return nr1 == nr2 - - # Compare next challenges, will resolve all IDs to names - def _compare_challenge_next(self, r1: str | int | None, r2: str | int | None) -> bool: - def normalize_next(r): - normalized = None - if type(r) == int: - if r > 0: - remote_challenge = self.load_installed_challenge(r) - if remote_challenge["id"] == r: - normalized = remote_challenge["name"] - else: - normalized = r - - return normalized - - 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.) + # that have to be fetched separately (e.g., flags, hints, requirements, etc.) # Note: files won't be included for two reasons: # 1. To avoid downloading them unnecessarily, e.g., when they are ignored # 2. Because it's dependent on the implementation whether to save them (mirror) or just compare (verify) - def _normalize_challenge(self, challenge_data: dict[str, Any]): - challenge = {} - - copy_keys = [ - "name", - "category", - "attribution", - "value", - "type", - "state", - "connection_info", - "logic", - ] - for key in copy_keys: - 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"]: - 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() - flags = r.json()["data"] - challenge["flags"] = [ - ( - f["content"] - if f["type"] == "static" and (f["data"] is None or f["data"] == "") - else { - "content": f["content"].strip().replace("\r\n", "\n"), - "type": f["type"], - "data": f["data"], - } - ) - for f in flags - ] - - # Add tags - r = self.api.get(f"/api/v1/challenges/{self.challenge_id}/tags") - r.raise_for_status() - tags = r.json()["data"] - challenge["tags"] = [t["value"] for t in tags] - - # Add hints - r = self.api.get(f"/api/v1/challenges/{self.challenge_id}/hints") - r.raise_for_status() - hints = r.json()["data"] - - # Determine which hints are part of a requirements chain: - # either they have prerequisites themselves, or are referenced as a prerequisite - referenced_ids = set() - for h in hints: - for pid in (h.get("requirements") or {}).get("prerequisites", []): - referenced_ids.add(pid) - hints_with_requirements = {h["id"] for h in hints if (h.get("requirements") or {}).get("prerequisites")} - needs_key = referenced_ids | hints_with_requirements - - id_to_key = {h["id"]: f"hint-{h['id']}" for h in hints} - normalized_hints = [] - for h in hints: - prerequisites = (h.get("requirements") or {}).get("prerequisites", []) - has_requirements = bool(prerequisites) - has_cost = h["cost"] > 0 - has_title = bool(h.get("title", "")) - in_requirements_chain = h["id"] in needs_key - - if not has_cost and not has_requirements and not has_title and not in_requirements_chain: - normalized_hints.append(h["content"]) - else: - hint_dict = {"content": h["content"]} - if in_requirements_chain: - hint_dict["key"] = id_to_key[h["id"]] - if has_title: - hint_dict["title"] = h["title"] - if has_cost: - hint_dict["cost"] = h["cost"] - if has_requirements: - hint_dict["requirements"] = [id_to_key[pid] for pid in prerequisites if pid in id_to_key] - normalized_hints.append(hint_dict) - - challenge["hints"] = normalized_hints - - # Add topics - r = self.api.get(f"/api/v1/challenges/{self.challenge_id}/topics") - r.raise_for_status() - topics = r.json()["data"] - challenge["topics"] = [t["value"] for t in topics] + def _normalize_challenge(self, challenge_data: dict[str, Any]) -> dict[str, Any]: + ctx = PropertyContext(self) - # Add requirements - r = self.api.get(f"/api/v1/challenges/{self.challenge_id}/requirements") - r.raise_for_status() - requirements = (r.json().get("data") or {}).get("prerequisites", []) - challenge["requirements"] = {"prerequisites": [], "anonymize": False} - if len(requirements) > 0: - # Prefer challenge names over IDs - r2 = self.api.get("/api/v1/challenges?view=admin") - 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) - - # Add next - nid = challenge_data.get("next_id") - if nid: - # Prefer challenge names over IDs - r = self.api.get(f"/api/v1/challenges/{nid}") - r.raise_for_status() - challenge["next"] = (r.json().get("data") or {}).get("name", None) - 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 + challenge = {} + for prop in PROPERTIES: + value = prop.pull(ctx, challenge_data) + if value is not NOT_PULLED: + challenge[prop.key] = value return challenge - # Create a dictionary of remote files in { basename: {"url": "", "location": ""} } format - def _normalize_remote_files(self, remote_files: list[str]) -> dict[str, dict[str, str]]: - normalized = {} - for f in remote_files: - file_parts = f.split("?token=")[0].split("/") - normalized[file_parts[-1]] = { - "url": f, - "location": f"{file_parts[-2]}/{file_parts[-1]}", - } - - return normalized - - # Create a dictionary of sha1sums in { location: sha1sum } format - def _get_files_sha1sums(self) -> dict[str, str]: - r = self.api.get("/api/v1/files?type=challenge") - r.raise_for_status() - return {f["location"]: f.get("sha1sum", None) for f in r.json()["data"]} - def sync(self, ignore: tuple[str] = ()) -> None: - challenge = self - if "name" in ignore: click.secho( "Attribute 'name' cannot be ignored when syncing a challenge", @@ -1033,132 +211,41 @@ def sync(self, ignore: tuple[str] = ()) -> None: if not self.get("name"): raise InvalidChallengeFile("Challenge does not provide a name") - if challenge.get("files", False) and "files" not in ignore: - # _validate_files will raise if file is not found - self._validate_files() + ctx = PropertyContext(self, ignore=ignore) - challenge_payload = self._get_initial_challenge_payload(ignore=ignore) + if self.get("files", False) and "files" not in ignore: + # validate will raise if a file is not found + get_property("files").validate(ctx) self._load_challenge_id() - remote_challenge = self.load_installed_challenge(self.challenge_id) - - # if value, category, type or description are ignored, revert them to the remote state in the initial payload - reset_properties_if_ignored = [ - "value", - "category", - "type", - "description", - "attribution", - ] - for p in reset_properties_if_ignored: - if p in ignore: - challenge_payload[p] = remote_challenge[p] - - # Update simple properties + ctx.remote_challenge = self.load_installed_challenge(self.challenge_id) + + # Update simple properties with a single PATCH, + # ignored properties are reverted to their remote values + challenge_payload = {} + for prop in PROPERTIES: + challenge_payload.update(prop.sync_payload(ctx)) + r = self.api.patch(f"/api/v1/challenges/{self.challenge_id}", json=challenge_payload) if r.ok is False: click.secho(f"Failed to sync challenge: ({r.status_code}) {r.text}", fg="red") r.raise_for_status() - # Update flags - if "flags" not in ignore: - self._delete_existing_flags() - if challenge.get("flags"): - self._create_flags() - - # Update topics - if "topics" not in ignore: - self._delete_existing_topics() - if challenge.get("topics"): - self._create_topics() - - # Update tags - if "tags" not in ignore: - self._delete_existing_tags() - if challenge.get("tags"): - self._create_tags() - - # Create / Upload files - if "files" not in ignore: - self["files"] = self.get("files") or [] - remote_challenge["files"] = remote_challenge.get("files") or [] - - # Get basenames of local files to compare against remote files - local_files = {f.split("/")[-1]: f for f in self["files"]} - remote_files = self._normalize_remote_files(remote_challenge["files"]) - - # Delete remote files which are no longer defined locally - for remote_file in remote_files: - if remote_file not in local_files: - self._delete_file(remote_files[remote_file]["location"]) - - # Only check for file changes if there are files to upload - if local_files: - sha1sums = self._get_files_sha1sums() - for local_file_name in local_files: - # Creating a new file - if local_file_name not in remote_files: - self._create_file(self.challenge_directory / local_files[local_file_name]) - continue - - # Updating an existing file - # sha1sum is present in CTFd 3.7+, use it instead of always re-uploading the file if possible - remote_file_sha1sum = sha1sums[remote_files[local_file_name]["location"]] - if remote_file_sha1sum is not None: - with open( - self.challenge_directory / local_files[local_file_name], - "rb", - ) as lf: - local_file_sha1sum = hash_file(lf) - - # Allow users to specify sha1sum in ignore to force reuploads - if "sha1sum" not in ignore: - if local_file_sha1sum == remote_file_sha1sum: - continue - - # if sha1sums are not present, or the hashes are different, re-upload the file - self._delete_file(remote_files[local_file_name]["location"]) - self._create_file(self.challenge_directory / local_files[local_file_name]) - - # Update hints - if "hints" not in ignore: - self._delete_existing_hints() - if challenge.get("hints"): - self._create_hints() - - # Update requirements - if challenge.get("requirements") and "requirements" not in ignore: - self._set_required_challenges() - - # Set next - _next = challenge.get("next", 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: - self._delete_existing_solution() - self._create_solution() + # Update properties stored in separate resources (flags, files, hints, etc.) + for prop in operation_order(): + prop.sync(ctx) make_challenge_visible = False # Bring back the challenge to be visible if: # 1. State is not ignored and set to visible, or defaults to visible if "state" not in ignore: - if challenge.get("state", "visible") == "visible": + if self.get("state", "visible") == "visible": make_challenge_visible = True # 2. State is ignored, but regardless of the local value, the remote state was visible else: - if remote_challenge.get("state") == "visible": + if ctx.remote_challenge.get("state") == "visible": make_challenge_visible = True if make_challenge_visible: @@ -1166,8 +253,6 @@ def sync(self, ignore: tuple[str] = ()) -> None: r.raise_for_status() def create(self, ignore: tuple[str] = ()) -> None: - challenge = self - for attr in ["name", "value"]: if attr in ignore: click.secho( @@ -1175,26 +260,21 @@ def create(self, ignore: tuple[str] = ()) -> None: fg="yellow", ) - if not challenge.get("name", False): + if not self.get("name", False): raise InvalidChallengeDefinition("Challenge does not provide a name") - if not challenge.get("value", False) and challenge.get("type", "standard") != "dynamic": + if not self.get("value", False) and self.get("type", "standard") != "dynamic": raise InvalidChallengeDefinition("Challenge does not provide a value") - if challenge.get("files", False) and "files" not in ignore: - # _validate_files will raise if file is not found - self._validate_files() + ctx = PropertyContext(self, ignore=ignore) - challenge_payload = self._get_initial_challenge_payload(ignore=ignore) + if self.get("files", False) and "files" not in ignore: + # validate will raise if a file is not found + get_property("files").validate(ctx) - # in the case of creation, value and type can't be ignored: - # value is required (unless the challenge is a dynamic value challenge), - # and the type will default to standard - # if category or description are ignored, set them to an empty string - reset_properties_if_ignored = ["category", "description", "attribution"] - for p in reset_properties_if_ignored: - if p in ignore: - challenge_payload[p] = "" + challenge_payload = {} + for prop in PROPERTIES: + challenge_payload.update(prop.create_payload(ctx)) r = self.api.post("/api/v1/challenges", json=challenge_payload) if r.ok is False: @@ -1203,329 +283,68 @@ def create(self, ignore: tuple[str] = ()) -> None: self.challenge_id = r.json()["data"]["id"] - # Create flags - if challenge.get("flags") and "flags" not in ignore: - self._create_flags() - - # Create topics - if challenge.get("topics") and "topics" not in ignore: - self._create_topics() - - # Create tags - if challenge.get("tags") and "tags" not in ignore: - self._create_tags() - - # Upload files - if challenge.get("files") and "files" not in ignore: - self._create_all_files() - - # Add hints - if challenge.get("hints") and "hints" not in ignore: - self._create_hints() - - # Add requirements - if challenge.get("requirements") and "requirements" not in ignore: - self._set_required_challenges() - - # Add next - _next = challenge.get("next", 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() + # Create properties stored in separate resources (flags, files, hints, etc.) + for prop in operation_order(): + prop.create(ctx) # Bring back the challenge if it's supposed to be visible # Either explicitly, or by assuming the default value (possibly because the state is ignored) - if challenge.get("state", "visible") == "visible" or "state" in ignore: + if self.get("state", "visible") == "visible" or "state" in ignore: r = self.api.patch(f"/api/v1/challenges/{self.challenge_id}", json={"state": "visible"}) r.raise_for_status() def lint(self, skip_hadolint=False, flag_format="flag{") -> bool: - challenge = self - - issues = {"fields": [], "dockerfile": [], "hadolint": [], "files": []} - - # Check if required fields are present - for field in [ - "name", - "author", - "category", - "description", - "attribution", - "value", - ]: - # value is allowed to be none if the challenge type is dynamic - if field == "value" and challenge.get("type") == "dynamic": - continue - - 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") - - # Check that Dockerfile exists and is EXPOSE'ing a port - if challenge.get("image") == ".": - dockerfile_path = self.challenge_directory / "Dockerfile" - has_dockerfile = dockerfile_path.is_file() - - if not has_dockerfile: - issues["dockerfile"].append("Dockerfile specified in 'image' field but no Dockerfile found") - - if has_dockerfile: - with open(dockerfile_path) as dockerfile: - dockerfile_source = dockerfile.read() - - if "EXPOSE" not in dockerfile_source: - issues["dockerfile"].append("Dockerfile is missing EXPOSE") - - if not skip_hadolint: - # Check Dockerfile with hadolint - hadolint = subprocess.run( - ["docker", "run", "--rm", "-i", "hadolint/hadolint"], - capture_output=True, - input=dockerfile_source.encode(), - ) - - if hadolint.returncode != 0: - issues["hadolint"].append(hadolint.stdout.decode()) - - else: - click.secho("Skipping Hadolint", fg="yellow") - - # Check that all files exist - files = self.get("files") or [] - for challenge_file in files: - challenge_file_path = self.challenge_directory / challenge_file - - if challenge_file_path.is_file() is False: - issues["files"].append( - f"Challenge file '{challenge_file}' specified, but not found at {challenge_file_path}" - ) - - # Check that the optional solution file exists - solution = self.get("solution", None) - if solution: - solution_file = None - solution_state = "hidden" - - if type(solution) == str: - solution_file = solution - elif type(solution) == dict: - solution_file = solution.get("path") - solution_state = solution.get("state", "hidden") - - if type(solution_state) != str or solution_state not in ["hidden", "visible", "solved"]: - issues["fields"].append("The solution state must be one of: hidden, visible, solved") - - else: - issues["fields"].append("The solution field must be a string path or an object with path and state") - - if type(solution_file) != str or not solution_file: - issues["fields"].append("The solution object must define a non-empty string path field") - else: - solution_file_path = self.challenge_directory / solution_file - if solution_file_path.is_file() is False: - issues["files"].append( - f"Solution file '{solution_file}' specified, but not found at {solution_file_path}" - ) - - # Check that files don't have a flag in them - for challenge_file in files: - challenge_file_path = self.challenge_directory / challenge_file - - if not challenge_file_path.exists(): - # The check for files present is above; this is only to look for flags in files that we do have - continue - - for s in strings(challenge_file_path): - if flag_format in s: - s = s.strip() - issues["files"].append(f"Potential flag found in distributed file '{challenge_file}':\n {s}") - - if any(messages for messages in issues.values() if len(messages) > 0): - raise LintException(issues=issues) - - return True + return lint_challenge(self, skip_hadolint=skip_hadolint, flag_format=flag_format) def mirror(self, files_directory_name: str = "dist", ignore: tuple[str] = ()) -> None: self._load_challenge_id() - remote_challenge = self.load_installed_challenge(self.challenge_id) - challenge = self._normalize_challenge(remote_challenge) - - remote_challenge["files"] = remote_challenge.get("files") or [] - challenge["files"] = challenge.get("files") or [] - - # Add files which are not handled in _normalize_challenge - if "files" not in ignore: - local_files = {Path(f).name: f for f in challenge["files"]} - - # Update files - for remote_file in remote_challenge["files"]: - # Get base file name - remote_file_name = remote_file.split("/")[-1].split("?token=")[0] - - # The file is only present on the remote - we have to download it, and assume a path - if remote_file_name not in local_files: - r = self.api.get(remote_file) - r.raise_for_status() - - # Ensure the directory for the challenge files exists - challenge_files_directory = self.challenge_directory / files_directory_name - challenge_files_directory.mkdir(parents=True, exist_ok=True) - - (challenge_files_directory / remote_file_name).write_bytes(r.content) - challenge["files"].append(f"{files_directory_name}/{remote_file_name}") + ctx = PropertyContext( + self, + ignore=ignore, + remote_challenge=self.load_installed_challenge(self.challenge_id), + options={"files_directory_name": files_directory_name}, + ) - # The file is already present in the challenge.yml - we know the desired path - else: - r = self.api.get(remote_file) - r.raise_for_status() - (self.challenge_directory / local_files[remote_file_name]).write_bytes(r.content) + normalized_challenge = self._normalize_challenge(ctx.remote_challenge) - # Soft-Delete files that are not present on the remote - # Remove them from challenge.yml but do not delete them from disk - remote_file_names = [f.split("/")[-1].split("?token=")[0] for f in remote_challenge["files"]] - challenge["files"] = [f for f in challenge["files"] if Path(f).name in remote_file_names] + # Some properties (e.g. files) are not part of the normalized challenge + # and amend it during mirror instead (e.g. downloading the files to disk) + for prop in PROPERTIES: + prop.mirror(ctx, normalized_challenge) - for key in challenge: + for key in normalized_challenge: if key not in ignore: - self[key] = challenge[key] + self[key] = normalized_challenge[key] self.save() def verify(self, ignore: tuple[str] = ()) -> bool: self._load_challenge_id() - challenge = self - remote_challenge = self.load_installed_challenge(self.challenge_id) - normalized_challenge = self._normalize_challenge(remote_challenge) - - remote_challenge["files"] = remote_challenge.get("files") or [] - challenge["files"] = challenge.get("files") or [] + ctx = PropertyContext(self, ignore=ignore, remote_challenge=self.load_installed_challenge(self.challenge_id)) + normalized_challenge = self._normalize_challenge(ctx.remote_challenge) for key in normalized_challenge: if key in ignore: continue # If challenge.yml doesn't have some property from the remote # Check if it's a default value that can be omitted - if key not in challenge: + if key not in self: if self.is_default_challenge_property(key, normalized_challenge[key]): continue - click.secho( - f"{key} is not in challenge.", - fg="yellow", - ) + click.secho(f"{key} is not in challenge.", fg="yellow") return False - if challenge[key] != normalized_challenge[key]: - if key == "requirements": - if type(challenge[key]) == dict: - cr = challenge[key]["prerequisites"] - ca = challenge[key].get("anonymize", False) - else: - cr = challenge[key] - ca = False - if ( - self._compare_challenge_requirements(cr, normalized_challenge[key]["prerequisites"]) - and ca == normalized_challenge[key]["anonymize"] - ): - continue - - 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 - - if key == "module" and self._compare_challenge_module(challenge[key], normalized_challenge[key]): - continue - - click.secho( - f"{key} comparison failed.", - fg="yellow", - ) + if not get_property(key).matches(ctx, self[key], normalized_challenge[key]): + click.secho(f"{key} comparison failed.", fg="yellow") return False - # Handle a special case for files, unless they are ignored - if "files" not in ignore: - # Check if files defined in challenge.yml are present - try: - self._validate_files() - local_files = {Path(f).name: f for f in challenge["files"]} - except InvalidChallengeFile: - click.secho( - "InvalidChallengeFile", - fg="yellow", - ) - return False - - remote_files = self._normalize_remote_files(remote_challenge["files"]) - # Check if there are no extra local files - for local_file in local_files: - if local_file not in remote_files: - click.secho( - f"{local_file} is not in remote challenge.", - fg="yellow", - ) - return False - - sha1sums = self._get_files_sha1sums() - # Check if all remote files are present locally - for remote_file_name in remote_files: - if remote_file_name not in local_files: - click.secho( - f"{remote_file_name} is not in local challenge.", - fg="yellow", - ) - return False - - # sha1sum is present in CTFd 3.7+, use it instead of downloading the file if possible - remote_file_sha1sum = sha1sums[remote_files[remote_file_name]["location"]] - if remote_file_sha1sum is not None: - with open(self.challenge_directory / local_files[remote_file_name], "rb") as lf: - local_file_sha1sum = hash_file(lf) - - if local_file_sha1sum != remote_file_sha1sum: - click.secho( - "sha1sum does not match with remote one.", - fg="yellow", - ) - return False - - return True - - # If sha1sum is not present, download the file and compare the contents - r = self.api.get(remote_files[remote_file_name]["url"]) - r.raise_for_status() - remote_file_contents = r.content - local_file_contents = (self.challenge_directory / local_files[remote_file_name]).read_bytes() - - if remote_file_contents != local_file_contents: - click.secho( - "the file content does not match with the remote one.", - fg="yellow", - ) - return False - - return True + # Some properties (e.g. files) are not part of the normalized challenge + # and implement their own whole-run verification instead + return all(prop.verify(ctx) for prop in PROPERTIES) def save(self): challenge_dict = dict(self) diff --git a/ctfcli/core/lint.py b/ctfcli/core/lint.py new file mode 100644 index 0000000..7f3dd69 --- /dev/null +++ b/ctfcli/core/lint.py @@ -0,0 +1,119 @@ +import subprocess + +import click + +from ctfcli.core.exceptions import LintException +from ctfcli.core.properties import PROPERTIES +from ctfcli.utils.tools import strings + + +def lint_challenge(challenge, skip_hadolint: bool = False, flag_format: str = "flag{") -> bool: + issues = {"fields": [], "dockerfile": [], "hadolint": [], "files": []} + + # Check if required fields are present + for field in [ + "name", + "author", + "category", + "description", + "attribution", + "value", + ]: + # value is allowed to be none if the challenge type is dynamic + if field == "value" and challenge.get("type") == "dynamic": + continue + + if challenge.get(field) is None: + issues["fields"].append(f"challenge.yml is missing required field: {field}") + + # Let each property validate its own value + for prop in PROPERTIES: + prop.lint(challenge, issues) + + # Check that the image field and Dockerfile match + if (challenge.challenge_directory / "Dockerfile").is_file() and challenge.get("image", "") != ".": + issues["dockerfile"].append("Dockerfile exists but image field does not point to it") + + # Check that Dockerfile exists and is EXPOSE'ing a port + if challenge.get("image") == ".": + dockerfile_path = challenge.challenge_directory / "Dockerfile" + has_dockerfile = dockerfile_path.is_file() + + if not has_dockerfile: + issues["dockerfile"].append("Dockerfile specified in 'image' field but no Dockerfile found") + + if has_dockerfile: + with open(dockerfile_path) as dockerfile: + dockerfile_source = dockerfile.read() + + if "EXPOSE" not in dockerfile_source: + issues["dockerfile"].append("Dockerfile is missing EXPOSE") + + if not skip_hadolint: + # Check Dockerfile with hadolint + hadolint = subprocess.run( + ["docker", "run", "--rm", "-i", "hadolint/hadolint"], + capture_output=True, + input=dockerfile_source.encode(), + ) + + if hadolint.returncode != 0: + issues["hadolint"].append(hadolint.stdout.decode()) + + else: + click.secho("Skipping Hadolint", fg="yellow") + + # Check that all files exist + files = challenge.get("files") or [] + for challenge_file in files: + challenge_file_path = challenge.challenge_directory / challenge_file + + if challenge_file_path.is_file() is False: + issues["files"].append( + f"Challenge file '{challenge_file}' specified, but not found at {challenge_file_path}" + ) + + # Check that the optional solution file exists + solution = challenge.get("solution", None) + if solution: + solution_file = None + solution_state = "hidden" + + if type(solution) == str: + solution_file = solution + elif type(solution) == dict: + solution_file = solution.get("path") + solution_state = solution.get("state", "hidden") + + if type(solution_state) != str or solution_state not in ["hidden", "visible", "solved"]: + issues["fields"].append("The solution state must be one of: hidden, visible, solved") + + else: + issues["fields"].append("The solution field must be a string path or an object with path and state") + + if type(solution_file) != str or not solution_file: + issues["fields"].append("The solution object must define a non-empty string path field") + else: + solution_file_path = challenge.challenge_directory / solution_file + if solution_file_path.is_file() is False: + issues["files"].append( + f"Solution file '{solution_file}' specified, but not found at {solution_file_path}" + ) + + # Check that files don't have a flag in them + for challenge_file in files: + challenge_file_path = challenge.challenge_directory / challenge_file + + if not challenge_file_path.exists(): + # The check for files present is above; this is only to look for flags in files that we do have + continue + + for s in strings(challenge_file_path): + if flag_format in s: + s = s.strip() + issues["files"].append(f"Potential flag found in distributed file '{challenge_file}':\n {s}") + + if any(messages for messages in issues.values() if len(messages) > 0): + raise LintException(issues=issues) + + return True diff --git a/ctfcli/core/properties/__init__.py b/ctfcli/core/properties/__init__.py new file mode 100644 index 0000000..0e0258a --- /dev/null +++ b/ctfcli/core/properties/__init__.py @@ -0,0 +1,91 @@ +from ctfcli.core.properties.base import NOT_PULLED, Property, PropertyContext +from ctfcli.core.properties.collections import ( + FlagsProperty, + HintsProperty, + TagsProperty, + TopicsProperty, +) +from ctfcli.core.properties.files import FilesProperty +from ctfcli.core.properties.image import ImageProperty +from ctfcli.core.properties.references import ( + ModuleProperty, + NextProperty, + RequirementsProperty, +) +from ctfcli.core.properties.scalars import ( + AttemptsProperty, + AttributionProperty, + ConnectionInfoProperty, + DescriptionProperty, + ExtraProperty, + LocalProperty, + LogicProperty, + NameProperty, + ScheduledAtProperty, + StateProperty, + TextProperty, + TypeProperty, + ValueProperty, +) +from ctfcli.core.properties.solution import SolutionProperty + +# The ordered registry of all challenge.yml properties. +# The order defines the order of keys written by Challenge.save(), +# and should match the order of the documented spec (spec/challenge-example.yml). +PROPERTIES: list[Property] = [ + NameProperty(), + LocalProperty("author"), + TextProperty("category"), + DescriptionProperty(), + AttributionProperty(), + ValueProperty(), + TypeProperty(), + ExtraProperty(), + ImageProperty(), + LocalProperty("protocol"), + LocalProperty("host"), + ConnectionInfoProperty(), + LocalProperty("healthcheck"), + SolutionProperty(), + AttemptsProperty(), + LogicProperty(), + FlagsProperty(), + FilesProperty(), + TopicsProperty(), + TagsProperty(), + HintsProperty(), + RequirementsProperty(), + NextProperty(), + ScheduledAtProperty(), + ModuleProperty(), + StateProperty(), + LocalProperty("version", newline_before=True), +] + +_PROPERTIES_BY_KEY: dict[str, Property] = {p.key: p for p in PROPERTIES} + + +def get_property(key: str) -> Property: + return _PROPERTIES_BY_KEY[key] + + +def has_property(key: str) -> bool: + return key in _PROPERTIES_BY_KEY + + +# Remote create/sync operations run in a stable, explicit order (op_order) - +# it intentionally differs from the yaml key order: e.g. the solution is +# uploaded last, and payload-only properties don't take part at all +def operation_order() -> list[Property]: + return sorted(PROPERTIES, key=lambda p: p.op_order) + + +__all__ = [ + "NOT_PULLED", + "PROPERTIES", + "Property", + "PropertyContext", + "get_property", + "has_property", + "operation_order", +] diff --git a/ctfcli/core/properties/base.py b/ctfcli/core/properties/base.py new file mode 100644 index 0000000..6f2278d --- /dev/null +++ b/ctfcli/core/properties/base.py @@ -0,0 +1,115 @@ +from typing import Any + +# Sentinel returned by Property.pull for properties that are not part of the +# normalized remote challenge (e.g. files, or keys that only exist locally) +NOT_PULLED = object() + + +class PropertyContext: + """Shared state for a single create / sync / mirror / verify run. + + Wraps the Challenge instance and caches remote lookups, so that multiple + properties do not re-fetch the same resources during one run. + """ + + def __init__( + self, + challenge, + ignore: tuple[str] = (), + remote_challenge: dict | None = None, + options: dict | None = None, + ): + self.challenge = challenge + self.ignore = ignore + self.remote_challenge = remote_challenge + # run-specific options (e.g. the files directory name during mirror) + self.options = options or {} + self._installed_challenges = None + + @property + def api(self): + return self.challenge.api + + @property + def challenge_id(self): + return self.challenge.challenge_id + + @property + def challenge_directory(self): + return self.challenge.challenge_directory + + def installed_challenges(self) -> list: + if self._installed_challenges is None: + self._installed_challenges = self.challenge.load_installed_challenges() + + return self._installed_challenges + + +class Property: + """A single challenge.yml attribute and its full lifecycle. + + The ordered registry of Property instances (ctfcli.core.properties.PROPERTIES) + defines the challenge.yml spec: Challenge.save() writes keys in registry order, + and create/sync/mirror/verify are loops over the registry. Adding a new + challenge attribute means adding one Property subclass and one registry entry. + """ + + # the challenge.yml key this property is responsible for + key: str = "" + + # whether Challenge.save() should put a blank line before this key + newline_before = False + + # relative position of this property's remote operations during create/sync; + # properties that only contribute to the initial payload keep the default 0 + op_order = 0 + + def ignored(self, ctx: PropertyContext) -> bool: + return self.key in ctx.ignore + + # whether the value is the implicit default, which save() omits and + # verify() accepts when the key is not present in challenge.yml + def is_default(self, value: Any) -> bool: + return False + + # contribution to the initial challenge POST payload + def create_payload(self, ctx: PropertyContext) -> dict: + return {} + + # contribution to the initial challenge PATCH payload + def sync_payload(self, ctx: PropertyContext) -> dict: + return self.create_payload(ctx) + + # remote operations to run after the challenge has been created + def create(self, ctx: PropertyContext) -> None: + pass + + # remote operations to run after the challenge has been patched + def sync(self, ctx: PropertyContext) -> None: + pass + + # normalized challenge.yml value built from remote challenge data, + # or NOT_PULLED when the property is not part of the normalized challenge + def pull(self, ctx: PropertyContext, remote_data: dict) -> Any: + return NOT_PULLED + + # whether a local value is equivalent to a pulled remote value + def matches(self, ctx: PropertyContext, local: Any, remote: Any) -> bool: + return local == remote + + # whole-run verification hook for properties that are not part of the + # normalized challenge and cannot be compared with pull() / matches() + # (e.g. files). Returns True when the property is in sync with the remote. + def verify(self, ctx: PropertyContext) -> bool: + return True + + # per-property validation of the local challenge.yml value, appending any + # problems to the shared lint issues dict (see ctfcli.core.lint) + def lint(self, challenge, issues: dict) -> None: + pass + + # hook to amend the normalized challenge during mirror, for properties that + # are not part of the normalized challenge and require side effects instead + # (e.g. files, which have to be downloaded and written to disk) + def mirror(self, ctx: PropertyContext, normalized: dict) -> None: + pass diff --git a/ctfcli/core/properties/collections.py b/ctfcli/core/properties/collections.py new file mode 100644 index 0000000..0f19bdf --- /dev/null +++ b/ctfcli/core/properties/collections.py @@ -0,0 +1,259 @@ +import click + +from ctfcli.core.properties.base import Property, PropertyContext + + +class CollectionProperty(Property): + """A list-valued attribute stored in its own CTFd collection + (flags, topics, tags, hints). Synced by deleting the existing remote items + and recreating them from challenge.yml.""" + + def create(self, ctx: PropertyContext) -> None: + if ctx.challenge.get(self.key) and not self.ignored(ctx): + self.create_items(ctx) + + def sync(self, ctx: PropertyContext) -> None: + if self.ignored(ctx): + return + + self.delete_existing(ctx) + if ctx.challenge.get(self.key): + self.create_items(ctx) + + def delete_existing(self, ctx: PropertyContext) -> None: + raise NotImplementedError + + def create_items(self, ctx: PropertyContext) -> None: + raise NotImplementedError + + +class FlagsProperty(CollectionProperty): + key = "flags" + newline_before = True + op_order = 10 + + def delete_existing(self, ctx: PropertyContext) -> None: + remote_flags = ctx.api.get("/api/v1/flags").json()["data"] + for flag in remote_flags: + if flag["challenge_id"] == ctx.challenge_id: + r = ctx.api.delete(f"/api/v1/flags/{flag['id']}") + r.raise_for_status() + + def create_items(self, ctx: PropertyContext) -> None: + for flag in ctx.challenge["flags"]: + if type(flag) == str: + flag_payload = { + "content": flag, + "type": "static", + "challenge_id": ctx.challenge_id, + } + else: + flag_payload = {**flag, "challenge_id": ctx.challenge_id} + + r = ctx.api.post("/api/v1/flags", json=flag_payload) + r.raise_for_status() + + def pull(self, ctx: PropertyContext, remote_data: dict): + r = ctx.api.get(f"/api/v1/challenges/{ctx.challenge_id}/flags") + r.raise_for_status() + flags = r.json()["data"] + return [ + ( + f["content"] + if f["type"] == "static" and (f["data"] is None or f["data"] == "") + else { + "content": f["content"].strip().replace("\r\n", "\n"), + "type": f["type"], + "data": f["data"], + } + ) + for f in flags + ] + + +class TopicsProperty(CollectionProperty): + key = "topics" + newline_before = True + op_order = 20 + + def is_default(self, value) -> bool: + return value == [] + + def delete_existing(self, ctx: PropertyContext) -> None: + remote_topics = ctx.api.get(f"/api/v1/challenges/{ctx.challenge_id}/topics").json()["data"] + for topic in remote_topics: + r = ctx.api.delete(f"/api/v1/topics?type=challenge&target_id={topic['id']}") + r.raise_for_status() + + def create_items(self, ctx: PropertyContext) -> None: + for topic in ctx.challenge["topics"]: + r = ctx.api.post( + "/api/v1/topics", + json={ + "value": topic, + "type": "challenge", + "challenge_id": ctx.challenge_id, + }, + ) + r.raise_for_status() + + def pull(self, ctx: PropertyContext, remote_data: dict): + r = ctx.api.get(f"/api/v1/challenges/{ctx.challenge_id}/topics") + r.raise_for_status() + topics = r.json()["data"] + return [t["value"] for t in topics] + + +class TagsProperty(CollectionProperty): + key = "tags" + newline_before = True + op_order = 30 + + def is_default(self, value) -> bool: + return value == [] + + def delete_existing(self, ctx: PropertyContext) -> None: + remote_tags = ctx.api.get("/api/v1/tags").json()["data"] + for tag in remote_tags: + if tag["challenge_id"] == ctx.challenge_id: + r = ctx.api.delete(f"/api/v1/tags/{tag['id']}") + r.raise_for_status() + + def create_items(self, ctx: PropertyContext) -> None: + for tag in ctx.challenge["tags"]: + r = ctx.api.post( + "/api/v1/tags", + json={"challenge_id": ctx.challenge_id, "value": tag}, + ) + r.raise_for_status() + + def pull(self, ctx: PropertyContext, remote_data: dict): + r = ctx.api.get(f"/api/v1/challenges/{ctx.challenge_id}/tags") + r.raise_for_status() + tags = r.json()["data"] + return [t["value"] for t in tags] + + +class HintsProperty(CollectionProperty): + key = "hints" + newline_before = True + op_order = 50 + + def is_default(self, value) -> bool: + return value == [] + + def delete_existing(self, ctx: PropertyContext) -> None: + remote_hints = ctx.api.get("/api/v1/hints").json()["data"] + for hint in remote_hints: + if hint["challenge_id"] == ctx.challenge_id: + r = ctx.api.delete(f"/api/v1/hints/{hint['id']}") + r.raise_for_status() + + def create_items(self, ctx: PropertyContext) -> None: + key_to_id = {} + target_hints = {} + + # Pass 1: create all hints; hints with requirements get blank content initially + # to prevent content from being exposed before prerequisites are enforced + for idx, hint in enumerate(ctx.challenge["hints"]): + if type(hint) == str: + hint_payload = { + "content": hint, + "title": "", + "cost": 0, + "challenge_id": ctx.challenge_id, + } + key = None + else: + has_requirements = bool(hint.get("requirements")) + hint_payload = { + "content": "" if has_requirements else hint["content"], + "title": hint.get("title", ""), + "cost": hint.get("cost", 0), + "challenge_id": ctx.challenge_id, + } + key = hint.get("key") + + r = ctx.api.post("/api/v1/hints", json=hint_payload) + r.raise_for_status() + + # Store IDs for processing later + target_hints[idx] = r.json()["data"]["id"] + if key is not None: + key_to_id[key] = r.json()["data"]["id"] + + # Pass 2: set requirements + for idx, hint in enumerate(ctx.challenge["hints"]): + if type(hint) == str: + continue + + requirements = hint.get("requirements", []) + if not requirements: + continue + + prerequisite_ids = [] + for req_key in requirements: + if req_key in key_to_id: + preq_hint_id = key_to_id[req_key] + prerequisite_ids.append(preq_hint_id) + else: + click.secho( + f'Hint key "{req_key}" not found. Skipping invalid hint requirement.', + fg="yellow", + ) + + hint_id = target_hints[idx] + + # Pass 3: fill in real content + if prerequisite_ids: + r = ctx.api.patch( + f"/api/v1/hints/{hint_id}", + json={"requirements": {"prerequisites": prerequisite_ids}}, + ) + r.raise_for_status() + + # Now safe to set the real content + r = ctx.api.patch( + f"/api/v1/hints/{hint_id}", + json={"content": hint["content"]}, + ) + r.raise_for_status() + + def pull(self, ctx: PropertyContext, remote_data: dict): + r = ctx.api.get(f"/api/v1/challenges/{ctx.challenge_id}/hints") + r.raise_for_status() + hints = r.json()["data"] + + # Determine which hints are part of a requirements chain: + # either they have prerequisites themselves, or are referenced as a prerequisite + referenced_ids = set() + for h in hints: + for pid in (h.get("requirements") or {}).get("prerequisites", []): + referenced_ids.add(pid) + hints_with_requirements = {h["id"] for h in hints if (h.get("requirements") or {}).get("prerequisites")} + needs_key = referenced_ids | hints_with_requirements + + id_to_key = {h["id"]: f"hint-{h['id']}" for h in hints} + normalized_hints = [] + for h in hints: + prerequisites = (h.get("requirements") or {}).get("prerequisites", []) + has_requirements = bool(prerequisites) + has_cost = h["cost"] > 0 + has_title = bool(h.get("title", "")) + in_requirements_chain = h["id"] in needs_key + + if not has_cost and not has_requirements and not has_title and not in_requirements_chain: + normalized_hints.append(h["content"]) + else: + hint_dict = {"content": h["content"]} + if in_requirements_chain: + hint_dict["key"] = id_to_key[h["id"]] + if has_title: + hint_dict["title"] = h["title"] + if has_cost: + hint_dict["cost"] = h["cost"] + if has_requirements: + hint_dict["requirements"] = [id_to_key[pid] for pid in prerequisites if pid in id_to_key] + normalized_hints.append(hint_dict) + + return normalized_hints diff --git a/ctfcli/core/properties/files.py b/ctfcli/core/properties/files.py new file mode 100644 index 0000000..375fc9e --- /dev/null +++ b/ctfcli/core/properties/files.py @@ -0,0 +1,235 @@ +from pathlib import Path + +import click + +from ctfcli.core.exceptions import InvalidChallengeFile +from ctfcli.core.properties.base import Property, PropertyContext +from ctfcli.utils.hashing import hash_file + + +class FilesProperty(Property): + """Distributed challenge files. Created in bulk on create; synced with a + diff against the remote (using sha1sums when the CTFd version provides them) + to avoid re-uploading unchanged files. + + Note: files are deliberately not part of pull() / the normalized challenge - + mirror() and verify() handle them separately, because downloading is only + wanted in some flows. + """ + + key = "files" + newline_before = True + op_order = 40 + + def is_default(self, value) -> bool: + return value == [] + + def validate(self, ctx: PropertyContext) -> None: + files = ctx.challenge.get("files") or [] + for challenge_file in files: + if not (ctx.challenge_directory / challenge_file).exists(): + raise InvalidChallengeFile(f"File {challenge_file} could not be loaded") + + def create(self, ctx: PropertyContext) -> None: + if ctx.challenge.get("files") and not self.ignored(ctx): + self.create_all_files(ctx) + + def sync(self, ctx: PropertyContext) -> None: + if self.ignored(ctx): + return + + ctx.challenge["files"] = ctx.challenge.get("files") or [] + ctx.remote_challenge["files"] = ctx.remote_challenge.get("files") or [] + + # Get basenames of local files to compare against remote files + local_files = {f.split("/")[-1]: f for f in ctx.challenge["files"]} + remote_files = self.normalize_remote_files(ctx.remote_challenge["files"]) + + # Delete remote files which are no longer defined locally + for remote_file in remote_files: + if remote_file not in local_files: + self.delete_file(ctx, remote_files[remote_file]["location"]) + + # Only check for file changes if there are files to upload + if not local_files: + return + + sha1sums = self.get_files_sha1sums(ctx) + for local_file_name in local_files: + # Creating a new file + if local_file_name not in remote_files: + self.create_file(ctx, ctx.challenge_directory / local_files[local_file_name]) + continue + + # Updating an existing file + # sha1sum is present in CTFd 3.7+, use it instead of always re-uploading the file if possible + remote_file_sha1sum = sha1sums[remote_files[local_file_name]["location"]] + if remote_file_sha1sum is not None: + with open(ctx.challenge_directory / local_files[local_file_name], "rb") as lf: + local_file_sha1sum = hash_file(lf) + + # Allow users to specify sha1sum in ignore to force reuploads + if "sha1sum" not in ctx.ignore and local_file_sha1sum == remote_file_sha1sum: + continue + + # if sha1sums are not present, or the hashes are different, re-upload the file + self.delete_file(ctx, remote_files[local_file_name]["location"]) + self.create_file(ctx, ctx.challenge_directory / local_files[local_file_name]) + + def mirror(self, ctx: PropertyContext, normalized: dict) -> None: + if self.ignored(ctx): + return + + ctx.remote_challenge["files"] = ctx.remote_challenge.get("files") or [] + normalized["files"] = normalized.get("files") or [] + + files_directory_name = ctx.options.get("files_directory_name", "dist") + local_files = {Path(f).name: f for f in normalized["files"]} + + # Update files + for remote_file in ctx.remote_challenge["files"]: + # Get base file name + remote_file_name = remote_file.split("/")[-1].split("?token=")[0] + + # The file is only present on the remote - we have to download it, and assume a path + if remote_file_name not in local_files: + r = ctx.api.get(remote_file) + r.raise_for_status() + + # Ensure the directory for the challenge files exists + challenge_files_directory = ctx.challenge_directory / files_directory_name + challenge_files_directory.mkdir(parents=True, exist_ok=True) + + (challenge_files_directory / remote_file_name).write_bytes(r.content) + normalized["files"].append(f"{files_directory_name}/{remote_file_name}") + + # The file is already present in the challenge.yml - we know the desired path + else: + r = ctx.api.get(remote_file) + r.raise_for_status() + (ctx.challenge_directory / local_files[remote_file_name]).write_bytes(r.content) + + # Soft-Delete files that are not present on the remote + # Remove them from challenge.yml but do not delete them from disk + remote_file_names = [f.split("/")[-1].split("?token=")[0] for f in ctx.remote_challenge["files"]] + normalized["files"] = [f for f in normalized["files"] if Path(f).name in remote_file_names] + + def verify(self, ctx: PropertyContext) -> bool: + if self.ignored(ctx): + return True + + ctx.challenge["files"] = ctx.challenge.get("files") or [] + ctx.remote_challenge["files"] = ctx.remote_challenge.get("files") or [] + + # Check if files defined in challenge.yml are present + try: + self.validate(ctx) + local_files = {Path(f).name: f for f in ctx.challenge["files"]} + except InvalidChallengeFile: + click.secho( + "InvalidChallengeFile", + fg="yellow", + ) + return False + + remote_files = self.normalize_remote_files(ctx.remote_challenge["files"]) + # Check if there are no extra local files + for local_file in local_files: + if local_file not in remote_files: + click.secho( + f"{local_file} is not in remote challenge.", + fg="yellow", + ) + return False + + sha1sums = self.get_files_sha1sums(ctx) + # Check if all remote files are present locally + for remote_file_name in remote_files: + if remote_file_name not in local_files: + click.secho( + f"{remote_file_name} is not in local challenge.", + fg="yellow", + ) + return False + + # sha1sum is present in CTFd 3.7+, use it instead of downloading the file if possible + remote_file_sha1sum = sha1sums[remote_files[remote_file_name]["location"]] + if remote_file_sha1sum is not None: + with open(ctx.challenge_directory / local_files[remote_file_name], "rb") as lf: + local_file_sha1sum = hash_file(lf) + + if local_file_sha1sum != remote_file_sha1sum: + click.secho( + "sha1sum does not match with remote one.", + fg="yellow", + ) + return False + + continue + + # If sha1sum is not present, download the file and compare the contents + r = ctx.api.get(remote_files[remote_file_name]["url"]) + r.raise_for_status() + remote_file_contents = r.content + local_file_contents = (ctx.challenge_directory / local_files[remote_file_name]).read_bytes() + + if remote_file_contents != local_file_contents: + click.secho( + "the file content does not match with the remote one.", + fg="yellow", + ) + return False + + return True + + def delete_file(self, ctx: PropertyContext, remote_location: str) -> None: + remote_files = ctx.api.get("/api/v1/files?type=challenge").json()["data"] + + for remote_file in remote_files: + if remote_file["location"] == remote_location: + r = ctx.api.delete(f"/api/v1/files/{remote_file['id']}") + r.raise_for_status() + + def create_file(self, ctx: PropertyContext, local_path: Path) -> None: + file_payload = {"challenge_id": ctx.challenge_id, "type": "challenge"} + + with local_path.open(mode="rb") as file_handle: + # Specifically use data= here to send multipart/form-data + r = ctx.api.post("/api/v1/files", files={"file": (local_path.name, file_handle)}, data=file_payload) + r.raise_for_status() + + def create_all_files(self, ctx: PropertyContext) -> None: + new_files = [] + for challenge_file in ctx.challenge["files"]: + file_path = ctx.challenge_directory / challenge_file + new_files.append(("file", (file_path.name, file_path.open("rb")))) + + files_payload = {"challenge_id": ctx.challenge_id, "type": "challenge"} + + # Specifically use data= here to send multipart/form-data + r = ctx.api.post("/api/v1/files", files=new_files, data=files_payload) + r.raise_for_status() + + # Close the file handles + for file_payload in new_files: + file_payload[1][1].close() + + # Create a dictionary of remote files in { basename: {"url": "", "location": ""} } format + @staticmethod + def normalize_remote_files(remote_files: list[str]) -> dict[str, dict[str, str]]: + normalized = {} + for f in remote_files: + file_parts = f.split("?token=")[0].split("/") + normalized[file_parts[-1]] = { + "url": f, + "location": f"{file_parts[-2]}/{file_parts[-1]}", + } + + return normalized + + # Create a dictionary of sha1sums in { location: sha1sum } format + @staticmethod + def get_files_sha1sums(ctx: PropertyContext) -> dict[str, str]: + r = ctx.api.get("/api/v1/files?type=challenge") + r.raise_for_status() + return {f["location"]: f.get("sha1sum", None) for f in r.json()["data"]} diff --git a/ctfcli/core/properties/image.py b/ctfcli/core/properties/image.py new file mode 100644 index 0000000..584baeb --- /dev/null +++ b/ctfcli/core/properties/image.py @@ -0,0 +1,69 @@ +import subprocess + +from slugify import slugify + +from ctfcli.core.exceptions import InvalidChallengeFile +from ctfcli.core.image import Image +from ctfcli.core.properties.base import Property, PropertyContext + + +class ImageProperty(Property): + """The challenge image. Only lives in challenge.yml - it is not synced with + the remote, but used by the deployment handlers. Resolving determines whether + the image is a registry reference, a local Dockerfile to be built, or a + pre-built local image.""" + + key = "image" + newline_before = True + + # Well-known registries recognized without an explicit registry:// prefix + known_registries = [ + "docker.io", + "gcr.io", + "ecr.aws", + "ghcr.io", + "azurecr.io", + "registry.digitalocean.com", + "registry.gitlab.com", + "registry.ctfd.io", + ] + + def resolve(self, ctx: PropertyContext) -> Image | None: + challenge_image = ctx.challenge.get("image") + + if not challenge_image: + return None + + # Check if challenge_image is explicitly marked with registry:// prefix + if challenge_image.startswith("registry://"): + challenge_image = challenge_image.replace("registry://", "") + return Image(challenge_image) + + # Check if it's a library image + if challenge_image.startswith("library/"): + return Image(f"docker.io/{challenge_image}") + + # Check if it defines a known registry + for registry in self.known_registries: + if registry in challenge_image: + return Image(challenge_image) + + # Check if it's a path to dockerfile to be built + if (ctx.challenge_directory / challenge_image / "Dockerfile").exists(): + return Image(slugify(ctx.challenge["name"]), ctx.challenge_directory / challenge_image) + + # Check if it's a local pre-built image + if ( + subprocess.call( + ["docker", "inspect", challenge_image], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + == 0 + ): + return Image(challenge_image) + + # If the image is set, but we fail to determine whether it's local / remote - raise an exception + raise InvalidChallengeFile( + f"Challenge file at {ctx.challenge.challenge_file_path} defines an image, but it couldn't be resolved" + ) diff --git a/ctfcli/core/properties/references.py b/ctfcli/core/properties/references.py new file mode 100644 index 0000000..6fd8df7 --- /dev/null +++ b/ctfcli/core/properties/references.py @@ -0,0 +1,260 @@ +import click + +from ctfcli.core.properties.base import Property, PropertyContext + + +class RequirementsProperty(Property): + key = "requirements" + newline_before = True + op_order = 60 + + def is_default(self, value) -> bool: + return value == [] or value == {"prerequisites": [], "anonymize": False} + + def create(self, ctx: PropertyContext) -> None: + if ctx.challenge.get("requirements") and not self.ignored(ctx): + self._set(ctx) + + def sync(self, ctx: PropertyContext) -> None: + self.create(ctx) + + def _set(self, ctx: PropertyContext) -> None: + requirements = ctx.challenge["requirements"] + required_challenges = [] + anonymize = False + if type(requirements) == dict: + rc = requirements.get("prerequisites", []) + anonymize = requirements.get("anonymize", False) + else: + rc = requirements + + for required_challenge in rc: + if type(required_challenge) == str: + # requirement by name + # find the challenge id from installed challenges + found = False + for remote_challenge in ctx.installed_challenges(): + if remote_challenge["name"] == required_challenge: + required_challenges.append(remote_challenge["id"]) + found = True + break + if found is False: + click.secho( + f'Challenge id cannot be found. Skipping invalid requirement name "{required_challenge}".', + fg="yellow", + ) + + elif type(required_challenge) == int: + # requirement by challenge id + # trust it and use it directly + required_challenges.append(required_challenge) + + required_challenge_ids = list(set(required_challenges)) + + if ctx.challenge_id in required_challenge_ids: + click.secho( + "Challenge cannot require itself. Skipping invalid requirement.", + fg="yellow", + ) + required_challenges.remove(ctx.challenge_id) + required_challenges.sort() + + requirements_payload = { + "requirements": { + "prerequisites": required_challenges, + "anonymize": anonymize, + } + } + r = ctx.api.patch(f"/api/v1/challenges/{ctx.challenge_id}", json=requirements_payload) + r.raise_for_status() + + def pull(self, ctx: PropertyContext, remote_data: dict): + r = ctx.api.get(f"/api/v1/challenges/{ctx.challenge_id}/requirements") + r.raise_for_status() + prerequisites = (r.json().get("data") or {}).get("prerequisites", []) + + requirements = {"prerequisites": [], "anonymize": False} + if len(prerequisites) > 0: + # Prefer challenge names over IDs + requirements["prerequisites"] = [c["name"] for c in ctx.installed_challenges() if c["id"] in prerequisites] + + requirements["anonymize"] = (r.json().get("data") or {}).get("anonymize", False) + return requirements + + # Compare challenge requirements, will resolve all IDs to names + def matches(self, ctx: PropertyContext, local, remote) -> bool: + if local == remote: + return True + + if type(local) == dict: + local_prerequisites = local["prerequisites"] + local_anonymize = local.get("anonymize", False) + else: + local_prerequisites = local + local_anonymize = False + + def normalize_requirements(requirements): + normalized = [] + for requirement in requirements: + if type(requirement) == int: + for remote_challenge in ctx.installed_challenges(): + if remote_challenge["id"] == requirement: + normalized.append(remote_challenge["name"]) + break + else: + normalized.append(requirement) + + return normalized + + nr1 = normalize_requirements(local_prerequisites) + nr1.sort() + nr2 = normalize_requirements(remote["prerequisites"]) + nr2.sort() + return nr1 == nr2 and local_anonymize == remote["anonymize"] + + +class NextProperty(Property): + key = "next" + op_order = 70 + + def is_default(self, value) -> bool: + return value is None + + def create(self, ctx: PropertyContext) -> None: + if not self.ignored(ctx): + self._set(ctx) + + def sync(self, ctx: PropertyContext) -> None: + self.create(ctx) + + def _set(self, ctx: PropertyContext) -> None: + _next = ctx.challenge.get("next", None) + + if type(_next) == str: + # next by name + # find the challenge id from installed challenges + resolved_next = None + for remote_challenge in ctx.installed_challenges(): + if remote_challenge["name"] == _next: + resolved_next = remote_challenge["id"] + break + if resolved_next is None: + click.secho( + "Challenge cannot find next challenge. Maybe it is invalid name or id. It will be cleared.", + fg="yellow", + ) + _next = resolved_next + elif type(_next) == int and _next > 0: + # next by challenge id + # trust it and use it directly + pass + else: + _next = None + + if ctx.challenge_id == _next: + click.secho( + "Challenge cannot set next challenge itself. Skipping invalid next challenge.", + fg="yellow", + ) + _next = None + + next_payload = {"next_id": _next} + r = ctx.api.patch(f"/api/v1/challenges/{ctx.challenge_id}", json=next_payload) + r.raise_for_status() + + def pull(self, ctx: PropertyContext, remote_data: dict): + nid = remote_data.get("next_id") + if not nid: + return None + + # Prefer the challenge name over the ID + r = ctx.api.get(f"/api/v1/challenges/{nid}") + r.raise_for_status() + return (r.json().get("data") or {}).get("name", None) + + # Compare next challenges, will resolve all IDs to names + def matches(self, ctx: PropertyContext, local, remote) -> bool: + def normalize_next(value): + if type(value) == int: + if value > 0: + remote_challenge = ctx.challenge.load_installed_challenge(value) + if remote_challenge["id"] == value: + return remote_challenge["name"] + return None + + return value + + return normalize_next(local) == normalize_next(remote) + + +class ModuleProperty(Property): + key = "module" + op_order = 80 + + def is_default(self, value) -> bool: + return value is None + + def create(self, ctx: PropertyContext) -> None: + if ctx.challenge.get("module") and not self.ignored(ctx): + self._set(ctx) + + def sync(self, ctx: PropertyContext) -> None: + # 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 ctx.challenge and not self.ignored(ctx): + self._set(ctx) + + def _set(self, ctx: PropertyContext) -> None: + module = ctx.challenge.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 = ctx.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 = ctx.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 = ctx.api.patch(f"/api/v1/challenges/{ctx.challenge_id}", json=module_payload) + r.raise_for_status() + + def pull(self, ctx: PropertyContext, remote_data: dict): + module_id = remote_data.get("module_id") + if not module_id: + return None + + # Prefer the module name over the ID + r = ctx.api.get(f"/api/v1/modules/{module_id}") + r.raise_for_status() + return (r.json().get("data") or {}).get("name", None) + + # 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 matches(self, ctx: PropertyContext, local, remote) -> bool: + def normalize_module(value): + return None if value is None else str(value) + + return normalize_module(local) == normalize_module(remote) diff --git a/ctfcli/core/properties/scalars.py b/ctfcli/core/properties/scalars.py new file mode 100644 index 0000000..8da9bce --- /dev/null +++ b/ctfcli/core/properties/scalars.py @@ -0,0 +1,284 @@ +from datetime import datetime, timezone +from typing import Any + +from ctfcli.core.exceptions import InvalidChallengeFile +from ctfcli.core.properties.base import NOT_PULLED, Property, PropertyContext + + +class LocalProperty(Property): + """A key that only lives in challenge.yml and is never synced with the remote + (e.g. author, image, healthcheck).""" + + def __init__(self, key: str, newline_before: bool = False): + self.key = key + self.newline_before = newline_before + + +class CopiedProperty(Property): + """A payload field that is copied verbatim from the remote challenge data + when pulling.""" + + def pull(self, ctx: PropertyContext, remote_data: dict): + if self.key in remote_data: + return remote_data[self.key] + + return NOT_PULLED + + +class NameProperty(CopiedProperty): + key = "name" + + def create_payload(self, ctx: PropertyContext) -> dict: + return {"name": ctx.challenge["name"]} + + +class TextProperty(CopiedProperty): + """category / description / attribution: sent as an empty string when missing, + reverted to the remote value when ignored during sync, and blanked when + ignored during create.""" + + def __init__(self, key: str): + self.key = key + + def create_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {self.key: ""} + + return {self.key: ctx.challenge.get(self.key, "")} + + def sync_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {self.key: ctx.remote_challenge[self.key]} + + return {self.key: ctx.challenge.get(self.key, "")} + + +class DescriptionProperty(TextProperty): + def __init__(self): + super().__init__("description") + + def pull(self, ctx: PropertyContext, remote_data: dict): + return remote_data["description"].strip().replace("\r\n", "\n").replace("\t", "") + + +class AttributionProperty(TextProperty): + def __init__(self): + super().__init__("attribution") + + def pull(self, ctx: PropertyContext, remote_data: dict): + attribution = remote_data.get("attribution", "") + if attribution: + attribution = attribution.strip().replace("\r\n", "\n").replace("\t", "") + + return attribution + + +class ValueProperty(CopiedProperty): + key = "value" + + def create_payload(self, ctx: PropertyContext) -> dict: + # Some challenge types (e.g., dynamic) override value. + # We can't send it to CTFd because we don't know the current value + value = ctx.challenge.get("value", None) + if value is None: + return {} + + # if value is an int as string, cast it + if type(value) == str and value.isdigit(): + value = int(value) + + return {"value": value} + + def sync_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {"value": ctx.remote_challenge["value"]} + + return self.create_payload(ctx) + + +class TypeProperty(CopiedProperty): + key = "type" + + def is_default(self, value) -> bool: + return value == "standard" + + def create_payload(self, ctx: PropertyContext) -> dict: + return {"type": ctx.challenge.get("type", "standard")} + + def sync_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {"type": ctx.remote_challenge["type"]} + + return self.create_payload(ctx) + + +class AttemptsProperty(Property): + key = "attempts" + newline_before = True + + def is_default(self, value) -> bool: + return value == 0 + + def create_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {} + + return {"max_attempts": ctx.challenge.get("attempts", 0)} + + def pull(self, ctx: PropertyContext, remote_data: dict): + return remote_data["max_attempts"] + + +class ConnectionInfoProperty(CopiedProperty): + key = "connection_info" + + def is_default(self, value) -> bool: + return value is None + + def create_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {} + + return {"connection_info": ctx.challenge.get("connection_info", None)} + + +class LogicProperty(CopiedProperty): + key = "logic" + + def create_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx) or not ctx.challenge.get("logic"): + return {} + + return {"logic": ctx.challenge.get("logic") or "any"} + + +class ExtraProperty(Property): + key = "extra" + newline_before = True + + def create_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {} + + return {**ctx.challenge.get("extra", {})} + + def pull(self, ctx: PropertyContext, remote_data: dict): + extra = {key: remote_data[key] for key in ["initial", "decay", "minimum"] if key in remote_data} + if not extra: + return NOT_PULLED + + return extra + + +class ScheduledAtProperty(Property): + """A timed release: a visible challenge stays hidden from players until this + moment passes. Always carries an explicit timezone offset - ctfcli never + assumes one, because guessing would silently release at the wrong time.""" + + key = "scheduled_at" + newline_before = True + + def is_default(self, value) -> bool: + return 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) + + @classmethod + def parse(cls, value: Any, challenge_file_path=None) -> "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 = cls._datetime_from_iso(value) + except ValueError as e: + raise InvalidChallengeFile( + f"Challenge file at {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 {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 {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 + + @classmethod + def normalize(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 cls._datetime_from_iso(value) + parsed = parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) + + return parsed.isoformat() + + def create_payload(self, ctx: PropertyContext) -> dict: + if self.ignored(ctx): + return {} + + # parse validates the timezone is explicit and raises otherwise + parsed = self.parse(ctx.challenge.get("scheduled_at"), ctx.challenge.challenge_file_path) + return {"scheduled_at": parsed.isoformat() if parsed else None} + + def pull(self, ctx: PropertyContext, remote_data: dict): + return self.normalize(remote_data.get("scheduled_at")) + + # Compare two scheduled_at values by the instant they represent, so that + # equivalent times written with different offsets compare as equal + def matches(self, ctx: PropertyContext, local, remote) -> bool: + path = ctx.challenge.challenge_file_path + local_parsed = self.parse(local, path) + remote_parsed = self.parse(remote, path) + + 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) + + # Check that scheduled_at, if present, carries an explicit timezone + def lint(self, challenge, issues: dict) -> None: + if challenge.get("scheduled_at") is None: + return + + try: + self.parse(challenge["scheduled_at"], challenge.challenge_file_path) + except InvalidChallengeFile as e: + issues["fields"].append(str(e)) + + +class StateProperty(CopiedProperty): + key = "state" + newline_before = True + + def is_default(self, value) -> bool: + return value == "visible" + + def create_payload(self, ctx: PropertyContext) -> dict: + # Hide the challenge for the duration of the sync / creation. + # Restoring the visibility afterwards is handled by the orchestration + # in Challenge.create / Challenge.sync + return {"state": "hidden"} diff --git a/ctfcli/core/properties/solution.py b/ctfcli/core/properties/solution.py new file mode 100644 index 0000000..5c5ff4f --- /dev/null +++ b/ctfcli/core/properties/solution.py @@ -0,0 +1,152 @@ +import logging +import re +from pathlib import Path + +import click + +from ctfcli.core.properties.base import Property, PropertyContext + +log = logging.getLogger("ctfcli.core.properties.solution") + + +class SolutionProperty(Property): + """A solution document uploaded to CTFd, with markdown images uploaded as + solution files and MkDocs-style snippet includes inlined.""" + + key = "solution" + op_order = 90 + + def create(self, ctx: PropertyContext) -> None: + if not self.ignored(ctx): + self.upsert(ctx) + + def sync(self, ctx: PropertyContext) -> None: + if self.ignored(ctx): + return + + if not self.resolve_path(ctx): + self.delete_existing(ctx) + self.upsert(ctx) + + def _parse_definition(self, ctx: PropertyContext) -> tuple[str, str] | None: + solution = ctx.challenge.get("solution", None) + if not solution: + return None + + if type(solution) == str: + return solution, "hidden" + + if type(solution) != dict: + click.secho( + "The solution field must be a string path or an object with path and state", + fg="red", + ) + return None + + solution_path = solution.get("path") + if type(solution_path) != str or not solution_path: + click.secho("The solution object must define a non-empty string path field", fg="red") + return None + + solution_state = solution.get("state", "hidden") + if type(solution_state) != str or solution_state not in ["hidden", "visible", "solved"]: + click.secho("The solution state must be one of: hidden, visible, solved", fg="red") + return None + + return solution_path, solution_state + + def resolve_path(self, ctx: PropertyContext) -> tuple[Path, str] | None: + parsed_solution = self._parse_definition(ctx) + if not parsed_solution: + return None + + solution_path_string, solution_state = parsed_solution + solution_path = ctx.challenge_directory / solution_path_string + if not solution_path.is_file(): + click.secho( + f"Solution file '{solution_path_string}' specified, but not found at {solution_path}", + fg="red", + ) + return None + + return solution_path, solution_state + + def delete_existing(self, ctx: PropertyContext) -> None: + remote_solutions = ctx.api.get("/api/v1/solutions").json()["data"] + for solution in remote_solutions: + if solution["challenge_id"] == ctx.challenge_id: + r = ctx.api.delete(f"/api/v1/solutions/{solution['id']}") + r.raise_for_status() + + def _get_existing_id(self, ctx: PropertyContext) -> int | None: + r = ctx.api.get("/api/v1/solutions") + r.raise_for_status() + remote_solutions = r.json().get("data") or [] + for solution in remote_solutions: + if solution["challenge_id"] == ctx.challenge_id: + return solution["id"] + return None + + def upsert(self, ctx: PropertyContext) -> None: + resolved_solution = self.resolve_path(ctx) + if not resolved_solution: + return + solution_path, solution_state = resolved_solution + + solution_id = self._get_existing_id(ctx) + if solution_id is None: + solution_payload_create = {"challenge_id": ctx.challenge_id, "state": solution_state, "content": ""} + + r = ctx.api.post("/api/v1/solutions", json=solution_payload_create) + r.raise_for_status() + solution_id = r.json()["data"]["id"] + else: + # Keep solution state in sync and clear stale content before rebuilding references. + r = ctx.api.patch( + f"/api/v1/solutions/{solution_id}", + json={"state": solution_state, "content": ""}, + ) + r.raise_for_status() + + with solution_path.open("r") as solution_file: + content = solution_file.read() + + # Find all images in the content (markdown format; ignore html format) + # Markdown format: ![alt text](image_url) + # Returns tuples: (full_match, alt_text, image_path) + markdown_images = re.findall(r"(!\[([^\]]*)\]\(([^\)]+)\))", content) + + # Find all snippet includes (MkDocs style: --8<-- "filename") + # Returns tuples: (full_match, filename) + snippet_includes = re.findall(r'(--8<--\s+["\']([^"\']+)["\'])', content) + + for mdx, alt, path in markdown_images: + local_path = solution_path.parent / path + file_payload = { + "type": "solution", + "solution_id": solution_id, + } + + with local_path.open(mode="rb") as file_handle: + # Specifically use data= here to send multipart/form-data + r = ctx.api.post("/api/v1/files", files={"file": (local_path.name, file_handle)}, data=file_payload) + r.raise_for_status() + resp = r.json() + server_location = resp["data"][0]["location"] + + content = content.replace(mdx, f"![{alt}](/files/{server_location})") + + # Process snippet includes (--8<-- "filename") + for full_match, filename in snippet_includes: + snippet_file_path = solution_path.parent / filename + if snippet_file_path.exists(): + with snippet_file_path.open("r") as snippet_file: + snippet_content = snippet_file.read() + # Replace the --8<-- directive with the actual file content + content = content.replace(full_match, snippet_content) + else: + log.warning(f"Snippet file not found: {filename}") + + solution_payload_patch = {"content": content} + r = ctx.api.patch(f"/api/v1/solutions/{solution_id}", json=solution_payload_patch) + r.raise_for_status() diff --git a/tests/core/deployment/test_cloud_deployment.py b/tests/core/deployment/test_cloud_deployment.py index 4e36966..91771d0 100644 --- a/tests/core/deployment/test_cloud_deployment.py +++ b/tests/core/deployment/test_cloud_deployment.py @@ -69,7 +69,7 @@ def test_fails_deployment_if_instance_does_not_support_deployments( @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_fails_deployment_if_image_build_failed( self, @@ -122,7 +122,7 @@ def mock_get(*args, **kwargs): @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_fails_deployment_if_image_push_failed( self, @@ -189,7 +189,7 @@ def mock_get(*args, **kwargs): return_value=b'Error response from daemon: Get "https://registry.ctfd.io/v2/": unauthorized: authentication required', # noqa ) @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_fails_deployment_if_registry_login_unsuccessful( self, @@ -271,7 +271,7 @@ def mock_get(*args, **kwargs): @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.cloud.subprocess.check_output") @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_fails_deployment_if_instance_url_is_not_ctfd_assigned( self, @@ -355,7 +355,7 @@ def mock_get(*args, **kwargs): @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.cloud.subprocess.check_output") @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_allows_skipping_registry_login( self, @@ -486,7 +486,7 @@ def mock_patch(*args, **kwargs): @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.cloud.subprocess.check_output", return_value=b"Login Succeeded") @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_deploys_challenge_with_existing_image_service( self, @@ -628,7 +628,7 @@ def mock_patch(*args, **kwargs): @mock.patch("ctfcli.core.deployment.cloud.subprocess.check_output", return_value=b"Login Succeeded") @mock.patch("ctfcli.core.deployment.cloud.time.sleep") @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_deploys_challenge_with_new_image_service( self, @@ -822,7 +822,7 @@ def mock_post(*args, **kwargs): @mock.patch("ctfcli.core.deployment.cloud.subprocess.check_output", return_value=b"Login Succeeded") @mock.patch("ctfcli.core.deployment.cloud.time.sleep") @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_fails_deployment_after_timeout( self, @@ -1013,7 +1013,7 @@ def mock_post(*args, **kwargs): @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.cloud.subprocess.check_output", return_value=b"Login Succeeded") @mock.patch("ctfcli.core.deployment.cloud.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") @mock.patch("ctfcli.core.deployment.cloud.API") def test_exposes_tcp_port( self, diff --git a/tests/core/deployment/test_registry_deployment.py b/tests/core/deployment/test_registry_deployment.py index 134538e..89491ad 100644 --- a/tests/core/deployment/test_registry_deployment.py +++ b/tests/core/deployment/test_registry_deployment.py @@ -18,7 +18,7 @@ class TestRegistryDeployment(unittest.TestCase): @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.registry.subprocess.run") @mock.patch("ctfcli.core.deployment.registry.Config") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_builds_and_pushes_image( self, mock_image_constructor: MagicMock, @@ -68,7 +68,7 @@ def test_fails_deployment_if_no_host_provided(self, mock_secho: MagicMock, *args @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.registry.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_deployment_if_no_registry_config( self, mock_image_constructor: MagicMock, @@ -93,7 +93,7 @@ def test_fails_deployment_if_no_registry_config( @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.registry.click.secho") @mock.patch("ctfcli.core.deployment.registry.Config") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_if_no_credentials( self, mock_image_constructor: MagicMock, @@ -122,7 +122,7 @@ def test_fails_if_no_credentials( @mock.patch("ctfcli.core.deployment.registry.click.secho") @mock.patch("ctfcli.core.deployment.registry.subprocess.run") @mock.patch("ctfcli.core.deployment.registry.Config") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_if_registry_credentials_invalid( self, mock_image_constructor: MagicMock, @@ -155,7 +155,7 @@ def test_fails_if_registry_credentials_invalid( @mock.patch("ctfcli.core.deployment.registry.subprocess.run") @mock.patch("ctfcli.core.deployment.registry.click.secho") @mock.patch("ctfcli.core.deployment.registry.Config") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_deployment_if_image_build_failed( self, mock_image_constructor: MagicMock, @@ -185,7 +185,7 @@ def test_fails_deployment_if_image_build_failed( @mock.patch("ctfcli.core.deployment.registry.subprocess.run") @mock.patch("ctfcli.core.deployment.registry.click.secho") @mock.patch("ctfcli.core.deployment.registry.Config") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_deployment_if_image_push_failed( self, mock_image_constructor: MagicMock, @@ -217,7 +217,7 @@ def test_fails_deployment_if_image_push_failed( @mock.patch("ctfcli.core.deployment.registry.click.secho") @mock.patch("ctfcli.core.deployment.registry.subprocess.run") @mock.patch("ctfcli.core.deployment.registry.Config") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_allows_skipping_login( self, mock_image_constructor: MagicMock, @@ -252,7 +252,7 @@ def test_allows_skipping_login( @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.registry.click.secho") @mock.patch("ctfcli.core.deployment.registry.subprocess.run") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_warns_about_logging_in_with_skip_login( self, mock_image_constructor: MagicMock, diff --git a/tests/core/deployment/test_ssh_deployment.py b/tests/core/deployment/test_ssh_deployment.py index 29911c9..41bc84f 100644 --- a/tests/core/deployment/test_ssh_deployment.py +++ b/tests/core/deployment/test_ssh_deployment.py @@ -19,7 +19,7 @@ class TestSSHDeployment(unittest.TestCase): @mock.patch.object(Path, "unlink") @mock.patch("ctfcli.core.deployment.ssh.subprocess.run") @mock.patch("ctfcli.core.deployment.ssh.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_builds_exports_and_copies_image( self, mock_image_constructor: MagicMock, @@ -95,7 +95,7 @@ def test_fails_deployment_if_no_host_provided(self, mock_secho: MagicMock, *args @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.ssh.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_deployment_if_image_build_failed( self, mock_image_constructor: MagicMock, @@ -118,7 +118,7 @@ def test_fails_deployment_if_image_build_failed( @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.ssh.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_deployment_if_image_export_failed( self, mock_image_constructor: MagicMock, @@ -143,7 +143,7 @@ def test_fails_deployment_if_image_export_failed( @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.ssh.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_deployment_if_no_exposed_port( self, mock_image_constructor: MagicMock, @@ -172,7 +172,7 @@ def test_fails_deployment_if_no_exposed_port( @mock.patch("ctfcli.core.config.Path.cwd", return_value=challenge_directory) @mock.patch("ctfcli.core.deployment.ssh.subprocess.run") @mock.patch("ctfcli.core.deployment.ssh.click.secho") - @mock.patch("ctfcli.core.challenge.Image") + @mock.patch("ctfcli.core.properties.image.Image") def test_fails_deployment_if_any_subprocess_exits( self, mock_image_constructor: MagicMock, diff --git a/tests/core/test_challenge.py b/tests/core/test_challenge.py index 5554a4d..e91bd69 100644 --- a/tests/core/test_challenge.py +++ b/tests/core/test_challenge.py @@ -14,6 +14,7 @@ RemoteChallengeNotFound, ) from ctfcli.core.image import Image +from ctfcli.core.properties import PropertyContext, get_property BASE_DIR = Path(__file__).parent.parent @@ -58,7 +59,7 @@ def test_load_challenge(self): self.assertEqual(challenge["name"], "Test Challenge") - @mock.patch("ctfcli.core.challenge.subprocess.call") + @mock.patch("ctfcli.core.properties.image.subprocess.call") def test_raises_if_image_defined_but_not_resolved(self, mock_call: MagicMock): mock_call.return_value = 1 challenge_path = BASE_DIR / "fixtures" / "challenges" / "test-challenge-minimal" / "challenge.yml" @@ -107,7 +108,7 @@ def test_recognizes_library_images(self): self.assertEqual(challenge.image.basename, "test-challenge") self.assertTrue(challenge.image.built) - @mock.patch("ctfcli.core.challenge.subprocess.call") + @mock.patch("ctfcli.core.properties.image.subprocess.call") def test_recognizes_local_prebuilt_images(self, mock_call: MagicMock): mock_call.return_value = 0 challenge_path = BASE_DIR / "fixtures" / "challenges" / "test-challenge-minimal" / "challenge.yml" @@ -141,7 +142,7 @@ class TestChallengeSolutions(unittest.TestCase): def test_resolves_solution_from_specified_path(self): challenge = Challenge(self.minimal_challenge, {"solution": "challenge.yml"}) - solution_path, solution_state = challenge._resolve_solution_path() + solution_path, solution_state = get_property("solution").resolve_path(PropertyContext(challenge)) self.assertEqual(solution_path, challenge.challenge_directory / "challenge.yml") self.assertEqual(solution_state, "hidden") @@ -155,7 +156,7 @@ def test_resolves_solution_object_from_specified_path(self): } }, ) - solution_path, solution_state = challenge._resolve_solution_path() + solution_path, solution_state = get_property("solution").resolve_path(PropertyContext(challenge)) self.assertEqual(solution_path, challenge.challenge_directory / "challenge.yml") self.assertEqual(solution_state, "solved") @@ -169,18 +170,18 @@ def test_resolves_solution_object_with_state_from_specified_path(self): } }, ) - solution_path, solution_state = challenge._resolve_solution_path() + solution_path, solution_state = get_property("solution").resolve_path(PropertyContext(challenge)) self.assertEqual(solution_path, challenge.challenge_directory / "challenge.yml") self.assertEqual(solution_state, "visible") def test_does_not_resolve_solution_if_not_specified(self): challenge = Challenge(self.minimal_challenge) - self.assertIsNone(challenge._resolve_solution_path()) + self.assertIsNone(get_property("solution").resolve_path(PropertyContext(challenge))) - @mock.patch("ctfcli.core.challenge.click.secho") + @mock.patch("ctfcli.core.properties.solution.click.secho") def test_does_not_resolve_solution_if_missing(self, mock_secho: MagicMock): challenge = Challenge(self.minimal_challenge, {"solution": "writeup/WRITEUP.md"}) - self.assertIsNone(challenge._resolve_solution_path()) + self.assertIsNone(get_property("solution").resolve_path(PropertyContext(challenge))) mock_secho.assert_called_once_with( f"Solution file 'writeup/WRITEUP.md' specified, but not found at " f"{challenge.challenge_directory / 'writeup/WRITEUP.md'}", @@ -214,7 +215,7 @@ def mock_post(*args, **kwargs): mock_api.get.side_effect = mock_get mock_api.post.side_effect = mock_post - challenge._create_solution() + get_property("solution").upsert(PropertyContext(challenge)) mock_api.post.assert_has_calls( [call("/api/v1/solutions", json={"challenge_id": 1, "state": "hidden", "content": ""})] @@ -248,7 +249,7 @@ def mock_post(*args, **kwargs): mock_api.get.side_effect = mock_get mock_api.post.side_effect = mock_post - challenge._create_solution() + get_property("solution").upsert(PropertyContext(challenge)) mock_api.post.assert_has_calls( [call("/api/v1/solutions", json={"challenge_id": 1, "state": "visible", "content": ""})] @@ -273,7 +274,7 @@ def mock_get(*args, **kwargs): mock_api: MagicMock = mock_api_constructor.return_value mock_api.get.side_effect = mock_get - challenge._create_solution() + get_property("solution").upsert(PropertyContext(challenge)) mock_api.post.assert_not_called() mock_api.patch.assert_has_calls( @@ -290,7 +291,7 @@ def test_does_not_create_solution_if_not_specified(self, mock_api_constructor: M challenge.challenge_id = 1 mock_api: MagicMock = mock_api_constructor.return_value - challenge._create_solution() + get_property("solution").upsert(PropertyContext(challenge)) mock_api.post.assert_not_called() mock_api.patch.assert_not_called() @@ -330,7 +331,7 @@ def mock_post(*args, **kwargs): mock_api.get.side_effect = mock_get mock_api.post.side_effect = mock_post - challenge._create_solution() + get_property("solution").upsert(PropertyContext(challenge)) mock_api.post.assert_has_calls( [ @@ -887,8 +888,8 @@ def test_updates_hints(self, mock_api_constructor: MagicMock): } mock_api.post.return_value.json.return_value = {"success": True, "data": {"id": 10}} - challenge._delete_existing_hints() - challenge._create_hints() + get_property("hints").delete_existing(PropertyContext(challenge)) + get_property("hints").create_items(PropertyContext(challenge)) self.assertEqual( mock_api.delete.call_args_list, @@ -1114,7 +1115,7 @@ def mock_get(*args, **kwargs): 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.properties.references.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"}) @@ -1233,7 +1234,7 @@ def test_does_not_touch_module_if_absent(self, mock_api_constructor: MagicMock, 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.properties.references.click.secho") @mock.patch("ctfcli.core.challenge.API") def test_challenge_cannot_require_itself( self, mock_api_constructor: MagicMock, mock_secho: MagicMock, *args, **kwargs @@ -1804,7 +1805,7 @@ def mock_post(*args, **kwargs): ) @mock.patch("ctfcli.core.challenge.Challenge.load_installed_challenges", return_value=installed_challenges) - @mock.patch("ctfcli.core.challenge.click.secho") + @mock.patch("ctfcli.core.properties.references.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"}) @@ -2016,7 +2017,7 @@ def test_validates_challenge_yml_does_not_point_to_dockerfile(self): } self.assertDictEqual(expected_lint_issues, e.exception.issues) - @mock.patch("ctfcli.core.challenge.click.secho") + @mock.patch("ctfcli.core.lint.click.secho") def test_validates_dockerfile_exposes_port(self, mock_secho: MagicMock): challenge = Challenge(self.invalid_dockerfile_challenge) @@ -2033,7 +2034,7 @@ def test_validates_dockerfile_exposes_port(self, mock_secho: MagicMock): mock_secho.assert_called_once_with("Skipping Hadolint", fg="yellow") self.assertDictEqual(expected_lint_issues, e.exception.issues) - @mock.patch("ctfcli.core.challenge.subprocess.run") + @mock.patch("ctfcli.core.lint.subprocess.run") def test_runs_hadolint(self, mock_run: MagicMock): class RunResult: def __init__(self, return_code): @@ -2058,8 +2059,8 @@ def __init__(self, return_code): } self.assertDictEqual(expected_lint_issues, e.exception.issues) - @mock.patch("ctfcli.core.challenge.subprocess.run") - @mock.patch("ctfcli.core.challenge.click.secho") + @mock.patch("ctfcli.core.lint.subprocess.run") + @mock.patch("ctfcli.core.lint.click.secho") def test_allows_for_skipping_hadolint(self, mock_secho: MagicMock, mock_run: MagicMock, *args, **kwargs): challenge = Challenge(self.dockerfile_challenge) result = challenge.lint(skip_hadolint=True) @@ -2440,13 +2441,15 @@ def test_compare_challenge_module(self, mock_api_constructor: MagicMock): 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")) + prop = get_property("module") + ctx = PropertyContext(challenge) + self.assertTrue(prop.matches(ctx, "Test Module", "Test Module")) + self.assertTrue(prop.matches(ctx, None, None)) + self.assertFalse(prop.matches(ctx, "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")) + self.assertTrue(prop.matches(ctx, 42, "42")) + self.assertFalse(prop.matches(ctx, 42, "Test Module")) @mock.patch("ctfcli.core.challenge.API") def test_verify_checks_if_challenge_is_the_same(self, mock_api_constructor: MagicMock): @@ -2576,90 +2579,93 @@ def test_additional_keys_are_appended(self): class TestChallengeScheduledAt(unittest.TestCase): minimal_challenge = BASE_DIR / "fixtures" / "challenges" / "test-challenge-minimal" / "challenge.yml" + def _prop(self): + return get_property("scheduled_at") + + def _payload(self, challenge, ignore=()): + return self._prop().create_payload(PropertyContext(challenge, ignore=ignore)) + 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"]) + parsed = self._prop().parse("2026-06-15T12:00:00+00:00") 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"]) + parsed = self._prop().parse("2026-06-15T12:00:00Z") 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"]) + parsed = self._prop().parse(datetime(2026, 6, 15, 14, 0, 0, tzinfo=timezone.utc)) 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("")) + self.assertIsNone(self._prop().parse(None)) + self.assertIsNone(self._prop().parse("")) 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._prop().parse("2026-06-15T12:00:00") 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"]) + self._prop().parse(datetime(2026, 6, 15, 12, 0, 0)) # noqa: DTZ001 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"]) + self._prop().parse("not-a-date") 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") + self.assertEqual(self._payload(challenge)["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"]) + self.assertIsNone(self._payload(challenge)["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) + self.assertNotIn("scheduled_at", self._payload(challenge, ignore=("scheduled_at",))) 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() + self._payload(challenge) def test_normalize_makes_utc_explicit(self): # CTFd returns naive UTC; ctfcli should write it back with an explicit offset + self.assertEqual(self._prop().normalize("2026-06-15T12:00:00"), "2026-06-15T12:00:00+00:00") + + def test_normalize_returns_none_for_none(self): + self.assertIsNone(self._prop().normalize(None)) + + def test_pull_normalizes_remote_value(self): + challenge = Challenge(self.minimal_challenge) + ctx = PropertyContext(challenge) self.assertEqual( - Challenge._normalize_scheduled_at("2026-06-15T12:00:00"), + self._prop().pull(ctx, {"scheduled_at": "2026-06-15T12:00:00"}), "2026-06-15T12:00:00+00:00", ) + self.assertIsNone(self._prop().pull(ctx, {})) - def test_normalize_returns_none_for_none(self): - self.assertIsNone(Challenge._normalize_scheduled_at(None)) + def test_is_default_only_for_none(self): + self.assertTrue(self._prop().is_default(None)) + self.assertFalse(self._prop().is_default("2026-06-15T12:00:00+00:00")) - def test_compare_equal_for_same_instant_different_offsets(self): - challenge = Challenge(self.minimal_challenge) + def test_matches_equal_for_same_instant_different_offsets(self): + ctx = PropertyContext(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")) + self.assertTrue(self._prop().matches(ctx, "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_matches_not_equal_for_different_instants(self): + ctx = PropertyContext(Challenge(self.minimal_challenge)) + self.assertFalse(self._prop().matches(ctx, "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_matches_handles_none(self): + ctx = PropertyContext(Challenge(self.minimal_challenge)) + self.assertTrue(self._prop().matches(ctx, None, None)) + self.assertFalse(self._prop().matches(ctx, "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"})