Skip to content

Authentication and authorization

Helpers are provided for OAuth and Open ID Connect flows:

Other flows are left for the application to implement. Sample code is included for reference. These can be used with or without deriving from OAuthClient

Authentication and Authorization.

Includes helpers for authenticating with Open ID Connect and OAuth and authorization providers for 10Duke API calls.

BearerTokenAuth

Bases: AuthBase

Bearer Token Auth hook - used with OAuth client to connect applications.

Source code in tenduke_core/auth/auth_provider.py 
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
class BearerTokenAuth(AuthBase):
    """Bearer Token Auth hook - used with OAuth client to connect applications."""

    def __init__(self, token_callable: Callable[[], str]) -> None:
        """Construct an IdTokenAuth provider (hook).

        Args:
            token_callable: A callable that returns the latest id_token.
        """
        if token_callable is None:
            raise InvalidArgumentError("bearer_token_callable")
        self.token_callable = token_callable

    def __call__(self, r: PreparedRequest):
        """Mutate outgoing request (adding authorization header).

        Args:
            r: Outgoing request.
        """
        r.headers["Authorization"] = f"Bearer {self.token_callable()}"
        return r

    def __eq__(self, other):
        """Return True if instances are equal; otherwise False.

        Equality is tested by retrieving the bearer token for each instance and comparing them.
        """
        return self.token_callable() == getattr(other, "token_callable", lambda: None)()

    def __ne__(self, other):
        """Return True if instances are not equal; otherwise False.

        Equality is tested by retrieving the id_token for each instance and comparing them.
        """
        return not self == other

__call__(r)

Mutate outgoing request (adding authorization header).

Parameters:

Name Type Description Default
r PreparedRequest

Outgoing request.

 required
Source code in tenduke_core/auth/auth_provider.py 
28
29
30
31
32
33
34
35
def __call__(self, r: PreparedRequest):
    """Mutate outgoing request (adding authorization header).

    Args:
        r: Outgoing request.
    """
    r.headers["Authorization"] = f"Bearer {self.token_callable()}"
    return r

__eq__(other)

Return True if instances are equal; otherwise False.

Equality is tested by retrieving the bearer token for each instance and comparing them.

Source code in tenduke_core/auth/auth_provider.py 
37
38
39
40
41
42
def __eq__(self, other):
    """Return True if instances are equal; otherwise False.

    Equality is tested by retrieving the bearer token for each instance and comparing them.
    """
    return self.token_callable() == getattr(other, "token_callable", lambda: None)()

__init__(token_callable)

Construct an IdTokenAuth provider (hook).

Parameters:

Name Type Description Default
token_callable Callable[[], str]

A callable that returns the latest id_token.

 required
Source code in tenduke_core/auth/auth_provider.py 
18
19
20
21
22
23
24
25
26
def __init__(self, token_callable: Callable[[], str]) -> None:
    """Construct an IdTokenAuth provider (hook).

    Args:
        token_callable: A callable that returns the latest id_token.
    """
    if token_callable is None:
        raise InvalidArgumentError("bearer_token_callable")
    self.token_callable = token_callable

__ne__(other)

Return True if instances are not equal; otherwise False.

Equality is tested by retrieving the id_token for each instance and comparing them.

Source code in tenduke_core/auth/auth_provider.py 
44
45
46
47
48
49
def __ne__(self, other):
    """Return True if instances are not equal; otherwise False.

    Equality is tested by retrieving the id_token for each instance and comparing them.
    """
    return not self == other

DeviceAuthorizationResponse dataclass

Data from Device Authorization Response.

Callers should use this data to initiate the user interaction phase of the Device Authorization Grant flow, before or simultaneously with starting polling for the token.

See https://www.rfc-editor.org/rfc/rfc8628#section-3.2.

Attributes:

Name Type Description
user_code str

The end-user verification code.
uri str

The end-user verification URI on the authorization server. The URI should be short and easy to remember as end users will be asked to manually type it into their user agent.
uri_complete str | None

A verification URI that includes the "user_code" (or other information with the same function as the "user_code"), which is designed for non-textual transmission.
Source code in tenduke_core/auth/device_auth_response.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@dataclass
class DeviceAuthorizationResponse:
    """
    Data from Device Authorization Response.

    Callers should use this data to initiate the user interaction phase
    of the Device Authorization Grant flow, before or simultaneously with
    starting polling for the token.

    See https://www.rfc-editor.org/rfc/rfc8628#section-3.2.

    Attributes:
        user_code: The end-user verification code.
        uri:
            The end-user verification URI on the authorization server. The URI should be short and
            easy to remember as end users will be asked to manually type it into their user agent.
        uri_complete:
            A verification URI that includes the "user_code" (or other information with the same
            function as the "user_code"), which is designed for non-textual transmission.
    """

    user_code: str
    uri: str
    uri_complete: str | None

DeviceFlowClient

Bases: OAuthClient

Helper class to perform OAuth device flow.

