|
1 | 1 | from __future__ import annotations
|
2 | 2 |
|
3 |
| -import base64 |
4 | 3 | from enum import Enum
|
5 | 4 | from typing import Optional
|
6 | 5 |
|
|
10 | 9 | from botocore.credentials import Credentials, ReadOnlyCredentials
|
11 | 10 |
|
12 | 11 |
|
13 |
| -def _authorization_header(client_id: str, client_secret: str) -> str: |
14 |
| - """ |
15 |
| - Generates the Authorization header for the request |
16 |
| -
|
17 |
| - Args: |
18 |
| - client_id (str): Client ID |
19 |
| - client_secret (str): Client Secret |
20 |
| -
|
21 |
| - Returns: |
22 |
| - str: Base64 encoded Authorization header |
23 |
| - """ |
24 |
| - auth_string = f"{client_id}:{client_secret}" |
25 |
| - encoded_auth_bytes = base64.b64encode(auth_string) |
26 |
| - encoded_auth_string = encoded_auth_bytes.decode("utf-8") |
27 |
| - return f"Basic {encoded_auth_string}" |
28 |
| - |
29 |
| - |
30 | 12 | class AWSServicePrefix(Enum):
|
31 | 13 | """
|
32 | 14 | AWS Service Prefixes - Enumerations of the supported service proxy types
|
@@ -125,52 +107,3 @@ def __init__(
|
125 | 107 |
|
126 | 108 | def __call__(self):
|
127 | 109 | return self.signed_request
|
128 |
| - |
129 |
| - |
130 |
| -class JWTAuth: |
131 |
| - def __init__( |
132 |
| - self, |
133 |
| - client_id: str, |
134 |
| - client_secret: str, |
135 |
| - auth_endpoint: str, |
136 |
| - provider: Enum = AuthProvider.COGNITO, |
137 |
| - audience: Optional[str] = None, |
138 |
| - scope: Optional[list] = None, |
139 |
| - ): |
140 |
| - self.client_id = client_id |
141 |
| - self.client_secret = client_secret |
142 |
| - self.auth_endpoint = auth_endpoint.removesuffix("/") |
143 |
| - self.headers = {"Content-Type": "application/x-www-form-urlencoded"} |
144 |
| - self.provider = provider |
145 |
| - self.audience = audience |
146 |
| - self.scope = scope |
147 |
| - |
148 |
| - self.body = { |
149 |
| - "client_id": self.client_id, |
150 |
| - "client_secret": self.client_secret, |
151 |
| - } |
152 |
| - |
153 |
| - if self.provider == AuthProvider.COGNITO.value: |
154 |
| - encoded_auth_string = _authorization_header(self.client_id, self.client_secret) |
155 |
| - self.headers["Authorization"] = f"Basic {encoded_auth_string}" |
156 |
| - self.body["grant_type"] = "client_credentials" |
157 |
| - if self.scope: |
158 |
| - self.body["scope"] = " ".join(self.scope) |
159 |
| - |
160 |
| - if self.provider == AuthProvider.AUTH0.value: |
161 |
| - self.body["client_id"] = self.client_id |
162 |
| - self.body["client_secret"] = self.client_secret |
163 |
| - self.body["grant_type"] = "client_credentials" |
164 |
| - self.body["audience"] = self.audience |
165 |
| - |
166 |
| - if self.provider == AuthProvider.OKTA.value: |
167 |
| - encoded_auth_string = _authorization_header(self.client_id, self.client_secret) |
168 |
| - self.headers["Accept"] = "application/json" |
169 |
| - self.headers["Authorization"] = f"Basic {encoded_auth_string}" |
170 |
| - self.headers["Cache-Control"] = "no-cache" |
171 |
| - |
172 |
| - self.body["grant_type"] = "client_credentials" |
173 |
| - if scope: |
174 |
| - self.body["scope"] = " ".join(self.scope) |
175 |
| - |
176 |
| - # response = _request_access_token(auth_endpoint=self.auth_endpoint, body=self.body, headers=self.headers) # noqa ERA001 |
0 commit comments