updated sheesh
This commit is contained in:
@@ -4,6 +4,7 @@ import subprocess
|
||||
import time
|
||||
import re
|
||||
import os
|
||||
import binascii
|
||||
|
||||
class Sheesh(Challenge):
|
||||
flag_location = 'flags/sheesh.txt'
|
||||
@@ -14,8 +15,10 @@ class Sheesh(Challenge):
|
||||
_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)
|
||||
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()
|
||||
@@ -30,23 +33,34 @@ class Sheesh(Challenge):
|
||||
bufsize=0,
|
||||
)
|
||||
|
||||
def _read_until(self, proc, token, timeout=5.0, max_bytes=1_000_000):
|
||||
def _read_exact_line(self, proc, timeout=5.0):
|
||||
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}")
|
||||
raise TimeoutError("Timeout waiting for a line")
|
||||
ch = r(1)
|
||||
if ch == "" and proc.poll() is not None:
|
||||
raise RuntimeError(f"Process ended while waiting for '{token}'. Output:\n{''.join(buf)}")
|
||||
raise RuntimeError("Process ended unexpectedly")
|
||||
buf.append(ch)
|
||||
if len(buf) > max_bytes:
|
||||
raise RuntimeError("Exceeded max read size")
|
||||
if token in "".join(buf):
|
||||
if ch == "\n":
|
||||
return "".join(buf)
|
||||
|
||||
def _read_until_prompt(self, proc, token: str, timeout=5.0):
|
||||
start = time.time()
|
||||
acc = []
|
||||
r = proc.stdout.read
|
||||
while True:
|
||||
if time.time() - start > timeout:
|
||||
raise TimeoutError(f"Timeout waiting for '{token}'")
|
||||
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()
|
||||
@@ -76,6 +90,7 @@ class Sheesh(Challenge):
|
||||
|
||||
def check(self):
|
||||
try:
|
||||
# parity check: host vs container flag
|
||||
with open(self.flag_location, 'r') as f:
|
||||
host_flag = f.read().strip()
|
||||
container_flag = self._read_container_flag()
|
||||
@@ -84,96 +99,66 @@ class Sheesh(Challenge):
|
||||
|
||||
proc = self._spawn()
|
||||
|
||||
def menu():
|
||||
self._read_until(proc, "> ", timeout=5.0)
|
||||
line1 = self._read_exact_line(proc, timeout=5.0)
|
||||
line2 = self._read_exact_line(proc, timeout=5.0)
|
||||
blank = self._read_exact_line(proc, timeout=5.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")
|
||||
|
||||
menu()
|
||||
self._read_until_prompt(proc, "blob: ", timeout=5.0)
|
||||
|
||||
self._send_line(proc, "1")
|
||||
self._read_until(proc, "pt: ", timeout=3.0)
|
||||
pt_hex = "414243444546"
|
||||
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")
|
||||
qA = "00" * 16
|
||||
self._send_line(proc, qA)
|
||||
lA = self._read_exact_line(proc, timeout=3.0)
|
||||
assert lA.startswith("blk:"), "Expected blk: for 16-byte blob"
|
||||
blk1 = self._expect_hex_field(lA, "blk")
|
||||
_ = self._read_exact_line(proc, timeout=3.0)
|
||||
|
||||
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")
|
||||
self._read_until_prompt(proc, "blob: ", timeout=5.0)
|
||||
self._send_line(proc, qA)
|
||||
lA2 = self._read_exact_line(proc, timeout=3.0)
|
||||
blk2 = self._expect_hex_field(lA2, "blk")
|
||||
_ = self._read_exact_line(proc, timeout=3.0)
|
||||
assert blk1 == blk2, "ECB-masked oracle not deterministic for same input"
|
||||
self.logger.info("[ok] blk determinism")
|
||||
|
||||
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)"
|
||||
self._read_until_prompt(proc, "blob: ", timeout=5.0)
|
||||
qB = "01" * 16
|
||||
self._send_line(proc, qB)
|
||||
lB = self._read_exact_line(proc, timeout=3.0)
|
||||
blk3 = self._expect_hex_field(lB, "blk")
|
||||
_ = self._read_exact_line(proc, timeout=3.0)
|
||||
assert blk3 != blk1, "ECB-masked oracle returned same block for different input"
|
||||
self.logger.info("[ok] blk diversity")
|
||||
|
||||
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)"
|
||||
self._read_until_prompt(proc, "blob: ", timeout=5.0)
|
||||
self._send_line(proc, "zzzz") # invalid hex
|
||||
line_err = self._read_exact_line(proc, timeout=3.0)
|
||||
assert "err" in line_err.lower(), "Expected 'err' for non-hex input"
|
||||
self.logger.info("[ok] error handling on non-hex")
|
||||
|
||||
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)")
|
||||
self._read_until_prompt(proc, "blob: ", timeout=5.0)
|
||||
self._send_line(proc, (iv_hex + ct_hex))
|
||||
ans1 = self._read_exact_line(proc, timeout=3.0).strip().lower()
|
||||
assert ans1 in ("ok", "no"), "Padding oracle did not return ok/no for iv||ct"
|
||||
self.logger.info(f"[ok] padding oracle responds ({ans1}) on banner bundle")
|
||||
|
||||
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")
|
||||
self._read_until_prompt(proc, "blob: ", timeout=5.0)
|
||||
iv_synth = os.urandom(16).hex()
|
||||
ct_synth = ("00" * (6 * 16))
|
||||
self._send_line(proc, iv_synth + ct_synth)
|
||||
ans2 = self._read_exact_line(proc, timeout=3.0).strip().lower()
|
||||
assert ans2 in ("ok", "no"), "Padding oracle did not return ok/no on synthetic 6-block ct"
|
||||
self.logger.info(f"[ok] padding oracle responds ({ans2}) on synthetic payload")
|
||||
|
||||
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")
|
||||
self._read_until_prompt(proc, "blob: ", timeout=5.0)
|
||||
|
||||
proc.kill()
|
||||
self.logger.info('Check passed for sheesh')
|
||||
return True
|
||||
|
||||
|
||||
Reference in New Issue
Block a user