Python-Keycloak Library

pip install python-keycloak

conda install -c conda-forge python-keycloak

from keycloak import KeycloakOpenID

keycloak_openid = KeycloakOpenID(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    client_id="your-client",
    client_secret_key="your-secret"
)

# Get a token
token = keycloak_openid.token("username", "password")

# Decode with audience verification disabled
options = {
    "verify_aud": False,      # Disable audience check
    "verify_signature": True,  # Keep signature verification
    "verify_exp": True         # Keep expiration check
}

token_info = keycloak_openid.decode_token(
    token['access_token'],
    options=options
)

# Now works without disabling audience verification
token_info = keycloak_openid.decode_token(token['access_token'])

from keycloak import KeycloakOpenID

keycloak_openid = KeycloakOpenID(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    client_id="your-client",
    client_secret_key="your-secret"  # Optional for public clients
)

# Get token with username/password
token = keycloak_openid.token("username", "password")

print(token['access_token'])
print(token['refresh_token'])
print(token['expires_in'])

# For service-to-service authentication
token = keycloak_openid.token(grant_type="client_credentials")

# Refresh an expired access token
new_token = keycloak_openid.refresh_token(token['refresh_token'])

# Decode and validate token
KEYCLOAK_PUBLIC_KEY = (
    "-----BEGIN PUBLIC KEY-----\n"
    + keycloak_openid.public_key()
    + "\n-----END PUBLIC KEY-----"
)

token_info = keycloak_openid.decode_token(
    token['access_token'],
    key=KEYCLOAK_PUBLIC_KEY,
    options={"verify_aud": False}  # See PyJWT Troubleshooting section
)

print(token_info['sub'])           # User ID
print(token_info['preferred_username'])
print(token_info['realm_access'])  # Realm roles

# Get user info from userinfo endpoint
userinfo = keycloak_openid.userinfo(token['access_token'])

print(userinfo['email'])
print(userinfo['name'])

# Validate token and get metadata
introspection = keycloak_openid.introspect(token['access_token'])

if introspection['active']:
    print(f"Token valid, expires: {introspection['exp']}")
else:
    print("Token is invalid or expired")

# Logout user (invalidate refresh token)
keycloak_openid.logout(token['refresh_token'])

from keycloak import KeycloakAdmin

keycloak_admin = KeycloakAdmin(
    server_url="https://your-cluster.app.skycloak.io",
    username="admin",
    password="admin-password",
    realm_name="master",           # Admin realm
    user_realm_name="your-realm"   # Target realm for operations
)

from keycloak import KeycloakAdmin, KeycloakOpenIDConnection

# Create connection with existing token
connection = KeycloakOpenIDConnection(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    token=existing_token  # Pre-acquired admin token
)

keycloak_admin = KeycloakAdmin(connection=connection)

from keycloak import KeycloakAdmin, KeycloakOpenIDConnection

# Use client credentials (service account must have admin roles)
connection = KeycloakOpenIDConnection(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    client_id="admin-cli",
    client_secret_key="your-service-account-secret"
)

keycloak_admin = KeycloakAdmin(connection=connection)

# Create a basic user
user_id = keycloak_admin.create_user({
    "username": "newuser",
    "email": "[email protected]",
    "enabled": True,
    "firstName": "New",
    "lastName": "User",
    "credentials": [{
        "type": "password",
        "value": "initial-password",
        "temporary": True  # Force password change on first login
    }]
})

print(f"Created user with ID: {user_id}")

user_id = keycloak_admin.create_user({
    "username": "enterprise-user",
    "email": "[email protected]",
    "enabled": True,
    "firstName": "Enterprise",
    "lastName": "User",
    "emailVerified": True,
    "attributes": {
        "company": ["Acme Corp"],
        "department": ["Engineering"],
        "employee_id": ["EMP-12345"]
    },
    "groups": ["/engineering", "/full-time"],
    "realmRoles": ["user", "viewer"]
})

# Get all users
users = keycloak_admin.get_users({})

# Get users with filtering
users = keycloak_admin.get_users({
    "email": "[email protected]"
})

# Get users with pagination
users = keycloak_admin.get_users({
    "first": 0,      # Offset
    "max": 100,       # Limit
    "search": "john"  # Search by username, email, first/last name
})

# Count total users
count = keycloak_admin.users_count()
print(f"Total users: {count}")

user = keycloak_admin.get_user(user_id)

print(user['username'])
print(user['email'])
print(user['attributes'])

# Update user properties
keycloak_admin.update_user(user_id, {
    "firstName": "Updated",
    "lastName": "Name",
    "attributes": {
        "department": ["Marketing"]  # Overwrites existing
    }
})

# Disable user
keycloak_admin.update_user(user_id, {"enabled": False})

