|
1 | 1 | from abc import ABC, abstractmethod |
| 2 | +from datetime import datetime, timezone |
| 3 | +from typing import Dict |
| 4 | + |
| 5 | +from zitadel_client.auth.open_id import OpenId |
2 | 6 |
|
3 | 7 |
|
4 | 8 | class Authenticator(ABC): |
5 | 9 | """ |
6 | | - Abstract base class for authentication strategies. |
7 | | -
|
8 | | - This class defines a common interface for implementing various |
9 | | - authentication mechanisms. Subclasses must implement the |
10 | | - `get_auth_headers` method to provide the necessary HTTP headers for |
11 | | - authenticated API requests, as well as the `refresh_token` method for |
12 | | - refreshing authentication tokens when needed. |
| 10 | + Abstract base class for authenticators. |
13 | 11 |
|
14 | | - Attributes: |
15 | | - host (str): The base URL for authentication endpoints. |
| 12 | + This class defines the basic structure for any authenticator by requiring the implementation |
| 13 | + of a method to retrieve authentication headers, and provides a way to store and retrieve the host. |
16 | 14 | """ |
17 | 15 |
|
18 | 16 | def __init__(self, host: str): |
19 | 17 | """ |
20 | | - Initialize the Authenticator with a host URL. |
| 18 | + Initializes the Authenticator with the specified host. |
21 | 19 |
|
22 | | - Args: |
23 | | - host (str): The base URL for all authentication endpoints. |
| 20 | + :param host: The base URL or endpoint for the service. |
24 | 21 | """ |
25 | 22 | self.host = host |
26 | 23 |
|
27 | 24 | @abstractmethod |
28 | | - def get_auth_headers(self) -> dict[str, str]: |
| 25 | + def get_auth_headers(self) -> Dict[str, str]: |
29 | 26 | """ |
30 | | - Generate and return the authentication headers required for API requests. |
| 27 | + Retrieves the authentication headers to be sent with requests. |
31 | 28 |
|
32 | | - Subclasses must implement this method to return a dictionary where the keys |
33 | | - are the header names and the values are the corresponding header values. |
| 29 | + Subclasses must override this method to return the appropriate headers. |
34 | 30 |
|
35 | | - Returns: |
36 | | - dict[str, str]: A dictionary containing the authentication header(s). |
| 31 | + :return: A dictionary mapping header names to their values. |
37 | 32 | """ |
38 | 33 | pass |
39 | 34 |
|
40 | | - @abstractmethod |
41 | | - def refresh_token(self) -> None: |
| 35 | + def get_host(self) -> str: |
| 36 | + """ |
| 37 | + Returns the stored host. |
| 38 | +
|
| 39 | + :return: The host as a string. |
42 | 40 | """ |
43 | | - Refresh the authentication token. |
| 41 | + return self.host |
| 42 | + |
| 43 | + |
| 44 | +class Token: |
| 45 | + def __init__(self, access_token: str, expires_at: datetime): |
| 46 | + self.access_token = access_token |
| 47 | + self.expires_at = expires_at |
| 48 | + |
| 49 | + def is_expired(self) -> bool: |
| 50 | + return datetime.now(timezone.utc) >= self.expires_at |
44 | 51 |
|
45 | | - This abstract method must be implemented by subclasses that require a mechanism |
46 | | - to refresh the authentication token. The implementation should update the token |
47 | | - used by the authenticator. |
48 | 52 |
|
49 | | - Returns: |
50 | | - None |
| 53 | +class OAuthAuthenticatorBuilder(ABC): |
| 54 | + """ |
| 55 | + Abstract builder class for constructing OAuth authenticator instances. |
| 56 | +
|
| 57 | + This builder provides common configuration options such as the OpenId instance and authentication scopes. |
| 58 | + """ |
| 59 | + |
| 60 | + def __init__(self, host: str): |
51 | 61 | """ |
52 | | - pass |
| 62 | + Initializes the OAuthAuthenticatorBuilder with a given host. |
53 | 63 |
|
54 | | - def get_host(self) -> str: |
| 64 | + :param host: The base URL for the OAuth provider. |
55 | 65 | """ |
56 | | - Retrieve the base host URL. |
| 66 | + self.open_id = OpenId(host) |
| 67 | + self.auth_scopes = "openid urn:zitadel:iam:org:project:id:zitadel:aud" |
57 | 68 |
|
58 | | - This method returns the host URL that was specified during the instantiation |
59 | | - of the Authenticator. It can be used by concrete classes or external consumers |
60 | | - to build full endpoint URLs. |
| 69 | + def scopes(self, auth_scopes: set) -> "OAuthAuthenticatorBuilder": |
| 70 | + """ |
| 71 | + Sets the authentication scopes for the OAuth authenticator. |
61 | 72 |
|
62 | | - Returns: |
63 | | - str: The base URL for authentication endpoints. |
| 73 | + :param auth_scopes: A set of scope strings. |
| 74 | + :return: The builder instance to allow for method chaining. |
64 | 75 | """ |
65 | | - return self.host |
| 76 | + self.auth_scopes = " ".join(auth_scopes) |
| 77 | + return self |
0 commit comments