0
# OpenID Connect Implementation
1

2
OpenID Connect implementation built on OAuth 2.0, providing identity layer with ID tokens, UserInfo endpoint, and discovery mechanisms. Supports all OpenID Connect flows: authorization code, implicit, and hybrid, with comprehensive claims validation and JWT-based identity tokens.
3

4
## Capabilities
5

6
### ID Token Validation
7

8
ID token claims validation for different OpenID Connect flows with support for standard and custom claims.
9

10
```python { .api }
11
class IDToken:
12
    """Base ID token claims validation."""
13
    
14
    def __init__(self, claims: dict, header: dict = None, options: dict = None, params: dict = None) -> None:
15
        """
16
        Initialize ID token validator.
17
        
18
        Args:
19
            claims: ID token claims dictionary
20
            header: JWT header dictionary
21
            options: Validation options
22
            params: Additional validation parameters
23
        """
24

25
    def validate(self, now: int = None) -> None:
26
        """
27
        Validate all ID token claims.
28
        
29
        Args:
30
            now: Current timestamp for time-based validations
31
        """
32

33
    def validate_iss(self) -> None:
34
        """Validate issuer claim."""
35

36
    def validate_sub(self) -> None:
37
        """Validate subject claim."""
38

39
    def validate_aud(self) -> None:
40
        """Validate audience claim."""
41

42
    def validate_exp(self, now: int = None, leeway: int = 0) -> None:
43
        """
44
        Validate expiration time claim.
45
        
46
        Args:
47
            now: Current timestamp
48
            leeway: Acceptable time skew in seconds
49
        """
50

51
    def validate_iat(self, now: int = None, leeway: int = 0) -> None:
52
        """
53
        Validate issued at claim.
54
        
55
        Args:
56
            now: Current timestamp
57
            leeway: Acceptable time skew in seconds
58
        """
59

60
    def validate_auth_time(self, max_age: int = None, now: int = None, leeway: int = 0) -> None:
61
        """
62
        Validate authentication time claim.
63
        
64
        Args:
65
            max_age: Maximum authentication age
66
            now: Current timestamp
67
            leeway: Acceptable time skew in seconds
68
        """
69

70
    def validate_nonce(self, nonce: str) -> None:
71
        """
72
        Validate nonce claim.
73
        
74
        Args:
75
            nonce: Expected nonce value
76
        """
77

78
    def validate_acr(self, values: list) -> None:
79
        """
80
        Validate Authentication Context Class Reference.
81
        
82
        Args:
83
            values: List of acceptable ACR values
84
        """
85

86
    def validate_amr(self, values: list) -> None:
87
        """
88
        Validate Authentication Methods References.
89
        
90
        Args:
91
            values: List of acceptable AMR values
92
        """
93

94
    def validate_azp(self, client_id: str) -> None:
95
        """
96
        Validate authorized party claim.
97
        
98
        Args:
99
            client_id: Expected client ID
100
        """
101

102
class CodeIDToken(IDToken):
103
    """ID token validation for authorization code flow."""
104
    
105
    def validate_at_hash(self, access_token: str, alg: str) -> None:
106
        """
107
        Validate access token hash claim.
108
        
109
        Args:
110
            access_token: Access token to verify hash against
111
            alg: JWT signing algorithm
112
        """
113

114
    def validate_c_hash(self, code: str, alg: str) -> None:
115
        """
116
        Validate authorization code hash claim.
117
        
118
        Args:
119
            code: Authorization code to verify hash against
120
            alg: JWT signing algorithm
121
        """
122

123
class ImplicitIDToken(IDToken):
124
    """ID token validation for implicit flow."""
125
    
126
    def validate_at_hash(self, access_token: str, alg: str) -> None:
127
        """
128
        Validate access token hash claim.
129
        
130
        Args:
131
            access_token: Access token to verify hash against
132
            alg: JWT signing algorithm
133
        """
134

135
class HybridIDToken(IDToken):
136
    """ID token validation for hybrid flow."""
137
    
138
    def validate_at_hash(self, access_token: str, alg: str) -> None:
139
        """
140
        Validate access token hash claim.
141
        
142
        Args:
143
            access_token: Access token to verify hash against
144
            alg: JWT signing algorithm
145
        """
146

147
    def validate_c_hash(self, code: str, alg: str) -> None:
148
        """
149
        Validate authorization code hash claim.
150
        
151
        Args:
152
            code: Authorization code to verify hash against
153
            alg: JWT signing algorithm
154
        """
155
```
156

