jcv-dev
551b42eab5
Close the gaps between the documented contract (API-STANDARD, TENANCY, ENTITLEMENTS) and the implementation, and remove committed build artifacts. Security: - verify + process inbound webhooks (HMAC/handler verify, size limit, timestamp tolerance, idempotency via InboundEvent.external_id) - real 2FA login flow (pre-auth challenge; tokens only after verify/recovery) - wire HybridPermission into security/audit views; add API-key rotate and users/<id>/permissions|roles endpoints - tenant-scoped throttling on by default; webhook replay protection - verify MercadoPago webhook signatures - login brute-force guard, configurable password policy, real ALTCHA PoW Correctness: - apply verified billing webhooks idempotently (subscription/entitlement/ invoice/PaymentTransaction); scheduled payment lifecycle jobs - capture audit update diffs automatically; add audit retention purge - working notification retries, per-channel rate limits, log retention - pluggable virus scanner, upload-size limit, pipeline toggle - feature rollout %/environment targeting; settings-driven registrations - workflow guards (instance cap, route depth, self-assignment, clone on re-entry) - wire every previously-dead INFRASYNTH_* setting; drop truly dead ones Delivery: - README + CHANGELOG; CI format check + coverage gate - keep test media out of the tree; untrack .coverage, __pycache__, egg-info, docs/ and invoice artifacts
109 lines
4.6 KiB
Python
109 lines
4.6 KiB
Python
from datetime import timedelta
|
|
|
|
from django.utils import timezone
|
|
|
|
from infrasynth.security.models import TwoFactorConfig
|
|
from infrasynth.security.two_factor.middleware import TwoFactorMiddleware
|
|
from infrasynth.security.two_factor.utils import (
|
|
generate_pre_auth_token,
|
|
get_pre_auth_expiry,
|
|
)
|
|
|
|
|
|
class TestPreAuthUtils:
|
|
def test_generate_pre_auth_token_length(self):
|
|
token = generate_pre_auth_token()
|
|
assert len(token) == 64
|
|
|
|
def test_generate_pre_auth_token_unique(self):
|
|
assert generate_pre_auth_token() != generate_pre_auth_token()
|
|
|
|
def test_get_pre_auth_expiry(self):
|
|
before = timezone.now()
|
|
expiry = get_pre_auth_expiry(minutes=5)
|
|
after = timezone.now()
|
|
assert before + timedelta(minutes=5) <= expiry <= after + timedelta(minutes=5)
|
|
|
|
|
|
class _Request:
|
|
def __init__(self, user, path="/api/v1/features/active/"):
|
|
self.user = user
|
|
self.path = path
|
|
|
|
|
|
class TestTwoFactorMiddleware:
|
|
def test_anonymous_passes(self, rf, db):
|
|
request = rf.get("/api/v1/features/active/")
|
|
request.user = type("Anon", (), {"is_authenticated": False})()
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|
|
|
|
def test_user_without_config_passes(self, rf, user):
|
|
request = rf.get("/api/v1/features/active/")
|
|
request.user = user
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|
|
|
|
def test_disabled_config_passes(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=False, is_configured=True)
|
|
request = rf.get("/api/v1/features/active/")
|
|
request.user = user
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|
|
|
|
def test_unconfigured_enabled_passes(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=False)
|
|
request = rf.get("/api/v1/features/active/")
|
|
request.user = user
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|
|
|
|
def test_api_path_returns_403_when_2fa_required(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True)
|
|
request = rf.get("/api/v1/features/active/")
|
|
request.user = user
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is not None
|
|
assert result.status_code == 403
|
|
import json as _json
|
|
|
|
assert _json.loads(result.content)["code"] == "AUTH_2FA_REQUIRED"
|
|
|
|
def test_non_api_path_redirects(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True)
|
|
request = rf.get("/some-page/")
|
|
request.user = user
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is not None
|
|
assert result.status_code == 302
|
|
assert result.url == "/api/v1/auth/2fa/verify/"
|
|
|
|
def test_jwt_authenticated_request_passes(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True)
|
|
request = rf.get("/api/v1/features/active/")
|
|
request.user = user
|
|
request.auth = object() # a minted token implies 2FA already happened
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|
|
|
|
def test_exempt_paths_not_redirected(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True)
|
|
request = rf.get("/api/v1/auth/check/")
|
|
request.user = user
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|
|
|
|
def test_admin_path_not_redirected(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True)
|
|
request = rf.get("/admin/")
|
|
request.user = user
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|
|
|
|
def test_verified_session_not_blocked(self, rf, user):
|
|
TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True)
|
|
request = rf.get("/api/v1/features/active/")
|
|
request.user = user
|
|
request.session = {"_2fa_verified": True}
|
|
result = TwoFactorMiddleware(get_response=lambda r: None).process_request(request)
|
|
assert result is None
|