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", "/home/ctf/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 (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