Gan, Jimmy
7fb92ccebf
Ensure auth file is properly initialized before testing replay protection. Add better error message to diagnose failures.
343 lines
12 KiB
Python
343 lines
12 KiB
Python
"""
|
|
Unit tests for auth.py - JWT, TOTP, password hashing, encryption.
|
|
"""
|
|
import pytest
|
|
import jwt
|
|
import pyotp
|
|
from datetime import datetime, timedelta, timezone
|
|
from auth_service import (
|
|
verify_password,
|
|
get_password_hash,
|
|
create_access_token,
|
|
create_refresh_token,
|
|
user_from_access_token,
|
|
verify_totp,
|
|
load_totp_secret,
|
|
save_totp_secret,
|
|
load_password_hash,
|
|
save_password_hash,
|
|
increment_token_version,
|
|
verify_token_version,
|
|
_encrypt,
|
|
_decrypt,
|
|
load_passkey_credentials,
|
|
save_passkey_credential,
|
|
clear_passkey_credentials,
|
|
)
|
|
from fastapi import HTTPException
|
|
|
|
|
|
class TestPasswordHashing:
|
|
"""Test password hashing and verification."""
|
|
|
|
def test_hash_and_verify_password(self):
|
|
"""Test that password hashing and verification works."""
|
|
password = "test_password_123"
|
|
hashed = get_password_hash(password)
|
|
|
|
assert hashed != password
|
|
assert verify_password(password, hashed)
|
|
|
|
def test_verify_wrong_password(self):
|
|
"""Test that wrong password fails verification."""
|
|
password = "correct_password"
|
|
wrong_password = "wrong_password"
|
|
hashed = get_password_hash(password)
|
|
|
|
assert not verify_password(wrong_password, hashed)
|
|
|
|
def test_different_hashes_for_same_password(self):
|
|
"""Test that same password produces different hashes (salt)."""
|
|
password = "test_password"
|
|
hash1 = get_password_hash(password)
|
|
hash2 = get_password_hash(password)
|
|
|
|
assert hash1 != hash2
|
|
assert verify_password(password, hash1)
|
|
assert verify_password(password, hash2)
|
|
|
|
|
|
class TestJWTTokens:
|
|
"""Test JWT token creation and validation."""
|
|
|
|
def test_create_access_token(self, mock_config):
|
|
"""Test access token creation."""
|
|
data = {"sub": "testuser", "role": "admin"}
|
|
token = create_access_token(data)
|
|
|
|
assert token is not None
|
|
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
|
assert payload["sub"] == "testuser"
|
|
assert payload["role"] == "admin"
|
|
assert payload["type"] == "access"
|
|
assert "exp" in payload
|
|
|
|
def test_create_access_token_with_custom_expiry(self, mock_config):
|
|
"""Test access token with custom expiration."""
|
|
data = {"sub": "testuser"}
|
|
expires_delta = timedelta(minutes=60)
|
|
token = create_access_token(data, expires_delta)
|
|
|
|
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
|
exp_time = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
# Should expire in approximately 60 minutes
|
|
time_diff = (exp_time - now).total_seconds()
|
|
assert 3500 < time_diff < 3700 # ~60 minutes with some tolerance
|
|
|
|
def test_create_refresh_token(self, mock_config, temp_auth_file):
|
|
"""Test refresh token creation."""
|
|
data = {"sub": "testuser", "role": "member"}
|
|
token = create_refresh_token(data)
|
|
|
|
assert token is not None
|
|
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
|
assert payload["sub"] == "testuser"
|
|
assert payload["role"] == "member"
|
|
assert payload["type"] == "refresh"
|
|
assert "tv" in payload # token version
|
|
assert "exp" in payload
|
|
|
|
def test_user_from_valid_access_token(self, mock_config, temp_rbac_file):
|
|
"""Test extracting user from valid access token."""
|
|
token = create_access_token({"sub": "testuser", "role": "admin"})
|
|
user = user_from_access_token(token)
|
|
|
|
assert user.username == "testuser"
|
|
assert user.role == "admin"
|
|
assert user.pages == "*"
|
|
assert user.sidebar_links == "*"
|
|
|
|
def test_user_from_expired_token(self, mock_config, temp_rbac_file):
|
|
"""Test that expired token raises HTTPException."""
|
|
token = create_access_token(
|
|
{"sub": "testuser", "role": "admin"},
|
|
expires_delta=timedelta(seconds=-1)
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
user_from_access_token(token)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
|
|
def test_user_from_invalid_token(self, mock_config, temp_rbac_file):
|
|
"""Test that invalid token raises HTTPException."""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
user_from_access_token("invalid.token.here")
|
|
|
|
assert exc_info.value.status_code == 401
|
|
|
|
def test_user_from_refresh_token_fails(self, mock_config, temp_rbac_file, temp_auth_file):
|
|
"""Test that refresh token cannot be used as access token."""
|
|
token = create_refresh_token({"sub": "testuser", "role": "admin"})
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
user_from_access_token(token)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
|
|
def test_user_from_token_without_username(self, mock_config, temp_rbac_file):
|
|
"""Test that token without username fails."""
|
|
# Create token without 'sub' field
|
|
token = jwt.encode(
|
|
{"role": "admin", "type": "access", "exp": datetime.now(timezone.utc) + timedelta(minutes=30)},
|
|
mock_config.SECRET_KEY,
|
|
algorithm=mock_config.ALGORITHM
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
user_from_access_token(token)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
|
|
|
|
class TestEncryption:
|
|
"""Test encryption and decryption functions."""
|
|
|
|
def test_encrypt_decrypt(self, mock_config):
|
|
"""Test basic encryption and decryption."""
|
|
plaintext = "my-secret-totp-key"
|
|
encrypted = _encrypt(plaintext)
|
|
|
|
assert encrypted != plaintext
|
|
assert encrypted != ""
|
|
|
|
decrypted = _decrypt(encrypted)
|
|
assert decrypted == plaintext
|
|
|
|
def test_encrypt_empty_string(self, mock_config):
|
|
"""Test encrypting empty string."""
|
|
encrypted = _encrypt("")
|
|
assert encrypted == ""
|
|
|
|
decrypted = _decrypt("")
|
|
assert decrypted == ""
|
|
|
|
def test_decrypt_invalid_ciphertext(self, mock_config):
|
|
"""Test decrypting invalid ciphertext returns original."""
|
|
invalid = "not-a-valid-encrypted-string"
|
|
decrypted = _decrypt(invalid)
|
|
|
|
# Should fallback to returning the original string
|
|
assert decrypted == invalid
|
|
|
|
def test_encrypt_decrypt_unicode(self, mock_config):
|
|
"""Test encryption with unicode characters."""
|
|
plaintext = "测试密钥🔐"
|
|
encrypted = _encrypt(plaintext)
|
|
decrypted = _decrypt(encrypted)
|
|
|
|
assert decrypted == plaintext
|
|
|
|
|
|
class TestTOTP:
|
|
"""Test TOTP functionality."""
|
|
|
|
def test_save_and_load_totp_secret(self, mock_config, temp_auth_file):
|
|
"""Test saving and loading TOTP secret."""
|
|
secret = "JBSWY3DPEHPK3PXP"
|
|
save_totp_secret(secret)
|
|
|
|
loaded = load_totp_secret()
|
|
assert loaded == secret
|
|
|
|
def test_load_totp_secret_from_env(self, mock_config, temp_auth_file, monkeypatch):
|
|
"""Test loading TOTP secret from environment variable."""
|
|
# Clear any existing secret in file first
|
|
save_totp_secret("")
|
|
|
|
monkeypatch.setenv("TOTP_SECRET", "ENV_SECRET_KEY")
|
|
import importlib
|
|
import config
|
|
importlib.reload(config)
|
|
|
|
# When no secret in file, should return env var
|
|
loaded = load_totp_secret()
|
|
assert loaded == "ENV_SECRET_KEY"
|
|
|
|
def test_verify_totp_valid_code(self, mock_config, temp_auth_file, sample_totp_secret):
|
|
"""Test verifying valid TOTP code."""
|
|
save_totp_secret(sample_totp_secret)
|
|
totp = pyotp.TOTP(sample_totp_secret)
|
|
valid_code = totp.now()
|
|
|
|
assert verify_totp(valid_code)
|
|
|
|
def test_verify_totp_invalid_code(self, mock_config, temp_auth_file, sample_totp_secret):
|
|
"""Test verifying invalid TOTP code."""
|
|
save_totp_secret(sample_totp_secret)
|
|
|
|
assert not verify_totp("000000")
|
|
|
|
def test_verify_totp_no_secret_enabled(self, mock_config, temp_auth_file, monkeypatch):
|
|
"""Test that TOTP verification passes when 2FA not enabled."""
|
|
# Clear any existing secret from previous tests
|
|
save_totp_secret("")
|
|
# Also clear environment variable fallback
|
|
monkeypatch.setenv("TOTP_SECRET", "")
|
|
import importlib
|
|
import config
|
|
importlib.reload(config)
|
|
# No secret saved, should return True (2FA disabled)
|
|
assert verify_totp("any_code")
|
|
|
|
def test_verify_totp_replay_protection(self, mock_config, temp_auth_file, sample_totp_secret):
|
|
"""Test that same TOTP code cannot be used twice."""
|
|
save_totp_secret(sample_totp_secret)
|
|
|
|
# Ensure auth file is properly initialized with last_totp_ts
|
|
import json
|
|
with open(temp_auth_file, 'r') as f:
|
|
data = json.load(f)
|
|
data['last_totp_ts'] = 0
|
|
with open(temp_auth_file, 'w') as f:
|
|
json.dump(data, f)
|
|
|
|
totp = pyotp.TOTP(sample_totp_secret)
|
|
valid_code = totp.now()
|
|
|
|
# First use should succeed
|
|
result = verify_totp(valid_code)
|
|
assert result, f"First TOTP verification failed for code {valid_code}"
|
|
|
|
# Second use of same code should fail (replay protection)
|
|
assert not verify_totp(valid_code)
|
|
|
|
|
|
class TestPasswordPersistence:
|
|
"""Test password hash persistence."""
|
|
|
|
def test_save_and_load_password_hash(self, mock_config, temp_auth_file):
|
|
"""Test saving and loading password hash."""
|
|
hashed = get_password_hash("new_password")
|
|
save_password_hash(hashed)
|
|
|
|
loaded = load_password_hash()
|
|
assert loaded == hashed
|
|
|
|
def test_load_password_hash_from_env(self, mock_config, temp_auth_file):
|
|
"""Test loading password hash from environment variable."""
|
|
# When no hash in file, should return env var
|
|
loaded = load_password_hash()
|
|
assert loaded == mock_config.ADMIN_PASSWORD_HASH
|
|
|
|
|
|
class TestPasskeyCredentials:
|
|
"""Test passkey credential management."""
|
|
|
|
def test_save_and_load_passkey_credentials(self, mock_config, temp_auth_file):
|
|
"""Test saving and loading passkey credentials."""
|
|
cred1 = {"id": "cred1", "public_key": "key1"}
|
|
cred2 = {"id": "cred2", "public_key": "key2"}
|
|
|
|
save_passkey_credential(cred1)
|
|
save_passkey_credential(cred2)
|
|
|
|
creds = load_passkey_credentials()
|
|
assert len(creds) == 2
|
|
assert cred1 in creds
|
|
assert cred2 in creds
|
|
|
|
def test_clear_passkey_credentials(self, mock_config, temp_auth_file):
|
|
"""Test clearing all passkey credentials."""
|
|
save_passkey_credential({"id": "cred1"})
|
|
save_passkey_credential({"id": "cred2"})
|
|
|
|
clear_passkey_credentials()
|
|
|
|
creds = load_passkey_credentials()
|
|
assert len(creds) == 0
|
|
|
|
|
|
class TestTokenVersion:
|
|
"""Test token version management for refresh token rotation."""
|
|
|
|
def test_increment_token_version(self, mock_config, temp_auth_file):
|
|
"""Test incrementing token version."""
|
|
v1 = increment_token_version()
|
|
assert v1 == 1
|
|
|
|
v2 = increment_token_version()
|
|
assert v2 == 2
|
|
|
|
v3 = increment_token_version()
|
|
assert v3 == 3
|
|
|
|
def test_verify_token_version(self, mock_config, temp_auth_file):
|
|
"""Test verifying token version."""
|
|
v1 = increment_token_version()
|
|
assert verify_token_version(v1)
|
|
|
|
v2 = increment_token_version()
|
|
assert verify_token_version(v2)
|
|
assert not verify_token_version(v1) # Old version should fail
|
|
|
|
def test_token_version_in_refresh_token(self, mock_config, temp_auth_file):
|
|
"""Test that refresh token includes current token version."""
|
|
current_version = increment_token_version()
|
|
token = create_refresh_token({"sub": "testuser"})
|
|
|
|
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
|
assert payload["tv"] == current_version
|