TL;DR
퍼뮤테이션의 역함수를 구현하고 해시 값에 맞는 블록 입력과 초기 상태를 역구성해 충돌 메시지를 자동화하는 문제
해시함수 - SPONGE(rate=1, capacity=1) + ANEMOI Permutation
Analysis
Code
challenge.py
#!/usr/bin/env sage
from Anemoi_primitive.anemoi import *
from copy import deepcopy
import os
import signal
from random import randint
TIMEOUT = 200
FLAG = os.getenv("FLAG", "FLAG{this_is_a_fake_flag_for_testing_purposes}")
PRIME = 280989701
# SPONGE
class SPONGE:
def __init__(self, prime):
self.prime = prime
self.n_bits_per_block = len(bin(prime)[2:]) - 1
# Anemoi params
self.l = 1
self.alpha = 3
self.n_rounds = 21
def hash(self, message, initial_c):
assert initial_c < self.prime
# ABSORBING PHASE
state = [[0], [initial_c]]
for b in message:
state[0][0] = (state[0][0] + b) % self.prime
temp = deepcopy(state)
permutation = ANEMOI(self.prime, self.alpha, self.n_rounds, self.l)
state = permutation(temp[0], temp[1])
return state[0][0]
# SERVICE
TEST = 100
def main():
S = SPONGE(PRIME)
print("For each of the given messages, you must give me a collision. Let's start!")
# create message
print(
"The messages should be represented as a list of values comma separated such as 2, 3, 4, 5, 6"
)
while True:
print(
"""
MENU:
1) Play
2) Exit
"""
)
choice = input("> ")
if choice == "1":
passed_tests = 0
while passed_tests < TEST:
message = [randint(0, PRIME - 1) for _ in range(5)]
initial_capacity = randint(0, PRIME - 1)
print(f"The chosen message is: {message}")
print(f"The initial capacity is: {initial_capacity}")
hash_value = S.hash(message, initial_capacity)
print(f"The corresponding hash value is {hash_value}")
solvable = int(input("Is there solution? (0 means NO, 1 means YES): "))
if solvable == 0:
continue
elif solvable > 1:
break
input_message = list(
map(int, input("Give me your message: ").strip().split(","))
)
if len(input_message) < 2:
print("Your message must have at least two blocks")
break
if not all([v < PRIME for v in input_message]):
print("Your message must be composed of values in the field!")
break
if "|".join(map(str, input_message)) in "|".join(map(str, message)):
print("Your message cannot contains subsequences of my message!")
break
input_initial_capacity = int(input("Give me your initial capacity: "))
your_hash_value = S.hash(input_message, input_initial_capacity)
if your_hash_value != hash_value:
print("The hashes do not match!")
break
else:
print("Congratulations!")
passed_tests += 1
if passed_tests == TEST:
print(f"Congratulations! Here is the flag: {FLAG}")
elif choice == "2":
break
if __name__ == "__main__":
signal.alarm(TIMEOUT)
main()
Anemoi_primitive/anemoi.py
from sage.all import *
from Anemoi_primitive.anemoi_utilities import *
from copy import deepcopy
"""
Simple ANEMOI implementation
"""
"""
Constants from https://github.com/anemoi-hash/anemoi-hash
"""
PI_0 = 1415926535897932384626433832795028841971693993751058209749445923078164062862089986280348253421170679
PI_1 = 8214808651328230664709384460955058223172535940812848111745028410270193852110555964462294895493038196
def euler_phi(v):
if is_prime(v):
return v - 1
else:
factors = factor(v)
phi = v
for f in factors:
phi *= (1 - 1 / f[0])
return phi
"""
Simple ANEMOI in F_q^2l
"""
class ANEMOI:
def __init__(self, q, alpha, nrounds, l):
self.q = q
self.alpha = alpha
self.nrounds = nrounds
self.l = l
# generate the field
self.F = GF(self.q)
if is_prime(self.q):
if gcd(self.alpha, self.q - 1) != 1:
raise Exception("alpha should be coprime with phi(q)")
self.degree = 2
self.to_field = lambda x: self.F(x)
self.from_field = lambda x: Integer(x)
else:
self.degree = 3
self.to_field = lambda x: self.F.fetch_int(x)
self.from_field = lambda x: x.integer_representation()
# get the generator and the inv_generator
self.g = self.F.multiplicative_generator()
self.beta = self.g
self.gamma = 0
self.delta = self.g ** (-1)
self.alpha_inv = pow(alpha, -1, euler_phi(self.q))
# Choosing constants: self.C and self.D are built from the
# digits of pi using an open butterfly
self.C = []
self.D = []
pi_F_0 = self.F(PI_0 % self.q)
pi_F_1 = self.F(PI_1 % self.q)
for r in range(0, self.nrounds + 1):
pi_0_r = pi_F_0 ** r
self.C.append([])
self.D.append([])
for i in range(0, self.l):
pi_1_i = pi_F_1 ** i
pow_alpha = (pi_0_r + pi_1_i) ** self.alpha
self.C[r].append(self.g * (pi_0_r) ** 2 + pow_alpha)
self.D[r].append(self.g * (pi_1_i) ** 2 + pow_alpha + self.delta)
self.mat = get_mds(self.F, self.l)
"""
X, Y: lists of l values in F_q
"""
def __call__(self, X, Y):
if len(X) != self.l or len(Y) != self.l:
raise Exception("wrong input size!")
else:
return self.hash([X, Y])
"""
:param state is a list of 2*l values in F_q
"""
def hash(self, state):
for i in range(self.nrounds):
state = self.forward_round(state, i)
state = self.add_constants(state, self.nrounds)
state = self.linear_layer(state)
return state
def verify(self, state, hash):
return None
"""
param: state -> list of elements in F_q
"""
def add_constants(self, state, round_index):
# pick the constants depending on alpha
temp_state = deepcopy(state)
for i in range(0, self.l):
state[0][i] += self.C[round_index][i]
state[1][i] += self.D[round_index][i]
return state
def linear_layer(self, state):
x, y = state[0], state[1]
x = self.mat * vector(x)
y = self.mat * vector(y[1:] + [y[0]])
# Pseudo-Hadamard transform on each (x,y) pair
y += x
x += y
return [list(x), list(y)]
def evaluate_sbox(self, _x, _y):
x, y = _x, _y
x -= self.beta * y ** self.degree
y -= x ** self.alpha_inv
x += (self.beta * y ** self.degree + self.delta)
return x, y
def evaluate_sbox_complete(self, state):
""" Flystel evaluation """
"""Applies an open Flystel to the full state. """
temp_state = deepcopy(state)
for i in range(0, self.l):
temp_state[0][i], temp_state[1][i] = self.evaluate_sbox(temp_state[0][i], temp_state[1][i])
return temp_state
def forward_round(self, state, round_index):
state = self.add_constants(state, round_index)
state = self.linear_layer(state)
state = self.evaluate_sbox_complete(state)
return state
Anemoi_primitive/anemoi_utilities.py
from sage.all import *
import itertools
# from original implementation https://github.com/anemoi-hash/anemoi-hash/blob/main/anemoi.sage
def is_mds(m):
# Uses the Laplace expansion of the determinant to calculate the (m+1)x(m+1) minors in terms of the mxm minors.
# Taken from https://github.com/mir-protocol/hash-constants/blob/master/mds_search.sage.
# 1-minors are just the elements themselves
if any(any(r == 0 for r in row) for row in m):
return False
N = m.nrows()
assert m.is_square() and N >= 2
det_cache = m
# Calculate all the nxn minors of m:
for n in range(2, N + 1):
new_det_cache = dict()
for rows in itertools.combinations(range(N), n):
for cols in itertools.combinations(range(N), n):
i, *rs = rows
# Laplace expansion along row i
det = 0
for j in range(n):
# pick out c = column j; the remaining columns are in cs
c = cols[j]
cs = cols[:j] + cols[j + 1:]
# Look up the determinant from the previous iteration
# and multiply by -1 if j is odd
cofactor = det_cache[(*rs, *cs)]
if j % 2 == 1:
cofactor = -cofactor
# update the determinant with the j-th term
det += m[i, c] * cofactor
if det == 0:
return False
new_det_cache[(*rows, *cols)] = det
det_cache = new_det_cache
return True
def M_2(x_input, b):
x = x_input[:]
x[0] += b * x[1]
x[1] += b * x[0]
return x
def M_3(x_input, b):
x = x_input[:]
t = x[0] + b * x[2]
x[2] += x[1]
x[2] += b * x[0]
x[0] = t + x[2]
x[1] += t
return x
def M_4(x_input, b):
x = x_input[:]
x[0] += x[1]
x[2] += x[3]
x[3] += b * x[0]
x[1] = b * (x[1] + x[2])
x[0] += x[1]
x[2] += b * x[3]
x[1] += x[2]
x[3] += x[0]
return x
def lfsr(x_input, b):
x = x_input[:]
l = len(x)
for r in range(0, l):
t = sum(b ** (2 ** i) * x[i] for i in range(0, l))
x = x[1:] + [t]
return x
def circulant_mds_matrix(field, l, coeff_upper_limit=None):
if coeff_upper_limit == None:
coeff_upper_limit = l + 1
assert (coeff_upper_limit > l)
for v in itertools.combinations_with_replacement(range(1, coeff_upper_limit), l):
mat = matrix.circulant(list(v)).change_ring(field)
if is_mds(mat):
return (mat)
# In some cases, the method won't return any valid matrix,
# hence the need to increase the limit further.
return circulant_mds_matrix(field, l, coeff_upper_limit + 1)
def get_mds(field, l):
if l == 1:
return identity_matrix(field, 1)
if l <= 4: # low addition case
a = field.multiplicative_generator()
b = field.one()
t = 0
while True:
# we construct the matrix
mat = []
b = b * a
t += 1
for i in range(0, l):
x_i = [field.one() * (j == i) for j in range(0, l)]
if l == 2:
mat.append(M_2(x_i, b))
elif l == 3:
mat.append(M_3(x_i, b))
elif l == 4:
mat.append(M_4(x_i, b))
mat = Matrix(field, l, l, mat).transpose()
if is_mds(mat):
return mat
else: # circulant matrix case
return circulant_mds_matrix(field, l)
Flow
proof of work:
hashcash -mCb28 <resource>
stamp:
처음에 접속하면 PoW 텍스트가 주어지는데
stamp에 hashcash 포맷을 전송해야한다. (rand/counter를 Base64 인코딩해야함)
- 서버는 매 라운드마다 길이 5의 무작위 메시지 $ M = [m0,...,m4] $와 초기 capacity $c0$을 생성한다.
- $hash(M, c0) = h$ 계산 후 $M, c0, h$를 출력한다.
- 해결 가능 여부를 0과 1중 하나 고른다.
- 1일 경우 새 메시지 $M'$와 초기 capacity $c'$를 제출한다.
- $hash(m', c')$가 동일한 $h$면 Congratulations! 출력
- 100회 성공해야 FLAG가 출력된다.
문자열 제약
if "|".join(map(str, input_message)) in "|".join(map(str, message)):
print("Your message cannot contains subsequences of my message!")
break
input_message 문자열로 a|b|c 처럼 이어 붙인 것이 서버가 만든 message를 x|y|z|... 로 붙인 문자열의 부분 문자열이면 실패
연속된 블록들이 원래 메시지의 인접한 구간과 정확히 일치하면 탈락한다.
HASH (SPONGE)
prime: p = 280989701 상에서 동작한다.
rate=1, capacity=1 (l=1)
absorb: 블록 b를 받아 $x \leftarrow (x+b)mod p$
최종 출력은 상태의 첫 좌표 $x$ 하나
블록 하나 흡수 후: $x \leftarrow x+b $ Permutation $F \rightarrow$ 새 상태, 이 과정을 블록마다 반복하며 마지막 상태의 $x$를 해시로 취한다.
HASH(ANEMOI(l=1))
prime 필드 $GF(p)$, 라운드 수 n_rounds = 21
라운드 구성
- x += C[r], y += D[r]
- LinearLayer
- S-box
마지막 라운드 후 추가로 AddConstant(r=n_rounds)와 LinearLayer를 한번 더 수행 (압축)
LinearLayer
def linear_layer(self, state):
x, y = state[0], state[1]
x = self.mat * vector(x)
y = self.mat * vector(y[1:] + [y[0]])
# Pseudo-Hadamard transform on each (x,y) pair
y += x
x += y
return [list(x), list(y)]
$ l=1$에서 다음 변환
$$ \begin{aligned} y' &= y + x,\\ x' &= x + y' = 2x + y. \end{aligned} \qquad\Longleftrightarrow\qquad \begin{bmatrix}x'\\y'\end{bmatrix} = \underbrace{\begin{bmatrix}2&1\\[2pt]1&1\end{bmatrix}}_{L} \begin{bmatrix}x\\y\end{bmatrix}. $$
행렬식 $\det L=1$이므로 가역이며,
$$ L^{-1}=\begin{bmatrix}1&-1\\[2pt]-1&2\end{bmatrix}\ (\bmod p), \quad\Rightarrow\quad \boxed{~x = x'-y',\ \ y = 2y'-x'~}. $$
S-box
def evaluate_sbox(self, _x, _y):
x, y = _x, _y
x -= self.beta * y ** self.degree
y -= x ** self.alpha_inv
x += (self.beta * y ** self.degree + self.delta)
return x, y
l=1에 맞추어 정리하면 다음과 같다.
$$
\begin{aligned}
t_1 &= x - \beta\,y^2,\\
y' &= y - t_1^{\alpha^{-1}},\\
x' &= t_1 + \big(\beta\,(y')^2 + \delta\big).
\end{aligned}
$$
이에 대한 역함수
$$
\begin{aligned}
t_1 &= x' - \big(\beta\,(y')^2 + \delta\big),\\
y &= y' + t_1^{\,\alpha^{-1}},\\
x &= t_1 + \beta\,y^2.
\end{aligned}
$$
모든 거듭제곱과 합, 곱은 $\mathbb{F}_p$에서 계산한다.
1라운드, 전체 Permutation
라운드 $r$의 전방 변환
$$ (x,y)\xrightarrow{+C[r],+D[r]}(x+C[r],\,y+D[r]) \xrightarrow{L}(x_1,y_1)\xrightarrow{\text{S-box}}(x',y'). $$
마지막에는 추가로 $(+C[n],+D[n])$ 후 $L$까지만 적용(S-box 없음)
따라서 전체 전방 $F:\mathbb{F}_p^2\to\mathbb{F}_p^2$의 역함수 $F^{-1}$는 다음과 같다.
- 마지막 $L^{-1}$ → $-C[n],-D[n]$,
- $r=n-1$부터 $0$까지 각 라운드에 대해: S-box$^{-1}$ → $L^{-1}$ → $-C[r],-D[r]$.
Exploitation
충돌 구성
초기 상태 $(x_0,y_0)=(0,c)$. 블록 $b$를 흡수하면 $x\leftarrow x+b$. 각 흡수 뒤에 $F$를 적용.
두 블록 메시지 $M'=[b_1,b_2]$와 초기 capacity $c'$의 전방 경로는 다음과 같다
$$
\begin{aligned}
(x,y) &\gets (0,c'),\\
(x,y) &\gets (b_1,c'), \quad (x_1,y_1) = F(b_1,c'),\\
(x,y) &\gets (x_1+b_2,\,y_1), \quad (x_2,y_2) = F(x_1+b_2,\,y_1)
\end{aligned}
$$
최종 해시는 $h=x_2$이다.
역구성 절차
목표 해시가 $h$일 때, $y_2\in\mathbb{F}_p$와 $b_2\in\mathbb{F}_p$를 임의로 선택한다.
- $(x_1+b_2,\,y_1)=F^{-1}(h,\,y_2)$
- $x_1\equiv (x_1+b_2)-b_2\pmod p$
- $(b_1,\,c')=F^{-1}(x_1,\,y_1)$
그럼 $M'=[b_1,b_2]$, 초기 capacity $c'$로 전방을 계산하면 다시 $x_2=h$가 되어 항상 동일 해시를 얻는다.
문자열 제약은 확률적으로 드물게 걸릴 수 있으므로, 충돌 시 $b_2\leftarrow b_2+1$로 조정하여 회피한다. (분석 첫 번째 참고)
Code
lin_inv 구현
def lin_inv(x1, y1):
y = (2*y1 - x1) % PRIME
x = (x1 - y1) % PRIME
return x, y
$L^{-1}:~x=x'-y',\ y=2y'-x'$
sbox_inv 구현
${Sbox}^{-1}$를 구현할 땐 순서를 주의해야한다.
def sbox_inv(x1, y1):
t1 = (x1 - (beta * pow(y1, 2, PRIME) + delta) % PRIME) % PRIME
ty = pow(t1, alpha_inv, PRIME)
y = (y1 + ty) % PRIME
x = (t1 + (beta * pow(y, 2, PRIME)) % PRIME) % PRIME
return x, y
$t_1 \rightarrow y \rightarrow x$
perm_inv
$F^{-1}$는 마무리 $L^{-1} \rightarrow $ 상수 제거 $\rightarrow$ 각 라운드에 대해 $ \text{Sbox}^{-1} \rightarrow L^{-1} \rightarrow $상수 제거 $\rightarrow$ perm_inv
def perm_inv(x, y):
x, y = lin_inv(x, y)
x = (x - C[N_ROUNDS]) % PRIME
y = (y - D[N_ROUNDS]) % PRIME
for r in range(N_ROUNDS-1, -1, -1):
x, y = sbox_inv(x, y)
x, y = lin_inv(x, y)
x = (x - C[r]) % PRIME
y = (y - D[r]) % PRIME
return x, y
make_collision
충돌 구성은 다음과 같이 구성한다.
$ (x_1{+}b_2,y_1)=F^{-1}(h,y_2)$, $x_1=(x_1{+}b_2)-b_2$, $(b_1,c')=F^{-1}(x_1,y_1) \rightarrow$ make_collision
def make_collision(target_hash: int, their_message_str: str):
y2 = random.randrange(PRIME)
x1_plus_b2, y1 = perm_inv(target_hash % PRIME, y2)
b2 = random.randrange(PRIME)
tries = 0
while True:
x1 = (x1_plus_b2 - b2) % PRIME
b1, c = perm_inv(x1, y1) # 첫 add 직후 -> b1, c
b1 %= PRIME; b2 %= PRIME; c %= PRIME
if f"{b1}, {b2}" not in their_message_str:
return b1, b2, c
b2 = (b2 + 1) % PRIME
tries += 1
if tries > 24:
b1 = (b1 + 1234567) % PRIME
return b1, b2, c
전체 구현
from pwn import *
import re, time, os, random, base64, subprocess, shutil, sys, signal
from hashlib import sha1
from multiprocessing import Process, Event, Queue, cpu_count
context.log_level = 'debug'
PRIME = 280989701
ALPHA = 3
N_ROUNDS = 21
PI_0 = 1415926535897932384626433832795028841971693993751058209749445923078164062862089986280348253421170679
PI_1 = 8214808651328230664709384460955058223172535940812848111745028410270193852110555964462294895493038196
RE_POW = re.compile(rb'hashcash\s+-mC?b(\d+)\s+([^\s]+)')
def leading_zero_bits_digest(d: bytes) -> int:
n = 0
for b in d:
if b == 0:
n += 8
continue
if b < 0x02: n += 7
elif b < 0x04: n += 6
elif b < 0x08: n += 5
elif b < 0x10: n += 4
elif b < 0x20: n += 3
elif b < 0x40: n += 2
elif b < 0x80: n += 1
return n
return n
def b64_nopad_from_bytes(b: bytes) -> str:
return base64.b64encode(b).decode().rstrip('=')
def counter_to_b64(counter: int) -> str:
if counter == 0:
return b64_nopad_from_bytes(b'\x00')
l = (counter.bit_length() + 7) // 8
return b64_nopad_from_bytes(counter.to_bytes(l, 'big'))
def verify_stamp(bits: int, resource: str, stamp: str) -> bool:
# 해시 선행 0비트 검증
parts = stamp.strip().split(':')
if len(parts) != 7:
return False
if parts[0] != '1':
return False
try:
if int(parts[1]) != bits:
return False
except:
return False
if parts[3] != resource:
return False
d = sha1(stamp.encode()).digest()
return leading_zero_bits_digest(d) >= bits
def try_system_hashcash(bits: int, resource: str, timeout_sec: int = 20) -> str | None:
if shutil.which('hashcash') is None:
return None
variants = [
['hashcash', '-m', f'-b{bits}', '-C', resource],
['hashcash', f'-mb{bits}', resource],
['hashcash', '-m', f'-b{bits}', resource],
['hashcash', f'-b{bits}', resource],
]
for cmd in variants:
try:
cp = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout_sec
)
out = (cp.stdout or '').strip()
if out:
stamp = out.splitlines()[0].strip()
if verify_stamp(bits, resource, stamp):
return stamp
except Exception:
continue
return None
def worker_pow(bits: int, resource: str, date_str: str, rand_b64: str,
start_ctr: int, step: int, stop_ev: Event, q: Queue):
prefix = f"1:{bits}:{date_str}:{resource}::{rand_b64}:"
ctr = start_ctr
_sha1 = sha1
_digest = leading_zero_bits_digest
_verify_bits = bits
while not stop_ev.is_set():
ctr_b64 = counter_to_b64(ctr)
s = prefix + ctr_b64
if _digest(_sha1(s.encode()).digest()) >= _verify_bits:
q.put(s)
stop_ev.set()
return
ctr += step
def solve_hashcash_parallel(bits: int, resource: str, use_time: bool = True) -> str:
date_str = time.strftime('%y%m%d%H%M%S' if use_time else '%y%m%d', time.gmtime())
rand_b64 = b64_nopad_from_bytes(os.urandom(8))
procs = []
stop_ev = Event()
q = Queue()
nproc = max(1, cpu_count() or 1)
start = random.randrange(0, 1 << 24)
for i in range(nproc):
p = Process(target=worker_pow,
args=(bits, resource, date_str, rand_b64, start + i, nproc, stop_ev, q),
daemon=True)
p.start()
procs.append(p)
stamp = q.get()
stop_ev.set()
for p in procs:
try:
p.join(timeout=0.2)
except Exception:
pass
return stamp
def solve_hashcash(bits: int, resource: str) -> str:
stamp = try_system_hashcash(bits, resource, timeout_sec=20)
if stamp:
return stamp
return solve_hashcash_parallel(bits, resource, use_time=True)
def modinv(a, m): return pow(a, -1, m)
def factorize(n):
f, d = {}, 2
while d*d <= n:
while n % d == 0:
f[d] = f.get(d, 0) + 1
n //= d
d = 3 if d == 2 else d + 2
if n > 1: f[n] = f.get(n, 0) + 1
return list(f.keys())
def primitive_root(p):
phi = p - 1
fac = factorize(phi)
for g in range(2, p):
if all(pow(g, phi//q, p) != 1 for q in fac):
return g
raise RuntimeError("nto found")
# anemoi l=1
g = primitive_root(PRIME)
beta = g % PRIME
delta = modinv(g, PRIME)
alpha_inv = modinv(ALPHA, PRIME - 1)
pi_F_0 = PI_0 % PRIME
C, D = [], []
for r in range(N_ROUNDS + 1):
pi_0_r = pow(pi_F_0, r, PRIME)
pow_alpha = pow((pi_0_r + 1) % PRIME, ALPHA, PRIME) # i=0
C.append((g * pow(pi_0_r, 2, PRIME) + pow_alpha) % PRIME)
D.append((g * 1 + pow_alpha + delta) % PRIME)
def lin_fwd(x, y):
y1 = (y + x) % PRIME
x1 = (x + y1) % PRIME
return x1, y1
def lin_inv(x1, y1):
y = (2*y1 - x1) % PRIME
x = (x1 - y1) % PRIME
return x, y
def sbox_fwd(x, y):
t1 = (x - (beta * pow(y, 2, PRIME)) % PRIME) % PRIME
ty = pow(t1, alpha_inv, PRIME)
y1 = (y - ty) % PRIME
x1 = (t1 + (beta * pow(y1, 2, PRIME) + delta) % PRIME) % PRIME
return x1, y1
def sbox_inv(x1, y1):
t1 = (x1 - (beta * pow(y1, 2, PRIME) + delta) % PRIME) % PRIME
ty = pow(t1, alpha_inv, PRIME)
y = (y1 + ty) % PRIME
x = (t1 + (beta * pow(y, 2, PRIME)) % PRIME) % PRIME
return x, y
def perm_inv(x, y):
x, y = lin_inv(x, y)
x = (x - C[N_ROUNDS]) % PRIME
y = (y - D[N_ROUNDS]) % PRIME
for r in range(N_ROUNDS-1, -1, -1):
x, y = sbox_inv(x, y)
x, y = lin_inv(x, y)
x = (x - C[r]) % PRIME
y = (y - D[r]) % PRIME
return x, y
def make_collision(target_hash: int, their_message_str: str):
y2 = random.randrange(PRIME)
x1_plus_b2, y1 = perm_inv(target_hash % PRIME, y2)
b2 = random.randrange(PRIME)
tries = 0
while True:
x1 = (x1_plus_b2 - b2) % PRIME
b1, c = perm_inv(x1, y1) # 첫 add 직후 -> b1, c
b1 %= PRIME; b2 %= PRIME; c %= PRIME
if f"{b1}, {b2}" not in their_message_str:
return b1, b2, c
b2 = (b2 + 1) % PRIME
tries += 1
if tries > 24:
b1 = (b1 + 1234567) % PRIME
return b1, b2, c
RE_MSG = re.compile(r"The chosen message is:\s*\[(.*?)\]")
RE_HASH = re.compile(r"The corresponding hash value is\s*(\d+)")
def main():
if sys.platform == 'win32':
import multiprocessing as mp
mp.set_start_method('spawn', force=True)
p = remote('주소블라인드', 포트블라인드, ssl=True, sni='주소블라인드')
p.sendlineafter(b'token', b'팀토큰블라인드')
buf = p.recvuntil(b'stamp:')
m = RE_POW.search(buf)
if not m:
raise RuntimeError("PoW line not found")
bits = int(m.group(1))
resource = m.group(2).decode()
stamp = solve_hashcash(bits, resource).encode()
p.sendline(stamp)
p.recvuntil(b"MENU:")
p.sendline(b"1")
passed = 0
while True:
data = p.recvuntil(b"Is there solution?").decode()
m_msg = RE_MSG.search(data); m_hash = RE_HASH.search(data)
assert m_msg and m_hash
their_msg_str = m_msg.group(1)
their_hash = int(m_hash.group(1))
p.sendline(b"1")
b1, b2, c = make_collision(their_hash, their_msg_str)
p.recvuntil(b"Give me your message:")
p.sendline(f"{b1}, {b2}".encode())
p.recvuntil(b"Give me your initial capacity:")
p.sendline(str(c).encode())
line = p.recvline(timeout=5)
if b"Congratulations!" not in line:
raise RuntimeError("Round failed; rerun script")
passed += 1
if passed >= 100:
print(p.recvall(timeout=5).decode(errors='ignore'))
break
if __name__ == "__main__":
main()
b'Give me your message: '
[DEBUG] Sent 0x14 bytes:
b'163045706, 94838410\n'
[DEBUG] Received 0x1f bytes:
b'Give me your initial capacity: '
[DEBUG] Sent 0xa bytes:
b'222391664\n'
[DEBUG] Received 0x83 bytes:
b'Congratulations!\n'
b'The chosen message is: [274833314, 239738239, 124344671, 124368038, 239627837]\n'
b'The initial capacity is: 170704170\n'
[DEBUG] Received 0x57 bytes:
b'The corresponding hash value is 87830836\n'
b'Is there solution? (0 means NO, 1 means YES): '
[DEBUG] Sent 0x2 bytes:
b'1\n'
[DEBUG] Received 0x16 bytes:
b'Give me your message: '
[DEBUG] Sent 0x13 bytes:
b'216436716, 2478868\n'
[DEBUG] Received 0x1f bytes:
b'Give me your initial capacity: '
[DEBUG] Sent 0xa bytes:
b'150231328\n'
[DEBUG] Received 0xda bytes:
b'Congratulations!\n'
b'Congratulations! Here is the flag: snakeCTF{resultant_computation_can_help_a_lot_if_some_roots_are_easy_to_find_부정행위방지를위한아이덴티티블라인드}\n'
b' \n'
b' MENU:\n'
b' 1) Play\n'
b' 2) Exit\n'
b' \n'
b'> '
[+] Receiving all data: Done (201B)
[*] Closed connection to 주소블라인드 port 포트블라인드
Congratulations! Here is the flag: snakeCTF{resultant_computation_can_help_a_lot_if_some_roots_are_easy_to_find_부정행위방지를위한아이덴티티블라인드}
MENU:
1) Play
2) Exit
최종 플래그
snakeCTF{resultant_computation_can_help_a_lot_if_some_roots_are_easy_to_find_******thisisteamblind******}
Ref.
'CTF' 카테고리의 다른 글
[PWN] scriptCTF 2025 Vault 3 write-up (0) | 2025.08.21 |
---|---|
[Rev] scriptCTF 2025 - plastic-shield 1, 2 write-up (3) | 2025.08.21 |
[Crypto] scriptCTF 2025 - Secure-Server-2 write-up (1) | 2025.08.21 |
[Crypto] scriptCTF 2025 - EaaS write-up (0) | 2025.08.21 |
[Crypto] CCE 2025 Qual - jokes write-up (2) | 2025.08.21 |