Source code in tenduke_core/auth/device_flow_client.py 
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
class DeviceFlowClient(OAuthClient):
    """Helper class to perform OAuth device flow."""

    class VerificationUri:
        """Verification URI/URL handling.

        As some implementations of device flow use URI and some use URL in the parameter name,
        we need to check for both.
        The logic in making the requests is that we should use the '_complete' variant of the URI
        if it is present/populated and use the incomplete version if it is not.
        """

        def __init__(self):
            """Construct a VerificationUri instance."""
            self._verification_uri_complete = ""
            self._verification_url_complete = ""
            self._verification_uri = ""
            self._verification_url = ""

        def read_from_dict(self, data: dict[str, str]):
            """Read the verification uri details from a dictionary.

            RFC8628 specifies verification_uri. At least one identity provider in the
            wild sends verification_url.
            Outside of this class, we don't need to worry about verification_url, this
            class will read whichever is present.

            Args:
                data: The contents of the Device Authorization Response.
            """
            self._verification_uri = data.get("verification_uri", "")
            self._verification_url = data.get("verification_url", "")
            self._verification_uri_complete = data.get("verification_uri_complete", "")
            self._verification_url_complete = data.get("verification_url_complete", "")

        def uri(self) -> str:
            """Return the token poll uri."""
            return self._verification_uri or self._verification_url

        def uri_complete(self) -> str:
            """Return the token poll complete uri."""
            return (
                self._verification_uri_complete
                or self._verification_url_complete
                or self._verification_uri
                or self._verification_url
            )

    def __init__(self, config: TendukeConfig, session_factory: SessionFactory):
        """Construct an instance of the DeviceFlowClient.

        Args:
            config:
                Configuration parameters for interacting with the OAuth / Open ID Authorization
                Server.
            session_factory:
                Used to create requests Session configured with the settings from config and with
                the configured User-Agent header value.
        """
        self._device_code = None
        self._expires_at = None
        self._user_code: str = ""
        self._interval = 5
        self._verification_uri = self.VerificationUri()
        super().__init__(config, session_factory)

    def authorize(self) -> DeviceAuthorizationResponse:
        """Make the authorization request and return the details to display to the user.

        Returns:
            A DeviceAuthorizationResponse object containing the code and uri needed to fetch a
            token.

        Raises:
            ValueError: Required configuration item idp_oauth_device_code_url was missing.
            OAuth2Error: Device Authorization Request was unsuccessful.
        """
        params = _device_authorization_request_parameters(
            self.config.idp_oauth_client_id,
            self.config.idp_oauth_client_secret,
            self.config.idp_oauth_scope,
        )
        url = self.config.idp_oauth_device_code_url
        if not url:
            raise DeviceCodeAuthorizationUrlMissingError()

        response = self.session.post(url, params)
        if not response.ok:
            raise_error_from_error_response(response)
        response_json = response.json()
        self._parse_authorize_response(response_json)
        return DeviceAuthorizationResponse(
            self._user_code,
            self._verification_uri.uri(),
            self._verification_uri.uri_complete(),
        )

    def _parse_authorize_response(self, data):
        self._device_code = data.get("device_code")
        if "expires_in" in data:
            seconds = data["expires_in"]
            self._expires_at = datetime.now(timezone.utc) + timedelta(seconds=seconds)
        self._user_code = data.get("user_code", "")
        self._interval = data.get("interval") or 5
        self._verification_uri.read_from_dict(data)

    async def poll_for_token_async(self):
        """Poll for an OAuth device flow token.

        The method waits for the specified interval between attempts.
        This method is asynchronous (non-blocking).

        Raises:
            OAuth2Error: Device Access Token Request was unsuccessful.
        """
        poll = True
        while poll:
            response = self._make_poll_request()
            if response.ok:
                poll = False
                token = TokenResponse.from_api(response.json())
                self.token = token
                return token

            poll = await self._handle_error_token_response_async(response)

    def poll_for_token(self):
        """Poll for an OAuth device flow token.

        The method waits for the specified interval between attempts.
        This method is synchronous (blocking).

        Raises:
            OAuth2Error: Device Access Token Request was unsuccessful.
        """
        return asyncio.run(self.poll_for_token_async())

    def _make_poll_request(self):
        params = _poll_params(
            self.config.idp_oauth_client_id,
            self._device_code,
            self.config.idp_oauth_client_secret,
        )
        url = self.config.idp_oauth_token_url
        if not url:
            raise TokenUrlMissingError()
        return self.session.post(url, data=params)

    async def _handle_error_token_response_async(self, response):
        json = response.json()
        error = json.get("error")
        error_code = json.get("error_code")

        continue_codes = ["authorization_pending", "slow_down"]
        if error in continue_codes or error_code in continue_codes:
            if "slow_down" in (error, error_code):
                self._interval += 5
            await asyncio.sleep(self._interval)
            # signal loop in caller to keep polling
            return True

        return raise_error_from_error_response(response)

VerificationUri

Verification URI/URL handling.

As some implementations of device flow use URI and some use URL in the parameter name, we need to check for both. The logic in making the requests is that we should use the '_complete' variant of the URI if it is present/populated and use the incomplete version if it is not.

