Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/auth/keycloak_mock.py
| Show All 11 Lines | |||||
| from swh.web.auth.utils import OIDC_SWH_WEB_CLIENT_ID | from swh.web.auth.utils import OIDC_SWH_WEB_CLIENT_ID | ||||
| from swh.web.config import get_config | from swh.web.config import get_config | ||||
| from .sample_data import oidc_profile, realm_public_key, userinfo | from .sample_data import oidc_profile, realm_public_key, userinfo | ||||
| class KeycloackOpenIDConnectMock(KeycloakOpenIDConnect): | class KeycloackOpenIDConnectMock(KeycloakOpenIDConnect): | ||||
| def __init__(self, auth_success=True): | def __init__(self, auth_success=True, exp=None): | ||||
| swhweb_config = get_config() | swhweb_config = get_config() | ||||
| super().__init__(swhweb_config['keycloak']['server_url'], | super().__init__(swhweb_config['keycloak']['server_url'], | ||||
| swhweb_config['keycloak']['realm_name'], | swhweb_config['keycloak']['realm_name'], | ||||
| OIDC_SWH_WEB_CLIENT_ID) | OIDC_SWH_WEB_CLIENT_ID) | ||||
| self.auth_success = auth_success | self.auth_success = auth_success | ||||
| self.exp = exp | |||||
| self._keycloak.public_key = lambda: realm_public_key | self._keycloak.public_key = lambda: realm_public_key | ||||
| self._keycloak.well_know = lambda: { | self._keycloak.well_know = lambda: { | ||||
| 'issuer': f'{self.server_url}realms/{self.realm_name}', | 'issuer': f'{self.server_url}realms/{self.realm_name}', | ||||
| 'authorization_endpoint': (f'{self.server_url}realms/' | 'authorization_endpoint': (f'{self.server_url}realms/' | ||||
| f'{self.realm_name}/protocol/' | f'{self.realm_name}/protocol/' | ||||
| 'openid-connect/auth'), | 'openid-connect/auth'), | ||||
| 'token_endpoint': (f'{self.server_url}realms/{self.realm_name}/' | 'token_endpoint': (f'{self.server_url}realms/{self.realm_name}/' | ||||
| 'protocol/openid-connect/token'), | 'protocol/openid-connect/token'), | ||||
| Show All 27 Lines | def decode_token(self, token): | ||||
| options = {} | options = {} | ||||
| if self.auth_success: | if self.auth_success: | ||||
| # skip signature expiration check as we use a static oidc_profile | # skip signature expiration check as we use a static oidc_profile | ||||
| # for the tests with expired tokens in it | # for the tests with expired tokens in it | ||||
| options['verify_exp'] = False | options['verify_exp'] = False | ||||
| decoded = super().decode_token(token, options) | decoded = super().decode_token(token, options) | ||||
| # tweak auth and exp time for tests | # tweak auth and exp time for tests | ||||
| expire_in = decoded['exp'] - decoded['auth_time'] | expire_in = decoded['exp'] - decoded['auth_time'] | ||||
| if self.exp is not None: | |||||
| decoded['exp'] = self.exp | |||||
| decoded['auth_time'] = self.exp - expire_in | |||||
| else: | |||||
| decoded['auth_time'] = int(timezone.now().timestamp()) | decoded['auth_time'] = int(timezone.now().timestamp()) | ||||
| decoded['exp'] = decoded['auth_time'] + expire_in | decoded['exp'] = decoded['auth_time'] + expire_in | ||||
| decoded['groups'] = ['/staff'] | decoded['groups'] = ['/staff'] | ||||
| return decoded | return decoded | ||||
| def mock_keycloak(mocker, auth_success=True): | def mock_keycloak(mocker, auth_success=True, exp=None): | ||||
| kc_oidc_mock = KeycloackOpenIDConnectMock(auth_success) | kc_oidc_mock = KeycloackOpenIDConnectMock(auth_success, exp) | ||||
| mock_get_oidc_client = mocker.patch( | mock_get_oidc_client = mocker.patch( | ||||
| 'swh.web.auth.views.get_oidc_client') | 'swh.web.auth.views.get_oidc_client') | ||||
| mock_get_oidc_client.return_value = kc_oidc_mock | mock_get_oidc_client.return_value = kc_oidc_mock | ||||
| mocker.patch('swh.web.auth.backends._oidc_client', kc_oidc_mock) | mocker.patch('swh.web.auth.backends._oidc_client', kc_oidc_mock) | ||||
| return kc_oidc_mock | return kc_oidc_mock | ||||