157
### UserInfo Endpoint
158

159
UserInfo endpoint implementation for serving user claims to relying parties.
160

161
```python { .api }
162
class UserInfo:
163
    """UserInfo claims representation."""
164
    
165
    def __init__(self, sub: str, **claims) -> None:
166
        """
167
        Initialize UserInfo with subject and claims.
168
        
169
        Args:
170
            sub: Subject identifier (required)
171
            **claims: Additional user claims
172
        """
173

174
    def __call__(self, claims: list = None) -> dict:
175
        """
176
        Get user claims, optionally filtered by requested claims.
177
        
178
        Args:
179
            claims: List of requested claims
180
            
181
        Returns:
182
            Dictionary of user claims
183
        """
184

185
class UserInfoEndpoint:
186
    """UserInfo endpoint implementation."""
187
    
188
    def __init__(self, query_token: callable, query_user: callable) -> None:
189
        """
190
        Initialize UserInfo endpoint.
191
        
192
        Args:
193
            query_token: Function to query and validate access token
194
            query_user: Function to query user by subject identifier
195
        """
196

197
    def create_userinfo_response(self, request: OAuth2Request) -> tuple:
198
        """
199
        Create UserInfo endpoint response.
200
        
201
        Args:
202
            request: OAuth2Request object
203
            
204
        Returns:
205
            Tuple of (status_code, body, headers)
206
        """
207

208
    def validate_userinfo_request(self, request: OAuth2Request) -> dict:
209
        """
210
        Validate UserInfo request and return token.
211
        
212
        Args:
213
            request: OAuth2Request object
214
            
215
        Returns:
216
            Token dictionary
217
        """
218

219
    def generate_userinfo(self, user: object, scope: str) -> dict:
220
        """
221
        Generate UserInfo response for user and scope.
222
        
223
        Args:
224
            user: User object
225
            scope: Access token scope
226
            
227
        Returns:
228
            UserInfo claims dictionary
229
        """
230
```
231

232
### OpenID Connect Grant Types
233

234
Enhanced OAuth 2.0 grants with OpenID Connect ID token support.
235

236
```python { .api }
237
class OpenIDCode:
238
    """OpenID Connect authorization code grant enhancement."""
239
    
240
    def create_id_token(self, token: dict, request: OAuth2Request, nonce: str = None) -> str:
241
        """
242
        Create ID token for authorization code flow.
243
        
244
        Args:
245
            token: Access token dictionary
246
            request: OAuth2Request object
247
            nonce: Nonce from authorization request
248
            
249
        Returns:
250
            ID token JWT string
251
        """
252

253
    def get_jwt_config(self, grant: object) -> dict:
254
        """
255
        Get JWT configuration for ID token.
256
        
257
        Args:
258
            grant: Grant object
259
            
260
        Returns:
261
            JWT configuration dictionary
262
        """
263

264
    def generate_user_info(self, user: object, scope: str) -> dict:
265
        """
266
        Generate user info claims for ID token.
267
        
268
        Args:
269
            user: User object
270
            scope: Token scope
271
            
272
        Returns:
273
            User info claims dictionary
274
        """
275

276
class OpenIDToken:
277
    """OpenID Connect token grant (refresh token with ID token)."""
278
    
279
    def create_id_token(self, token: dict, request: OAuth2Request, nonce: str = None) -> str:
280
        """
281
        Create ID token for token refresh.
282
        
283
        Args:
284
            token: Access token dictionary
285
            request: OAuth2Request object
286
            nonce: Nonce from original authorization
287
            
288
        Returns:
289
            ID token JWT string
290
        """
291

292
class OpenIDImplicitGrant:
293
    """OpenID Connect implicit grant implementation."""
294
    
295
    RESPONSE_TYPES: list = ['id_token', 'id_token token']
296
    
297
    def create_id_token(self, grant_user: callable, request: OAuth2Request) -> str:
298
        """
299
        Create ID token for implicit flow.
300
        
301
        Args:
302
            grant_user: Function to grant user
303
            request: OAuth2Request object
304
            
305
        Returns:
306
            ID token JWT string
307
        """
308

309
    def exists_nonce(self, nonce: str, request: OAuth2Request) -> bool:
310
        """
311
        Check if nonce exists and is valid.
312
        
313
        Args:
314
            nonce: Nonce value
315
            request: OAuth2Request object
316
            
317
        Returns:
318
            True if nonce is valid
319
        """
320

321
class OpenIDHybridGrant:
322
    """OpenID Connect hybrid grant implementation."""
323
    
324
    RESPONSE_TYPES: list = ['code id_token', 'code token', 'code id_token token']
325
    
326
    def create_id_token(self, token: dict, request: OAuth2Request, nonce: str = None, code: str = None) -> str:
327
        """
328
        Create ID token for hybrid flow.
329
        
330
        Args:
331
            token: Access token dictionary (if issued)
332
            request: OAuth2Request object
333
            nonce: Nonce from authorization request
334
            code: Authorization code (if issued)
335
            
336
        Returns:
337
            ID token JWT string
338
        """
339
```
340

