CAIP-79 exteneded tests

Author: broHeryk

symphony/bdk/core/auth/auth_session.py

@@ -5,7 +5,7 @@
 import logging
 
 from symphony.bdk.core.auth.exception import AuthInitializationError
-from symphony.bdk.core.auth.jwt_helper import extract_session_token_claims
+from symphony.bdk.core.auth.jwt_helper import extract_token_claims
 
 
 logger = logging.getLogger(__name__)
@@ -101,7 +101,7 @@ def key_manager_token(self, value):
     @property
     async def skd_enabled(self):
 
-        token_data = extract_session_token_claims(await self.session_token)
+        token_data = extract_token_claims(await self.session_token)
         if not token_data.get(SKD_FLAG_NAME, False):
             return False
         return await self._authenticator.agent_version_service.is_skd_supported()

symphony/bdk/core/auth/jwt_helper.py

@@ -85,8 +85,11 @@ def _parse_public_key_from_x509_cert(certificate: str) -> str:
         raise AuthInitializationError("Unable to parse the certificate. Check certificate format.") from exc
 
 
-def extract_session_token_claims(session_token):
-    return jwt.decode(session_token,
-                      algorithms=["RS512"],
+def extract_token_claims(session_token):
+    try:
+        return jwt.decode(session_token,
+                      algorithms=[JWT_ENCRYPTION_ALGORITHM],
                       options={"verify_signature": False}
-                      )
+                      )
+    except DecodeError:
+        return {}

symphony/bdk/core/service/version/agent_version_service.py

