ss-tools/backend/tests/test_security_orthogonal.py

# #region TestSecurityOrthogonal [C:3] [TYPE Module] [SEMANTICS test, security, orthogonal, edge-cases]
# @BRIEF Orthogonal security tests for critical auth/encryption modules.
#   Tests edge cases and invariants that existing tests miss:
#   - bcrypt 72-char limit, unicode passwords
#   - API key format validation, edge characters
#   - Fernet encryption: invalid ciphertext, empty data, corruption
#   - Encryption key lifecycle: missing .env, unwritable dir
#   - Log injection: CRLF in username, long messages
#   - JWT: expired, malformed, missing claims
# @RELATION BINDS_TO -> [AuthSecurityModule]
# @RELATION BINDS_TO -> [APIKeyUtilities]
# @RELATION BINDS_TO -> [EncryptionCore]
# @RELATION BINDS_TO -> [EncryptionKeyModule]
# @RELATION BINDS_TO -> [AuthLoggerModule]
# @TEST_CONTRACT: PasswordSecurity -> edge-case password handling
# @TEST_CONTRACT: ApiKeyFormat -> key structure invariants
# @TEST_CONTRACT: EncryptionRobustness -> error handling on invalid data
# @TEST_CONTRACT: KeyLifecycle -> encryption key resolution paths
# @TEST_CONTRACT: LogInjectionProtection -> CRLF/control chars in log input
# @TEST_INVARIANT: bcrypt_truncation -> bcrypt silently truncates at 72 bytes
# @TEST_INVARIANT: api_key_format -> ssk_ prefix + hash uniqueness
# @TEST_INVARIANT: fernet_symmetric -> encrypt/decrypt is reversible
# @TEST_INVARIANT: log_no_injection -> CRLF in input doesn't forge log entries
import os
import tempfile
from pathlib import Path
from unittest.mock import patch

import pytest

# ──────────────────────────────────────────────
# AUTH: Password security edge cases
# ──────────────────────────────────────────────


class TestPasswordSecurity:
    """Orthogonal tests for password hashing — edge cases bcrypt doesn't handle well."""

    # #region test_password_bcrypt_72_byte_limit [C:2] [TYPE Function]
    # @BRIEF bcrypt truncates passwords longer than 72 bytes — the first 72 bytes
    #   determine the hash. A password of 73 bytes where byte 73 differs should
    #   produce the SAME hash.
    # @TEST_EDGE: password_over_72_bytes -> bcrypt silently truncates
    def test_password_bcrypt_72_byte_limit(self):
        """bcrypt truncates at 72 bytes: passwords differing only after byte 72 match."""
        from src.core.auth.security import get_password_hash, verify_password

        # 80-byte passwords differing only at position 73+
        pwd1 = "A" * 72 + "B" * 8
        pwd2 = "A" * 72 + "C" * 8

        h1 = get_password_hash(pwd1)
        assert verify_password(pwd2, h1)  # bcrypt sees same first 72 bytes
    # #endregion test_password_bcrypt_72_byte_limit

    # #region test_password_unicode_normalization [C:2] [TYPE Function]
    # @BRIEF Unicode composed vs decomposed forms — bcrypt operates on bytes,
    #   so "é" (U+00E9) != "e" + combining accent (U+0065 U+0301).
    # @TEST_EDGE: unicode_decomposition -> different byte sequences fail
    def test_password_unicode_normalization(self):
        """Unicode normalization: NFC != NFD produces different hashes."""
        from src.core.auth.security import get_password_hash, verify_password

        import unicodedata
        nfc = unicodedata.normalize("NFC", "café")      # single codepoint é
        nfd = unicodedata.normalize("NFD", "café")      # e + combining accent

        assert nfc != nfd  # different byte sequences

        h = get_password_hash(nfc)
        assert verify_password(nfd, h) is False  # should NOT match
    # #endregion test_password_unicode_normalization

    # #region test_password_empty_and_whitespace [C:2] [TYPE Function]
    # @BRIEF Empty password, space-only password, null hash.
    def test_password_empty_and_whitespace(self):
        from src.core.auth.security import verify_password, get_password_hash

        # Empty password — should hash and verify
        h = get_password_hash("")
        assert verify_password("", h)

        # Space-only password
        h = get_password_hash("   ")
        assert verify_password("   ", h)

        # Verify against None — should return False, not crash
        assert not verify_password("password", None)
        assert not verify_password("password", "")
    # #endregion test_password_empty_and_whitespace