341
### OpenID Connect Discovery
342

343
Provider metadata and discovery implementation following OpenID Connect Discovery 1.0.
344

345
```python { .api }
346
class OpenIDProviderMetadata:
347
    """OpenID Connect provider metadata."""
348
    
349
    def __init__(self, issuer: str, **metadata) -> None:
350
        """
351
        Initialize provider metadata.
352
        
353
        Args:
354
            issuer: Provider issuer identifier
355
            **metadata: Additional metadata parameters
356
        """
357

358
    def validate(self) -> None:
359
        """Validate provider metadata for required fields."""
360

361
    def get_jwks_uri(self) -> str:
362
        """Get JSON Web Key Set URI."""
363

364
    def get_supported_scopes(self) -> list:
365
        """Get list of supported scopes."""
366

367
    def get_supported_response_types(self) -> list:
368
        """Get list of supported response types."""
369

370
    def get_supported_grant_types(self) -> list:
371
        """Get list of supported grant types."""
372

373
    def get_supported_id_token_signing_alg_values(self) -> list:
374
        """Get list of supported ID token signing algorithms."""
375

376
    def get_supported_claims(self) -> list:
377
        """Get list of supported claims."""
378

379
def get_well_known_url(issuer: str, external: bool = False) -> str:
380
    """
381
    Generate OpenID Connect well-known configuration URL.
382
    
383
    Args:
384
        issuer: Provider issuer identifier
385
        external: Whether to use external URL format
386
        
387
    Returns:
388
        Well-known configuration URL
389
    """
390
```
391

392
### Client Registration
393

394
Dynamic client registration following OpenID Connect Registration 1.0.
395

396
```python { .api }
397
class ClientMetadataClaims:
398
    """Client metadata claims validation."""
399
    
400
    def __init__(self, claims: dict, **params) -> None:
401
        """
402
        Initialize client metadata validator.
403
        
404
        Args:
405
            claims: Client metadata claims
406
            **params: Additional validation parameters
407
        """
408

409
    def validate(self) -> None:
410
        """Validate all client metadata claims."""
411

412
    def validate_redirect_uris(self) -> None:
413
        """Validate redirect URIs."""
414

415
    def validate_response_types(self) -> None:
416
        """Validate response types."""
417

418
    def validate_grant_types(self) -> None:
419
        """Validate grant types."""
420

421
    def validate_application_type(self) -> None:
422
        """Validate application type (web or native)."""
423

424
    def validate_client_name(self) -> None:
425
        """Validate client name."""
426

427
    def validate_client_uri(self) -> None:
428
        """Validate client URI."""
429

430
    def validate_logo_uri(self) -> None:
431
        """Validate logo URI."""
432

433
    def validate_scope(self) -> None:
434
        """Validate requested scopes."""
435

436
    def validate_contacts(self) -> None:
437
        """Validate contact information."""
438

439
    def validate_tos_uri(self) -> None:
440
        """Validate terms of service URI."""
441

442
    def validate_policy_uri(self) -> None:
443
        """Validate privacy policy URI."""
444

445
    def validate_jwks_uri(self) -> None:
446
        """Validate JWKS URI."""
447

448
    def validate_jwks(self) -> None:
449
        """Validate embedded JWKS."""
450

451
    def validate_software_id(self) -> None:
452
        """Validate software ID."""
453

454
    def validate_software_version(self) -> None:
455
        """Validate software version."""
456
```
457

