Files
ss-tools/backend/tests/test_security_orthogonal.py

538 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# #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
from pathlib import Path
import pytest
import tempfile
from unittest.mock import patch
# ──────────────────────────────────────────────
# 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
@pytest.mark.skip(reason="bcrypt ≥4.1 raises ValueError on >72 bytes instead of silently truncating — test needs password[:72] in production code")
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."""
import unicodedata
from src.core.auth.security import get_password_hash, verify_password
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 get_password_hash, verify_password
# 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
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 cryptography.fernet import Fernet
from src.core.encryption_key import ensure_encryption_key
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_missing_raises [C:2] [TYPE Function]
# @BRIEF When neither env nor .env has a key, crash-early with RuntimeError.
# @TEST_EDGE: missing_encryption_key -> RuntimeError with clear message
def test_ensure_encryption_key_missing_raises(self):
from src.core.encryption_key import ensure_encryption_key
with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f:
f.write("# empty .env\n")
env_path = Path(f.name)
try:
old = os.environ.pop("ENCRYPTION_KEY", None)
try:
with pytest.raises(RuntimeError, match="ENCRYPTION_KEY"):
ensure_encryption_key(env_path)
finally:
if old is not None:
os.environ["ENCRYPTION_KEY"] = old
finally:
env_path.unlink(missing_ok=True)
# #endregion test_ensure_encryption_key_missing_raises
# #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 cryptography.fernet import Fernet
from src.core.encryption_key import ensure_encryption_key
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):
# Create an already-expired token
import time
from jose import JWTError, jwt
from src.core.auth.config import auth_config
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
# ──────────────────────────────────────────────
# RATE LIMITER
# ──────────────────────────────────────────────
class TestRateLimiter:
"""Orthogonal tests for in-memory rate limiter."""
# #region test_rate_limiter_allows_under_limit [C:2] [TYPE Function]
# @BRIEF Under limit: not banned, attempts recorded.
def test_rate_limiter_allows_under_limit(self):
from src.core.rate_limiter import RateLimiter
rl = RateLimiter()
assert not rl.is_banned("1.2.3.4")
for _ in range(5):
rl.record_attempt("1.2.3.4")
assert not rl.is_banned("1.2.3.4")
# #endregion test_rate_limiter_allows_under_limit
# #region test_rate_limiter_bans_over_limit [C:2] [TYPE Function]
# @BRIEF Over limit: IP is banned, success clears history.
def test_rate_limiter_bans_over_limit(self):
from src.core.rate_limiter import MAX_ATTEMPTS, RateLimiter
rl = RateLimiter()
ip = "5.6.7.8"
for _ in range(MAX_ATTEMPTS + 1):
rl.record_attempt(ip)
assert rl.is_banned(ip)
# After ban, success should clear (but may still be banned)
rl.record_success(ip)
# #endregion test_rate_limiter_bans_over_limit
# #region test_rate_limiter_success_clears_history [C:2] [TYPE Function]
# @BRIEF Successful auth clears attempt history for that IP.
def test_rate_limiter_success_clears_history(self):
from src.core.rate_limiter import RateLimiter
rl = RateLimiter()
rl.record_attempt("9.9.9.9")
rl.record_attempt("9.9.9.9")
rl.record_success("9.9.9.9")
# After success, the attempts dict is cleared — we can make more without ban
for _ in range(5):
rl.record_attempt("9.9.9.9")
assert not rl.is_banned("9.9.9.9")
# #endregion test_rate_limiter_success_clears_history
# #region test_rate_limiter_different_ips_independent [C:2] [TYPE Function]
# @BRIEF Different IPs are tracked independently.
def test_rate_limiter_different_ips_independent(self):
from src.core.rate_limiter import MAX_ATTEMPTS, RateLimiter
rl = RateLimiter()
# One IP gets banned
for _ in range(MAX_ATTEMPTS + 1):
rl.record_attempt("bad-ip")
assert rl.is_banned("bad-ip")
# Other IP is not affected
assert not rl.is_banned("good-ip")
# #endregion test_rate_limiter_different_ips_independent
# ──────────────────────────────────────────────
# PASSWORD POLICY
# ──────────────────────────────────────────────
class TestPasswordPolicy:
"""Orthogonal tests for password strength validation."""
# #region test_password_too_short [C:2] [TYPE Function]
def test_password_too_short(self):
from pydantic import ValidationError
from src.schemas.auth import UserCreate
with pytest.raises(ValidationError):
UserCreate(username="test", password="Ab1") # only 3 chars
# #endregion test_password_too_short
# #region test_password_no_uppercase [C:2] [TYPE Function]
def test_password_no_uppercase(self):
from pydantic import ValidationError
from src.schemas.auth import UserCreate
with pytest.raises(ValidationError):
UserCreate(username="test", password="abcdefgh1") # no uppercase
# #endregion test_password_no_uppercase
# #region test_password_no_digit [C:2] [TYPE Function]
def test_password_no_digit(self):
from pydantic import ValidationError
from src.schemas.auth import UserCreate
with pytest.raises(ValidationError):
UserCreate(username="test", password="Abcdefgh") # no digit
# #endregion test_password_no_digit
# #region test_password_valid [C:2] [TYPE Function]
def test_password_valid(self):
from src.schemas.auth import UserCreate
u = UserCreate(username="test", password="ValidPass1")
assert u.password == "ValidPass1"
# #endregion test_password_valid
# ──────────────────────────────────────────────
# LOG MASKING
# ──────────────────────────────────────────────
class TestLogMasking:
"""Orthogonal tests for sensitive field masking in log_security_event."""
# #region test_mask_sensitive_fields [C:2] [TYPE Function]
def test_mask_sensitive_fields(self):
from src.core.auth.logger import _mask_details
masked = _mask_details({
"username": "john",
"password": "supersecret",
"token": "eyJhbGciOiJIUzI1NiJ9.xxx",
"api_key": "ssk_abc123",
"source": "LOCAL",
})
assert masked["username"] == "john"
assert masked["password"] == "***"
assert masked["token"] == "***"
assert masked["api_key"] == "***"
assert masked["source"] == "LOCAL"
# #endregion test_mask_sensitive_fields
# #region test_mask_nested_details [C:2] [TYPE Function]
def test_mask_nested_details(self):
from src.core.auth.logger import _mask_details
masked = _mask_details({
"user": {"username": "john", "password": "secret"},
})
assert masked["user"]["username"] == "john"
assert masked["user"]["password"] == "***"
# #endregion test_mask_nested_details
# #endregion TestSecurityOrthogonal