Source code in tenduke_core/auth/device_flow_client.py 
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
class VerificationUri:
    """Verification URI/URL handling.

    As some implementations of device flow use URI and some use URL in the parameter name,
    we need to check for both.
    The logic in making the requests is that we should use the '_complete' variant of the URI
    if it is present/populated and use the incomplete version if it is not.
    """

    def __init__(self):
        """Construct a VerificationUri instance."""
        self._verification_uri_complete = ""
        self._verification_url_complete = ""
        self._verification_uri = ""
        self._verification_url = ""

    def read_from_dict(self, data: dict[str, str]):
        """Read the verification uri details from a dictionary.

        RFC8628 specifies verification_uri. At least one identity provider in the
        wild sends verification_url.
        Outside of this class, we don't need to worry about verification_url, this
        class will read whichever is present.

        Args:
            data: The contents of the Device Authorization Response.
        """
        self._verification_uri = data.get("verification_uri", "")
        self._verification_url = data.get("verification_url", "")
        self._verification_uri_complete = data.get("verification_uri_complete", "")
        self._verification_url_complete = data.get("verification_url_complete", "")

    def uri(self) -> str:
        """Return the token poll uri."""
        return self._verification_uri or self._verification_url

    def uri_complete(self) -> str:
        """Return the token poll complete uri."""
        return (
            self._verification_uri_complete
            or self._verification_url_complete
            or self._verification_uri
            or self._verification_url
        )

__init__()

Construct a VerificationUri instance.

Source code in tenduke_core/auth/device_flow_client.py 
54
55
56
57
58
59
def __init__(self):
    """Construct a VerificationUri instance."""
    self._verification_uri_complete = ""
    self._verification_url_complete = ""
    self._verification_uri = ""
    self._verification_url = ""

read_from_dict(data)

Read the verification uri details from a dictionary.

RFC8628 specifies verification_uri. At least one identity provider in the wild sends verification_url. Outside of this class, we don't need to worry about verification_url, this class will read whichever is present.

Parameters:

Name Type Description Default
data dict[str, str]

The contents of the Device Authorization Response.

 required
Source code in tenduke_core/auth/device_flow_client.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def read_from_dict(self, data: dict[str, str]):
    """Read the verification uri details from a dictionary.

    RFC8628 specifies verification_uri. At least one identity provider in the
    wild sends verification_url.
    Outside of this class, we don't need to worry about verification_url, this
    class will read whichever is present.

    Args:
        data: The contents of the Device Authorization Response.
    """
    self._verification_uri = data.get("verification_uri", "")
    self._verification_url = data.get("verification_url", "")
    self._verification_uri_complete = data.get("verification_uri_complete", "")
    self._verification_url_complete = data.get("verification_url_complete", "")

uri()

Return the token poll uri.

Source code in tenduke_core/auth/device_flow_client.py 
77
78
79
def uri(self) -> str:
    """Return the token poll uri."""
    return self._verification_uri or self._verification_url

uri_complete()

Return the token poll complete uri.

Source code in tenduke_core/auth/device_flow_client.py 
81
82
83
84
85
86
87
88
def uri_complete(self) -> str:
    """Return the token poll complete uri."""
    return (
        self._verification_uri_complete
        or self._verification_url_complete
        or self._verification_uri
        or self._verification_url
    )

__init__(config, session_factory)

Construct an instance of the DeviceFlowClient.

Parameters:

Name Type Description Default
config TendukeConfig

Configuration parameters for interacting with the OAuth / Open ID Authorization Server.

 required
session_factory SessionFactory

Used to create requests Session configured with the settings from config and with the configured User-Agent header value.

 required
Source code in tenduke_core/auth/device_flow_client.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def __init__(self, config: TendukeConfig, session_factory: SessionFactory):
    """Construct an instance of the DeviceFlowClient.

    Args:
        config:
            Configuration parameters for interacting with the OAuth / Open ID Authorization
            Server.
        session_factory:
            Used to create requests Session configured with the settings from config and with
            the configured User-Agent header value.
    """
    self._device_code = None
    self._expires_at = None
    self._user_code: str = ""
    self._interval = 5
    self._verification_uri = self.VerificationUri()
    super().__init__(config, session_factory)

authorize()

Make the authorization request and return the details to display to the user.

Returns:

Type Description
DeviceAuthorizationResponse

A DeviceAuthorizationResponse object containing the code and uri needed to fetch a
DeviceAuthorizationResponse

token.

Raises:

Type Description
ValueError

Required configuration item idp_oauth_device_code_url was missing.
OAuth2Error

Device Authorization Request was unsuccessful.
Source code in tenduke_core/auth/device_flow_client.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def authorize(self) -> DeviceAuthorizationResponse:
    """Make the authorization request and return the details to display to the user.

    Returns:
        A DeviceAuthorizationResponse object containing the code and uri needed to fetch a
        token.

    Raises:
        ValueError: Required configuration item idp_oauth_device_code_url was missing.
        OAuth2Error: Device Authorization Request was unsuccessful.
    """
    params = _device_authorization_request_parameters(
        self.config.idp_oauth_client_id,
        self.config.idp_oauth_client_secret,
        self.config.idp_oauth_scope,
    )
    url = self.config.idp_oauth_device_code_url
    if not url:
        raise DeviceCodeAuthorizationUrlMissingError()

    response = self.session.post(url, params)
    if not response.ok:
        raise_error_from_error_response(response)
    response_json = response.json()
    self._parse_authorize_response(response_json)
    return DeviceAuthorizationResponse(
        self._user_code,
        self._verification_uri.uri(),
        self._verification_uri.uri_complete(),
    )

poll_for_token()

Poll for an OAuth device flow token.

The method waits for the specified interval between attempts. This method is synchronous (blocking).

Raises:

Type Description
OAuth2Error

