1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
"""
oauthlib.openid.connect.core.endpoints.userinfo
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
This module is an implementation of userinfo endpoint.
"""
import json
import logging
from oauthlib.common import Request
from oauthlib.oauth2.rfc6749 import errors
from oauthlib.oauth2.rfc6749.endpoints.base import (
BaseEndpoint, catch_errors_and_unavailability,
)
from oauthlib.oauth2.rfc6749.tokens import BearerToken
log = logging.getLogger(__name__)
class UserInfoEndpoint(BaseEndpoint):
"""Authorizes access to userinfo resource.
"""
def __init__(self, request_validator):
self.bearer = BearerToken(request_validator, None, None, None)
self.request_validator = request_validator
BaseEndpoint.__init__(self)
@catch_errors_and_unavailability
def create_userinfo_response(self, uri, http_method='GET', body=None, headers=None):
"""Validate BearerToken and return userinfo from RequestValidator
The UserInfo Endpoint MUST return a
content-type header to indicate which format is being returned. The
content-type of the HTTP response MUST be application/json if the
response body is a text JSON object; the response body SHOULD be encoded
using UTF-8.
"""
request = Request(uri, http_method, body, headers)
request.scopes = ["openid"]
self.validate_userinfo_request(request)
claims = self.request_validator.get_userinfo_claims(request)
if claims is None:
log.error('Userinfo MUST have claims for %r.', request)
raise errors.ServerError(status_code=500)
if isinstance(claims, dict):
resp_headers = {
'Content-Type': 'application/json'
}
if "sub" not in claims:
log.error('Userinfo MUST have "sub" for %r.', request)
raise errors.ServerError(status_code=500)
body = json.dumps(claims)
elif isinstance(claims, str):
resp_headers = {
'Content-Type': 'application/jwt'
}
body = claims
else:
log.error('Userinfo return unknown response for %r.', request)
raise errors.ServerError(status_code=500)
log.debug('Userinfo access valid for %r.', request)
return resp_headers, body, 200
def validate_userinfo_request(self, request):
"""Ensure the request is valid.
5.3.1. UserInfo Request
The Client sends the UserInfo Request using either HTTP GET or HTTP
POST. The Access Token obtained from an OpenID Connect Authentication
Request MUST be sent as a Bearer Token, per `Section 2`_ of OAuth 2.0
Bearer Token Usage [RFC6750].
It is RECOMMENDED that the request use the HTTP GET method and the
Access Token be sent using the Authorization header field.
The following is a non-normative example of a UserInfo Request:
.. code-block:: http
GET /userinfo HTTP/1.1
Host: server.example.com
Authorization: Bearer SlAV32hkKG
5.3.3. UserInfo Error Response
When an error condition occurs, the UserInfo Endpoint returns an Error
Response as defined in `Section 3`_ of OAuth 2.0 Bearer Token Usage
[RFC6750]. (HTTP errors unrelated to RFC 6750 are returned to the User
Agent using the appropriate HTTP status code.)
The following is a non-normative example of a UserInfo Error Response:
.. code-block:: http
HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer error="invalid_token",
error_description="The Access Token expired"
.. _`Section 2`: https://datatracker.ietf.org/doc/html/rfc6750#section-2
.. _`Section 3`: https://datatracker.ietf.org/doc/html/rfc6750#section-3
"""
if not self.bearer.validate_request(request):
raise errors.InvalidTokenError()
if "openid" not in request.scopes:
raise errors.InsufficientScopeError()
|
