SAML Authentication — Browser-Based Single Sign-On for Enterprise APIs

from flask import Flask, request, redirect, jsonify
from onelogin.saml2.auth import OneLogin_Saml2_Auth
from onelogin.saml2.utils import OneLogin_Saml2_Utils
import jwt, datetime, secrets, os

app = Flask(__name__)
SECRET = os.environ.get("JWT_SECRET", "dev-secret")

# SAML configuration
SAML_CONFIG = {
    "strict": True,
    "debug": True,
    "sp": {
        "entityId": "https://api.durga-antivirus.com/saml/metadata",
        "assertionConsumerService": {
            "url": "https://api.durga-antivirus.com/api/auth/saml/acs",
            "binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
        },
        "singleLogoutService": {
            "url": "https://api.durga-antivirus.com/api/auth/saml/sls",
            "binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
        },
        "NameIDFormat": "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
    },
    "idp": {
        "entityId": os.environ.get("SAML_IDP_ENTITY_ID"),
        "singleSignOnService": {
            "url": os.environ.get("SAML_SSO_URL"),
            "binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
        },
        "x509cert": os.environ.get("SAML_IDP_CERT", "").replace("\\n", "\n")
    }
}

def init_saml_auth(req):
    """Initialize SAML authentication from Flask request."""
    auth = OneLogin_Saml2_Auth(req, SAML_CONFIG)
    return auth

def prepare_flask_request(request):
    """Convert Flask request to format expected by python3-saml."""
    url_data = {
        "http_host": request.host,
        "server_port": request.host.split(":")[1] if ":" in request.host else "443",
        "script_name": request.path,
        "get_data": request.args.to_dict(),
        "post_data": request.form.to_dict(),
        "query_string": request.query_string.decode() if request.query_string else ""
    }
    return url_data

@app.route("/api/auth/saml/login")
def saml_login():
    """Initiate SP-initiated SAML SSO."""
    req = prepare_flask_request(request)
    auth = init_saml_auth(req)
    return redirect(auth.login())

@app.route("/api/auth/saml/acs", methods=["POST"])
def saml_acs():
    """Assertion Consumer Service — receive SAML response."""
    req = prepare_flask_request(request)
    auth = init_saml_auth(req)

    # Process SAML response
    auth.process_response()
    errors = auth.get_errors()

    if errors:
        return jsonify({
            "error": "SAML authentication failed",
            "details": errors
        }), 401

    if not auth.is_authenticated():
        return jsonify({"error": "Not authenticated"}), 401

    # Extract user attributes from SAML assertion
    attributes = auth.get_attributes()
    name_id = auth.get_nameid()

    # Build user profile from SAML claims
    user_data = {
        "name_id": name_id,
        "email": attributes.get("email", [name_id])[0],
        "first_name": attributes.get("givenName", [""])[0],
        "last_name": attributes.get("sn", [""])[0],
        "groups": attributes.get("memberOf", []),
        "idp": SAML_CONFIG["idp"]["entityId"]
    }

    # Issue API token
    access_token = jwt.encode({
        "sub": user_data["email"],
        "name": f"{user_data['first_name']} {user_data['last_name']}".strip(),
        "groups": user_data["groups"],
        "auth_method": "saml",
        "iat": datetime.datetime.utcnow(),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=24)
    }, SECRET, algorithm="HS256")

    return jsonify({
        "access_token": access_token,
        "token_type": "Bearer",
        "expires_in": 86400,
        "user": {
            "email": user_data["email"],
            "groups": user_data["groups"]
        }
    })

@app.route("/api/auth/saml/metadata")
def saml_metadata():
    """Return SAML SP metadata XML for IdP configuration."""
    req = prepare_flask_request(request)
    auth = init_saml_auth(req)
    metadata = auth.get_settings().get_sp_metadata()
    errors = auth.get_settings().validate_metadata(metadata)

    if errors:
        return jsonify({"error": "Invalid metadata", "details": errors}), 500

    return metadata, 200, {"Content-Type": "application/xml"}

@app.route("/api/auth/saml/sls", methods=["GET", "POST"])
def saml_sls():
    """Single Logout Service."""
    req = prepare_flask_request(request)
    auth = init_saml_auth(req)

    url = auth.process_slo(delete_session_cb=lambda: clear_session())
    errors = auth.get_errors()

    if url:
        return redirect(url)

    return jsonify({"message": "Logged out"})

def clear_session():
    """Clear user session on SAML logout."""
    # Invalidate any active tokens for this session
    pass

# API token exchange for SAML-authenticated requests
@app.route("/api/auth/saml/exchange", methods=["POST"])
def exchange_saml_token():
    """Exchange a SAML assertion for a scoped API token."""
    assertion = request.json.get("saml_assertion")
    requested_scopes = request.json.get("scopes", [])
    saml_settings = OneLogin_Saml2_Settings(SAML_CONFIG)

    # Validate and decrypt assertion
    response = OneLogin_Saml2_Response(
        saml_settings, assertion
    )

    if not response.is_valid():
        return jsonify({"error": "Invalid SAML assertion"}), 401

    # Extract attributes and issue scoped token
    attributes = response.get_attributes()
    user_groups = attributes.get("memberOf", [])

    # Map SAML groups to API scopes
    allowed_scopes = map_groups_to_scopes(user_groups)
    effective_scopes = [s for s in requested_scopes if s in allowed_scopes]

    token = jwt.encode({
        "sub": response.get_nameid(),
        "scopes": effective_scopes,
        "auth_method": "saml_exchange",
        "iat": datetime.datetime.utcnow(),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    }, SECRET, algorithm="HS256")

    return jsonify({"access_token": token, "scopes": effective_scopes})