Device Access Token Request was unsuccessful.
Source code in tenduke_core/auth/device_flow_client.py 
168
169
170
171
172
173
174
175
176
177
def poll_for_token(self):
    """Poll for an OAuth device flow token.

    The method waits for the specified interval between attempts.
    This method is synchronous (blocking).

    Raises:
        OAuth2Error: Device Access Token Request was unsuccessful.
    """
    return asyncio.run(self.poll_for_token_async())

poll_for_token_async() async

Poll for an OAuth device flow token.

The method waits for the specified interval between attempts. This method is asynchronous (non-blocking).

Raises:

Type Description
OAuth2Error

Device Access Token Request was unsuccessful.
Source code in tenduke_core/auth/device_flow_client.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
async def poll_for_token_async(self):
    """Poll for an OAuth device flow token.

    The method waits for the specified interval between attempts.
    This method is asynchronous (non-blocking).

    Raises:
        OAuth2Error: Device Access Token Request was unsuccessful.
    """
    poll = True
    while poll:
        response = self._make_poll_request()
        if response.ok:
            poll = False
            token = TokenResponse.from_api(response.json())
            self.token = token
            return token

        poll = await self._handle_error_token_response_async(response)

IdTokenAuth

Bases: AuthBase

ID Token Auth hook - used with Open ID Connect in public applications.

Source code in tenduke_core/auth/auth_provider.py 
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
class IdTokenAuth(AuthBase):
    """ID Token Auth hook - used with Open ID Connect in public applications."""

    def __init__(self, id_token_callable: Callable[[], str]) -> None:
        """Construct an IdTokenAuth provider (hook).

        Args:
            id_token_callable: A callable that returns the latest id_token.
        """
        if id_token_callable is None:
            raise InvalidArgumentError("id_token_callable")
        self.id_token_callable = id_token_callable

    def __call__(self, r: PreparedRequest):
        """Mutate outgoing request (adding authorization header).

        Args:
            r: Outgoing request.
        """
        r.headers["Authorization"] = f"IdToken {self.id_token_callable()}"
        return r

    def __eq__(self, other):
        """Return True if instances are equal; otherwise False.

        Equality is tested by retrieving the id_token for each instance and comparing them.
        """
        return self.id_token_callable() == getattr(other, "id_token_callable", lambda: None)()

    def __ne__(self, other):
        """Return True if instances are not equal; otherwise False.

        Equality is tested by retrieving the id_token for each instance and comparing them.
        """
        return not self == other

__call__(r)

Mutate outgoing request (adding authorization header).

Parameters:

Name Type Description Default
r PreparedRequest

Outgoing request.

 required
Source code in tenduke_core/auth/auth_provider.py 
65
66
67
68
69
70
71
72
def __call__(self, r: PreparedRequest):
    """Mutate outgoing request (adding authorization header).

    Args:
        r: Outgoing request.
    """
    r.headers["Authorization"] = f"IdToken {self.id_token_callable()}"
    return r

__eq__(other)

Return True if instances are equal; otherwise False.

Equality is tested by retrieving the id_token for each instance and comparing them.

Source code in tenduke_core/auth/auth_provider.py 
74
75
76
77
78
79
def __eq__(self, other):
    """Return True if instances are equal; otherwise False.

    Equality is tested by retrieving the id_token for each instance and comparing them.
    """
    return self.id_token_callable() == getattr(other, "id_token_callable", lambda: None)()

__init__(id_token_callable)

Construct an IdTokenAuth provider (hook).

Parameters:

Name Type Description Default
id_token_callable Callable[[], str]

A callable that returns the latest id_token.

 required
Source code in tenduke_core/auth/auth_provider.py 
55
56
57
58
59
60
61
62
63
def __init__(self, id_token_callable: Callable[[], str]) -> None:
    """Construct an IdTokenAuth provider (hook).

    Args:
        id_token_callable: A callable that returns the latest id_token.
    """
    if id_token_callable is None:
        raise InvalidArgumentError("id_token_callable")
    self.id_token_callable = id_token_callable

__ne__(other)

Return True if instances are not equal; otherwise False.

Equality is tested by retrieving the id_token for each instance and comparing them.

Source code in tenduke_core/auth/auth_provider.py 
81
82
83
84
85
86
def __ne__(self, other):
    """Return True if instances are not equal; otherwise False.

    Equality is tested by retrieving the id_token for each instance and comparing them.
    """
    return not self == other

OAuthClient

OAuth / Open ID Connect client.