@@ -27,6 +27,8 @@ async def is_skd_supported(self) -> bool:
         """
         try:
             agent_info = await self._signals_api.v1_info_get()
+            if not agent_info or not agent_info.version:
+                return False
         except ApiException:
             return False
         agent_major_version, agent_minor_version = self._parse_version(agent_info.version)
@@ -38,8 +40,9 @@ async def is_skd_supported(self) -> bool:
 
     @staticmethod
     def _parse_version(version_string):
-        match = re.match(r"Agent-(\d+)\.(\d+)\..*", version_string)
+        if not version_string:
+            return None, None
+        match = re.match(VERSION_REGEXP, version_string)
         if match:
             return int(match.group(1)), int(match.group(2))
-
         return None, None

tests/conftest.py

@@ -16,39 +16,47 @@
 from symphony.bdk.core.auth.auth_session import SKD_FLAG_NAME
 
 
-@pytest.fixture(name="root_key", scope="session")  # the fixture will be created only once for entire test session.
+@pytest.fixture(
+    name="root_key", scope="session"
+)  # the fixture will be created only once for entire test session.
 def fixture_root_key():
     return rsa.generate_private_key(
-        public_exponent=65537,
-        key_size=4096,
-        backend=default_backend())
+        public_exponent=65537, key_size=4096, backend=default_backend()
+    )
 
 
 @pytest.fixture(name="rsa_key", scope="session")
 def fixture_rsa_key(root_key):
     return root_key.private_bytes(
         encoding=serialization.Encoding.PEM,
         format=serialization.PrivateFormat.PKCS8,
-        encryption_algorithm=serialization.NoEncryption()).decode("utf-8")
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode("utf-8")
 
 
 @pytest.fixture(name="certificate", scope="session")
 def fixture_certificate(root_key):
-    subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, u"commonName")])
+    subject = issuer = x509.Name(
+        [x509.NameAttribute(NameOID.COMMON_NAME, "commonName")]
+    )
     now = datetime.datetime.utcnow()
-    cert = x509.CertificateBuilder() \
-        .subject_name(subject) \
-        .issuer_name(issuer) \
-        .public_key(root_key.public_key()) \
-        .serial_number(x509.random_serial_number()) \
-        .not_valid_before(now) \
-        .not_valid_after(now + datetime.timedelta(days=30)) \
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(issuer)
+        .public_key(root_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now)
+        .not_valid_after(now + datetime.timedelta(days=30))
         .sign(root_key, hashes.SHA512(), default_backend())
+    )
     return cert.public_bytes(encoding=serialization.Encoding.PEM).decode("utf-8")
 
 
 @pytest.fixture(autouse=True)
 def mock_jwt_decode_for_skd():
-    with patch("symphony.bdk.core.auth.auth_session.extract_session_token_claims",
-               return_value={SKD_FLAG_NAME: False}) as mock:
-        yield mock
+    with patch(
+        "symphony.bdk.core.auth.auth_session.extract_token_claims",
+        return_value={SKD_FLAG_NAME: False},
+    ) as mock:
+        yield mock

tests/core/auth/auth_session_test.py

@@ -1,15 +1,36 @@
 from datetime import datetime, timezone
 
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from symphony.bdk.core.auth.auth_session import AuthSession, OboAuthSession, AppAuthSession
+from symphony.bdk.core.auth.auth_session import (
+    AuthSession,
+    OboAuthSession,
+    AppAuthSession,
+    SKD_FLAG_NAME,
+)
 from symphony.bdk.core.auth.exception import AuthInitializationError
 from symphony.bdk.gen.login_model.token import Token
 from symphony.bdk.gen.login_model.extension_app_tokens import ExtensionAppTokens
 
 
+@pytest.fixture
+def mock_authenticator():
+    authenticator = MagicMock()
+    authenticator.agent_version_service = AsyncMock()
+    authenticator.retrieve_session_token = AsyncMock(
+        return_value="session_token_string"
+    )
+    authenticator.retrieve_key_manager_token = AsyncMock(return_value="km_token_string")
+    return authenticator
+
+
+@pytest.fixture
+def auth_session(mock_authenticator):
+    return AuthSession(mock_authenticator)
+
+
 @pytest.mark.asyncio
 async def test_refresh():
     mock_bot_authenticator = AsyncMock()
@@ -57,8 +78,14 @@ async def test_auth_token():
 @pytest.mark.asyncio
 async def test_refresh_obo():
     mock_obo_authenticator = AsyncMock()
-    mock_obo_authenticator.retrieve_obo_session_token_by_user_id.side_effect = ["session_token1", "session_token2"]
-    mock_obo_authenticator.retrieve_obo_session_token_by_username.side_effect = ["session_token3", "session_token4"]
+    mock_obo_authenticator.retrieve_obo_session_token_by_user_id.side_effect = [
+        "session_token1",
+        "session_token2",
+    ]
+    mock_obo_authenticator.retrieve_obo_session_token_by_username.side_effect = [
+        "session_token3",
+        "session_token4",
+    ]
 
     obo_session = OboAuthSession(mock_obo_authenticator, user_id=1234)
 
@@ -95,14 +122,96 @@ async def test_app_auth_session():
     expire_at = 1539636528288
 
     ext_app_authenticator = AsyncMock()
-    ext_app_authenticator.authenticate_and_retrieve_tokens.return_value = \
-        ExtensionAppTokens(app_id="app_id", app_token=retrieved_app_token, symphony_token=symphony_token,
-                           expire_at=expire_at)
+    ext_app_authenticator.authenticate_and_retrieve_tokens.return_value = (
+        ExtensionAppTokens(
+            app_id="app_id",
+            app_token=retrieved_app_token,
+            symphony_token=symphony_token,
+            expire_at=expire_at,
+        )
+    )
 
     session = AppAuthSession(ext_app_authenticator, input_app_token)
     await session.refresh()
 
-    ext_app_authenticator.authenticate_and_retrieve_tokens.assert_called_once_with(input_app_token)
+    ext_app_authenticator.authenticate_and_retrieve_tokens.assert_called_once_with(
+        input_app_token
+    )
     assert session.app_token == retrieved_app_token
     assert session.symphony_token == symphony_token
     assert session.expire_at == expire_at
+
+
+@pytest.mark.asyncio
+async def test_skd_disabled_if_claim_is_missing(auth_session):
+    # Given: The token claims do not contain the SKD flag
+    with patch(
+        "symphony.bdk.core.auth.auth_session.extract_token_claims", return_value={}
+    ):
+        # When: skd_enabled is checked
+        is_enabled = await auth_session.skd_enabled
+        # Then: The result is False
+        assert is_enabled is False
+
+
+@pytest.mark.asyncio
+async def test_skd_disabled_if_agent_not_supported(auth_session, mock_authenticator):
+    # Given: The token has the SKD flag but the agent does not support it
+    mock_authenticator.agent_version_service.is_skd_supported.return_value = False
+    claims_with_skd = {SKD_FLAG_NAME: True}
+    with patch(
+        "symphony.bdk.core.auth.auth_session.extract_token_claims",
+        return_value=claims_with_skd,
+    ):
+        # When: skd_enabled is checked
+        is_enabled = await auth_session.skd_enabled
+        # Then: The result is False and the agent version was checked
+        assert is_enabled is False
+        mock_authenticator.agent_version_service.is_skd_supported.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_skd_enabled_when_fully_supported(auth_session, mock_authenticator):
+    # Given: The token has the SKD flag AND the agent supports it
+    mock_authenticator.agent_version_service.is_skd_supported.return_value = True
+    claims_with_skd = {SKD_FLAG_NAME: True}
+    with patch(
+        "symphony.bdk.core.auth.auth_session.extract_token_claims",
+        return_value=claims_with_skd,
+    ):
+        # When: skd_enabled is checked
+        is_enabled = await auth_session.skd_enabled
+        # Then: The result is True and the agent version was checked
+        assert is_enabled is True
+        mock_authenticator.agent_version_service.is_skd_supported.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_km_token_is_empty_when_skd_enabled(auth_session, mock_authenticator):
+    # Given: SKD is fully enabled
+    mock_authenticator.agent_version_service.is_skd_supported.return_value = True
+    claims_with_skd = {SKD_FLAG_NAME: True}
+    with patch(
+        "symphony.bdk.core.auth.auth_session.extract_token_claims",
+        return_value=claims_with_skd,
+    ):
+        # When: The key manager token is requested
+        km_token = await auth_session.key_manager_token
+        # Then: The token is an empty string and the real retrieval method was NOT called
+        assert km_token == ""
+        mock_authenticator.retrieve_key_manager_token.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_km_token_is_retrieved_when_skd_disabled(
+    auth_session, mock_authenticator
+):
+    # Given: SKD is disabled because the token claim is missing
+    with patch(
+        "symphony.bdk.core.auth.auth_session.extract_token_claims", return_value={}
+    ):
+        # When: The key manager token is requested
+        km_token = await auth_session.key_manager_token
+        # Then: The real token is returned and the retrieval method was called
+        assert km_token == "km_token_string"
+        mock_authenticator.retrieve_key_manager_token.assert_called_once()

tests/core/auth/jwt_helper_test.py

@@ -5,7 +5,12 @@
 from jwt import InvalidAudienceError
 
 from symphony.bdk.core.auth.exception import AuthInitializationError
-from symphony.bdk.core.auth.jwt_helper import create_signed_jwt, validate_jwt, create_signed_jwt_with_claims
+from symphony.bdk.core.auth.jwt_helper import (
+    create_signed_jwt,
+    validate_jwt,
+    create_signed_jwt_with_claims,
+    extract_token_claims,
+)
 from symphony.bdk.core.config.model.bdk_rsa_key_config import BdkRsaKeyConfig
 
 AUDIENCE = "app-id"
@@ -21,7 +26,7 @@ def fixture_jwt_payload():
     return {
         "sub": "bot-user",
         "exp": (datetime.datetime.now(datetime.timezone.utc).timestamp() + (5 * 60)),
-        "aud": AUDIENCE
+        "aud": AUDIENCE,
     }
 
 
@@ -31,7 +36,7 @@ def test_create_signed_jwt_from_path(key_config, rsa_key):
 
     mock_open = mock.mock_open(read_data=rsa_key)
 
-    with mock.patch('builtins.open', mock_open):
+    with mock.patch("builtins.open", mock_open):
         assert create_signed_jwt(key_config, "test_bot") is not None
         mock_open.assert_called_with(private_key_path, "r")
 
@@ -49,16 +54,14 @@ def test_validate_jwt(jwt_payload, certificate, rsa_key):
     assert claims == jwt_payload
 
 
-
 def test_validate_expired_jwt(jwt_payload, certificate, rsa_key):
-    jwt_payload["exp"] = (datetime.datetime.now(datetime.timezone.utc).timestamp() - 10)
+    jwt_payload["exp"] = datetime.datetime.now(datetime.timezone.utc).timestamp() - 10
     signed_jwt = create_signed_jwt_with_claims(rsa_key, jwt_payload)
 
     with pytest.raises(AuthInitializationError):
         validate_jwt(signed_jwt, certificate, AUDIENCE)
 
 
-
 def test_validate_jwt_with_empty_sub(jwt_payload, certificate, rsa_key):
     jwt_payload["sub"] = None
     signed_jwt = create_signed_jwt_with_claims(rsa_key, jwt_payload)
@@ -84,3 +87,33 @@ def test_validate_jwt_with_invalid_cert(jwt_payload, rsa_key):
 def test_validate_jwt_with_invalid_jwt(certificate):
     with pytest.raises(AuthInitializationError):
         validate_jwt("invalid jwt", certificate, AUDIENCE)
+
+
+def test_extract_claims_from_valid_token(rsa_key):
+    # Given: A valid jwt token
+    payload = {"sub": "test-bot", "skd": True, "userId": 12345}
+    token = create_signed_jwt_with_claims(rsa_key, payload)
+    # When: extract JWT claims is called with a valid token
+    claims = extract_token_claims(token)
+    # Then: fields are extracted as expected
+    assert claims["sub"] == "test-bot"
+    assert claims["skd"] is True
+    assert claims["userId"] == 12345
+
+
+@pytest.mark.parametrize(
+    "invalid_token",
+    [
+        # Given: invalid JWT to be extracted
+        "not-a-jwt",
+        "a.b.c",  # malformed JWT
+        "a.b",  # not enough segments
+        "",  # empty string
+        None,  # None value
+    ],
+)
+def test_extract_claims_from_invalid_token(invalid_token):
+    # When: extract JWT claims is called
+    claims = extract_token_claims(invalid_token)
+    # Then: empty response is returned
+    assert claims == {}