Add files via upload

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