TL;DR

CBC 상태를 재활용하는 encryption oracle을 이용해 secret을 바이트 단위로 추출했다. PoW를 브루트포스로 해결하고 연속 블록 충돌 조건으로 후보를 검증했다. 후보 탐색은 printable 우선 집합과 전체 바이트 집합을 단계적으로 사용했다.

Additional: 딜레이가 심한 관계로 인도 지역에 AWS로 공격했다.

Analysis

1
def proof_of_work(self):
2
    proof = ''.join([random.choice(string.ascii_letters+string.digits) for _ in range(20)])
3
    _hexdigest = sha256(proof.encode()).hexdigest()
4
    self.send(f"sha256(XXXX+{proof[4:]}) == {_hexdigest}".encode())
5
    x = self.recv(prompt=b'Give me XXXX: ')
6
    if len(x) != 4 or sha256(x+proof[4:].encode()).hexdigest() != _hexdigest:
7
        return False
8
    return True

세션 시작은 4바이트 prefix를 찾는 PoW이다. 허용 alphabet이 $62$ 문자라 탐색 공간은 $62^4$ 이며 전수 검사로 해결 가능하다.

1
secret = get_secret()
2
key = get_key()
3
IV = get_IV()
4
aes = AES.new(key, mode=AES.MODE_CBC, iv=IV)
5
self.send(f"Kakashi says — IV is: {IV.hex()}".encode())

세션별로 secret key IV를 고정한다. AES.new로 만든 CBC 객체를 재사용하므로 내부 IV는 매 암호화 후 마지막 블록으로 갱신된다. 서버는 처음 연결 시 초기 IV만 알려준다.

1
def pad(self, data):
2
    pad_len = 16 - len(data)%16
3
    return data + bytes([pad_len])*pad_len

패딩은 PKCS7이다. 사용자가 보낸 메시지 뒤에 secret을 이어붙이고 padding을 적용한다고 가정한다. 빈 입력을 보낼 때 ciphertext 길이가 항상 64바이트였으므로 secret 길이 48바이트라는 결론에 도달했다. 이 가정은 ciphertext 길이 관찰을 통해 검증했다.

1
if choice == b"1":
2
    msg = self.recvhex(prompt=b"Your jutsu (in hex): ")
3
    cipher = perform_encryption(aes, msg, secret, self.pad)
4
    self.send(cipher.hex().encode())

메뉴 1이 encryption oracle을 제공한다. perform_encryption 구현은 주어지지 않았지만 동일 세션에서 IV가 연속적으로 갱신되는 점과 ciphertext 길이 분석으로 plaintext 구성이 pad(msg + secret)임을 추측했다. secret이 블록 끝에 오도록 prefix 길이를 조절하면 공격 대상 바이트를 마지막 위치에 둘 수 있다.

1
elif choice == b"2":
2
    guess = self.recvhex(prompt=b"What did your Sharingan see? (in hex): ")
3
    if guess == secret:
4
        self.send(b"Kakashi: Impressive. Here is your scroll : " + flag)

복구한 secret을 메뉴 2로 제출하면 flag를 획득한다. 따라서 목표는 encryption oracle로 secret을 추출하는 것이다.

1
def recover_single_session():
2
    conn = Conn()
3
    conn.connect_and_setup()
4
    recovered = bytearray()
5
    while idx < SECRET_LEN:
6
        pad_len = (-(idx + 1)) % 16
7
        if pad_len == 0:
8
            pad_len = 16
9
        prefix = b"\x00" * pad_len
10
        block_idx = (pad_len + idx) // 16

secret의 $idx$ 번째 바이트를 노릴 때 prefix 길이를 조정해 해당 바이트를 어떤 블록의 끝으로 보낸다. pad_len이 0이면 빈 메시지가 서버에서 세션 종료를 유발해 16으로 대체한다. 이는 padding 규칙을 유지하므로 공격에는 영향이 없다.

1
state_before = bytes(conn.state_iv)
2
ctA = conn.encrypt(prefix)
3
blocksA = chunk_blocks(ctA)
4
target_block = blocksA[block_idx]
5
prev_block = state_before if block_idx == 0 else blocksA[block_idx - 1]
6
IV_prime = bytes(conn.state_iv)

baseline 조회로 목표 블록 target_block과 chaining 값 prev_block을 수집한다. state_iv는 매 요청 후 마지막 블록으로 갱신되므로 IV_prime이 다음 요청의 초기 IV가 된다.

