Add files via upload
This commit is contained in:
itoid
@@ -0,0 +1,191 @@
|
||||
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", "/usr/local/bin/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
|
||||
Reference in New Issue
Block a user