# Enable user
keycloak_admin.update_user(user_id, {"enabled": True})

keycloak_admin.delete_user(user_id)

# Set permanent password
keycloak_admin.set_user_password(
    user_id=user_id,
    password="new-secure-password",
    temporary=False
)

# Set temporary password (forces change on next login)
keycloak_admin.set_user_password(
    user_id=user_id,
    password="temp-password",
    temporary=True
)

# Send email verification
keycloak_admin.send_verify_email(user_id)

# Send password reset email
keycloak_admin.send_update_account(
    user_id=user_id,
    payload=["UPDATE_PASSWORD"]
)

# Get all realm roles
roles = keycloak_admin.get_realm_roles()

for role in roles:
    print(f"{role['name']}: {role.get('description', 'No description')}")

# Get specific role
admin_role = keycloak_admin.get_realm_role("admin")

keycloak_admin.create_realm_role({
    "name": "api_user",
    "description": "Can access API endpoints",
    "composite": False
})

# Get role objects
roles = keycloak_admin.get_realm_roles()
user_role = next(r for r in roles if r['name'] == 'user')
api_role = next(r for r in roles if r['name'] == 'api_user')

# Assign roles
keycloak_admin.assign_realm_roles(
    user_id=user_id,
    roles=[user_role, api_role]
)

user_roles = keycloak_admin.get_realm_roles_of_user(user_id)

for role in user_roles:
    print(role['name'])

keycloak_admin.delete_realm_roles_of_user(
    user_id=user_id,
    roles=[api_role]
)

# Get client ID (internal ID, not client_id)
clients = keycloak_admin.get_clients()
my_client = next(c for c in clients if c['clientId'] == 'my-app')
client_internal_id = my_client['id']

# Get client roles
client_roles = keycloak_admin.get_client_roles(client_internal_id)

# Create client role
keycloak_admin.create_client_role(
    client_internal_id,
    {"name": "editor", "description": "Can edit content"}
)

# Assign client role to user
editor_role = keycloak_admin.get_client_role(client_internal_id, "editor")
keycloak_admin.assign_client_role(
    user_id=user_id,
    client_id=client_internal_id,
    roles=[editor_role]
)

# Password grant (user authentication)
token = keycloak_openid.token("username", "password")

# Client credentials (service authentication)
token = keycloak_openid.token(grant_type="client_credentials")

# Authorization code (web flow)
# First, redirect user to:
auth_url = keycloak_openid.auth_url(
    redirect_uri="https://your-app.com/callback",
    scope="openid profile email"
)

# Then exchange code for token:
token = keycloak_openid.token(
    grant_type="authorization_code",
    code="authorization-code-from-callback",
    redirect_uri="https://your-app.com/callback"
)

import requests
from jose import jwt

# Get JWKS (JSON Web Key Set)
jwks_url = f"{keycloak_openid.well_known()['jwks_uri']}"
jwks = requests.get(jwks_url).json()

# Validate token with JWKS
try:
    claims = jwt.decode(
        token['access_token'],
        jwks,
        algorithms=['RS256'],
        audience=keycloak_openid.client_id,
        issuer=keycloak_openid.well_known()['issuer']
    )
    print(f"Valid token for user: {claims['sub']}")
except jwt.JWTError as e:
    print(f"Invalid token: {e}")

import time

class TokenManager:
    def __init__(self, keycloak_openid):
        self.kc = keycloak_openid
        self.token = None
        self.token_expiry = 0

    def get_token(self):
        """Get valid access token, refreshing if needed."""
        current_time = time.time()

        # No token yet
        if self.token is None:
            raise ValueError("No token available. Call authenticate() first.")

        # Token expired or about to expire (30s buffer)
        if current_time >= self.token_expiry - 30:
            self._refresh_token()

        return self.token['access_token']

    def authenticate(self, username, password):
        """Initial authentication."""
        self.token = self.kc.token(username, password)
        self.token_expiry = time.time() + self.token['expires_in']

    def _refresh_token(self):
        """Refresh the access token."""
        try:
            self.token = self.kc.refresh_token(self.token['refresh_token'])
            self.token_expiry = time.time() + self.token['expires_in']
        except Exception as e:
            # Refresh token expired, need re-authentication
            raise ValueError(f"Token refresh failed: {e}")

from keycloak.exceptions import KeycloakAuthenticationError

def make_authenticated_request(token_manager, request_func):
    """Make request with automatic token refresh."""
    try:
        token = token_manager.get_token()
        return request_func(token)
    except KeycloakAuthenticationError:
        # Token invalid, try refresh
        token_manager._refresh_token()
        token = token_manager.get_token()
        return request_func(token)