1
G_full = prefix + bytes(recovered) + bytes([candidate])
2
G = G_full[-16:]
3
first_plain = xor_bytes(xor_bytes(G, prev_block), IV_prime)
4
ctB = conn.encrypt(first_plain)
5
if chunk_blocks(ctB)[0] == target_block:
6
    return candidate

secret 후보 바이트를 끝에 붙여 16바이트 조각 G를 만들고 CBC 역산으로 첫 블록 plaintext를 구성한다. 후보가 참이면 첫 ciphertext 블록이 baseline의 target_block과 일치한다. 후보가 틀리면 충돌이 사라진다. 이 방식은 secret 바이트를 앞에서부터 순차적으로 결정한다.

1
pri, rest = make_two_stage_candidates()
2
c = try_candidates(pri)
3
...
4
c = try_candidates(rest)

후보는 printable 문자 우선 집합과 나머지 바이트 집합으로 나누어 검사한다. secret이 ASCII flag일 가능성을 활용해 쿼리 수를 줄인다. 모든 후보가 실패하면 세션을 포기하고 재연결한다. 이는 perform_encryption 구조에 대한 추정이 틀렸을 경우 대비책으로 사용된다.

Exploit

PoW 해결: 서버 배너에서 suffix와 digest를 추출해 $62^4$ 공간을 탐색한다. Python itertools.product와 sha256 비교로 약 수백만 해시를 계산하면 충분하다.
상태 추적: 각 encryption 응답의 마지막 블록을 내부 state_iv에 저장해 다음 요청의 chaining 값으로 사용한다. 이는 PyCryptodome CBC가 연속 스트림처럼 동작한다는 사실에 기반한 필수 조건이다.
바이트 복구: 목표 인덱스마다 prefix를 설정해 secret 바이트를 블록 끝에 놓고 baseline과 probe를 비교한다. printable 후보 집합 후 전체 바이트 집합을 시도한다. 후보가 일치하면 secret 바이트를 확정하고 다음 인덱스로 이동한다.
제출: 48바이트 secret을 모두 얻으면 메뉴 2로 전송한다. flag를 받기 전까지 세션이 끊길 수 있으므로 최대 8회까지 자동 재시도하도록 구성했다.

1
import time
2
import re
3
import string
4
import itertools
5
from hashlib import sha256
6
from typing import List
7

8
from pwn import context, remote
9
from pwnlib.exception import PwnlibException
10

11
HOST = "play.h7tex.com"
12
PORT = 44056
13
SECRET_LEN = 48
14

15
POW_RE = re.compile(br"sha256\(XXXX\+(.+?)\) == ([0-9a-f]{64})")
16
POW_ALPHABET = string.ascii_letters + string.digits
17

18
SOCK_TIMEOUT = 10.0
19
RETRY_DELAY = 0.4
20
MAX_CALLS_PER_SESSION = 200000
21
MAX_SESS_RESTARTS = 8
22

23
context.log_level = "debug"
24
DEBUG = True
25

26
def build_priority_candidates() -> List[int]:
27
    pri = []
28
    extras = [ord('{'), ord('}'), ord('_'), ord('-'), ord(':')]
29
    for b in extras:
30
        if b not in pri:
31
            pri.append(b)
32
    for b in range(0x20, 0x7f):
33
        if b not in pri:
34
            pri.append(b)
35
    return pri
36

37
PRIORITY_CANDIDATES = build_priority_candidates()
38
FULL_CANDIDATES = list(range(256))
39

40
def make_two_stage_candidates():
41
    pri = PRIORITY_CANDIDATES.copy()
42
    rest = [b for b in FULL_CANDIDATES if b not in pri]
43
    return pri, rest
44

45
def xor_bytes(a: bytes, b: bytes) -> bytes:
46
    return bytes(x ^ y for x, y in zip(a, b))
47

48
def chunk_blocks(data: bytes, bs: int = 16):
49
    return [data[i:i+bs] for i in range(0, len(data), bs)]
50

51
def find_longest_hex_token(data: bytes) -> bytes:
52
    text = data.decode(errors="ignore")
53
    tokens = []
54
    for line in text.splitlines():
55
        t = line.strip()
56
        if not t:
57
            continue
58
        if all(ch in "0123456789abcdefABCDEF" for ch in t) and (len(t) % 2 == 0):
