[DOP-21482] - implement unit tests for KeycloakAuthProvider

.env.docker

@@ -20,7 +20,7 @@ SYNCMASTER__DATABASE__URL=postgresql+asyncpg://syncmaster:changeme@db:5432/syncm
 
 # TODO: add to KeycloakAuthProvider documentation about creating new realms, add users, etc.
 # KEYCLOAK Auth
-SYNCMASTER__AUTH__SERVER_URL=http://keycloak:8080/
+SYNCMASTER__AUTH__SERVER_URL=http://keycloak:8080
 SYNCMASTER__AUTH__REALM_NAME=manually_created
 SYNCMASTER__AUTH__CLIENT_ID=manually_created
 SYNCMASTER__AUTH__CLIENT_SECRET=generated_by_keycloak

.env.local

@@ -18,7 +18,16 @@ export SYNCMASTER__CRYPTO_KEY=UBgPTioFrtH2unlC4XFDiGf5sYfzbdSf_VgiUSaQc94=
 # Postgres
 export SYNCMASTER__DATABASE__URL=postgresql+asyncpg://syncmaster:changeme@localhost:5432/syncmaster
 
-# Auth
+# Keycloack Auth
+export SYNCMASTER__AUTH__SERVER_URL=http://keycloak:8080
+export SYNCMASTER__AUTH__REALM_NAME=manually_created
+export SYNCMASTER__AUTH__CLIENT_ID=manually_created
+export SYNCMASTER__AUTH__CLIENT_SECRET=generated_by_keycloak
+export SYNCMASTER__AUTH__REDIRECT_URI=http://localhost:8000/auth/callback
+export SYNCMASTER__AUTH__SCOPE=email
+export SYNCMASTER__AUTH__PROVIDER=syncmaster.backend.providers.auth.keycloak_provider.KeycloakAuthProvider
+
+# Dummy Auth
 export SYNCMASTER__AUTH__PROVIDER=syncmaster.backend.providers.auth.dummy_provider.DummyAuthProvider
 export SYNCMASTER__AUTH__ACCESS_TOKEN__SECRET_KEY=secret
 

pyproject.toml

@@ -136,6 +136,7 @@ onetl = {extras = ["spark", "s3", "hdfs"], version = "^0.12.0"}
 faker = ">=28.4.1,<34.0.0"
 coverage = "^7.6.1"
 gevent = "^24.2.1"
+responses = "*"
 
 [tool.poetry.group.dev.dependencies]
 mypy = "^1.11.2"

syncmaster/backend/providers/auth/keycloak_provider.py

@@ -81,6 +81,8 @@ async def get_current_user(self, access_token: str, *args, **kwargs) -> Any:
             self.redirect_to_auth(request.url.path)
 
         try:
+            # if user is disabled or blocked in Keycloak after the token is issued, he will
+            # remain authorized until the token expires (not more than 15 minutes in MTS SSO)
             token_info = self.keycloak_openid.decode_token(token=access_token)
         except Exception as e:
             log.info("Access token is invalid or expired: %s", e)

tests/conftest.py

