def recv_until(s: socket.socket, token: bytes) -> bytes: data = b"" while token not in data: chunk = s.recv(4096) if not chunk: break data += chunk return data
def recv_line(s: socket.socket) -> bytes: data = b"" while True: ch = s.recv(1) if not ch: break data += ch if ch == b"\n": break return data
def eq_expr(a, v: int): # (a >= v) and (a <= v) => a == v return { "op": "and", "arg1": {"op": ">=", "arg1": a, "arg2": v}, "arg2": {"op": "<=", "arg1": a, "arg2": v}, }
def run_pow_k(k: int, exp: int): """ Chạy pow BASE**exp đúng k lần theo chuỗi and-nested, và trả về 1. Không cộng big-int, chỉ ép evaluate thực sự thực hiện pow. """ expr = 1 pow_node = {"op": "**", "arg1": BASE, "arg2": exp} for _ in range(k): expr = {"op": "and", "arg1": pow_node, "arg2": expr} return expr
def oracle_for_chunk(i: int, exp: int): """ Leak v=(chunk) in [0..7] bằng 8 nhánh: if chunk==v -> chạy (v+1) lần pow, trả 1 -> OR short-circuit dừng """ c = chunk_expr(CHUNK_BITS * i) branches = [] for v in range(8): cond = eq_expr(c, v) body = run_pow_k(v + 1, exp) # luôn >=1 pow để decode theo t1 + v*unit branches.append({"op": "and", "arg1": cond, "arg2": body})
expr = branches[0] for b in branches[1:]: expr = {"op": "or", "arg1": expr, "arg2": b} return expr
def main(): used = 0 x_guess = 0
with socket.create_connection((HOST, PORT)) as s: # ---- dt0 baseline: lấy median 3 lần để giảm jitter ---- dt0_samples = [] for _ in range(3): dt, ans = ask_expr(s, 0); used += 1 if ans == b"": print("[!] Server closed during baseline.") return dt0_samples.append(dt) dt0 = median(dt0_samples)
# ---- chọn exponent sao cho 1 pow đủ “nặng” nhưng không quá chậm ---- chosen_exp = None chosen_t1 = None
# Ta đo t1 = time(run_pow_k(1)) ~ dt0 + pow_time # target pow_time khoảng 0.12..0.35s (đủ tách mức, không kill) for exp in EXP_CANDIDATES: t1s = [] for _ in range(2): dt, ans = ask_expr(s, run_pow_k(1, exp)); used += 1 if ans == b"": print("[!] Server closed during exp probe. Try smaller EXP list.") return t1s.append(dt) t1 = median(t1s) pow_time = max(0.0, t1 - dt0) print(f"[probe] exp={exp:>4} t1≈{t1:.3f}s pow_time≈{pow_time:.3f}s")
# nếu đã quá chậm, dừng và lấy exp hiện tại nhỏ hơn (không tăng nữa) if pow_time > 0.60 and chosen_exp is None: chosen_exp = exp chosen_t1 = t1 break
if chosen_exp is None: # fallback: lấy exp lớn nhất đã thử chosen_exp = EXP_CANDIDATES[-1] chosen_t1 = t1
# ---- đo unit = time(2 pow) - time(1 pow) ---- t2s = [] for _ in range(2): dt, ans = ask_expr(s, run_pow_k(2, chosen_exp)); used += 1 if ans == b"": print("[!] Server closed during unit calibration.") return t2s.append(dt) t2 = median(t2s)
# ---- budget check ---- # used so far: 3 + (<= 2*len(cands)) + 2 (thường ~3+4+2=9) # main queries: # - 33 chunk # - bit99 # - top-check # => 35 # total ~ 44, còn ~6 query cho retry/filler retry_budget = max(0, QUOTA - (used + CHUNKS + 2 + 1)) # +1 guess prompt not in quota # (CHUNKS + bit99 + top-check) print(f"[i] Retry budget (chunk re-ask): {retry_budget}")
# ---- decode chunks ---- for i in range(CHUNKS): expr = oracle_for_chunk(i, chosen_exp) dt, ans = ask_expr(s, expr); used += 1 if ans == b"": print(f"[!] Server closed at chunk {i}.") return
# decode v from dt ≈ t1 + v*unit v = int(round((dt - chosen_t1) / unit)) if v < 0: v = 0 if v > 7: v = 7 pred = chosen_t1 + v * unit resid = abs(dt - pred)
# retry nếu residual quá lớn (mơ hồ) và còn budget # ngưỡng: max(0.45*unit, 0.12s) if retry_budget > 0 and resid > max(0.45 * unit, 0.12): dt2, ans2 = ask_expr(s, expr); used += 1 retry_budget -= 1 if ans2 == b"": print(f"[!] Server closed at chunk {i} retry.") return dtm = median([dt, dt2]) v2 = int(round((dtm - chosen_t1) / unit)) v2 = 0 if v2 < 0 else (7 if v2 > 7 else v2) dt, v = dtm, v2 pred = chosen_t1 + v * unit resid = abs(dt - pred)
# ---- bit99 exact ---- dt, ans = ask_expr(s, bit_expr(99)); used += 1 if ans == b"": print("[!] Server closed at bit99.") return b99 = parse_yes(ans) if b99: x_guess |= (1 << 99) else: x_guess &= ~(1 << 99) print(f"[bit99] {b99}")
# ---- rare top value x==2^100 ---- dt, ans = ask_expr(s, {"op": ">=", "arg1": "x", "arg2": 1 << 100}); used += 1 if ans == b"": print("[!] Server closed at top-check.") return if b"Yes" in ans: x_guess = 1 << 100 print("[top] x == 2^100")
# ---- fill remaining expressions ---- while used < QUOTA: dt, ans = ask_expr(s, 0); used += 1 if ans == b"": print("[!] Server closed during filler.") return
# ---- guess ---- p = recv_until(s, b": ") if not p.endswith(b": "): print("[!] Did not reach guess prompt.") return send_line(s, str(x_guess)) out = s.recv(65535) print("\n[server output]\n" + out.decode(errors="ignore"))
