endurain/backend/tests/session/test_token_manager.py

from datetime import datetime, timezone

import pytest
from fastapi import HTTPException

import session.token_manager as session_token_manager


class TestTokenManagerSecurity:
    """
    Test suite for the TokenManager class, focusing on security-related aspects of token creation, decoding, validation, and CSRF token generation.

    This class contains tests that verify:
    - Correct extraction of claims from tokens.
    - Proper handling of missing or invalid claims.
    - Decoding of valid and invalid tokens, including detection of tampering and use of incorrect secrets.
    - Validation of token expiration, including handling of expired tokens.
    - Creation of access and refresh tokens, ensuring correct types, expiration times, and uniqueness.
    - Security properties of CSRF tokens, such as uniqueness and sufficient length.
    - Robustness against empty or None tokens.
    - Use of secure algorithms for token signing.
    - Consistency of token expiration times in UTC.
    - Uniqueness of tokens for different session IDs and for repeated calls with the same user/session.

    Each test ensures that the TokenManager behaves securely and correctly under various scenarios, raising appropriate exceptions and maintaining cryptographic standards.
    """

    def test_get_token_claim_returns_correct_value(
        self, token_manager, sample_user_read
    ):
        """
        Tests that the `get_token_claim` method of the token manager returns the correct values for specific claims.
        This test creates a token for a given session and user, then verifies that:
        - The "sub" claim matches the user's ID.
        - The "sid" claim matches the session ID.
        """
        session_id = "test-session-123"
        _, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        sub_claim = token_manager.get_token_claim(token, "sub")
        sid_claim = token_manager.get_token_claim(token, "sid")

        assert (
            sub_claim == sample_user_read.id
        ), "sub claim should match user ID (as int)"
        assert sid_claim == session_id, "sid claim should match session ID"

    def test_get_token_claim_with_missing_claim(self, token_manager, sample_user_read):
        """
        Test that get_token_claim raises an HTTPException with status code 401 when attempting to retrieve a claim
        that does not exist in the token. Verifies that the exception detail message indicates the claim is missing.
        """
        session_id = "test-session-id"
        _, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        with pytest.raises(HTTPException) as exc_info:
            token_manager.get_token_claim(token, "nonexistent_claim")
        assert exc_info.value.status_code == 401
        assert "missing" in exc_info.value.detail.lower()

    def test_decode_valid_token(self, token_manager, sample_user_read):
        """
        Tests that a valid token generated by the token manager can be successfully decoded.

        This test verifies that:
        - A token created with a valid session ID and user information can be decoded.
        - The decoded payload is not None.
        - The decoded payload has a 'claims' attribute.
        - The 'claims' attribute of the payload is not None.

        Args:
            token_manager: The token manager instance used to create and decode tokens.
            sample_user_read: A sample user object used for token creation.
        """
        session_id = "test-session-id"
        _, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        payload = token_manager.decode_token(token)
        assert payload is not None, "Decoded payload should not be None"
        assert hasattr(payload, "claims"), "Payload should have claims attribute"
        assert payload.claims is not None, "Claims should not be None"

    def test_decode_token_contains_expected_claims(
        self, token_manager, sample_user_read
    ):
        """
        Test that the decoded token contains the expected claims.

        This test verifies that when a token is created using the token manager,
        the decoded payload includes the required claims:
        - 'sub': the user ID (subject)
        - 'sid': the session ID
        - 'exp': the expiration timestamp

        Args:
            token_manager: The token manager instance used to create and decode tokens.
            sample_user_read: A sample user object used for token creation.

        Asserts:
            - The 'sub' claim is present in the token payload.
            - The 'sid' claim is present in the token payload.
            - The 'exp' claim is present in the token payload.
        """
        session_id = "test-session-id"
        _, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        payload = token_manager.decode_token(token)
        claims = payload.claims

        assert "sub" in claims, "Token should have 'sub' claim (user_id)"
        assert "sid" in claims, "Token should have 'sid' claim (session_id)"
        assert "exp" in claims, "Token should have 'exp' claim"
        # Note: 'type' claim is not actually used in token creation

    def test_decode_token_with_invalid_token(self, token_manager):
        """
        Tests that the `decode_token` method of the token manager raises an HTTPException with status code 401
        when provided with various invalid JWT tokens.

        Args:
            token_manager: An instance of the token manager to be tested.

        Asserts:
            - An HTTPException is raised for each invalid token.
            - The raised exception has a status code of 401.
        """
        invalid_tokens = [
            "invalid.token.here",
            "not-a-jwt",
            "",
            "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.invalid",
        ]

        for invalid_token in invalid_tokens:
            with pytest.raises(HTTPException) as exc_info:
                token_manager.decode_token(invalid_token)
            assert exc_info.value.status_code == 401

    def test_decode_token_with_wrong_secret(self, sample_user_read):
        """
        Test that decoding a token with a different secret key than the one used to create it raises an HTTP 401 Unauthorized exception.

        This test creates a token using one instance of TokenManager with a specific secret key, then attempts to decode the token using another TokenManager instance with a different secret key. It asserts that an HTTPException with status code 401 is raised, indicating unauthorized access due to the wrong secret.
        """
        # Create token with one manager
        manager1 = session_token_manager.TokenManager(
            secret_key="secret-key-one-min-32-characters-long"
        )
        _, token = manager1.create_token(
            "session-id", sample_user_read, session_token_manager.TokenType.ACCESS
        )

        # Try to decode with different manager
        manager2 = session_token_manager.TokenManager(
            secret_key="secret-key-two-min-32-characters-long"
        )
        with pytest.raises(HTTPException) as exc_info:
            manager2.decode_token(token)
        assert exc_info.value.status_code == 401

    def test_validate_token_expiration_with_valid_token(
        self, token_manager, sample_user_read
    ):
        """
        Test that `validate_token_expiration` does not raise an exception when provided with a valid (non-expired) token.

        This test creates a valid access token using the token manager and verifies that calling
        `validate_token_expiration` with this token does not raise an `HTTPException`. If an exception is raised,
        the test fails, indicating that valid tokens are incorrectly being marked as expired.
        """
        session_id = "test-session-id"
        _, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        # Should not raise an exception
        try:
            token_manager.validate_token_expiration(token)
        except HTTPException:
            pytest.fail("Valid token should not raise HTTPException")

    def test_validate_token_expiration_with_expired_token(self, token_manager):
        """
        Test that validate_token_expiration raises an HTTPException with status code 401
        when provided with an expired JWT token.

        Args:
            self: The test class instance.
            token_manager: The token manager instance to be tested.

        Asserts:
            - An HTTPException is raised when an expired token is validated.
            - The exception's status code is 401 (Unauthorized).
        """
        token = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzaWQiOiJzZXNzaW9uLWlkIiwiaXNzIjoiaHR0cDovL2xvY2FsaG9zdDo4MDgwIiwiYXVkIjoiaHR0cDovL2xvY2FsaG9zdDo4MDgwIiwic3ViIjoxLCJzY29wZSI6WyJwcm9maWxlIiwidXNlcnM6cmVhZCIsImdlYXJzOnJlYWQiLCJnZWFyczp3cml0ZSIsImFjdGl2aXRpZXM6cmVhZCIsImFjdGl2aXRpZXM6d3JpdGUiLCJoZWFsdGg6cmVhZCIsImhlYWx0aDp3cml0ZSIsImhlYWx0aF90YXJnZXRzOnJlYWQiLCJoZWFsdGhfdGFyZ2V0czp3cml0ZSJdLCJpYXQiOjE3NTk5NTMxODUsIm5iZiI6MTc1OTk1MzE4NSwiZXhwIjoxNzU5OTU0MDg1LCJqdGkiOiI3OWY2NDJlZC00NzNkLTQxMGYtYWMyNS0yMjYxMDU5YTM4NjIifQ.VSizGzvIIi_EJYD_YmfZBEBE_9aJbhLW-25cD1kEOeM"

        with pytest.raises(HTTPException) as excinfo:
            token_manager.validate_token_expiration(token)

        assert excinfo.value.status_code == 401

    def test_create_access_token(self, token_manager, sample_user_read):
        """
        Tests the creation of an access token using the token manager.

        This test verifies that:
        - The generated token is not None.
        - The token is a non-empty string.
        - The expiration time is a datetime object.
        - The expiration time is set in the future.

        Args:
            token_manager: The token manager instance used to create tokens.
            sample_user_read: A sample user object for whom the token is created.
        """
        session_id = "test-session-id"
        exp_time, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        assert token is not None, "Token should not be None"
        assert isinstance(token, str), "Token should be a string"
        assert len(token) > 0, "Token should not be empty"
        assert isinstance(exp_time, datetime), "Expiration should be a datetime"
        assert exp_time > datetime.now(
            timezone.utc
        ), "Expiration should be in the future"

    def test_create_refresh_token(self, token_manager, sample_user_read):
        """
        Test the creation of a refresh token using the token manager.

        This test verifies that the `create_token` method of the token manager:
        - Returns a non-None, non-empty string token.
        - Returns an expiration time as a `datetime` object.
        - Ensures the expiration time is set in the future.

        Args:
            token_manager: The token manager instance used to create the token.
            sample_user_read: A sample user object to associate with the token.

        Asserts:
            - The generated token is not None.
            - The token is a string and not empty.
            - The expiration time is a `datetime` object and is set in the future.
        """
        session_id = "test-session-id"
        exp_time, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.REFRESH
        )

        assert token is not None, "Token should not be None"
        assert isinstance(token, str), "Token should be a string"
        assert len(token) > 0, "Token should not be empty"
        assert isinstance(exp_time, datetime), "Expiration should be a datetime"
        assert exp_time > datetime.now(
            timezone.utc
        ), "Expiration should be in the future"

    def test_access_token_shorter_expiration_than_refresh(
        self, token_manager, sample_user_read
    ):
        """
        Test that the access token has a shorter expiration time than the refresh token.

        This test verifies that when creating both an access token and a refresh token for the same session and user,
        the expiration time of the access token is less than that of the refresh token, ensuring correct token lifetimes.
        """
        session_id = "test-session-id"
        access_exp, _ = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )
        refresh_exp, _ = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.REFRESH
        )

        assert (
            access_exp < refresh_exp
        ), "Access token should expire before refresh token"

    def test_create_csrf_token_generates_unique_tokens(self, token_manager):
        """
        Test that `create_csrf_token` generates unique CSRF tokens.

        This test calls the `create_csrf_token` method of the token manager multiple times
        and asserts that all generated tokens are unique, ensuring cryptographic security.
        """
        tokens = [token_manager.create_csrf_token() for _ in range(10)]
        unique_tokens = set(tokens)

        assert (
            len(unique_tokens) == 10
        ), "CSRF tokens should be unique (cryptographically secure)"

    def test_create_csrf_token_has_sufficient_length(self, token_manager):
        """
        Test that the CSRF token generated by the token manager has a sufficient length.

        This test ensures that the `create_csrf_token` method of the token manager
        returns a token string that is at least 32 characters long, which is important
        for maintaining adequate security against brute-force attacks.

        Args:
            token_manager: An instance of the token manager responsible for generating CSRF tokens.

        Asserts:
            The generated CSRF token has a length of at least 32 characters.
        """
        token = token_manager.create_csrf_token()

        assert len(token) >= 32, "CSRF token should be at least 32 characters long"

    def test_tokens_are_different_for_same_user(self, token_manager, sample_user_read):
        """
        Test that creating multiple tokens for the same user and session results in unique tokens.

        This test ensures that even when the same user and session ID are provided to the token manager,
        each call to `create_token` generates a distinct token, verifying that token generation does not
        produce duplicate tokens for identical input parameters.

        Args:
            token_manager: The token manager instance responsible for creating tokens.
            sample_user_read: A sample user object used for token creation.

        Asserts:
            The two generated tokens are not equal, confirming token uniqueness.
        """
        session_id = "test-session-id"
        _, token1 = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )
        _, token2 = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        assert token1 != token2, "Tokens should be unique even for the same user"

    def test_token_tampering_detection(self, token_manager, sample_user_read):
        """
        Tests that the token manager correctly detects and rejects tampered tokens.

        This test creates a valid token, deliberately modifies (tampers with) its contents,
        and then attempts to decode it. The expected behavior is that the token manager
        raises an HTTPException with a 401 status code, indicating unauthorized access
        due to token tampering.

        Args:
            token_manager: The token manager instance used to create and decode tokens.
            sample_user_read: A sample user object used for token creation.

        Raises:
            HTTPException: If the tampered token is detected as invalid, with status code 401.
        """
        session_id = "test-session-id"
        _, token = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        # Tamper with the token
        tampered_token = token[:-5] + "XXXXX"

        with pytest.raises(HTTPException) as exc_info:
            token_manager.decode_token(tampered_token)
        assert exc_info.value.status_code == 401

    def test_empty_token_handling(self, token_manager):
        """
        Test that the token manager raises an HTTPException with status code 401
        when attempting to decode an empty token string.
        """
        with pytest.raises(HTTPException) as exc_info:
            token_manager.decode_token("")
        assert exc_info.value.status_code == 401

    def test_none_token_handling(self, token_manager):
        """
        Test that the token_manager.decode_token method raises an appropriate exception
        (HTTPException, AttributeError, or TypeError) when called with a None token.
        """
        with pytest.raises((HTTPException, AttributeError, TypeError)):
            token_manager.decode_token(None)

    def test_token_algorithm_is_secure(self, token_manager):
        """
        Test to ensure that the token manager uses a secure algorithm.

        This test asserts that the default algorithm used by the token manager is "HS256",
        which is considered secure for signing tokens. If a weaker algorithm is used,
        the test will fail, indicating a potential security risk.
        """
        assert (
            token_manager.algorithm == "HS256"
        ), "Default algorithm should be HS256 or stronger"

    def test_different_session_ids_produce_different_tokens(
        self, token_manager, sample_user_read
    ):
        """
        Test that creating tokens with different session IDs produces different token values.

        This test ensures that the token manager generates unique tokens for different session IDs,
        even when the user and token type are the same. It verifies that the tokens created for
        distinct session IDs are not equal, which is important for session isolation and security.

        Args:
            token_manager: The token manager instance used to create tokens.
            sample_user_read: A sample user object used for token creation.

        Asserts:
            The tokens generated for different session IDs are not equal.
        """
        _, token1 = token_manager.create_token(
            "session-id-1", sample_user_read, session_token_manager.TokenType.ACCESS
        )
        _, token2 = token_manager.create_token(
            "session-id-2", sample_user_read, session_token_manager.TokenType.ACCESS
        )

        assert token1 != token2, "Different session IDs should produce different tokens"

    def test_token_expiration_time_is_in_utc(self, token_manager, sample_user_read):
        """
        Test that the token expiration time generated by the token manager is set in the UTC timezone.

        This test verifies that when a token is created using the token manager, the returned expiration time (`exp_time`)
        has its `tzinfo` attribute set to `timezone.utc`, ensuring that all expiration times are consistently in UTC.
        """
        session_id = "test-session-id"
        exp_time, _ = token_manager.create_token(
            session_id, sample_user_read, session_token_manager.TokenType.ACCESS
        )

        assert (
            exp_time.tzinfo == timezone.utc
        ), "Expiration time should be in UTC timezone"