from .Challenge import Challenge import subprocess import time import re import os import binascii import secrets class Sheesh(Challenge): flag_location = 'flags/sheesh.txt' history_location = 'history/sheesh.txt' _CONTAINER = "sheesh_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_exact_line(self, proc, timeout=6.0): start = time.time() buf = [] r = proc.stdout.read while True: if time.time() - start > timeout: raise TimeoutError("Timeout waiting for a line") ch = r(1) if ch == "" and proc.poll() is not None: raise RuntimeError("Process ended unexpectedly") buf.append(ch) if ch == "\n": return "".join(buf) def _read_until_prompt(self, proc, token: str, timeout=6.0): start = time.time() acc = [] r = proc.stdout.read while True: if time.time() - start > timeout: raise TimeoutError(f"Timeout waiting for '{token}'")inp ch = r(1) if ch == "" and proc.poll() is not None: raise RuntimeError("Process ended while waiting for prompt") acc.append(ch) if "".join(acc).endswith(token): return "".join(acc) 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 test1(self, proc, b16_hex: str) -> str: assert len(b16_hex) == 32 self._read_until_prompt(proc, "inp: ", timeout=6.0) self._send_line(proc, b16_hex) line = self._read_exact_line(proc, timeout=4.0) assert line.startswith("blk:"), "Expected blk: for 16-byte input" hx = self._expect_hex_field(line, "blk") # chall prints an extra blank line after blk: _ = self._read_exact_line(proc, timeout=4.0) return hx.lower() def test2(self, proc, iv_hex: str, ct_hex: str) -> str: self._read_until_prompt(proc, "inp: ", timeout=6.0) self._send_line(proc, iv_hex + ct_hex) ans = self._read_exact_line(proc, timeout=4.0).strip().lower() assert ans in ("ok", "zzz"), "Padding oracle did not return ok/zzz" return ans def test3(self, proc, iv_hex: str, ct_hex: str, trials=9) -> tuple[int, int, str]: ok_cnt = 0 zzz_cnt = 0 for _ in range(trials): ans = self.test2(proc, iv_hex, ct_hex) if ans == "ok": ok_cnt += 1 else: zzz_cnt += 1 final = "ok" if ok_cnt > zzz_cnt else "zzz" return ok_cnt, zzz_cnt, final 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() line1 = self._read_exact_line(proc, timeout=6.0) line2 = self._read_exact_line(proc, timeout=6.0) blank = self._read_exact_line(proc, timeout=6.0) assert line1.startswith("iv:") and line2.startswith("ct:"), "Missing banner" iv_hex = self._expect_hex_field(line1, "iv") ct_hex = self._expect_hex_field(line2, "ct") iv0 = bytes.fromhex(iv_hex) ct0 = bytes.fromhex(ct_hex) assert len(iv0) == 16 and len(ct0) >= 16 and len(ct0) % 16 == 0, "Bad banner sizes" self.logger.info("[ok] banner parsed") block00 = "00" * 16 blk1 = self.test1(proc, block00) blk2 = self.test1(proc, block00) blk3 = self.test1(proc, block00) assert blk1 == blk2 == blk3, "ECB-masked oracle not deterministic for same input" self.logger.info("[ok] blk determinism (3x same input)") block01 = "01" * 16 blk4 = self.test1(proc, block01) assert blk4 != blk1, "ECB-masked oracle returned same block for different input" self.logger.info("[ok] blk diversity") self._read_until_prompt(proc, "inp: ", timeout=6.0) self._send_line(proc, "zzzz") # invalid hex line_err = self._read_exact_line(proc, timeout=4.0) assert "err" in line_err.lower(), "Expected 'err' for non-hex input" maybe_hmmm = self._read_exact_line(proc, timeout=4.0).strip().lower() if "hmmm" not in maybe_hmmm: self.logger.warning(f"[warn] expected 'hmmm' after err, got: {maybe_hmmm!r}") self.logger.info("[ok] error handling on non-hex input") ok_cnt, zzz_cnt, final = self.test3(proc, iv_hex, ct_hex, trials=9) self.logger.info(f"[ok] padding oracle responded on banner (ok={ok_cnt}, zzz={zzz_cnt}, final={final})") iv_synth = secrets.token_hex(16) ct_synth = "00" * (6 * 16) # 6 blocks ok_cnt2, zzz_cnt2, final2 = self.test3(proc, iv_synth, ct_synth, trials=9) self.logger.info(f"[ok] padding oracle responded on synthetic (ok={ok_cnt2}, zzz={zzz_cnt2}, final={final2})") self._read_until_prompt(proc, "inp: ", timeout=6.0) proc.kill() self.logger.info('Check passed for sheesh') return True except Exception as e: self.logger.error(f'Could not check sheesh: {e}') return False