458
### Authorization Code Mixin
459

460
Enhanced authorization code mixin with OpenID Connect support.
461

462
```python { .api }
463
class AuthorizationCodeMixin:
464
    """Mixin for OIDC authorization code model."""
465
    
466
    code: str  # Authorization code
467
    client_id: str  # Client identifier
468
    redirect_uri: str  # Redirect URI
469
    scope: str  # Authorized scope
470
    user_id: str  # User identifier
471
    code_challenge: str  # PKCE code challenge
472
    code_challenge_method: str  # PKCE challenge method
473
    nonce: str  # OpenID Connect nonce
474
    
475
    def get_nonce(self) -> str:
476
        """
477
        Get OpenID Connect nonce.
478
        
479
        Returns:
480
            Nonce value
481
        """
482

483
    def get_auth_time(self) -> int:
484
        """
485
        Get authentication time.
486
        
487
        Returns:
488
            Authentication timestamp
489
        """
490
```
491

492
### Utility Functions
493

494
Helper functions for OpenID Connect flows and claim handling.
495

496
```python { .api }
497
def get_claim_cls_by_response_type(response_type: str) -> type:
498
    """
499
    Get appropriate ID token claims class by response type.
500
    
501
    Args:
502
        response_type: OAuth 2.0/OIDC response type
503
        
504
    Returns:
505
        ID token claims validation class
506
    """
507
```
508

509
## Usage Examples
510

511
### ID Token Validation
512

513
```python
514
from authlib.oidc.core import IDToken, CodeIDToken
515
from authlib.jose import JsonWebToken
516

517
# Decode and validate ID token
518
id_token_string = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9...'
519
jwt = JsonWebToken(['RS256'])
520

521
# Decode JWT
522
claims = jwt.decode(id_token_string, public_key)
523

524
# Validate ID token claims for authorization code flow
525
id_token = CodeIDToken(claims, header={'alg': 'RS256'})
526
id_token.validate()
527

528
# Validate access token hash (for code flow)
529
id_token.validate_at_hash('access-token-value', 'RS256')
530

531
# Validate with specific parameters
532
id_token = IDToken(claims, options={
533
    'iss': {'essential': True, 'value': 'https://provider.com'},
534
    'aud': {'essential': True, 'value': 'client-id'}
535
})
536
id_token.validate()
537
```
538

539
### UserInfo Endpoint
540

541
```python
542
from authlib.oidc.core import UserInfo, UserInfoEndpoint
543
from authlib.oauth2 import OAuth2Request
544

545
# Create UserInfo representation
546
user_info = UserInfo(
547
    sub='user123',
548
    name='John Doe',
549
    email='john@example.com',
550
    email_verified=True,
551
    picture='https://example.com/photo.jpg'
552
)
553

554
# Get filtered claims
555
profile_claims = user_info(['name', 'picture'])
556
# Returns: {'sub': 'user123', 'name': 'John Doe', 'picture': '...'}
557

558
# Server-side UserInfo endpoint
559
def query_token(access_token):
560
    # Validate and return token info
561
    return get_token_by_value(access_token)
562

563
def query_user(sub):
564
    # Return user by subject
565
    return get_user_by_id(sub)
566

567
userinfo_endpoint = UserInfoEndpoint(
568
    query_token=query_token,
569
    query_user=query_user
570
)
571

572
# Handle UserInfo request
573
request = OAuth2Request('GET', '/userinfo', headers={'Authorization': 'Bearer token'})
574
status_code, body, headers = userinfo_endpoint.create_userinfo_response(request)
575
```
576

577
### OpenID Connect Authorization Server
578