59
            tokens.append(t)
60
    if not tokens:
61
        return b""
62
    return max(tokens, key=len).encode()
63

64
class ConnectionDropped(Exception):
65
    pass
66

67
class Conn:
68
    def __init__(self):
69
        self.sock = None
70
        self.state_iv = None
71
        self.encrypt_calls = 0
72

73
    def connect_and_setup(self):
74
        if self.sock:
75
            try:
76
                self.sock.close()
77
            except Exception:
78
                pass
79
        self.sock = remote(HOST, PORT, timeout=SOCK_TIMEOUT)
80

81
        # PoW
82
        data = self.sock.recvuntil(b"Give me XXXX: ", timeout=SOCK_TIMEOUT)
83
        m = POW_RE.search(data)
84
        if not m:
85
            raise RuntimeError("PoW prompt not found")
86
        suffix, digest_hex = m.groups()
87
        target = digest_hex.decode()
88
        solved = False
89
        for cand in itertools.product(POW_ALPHABET, repeat=4):
90
            prefix = "".join(cand).encode()
91
            if sha256(prefix + suffix).hexdigest() == target:
92
                self.sock.sendline(prefix)
93
                solved = True
94
                break
95
        if not solved:
96
            raise RuntimeError("PoW unsolved")
97

98
        # menu + IV
99
        menu = self.sock.recvuntil(b"> ", timeout=SOCK_TIMEOUT)
100
        text = menu.decode(errors="ignore")
101
        iv = None
102
        for line in text.splitlines():
103
            if "IV is:" in line:
104
                iv_hex = line.split("IV is:")[1].strip()
105
                iv = bytes.fromhex(iv_hex)
106
                break
107
        if iv is None:
108
            raise RuntimeError("IV not found in menu")
109
        self.state_iv = bytearray(iv)
110
        self.encrypt_calls = 0
111
        if DEBUG:
112
            print("[connect] IV:", iv.hex())
113

114
    def encrypt(self, payload: bytes, timeout: float = SOCK_TIMEOUT) -> bytes:
115
        try:
116
            self.sock.sendline(b"1")
117
            self.sock.recvuntil(b"(in hex): ", timeout=timeout)
118
            self.sock.sendline(payload.hex().encode())
119
            data = self.sock.recvuntil(b"> ", timeout=timeout)
120
            token = find_longest_hex_token(data)
121
            if not token:
122
                raise ConnectionDropped("no hex token in reply")
123
            ct = bytes.fromhex(token.decode())
124
            if len(ct) < 16:
125
                raise ConnectionDropped("cipher too short")
126
            self.state_iv[:] = ct[-16:]
127
            self.encrypt_calls += 1
128
            if DEBUG and (self.encrypt_calls < 20 or self.encrypt_calls % 200 == 0):
129
                print(f"[encrypt #{self.encrypt_calls}] payload_len={len(payload)} ct_len={len(ct)} iv={bytes(self.state_iv).hex()}")
130
            if self.encrypt_calls > MAX_CALLS_PER_SESSION:
131
                raise ConnectionDropped("exceeded max calls per session")
132
            return ct
133
        except (BrokenPipeError, ConnectionResetError, EOFError, ConnectionDropped) as e:
134
            raise ConnectionDropped(e)
135
        except PwnlibException as e:
136
            raise ConnectionDropped(e)
137
        except Exception as e:
138
            raise ConnectionDropped(e)
139

140
    def submit(self, secret: bytes) -> str:
141
        try:
142
            self.sock.sendline(b"2")
143
            self.sock.recvuntil(b"(in hex): ", timeout=5)
144
            self.sock.sendline(secret.hex().encode())
145
            line = self.sock.recvline(timeout=3)
146
            return line.decode(errors="ignore").strip() if line else ""
147
        except Exception:
148
            return ""
149

150
def recover_single_session():
151
    conn = Conn()
152
    conn.connect_and_setup()
153

154
    recovered = bytearray()
155
    idx = 0
156

157
    while idx < SECRET_LEN:
158
        pad_len = (-(idx + 1)) % 16
159
        if pad_len == 0:
160
            pad_len = 16
161
        prefix = b"\x00" * pad_len
162
        block_idx = (pad_len + idx) // 16
163

164
        if DEBUG:
165
            print(f"[IDX {idx}] pad_len={pad_len} block_idx={block_idx} iv={bytes(conn.state_iv).hex()} enc_calls={conn.encrypt_calls}")