Source code in tenduke_core/auth/oauth_client.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
class OAuthClient:
    """OAuth / Open ID Connect client."""

    def __init__(self, config: TendukeConfig, session_factory: SessionFactory):
        """
        Construct an instance of the OAuth Client.

        Args:
            config:
                Configuration parameters for interacting with the OAuth / Open ID Authorization
                Server.
            session_factory:
                Used to create requests Session configured with the settings from config and with
                the configured User-Agent header value.
        """
        self.config = config
        self._idp_jwks_client = None
        if config.idp_jwks_uri:
            self._idp_jwks_client = PyJWKClient(
                config.idp_jwks_uri, cache_keys=True, lifespan=345000
            )
        self._try_to_refresh = True
        self._token: TokenResponse | None = None
        self._id_token: IdToken | None = None
        self.session = session_factory.create()

    @property
    def token(self) -> TokenResponse | None:
        """The current token response."""
        return self._token

    @token.setter
    def token(self, value: TokenResponse):
        if value.id_token is not None:
            self._id_token = IdToken.from_token(
                value, self._idp_jwks_client, self.config.idp_oauth_client_id
            )
        self._try_to_refresh = value.refresh_token is not None and value.id_token is not None
        self._token = value

    @property
    def id_token(self) -> str | None:
        """The current identity token (if any)."""
        if not self._token:
            return None
        self._refresh_token()
        return self._id_token.token if self._id_token else None

    def get_user_info(self) -> UserInfo:
        """Retrieve the details of the user from the userinfo endpoint.

        Returns:
            The information for the authenticated user.

        Raises:
            ValueError: User info URL missing from configuration.
            OAuth2Error: The user info request was unsuccessful.
        """
        headers = {"Authorization": f"Bearer {self._token.access_token}"} if self._token else {}
        if not self.config.idp_userinfo_url:
            raise UserInfoUrlMissingError()

        response = self.session.get(self.config.idp_userinfo_url, headers=headers)
        if not response.ok:
            raise_error_from_error_response(response)

        info = UserInfo.from_api(response.json())
        return info

    def auth(self) -> IdTokenAuth:
        """Get an authorization implementation based on the current identity.

        Returns:
            An authorization provider hook that sets the authorization header of HTTP requests
            using the current id_token.
        """
        if not self.id_token:
            raise IdTokenMissingError()

        return IdTokenAuth(lambda: self.id_token or "")

    def _can_refresh(self):
        return (
            self._try_to_refresh
            and self._token is not None
            and self._token.refresh_token is not None
            and self.config.idp_oauth_token_url is not None
        )

    def _refresh_token(self):
        # if we know the expiry of the id_token, check it is still valid
        # if it is close to expiry and with have a token url, see if we can get a new one
        is_expired = _is_expired(self._id_token, self.config.token_refresh_leeway_seconds)
        can_refesh = self._can_refresh()

        # type checking disabled for members already checked for None in
        #  _is_expired and _can_refresh
        if is_expired and can_refesh:
            data = {
                "grant_type": "refresh_token",
                "refresh_token": self._token.refresh_token,  # pyright: ignore[reportOptionalMemberAccess]
                "client_id": self.config.idp_oauth_client_id,
            }
            if self.config.idp_oauth_client_secret:
                data["client_secret"] = self.config.idp_oauth_client_secret
            response = self.session.post(self.config.idp_oauth_token_url, data=data)  # pyright: ignore[reportArgumentType]
            if response.ok:
                token = TokenResponse.from_api(response.json())
                self.token = token

id_token property

The current identity token (if any).

token property writable

The current token response.

__init__(config, session_factory)

Construct an instance of the OAuth Client.

Parameters:

Name Type Description Default
config TendukeConfig

Configuration parameters for interacting with the OAuth / Open ID Authorization Server.

 required
session_factory SessionFactory

Used to create requests Session configured with the settings from config and with the configured User-Agent header value.

 required
Source code in tenduke_core/auth/oauth_client.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(self, config: TendukeConfig, session_factory: SessionFactory):
    """
    Construct an instance of the OAuth Client.

    Args:
        config:
            Configuration parameters for interacting with the OAuth / Open ID Authorization
            Server.
        session_factory:
            Used to create requests Session configured with the settings from config and with
            the configured User-Agent header value.
    """
    self.config = config
    self._idp_jwks_client = None
    if config.idp_jwks_uri:
        self._idp_jwks_client = PyJWKClient(
            config.idp_jwks_uri, cache_keys=True, lifespan=345000
        )
    self._try_to_refresh = True
    self._token: TokenResponse | None = None
    self._id_token: IdToken | None = None
    self.session = session_factory.create()

auth()

Get an authorization implementation based on the current identity.

Returns:

Type Description
IdTokenAuth

An authorization provider hook that sets the authorization header of HTTP requests
IdTokenAuth

using the current id_token.
Source code in tenduke_core/auth/oauth_client.py 
152
153
154
155
156
157
158
159
160
161
162
def auth(self) -> IdTokenAuth:
    """Get an authorization implementation based on the current identity.

    Returns:
        An authorization provider hook that sets the authorization header of HTTP requests
        using the current id_token.
    """
    if not self.id_token:
        raise IdTokenMissingError()

    return IdTokenAuth(lambda: self.id_token or "")

get_user_info()

Retrieve the details of the user from the userinfo endpoint.

Returns:

Type Description
UserInfo

The information for the authenticated user.

Raises:

Type Description
ValueError

User info URL missing from configuration.
OAuth2Error

The user info request was unsuccessful.
Source code in tenduke_core/auth/oauth_client.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def get_user_info(self) -> UserInfo:
    """Retrieve the details of the user from the userinfo endpoint.

    Returns:
        The information for the authenticated user.

    Raises:
        ValueError: User info URL missing from configuration.
        OAuth2Error: The user info request was unsuccessful.
    """
    headers = {"Authorization": f"Bearer {self._token.access_token}"} if self._token else {}
    if not self.config.idp_userinfo_url:
        raise UserInfoUrlMissingError()

    response = self.session.get(self.config.idp_userinfo_url, headers=headers)
    if not response.ok:
        raise_error_from_error_response(response)

    info = UserInfo.from_api(response.json())
    return info

PkceFlowClient

Bases: OAuthClient

Client for Proof Key for Code Exchange (PKCE) flow.

