|
8 | 8 |
|
9 | 9 | from typing import Optional, Any, Callable |
10 | 10 | from authlib.integrations.flask_oauth2 import ResourceProtector |
| 11 | +from authlib.oauth2.rfc6750 import BearerTokenValidator |
11 | 12 | from ..auth.validation import Auth0JWTBearerTokenValidator |
12 | 13 | from ..utils.config_loader import get_config, get_config_value |
13 | 14 |
|
14 | 15 |
|
| 16 | +class StaticBearerToken: |
| 17 | + """Minimal token object for test-only bearer token validation.""" |
| 18 | + |
| 19 | + def __init__(self, token_string: str, scope: str = ""): |
| 20 | + self.token_string = token_string |
| 21 | + self.scope = scope |
| 22 | + |
| 23 | + def is_expired(self) -> bool: |
| 24 | + return False |
| 25 | + |
| 26 | + def is_revoked(self) -> bool: |
| 27 | + return False |
| 28 | + |
| 29 | + def get_scope(self) -> str: |
| 30 | + return self.scope |
| 31 | + |
| 32 | + |
| 33 | +class StaticBearerTokenValidator(BearerTokenValidator): |
| 34 | + """Accept a single configured bearer token for test environments.""" |
| 35 | + |
| 36 | + def __init__(self, expected_token: str): |
| 37 | + super().__init__() |
| 38 | + self.expected_token = expected_token |
| 39 | + |
| 40 | + def authenticate_token( |
| 41 | + self, token_string: Optional[str] |
| 42 | + ) -> Optional[StaticBearerToken]: |
| 43 | + if token_string == self.expected_token: |
| 44 | + return StaticBearerToken(token_string) |
| 45 | + return None |
| 46 | + |
| 47 | + |
15 | 48 | class NoOpDecorator: |
16 | 49 | """ |
17 | 50 | No-operation decorator used when authentication is disabled. |
@@ -63,14 +96,22 @@ def _setup_authentication(self) -> None: |
63 | 96 | """ |
64 | 97 | # Check if Auth0 is explicitly enabled via configuration |
65 | 98 | self._auth_enabled = get_config_value("auth.enabled", False) |
| 99 | + app_environment = get_config_value("app.environment", "") |
| 100 | + auth0_test_token = get_config_value("auth.auth0.test_token", "") |
66 | 101 |
|
67 | 102 | # Get Auth0 configuration values |
68 | 103 | auth0_address = get_config_value("auth.auth0.address", "") |
69 | 104 | auth0_audience = get_config_value("auth.auth0.audience", "") |
70 | 105 |
|
71 | 106 | # Initialize the appropriate decorator |
72 | 107 | if self._auth_enabled: |
73 | | - if auth0_address and auth0_audience: |
| 108 | + if app_environment == "test_with_auth" and auth0_test_token: |
| 109 | + resource_protector = ResourceProtector() |
| 110 | + resource_protector.register_token_validator( |
| 111 | + StaticBearerTokenValidator(auth0_test_token) |
| 112 | + ) |
| 113 | + self._decorator = resource_protector |
| 114 | + elif auth0_address and auth0_audience: |
74 | 115 | # Set up real Auth0 authentication |
75 | 116 | resource_protector = ResourceProtector() |
76 | 117 | validator = Auth0JWTBearerTokenValidator( |
|
0 commit comments