Module 06: Broken Access Control¶

# VULNERABLE: IDOR in API endpoint
from flask import Flask, request, jsonify

@app.route('/api/invoice/<int:invoice_id>')
def get_invoice(invoice_id):
    # MISSING: check that the current user owns this invoice
    invoice = db.get_invoice(invoice_id)
    return jsonify(invoice)
# Attacker logs in as user 1, requests /api/invoice/2 → sees user 2's invoice

# SAFE: enforce ownership check
from flask_login import current_user

@app.route('/api/invoice/<int:invoice_id>')
def get_invoice_safe(invoice_id):
    invoice = db.get_invoice(invoice_id)
    # Verify the authenticated user owns this resource
    if invoice is None or invoice.user_id != current_user.id:
        return jsonify({"error": "Not found"}), 404
    return jsonify(invoice)

# VULNERABLE: Vertical privilege escalation via hidden parameter
@app.route('/update_user', methods=['POST'])
def update_user():
    user_id = request.form['user_id']
    new_role = request.form.get('role', 'user')  # role comes from client
    # VULNERABLE: client controls the role assignment
    db.update_user_role(user_id, new_role)

# Attack: POST /update_user with role=admin
# The attacker promotes themselves to admin

# SAFE: role changes are admin-only operations
from functools import wraps

def admin_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if not current_user.is_admin:
            return jsonify({"error": "Forbidden"}), 403
        return f(*args, **kwargs)
    return decorated

@app.route('/update_user', methods=['POST'])
@admin_required  # Only admins can change roles
def update_user_safe():
    user_id = request.form['user_id']
    new_role = request.form['role']
    # Server-side validation: only allow valid role values
    if new_role not in ['user', 'moderator', 'admin']:
        return jsonify({"error": "Invalid role"}), 400
    db.update_user_role(user_id, new_role)

# VULNERABLE: Mass assignment in a user profile update
@app.route('/api/profile', methods=['PUT'])
def update_profile():
    user_data = request.json  # Entire request body
    # VULNERABLE: updates ALL fields including is_admin, balance, etc.
    current_user.update(**user_data)
    return jsonify({"success": True})

# Attack: PUT /api/profile with {"name": "Hacker", "is_admin": true}

# SAFE: allowlist the fields that users can modify
ALLOWED_UPDATE_FIELDS = {'name', 'email', 'bio', 'avatar_url'}

@app.route('/api/profile_safe', methods=['PUT'])
def update_profile_safe():
    user_data = request.json
    # Only update fields in the allowlist
    safe_data = {k: v for k, v in user_data.items() if k in ALLOWED_UPDATE_FIELDS}
    current_user.update(**safe_data)
    return jsonify({"success": True})

import jwt  # PyJWT library

# VULNERABLE 1: Algorithm confusion — "none" algorithm
# Some libraries accept "alg": "none" which means no signature verification
# Attacker modifies payload and sets alg to "none"
# Fix: explicitly specify allowed algorithms in verification

# WRONG — allows any algorithm including "none"
decoded = jwt.decode(token, options={"verify_signature": False})

# RIGHT — specify exactly which algorithms are allowed
decoded = jwt.decode(token, secret_key, algorithms=["HS256"])

# VULNERABLE 2: Weak secret
# If the JWT is signed with a weak or default secret (e.g., "secret", "password"),
# it can be cracked offline using tools like hashcat or john
# Attack: hashcat -a 0 -m 16500 <jwt> /usr/share/wordlists/rockyou.txt
# Fix: use cryptographically random secrets of at least 256 bits

import secrets
# Generate a strong JWT secret
strong_secret = secrets.token_hex(32)  # 256-bit random secret

# VULNERABLE 3: Algorithm substitution (RS256 → HS256)
# If server uses RSA (RS256), attacker can sign with the PUBLIC KEY as HMAC secret
# by changing the algorithm to HS256 (symmetric HMAC)
# Fix: server must verify the algorithm matches what it expects

# Simple RBAC implementation
from enum import Enum

class Role(Enum):
    VIEWER = 1
    EDITOR = 2
    ADMIN = 3

class Permission(Enum):
    READ = 'read'
    WRITE = 'write'
    DELETE = 'delete'
    MANAGE_USERS = 'manage_users'

ROLE_PERMISSIONS = {
    Role.VIEWER: {Permission.READ},
    Role.EDITOR: {Permission.READ, Permission.WRITE},
    Role.ADMIN: {Permission.READ, Permission.WRITE, Permission.DELETE, Permission.MANAGE_USERS},
}

def has_permission(user, permission: Permission) -> bool:
    """Check if user has the required permission."""
    user_role = Role(user.role)
    return permission in ROLE_PERMISSIONS.get(user_role, set())

# Usage in a view
def delete_post(post_id):
    if not has_permission(current_user, Permission.DELETE):
        return {"error": "Forbidden"}, 403
    # Proceed with deletion only if authorized
    db.delete_post(post_id)

# Test script for vertical privilege escalation (authorized lab only)
import requests

# Two sessions: regular user and admin user
user_session = requests.Session()
admin_session = requests.Session()

BASE_URL = "http://localhost"  # Local DVWA or authorized target

# Log in as regular user
user_session.post(f"{BASE_URL}/login", data={"username": "user", "password": "user123"})

# Log in as admin for comparison
admin_session.post(f"{BASE_URL}/login", data={"username": "admin", "password": "admin123"})

# Get list of admin-only endpoints to test
admin_endpoints = [
    "/admin/users",
    "/admin/settings",
    "/api/admin/stats",
    "/management/",
]

print("Testing vertical privilege escalation:")
for endpoint in admin_endpoints:
    admin_r = admin_session.get(f"{BASE_URL}{endpoint}")
    user_r = user_session.get(f"{BASE_URL}{endpoint}")

    if admin_r.status_code == 200 and user_r.status_code == 200:
        print(f"[!] POSSIBLE ESCALATION: {endpoint} — admin=200, user=200")
    elif user_r.status_code == 403:
        print(f"[OK] {endpoint} — correctly returns 403 for regular user")
    else:
        print(f"[?] {endpoint} — admin={admin_r.status_code}, user={user_r.status_code}")