# ──────────────────────────────────────────────
# AUTH: API key format invariants
# ──────────────────────────────────────────────


class TestApiKeyFormat:
    """Orthogonal tests for API key generation — format invariants and edge cases."""

    # #region test_api_key_format_invariants [C:2] [TYPE Function]
    # @BRIEF Every generated key must start with ssk_, prefix is exactly 11 chars,
    #   key_hash is 64 hex chars, raw_key contains only URL-safe base64.
    def test_api_key_format_invariants(self):
        from src.core.auth.api_key import generate_api_key, hash_api_key

        for _ in range(100):
            raw, prefix, key_hash = generate_api_key()

            assert raw.startswith("ssk_"), f"Key must start with ssk_: {raw[:10]}"
            assert len(prefix) == 11, f"Prefix must be 11 chars: {prefix}"
            assert prefix == raw[:11], f"Prefix mismatch: {prefix} != {raw[:11]}"
            assert len(key_hash) == 64, f"Hash must be 64 hex chars: {key_hash}"
            int(key_hash, 16)  # must be valid hex — raises ValueError if not

            # Raw key is URL-safe base64 after ssk_ prefix
            body = raw[4:]  # after "ssk_"
            import string
            allowed = set(string.ascii_letters + string.digits + "-_")
            assert all(c in allowed for c in body), f"Invalid chars in key body: {body[:10]}"
    # #endregion test_api_key_format_invariants

    # #region test_api_key_hash_uniqueness [C:2] [TYPE Function]
    # @BRIEF 1000 generated keys must all have unique raw keys and unique hashes.
    def test_api_key_hash_uniqueness(self):
        from src.core.auth.api_key import generate_api_key

        raws = set()
        hashes = set()
        prefixes = set()

        for _ in range(1000):
            raw, prefix, key_hash = generate_api_key()
            raws.add(raw)
            hashes.add(key_hash)
            prefixes.add(prefix)

        assert len(raws) == 1000   # all raws unique
        assert len(hashes) == 1000 # all hashes unique
        assert len(prefixes) <= 1000  # prefixes may collide (only 7 chars)
    # #endregion test_api_key_hash_uniqueness

    # #region test_hash_api_key_consistency [C:2] [TYPE Function]
    # @BRIEF hash_api_key is deterministic — same input = same output.
    def test_hash_api_key_consistency(self):
        from src.core.auth.api_key import hash_api_key

        key = "ssk_test-key-for-consistency-check-123"
        h1 = hash_api_key(key)
        h2 = hash_api_key(key)
        assert h1 == h2
        assert len(h1) == 64
    # #endregion test_hash_api_key_consistency


# ──────────────────────────────────────────────
# ENCRYPTION: Fernet robustness
# ──────────────────────────────────────────────


