1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
| import itertools
import hashlib
import re
import socket
import string
from typing import List, Tuple, Dict
from MySQLdb.constants.ER import HOSTNAME
HOST = "47.93.217.138"
PORT = 30748
class CTFCodeSolver:
"""
A client for solving a CTF challenge based on error-correcting codes and PoW.
"""
CHARSET = string.ascii_letters + string.digits
GENERATOR_MATRIX = [
[0,1,1,0,1,0,0,0,1,1,0,0,0,1,0,1,1],
[1,0,1,1,0,1,0,0,0,1,1,0,0,0,1,0,1],
[1,1,0,1,1,0,1,0,0,0,1,1,0,0,0,1,0],
[0,1,1,0,1,1,0,1,0,0,0,1,1,0,0,0,1],
[1,0,1,1,0,1,1,0,1,0,0,0,1,1,0,0,0],
[0,1,0,1,1,0,1,1,0,1,0,0,0,1,1,0,0],
[0,0,1,0,1,1,0,1,1,0,1,0,0,0,1,1,0],
[0,0,0,1,0,1,1,0,1,1,0,1,0,0,0,1,1],
]
SECRET_BITS = 8
CODEWORD_BITS = 17
MAX_ERRORS = 2
def __init__(self, server_ip: str, server_port: int):
"""
Initialize the solver and precompute all required codewords and questions.
"""
self.server_ip = server_ip
self.server_port = server_port
self.sock = None
self.codeword_map = self._precompute_codewords()
self.questions = self._generate_questions()
print("[*] Solver initialized: codewords and questions prepared.")
def _precompute_codewords(self) -> Dict[Tuple[int, ...], Tuple[int, ...]]:
"""
Generate all possible 8-bit secret codewords and verify the minimum Hamming distance.
"""
codeword_map = {}
all_codewords = []
for secret in itertools.product((0, 1), repeat=self.SECRET_BITS):
codeword = self._encode(secret)
if codeword in codeword_map:
raise ValueError("Generator matrix produced a collision for different secrets.")
codeword_map[codeword] = secret
all_codewords.append(codeword)
min_distance = self.CODEWORD_BITS
for i in range(len(all_codewords)):
for j in range(i + 1, len(all_codewords)):
dist = sum(b1 ^ b2 for b1, b2 in zip(all_codewords[i], all_codewords[j]))
min_distance = min(min_distance, dist)
required_distance = 2 * self.MAX_ERRORS + 1
if min_distance < required_distance:
raise ValueError(
f"Hamming distance too small! Minimum distance={min_distance}, required={required_distance}"
)
print(f"[+] Precomputed {len(all_codewords)} codewords, minimum Hamming distance={min_distance}")
return codeword_map
def _generate_questions(self) -> List[str]:
"""
Generate logical expressions (questions) from the transposed generator matrix.
"""
transposed = list(zip(*self.GENERATOR_MATRIX))
questions = []
for col in transposed:
active_bits = [i for i, b in enumerate(col) if b]
questions.append(self._create_xor_expression(active_bits))
return questions
def _encode(self, secret: Tuple[int, ...]) -> Tuple[int, ...]:
"""
Encode an 8-bit secret vector into a 17-bit codeword.
"""
codeword = []
for col in range(self.CODEWORD_BITS):
val = sum(secret[row] * self.GENERATOR_MATRIX[row][col] for row in range(self.SECRET_BITS)) % 2
codeword.append(val)
return tuple(codeword)
def _create_xor_expression(self, indices: List[int]) -> str:
"""
Create a nested XOR expression from a list of indices.
"""
if not indices:
return "0"
if len(indices) == 1:
return f"S{indices[0]}"
expr = f"( ( S{indices[0]} == S{indices[1]} ) == 0 )"
for idx in indices[2:]:
expr = f"( ( {expr} == S{idx} ) == 0 )"
return expr
def _solve_pow(self, challenge: str) -> str:
"""
Solve the server's SHA256 PoW challenge.
"""
match = re.match(r"sha256\(XXXX\+([A-Za-z0-9]+)\) == ([0-9a-f]{64})", challenge.strip())
if not match:
raise ValueError(f"Invalid PoW format: {challenge}")
suffix, target_hash = match.groups()
print(f"[*] Solving PoW: sha256(XXXX+{suffix}) == {target_hash[:10]}...")
for cand in itertools.product(self.CHARSET, repeat=4):
prefix = "".join(cand)
if hashlib.sha256((prefix + suffix).encode()).hexdigest() == target_hash:
print(f"[+] PoW solved: prefix='{prefix}'")
return prefix
raise RuntimeError("PoW solution not found.")
def _decode_response(self, received: List[int]) -> Tuple[int, ...]:
"""
Decode the 17-bit response, correcting up to MAX_ERRORS errors.
"""
recv_tuple = tuple(received)
if recv_tuple in self.codeword_map:
return self.codeword_map[recv_tuple]
positions = range(self.CODEWORD_BITS)
for num_errors in range(1, self.MAX_ERRORS + 1):
for error_indices in itertools.combinations(positions, num_errors):
tmp = list(recv_tuple)
for i in error_indices:
tmp[i] ^= 1
corrected = tuple(tmp)
if corrected in self.codeword_map:
print(f"[DEBUG] Corrected bits at positions {error_indices}")
return self.codeword_map[corrected]
raise ValueError(f"Decoding failed: exceeded max error correction ({self.MAX_ERRORS})")
def _read_line(self) -> str:
"""
Read a single line from the server, decode and strip it.
"""
buffer = b""
while not buffer.endswith(b"\n"):
chunk = self.sock.recv(1)
if not chunk:
raise ConnectionAbortedError("Server closed the connection.")
buffer += chunk
line = buffer.decode().strip()
print(f"[RECV] {line}")
return line
def _send_line(self, message: str):
"""
Send a line to the server.
"""
print(f"[SEND] {message}")
self.sock.sendall((message + "\n").encode())
def _handle_pow(self):
"""
Perform the PoW exchange with the server.
"""
pow_challenge = self._read_line()
solution = self._solve_pow(pow_challenge)
self._read_line()
self._send_line(solution)
def _play_round(self) -> bool:
"""
Perform one full Q&A round.
Returns True if flag is received, False otherwise.
"""
line = ""
while "Ask your question:" not in line:
line = self._read_line()
answers = []
for question in self.questions:
while "Ask your question:" not in line:
line = self._read_line()
self._send_line(question)
response_line = ""
while "Prisoner's response:" not in response_line:
response_line = self._read_line()
answer_str = response_line.split(":", 1)[1].strip().rstrip("!").strip()
answers.append(1 if answer_str == "True" else 0)
line = self._read_line()
if "Now reveal" in line:
break
if len(answers) != self.CODEWORD_BITS:
raise ValueError(f"Answer count mismatch: expected {self.CODEWORD_BITS}, got {len(answers)}")
while "Now reveal" not in line:
line = self._read_line()
decoded_secret = self._decode_response(answers)
self._send_line(" ".join(map(str, decoded_secret)))
result_line = self._read_line()
return "flag" in result_line.lower()
def solve(self):
"""
Connect to the server and execute the full solving process.
"""
try:
with socket.create_connection((self.server_ip, self.server_port), timeout=10) as conn:
self.sock = conn
self._handle_pow()
while not self._play_round():
print("\n[*] Flag not found, starting a new round...\n")
print("\n[+] Challenge completed: Flag found!")
except Exception as e:
print(f"\n[ERROR] Execution error: {e}")
if __name__ == "__main__":
solver = CTFCodeSolver(HOST,PORT)
solver.solve()
|
