Advanced SSO Integration

# Navigate to project directory
cd ~/identity-stack

# Create directory for SCIM mock server
mkdir -p scim-mock
cd scim-mock

# Create Python virtual environment
python3 -m venv venv
source venv/bin/activate

# Install required packages
pip install flask gunicorn

#!/usr/bin/env python3
"""
============================================
MOCK SCIM 2.0 SERVER
Test Authentik SCIM provisioning locally
============================================
"""

from flask import Flask, request, jsonify
import sqlite3
import uuid
from datetime import datetime
from functools import wraps

app = Flask(__name__)

# ============================================
# CONFIGURATION
# ============================================
SCIM_TOKEN = "your-secure-token-here"  # Change this!

# ============================================
# DATABASE INITIALIZATION
# ============================================
def init_db():
    """Create SQLite database for storing provisioned users"""
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id TEXT PRIMARY KEY,
            userName TEXT UNIQUE NOT NULL,
            givenName TEXT,
            familyName TEXT,
            email TEXT,
            active INTEGER DEFAULT 1,
            created_at TEXT,
            updated_at TEXT
        )
    ''')
    conn.commit()
    conn.close()
    print("Database initialized")

# ============================================
# AUTHENTICATION DECORATOR
# ============================================
def require_auth(f):
    """Verify Bearer token for SCIM requests"""
    @wraps(f)
    def decorated(*args, **kwargs):
        auth_header = request.headers.get('Authorization', '')
        if not auth_header.startswith('Bearer '):
            return jsonify({'error': 'Missing Authorization header'}), 401
        
        token = auth_header[7:]
        if token != SCIM_TOKEN:
            return jsonify({'error': 'Invalid token'}), 401
        
        return f(*args, **kwargs)
    return decorated

# ============================================
# SCIM 2.0 ENDPOINTS
# ============================================

@app.route('/scim/v2/Users', methods=['POST'])
@require_auth
def create_user():
    """
    SCIM Create User
    Called when a user is added to a provisioned group in Authentik
    """
    data = request.get_json()
    user_id = str(uuid.uuid4())
    now = datetime.utcnow().isoformat()
    
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    
    try:
        cursor.execute('''
            INSERT INTO users (id, userName, givenName, familyName, email, active, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            user_id,
            data.get('userName'),
            data.get('name', {}).get('givenName', ''),
            data.get('name', {}).get('familyName', ''),
            data.get('emails', [{}])[0].get('value', ''),
            1 if data.get('active', True) else 0,
            now,
            now
        ))
        conn.commit()
        
        print(f" User created: {data.get('userName')}")
        
        return jsonify({
            'schemas': ['urn:ietf:params:scim:schemas:core:2.0:User'],
            'id': user_id,
            'userName': data.get('userName'),
            'meta': {
                'resourceType': 'User',
                'created': now,
                'lastModified': now
            }
        }), 201
        
    except sqlite3.IntegrityError:
        return jsonify({'error': 'User already exists'}), 409
    finally:
        conn.close()

@app.route('/scim/v2/Users/<user_id>', methods=['PATCH'])
@require_auth
def update_user(user_id):
    """
    SCIM Update User
    Called when user attributes change in Authentik
    """
    data = request.get_json()
    now = datetime.utcnow().isoformat()
    
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    
    # Process SCIM patch operations
    for op in data.get('Operations', []):
        if op.get('path') == 'active':
            active = 1 if op.get('value') else 0
            cursor.execute(
                'UPDATE users SET active = ?, updated_at = ? WHERE id = ?',
                (active, now, user_id)
            )
            status = "activated" if active else "deactivated"
            print(f" User {user_id} {status}")
    
    conn.commit()
    conn.close()
    
    return jsonify({'status': 'updated'}), 200

@app.route('/scim/v2/Users/<user_id>', methods=['DELETE'])
@require_auth
def delete_user(user_id):
    """
    SCIM Delete User
    Called when user is removed from provisioned group
    """
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
    conn.commit()
    conn.close()
    
    print(f" User deleted: {user_id}")
    return '', 204

@app.route('/scim/v2/Users', methods=['GET'])
@require_auth
def list_users():
    """List all provisioned users"""
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')
    users = cursor.fetchall()
    conn.close()
    
    return jsonify({
        'schemas': ['urn:ietf:params:scim:api:messages:2.0:ListResponse'],
        'totalResults': len(users),
        'Resources': [{
            'id': u[0],
            'userName': u[1],
            'active': bool(u[5])
        } for u in users]
    })

if __name__ == '__main__':
    init_db()
    app.run(host='0.0.0.0', port=5000, debug=True)

# Ensure virtual environment is active
cd ~/identity-stack/scim-mock
source venv/bin/activate

# Start the SCIM server in the background
# Use screen or tmux to keep it running
python scim_server.py &

# Or use gunicorn for production-like setup
gunicorn -w 2 -b 0.0.0.0:5000 scim_server:app &