From c9a2792481bf550665cee9ed9fcb5fc5b70f27c0 Mon Sep 17 00:00:00 2001 From: itoid Date: Sat, 25 Oct 2025 17:39:54 +0700 Subject: [PATCH] Add files via upload --- receiver/challenges/Phew.py | 192 ++++++++++++++++++++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 receiver/challenges/Phew.py diff --git a/receiver/challenges/Phew.py b/receiver/challenges/Phew.py new file mode 100644 index 0000000..f7dc767 --- /dev/null +++ b/receiver/challenges/Phew.py @@ -0,0 +1,192 @@ +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 = "phew_container" + _SERVICE_CMD = ["docker", "exec", "-i", _CONTAINER, "python3", "/home/ctfuser/chall/src/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 (phew)') + + def run_encrypt_once(pt_hex: str) -> str: + proc = self._spawn() + try: + self._read_until(proc, "> ", timeout=5.0) + 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, pt_hex) + out = self._read_until(proc, "\n", timeout=3.0) + ct_hex = self._expect_hex_field(out, "ct") + self._read_until(proc, "> ", timeout=3.0) + 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 (encrypt)") + return ct_hex + finally: + if proc.poll() is None: + proc.kill() + + def run_decrypt_once(ct_hex: str) -> str: + proc = self._spawn() + try: + self._read_until(proc, "> ", timeout=5.0) + 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 = self._read_until(proc, "\n", timeout=3.0) + pt_hex = self._expect_hex_field(out, "pt") + self._read_until(proc, "> ", timeout=3.0) + 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 (decrypt)") + return pt_hex + finally: + if proc.poll() is None: + proc.kill() + + def run_bingo_once() -> str: + proc = self._spawn() + try: + self._read_until(proc, "> ", timeout=5.0) + self._send_line(proc, "2") + out = self._read_until(proc, "\n", timeout=3.0) + ct_hex = self._expect_hex_field(out, "ct") + # exit fast + self._read_until(proc, "> ", timeout=3.0) + 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 (bingo)") + return ct_hex + finally: + if proc.poll() is None: + proc.kill() + + ct1 = run_encrypt_once("414243444546") # "ABCDEF" + assert ct1 and self._HEX_RE.match(ct1), "encrypt(1) did not return hex" + self.logger.info("[ok] encrypt produced hex") + + pt_back = run_decrypt_once(ct1) + assert pt_back.strip() != "", "decrypt returned empty output" + assert self._HEX_RE.match(pt_back), "decrypt(3) did not return hex" + self.logger.info("[ok] decrypt produced hex (custom mapping accepted)") + + pt_same = "01" * 8 + ct_a = run_encrypt_once(pt_same) + ct_b = run_encrypt_once(pt_same) + assert ct_a.lower() != ct_b.lower(), "Encryption appears deterministic for same plaintext" + self.logger.info("[ok] encrypt randomness") + + b1 = run_bingo_once() + b2 = run_bingo_once() + assert b1.lower() != b2.lower(), "Bingo ciphertexts reused randomness" + self.logger.info("[ok] bingo randomness") + + import time as _time + start = _time.time() + proc_alarm = self._spawn() + try: + try: + proc_alarm.wait(timeout=20.0) + except subprocess.TimeoutExpired: + proc_alarm.kill() + raise AssertionError("Alarm did not fire within 20s (possible patch/removal)") + elapsed = _time.time() - start + self.logger.info(f"[info] observed lifetime for alarm: {elapsed:.2f}s") + if not (8.0 <= elapsed <= 12.0): + raise AssertionError(f"Alarm window changed (expected ~10s): observed {elapsed:.2f}s") + self.logger.info("[ok] signal.alarm(10) integrity") + finally: + if proc_alarm.poll() is None: + proc_alarm.kill() + + self.logger.info('Check passed for phew (custom cipher + alarm intact)') + return True + + except Exception as e: + self.logger.error(f'Could not check phew: {e}') + return False