import asyncio
from keycloak import KeycloakOpenID

async def async_auth():
    keycloak_openid = KeycloakOpenID(
        server_url="https://your-cluster.app.skycloak.io",
        realm_name="your-realm",
        client_id="your-client",
        client_secret_key="your-secret"
    )

    # Async token request
    token = await keycloak_openid.a_token("username", "password")

    # Async userinfo
    userinfo = await keycloak_openid.a_userinfo(token['access_token'])

    # Async introspect
    introspection = await keycloak_openid.a_introspect(token['access_token'])

    return userinfo

# Run async
result = asyncio.run(async_auth())

import secrets
import hashlib
import base64

# Generate PKCE verifier and challenge
code_verifier = secrets.token_urlsafe(32)
code_challenge = base64.urlsafe_b64encode(
    hashlib.sha256(code_verifier.encode()).digest()
).rstrip(b'=').decode()

# Get auth URL with PKCE
auth_url = keycloak_openid.auth_url(
    redirect_uri="https://your-app.com/callback",
    scope="openid profile email",
    code_challenge=code_challenge,
    code_challenge_method="S256"
)

# Exchange code with verifier
token = keycloak_openid.token(
    grant_type="authorization_code",
    code="authorization-code",
    redirect_uri="https://your-app.com/callback",
    code_verifier=code_verifier
)

# Get UMA permissions for a resource
permissions = keycloak_openid.uma_permissions(
    token['access_token'],
    permissions=["resource_id#scope"]
)

# Check if user has permission
has_permission = keycloak_openid.has_uma_access(
    token['access_token'],
    "resource_id#read"
)

keycloak_openid = KeycloakOpenID(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    client_id="your-client",
    custom_headers={
        "X-Custom-Header": "value",
        "User-Agent": "MyApp/1.0"
    },
    proxies={
        "http": "http://proxy:8080",
        "https": "https://proxy:8080"
    },
    verify=True  # SSL verification (set to False for self-signed certs)
)

keycloak_admin = KeycloakAdmin(
    server_url="https://your-cluster.app.skycloak.io",
    username="admin",
    password="admin-password",
    realm_name="your-realm",
    timeout=60,           # Request timeout in seconds
    pool_maxsize=10,      # Max connections in pool
    pool_connections=5    # Pool connections to keep
)

import os
from keycloak import KeycloakOpenID, KeycloakAdmin

keycloak_openid = KeycloakOpenID(
    server_url=os.environ['KEYCLOAK_SERVER_URL'],
    realm_name=os.environ['KEYCLOAK_REALM'],
    client_id=os.environ['KEYCLOAK_CLIENT_ID'],
    client_secret_key=os.environ.get('KEYCLOAK_CLIENT_SECRET')
)

keycloak_admin = KeycloakAdmin(
    server_url=os.environ['KEYCLOAK_SERVER_URL'],
    username=os.environ.get('KEYCLOAK_ADMIN_USER'),
    password=os.environ.get('KEYCLOAK_ADMIN_PASSWORD'),
    realm_name=os.environ['KEYCLOAK_REALM']
)

from keycloak import KeycloakAdmin, KeycloakOpenIDConnection
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Create connection with retry logic
connection = KeycloakOpenIDConnection(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    client_id="admin-cli",
    client_secret_key="secret",
    timeout=30  # 30 second timeout
)

# Add retry adapter
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
connection.session.mount("https://", adapter)
connection.session.mount("http://", adapter)

keycloak_admin = KeycloakAdmin(connection=connection)

# Production: Always verify SSL (default)
keycloak_openid = KeycloakOpenID(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    client_id="your-client",
    verify=True  # Default
)

# Development with self-signed cert (NOT for production)
keycloak_openid = KeycloakOpenID(
    server_url="https://localhost:8443",
    realm_name="your-realm",
    client_id="your-client",
    verify=False  # Disable SSL verification
)

# Custom CA certificate
keycloak_openid = KeycloakOpenID(
    server_url="https://your-cluster.app.skycloak.io",
    realm_name="your-realm",
    client_id="your-client",
    verify="/path/to/ca-bundle.crt"
)

from keycloak.exceptions import (
    KeycloakAuthenticationError,
    KeycloakConnectionError,
    KeycloakGetError,
    KeycloakPostError
)

def safe_create_user(admin, user_data):
    """Create user with comprehensive error handling."""
    try:
        user_id = admin.create_user(user_data)
        return {"success": True, "user_id": user_id}

    except KeycloakAuthenticationError as e:
        # Admin credentials invalid or expired
        return {"success": False, "error": "Authentication failed", "details": str(e)}

    except KeycloakConnectionError as e:
        # Network or connectivity issues
        return {"success": False, "error": "Connection failed", "details": str(e)}

    except KeycloakPostError as e:
        # API returned an error (e.g., user exists)
        if e.response_code == 409:
            return {"success": False, "error": "User already exists"}
        return {"success": False, "error": "API error", "details": str(e)}

    except Exception as e:
        return {"success": False, "error": "Unexpected error", "details": str(e)}