579
```python
580
from authlib.oauth2 import AuthorizationServer
581
from authlib.oauth2.rfc6749.grants import AuthorizationCodeGrant
582
from authlib.oidc.core.grants import OpenIDCode
583
from authlib.oidc.core import UserInfo
584

585
# Enhance authorization code grant with OpenID Connect
586
class AuthorizationCodeGrantWithOpenID(AuthorizationCodeGrant):
587
    def __init__(self, request, server):
588
        super().__init__(request, server)
589
        # Add OpenID Connect support
590
        self.openid_code = OpenIDCode()
591

592
    def create_token_response(self):
593
        token = self.generate_token()
594
        
595
        # Add ID token for OpenID Connect requests
596
        if 'openid' in self.request.scope:
597
            id_token = self.openid_code.create_id_token(
598
                token=token,
599
                request=self.request,
600
                nonce=self.request.data.get('nonce')
601
            )
602
            token['id_token'] = id_token
603
        
604
        return token
605

606
# Register enhanced grant
607
authorization_server = AuthorizationServer(query_client, save_token)
608
authorization_server.register_grant(AuthorizationCodeGrantWithOpenID)
609

610
# Add UserInfo endpoint
611
userinfo_endpoint = UserInfoEndpoint(query_token, query_user)
612
authorization_server.register_endpoint(userinfo_endpoint)
613
```
614

615
### Discovery Endpoint
616

617
```python
618
from authlib.oidc.discovery import OpenIDProviderMetadata, get_well_known_url
619

620
# Create provider metadata
621
metadata = OpenIDProviderMetadata(
622
    issuer='https://provider.com',
623
    authorization_endpoint='https://provider.com/authorize',
624
    token_endpoint='https://provider.com/token',
625
    userinfo_endpoint='https://provider.com/userinfo',
626
    jwks_uri='https://provider.com/.well-known/jwks.json',
627
    scopes_supported=['openid', 'profile', 'email'],
628
    response_types_supported=['code', 'token', 'id_token', 'code token', 'code id_token', 'id_token token', 'code id_token token'],
629
    subject_types_supported=['public'],
630
    id_token_signing_alg_values_supported=['RS256', 'HS256'],
631
    claims_supported=['sub', 'iss', 'aud', 'exp', 'iat', 'name', 'email', 'email_verified']
632
)
633

634
# Validate metadata
635
metadata.validate()
636

637
# Generate well-known URL
638
well_known_url = get_well_known_url('https://provider.com')
639
# Returns: https://provider.com/.well-known/openid_configuration
640

641
# Serve discovery endpoint
642
@app.route('/.well-known/openid_configuration')
643
def openid_configuration():
644
    return jsonify(metadata.__dict__)
645
```
646

647
### Client Integration
648

649
```python
650
from authlib.integrations.requests_client import OAuth2Session
651
from authlib.oidc.core import IDToken
652
from authlib.jose import JsonWebToken
653

654
# OpenID Connect client flow
655
client = OAuth2Session(
656
    client_id='client-id',
657
    client_secret='client-secret',
658
    scope='openid profile email'
659
)
660

661
# Authorization with nonce
662
import secrets
663
nonce = secrets.token_urlsafe(32)
664

665
auth_url, state = client.create_authorization_url(
666
    'https://provider.com/authorize',
667
    nonce=nonce
668
)
669

670
# Exchange code for tokens (including ID token)
671
token = client.fetch_token(
672
    'https://provider.com/token',
673
    authorization_response='https://callback.com?code=...'
674
)
675

676
# Validate ID token
677
if 'id_token' in token:
678
    jwt = JsonWebToken(['RS256'])
679
    # Get provider's public key from JWKS endpoint
680
    public_key = get_provider_public_key()
681
    
682
    # Decode and validate ID token
683
    claims = jwt.decode(token['id_token'], public_key)
684
    id_token = IDToken(claims)
685
    id_token.validate()
686
    id_token.validate_nonce(nonce)
687
    
688
    print(f"User ID: {claims['sub']}")
689
    print(f"User name: {claims.get('name')}")
690

691
# Get additional user info
692
userinfo_response = client.get('https://provider.com/userinfo')
693
user_claims = userinfo_response.json()
694
```