Source code in tenduke_core/auth/pkce_flow_client.py 
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class PkceFlowClient(OAuthClient):
    """Client for Proof Key for Code Exchange (PKCE) flow."""

    class RedirectHttpServer(HTTPServer):
        """HTTP Server to handle redirect with code."""

        def __init__(
            self,
            *args,
            success_http: str | None = AUTH_SUCCESS_MESSAGE,
            **kwargs,
        ):
            """Construct an instance of the HTTP Server.

            Args:
                success_http: HTTP message to send once redirect request is received.
            """
            self.success_http = success_http
            self.authorization_response = ""
            super().__init__(*args, **kwargs)

    class RedirectHandler(BaseHTTPRequestHandler):
        """Handler class for redirect with code."""

        def do_GET(self):  # pylint: disable=invalid-name
            """Handle OAuth redirect."""
            self.wfile.write(self.server.success_http.encode("UTF-8"))  # pyright:ignore[reportAttributeAccessIssue]
            self.wfile.flush()
            self.server.authorization_response = self.path  # pyright:ignore[reportAttributeAccessIssue]

    def __init__(self, config: TendukeConfig, session_factory: SessionFactory, *args, **kwargs):
        """Construct an instance of the PkceFlowClient.

        Args:
            config:
                Configuration parameters for interacting with the OAuth / Open ID Authorization
                Server.
            session_factory:
                Used to create requests Session configured with the settings from config and with
                the configured User-Agent header value.
        """
        self.code_verifier = kwargs.pop("code_verifier", None)
        self.state = kwargs.pop("state", None)
        self._client = None
        super().__init__(config, session_factory)

    def _build_redirect_uri(self, port: int | None = None):
        port = port or self.config.auth_redirect_port
        redirect_uri = self.config.auth_redirect_uri
        parsed_result = urlparse(redirect_uri)
        if parsed_result.hostname in (
            "localhost",
            "127.0.0.1",
            "::1",
        ) and _is_default_port(parsed_result):
            # ParsedResult.host_name returns the hostname only, we need to escape as per RFC2732
            host_name = "[::1]" if parsed_result.hostname == "::1" else parsed_result.hostname
            # urllib.parse is more flexible with types than mypy gives it credit for
            new_uri = f"{host_name}:{port}"  # type: ignore[str-bytes-safe]
            updated = parsed_result._replace(netloc=new_uri)  # type: ignore[arg-type]
            redirect_uri = urlunparse(updated)  # type: ignore[assignment]
        if not redirect_uri:
            port = port or self.config.auth_redirect_port
            redirect_uri = urljoin(f"http://localhost:{port}", self.config.auth_redirect_path)
        return redirect_uri

    def _resolve_port(self):
        redirect_uri = self.config.auth_redirect_uri
        parsed_result = urlparse(redirect_uri)
        if parsed_result.port is not None and parsed_result.port not in (0, 80, 443):
            return parsed_result.port
        return self.config.auth_redirect_port

    def create_authorization_url(self, port: int | None = None) -> str:
        """Generate and return authorization url.

        Args:
            port: The port number to use in the redirect URI

        Returns:
            URL for authorization request.
        """
        self.code_verifier = generate_token(128)

        redirect_uri = self._build_redirect_uri(port)

        self._client = OAuth2Session(
            client_id=self.config.idp_oauth_client_id,
            scope=self.config.idp_oauth_scope,
            redirect_uri=redirect_uri,
            code_challenge_method="S256",
        )
        url, state = self._client.create_authorization_url(  # type: ignore[attr-defined]
            self.config.idp_oauth_authorization_url, code_verifier=self.code_verifier
        )
        self.state = state
        return url

    def fetch_token(
        self, authorization_response: str | None, port: int | None = None
    ) -> TokenResponse:
        """Fetch token based on authorization response.

        Args:
            authorization_response:
                Redirect URL request with authorization code to exchange for token.

        Returns:
            TokenResponse object for the authorization code in the authorization_reponse parameter.
        """
        redirect_uri = self._build_redirect_uri(port)

        if self._client is None:
            self._client = OAuth2Session(
                client_id=self.config.idp_oauth_client_id,
                scope=self.config.idp_oauth_scope,
                redirect_uri=redirect_uri,
                code_challenge_method="S256",
            )
        token = self._client.fetch_token(  # type: ignore[attr-defined]
            self.config.idp_oauth_token_url,
            authorization_response=authorization_response,
            state=self.state,
            code_verifier=self.code_verifier,
        )
        token = TokenResponse.from_api(token)
        self.token = token
        return token

    def login_with_default_browser(self) -> TokenResponse:
        """Launch system default browser for user to log in.

        Returns:
            TokenResponse object with token for the authenticated user.

        Raises:
            OAuth2Error: Authentication failed.
        """
        authorization_response = None
        success_http = AUTH_SUCCESS_MESSAGE

        if self.config.auth_success_message:
            success_http = Path(self.config.auth_success_message).read_text("UTF-8")

        port = self._resolve_port()

        with self.RedirectHttpServer(
            ("", port),
            self.RedirectHandler,
            success_http=success_http,
        ) as httpd:
            url = self.create_authorization_url(httpd.server_port)
            webbrowser.open(url)
            httpd.handle_request()
            authorization_response = httpd.authorization_response

        if authorization_response:
            return self.fetch_token(authorization_response, httpd.server_port)

        raise AuthenticationFailed()

RedirectHandler

Bases: BaseHTTPRequestHandler

Handler class for redirect with code.