from unittest.mock import Mock, patch
import pytest

@pytest.fixture
def mock_keycloak_openid():
    """Create a mock KeycloakOpenID instance."""
    mock = Mock()

    # Mock token response
    mock.token.return_value = {
        'access_token': 'mock-access-token',
        'refresh_token': 'mock-refresh-token',
        'expires_in': 300,
        'token_type': 'Bearer'
    }

    # Mock userinfo response
    mock.userinfo.return_value = {
        'sub': 'user-123',
        'preferred_username': 'testuser',
        'email': '[email protected]',
        'email_verified': True
    }

    # Mock introspect response
    mock.introspect.return_value = {
        'active': True,
        'sub': 'user-123',
        'exp': 1234567890
    }

    return mock

def test_authentication(mock_keycloak_openid):
    """Test authentication flow."""
    token = mock_keycloak_openid.token("testuser", "password")

    assert token['access_token'] == 'mock-access-token'
    mock_keycloak_openid.token.assert_called_once_with("testuser", "password")

import pytest
import time

@pytest.fixture
def valid_token_response():
    return {
        'access_token': 'eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...',
        'refresh_token': 'eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...',
        'expires_in': 300,
        'refresh_expires_in': 1800,
        'token_type': 'Bearer',
        'scope': 'openid profile email'
    }

@pytest.fixture
def decoded_token():
    return {
        'sub': 'user-uuid-123',
        'preferred_username': 'testuser',
        'email': '[email protected]',
        'realm_access': {
            'roles': ['user', 'admin']
        },
        'resource_access': {
            'my-app': {
                'roles': ['editor']
            }
        },
        'exp': int(time.time()) + 300,
        'iat': int(time.time()),
        'iss': 'https://keycloak.example.com/realms/test'
    }

import pytest
import os

@pytest.fixture(scope="module")
def keycloak_openid():
    """Real Keycloak connection for integration tests."""
    from keycloak import KeycloakOpenID

    return KeycloakOpenID(
        server_url=os.environ.get('TEST_KEYCLOAK_URL', 'http://localhost:8080'),
        realm_name=os.environ.get('TEST_KEYCLOAK_REALM', 'test'),
        client_id=os.environ.get('TEST_KEYCLOAK_CLIENT', 'test-client'),
        client_secret_key=os.environ.get('TEST_KEYCLOAK_SECRET')
    )

@pytest.mark.integration
def test_real_authentication(keycloak_openid):
    """Integration test with real Keycloak."""
    token = keycloak_openid.token(
        os.environ['TEST_USER'],
        os.environ['TEST_PASSWORD']
    )

    assert 'access_token' in token

    userinfo = keycloak_openid.userinfo(token['access_token'])
    assert userinfo['preferred_username'] == os.environ['TEST_USER']

# Check credentials
print(f"Client ID: {keycloak_openid.client_id}")
print(f"Realm: {keycloak_openid.realm_name}")

# Verify client secret is set for confidential clients
if not keycloak_openid.client_secret_key:
    print("Warning: No client secret set")

# Check token expiration
import time
if token_info['exp'] < time.time():
    print("Token has expired")

# Check user roles in token
print(f"Realm roles: {token_info.get('realm_access', {}).get('roles', [])}")
print(f"Client roles: {token_info.get('resource_access', {})}")

# For admin operations, ensure service account has:
# - realm-management client roles
# - manage-users, manage-clients, etc.

# Verify server URL format (no trailing slash)
server_url = "https://your-cluster.app.skycloak.io"  # Correct
server_url = "https://your-cluster.app.skycloak.io/"  # May cause issues

# Test connectivity
import requests
try:
    response = requests.get(f"{server_url}/realms/{realm_name}/.well-known/openid-configuration")
    print(f"Status: {response.status_code}")
except requests.exceptions.SSLError:
    print("SSL certificate verification failed")
except requests.exceptions.ConnectionError:
    print("Cannot connect to server")

# Get well-known configuration
well_known = keycloak_openid.well_known()
print(f"Supported algorithms: {well_known.get('id_token_signing_alg_values_supported')}")

# Verify JWKS is accessible
import requests
jwks_response = requests.get(well_known['jwks_uri'])
print(f"JWKS status: {jwks_response.status_code}")

# Check token format
parts = token['access_token'].split('.')
if len(parts) != 3:
    print("Token is malformed (should have 3 parts)")