OAuth2 Back-Channel Logout — RFC 7009 Session Termination Across Providers

import jwt
from datetime import datetime, timedelta, timezone

def create_logout_token(session_id, user_id, issuer, private_key,
                        events=None, sid=None):
    """Create a logout token per OpenID Connect Back-Channel Logout spec."""
    if events is None:
        events = {
            'http://schemas.openid.net/event/backchannel-logout': {}
        }

    now = datetime.now(timezone.utc)
    payload = {
        'iss': issuer,
        'sub': user_id,
        'aud': issuer,  # Each RP will verify with its own client_id
        'iat': now,
        'exp': now + timedelta(minutes=5),
        'jti': str(uuid.uuid4()),
        'events': events,
        'sid': sid or session_id
    }

    token = jwt.encode(payload, private_key, algorithm='RS256')
    print(f"Logout token issued for session {session_id[:8]}...")
    return token

# Usage
logout_token = create_logout_token(
    session_id='sess_abc123',
    user_id='user_456',
    issuer='https://auth.dodatech.com',
    private_key=PRIVATE_KEY
)

from flask import Flask, request, jsonify
import jwt
from jwt import PyJWKClient

app = Flask(__name__)
jwks_client = PyJWKClient('https://auth.dodatech.com/.well-known/jwks.json')

@app.route('/backchannel-logout', methods=['POST'])
def backchannel_logout():
    """Receive back-channel logout request from the OP."""
    logout_token = request.form.get('logout_token')
    if not logout_token:
        return jsonify({'error': 'missing_logout_token'}), 400

    try:
        # Validate logout token
        signing_key = jwks_client.get_signing_key_from_jwt(logout_token)
        payload = jwt.decode(
            logout_token,
            signing_key.key,
            algorithms=['RS256'],
            issuer='https://auth.dodatech.com',
            audience=CLIENT_ID,
            options={'require': ['events', 'sub', 'sid', 'iat']}
        )

        # Verify it's a logout event
        events = payload.get('events', {})
        if 'http://schemas.openid.net/event/backchannel-logout' not in events:
            return jsonify({'error': 'invalid_logout_event'}), 400

        # Terminate local session
        session_id = payload['sid']
        user_id = payload['sub']

        if terminate_local_session(session_id, user_id):
            print(f"Session terminated: user={user_id}, session={session_id[:8]}...")
            return jsonify({'result': 'ok'}), 200
        else:
            # Session not found — still return 200 per spec
            return jsonify({'result': 'ok'}), 200

    except jwt.ExpiredSignatureError:
        return jsonify({'error': 'expired_logout_token'}), 400
    except Exception as e:
        print(f"Logout token validation failed: {e}")
        return jsonify({'error': 'invalid_logout_token'}), 400

def terminate_local_session(session_id, user_id):
    """Terminate the local session. Return False if session not found."""
    session = find_local_session(session_id, user_id)
    if session:
        delete_local_session(session_id)
        # Invalidate all tokens issued for this session
        invalidate_session_tokens(session_id)
        return True
    return False

@app.route('/logout')
def rp_initiated_logout():
    """Handle RP-initiated logout per OpenID Connect RP-Initiated Logout."""
    id_token_hint = request.args.get('id_token_hint')
    post_logout_redirect_uri = request.args.get('post_logout_redirect_uri')
    state = request.args.get('state')

    # Validate the logout request
    if id_token_hint:
        try:
            payload = validate_id_token(id_token_hint)
            user_id = payload['sub']
            session_id = payload.get('sid')

            # Clear local session
            clear_user_session(user_id)

            # Redirect to OP for global logout
            op_logout_url = build_op_logout_url(
                id_token_hint,
                post_logout_redirect_uri,
                state
            )
            return redirect(op_logout_url)

        except Exception as e:
            print(f"Invalid id_token_hint: {e}")

    # Simple local logout
    clear_session(request)
    return redirect('/goodbye')

def build_op_logout_url(id_token_hint, post_logout_redirect_uri, state):
    """Build logout URL to send the user to the OP for global logout."""
    params = {
        'id_token_hint': id_token_hint,
        'post_logout_redirect_uri': post_logout_redirect_uri or
            'https://dashboard.dodatech.com/logged-out',
        'state': state or secrets.token_urlsafe(16)
    }
    query = urlencode(params)
    return f"https://auth.dodatech.com/logout?{query}"