166

167
        pri, rest = make_two_stage_candidates()
168

169
        def try_candidates(iterable):
170
            nonlocal conn, idx, prefix, block_idx, recovered
171
            for c in iterable:
172
                try:
173
                    state_before = bytes(conn.state_iv)
174
                    ctA = conn.encrypt(prefix)
175
                except ConnectionDropped as e:
176
                    if DEBUG:
177
                        print("[*] baseline: connection dropped during encrypt:", e)
178
                    raise ConnectionDropped(e)
179

180
                blocksA = chunk_blocks(ctA)
181
                if block_idx >= len(blocksA):
182
                    if DEBUG:
183
                        print(f"[!] ctA too short (need block {block_idx} have {len(blocksA)}) -> abort session")
184
                    raise ConnectionDropped("ctA too short")
185

186
                target_block = blocksA[block_idx]
187
                prev_block = state_before if block_idx == 0 else blocksA[block_idx - 1]
188
                IV_prime = bytes(conn.state_iv)
189

190
                G_full = prefix + bytes(recovered) + bytes([c])
191
                G = G_full[-16:]
192
                first_plain = xor_bytes(xor_bytes(G, prev_block), IV_prime)
193

194
                try:
195
                    ctB = conn.encrypt(first_plain)
196
                except ConnectionDropped as e:
197
                    if DEBUG:
198
                        print("[*] probe: connection dropped during encrypt:", e)
199
                    raise ConnectionDropped(e)
200

201
                blocksB = chunk_blocks(ctB)
202
                if not blocksB:
203
                    if DEBUG:
204
                        print("[!] ctB empty -> abort session")
205
                    raise ConnectionDropped("ctB empty")
206

207
                if blocksB[0] == target_block:
208
                    return c
209
            return None
210

211
        c = try_candidates(pri)
212
        if c is not None:
213
            recovered.append(c)
214
            print(f"[{idx+1:02d}/{SECRET_LEN}] byte = 0x{c:02x} ({chr(c) if 32<=c<=126 else '.'})  calls={conn.encrypt_calls} (from PRIORITY set)")
215
            idx += 1
216
            continue
217

218
        if DEBUG:
219
            print(f"[IDX {idx}] no match in PRIORITY set -> trying FULL 0..255 set (this will be slower)")
220

221
        c = try_candidates(rest)
222
        if c is not None:
223
            recovered.append(c)
224
            print(f"[{idx+1:02d}/{SECRET_LEN}] byte = 0x{c:02x} ({chr(c) if 32<=c<=126 else '.'})  calls={conn.encrypt_calls} (from FULL set)")
225
            idx += 1
226
            continue
227

228
        if DEBUG:
229
            print(f"[!] exhausted all candidates at idx={idx} -> abort session")
230
        raise RuntimeError("No candidate matched in session (exhausted full set)")
231

232
    return bytes(recovered), conn
233

234
def run_recover():
235
    attempts = 0
236
    while attempts < MAX_SESS_RESTARTS:
237
        attempts += 1
238
        try:
239
            secret, conn = recover_single_session()
240
            print("[+] recovered secret (hex):", secret.hex())
241
            resp = conn.submit(secret)
242
            if resp:
243
                print("[+] submit response:", resp)
244
            return
245
        except ConnectionDropped as e:
246
            print(f"[!] session dropped (attempt {attempts}/{MAX_SESS_RESTARTS}): {e}")
247
            time.sleep(RETRY_DELAY)
248
            continue
249
        except RuntimeError as e:
250
            print(f"[!] session runtime error (attempt {attempts}/{MAX_SESS_RESTARTS}): {e}")
251
            time.sleep(RETRY_DELAY)
252
            continue
253
        except Exception as e:
254
            print("[!] unexpected error:", type(e).__name__, e)
255
            time.sleep(RETRY_DELAY)
256
            continue
257
    print("[FATAL] all session attempts exhausted. Consider raising MAX_SESS_RESTARTS or analyzing server behavior more closely.")
258

259
if __name__ == "__main__":
260
    run_recover()

1
H7CTF{K4K4SH1_5H4R1NG4N_$33_411_258d69d7-2492-4a56-ac26-1d0b91117393}

[H7CTF 2025] Kakashi's secrect jutsu writeup

TL;DR

Analysis

Exploit