|
1 | 1 | import logging |
| 2 | +import ssl |
2 | 3 | from datetime import datetime, timedelta |
3 | 4 | from typing import TYPE_CHECKING, Any, Dict, List, Optional |
| 5 | +from typing import TYPE_CHECKING, Any, Dict, List, NotRequired, Optional, TypedDict |
4 | 6 |
|
5 | 7 | import jwt |
6 | 8 | from fastapi import HTTPException, status |
|
12 | 14 | log = logging.getLogger('fastapi_azure_auth') |
13 | 15 |
|
14 | 16 |
|
| 17 | +class HttpClientConfig(TypedDict): |
| 18 | + verify: NotRequired[ssl.SSLContext] |
| 19 | + |
| 20 | + |
15 | 21 | class OpenIdConfig: |
16 | 22 | def __init__( |
17 | 23 | self, |
18 | 24 | tenant_id: Optional[str] = None, |
19 | 25 | multi_tenant: bool = False, |
20 | 26 | app_id: Optional[str] = None, |
21 | 27 | config_url: Optional[str] = None, |
| 28 | + http_client_config: Optional[HttpClientConfig] = None, |
22 | 29 | ) -> None: |
23 | 30 | self.tenant_id: Optional[str] = tenant_id |
24 | 31 | self._config_timestamp: Optional[datetime] = None |
25 | 32 | self.multi_tenant: bool = multi_tenant |
26 | 33 | self.app_id = app_id |
27 | 34 | self.config_url = config_url |
| 35 | + self.http_client_config: HttpClientConfig = ( |
| 36 | + http_client_config or HttpClientConfig() |
| 37 | + ) |
28 | 38 |
|
29 | 39 | self.authorization_endpoint: str |
30 | 40 | self.signing_keys: dict[str, 'AllowedPublicKeys'] |
@@ -72,7 +82,7 @@ async def _load_openid_config(self) -> None: |
72 | 82 | if self.app_id: |
73 | 83 | config_url += f'?appid={self.app_id}' |
74 | 84 |
|
75 | | - async with AsyncClient(timeout=10) as client: |
| 85 | + async with AsyncClient(timeout=10, **self.http_client_config) as client: |
76 | 86 | log.info('Fetching OpenID Connect config from %s', config_url) |
77 | 87 | openid_response = await client.get(config_url) |
78 | 88 | openid_response.raise_for_status() |
|
0 commit comments