Source code in tenduke_core/auth/pkce_flow_client.py 
66
67
68
69
70
71
72
73
class RedirectHandler(BaseHTTPRequestHandler):
    """Handler class for redirect with code."""

    def do_GET(self):  # pylint: disable=invalid-name
        """Handle OAuth redirect."""
        self.wfile.write(self.server.success_http.encode("UTF-8"))  # pyright:ignore[reportAttributeAccessIssue]
        self.wfile.flush()
        self.server.authorization_response = self.path  # pyright:ignore[reportAttributeAccessIssue]

do_GET()

Handle OAuth redirect.

Source code in tenduke_core/auth/pkce_flow_client.py 
69
70
71
72
73
def do_GET(self):  # pylint: disable=invalid-name
    """Handle OAuth redirect."""
    self.wfile.write(self.server.success_http.encode("UTF-8"))  # pyright:ignore[reportAttributeAccessIssue]
    self.wfile.flush()
    self.server.authorization_response = self.path  # pyright:ignore[reportAttributeAccessIssue]

RedirectHttpServer

Bases: HTTPServer

HTTP Server to handle redirect with code.

Source code in tenduke_core/auth/pkce_flow_client.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class RedirectHttpServer(HTTPServer):
    """HTTP Server to handle redirect with code."""

    def __init__(
        self,
        *args,
        success_http: str | None = AUTH_SUCCESS_MESSAGE,
        **kwargs,
    ):
        """Construct an instance of the HTTP Server.

        Args:
            success_http: HTTP message to send once redirect request is received.
        """
        self.success_http = success_http
        self.authorization_response = ""
        super().__init__(*args, **kwargs)

__init__(*args, success_http=AUTH_SUCCESS_MESSAGE, **kwargs)

Construct an instance of the HTTP Server.

Parameters:

Name Type Description Default
success_http str | None

HTTP message to send once redirect request is received.

 AUTH_SUCCESS_MESSAGE
Source code in tenduke_core/auth/pkce_flow_client.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    *args,
    success_http: str | None = AUTH_SUCCESS_MESSAGE,
    **kwargs,
):
    """Construct an instance of the HTTP Server.

    Args:
        success_http: HTTP message to send once redirect request is received.
    """
    self.success_http = success_http
    self.authorization_response = ""
    super().__init__(*args, **kwargs)

__init__(config, session_factory, *args, **kwargs)

Construct an instance of the PkceFlowClient.

Parameters:

Name Type Description Default
config TendukeConfig

Configuration parameters for interacting with the OAuth / Open ID Authorization Server.

 required
session_factory SessionFactory

Used to create requests Session configured with the settings from config and with the configured User-Agent header value.

 required
Source code in tenduke_core/auth/pkce_flow_client.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __init__(self, config: TendukeConfig, session_factory: SessionFactory, *args, **kwargs):
    """Construct an instance of the PkceFlowClient.

    Args:
        config:
            Configuration parameters for interacting with the OAuth / Open ID Authorization
            Server.
        session_factory:
            Used to create requests Session configured with the settings from config and with
            the configured User-Agent header value.
    """
    self.code_verifier = kwargs.pop("code_verifier", None)
    self.state = kwargs.pop("state", None)
    self._client = None
    super().__init__(config, session_factory)

create_authorization_url(port=None)

Generate and return authorization url.

Parameters:

Name Type Description Default
port int | None

The port number to use in the redirect URI

 None

Returns:

Type Description
str

URL for authorization request.
Source code in tenduke_core/auth/pkce_flow_client.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def create_authorization_url(self, port: int | None = None) -> str:
    """Generate and return authorization url.

    Args:
        port: The port number to use in the redirect URI

    Returns:
        URL for authorization request.
    """
    self.code_verifier = generate_token(128)

    redirect_uri = self._build_redirect_uri(port)

    self._client = OAuth2Session(
        client_id=self.config.idp_oauth_client_id,
        scope=self.config.idp_oauth_scope,
        redirect_uri=redirect_uri,
        code_challenge_method="S256",
    )
    url, state = self._client.create_authorization_url(  # type: ignore[attr-defined]
        self.config.idp_oauth_authorization_url, code_verifier=self.code_verifier
    )
    self.state = state
    return url

fetch_token(authorization_response, port=None)

Fetch token based on authorization response.

Parameters:

Name Type Description Default
authorization_response str | None

Redirect URL request with authorization code to exchange for token.

 required

Returns:

Type Description
TokenResponse

TokenResponse object for the authorization code in the authorization_reponse parameter.
Source code in tenduke_core/auth/pkce_flow_client.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def fetch_token(
    self, authorization_response: str | None, port: int | None = None
) -> TokenResponse:
    """Fetch token based on authorization response.

    Args:
        authorization_response:
            Redirect URL request with authorization code to exchange for token.

    Returns:
        TokenResponse object for the authorization code in the authorization_reponse parameter.
    """
    redirect_uri = self._build_redirect_uri(port)

    if self._client is None:
        self._client = OAuth2Session(
            client_id=self.config.idp_oauth_client_id,
            scope=self.config.idp_oauth_scope,
            redirect_uri=redirect_uri,
            code_challenge_method="S256",
        )
    token = self._client.fetch_token(  # type: ignore[attr-defined]
        self.config.idp_oauth_token_url,
        authorization_response=authorization_response,
        state=self.state,
        code_verifier=self.code_verifier,
    )
    token = TokenResponse.from_api(token)
    self.token = token
    return token