@@ -32,6 +32,7 @@
 
 pytest_plugins = [
     "tests.test_unit.test_transfers.transfer_fixtures",
+    "tests.test_unit.test_auth.auth_fixtures",
     "tests.test_unit.test_runs.run_fixtures",
     "tests.test_unit.test_connections.connection_fixtures",
     "tests.test_unit.test_scheduler.scheduler_fixtures",
@@ -64,9 +65,9 @@ def event_loop():
     loop.close()
 
 
-@pytest.fixture(scope="session")
-def settings():
-    return Settings()
+@pytest.fixture(scope="session", params=[{}])
+def settings(request: pytest.FixtureRequest) -> Settings:
+    return Settings.parse_obj(request.param)
 
 
 @pytest.fixture(scope="session")

tests/test_unit/test_auth/auth_fixtures/__init__.py

@@ -0,0 +1,7 @@
+from tests.test_unit.test_auth.auth_fixtures.keycloak_fixture import (
+    create_session_cookie,
+    mock_keycloak_realm,
+    mock_keycloak_token_refresh,
+    mock_keycloak_well_known,
+    rsa_keys,
+)

tests/test_unit/test_auth/auth_fixtures/keycloak_fixture.py

@@ -0,0 +1,158 @@
+import json
+import time
+from base64 import b64encode
+
+import pytest
+import responses
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+from itsdangerous import TimestampSigner
+from jose import jwt
+
+
+@pytest.fixture(scope="session")
+def rsa_keys():
+    # create private & public keys to emulate Keycloak signing
+    private_key = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+    )
+    private_pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    public_key = private_key.public_key()
+
+    return {
+        "private_key": private_key,
+        "private_pem": private_pem,
+        "public_key": public_key,
+    }
+
+
+def get_public_key_pem(public_key):
+    public_pem = public_key.public_bytes(
+        encoding=Encoding.PEM,
+        format=PublicFormat.SubjectPublicKeyInfo,
+    )
+    public_pem_str = public_pem.decode("utf-8")
+    public_pem_str = public_pem_str.replace("-----BEGIN PUBLIC KEY-----\n", "")
+    public_pem_str = public_pem_str.replace("-----END PUBLIC KEY-----\n", "")
+    public_pem_str = public_pem_str.replace("\n", "")
+    return public_pem_str
+
+
+@pytest.fixture
+def create_session_cookie(rsa_keys, settings):
+    def _create_session_cookie(user, expire_in_msec=1000) -> str:
+        private_pem = rsa_keys["private_pem"]
+        session_secret_key = settings.server.session.secret_key
+
+        payload = {
+            "sub": str(user.id),
+            "preferred_username": user.username,
+            "email": user.email,
+            "given_name": user.first_name,
+            "middle_name": user.middle_name,
+            "family_name": user.last_name,
+            "exp": int(time.time()) + (expire_in_msec / 1000),
+        }
+
+        access_token = jwt.encode(payload, private_pem, algorithm="RS256")
+        refresh_token = "mock_refresh_token"
+
+        session_data = {
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+        }
+
+        signer = TimestampSigner(session_secret_key)
+        json_bytes = json.dumps(session_data).encode("utf-8")
+        base64_bytes = b64encode(json_bytes)
+        signed_data = signer.sign(base64_bytes)
+        session_cookie = signed_data.decode("utf-8")
+
+        return session_cookie
+
+    return _create_session_cookie
+
+
+@pytest.fixture
+def mock_keycloak_well_known(settings):
+    server_url = settings.auth.server_url
+    realm_name = settings.auth.client_id
+    well_known_url = f"{server_url}/realms/{realm_name}/.well-known/openid-configuration"
+
+    responses.add(
+        responses.GET,
+        well_known_url,
+        json={
+            "authorization_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/auth",
+            "token_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/token",
+            "userinfo_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/userinfo",
+            "end_session_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/logout",
+            "jwks_uri": f"{server_url}/realms/{realm_name}/protocol/openid-connect/certs",
+            "issuer": f"{server_url}/realms/{realm_name}",
+        },
+        status=200,
+        content_type="application/json",
+    )
+
+
+@pytest.fixture
+def mock_keycloak_realm(settings, rsa_keys):
+    server_url = settings.auth.server_url
+    realm_name = settings.auth.client_id
+    realm_url = f"{server_url}/realms/{realm_name}"
+    public_pem_str = get_public_key_pem(rsa_keys["public_key"])
+
+    responses.add(
+        responses.GET,
+        realm_url,
+        json={
+            "realm": realm_name,
+            "public_key": public_pem_str,
+            "token-service": f"{server_url}/realms/{realm_name}/protocol/openid-connect/token",
+            "account-service": f"{server_url}/realms/{realm_name}/account",
+        },
+        status=200,
+        content_type="application/json",
+    )
+
+
+@pytest.fixture
+def mock_keycloak_token_refresh(settings, rsa_keys):
+    server_url = settings.auth.server_url
+    realm_name = settings.auth.client_id
+    token_url = f"{server_url}/realms/{realm_name}/protocol/openid-connect/token"
+
+    # generate new access and refresh tokens
+    expires_in = int(time.time()) + 1000
+    private_pem = rsa_keys["private_pem"]
+    payload = {
+        "sub": "mock_user_id",
+        "preferred_username": "mock_username",
+        "email": "mock_email@example.com",
+        "given_name": "Mock",
+        "middle_name": "User",
+        "family_name": "Name",
+        "exp": expires_in,
+    }
+
+    new_access_token = jwt.encode(payload, private_pem, algorithm="RS256")
+    new_refresh_token = "mock_new_refresh_token"
+
+    responses.add(
+        responses.POST,
+        token_url,
+        json={
+            "access_token": new_access_token,
+            "refresh_token": new_refresh_token,
+            "token_type": "bearer",
+            "expires_in": expires_in,
+        },
+        status=200,
+        content_type="application/json",
+    )