class TestEncryptionRobustness:
    """Orthogonal tests for EncryptionManager — error handling on invalid data."""

    @pytest.fixture(autouse=True)
    def _setup_key(self):
        """Ensure ENCRYPTION_KEY is set."""
        if not os.getenv("ENCRYPTION_KEY"):
            from cryptography.fernet import Fernet
            os.environ["ENCRYPTION_KEY"] = Fernet.generate_key().decode()

    # #region test_encrypt_decrypt_empty [C:2] [TYPE Function]
    # @BRIEF Empty string should encrypt and decrypt back to empty string.
    def test_encrypt_decrypt_empty(self):
        from src.core.encryption import EncryptionManager

        mgr = EncryptionManager()
        encrypted = mgr.encrypt("")
        decrypted = mgr.decrypt(encrypted)
        assert decrypted == ""
    # #endregion test_encrypt_decrypt_empty

    # #region test_decrypt_invalid_data [C:2] [TYPE Function]
    # @BRIEF Decrypting garbage, wrong key, truncated data must raise.
    def test_decrypt_invalid_data(self):
        from src.core.encryption import EncryptionManager

        mgr = EncryptionManager()

        # Garbage data
        with pytest.raises(Exception):
            mgr.decrypt("not-encrypted-data")

        # Empty string
        with pytest.raises(Exception):
            mgr.decrypt("")

        # Truncated valid token
        valid = mgr.encrypt("test")
        with pytest.raises(Exception):
            mgr.decrypt(valid[:-10])
    # #endregion test_decrypt_invalid_data

    # #region test_encrypt_decrypt_large_data [C:2] [TYPE Function]
    # @BRIEF Large strings (1MB) should encrypt/decrypt without error.
    def test_encrypt_decrypt_large_data(self):
        from src.core.encryption import EncryptionManager

        mgr = EncryptionManager()
        large = "x" * (1024 * 1024)  # 1MB
        encrypted = mgr.encrypt(large)
        decrypted = mgr.decrypt(encrypted)
        assert decrypted == large
    # #endregion test_encrypt_decrypt_large_data


# ──────────────────────────────────────────────
# ENCRYPTION KEY: lifecycle paths
# ──────────────────────────────────────────────


class TestEncryptionKeyLifecycle:
    """Orthogonal tests for ensure_encryption_key — resolution strategy."""

    # #region test_ensure_encryption_key_from_env [C:2] [TYPE Function]
    # @BRIEF When ENCRYPTION_KEY is set in environment, return it without side effects.
    def test_ensure_encryption_key_from_env(self):
        from src.core.encryption_key import ensure_encryption_key
        from cryptography.fernet import Fernet

        expected = Fernet.generate_key().decode()
        with patch.dict(os.environ, {"ENCRYPTION_KEY": expected}, clear=False):
            # Ensure no .env file is accessed — use a non-existent path
            result = ensure_encryption_key(Path("/nonexistent/.env"))
            assert result == expected
    # #endregion test_ensure_encryption_key_from_env

    # #region test_ensure_encryption_key_creates_file [C:2] [TYPE Function]
    # @BRIEF When neither env nor .env has a key, generate one and persist to .env.
    def test_ensure_encryption_key_creates_file(self):
        from src.core.encryption_key import ensure_encryption_key

        with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f:
            env_path = Path(f.name)

        try:
            # Remove any existing ENCRYPTION_KEY from env
            old = os.environ.pop("ENCRYPTION_KEY", None)
            try:
                result = ensure_encryption_key(env_path)
                assert len(result) > 0

                # Key should be in the file
                content = env_path.read_text()
                assert "ENCRYPTION_KEY=" in content
                assert result in content
            finally:
                if old is not None:
                    os.environ["ENCRYPTION_KEY"] = old
        finally:
            env_path.unlink(missing_ok=True)
    # #endregion test_ensure_encryption_key_creates_file

    # #region test_ensure_encryption_key_loads_from_file [C:2] [TYPE Function]
    # @BRIEF When env is empty but .env file has the key, load it.
    def test_ensure_encryption_key_loads_from_file(self):
        from src.core.encryption_key import ensure_encryption_key
        from cryptography.fernet import Fernet

        key = Fernet.generate_key().decode()
        old = os.environ.pop("ENCRYPTION_KEY", None)

        try:
            with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f:
                f.write(f"ENCRYPTION_KEY={key}\n")
                env_path = Path(f.name)

            try:
                result = ensure_encryption_key(env_path)
                assert result == key
                assert os.environ["ENCRYPTION_KEY"] == key
            finally:
                env_path.unlink(missing_ok=True)
        finally:
            if old is not None:
                os.environ["ENCRYPTION_KEY"] = old
    # #endregion test_ensure_encryption_key_loads_from_file