login_with_default_browser()

Launch system default browser for user to log in.

Returns:

Type Description
TokenResponse

TokenResponse object with token for the authenticated user.

Raises:

Type Description
OAuth2Error

Authentication failed.
Source code in tenduke_core/auth/pkce_flow_client.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def login_with_default_browser(self) -> TokenResponse:
    """Launch system default browser for user to log in.

    Returns:
        TokenResponse object with token for the authenticated user.

    Raises:
        OAuth2Error: Authentication failed.
    """
    authorization_response = None
    success_http = AUTH_SUCCESS_MESSAGE

    if self.config.auth_success_message:
        success_http = Path(self.config.auth_success_message).read_text("UTF-8")

    port = self._resolve_port()

    with self.RedirectHttpServer(
        ("", port),
        self.RedirectHandler,
        success_http=success_http,
    ) as httpd:
        url = self.create_authorization_url(httpd.server_port)
        webbrowser.open(url)
        httpd.handle_request()
        authorization_response = httpd.authorization_response

    if authorization_response:
        return self.fetch_token(authorization_response, httpd.server_port)

    raise AuthenticationFailed()

TokenResponse dataclass

Bases: Model

Token response data model.

https://www.rfc-editor.org/rfc/rfc6749#section-4.2.2

Attributes:

Name Type Description
access_token str

The access token issued by the authorization server.
token_type str

The type of the token issued.
expires_in int

The lifetime in seconds of the access token.
expires_at datetime

The datetime when the token will expire Derived from expires_in and the current system time when the token was received.
refresh_token str | None

The refresh token, which can be used to obtain new access tokens.
id_token str | None

ID Token value associated with the authenticated session.
Source code in tenduke_core/auth/token_response.py 
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
@dataclass
class TokenResponse(Model):
    """Token response data model.

    https://www.rfc-editor.org/rfc/rfc6749#section-4.2.2

    Attributes:
        access_token: The access token issued by the authorization server.
        token_type: The type of the token issued.
        expires_in: The lifetime in seconds of the access token.
        expires_at:
            The datetime when the token will expire Derived from expires_in and the current system
            time when the token was received.
        refresh_token: The refresh token, which can be used to obtain new access tokens.
        id_token: ID Token value associated with the authenticated session.
    """

    access_token: str
    token_type: str
    expires_in: int
    expires_at: datetime = field(init=False)
    refresh_token: str | None = None
    # id_token is not mandatory in the RFC 6749 Access Token Response
    # but Open ID Connect id_token is required for ID Token auth
    id_token: str | None = None

    def __post_init__(self):
        """Initialize expires_at based on expires_in and current time."""
        self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=self.expires_in)

__post_init__()

Initialize expires_at based on expires_in and current time.

Source code in tenduke_core/auth/token_response.py 
39
40
41
def __post_init__(self):
    """Initialize expires_at based on expires_in and current time."""
    self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=self.expires_in)

UserInfo dataclass

Bases: Model

User info response data model.

Attributes:

Name Type Description
sub str

Subject identitfier. A locally unique and never reassigned identifier within the issuer for the End-User.
name str | None

End-User's full name in displayable form including all name parts, possibly including titles and suffixes, ordered according to the End-User's locale and preferences.
given_name str | None

Given name(s) or first name(s) of the End-User. Note that in some cultures, people can have multiple given names; all can be present, with the names being separated by space characters.
family_name str | None

Surname(s) or last name(s) of the End-User. Note that in some cultures, people can have multiple family names or no family name; all can be present, with the names being separated by space characters.
email str | None

End-User's preferred e-mail address.
formatted_name str | None

Combination of names for display purposes. Shall contain a value if any identifier is populated.
Source code in tenduke_core/auth/user_info.py 
 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
@dataclass
class UserInfo(Model):
    """User info response data model.

    Attributes:
        sub:
            Subject identitfier. A locally unique and never reassigned identifier within the issuer
            for the End-User.
        name:
            End-User's full name in displayable form including all name parts, possibly including
            titles and suffixes, ordered according to the End-User's locale and preferences.
        given_name:
            Given name(s) or first name(s) of the End-User. Note that in some cultures, people can
            have multiple given names; all can be present, with the names being separated by space
            characters.
        family_name:
            Surname(s) or last name(s) of the End-User. Note that in some cultures, people can have
            multiple family names or no family name; all can be present, with the names being
            separated by space characters.
        email: End-User's preferred e-mail address.
        formatted_name:
            Combination of names for display purposes. Shall contain a value if any identifier is
            populated.
    """

    sub: str
    name: str | None = None
    given_name: str | None = None
    family_name: str | None = None
    email: str | None = None
    formatted_name: str | None = field(init=False)

    def __post_init__(self):
        """Initialize formatted name based on other values."""
        if self.given_name and self.family_name:
            self.formatted_name = f"{self.given_name} {self.family_name}"
        else:
            self.formatted_name = self.name or self.email or self.sub

__post_init__()

Initialize formatted name based on other values.

Source code in tenduke_core/auth/user_info.py 
40
41
42
43
44
45
def __post_init__(self):
    """Initialize formatted name based on other values."""
    if self.given_name and self.family_name:
        self.formatted_name = f"{self.given_name} {self.family_name}"
    else:
        self.formatted_name = self.name or self.email or self.sub