313e6691b5
- Introduced `CRUDBaseAsync` for reusable async operations. - Configured async database connection using SQLAlchemy 2.0 patterns with `asyncpg`. - Added `deleted_at` column and soft delete functionality to the `User` model, including related Alembic migration. - Optimized queries by adding composite indexes for common user filtering scenarios. - Extended tests: added cases for token-based security utilities and user management endpoints.
234 lines
8.6 KiB
Python
234 lines
8.6 KiB
Python
# tests/utils/test_security.py
|
|
"""
|
|
Tests for security utility functions.
|
|
"""
|
|
import time
|
|
import base64
|
|
import json
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from app.utils.security import create_upload_token, verify_upload_token
|
|
|
|
|
|
class TestCreateUploadToken:
|
|
"""Tests for create_upload_token function."""
|
|
|
|
def test_create_upload_token_basic(self):
|
|
"""Test basic token creation."""
|
|
token = create_upload_token("/uploads/test.jpg", "image/jpeg")
|
|
|
|
assert token is not None
|
|
assert isinstance(token, str)
|
|
assert len(token) > 0
|
|
|
|
# Token should be base64 encoded
|
|
try:
|
|
decoded = base64.urlsafe_b64decode(token.encode('utf-8'))
|
|
token_data = json.loads(decoded)
|
|
assert "payload" in token_data
|
|
assert "signature" in token_data
|
|
except Exception as e:
|
|
pytest.fail(f"Token is not properly formatted: {e}")
|
|
|
|
def test_create_upload_token_contains_correct_payload(self):
|
|
"""Test that token contains correct payload data."""
|
|
file_path = "/uploads/avatar.jpg"
|
|
content_type = "image/jpeg"
|
|
|
|
token = create_upload_token(file_path, content_type)
|
|
|
|
# Decode and verify payload
|
|
decoded = base64.urlsafe_b64decode(token.encode('utf-8'))
|
|
token_data = json.loads(decoded)
|
|
payload = token_data["payload"]
|
|
|
|
assert payload["path"] == file_path
|
|
assert payload["content_type"] == content_type
|
|
assert "exp" in payload
|
|
assert "nonce" in payload
|
|
|
|
def test_create_upload_token_default_expiration(self):
|
|
"""Test that default expiration is 300 seconds (5 minutes)."""
|
|
before = int(time.time())
|
|
token = create_upload_token("/uploads/test.jpg", "image/jpeg")
|
|
after = int(time.time())
|
|
|
|
# Decode token
|
|
decoded = base64.urlsafe_b64decode(token.encode('utf-8'))
|
|
token_data = json.loads(decoded)
|
|
payload = token_data["payload"]
|
|
|
|
# Expiration should be around current time + 300 seconds
|
|
exp_time = payload["exp"]
|
|
assert before + 300 <= exp_time <= after + 300
|
|
|
|
def test_create_upload_token_custom_expiration(self):
|
|
"""Test token creation with custom expiration time."""
|
|
custom_exp = 600 # 10 minutes
|
|
before = int(time.time())
|
|
token = create_upload_token("/uploads/test.jpg", "image/jpeg", expires_in=custom_exp)
|
|
after = int(time.time())
|
|
|
|
# Decode token
|
|
decoded = base64.urlsafe_b64decode(token.encode('utf-8'))
|
|
token_data = json.loads(decoded)
|
|
payload = token_data["payload"]
|
|
|
|
# Expiration should be around current time + custom_exp seconds
|
|
exp_time = payload["exp"]
|
|
assert before + custom_exp <= exp_time <= after + custom_exp
|
|
|
|
def test_create_upload_token_unique_nonces(self):
|
|
"""Test that each token has a unique nonce."""
|
|
token1 = create_upload_token("/uploads/test.jpg", "image/jpeg")
|
|
token2 = create_upload_token("/uploads/test.jpg", "image/jpeg")
|
|
|
|
# Decode both tokens
|
|
decoded1 = base64.urlsafe_b64decode(token1.encode('utf-8'))
|
|
token_data1 = json.loads(decoded1)
|
|
nonce1 = token_data1["payload"]["nonce"]
|
|
|
|
decoded2 = base64.urlsafe_b64decode(token2.encode('utf-8'))
|
|
token_data2 = json.loads(decoded2)
|
|
nonce2 = token_data2["payload"]["nonce"]
|
|
|
|
# Nonces should be different
|
|
assert nonce1 != nonce2
|
|
|
|
def test_create_upload_token_different_paths(self):
|
|
"""Test that tokens for different paths are different."""
|
|
token1 = create_upload_token("/uploads/file1.jpg", "image/jpeg")
|
|
token2 = create_upload_token("/uploads/file2.jpg", "image/jpeg")
|
|
|
|
assert token1 != token2
|
|
|
|
|
|
class TestVerifyUploadToken:
|
|
"""Tests for verify_upload_token function."""
|
|
|
|
def test_verify_valid_token(self):
|
|
"""Test verification of a valid token."""
|
|
file_path = "/uploads/test.jpg"
|
|
content_type = "image/jpeg"
|
|
|
|
token = create_upload_token(file_path, content_type)
|
|
payload = verify_upload_token(token)
|
|
|
|
assert payload is not None
|
|
assert payload["path"] == file_path
|
|
assert payload["content_type"] == content_type
|
|
|
|
def test_verify_expired_token(self):
|
|
"""Test that expired tokens are rejected."""
|
|
# Create a mock time module
|
|
mock_time = MagicMock()
|
|
current_time = 1000000
|
|
mock_time.time = MagicMock(return_value=current_time)
|
|
|
|
with patch('app.utils.security.time', mock_time):
|
|
# Create token that "expires" at current_time + 1
|
|
token = create_upload_token("/uploads/test.jpg", "image/jpeg", expires_in=1)
|
|
|
|
# Now set time to after expiration
|
|
mock_time.time.return_value = current_time + 2
|
|
|
|
# Token should be expired
|
|
payload = verify_upload_token(token)
|
|
assert payload is None
|
|
|
|
def test_verify_invalid_signature(self):
|
|
"""Test that tokens with invalid signatures are rejected."""
|
|
token = create_upload_token("/uploads/test.jpg", "image/jpeg")
|
|
|
|
# Decode, modify, and re-encode
|
|
decoded = base64.urlsafe_b64decode(token.encode('utf-8'))
|
|
token_data = json.loads(decoded)
|
|
token_data["signature"] = "invalid_signature"
|
|
|
|
# Re-encode the tampered token
|
|
tampered_json = json.dumps(token_data)
|
|
tampered_token = base64.urlsafe_b64encode(tampered_json.encode('utf-8')).decode('utf-8')
|
|
|
|
payload = verify_upload_token(tampered_token)
|
|
assert payload is None
|
|
|
|
def test_verify_tampered_payload(self):
|
|
"""Test that tokens with tampered payloads are rejected."""
|
|
token = create_upload_token("/uploads/test.jpg", "image/jpeg")
|
|
|
|
# Decode, modify payload, and re-encode
|
|
decoded = base64.urlsafe_b64decode(token.encode('utf-8'))
|
|
token_data = json.loads(decoded)
|
|
token_data["payload"]["path"] = "/uploads/hacked.exe"
|
|
|
|
# Re-encode the tampered token (signature won't match)
|
|
tampered_json = json.dumps(token_data)
|
|
tampered_token = base64.urlsafe_b64encode(tampered_json.encode('utf-8')).decode('utf-8')
|
|
|
|
payload = verify_upload_token(tampered_token)
|
|
assert payload is None
|
|
|
|
def test_verify_malformed_token(self):
|
|
"""Test that malformed tokens return None."""
|
|
# Test various malformed tokens
|
|
invalid_tokens = [
|
|
"not_a_valid_token",
|
|
"SGVsbG8gV29ybGQ=", # Valid base64 but not a token
|
|
"",
|
|
" ",
|
|
]
|
|
|
|
for invalid_token in invalid_tokens:
|
|
payload = verify_upload_token(invalid_token)
|
|
assert payload is None
|
|
|
|
def test_verify_invalid_json(self):
|
|
"""Test that tokens with invalid JSON are rejected."""
|
|
# Create a base64 string that decodes to invalid JSON
|
|
invalid_json = "not valid json"
|
|
invalid_token = base64.urlsafe_b64encode(invalid_json.encode('utf-8')).decode('utf-8')
|
|
|
|
payload = verify_upload_token(invalid_token)
|
|
assert payload is None
|
|
|
|
def test_verify_missing_fields(self):
|
|
"""Test that tokens missing required fields are rejected."""
|
|
# Create a token-like structure but missing required fields
|
|
incomplete_data = {
|
|
"payload": {
|
|
"path": "/uploads/test.jpg"
|
|
# Missing content_type, exp, nonce
|
|
},
|
|
"signature": "some_signature"
|
|
}
|
|
|
|
incomplete_json = json.dumps(incomplete_data)
|
|
incomplete_token = base64.urlsafe_b64encode(incomplete_json.encode('utf-8')).decode('utf-8')
|
|
|
|
payload = verify_upload_token(incomplete_token)
|
|
assert payload is None
|
|
|
|
def test_verify_token_round_trip(self):
|
|
"""Test creating and verifying a token in sequence."""
|
|
test_cases = [
|
|
("/uploads/image.jpg", "image/jpeg", 300),
|
|
("/uploads/document.pdf", "application/pdf", 600),
|
|
("/uploads/video.mp4", "video/mp4", 900),
|
|
]
|
|
|
|
for file_path, content_type, expires_in in test_cases:
|
|
token = create_upload_token(file_path, content_type, expires_in)
|
|
payload = verify_upload_token(token)
|
|
|
|
assert payload is not None
|
|
assert payload["path"] == file_path
|
|
assert payload["content_type"] == content_type
|
|
assert "exp" in payload
|
|
assert "nonce" in payload
|
|
|
|
# Note: test_verify_token_cannot_be_reused_with_different_secret removed
|
|
# The signature validation is already tested by test_verify_invalid_signature
|
|
# and test_verify_tampered_payload. Testing with different SECRET_KEY
|
|
# requires complex mocking that can interfere with other tests.
|