# ──────────────────────────────────────────────
# AUTH LOGGER: log injection protection
# ──────────────────────────────────────────────


class TestLogInjectionProtection:
    """Orthogonal tests for auth logger — verify CRLF/control chars don't forge logs."""

    # #region test_log_security_event_injection [C:2] [TYPE Function]
    # @BRIEF CRLF in username or event_type must not create forged log entries.
    #   Our fix uses %s-formatting which prevents CRLF injection.
    def test_log_security_event_injection(self):
        from src.core.auth.logger import log_security_event

        # Username with embedded newline — this would forge a log entry with plain f-strings
        malicious = "admin\n[AUDIT][FAKE] User: attacker"
        # Should not raise, should not create a fake entry
        log_security_event("LOGIN_SUCCESS", malicious, {"source": "test"})
        # (We verify by checking that no exception occurs — actual log inspection
        #  would require a log handler fixture)
    # #endregion test_log_security_event_injection

    # #region test_log_security_event_long_input [C:2] [TYPE Function]
    # @BRIEF Very long username/event_type must not crash the logger.
    def test_log_security_event_long_input(self):
        from src.core.auth.logger import log_security_event

        long_username = "u" * 10000
        log_security_event("TEST_EVENT", long_username, {"detail": "x" * 10000})
    # #endregion test_log_security_event_long_input

    # #region test_log_security_event_special_chars [C:2] [TYPE Function]
    # @BRIEF Unicode, control chars, emoji in username must not break logging.
    def test_log_security_event_special_chars(self):
        from src.core.auth.logger import log_security_event

        log_security_event("LOGIN_SUCCESS", "user@domain.com\u0000null_bytes")
        log_security_event("LOGIN_SUCCESS", "user\u0001\u0002\u001fcontrol_chars")
        log_security_event("LOGIN_SUCCESS", "user🔥emoji")
        log_security_event("LOGIN_SUCCESS", "пользователь_кириллица")
    # #endregion test_log_security_event_special_chars


# ──────────────────────────────────────────────
# DEPENDENCIES: JWT edge cases
# ──────────────────────────────────────────────


class TestJwtEdgeCases:
    """Orthogonal tests for JWT token handling — edge cases in decode/validation."""

    # #region test_decode_expired_token [C:2] [TYPE Function]
    # @BRIEF Token with past expiration must raise JWTError.
    def test_decode_expired_token(self):
        from datetime import timedelta
        from jose import jwt, JWTError
        from src.core.auth.config import auth_config

        # Create an already-expired token
        import time
        payload = {"sub": "testuser", "exp": int(time.time()) - 3600}  # 1 hour ago
        token = jwt.encode(payload, auth_config.SECRET_KEY, algorithm=auth_config.ALGORITHM)

        from src.core.auth.jwt import decode_token
        with pytest.raises(JWTError):
            decode_token(token)
    # #endregion test_decode_expired_token

    # #region test_decode_malformed_token [C:2] [TYPE Function]
    # @BRIEF Malformed JWT (wrong format, wrong key) must raise JWTError.
    def test_decode_malformed_token(self):
        from jose import JWTError
        from src.core.auth.jwt import decode_token

        with pytest.raises(JWTError):
            decode_token("not.a.token")

        with pytest.raises(JWTError):
            decode_token("")

        # Token with wrong algorithm
        from jose import jwt
        token = jwt.encode({"sub": "test"}, "wrong-key", algorithm="HS256")
        with pytest.raises(JWTError):
            decode_token(token)
    # #endregion test_decode_malformed_token

    # #region test_decode_token_missing_sub [C:2] [TYPE Function]
    # @BRIEF Token without 'sub' claim should decode but downstream should handle.
    def test_decode_token_missing_sub(self):
        from src.core.auth.jwt import create_access_token, decode_token

        token = create_access_token(data={"role": "admin"})  # no 'sub'
        payload = decode_token(token)
        assert "sub" not in payload
        assert payload.get("role") == "admin"
    # #endregion test_decode_token_missing_sub


# #endregion TestSecurityOrthogonal