From 0d216d77eb58d6f41e4dc31aac3c50a69bc21e3c Mon Sep 17 00:00:00 2001 From: rafliher Date: Sat, 25 Oct 2025 14:48:20 +0700 Subject: [PATCH] try checker --- .../sla.py => receiver/challenges/Blogspot.py | 6 +- .../cdn/sla.py => receiver/challenges/CDN.py | 25 +- receiver/challenges/Challenge.py | 1 - .../{test_phew_standalone.py => Phew.py} | 364 ++++++++--------- .../{test_sheesh_standalone.py => Sheesh.py} | 382 +++++++++--------- receiver/challenges/Wanderer.py | 88 ---- receiver/challenges/Warmup.py | 41 ++ receiver/main.py | 16 +- services/docker-compose.yml | 2 - starter.py | 12 +- 10 files changed, 457 insertions(+), 480 deletions(-) rename services/blogpost/sla.py => receiver/challenges/Blogspot.py (98%) rename services/cdn/sla.py => receiver/challenges/CDN.py (90%) rename receiver/challenges/{test_phew_standalone.py => Phew.py} (97%) rename receiver/challenges/{test_sheesh_standalone.py => Sheesh.py} (97%) delete mode 100644 receiver/challenges/Wanderer.py create mode 100644 receiver/challenges/Warmup.py diff --git a/services/blogpost/sla.py b/receiver/challenges/Blogspot.py similarity index 98% rename from services/blogpost/sla.py rename to receiver/challenges/Blogspot.py index 43ffbd3..ad478cf 100644 --- a/services/blogpost/sla.py +++ b/receiver/challenges/Blogspot.py @@ -15,7 +15,7 @@ from PIL import Image # pillow for tiny PNG generation from .Challenge import Challenge -class WebAppSLA(Challenge): +class Blogspot(Challenge): """ SLA checker for the provided Flask challenge app. @@ -26,8 +26,8 @@ class WebAppSLA(Challenge): - Login flow at /login, register at /register - Create post at /create, view post at /post/, uploads at /uploads/ """ - flag_location = 'flags/webapp.txt' # Host copy (used by your orchestrator) - history_location = 'history/webapp.txt' + flag_location = 'flags/blogspot.txt' # Host copy (used by your orchestrator) + history_location = 'history/blogspot.txt' container_flag_path = '/flag.txt' container_name = 'blogpost_container' # <-- set to your actual container name diff --git a/services/cdn/sla.py b/receiver/challenges/CDN.py similarity index 90% rename from services/cdn/sla.py rename to receiver/challenges/CDN.py index 9c22e9c..df9f8cd 100644 --- a/services/cdn/sla.py +++ b/receiver/challenges/CDN.py @@ -12,7 +12,7 @@ from PIL import Image from .Challenge import Challenge -class WebAppSLA(Challenge): +class CDN(Challenge): """ SLA checker for the Flask app. @@ -23,8 +23,8 @@ class WebAppSLA(Challenge): """ # Paths & infra knobs - flag_location = 'flags/webapp.txt' # host copy (written by your orchestrator) - history_location = 'history/webapp.txt' + flag_location = 'flags/cdn.txt' # host copy (written by your orchestrator) + history_location = 'history/cdn.txt' container_flag_path = '/flag.txt' container_name = 'chal_app' # adjust if your container name differs @@ -34,6 +34,25 @@ class WebAppSLA(Challenge): 'Date Created', ) + # --- Flag distribution hook (optional, mirrors your example) --- + def distribute(self, flag: str) -> bool: + """ + Writes/records the current flag on the host. Your infra may separately + mount/copy it into the container; this class *also* verifies existence + inside the container during .check(). + """ + self._make_logger() + try: + with open(self.flag_location, 'w') as f: + f.write(flag) + with open(self.history_location, 'a') as f: + f.write(flag + '\n') + self.logger.info(f"Flag '{flag}' written to {self.flag_location}") + return True + except Exception as e: + self.logger.error(f"Failed writing host flag: {e}") + return False + def _make_logger(self): if not hasattr(self, 'logger') or self.logger is None: self.logger = logging.getLogger(self.__class__.__name__) diff --git a/receiver/challenges/Challenge.py b/receiver/challenges/Challenge.py index be7d8e0..c90ed1f 100644 --- a/receiver/challenges/Challenge.py +++ b/receiver/challenges/Challenge.py @@ -4,7 +4,6 @@ import string from config import get_settings - class Challenge(object): name = __name__ settings = get_settings() diff --git a/receiver/challenges/test_phew_standalone.py b/receiver/challenges/Phew.py similarity index 97% rename from receiver/challenges/test_phew_standalone.py rename to receiver/challenges/Phew.py index 6f10e36..45e2889 100644 --- a/receiver/challenges/test_phew_standalone.py +++ b/receiver/challenges/Phew.py @@ -1,182 +1,182 @@ -from .Challenge import Challenge - -import subprocess -import time -import re -import os - -class PailierBingo(Challenge): - flag_location = 'flags/phew.txt' - history_location = 'history/phew.txt' - - _CONTAINER = "pailier_container" # <-- set to your container name - _SERVICE_CMD = ["docker", "exec", "-i", _CONTAINER, "python3", "/usr/local/bin/chall.py"] - _HEX_RE = re.compile(r'^[0-9a-fA-F]+$') - - def _read_container_flag(self) -> str: - out = subprocess.run(["docker", "exec", self._CONTAINER, "cat", "/flag.txt"], - capture_output=True, text=True) - if out.returncode != 0 or not out.stdout.strip(): - raise FileNotFoundError("Flag not found in container (/flag.txt)") - return out.stdout.strip() - - def _spawn(self): - return subprocess.Popen( - self._SERVICE_CMD, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - bufsize=0, - ) - - def _read_until(self, proc, token, timeout=5.0, max_bytes=1_000_000): - start = time.time() - buf = [] - r = proc.stdout.read - while True: - if time.time() - start > timeout: - tail = ''.join(buf)[-500:] - raise TimeoutError(f"Timeout waiting for '{token}'. Got so far:\n{tail}") - ch = r(1) - if ch == "" and proc.poll() is not None: - raise RuntimeError(f"Process ended while waiting for '{token}'. Output:\n{''.join(buf)}") - buf.append(ch) - if len(buf) > max_bytes: - raise RuntimeError("Exceeded max read size") - if token in "".join(buf): - return "".join(buf) - - def _send_line(self, proc, s: str): - proc.stdin.write(s + "\n") - proc.stdin.flush() - - def _expect_hex_field(self, text: str, label: str) -> str: - """ - Extracts 'ct : ' or 'pt : ' (label = 'ct' or 'pt') - The service prints with spaces around the colon: 'ct : ' - """ - m = re.search(rf"{re.escape(label)}\s*:\s*([0-9a-fA-F]+)", text) - assert m, f"Missing '{label}' in output. Tail:\n{text[-400:]}" - hx = m.group(1) - assert self._HEX_RE.match(hx), f"{label} is not hex" - return hx - - def distribute(self, flag): - try: - os.makedirs(os.path.dirname(self.flag_location), exist_ok=True) - with open(self.flag_location, 'w') as f: - f.write(flag) - - os.makedirs(os.path.dirname(self.history_location), exist_ok=True) - with open(self.history_location, 'a') as f: - f.write(flag + '\n') - - self.logger.info(f'Flag {flag} written to {self.flag_location}') - return True - except Exception as e: - self.logger.error(f'Could not write flag to {self.flag_location}: {e}') - return False - - def check(self): - try: - with open(self.flag_location, 'r') as f: - host_flag = f.read().strip() - container_flag = self._read_container_flag() - assert host_flag == container_flag, 'Flag mismatch between host and container' - self.logger.info('[ok] flag parity (phew)') - - proc = self._spawn() - - def menu(): - self._read_until(proc, "> ", timeout=5.0) - - menu() - self._send_line(proc, "1") - self._read_until(proc, "pt (hex)", timeout=3.0) - self._read_until(proc, "> ", timeout=3.0) - pt_hex = "414243444546" # "ABCDEF" - self._send_line(proc, pt_hex) - out = self._read_until(proc, "\n", timeout=3.0) - ct_hex = self._expect_hex_field(out, "ct") - self.logger.info("[ok] encrypt(1) produced hex CT") - - menu() - self._send_line(proc, "3") - self._read_until(proc, "ct (hex)", timeout=3.0) - self._read_until(proc, "> ", timeout=3.0) - self._send_line(proc, ct_hex) - out_dec = self._read_until(proc, "\n", timeout=3.0) - pt_back_hex = self._expect_hex_field(out_dec, "pt") - # Normalize leading zeros for comparison - def norm(h): return (h.lstrip("0") or "0").lower() - assert norm(pt_back_hex) == norm(pt_hex), f"decrypt(encrypt(pt)) mismatch: {pt_back_hex} != {pt_hex}" - self.logger.info("[ok] decrypt(3) round-trip") - - menu() - self._send_line(proc, "1") - self._read_until(proc, "pt (hex)", timeout=3.0) - self._read_until(proc, "> ", timeout=3.0) - same_pt = "01" * 8 - self._send_line(proc, same_pt) - out_a = self._read_until(proc, "\n", timeout=3.0) - ct_a = self._expect_hex_field(out_a, "ct") - - menu() - self._send_line(proc, "1") - self._read_until(proc, "pt (hex)", timeout=3.0) - self._read_until(proc, "> ", timeout=3.0) - self._send_line(proc, same_pt) - out_b = self._read_until(proc, "\n", timeout=3.0) - ct_b = self._expect_hex_field(out_b, "ct") - - assert ct_a.lower() != ct_b.lower(), "Encryption appears deterministic for same plaintext" - self.logger.info("[ok] encrypt(1) randomness") - - menu() - self._send_line(proc, "2") - out2a = self._read_until(proc, "\n", timeout=3.0) - bingo1 = self._expect_hex_field(out2a, "ct") - - menu() - self._send_line(proc, "2") - out2b = self._read_until(proc, "\n", timeout=3.0) - bingo2 = self._expect_hex_field(out2b, "ct") - - assert bingo1.lower() != bingo2.lower(), "Bingo ciphertexts reused randomness" - self.logger.info("[ok] bingo(2) randomness") - - menu() - self._send_line(proc, "1") - self._read_until(proc, "pt (hex)", timeout=3.0) - self._read_until(proc, "> ", timeout=3.0) - pt_hex2 = "deadbeef" - self._send_line(proc, pt_hex2) - out_enc2 = self._read_until(proc, "\n", timeout=3.0) - ct2 = self._expect_hex_field(out_enc2, "ct") - - menu() - self._send_line(proc, "3") - self._read_until(proc, "ct (hex)", timeout=3.0) - self._read_until(proc, "> ", timeout=3.0) - self._send_line(proc, ct2) - out_dec2 = self._read_until(proc, "\n", timeout=3.0) - pt2_back = self._expect_hex_field(out_dec2, "pt") - assert norm(pt2_back) == norm(pt_hex2), f"decrypt(encrypt(pt2)) mismatch: {pt2_back} != {pt_hex2}" - self.logger.info("[ok] decrypt(3) round-trip (case 2)") - - menu() - self._send_line(proc, "4") - try: - proc.wait(timeout=2.0) - except subprocess.TimeoutExpired: - proc.kill() - raise AssertionError("Program did not exit after option 4") - self.logger.info("[ok] service exit on 4") - - self.logger.info('Check passed for phew') - return True - - except Exception as e: - self.logger.error(f'Could not check phew: {e}') - return False +from .Challenge import Challenge + +import subprocess +import time +import re +import os + +class Phew(Challenge): + flag_location = 'flags/phew.txt' + history_location = 'history/phew.txt' + + _CONTAINER = "pailier_container" # <-- set to your container name + _SERVICE_CMD = ["docker", "exec", "-i", _CONTAINER, "python3", "/usr/local/bin/chall.py"] + _HEX_RE = re.compile(r'^[0-9a-fA-F]+$') + + def _read_container_flag(self) -> str: + out = subprocess.run(["docker", "exec", self._CONTAINER, "cat", "/flag.txt"], + capture_output=True, text=True) + if out.returncode != 0 or not out.stdout.strip(): + raise FileNotFoundError("Flag not found in container (/flag.txt)") + return out.stdout.strip() + + def _spawn(self): + return subprocess.Popen( + self._SERVICE_CMD, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=0, + ) + + def _read_until(self, proc, token, timeout=5.0, max_bytes=1_000_000): + start = time.time() + buf = [] + r = proc.stdout.read + while True: + if time.time() - start > timeout: + tail = ''.join(buf)[-500:] + raise TimeoutError(f"Timeout waiting for '{token}'. Got so far:\n{tail}") + ch = r(1) + if ch == "" and proc.poll() is not None: + raise RuntimeError(f"Process ended while waiting for '{token}'. Output:\n{''.join(buf)}") + buf.append(ch) + if len(buf) > max_bytes: + raise RuntimeError("Exceeded max read size") + if token in "".join(buf): + return "".join(buf) + + def _send_line(self, proc, s: str): + proc.stdin.write(s + "\n") + proc.stdin.flush() + + def _expect_hex_field(self, text: str, label: str) -> str: + """ + Extracts 'ct : ' or 'pt : ' (label = 'ct' or 'pt') + The service prints with spaces around the colon: 'ct : ' + """ + m = re.search(rf"{re.escape(label)}\s*:\s*([0-9a-fA-F]+)", text) + assert m, f"Missing '{label}' in output. Tail:\n{text[-400:]}" + hx = m.group(1) + assert self._HEX_RE.match(hx), f"{label} is not hex" + return hx + + def distribute(self, flag): + try: + os.makedirs(os.path.dirname(self.flag_location), exist_ok=True) + with open(self.flag_location, 'w') as f: + f.write(flag) + + os.makedirs(os.path.dirname(self.history_location), exist_ok=True) + with open(self.history_location, 'a') as f: + f.write(flag + '\n') + + self.logger.info(f'Flag {flag} written to {self.flag_location}') + return True + except Exception as e: + self.logger.error(f'Could not write flag to {self.flag_location}: {e}') + return False + + def check(self): + try: + with open(self.flag_location, 'r') as f: + host_flag = f.read().strip() + container_flag = self._read_container_flag() + assert host_flag == container_flag, 'Flag mismatch between host and container' + self.logger.info('[ok] flag parity (phew)') + + proc = self._spawn() + + def menu(): + self._read_until(proc, "> ", timeout=5.0) + + menu() + self._send_line(proc, "1") + self._read_until(proc, "pt (hex)", timeout=3.0) + self._read_until(proc, "> ", timeout=3.0) + pt_hex = "414243444546" # "ABCDEF" + self._send_line(proc, pt_hex) + out = self._read_until(proc, "\n", timeout=3.0) + ct_hex = self._expect_hex_field(out, "ct") + self.logger.info("[ok] encrypt(1) produced hex CT") + + menu() + self._send_line(proc, "3") + self._read_until(proc, "ct (hex)", timeout=3.0) + self._read_until(proc, "> ", timeout=3.0) + self._send_line(proc, ct_hex) + out_dec = self._read_until(proc, "\n", timeout=3.0) + pt_back_hex = self._expect_hex_field(out_dec, "pt") + # Normalize leading zeros for comparison + def norm(h): return (h.lstrip("0") or "0").lower() + assert norm(pt_back_hex) == norm(pt_hex), f"decrypt(encrypt(pt)) mismatch: {pt_back_hex} != {pt_hex}" + self.logger.info("[ok] decrypt(3) round-trip") + + menu() + self._send_line(proc, "1") + self._read_until(proc, "pt (hex)", timeout=3.0) + self._read_until(proc, "> ", timeout=3.0) + same_pt = "01" * 8 + self._send_line(proc, same_pt) + out_a = self._read_until(proc, "\n", timeout=3.0) + ct_a = self._expect_hex_field(out_a, "ct") + + menu() + self._send_line(proc, "1") + self._read_until(proc, "pt (hex)", timeout=3.0) + self._read_until(proc, "> ", timeout=3.0) + self._send_line(proc, same_pt) + out_b = self._read_until(proc, "\n", timeout=3.0) + ct_b = self._expect_hex_field(out_b, "ct") + + assert ct_a.lower() != ct_b.lower(), "Encryption appears deterministic for same plaintext" + self.logger.info("[ok] encrypt(1) randomness") + + menu() + self._send_line(proc, "2") + out2a = self._read_until(proc, "\n", timeout=3.0) + bingo1 = self._expect_hex_field(out2a, "ct") + + menu() + self._send_line(proc, "2") + out2b = self._read_until(proc, "\n", timeout=3.0) + bingo2 = self._expect_hex_field(out2b, "ct") + + assert bingo1.lower() != bingo2.lower(), "Bingo ciphertexts reused randomness" + self.logger.info("[ok] bingo(2) randomness") + + menu() + self._send_line(proc, "1") + self._read_until(proc, "pt (hex)", timeout=3.0) + self._read_until(proc, "> ", timeout=3.0) + pt_hex2 = "deadbeef" + self._send_line(proc, pt_hex2) + out_enc2 = self._read_until(proc, "\n", timeout=3.0) + ct2 = self._expect_hex_field(out_enc2, "ct") + + menu() + self._send_line(proc, "3") + self._read_until(proc, "ct (hex)", timeout=3.0) + self._read_until(proc, "> ", timeout=3.0) + self._send_line(proc, ct2) + out_dec2 = self._read_until(proc, "\n", timeout=3.0) + pt2_back = self._expect_hex_field(out_dec2, "pt") + assert norm(pt2_back) == norm(pt_hex2), f"decrypt(encrypt(pt2)) mismatch: {pt2_back} != {pt_hex2}" + self.logger.info("[ok] decrypt(3) round-trip (case 2)") + + menu() + self._send_line(proc, "4") + try: + proc.wait(timeout=2.0) + except subprocess.TimeoutExpired: + proc.kill() + raise AssertionError("Program did not exit after option 4") + self.logger.info("[ok] service exit on 4") + + self.logger.info('Check passed for phew') + return True + + except Exception as e: + self.logger.error(f'Could not check phew: {e}') + return False diff --git a/receiver/challenges/test_sheesh_standalone.py b/receiver/challenges/Sheesh.py similarity index 97% rename from receiver/challenges/test_sheesh_standalone.py rename to receiver/challenges/Sheesh.py index 3512f6d..3997dbc 100644 --- a/receiver/challenges/test_sheesh_standalone.py +++ b/receiver/challenges/Sheesh.py @@ -1,191 +1,191 @@ -from .Challenge import Challenge - -import subprocess -import time -import re -import os - -class Sheesh(Challenge): - flag_location = 'flags/sheesh.txt' - history_location = 'history/sheesh.txt' - - _CONTAINER = "sheesh_container" - _SERVICE_CMD = ["docker", "exec", "-i", _CONTAINER, "python3", "/usr/local/bin/chall.py"] - _HEX_RE = re.compile(r'^[0-9a-fA-F]+$') - - def _read_container_flag(self) -> str: - out = subprocess.run(["docker", "exec", self._CONTAINER, "cat", "/flag.txt"], - capture_output=True, text=True) - if out.returncode != 0 or not out.stdout.strip(): - raise FileNotFoundError("Flag not found in container (/flag.txt)") - return out.stdout.strip() - - def _spawn(self): - return subprocess.Popen( - self._SERVICE_CMD, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - bufsize=0, - ) - - def _read_until(self, proc, token, timeout=5.0, max_bytes=1_000_000): - start = time.time() - buf = [] - r = proc.stdout.read - while True: - if time.time() - start > timeout: - tail = ''.join(buf)[-500:] - raise TimeoutError(f"Timeout waiting for '{token}'. Got so far:\n{tail}") - ch = r(1) - if ch == "" and proc.poll() is not None: - raise RuntimeError(f"Process ended while waiting for '{token}'. Output:\n{''.join(buf)}") - buf.append(ch) - if len(buf) > max_bytes: - raise RuntimeError("Exceeded max read size") - if token in "".join(buf): - return "".join(buf) - - def _send_line(self, proc, s: str): - proc.stdin.write(s + "\n") - proc.stdin.flush() - - def _expect_hex_field(self, text: str, label: str) -> str: - m = re.search(rf"{re.escape(label)}\s*:\s*([0-9a-fA-F]+)", text) - assert m, f"Missing '{label}' in output. Tail:\n{text[-400:]}" - hx = m.group(1) - assert self._HEX_RE.match(hx), f"{label} is not hex" - return hx - - def distribute(self, flag): - try: - os.makedirs(os.path.dirname(self.flag_location), exist_ok=True) - with open(self.flag_location, 'w') as f: - f.write(flag) - - os.makedirs(os.path.dirname(self.history_location), exist_ok=True) - with open(self.history_location, 'a') as f: - f.write(flag + '\n') - - self.logger.info(f'Flag {flag} written to {self.flag_location}') - return True - except Exception as e: - self.logger.error(f'Could not write flag to {self.flag_location}: {e}') - return False - - def check(self): - try: - with open(self.flag_location, 'r') as f: - host_flag = f.read().strip() - container_flag = self._read_container_flag() - assert host_flag == container_flag, 'Flag mismatch between host and container' - self.logger.info('[ok] flag parity (sheesh)') - - proc = self._spawn() - - def menu(): - self._read_until(proc, "> ", timeout=5.0) - - menu() - - self._send_line(proc, "1") - self._read_until(proc, "pt: ", timeout=3.0) - pt_hex = "414243444546" # "ABCDEF" - self._send_line(proc, pt_hex) - out = self._read_until(proc, "\n\n", timeout=3.0) - ct_hex = self._expect_hex_field(out, "ct") - ct = bytes.fromhex(ct_hex) - assert len(ct) == 16 + len(bytes.fromhex(pt_hex)), "CFB ct length mismatch (IV+PT)" - iv_a = ct[:16] - self.logger.info("[ok] encrypt(1) basic") - - menu() - self._send_line(proc, "1") - self._read_until(proc, "pt: ", timeout=3.0) - pt_hex2 = "00" * 8 - self._send_line(proc, pt_hex2) - out2 = self._read_until(proc, "\n\n", timeout=3.0) - ct2_hex = self._expect_hex_field(out2, "ct") - ct2 = bytes.fromhex(ct2_hex) - assert len(ct2) == 16 + len(bytes.fromhex(pt_hex2)), "Second CFB ct length mismatch" - iv_b = ct2[:16] - assert iv_a != iv_b, "CFB IV appears reused" - self.logger.info("[ok] encrypt(1) IV rotates") - - menu() - self._send_line(proc, "3") - self._read_until(proc, "pt: ", timeout=3.0) - self._send_line(proc, "00" * 15) - out3a = self._read_until(proc, "\n", timeout=3.0) - assert "hmmm" in out3a.lower(), "Option 3 must reject non-16-byte input (15B)" - - menu() - self._send_line(proc, "3") - self._read_until(proc, "pt: ", timeout=3.0) - self._send_line(proc, "00" * 17) - out3b = self._read_until(proc, "\n", timeout=3.0) - assert "hmmm" in out3b.lower(), "Option 3 must reject non-16-byte input (17B)" - - menu() - self._send_line(proc, "3") - self._read_until(proc, "pt: ", timeout=3.0) - self._send_line(proc, "11" * 16) - out3 = self._read_until(proc, "\n\n", timeout=3.0) - ct3_hex = self._expect_hex_field(out3, "ct") - ct3 = bytes.fromhex(ct3_hex) - assert len(ct3) in (16, 32), f"Unexpected CT length for option 3: {len(ct3)} (expected 16 or 32)" - if len(ct3) == 16: - self.logger.info("[ok] get third(3): vuln flavor (16-byte CT)") - else: - self.logger.info("[ok] get third(3): patched flavor (32-byte CT)") - - menu() - self._send_line(proc, "3") - out4 = self._read_until(proc, "\n", timeout=3.0) - assert "sheesh" in out4.lower(), "Second call to option 3 should be rejected with 'sheesh'" - self.logger.info("[ok] get third(3) lockout") - - menu() - self._send_line(proc, "2") - out5 = self._read_until(proc, "\n\n", timeout=3.0) - iv1_hex = self._expect_hex_field(out5, "iv1") - iv2_hex = self._expect_hex_field(out5, "iv2") - ct5_hex = self._expect_hex_field(out5, "ct") - assert len(bytes.fromhex(iv1_hex)) == 16, "iv1 length invalid" - assert len(bytes.fromhex(iv2_hex)) == 16, "iv2 length invalid" - assert len(bytes.fromhex(ct5_hex)) >= 16 and len(bytes.fromhex(ct5_hex)) % 16 == 0, "CBC ct invalid length" - self.logger.info("[ok] profit(2) first call") - - menu() - self._send_line(proc, "2") - out6 = self._read_until(proc, "\n\n", timeout=3.0) - iv1_hex_2 = self._expect_hex_field(out6, "iv1") - iv2_hex_2 = self._expect_hex_field(out6, "iv2") - assert iv1_hex_2 != iv1_hex or iv2_hex_2 != iv2_hex, "IVs did not change between consecutive '2' calls" - self.logger.info("[ok] profit(2) fresh IVs") - - menu() - self._send_line(proc, "4") - try: - proc.wait(timeout=2.0) - except subprocess.TimeoutExpired: - proc.kill() - raise AssertionError("Program did not exit after option 4") - self.logger.info("[ok] service exit on 4") - - proc_alarm = self._spawn() - self._read_until(proc_alarm, "zzz", timeout=190.0) # 180s + slack - try: - proc_alarm.wait(timeout=5.0) - except subprocess.TimeoutExpired: - proc_alarm.kill() - raise AssertionError("Alarm fired but process did not exit") - self.logger.info("[ok] alarm fired ('zzz') and process self-terminated") - - self.logger.info('Check passed for sheesh') - return True - - except Exception as e: - self.logger.error(f'Could not check sheesh: {e}') - return False +from .Challenge import Challenge + +import subprocess +import time +import re +import os + +class Sheesh(Challenge): + flag_location = 'flags/sheesh.txt' + history_location = 'history/sheesh.txt' + + _CONTAINER = "sheesh_container" + _SERVICE_CMD = ["docker", "exec", "-i", _CONTAINER, "python3", "/usr/local/bin/chall.py"] + _HEX_RE = re.compile(r'^[0-9a-fA-F]+$') + + def _read_container_flag(self) -> str: + out = subprocess.run(["docker", "exec", self._CONTAINER, "cat", "/flag.txt"], + capture_output=True, text=True) + if out.returncode != 0 or not out.stdout.strip(): + raise FileNotFoundError("Flag not found in container (/flag.txt)") + return out.stdout.strip() + + def _spawn(self): + return subprocess.Popen( + self._SERVICE_CMD, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=0, + ) + + def _read_until(self, proc, token, timeout=5.0, max_bytes=1_000_000): + start = time.time() + buf = [] + r = proc.stdout.read + while True: + if time.time() - start > timeout: + tail = ''.join(buf)[-500:] + raise TimeoutError(f"Timeout waiting for '{token}'. Got so far:\n{tail}") + ch = r(1) + if ch == "" and proc.poll() is not None: + raise RuntimeError(f"Process ended while waiting for '{token}'. Output:\n{''.join(buf)}") + buf.append(ch) + if len(buf) > max_bytes: + raise RuntimeError("Exceeded max read size") + if token in "".join(buf): + return "".join(buf) + + def _send_line(self, proc, s: str): + proc.stdin.write(s + "\n") + proc.stdin.flush() + + def _expect_hex_field(self, text: str, label: str) -> str: + m = re.search(rf"{re.escape(label)}\s*:\s*([0-9a-fA-F]+)", text) + assert m, f"Missing '{label}' in output. Tail:\n{text[-400:]}" + hx = m.group(1) + assert self._HEX_RE.match(hx), f"{label} is not hex" + return hx + + def distribute(self, flag): + try: + os.makedirs(os.path.dirname(self.flag_location), exist_ok=True) + with open(self.flag_location, 'w') as f: + f.write(flag) + + os.makedirs(os.path.dirname(self.history_location), exist_ok=True) + with open(self.history_location, 'a') as f: + f.write(flag + '\n') + + self.logger.info(f'Flag {flag} written to {self.flag_location}') + return True + except Exception as e: + self.logger.error(f'Could not write flag to {self.flag_location}: {e}') + return False + + def check(self): + try: + with open(self.flag_location, 'r') as f: + host_flag = f.read().strip() + container_flag = self._read_container_flag() + assert host_flag == container_flag, 'Flag mismatch between host and container' + self.logger.info('[ok] flag parity (sheesh)') + + proc = self._spawn() + + def menu(): + self._read_until(proc, "> ", timeout=5.0) + + menu() + + self._send_line(proc, "1") + self._read_until(proc, "pt: ", timeout=3.0) + pt_hex = "414243444546" # "ABCDEF" + self._send_line(proc, pt_hex) + out = self._read_until(proc, "\n\n", timeout=3.0) + ct_hex = self._expect_hex_field(out, "ct") + ct = bytes.fromhex(ct_hex) + assert len(ct) == 16 + len(bytes.fromhex(pt_hex)), "CFB ct length mismatch (IV+PT)" + iv_a = ct[:16] + self.logger.info("[ok] encrypt(1) basic") + + menu() + self._send_line(proc, "1") + self._read_until(proc, "pt: ", timeout=3.0) + pt_hex2 = "00" * 8 + self._send_line(proc, pt_hex2) + out2 = self._read_until(proc, "\n\n", timeout=3.0) + ct2_hex = self._expect_hex_field(out2, "ct") + ct2 = bytes.fromhex(ct2_hex) + assert len(ct2) == 16 + len(bytes.fromhex(pt_hex2)), "Second CFB ct length mismatch" + iv_b = ct2[:16] + assert iv_a != iv_b, "CFB IV appears reused" + self.logger.info("[ok] encrypt(1) IV rotates") + + menu() + self._send_line(proc, "3") + self._read_until(proc, "pt: ", timeout=3.0) + self._send_line(proc, "00" * 15) + out3a = self._read_until(proc, "\n", timeout=3.0) + assert "hmmm" in out3a.lower(), "Option 3 must reject non-16-byte input (15B)" + + menu() + self._send_line(proc, "3") + self._read_until(proc, "pt: ", timeout=3.0) + self._send_line(proc, "00" * 17) + out3b = self._read_until(proc, "\n", timeout=3.0) + assert "hmmm" in out3b.lower(), "Option 3 must reject non-16-byte input (17B)" + + menu() + self._send_line(proc, "3") + self._read_until(proc, "pt: ", timeout=3.0) + self._send_line(proc, "11" * 16) + out3 = self._read_until(proc, "\n\n", timeout=3.0) + ct3_hex = self._expect_hex_field(out3, "ct") + ct3 = bytes.fromhex(ct3_hex) + assert len(ct3) in (16, 32), f"Unexpected CT length for option 3: {len(ct3)} (expected 16 or 32)" + if len(ct3) == 16: + self.logger.info("[ok] get third(3): vuln flavor (16-byte CT)") + else: + self.logger.info("[ok] get third(3): patched flavor (32-byte CT)") + + menu() + self._send_line(proc, "3") + out4 = self._read_until(proc, "\n", timeout=3.0) + assert "sheesh" in out4.lower(), "Second call to option 3 should be rejected with 'sheesh'" + self.logger.info("[ok] get third(3) lockout") + + menu() + self._send_line(proc, "2") + out5 = self._read_until(proc, "\n\n", timeout=3.0) + iv1_hex = self._expect_hex_field(out5, "iv1") + iv2_hex = self._expect_hex_field(out5, "iv2") + ct5_hex = self._expect_hex_field(out5, "ct") + assert len(bytes.fromhex(iv1_hex)) == 16, "iv1 length invalid" + assert len(bytes.fromhex(iv2_hex)) == 16, "iv2 length invalid" + assert len(bytes.fromhex(ct5_hex)) >= 16 and len(bytes.fromhex(ct5_hex)) % 16 == 0, "CBC ct invalid length" + self.logger.info("[ok] profit(2) first call") + + menu() + self._send_line(proc, "2") + out6 = self._read_until(proc, "\n\n", timeout=3.0) + iv1_hex_2 = self._expect_hex_field(out6, "iv1") + iv2_hex_2 = self._expect_hex_field(out6, "iv2") + assert iv1_hex_2 != iv1_hex or iv2_hex_2 != iv2_hex, "IVs did not change between consecutive '2' calls" + self.logger.info("[ok] profit(2) fresh IVs") + + menu() + self._send_line(proc, "4") + try: + proc.wait(timeout=2.0) + except subprocess.TimeoutExpired: + proc.kill() + raise AssertionError("Program did not exit after option 4") + self.logger.info("[ok] service exit on 4") + + proc_alarm = self._spawn() + self._read_until(proc_alarm, "zzz", timeout=190.0) # 180s + slack + try: + proc_alarm.wait(timeout=5.0) + except subprocess.TimeoutExpired: + proc_alarm.kill() + raise AssertionError("Alarm fired but process did not exit") + self.logger.info("[ok] alarm fired ('zzz') and process self-terminated") + + self.logger.info('Check passed for sheesh') + return True + + except Exception as e: + self.logger.error(f'Could not check sheesh: {e}') + return False diff --git a/receiver/challenges/Wanderer.py b/receiver/challenges/Wanderer.py deleted file mode 100644 index 457f34c..0000000 --- a/receiver/challenges/Wanderer.py +++ /dev/null @@ -1,88 +0,0 @@ -from .Challenge import Challenge - -import io -import requests -import random -import subprocess -import requests -import re - -class Wanderer(Challenge): - flag_location = 'flags/wanderer.txt' - history_location = 'history/wanderer.txt' - - def distribute(self, flag): - try: - with open(self.flag_location, 'w') as f: - f.write(flag) - - with open(self.history_location, 'a') as f: - f.write(flag + '\n') - - self.logger.info(f'Flag {flag} written to {self.flag_location}') - return True - - except Exception as e: - self.logger.error(f'Could not write flag to {self.flag_location}: {e}') - return False - - def check(self): - try: - session = requests.Session() - # Step 1: Check Register - register_url = f'http://localhost:{self.port}/index.php?module=user&action=register' - registration_data = {'username': 'adminwreckitchecker','password': 'adminwreckitchecker'} - response = session.post(register_url, data=registration_data) - assert "username already exists" in response.text or "Registration Success" in response.text or response.status_code == 200 , 'Function Register Failed' - - # Step 2: Check Login - login_url = f'http://localhost:{self.port}/index.php?module=user&action=login' - login_data = {'username': 'adminwreckitchecker', 'password': 'adminwreckitchecker'} - response = session.post(login_url, data=login_data) - assert "Login Success" in response.text or response.status_code == 200, 'Login function failed' - - # Step 3: Check UUID Not hidden - view_page_url = f'http://localhost:{self.port}/index.php?module=page&action=viewPage' - response = session.get(view_page_url) - uuid_pattern = re.compile(r'[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}') - found_uuids = uuid_pattern.findall(response.text) - assert found_uuids, 'UUID not found' - - # Step 4: Check create sample page - sample_page_url = f'http://localhost:{self.port}/index.php?module=page&action=samplePage' - response = session.get(sample_page_url) - assert "Create Success" in response.text or response.status_code == 200, 'Create Sample Failed' - - # Step 5: Check Edit HTML Functionality - edit_url_html = f'http://localhost:{self.port}/index.php?module=user&action=edit&type=html' - response = session.get(edit_url_html) - assert response.status_code == 200, 'Edit html page not accessible' - - # Step 6: Check Edit JS Functionality - edit_url_js = f'http://localhost:{self.port}/index.php?module=user&action=edit&type=js' - response = session.get(edit_url_js) - assert response.status_code == 200, 'Edit js page not accessible' - - # Step 7: Check Edit CSS Functionality - edit_url_css = f'http://localhost:{self.port}/index.php?module=user&action=edit&type=css' - response = session.get(edit_url_css) - assert response.status_code == 200, 'Edit css page not accessible' - - # Step 8: Check flag - with open(self.flag_location, 'r') as f: - host_flag = f.read().strip() - - container_flag = subprocess.run( - ["docker", "exec", "wanderer_container", "cat", "/flag.txt"], - capture_output=True, - text=True - ).stdout.strip() - - assert host_flag == container_flag, 'Flag mismatch between host and container' - - self.logger.info('Check passed for wanderer') - return True - - except Exception as e: - self.logger.error(f'Could not check wanderer: {e}') - return False diff --git a/receiver/challenges/Warmup.py b/receiver/challenges/Warmup.py new file mode 100644 index 0000000..e4ced12 --- /dev/null +++ b/receiver/challenges/Warmup.py @@ -0,0 +1,41 @@ +from .Challenge import Challenge + +import io +import requests +import random +import subprocess + +class Warmup(Challenge): + flag_location = 'flags/warmup.txt' + history_location = 'history/warmup.txt' + + def distribute(self, flag): + try: + with open(self.flag_location, 'w') as f: + f.write(flag) + with open(self.history_location, 'a') as f: + f.write(flag + '\n') + self.logger.info(f'Flag {flag} written to {self.flag_location}') + return True + except Exception as e: + self.logger.error(f'Could not write flag to {self.flag_location}: {e}') + return False + + def check(self): + try: + url = f'http://localhost:{self.port}/' + r = requests.get(url, timeout=5) + assert "Simple file viewing application" in r.text.lower(), 'Warmup app not available' + + with open(self.flag_location, 'r') as f: + host_flag = f.read().strip() + container_flag = subprocess.run([ + "docker", "exec", "warmup_container", "cat", "/flag.txt" + ], capture_output=True, text=True).stdout.strip() + assert host_flag == container_flag, 'Flag mismatch between host and container' + + self.logger.info('Check passed for warmup') + return True + except Exception as e: + self.logger.error(f'Could not check warmup: {e}') + return False \ No newline at end of file diff --git a/receiver/main.py b/receiver/main.py index 8d3804c..cf34af3 100644 --- a/receiver/main.py +++ b/receiver/main.py @@ -3,8 +3,12 @@ from pydantic import BaseModel from fastapi.security import HTTPBasic, HTTPBasicCredentials from config import get_settings -from challenges.Poke import Poke -from challenges.Blinkpdf import BlinkPDF +from challenges.Blogspot import Blogspot +from challenges.Carbeat import Carbeat +from challenges.CDN import CDN +from challenges.Phew import Phew +from challenges.Sheesh import Sheesh +from challenges.Warmup import Warmup import os @@ -13,8 +17,12 @@ security = HTTPBasic() settings = get_settings() challenges = { - "blogpost": Poke(10000), - "cdn": BlinkPDF(11000), + "blogpost": Blogspot(10000), + "carbeat": Carbeat(11000), + "cdn": CDN(12000), + "phew": Phew(13000), + "sheesh": Sheesh(14000), + "warmup": Warmup(15000), } class Flag(BaseModel): diff --git a/services/docker-compose.yml b/services/docker-compose.yml index 6e1c880..ae5a776 100644 --- a/services/docker-compose.yml +++ b/services/docker-compose.yml @@ -1,5 +1,3 @@ -version: '3.1' - services: # --- blogpost --- blogpost: diff --git a/starter.py b/starter.py index bccab3a..f10d743 100644 --- a/starter.py +++ b/starter.py @@ -35,12 +35,12 @@ def main(): os.chdir(os.path.join(cwd, 'services')) os.system(f'docker compose -f docker-compose.yml up --build -d {args.challenges}') - # os.chdir(os.path.join(cwd, 'receiver')) - # os.system('apt-get install -y gcc python3-dev libgmp3-dev libssl-dev libffi-dev build-essential python3-venv') - # os.system('python3 -m venv .') - # os.system('sudo ./bin/activate') - # os.system('sudo ./bin/python3 -m pip install -r requirements.txt') - # os.system('sudo ./bin/python3 -m uvicorn main:app --reload --host 0.0.0.0 --port 80') + os.chdir(os.path.join(cwd, 'receiver')) + os.system('apt-get install -y gcc python3-dev libgmp3-dev libssl-dev libffi-dev build-essential python3-venv') + os.system('python3 -m venv .') + os.system('sudo ./bin/activate') + os.system('sudo ./bin/python3 -m pip install -r requirements.txt') + os.system('sudo ./bin/python3 -m uvicorn main:app --reload --host 0.0.0.0 --port 80') if __name__ == '__main__': main()