Delete receiver/challenges/Phew.py
This commit is contained in:
itoid
@@ -1,182 +0,0 @@
|
|||||||
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" # <-- set to your container name
|
|
||||||
_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:
|
|
||||||
"""
|
|
||||||
Extracts 'ct : <hex>' or 'pt : <hex>' (label = 'ct' or 'pt')
|
|
||||||
The service prints with spaces around the colon: 'ct : <hex>'
|
|
||||||
"""
|
|
||||||
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
|
|
||||||
Reference in New Issue
Block a user