OAuth2 Scopes and Permissions — Fine-Grained Access Control with Token Scoping

import jwt, datetime
from flask import Flask, request, jsonify, g
from functools import wraps

app = Flask(__name__)
SECRET = "scope-secret"

# Scope hierarchy: admin includes all permissions
SCOPE_HIERARCHY = {
    "threat:read": ["threat:read"],
    "threat:write": ["threat:read", "threat:write"],
    "threat:delete": ["threat:read", "threat:write", "threat:delete"],
    "admin": ["threat:read", "threat:write", "threat:delete", "admin:config"]
}

def require_scope(required_scope):
    """Decorator to enforce scope requirement."""
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            token_scopes = getattr(g, "scopes", "").split()
            required_scopes = SCOPE_HIERARCHY.get(required_scope, [required_scope])

            if not any(s in token_scopes for s in required_scopes):
                return jsonify({
                    "error": "insufficient_scope",
                    "required_scope": required_scope,
                    "current_scopes": token_scopes
                }), 403

            return f(*args, **kwargs)
        return decorated
    return decorator

@app.before_request
def authenticate():
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return

    try:
        payload = jwt.decode(auth[7:], SECRET, algorithms=["HS256"])
        g.user = payload.get("sub")
        g.scopes = payload.get("scope", "")
    except jwt.InvalidTokenError:
        return jsonify({"error": "Invalid token"}), 401

@app.route("/api/v1/threats", methods=["GET"])
@require_scope("threat:read")
def list_threats():
    return jsonify({"threats": [], "scopes": g.scopes})

@app.route("/api/v1/threats", methods=["POST"])
@require_scope("threat:write")
def create_threat():
    return jsonify({"status": "created"})

@app.route("/api/v1/config", methods=["GET"])
@require_scope("admin")
def get_config():
    return jsonify({"config": {}})

@app.route("/authorize", methods=["GET"])
def authorize():
    """Authorization endpoint with scope handling."""
    client_id = request.args.get("client_id")
    requested_scope = request.args.get("scope", "")
    redirect_uri = request.args.get("redirect_uri")

    client = get_client(client_id)
    if not client:
        return "Invalid client", 400

    # Check if requested scopes are within allowed scopes
    requested_scopes = set(requested_scope.split())
    allowed_scopes = set(client["allowed_scopes"])
    denied_scopes = requested_scopes - allowed_scopes

    if denied_scopes:
        # Negotiate scopes down to allowed set
        negotiated_scope = " ".join(requested_scopes & allowed_scopes)
        if not negotiated_scope:
            return jsonify({
                "error": "invalid_scope",
                "scope": " ".join(denied_scopes)
            }), 400
        requested_scope = negotiated_scope

    # Show consent screen (simplified)
    # In production, redirect to a consent page

    auth_code = generate_auth_code(client_id, redirect_uri, requested_scope)
    return jsonify({
        "redirect": f"{redirect_uri}?code={auth_code}&scope={requested_scope}"
    })

def generate_auth_code(client_id, redirect_uri, scope):
    """Generate authorization code with scope."""
    code = secrets.token_urlsafe(32)
    authorization_codes[code] = {
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "scope": scope,
        "expires_at": time.time() + 60
    }
    return code

# Step-up authentication — expand scope after MFA
def issue_token_with_scope(user_id, scope):
    """Issue token with specified scope."""
    return jwt.encode({
        "sub": user_id,
        "scope": scope,
        "mfa_verified": "mfa" in scope,
        "iat": datetime.datetime.utcnow(),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    }, SECRET, algorithm="HS256")

@app.route("/api/auth/step-up", methods=["POST"])
def step_up_scope():
    """Elevate token scope after MFA verification."""
    current_token = request.headers.get("Authorization", "").replace("Bearer ", "")
    current_payload = jwt.decode(current_token, SECRET, algorithms=["HS256"])
    current_user = current_payload.get("sub")

    # Verify MFA
    mfa_code = request.json.get("mfa_code")
    if not verify_mfa(current_user, mfa_code):
        return jsonify({"error": "MFA verification failed"}), 401

    # Expand scope — add threat:write to existing scopes
    existing_scopes = current_payload.get("scope", "")
    expanded_scope = f"{existing_scopes} threat:write mfa"

    new_token = issue_token_with_scope(current_user, expanded_scope)
    return jsonify({
        "access_token": new_token,
        "scope": expanded_scope,
        "expires_in": 3600
    })