[DOP-21482] - implement unit tests for KeycloakAuthProvider

Author: maxim-lixakov

.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
 

poetry.lock

@@ -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"

pyproject.toml

@@ -64,9 +64,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/conftest.py

@@ -0,0 +1,118 @@
+import json
+from base64 import b64encode
+
+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
+
+# copied from .env.docker as backend tries to send requests to corresponding
+KEYCLOAK_CONFIG = {
+    "server_url": "http://keycloak:8080",
+    "realm_name": "manually_created",
+    "redirect_uri": "http://localhost:8000/v1/auth/callback",
+    "client_secret": "generated_by_keycloak",
+    "scope": "email",
+    "client_id": "test-client",
+}
+# 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()
+
+
+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
+
+
+def create_session_cookie(payload: dict, session_secret_key: str) -> str:
+    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)
+    return signed_data.decode("utf-8")
+
+
+def mock_keycloak_well_known(responses_mock):
+    server_url = KEYCLOAK_CONFIG.get("server_url")
+    realm_name = KEYCLOAK_CONFIG.get("realm_name")
+    well_known_url = f"{server_url}/realms/{realm_name}/.well-known/openid-configuration"
+
+    responses_mock.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",
+    )
+
+
+def mock_keycloak_token_endpoint(responses_mock, access_token: str, refresh_token: str):
+    server_url = KEYCLOAK_CONFIG.get("server_url")
+    realm_name = KEYCLOAK_CONFIG.get("realm_name")
+    token_url = f"{server_url}/realms/{realm_name}/protocol/openid-connect/token"
+
+    responses_mock.add(
+        responses.POST,
+        token_url,
+        body=json.dumps(
+            {
+                "access_token": access_token,
+                "refresh_token": refresh_token,
+                "token_type": "bearer",
+                "expires_in": 3600,
+            },
+        ),
+        status=200,
+        content_type="application/json",
+    )
+
+
+def mock_keycloak_realm(responses_mock):
+    server_url = KEYCLOAK_CONFIG.get("server_url")
+    realm_name = KEYCLOAK_CONFIG.get("realm_name")
+    realm_url = f"{server_url}/realms/{realm_name}"
+
+    responses_mock.add(
+        responses.GET,
+        realm_url,
+        json={
+            "realm": realm_name,
+            "public_key": get_public_key_pem(PUBLIC_KEY),
+            "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",
+    )

tests/test_unit/test_auth/__init__.py

@@ -0,0 +1,147 @@
+import time
+
+import pytest
+import responses
+from httpx import AsyncClient
+
+from syncmaster.backend.settings import BackendSettings as Settings
+from tests.mocks import MockUser
+from tests.test_unit.test_auth.mocks.keycloak import (
+    create_session_cookie,
+    mock_keycloak_realm,
+    mock_keycloak_well_known,
+)
+
+KEYCLOAK_PROVIDER = "syncmaster.backend.providers.auth.keycloak_provider.KeycloakAuthProvider"
+pytestmark = [pytest.mark.asyncio, pytest.mark.backend]
+
+
+@responses.activate
+@pytest.mark.parametrize("settings", [{"auth": {"provider": KEYCLOAK_PROVIDER}}], indirect=True)
+async def test_get_keycloak_user_unauthorized(client: AsyncClient):
+    mock_keycloak_well_known(responses)
+
+    response = await client.get("/v1/users/some_user_id")
+
+    # redirect unauthorized user to Keycloak
+    assert response.status_code == 307
+    assert "protocol/openid-connect/auth?" in str(
+        response.next_request.url,
+    )
+
+
+@responses.activate
+@pytest.mark.parametrize("settings", [{"auth": {"provider": KEYCLOAK_PROVIDER}}], indirect=True)
+async def test_get_keycloak_user_authorized(
+    client: AsyncClient,
+    simple_user: MockUser,
+    settings: Settings,
+    access_token_factory,
+):
+    payload = {
+        "sub": str(simple_user.id),
+        "preferred_username": simple_user.username,
+        "email": simple_user.email,
+        "given_name": simple_user.first_name,
+        "middle_name": simple_user.middle_name,
+        "family_name": simple_user.last_name,
+        "exp": time.time() + 1000,
+    }
+
+    mock_keycloak_well_known(responses)
+    mock_keycloak_realm(responses)
+
+    session_cookie = create_session_cookie(payload, settings.server.session.secret_key)
+    headers = {
+        "Cookie": f"session={session_cookie}",
+    }
+    response = await client.get(
+        f"/v1/users/{simple_user.id}",
+        headers=headers,
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "id": simple_user.id,
+        "is_superuser": simple_user.is_superuser,
+        "username": simple_user.username,
+    }
+
+
+@responses.activate
+@pytest.mark.parametrize("settings", [{"auth": {"provider": KEYCLOAK_PROVIDER}}], indirect=True)
+async def test_get_keycloak_deleted_user(
+    client: AsyncClient,
+    simple_user: MockUser,
+    deleted_user: MockUser,
+    settings: Settings,
+):
+    payload = {
+        "sub": str(simple_user.id),
+        "preferred_username": simple_user.username,
+        "email": simple_user.email,
+        "given_name": simple_user.first_name,
+        "middle_name": simple_user.middle_name,
+        "family_name": simple_user.last_name,
+        "exp": time.time() + 1000,
+    }
+
+    mock_keycloak_well_known(responses)
+    mock_keycloak_realm(responses)
+
+    session_cookie = create_session_cookie(payload, settings.server.session.secret_key)
+    headers = {
+        "Cookie": f"session={session_cookie}",
+    }
+    response = await client.get(
+        f"/v1/users/{deleted_user.id}",
+        headers=headers,
+    )
+    assert response.status_code == 404
+    assert response.json() == {
+        "error": {
+            "code": "not_found",
+            "message": "User not found",
+            "details": None,
+        },
+    }
+
+
+@responses.activate
+@pytest.mark.parametrize("settings", [{"auth": {"provider": KEYCLOAK_PROVIDER}}], indirect=True)
+async def test_get_keycloak_user_inactive(
+    client: AsyncClient,
+    simple_user: MockUser,
+    inactive_user: MockUser,
+    settings: Settings,
+):
+    payload = {
+        "sub": str(inactive_user.id),
+        "preferred_username": inactive_user.username,
+        "email": inactive_user.email,
+        "given_name": inactive_user.first_name,
+        "middle_name": inactive_user.middle_name,
+        "family_name": inactive_user.last_name,
+        "exp": time.time() + 1000,
+    }
+
+    mock_keycloak_well_known(responses)
+    mock_keycloak_realm(responses)
+
+    session_cookie = create_session_cookie(payload, settings.server.session.secret_key)
+    headers = {
+        "Cookie": f"session={session_cookie}",
+    }
+
+    response = await client.get(
+        f"/v1/users/{simple_user.id}",
+        headers=headers,
+    )
+    assert response.status_code == 403
+    assert response.json() == {
+        "error": {
+            "code": "forbidden",
+            "message": "You have no power here",
+            "details": None,
+        },
+    }