From 21731b9887c9d4b954662b3eed6c9d439ac8f1c2 Mon Sep 17 00:00:00 2001 From: jcv-dev Date: Thu, 24 Sep 2026 10:49:44 -0500 Subject: [PATCH] feat(gates): composable per-endpoint gating extension API Make access gating a first-class, pip-consumable extension point so a consuming app can gate any of its own views behind 2FA / ALTCHA / entitlement / feature flag / permission, or gate nothing, without editing the kit. - infrasynth.gates: Gate, GateResult, GatePermission, @gated and built-ins TwoFactorGate, AltchaGate, EntitlementGate, FeatureGate, PermissionGate; denials raise the correct namespaced error/status (per-endpoint, opt-in, default is no gating) - mint a `2fa` JWT claim only after verification (preserved across workspace selection) so TwoFactorGate is meaningful for API/multi-workspace clients - GatePermission added to DEFAULT_PERMISSION_CLASSES; HybridPermission evaluates declared gates so kit permissions gate automatically - document the extension surface and stable import paths in README --- CHANGELOG.md | 10 + PLAN.md | 1 + README.md | 73 +++++ config/settings/base.py | 2 + infrasynth/gates.py | 381 +++++++++++++++++++++++++ infrasynth/security/permissions.py | 4 + infrasynth/security/views.py | 32 ++- tests/test_gates.py | 196 +++++++++++++ tests/test_security/test_two_factor.py | 22 ++ 9 files changed, 710 insertions(+), 11 deletions(-) create mode 100644 infrasynth/gates.py create mode 100644 tests/test_gates.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6afb666..78eb555 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,16 @@ not hand-pick a version (see `../AGENTS.backend-packages.md` §8). ## [Unreleased] ### Added +- **Composable per-endpoint gates** (`infrasynth.gates`): declare + `infrasynth_gates = [...]` (and/or `@gated(...)` on a viewset action) with + `TwoFactorGate`, `AltchaGate`, `EntitlementGate`, `FeatureGate`, + `PermissionGate`, or a custom `Gate`. Access is evaluated per endpoint, the + default is "gate nothing", and denials raise the correct namespaced error + (`AUTH_2FA_REQUIRED`, `ENTITLEMENT_PLAN_UPGRADE_REQUIRED`, + `VALIDATION_ALTCHA_REQUIRED`, …). `GatePermission` is a default permission + class and the kit's `HybridPermission` evaluates declared gates too. +- **`2fa` JWT claim** minted only after successful verification and preserved + across workspace selection, so `TwoFactorGate` works for multi-workspace users. - **Verified inbound webhooks.** `InboundReceiveView` now enforces the shared HMAC signature (or a provider-specific `BaseInboundHandler.verify`), payload size limits, timestamp tolerance, and idempotent re-delivery via diff --git a/PLAN.md b/PLAN.md index 67efcfd..0c5e875 100644 --- a/PLAN.md +++ b/PLAN.md @@ -174,6 +174,7 @@ InfraSynth Base es un conjunto de Django apps reutilizables que proveen la infra - [x] Guardas de workflow (`MAX_INSTANCES_PER_WORKFLOW`, `ROUTE_MAX_DEPTH`, `ALLOW_SELF_ASSIGNMENT`, `AUTO_CLONE_ASSIGNEES_ON_REENTRY`) - [x] Throttling tenant-scoped por defecto; `CELERY_BEAT_SCHEDULE` con trabajos periódicos - [x] README + CHANGELOG; CI con `ruff format --check` y umbral de cobertura +- [x] `infrasynth.gates` — gates por endpoint (`TwoFactorGate`, `AltchaGate`, `EntitlementGate`, `FeatureGate`, `PermissionGate`) declarables sin tocar el kit; `GatePermission` por defecto; claim `2fa` en el JWT ## 1. Estructura del Paquete diff --git a/README.md b/README.md index c7fc61f..f722191 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,79 @@ class SlackChannel(BaseChannel): Registries are populated in `apps.py:ready()`: `FeatureRegistry`, `EventRegistry`, `VariableResolverRegistry`, `DataValidatorRegistry`. +--- + +## Gating your own endpoints (no kit edits) + +`infrasynth.gates` is the composable, per-endpoint access layer. A view declares +exactly what it needs; the default is **gate nothing**: + +```python +from rest_framework.permissions import AllowAny, IsAuthenticated +from infrasynth.gates import ( + GatePermission, gated, + TwoFactorGate, AltchaGate, EntitlementGate, FeatureGate, PermissionGate, +) + +# Public form: proof-of-work, no login. +class SignupView(APIView): + permission_classes = [AllowAny, GatePermission] + infrasynth_gates = [AltchaGate()] + +# Sensitive action: step-up 2FA + commercial right + codename. +class PayoutView(APIView): + permission_classes = [IsAuthenticated, GatePermission] + infrasynth_gates = [ + TwoFactorGate(), # passes users without 2FA; + EntitlementGate("billing", feature="payouts"), # use require_configured=True to demand setup + PermissionGate("billing.payout"), + ] + +class TicketViewSet(ModelViewSet): + permission_classes = [IsAuthenticated, GatePermission] + infrasynth_gates = [FeatureGate("ticketing")] # 404 when the flag is off + + @gated(TwoFactorGate()) + @action(detail=True, methods=["post"]) + def close(self, request, pk=None): ... +``` + +- Gates run in order; the first denial raises the matching namespaced error + (`AUTH_*`, `ENTITLEMENT_*`, `VALIDATION_*`, `NOT_FOUND`) so the envelope gets + the right code and status. No gates declared ⇒ the permission is a no-op. +- `GatePermission` is in `DEFAULT_PERMISSION_CLASSES`; the kit's own + `HybridPermission`/`IsAuthenticatedAndPermitted` also evaluate declared gates, + so you only add it explicitly on views that use plain DRF permissions. +- `AltchaGate` accepts the solution in a JSON `altcha` object, flat body/query + keys, or the `X-Altcha: ::` header; clients get + a challenge from `/api/v1/auth/altcha/challenge/`. +- `TwoFactorGate` verifies a JWT `2fa` claim (minted at verification and carried + across workspace selection) or a verified session. Users without 2FA pass by + default; `require_configured=True` denies them with `AUTH_2FA_SETUP_REQUIRED`. + +Every gate is also a plain class implementing `check(request, view) -> GateResult`, +so an app can ship its own (e.g. an IP allow-list) and pass it to `@gated(...)`. + +--- + +## Stable import paths + +Consumers import from the public modules, never internal helpers: + +| Need | Import | +|---|---| +| Gating | `from infrasynth.gates import GatePermission, AltchaGate, …` | +| Permissions/auth | `from infrasynth.security.permissions import HybridPermission` | +| JWT/API-key auth | `from infrasynth.security.auth.cookies import CookieJWTAuthentication` | +| Authorization | `from infrasynth.security.services import AuthorizationService` | +| Tenant context/scoping | `from infrasynth.tenancy.managers import TenantManager` | +| Entitlements | `from infrasynth.billing.entitlements import EntitlementService` | +| Feature flags | `from infrasynth.features.services import FeatureService` | +| Storage | `from infrasynth.files.storage import get_storage_backend` | +| Errors/envelope | `from infrasynth.shared.exceptions import EntitlementError` | +| Wire format | `from infrasynth.api.renderers import EnvelopeJSONRenderer` | + + --- ## Scheduled work diff --git a/config/settings/base.py b/config/settings/base.py index 4928dc8..ddb696d 100644 --- a/config/settings/base.py +++ b/config/settings/base.py @@ -109,6 +109,7 @@ REST_FRAMEWORK = { ], "DEFAULT_PERMISSION_CLASSES": [ "rest_framework.permissions.IsAuthenticated", + "infrasynth.gates.GatePermission", ], "DEFAULT_RENDERER_CLASSES": [ "infrasynth.api.renderers.EnvelopeJSONRenderer", @@ -191,6 +192,7 @@ INFRASYNTH_SECURITY = { "ALTCHA_DIFFICULTY": 10000, "ALTCHA_CHALLENGE_EXPIRY_SECONDS": 300, "ALTCHA_PROTECT_LOGIN": False, + "ALTCHA_HEADER": "X-Altcha", "API_KEY_PREFIX_LENGTH": 8, "API_KEY_HASH_ALGORITHM": "pbkdf2_sha256", "API_KEY_DEFAULT_EXPIRY_DAYS": 365, diff --git a/infrasynth/gates.py b/infrasynth/gates.py new file mode 100644 index 0000000..e3f08ca --- /dev/null +++ b/infrasynth/gates.py @@ -0,0 +1,381 @@ +"""Composable, per-endpoint access gates. + +This is the kit's **extension surface** for access control. A consuming app +(installed from PyPI) declares what a view needs — and nothing else — without +editing the kit:: + + from infrasynth.gates import ( + GatePermission, gated, + TwoFactorGate, AltchaGate, EntitlementGate, FeatureGate, PermissionGate, + ) + + class PublicSignupView(APIView): + permission_classes = [AllowAny, GatePermission] + infrasynth_gates = [AltchaGate()] # proven human, no login + + class PayoutView(APIView): + permission_classes = [IsAuthenticated, GatePermission] + infrasynth_gates = [ + TwoFactorGate(), # step-up second factor + EntitlementGate("billing", feature="payouts"), + PermissionGate("billing.payout"), + ] + + class TicketViewSet(ModelViewSet): + permission_classes = [IsAuthenticated, GatePermission] + infrasynth_gates = [FeatureGate("ticketing")] + + @gated(TwoFactorGate()) + @action(detail=True, methods=["post"]) + def close(self, request, pk=None): ... + +A view with no gates is ungated — the default is "gate nothing". Gates are +evaluated in order and the first denial raises the matching namespaced +:class:`~infrasynth.shared.exceptions.AppError`, so the response carries the +correct code and HTTP status (``AUTH_*``/``ENTITLEMENT_*``/``VALIDATION_*``). + +The kit's own permission classes (:class:`infrasynth.security.permissions. +HybridPermission`) call :func:`evaluate_gates` too, so declaring gates on a view +that already uses kit permissions is enough. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any, TypeVar + +from rest_framework.permissions import BasePermission + +__all__ = [ + "Gate", + "GateResult", + "GatePermission", + "evaluate_gates", + "gated", + "TwoFactorGate", + "AltchaGate", + "EntitlementGate", + "FeatureGate", + "PermissionGate", +] + +_F = TypeVar("_F", bound=Callable[..., Any]) + + +@dataclass(frozen=True) +class GateResult: + """Outcome of a single gate check.""" + + allowed: bool + code: str = "" + message: str = "Access denied." + details: list[dict[str, Any]] = field(default_factory=list) + status: int = 403 + + @classmethod + def allow(cls) -> GateResult: + return cls(allowed=True) + + @classmethod + def deny( + cls, + code: str, + message: str, + *, + status: int = 403, + details: list[dict[str, Any]] | None = None, + ) -> GateResult: + return cls(allowed=False, code=code, message=message, status=status, details=details or []) + + +class Gate: + """Base class for a single access condition.""" + + def check(self, request: Any, view: Any) -> GateResult: # pragma: no cover - interface + raise NotImplementedError + + +# --- built-in gates --------------------------------------------------------- + + +class TwoFactorGate(Gate): + """Requires the current session/token to have passed the second factor. + + * A user with **no** configured 2FA passes by default (there is nothing to + enforce). Set ``require_configured=True`` to instead demand setup. + * A user **with** 2FA must present proof: a JWT carrying the ``2fa`` claim + (minted only after verification, including across workspace selection) or + a verified session. + """ + + def __init__(self, *, require_configured: bool = False) -> None: + self.require_configured = require_configured + + def check(self, request: Any, view: Any) -> GateResult: + user = getattr(request, "user", None) + if not user or not getattr(user, "is_authenticated", False): + # Authentication is the auth class's job, not the gate's. + return GateResult.allow() + + from infrasynth.security.models import TwoFactorConfig + + config = TwoFactorConfig.objects.filter(user=user).first() + configured = bool(config and config.is_enabled and config.is_configured) + if not configured: + if self.require_configured: + return GateResult.deny( + "AUTH_2FA_SETUP_REQUIRED", + "Second-factor authentication must be configured for this action.", + details=[{"field": "2fa", "issue": "setup_required"}], + ) + return GateResult.allow() + + if _token_has_second_factor(request): + return GateResult.allow() + session = getattr(request, "session", None) + if session is not None and session.get("_2fa_verified"): + return GateResult.allow() + return GateResult.deny( + "AUTH_2FA_REQUIRED", + "Second-factor verification is required for this action.", + details=[{"field": "2fa", "issue": "verification_required"}], + ) + + +class AltchaGate(Gate): + """Requires a valid ALTCHA proof-of-work solution on the request. + + The client fetches a challenge from ``/api/v1/auth/altcha/challenge/`` and + submits the solution in any of these places: + + * JSON body: ``{"altcha": {"challenge_id", "solution", "number"}}`` + * JSON body: flat ``altcha_challenge_id`` / ``altcha_solution`` / ``altcha_number`` + * header ``X-Altcha: ::`` + * query string: the three flat names + + No login is required — that is the point (signup, contact, public forms). + """ + + def __init__(self, *, required: bool = True) -> None: + self.required = required + + def check(self, request: Any, view: Any) -> GateResult: + challenge_id, solution, number = _extract_altcha(request) + if not challenge_id or not solution or number is None: + if not self.required: + return GateResult.allow() + return GateResult.deny( + "VALIDATION_ALTCHA_REQUIRED", + "An ALTCHA proof-of-work solution is required.", + status=400, + details=[{"field": "altcha", "issue": "required"}], + ) + from infrasynth.security.altcha.services import ALTCHAService + + if ALTCHAService().verify(challenge_id, solution, number): + return GateResult.allow() + return GateResult.deny( + "VALIDATION_ALTCHA_INVALID", + "The ALTCHA proof-of-work solution is missing, expired, or invalid.", + status=400, + details=[{"field": "altcha", "issue": "invalid"}], + ) + + +class EntitlementGate(Gate): + """Requires the current tenant to be commercially entitled (``ENTITLEMENT_*``).""" + + def __init__(self, app: str, *, feature: str | None = None) -> None: + self.app = app + self.feature = feature + + def check(self, request: Any, view: Any) -> GateResult: + from infrasynth.billing.entitlements import EntitlementService + from infrasynth.tenancy.context import get_current_tenant + + tenant = get_current_tenant() + service = EntitlementService() + if service.is_entitled(tenant, self.app, feature=self.feature): + return GateResult.allow() + + if tenant is not None and service.get(tenant, self.app) is not None and self.feature: + code = "ENTITLEMENT_PLAN_UPGRADE_REQUIRED" + message = f"The current plan does not include '{self.feature}'." + else: + code = "ENTITLEMENT_APP_NOT_OWNED" + message = f"This workspace is not entitled to '{self.app}'." + return GateResult.deny( + code, + message, + status=402, + details=[{"app": self.app, "feature": self.feature} if self.feature else {"app": self.app}], + ) + + +class FeatureGate(Gate): + """Requires an operational feature flag to be enabled (``404`` when off).""" + + def __init__(self, slug: str, *, default: bool | None = None) -> None: + self.slug = slug + self.default = default + + def check(self, request: Any, view: Any) -> GateResult: + from infrasynth.features.services import FeatureService + from infrasynth.tenancy.context import get_current_tenant + + tenant = get_current_tenant() + enabled = FeatureService().is_enabled( + self.slug, + user=getattr(request, "user", None), + tenant_id=tenant.pk if tenant is not None else None, + default=self.default, + ) + if enabled: + return GateResult.allow() + # Hide existence behind the flag: 404, not 403. + return GateResult.deny("NOT_FOUND", "The requested resource was not found.", status=404) + + +class PermissionGate(Gate): + """Requires permission codename(s) through ``AuthorizationService``.""" + + def __init__(self, *codenames: str, require_all: bool = False) -> None: + self.codenames = codenames + self.require_all = require_all + + def check(self, request: Any, view: Any) -> GateResult: + from infrasynth.security.services import AuthorizationService + + user = getattr(request, "user", None) + authz = AuthorizationService() + if self.require_all: + allowed = authz.has_all_permissions(user, list(self.codenames)) + else: + allowed = authz.has_any_permission(user, list(self.codenames)) + if allowed: + return GateResult.allow() + return GateResult.deny( + "AUTH_FORBIDDEN", + "You do not have permission to perform this action.", + details=[{"field": "permission", "issue": ", ".join(self.codenames)}], + ) + + +# --- evaluation ------------------------------------------------------------- + + +def _gates_for(view: Any) -> list[Gate]: + gates: list[Gate] = list(getattr(view, "infrasynth_gates", []) or []) + action = getattr(view, "action", None) + if action: + handler = getattr(view, action, None) + gates += list(getattr(handler, "infrasynth_gates", []) or []) + getter = getattr(view, "get_infrasynth_gates", None) + if callable(getter): + gates += list(getter() or []) + return gates + + +def evaluate_gates(request: Any, view: Any) -> None: + """Runs every declared gate, raising the namespaced error on the first denial.""" + for gate in _gates_for(view): + result = gate.check(request, view) + if not result.allowed: + _raise_denial(result) + + +def _raise_denial(result: GateResult) -> None: + from infrasynth.shared.exceptions import ( + AppError, + AuthError, + EntitlementError, + NotFoundError, + ValidationAppError, + ) + + code = result.code + if code.startswith("ENTITLEMENT_"): + exc: type[AppError] = EntitlementError + elif code.startswith("VALIDATION_"): + exc = ValidationAppError + elif code.startswith("NOT_FOUND"): + exc = NotFoundError + else: + exc = AuthError + raise exc(result.message, code=code, status=result.status, details=result.details) + + +def gated(*gates: Gate) -> Callable[[_F], _F]: + """Decorator adding gate(s) to a view class or a viewset action method.""" + + def decorator(func: _F) -> _F: + existing = list(getattr(func, "infrasynth_gates", []) or []) + func.infrasynth_gates = existing + list(gates) # type: ignore[attr-defined] + return func + + return decorator + + +class GatePermission(BasePermission): + """DRF permission that evaluates a view's declared gates (no gates ⇒ allow).""" + + message = "Access denied by a gate." + + def has_permission(self, request: Any, view: Any) -> bool: + evaluate_gates(request, view) + return True + + +# --- helpers ---------------------------------------------------------------- + + +def _token_has_second_factor(request: Any) -> bool: + auth = getattr(request, "auth", None) + if auth is None or not hasattr(auth, "get"): + return False + try: + return bool(auth.get("2fa")) + except Exception: # noqa: BLE001 - opaque token objects + return False + + +def _extract_altcha(request: Any) -> tuple[str | None, str | None, int | None]: + from infrasynth.shared.settings_utils import get_setting + + data = getattr(request, "data", None) or {} + payload = data.get("altcha") if hasattr(data, "get") else None + if isinstance(payload, dict): + return ( + payload.get("challenge_id") or payload.get("challengeId"), + payload.get("solution"), + _as_int(payload.get("number")), + ) + + header_name = str(get_setting("INFRASYNTH_SECURITY", "ALTCHA_HEADER", "X-Altcha")) + headers = getattr(request, "headers", {}) or {} + raw = headers.get(header_name) or headers.get(header_name.lower()) + if raw and ":" in raw: + challenge_id, solution, number = (raw.split(":", 2) + [""])[:3] + return challenge_id, solution, _as_int(number) + + def _first(*keys: str): + for key in keys: + if hasattr(data, "get") and data.get(key) is not None: + return data.get(key) + if hasattr(request, "query_params") and request.query_params.get(key) is not None: + return request.query_params.get(key) + return None + + return ( + _first("altcha_challenge_id", "altchaChallengeId"), + _first("altcha_solution", "altchaSolution"), + _as_int(_first("altcha_number", "altchaNumber")), + ) + + +def _as_int(value: Any) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None diff --git a/infrasynth/security/permissions.py b/infrasynth/security/permissions.py index 8c5dd52..566d9ad 100644 --- a/infrasynth/security/permissions.py +++ b/infrasynth/security/permissions.py @@ -21,6 +21,8 @@ from typing import Any from rest_framework.permissions import BasePermission +from infrasynth.gates import evaluate_gates + from .services import AuthorizationService @@ -52,6 +54,8 @@ class HybridPermission(BasePermission): user = getattr(request, "user", None) if not user or not getattr(user, "is_authenticated", False): return False + # Declared gates apply to everyone, including owners and superusers. + evaluate_gates(request, view) if getattr(user, "is_superuser", False): return True if is_tenant_owner(user): diff --git a/infrasynth/security/views.py b/infrasynth/security/views.py index ce9cbef..9dc4287 100644 --- a/infrasynth/security/views.py +++ b/infrasynth/security/views.py @@ -50,10 +50,12 @@ class _AuthSupport: """Cookie/JWT/2FA helpers shared by the auth and 2FA viewsets.""" @staticmethod - def _mint_tokens(user, tenant): + def _mint_tokens(user, tenant, *, two_factor: bool = False): refresh = RefreshToken.for_user(user) if tenant is not None: refresh["tenant"] = str(tenant.pk) + # ``2fa`` lets per-endpoint gates require a step-up second factor. + refresh["2fa"] = bool(two_factor) return str(refresh.access_token), str(refresh) def _cookie_config(self): @@ -116,27 +118,29 @@ class _AuthSupport: ) def _user_from_refresh_cookie(self, request): + """Returns ``(user, claims)`` from the refresh cookie, or ``(None, {})``.""" raw = request.COOKIES.get(get_setting("INFRASYNTH_SECURITY", "REFRESH_COOKIE_NAME", "refresh_token")) if not raw: - return None + return None, {} try: refresh = RefreshToken(self._decrypt(raw)) # type: ignore[arg-type] user_id = refresh.get("user_id") + claims = {"2fa": bool(refresh.get("2fa"))} except Exception: - return None - return UserModel.objects.filter(pk=user_id).first() + return None, {} + return UserModel.objects.filter(pk=user_id).first(), claims - def _complete_login(self, request, user): + def _complete_login(self, request, user, *, two_factor: bool = False): """Issues tenant-bound tokens or a workspace-picker challenge.""" memberships = TenantService().get_active_memberships(user) if len(memberships) == 1: tenant = memberships[0].tenant - access, refresh = self._mint_tokens(user, tenant) + access, refresh = self._mint_tokens(user, tenant, two_factor=two_factor) response = Response({"detail": "Login successful.", "tenant": str(tenant.pk)}) self._set_auth_cookies(response, access, refresh) return response if len(memberships) > 1: - _, refresh = self._mint_tokens(user, None) + _, refresh = self._mint_tokens(user, None, two_factor=two_factor) response = Response( { "detail": "Select a workspace.", @@ -204,7 +208,7 @@ class _AuthSupport: request.session.pop("_2fa_pre_auth_token", None) request.session.pop("_2fa_started_at", None) two_factor_verified.send(sender=self.__class__, user=user, method="totp") - return self._complete_login(request, user) + return self._complete_login(request, user, two_factor=True) class AuthViewSet(_AuthSupport, viewsets.GenericViewSet): @@ -262,7 +266,7 @@ class AuthViewSet(_AuthSupport, viewsets.GenericViewSet): @action(detail=False, methods=["post"], url_path="select-workspace") def select_workspace(self, request): - user = self._user_from_refresh_cookie(request) + user, claims = self._user_from_refresh_cookie(request) if user is None: raise AuthenticationFailed("No pending workspace selection.") tenant_id = request.data.get("tenantId") or request.data.get("tenant_id") @@ -272,7 +276,7 @@ class AuthViewSet(_AuthSupport, viewsets.GenericViewSet): tenant = TenantService().select_tenant(user, tenant_id) except NotFoundError as exc: raise NotFound(str(exc)) from exc - access, refresh = self._mint_tokens(user, tenant) + access, refresh = self._mint_tokens(user, tenant, two_factor=claims.get("2fa", False)) response = Response({"detail": "Workspace selected.", "tenant": str(tenant.pk)}) self._set_auth_cookies(response, access, refresh) return response @@ -297,7 +301,13 @@ class AuthViewSet(_AuthSupport, viewsets.GenericViewSet): ) except NotFoundError as exc: raise NotFound(str(exc)) from exc - access, refresh = self._mint_tokens(request.user, tenant) + auth = getattr(request, "auth", None) + two_factor = ( + bool(auth.get("2fa")) + if auth is not None and hasattr(auth, "get") + else bool(getattr(request, "session", None) and request.session.get("_2fa_verified")) + ) + access, refresh = self._mint_tokens(request.user, tenant, two_factor=two_factor) response = Response({"detail": "Workspace switched.", "tenant": str(tenant.pk)}) self._set_auth_cookies(response, access, refresh) return response diff --git a/tests/test_gates.py b/tests/test_gates.py new file mode 100644 index 0000000..ab32a46 --- /dev/null +++ b/tests/test_gates.py @@ -0,0 +1,196 @@ +"""Tests for the composable per-endpoint gate layer.""" + +import pytest + +from infrasynth.gates import ( + AltchaGate, + EntitlementGate, + FeatureGate, + GatePermission, + PermissionGate, + TwoFactorGate, + evaluate_gates, + gated, +) +from infrasynth.shared.exceptions import AppError, AuthError, EntitlementError, NotFoundError + +pytestmark = pytest.mark.django_db + + +class _Anon: + is_authenticated = False + + +class FakeRequest: + def __init__(self, user=None, *, auth=None, session=None, data=None, headers=None, query_params=None): + self.user = user if user is not None else _Anon() + self.auth = auth + self.session = session + self.data = data or {} + self.headers = headers or {} + self.query_params = query_params or {} + + +class FakeView: + def __init__(self, *gates): + self.infrasynth_gates = list(gates) + + +class TestGatePermission: + def test_no_gates_allows(self, user): + assert GatePermission().has_permission(FakeRequest(user), FakeView()) is True + + def test_decorator_merges_gates(self): + class V: + pass + + @gated(FeatureGate("a")) + @gated(FeatureGate("b")) + def action(self): + return None + + assert len(action.infrasynth_gates) == 2 + + +class TestTwoFactorGate: + def test_no_config_passes_by_default(self, user): + evaluate_gates(FakeRequest(user), FakeView(TwoFactorGate())) + + def test_no_config_required_denies(self, user): + with pytest.raises(AuthError) as exc: + evaluate_gates(FakeRequest(user), FakeView(TwoFactorGate(require_configured=True))) + assert exc.value.code == "AUTH_2FA_SETUP_REQUIRED" + + def test_configured_without_proof_denies(self, user): + from infrasynth.security.models import TwoFactorConfig + + TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True, secret_key_encrypted="x") + with pytest.raises(AuthError) as exc: + evaluate_gates(FakeRequest(user), FakeView(TwoFactorGate())) + assert exc.value.code == "AUTH_2FA_REQUIRED" + + def test_configured_with_token_claim_passes(self, user): + from infrasynth.security.models import TwoFactorConfig + + TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True, secret_key_encrypted="x") + request = FakeRequest(user, auth={"2fa": True}) + evaluate_gates(request, FakeView(TwoFactorGate())) + + def test_configured_with_session_passes(self, user): + from infrasynth.security.models import TwoFactorConfig + + TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True, secret_key_encrypted="x") + request = FakeRequest(user, session={"_2fa_verified": True}) + evaluate_gates(request, FakeView(TwoFactorGate())) + + +class TestAltchaGate: + def test_missing_token_denies(self): + with pytest.raises(AppError) as exc: + evaluate_gates(FakeRequest(), FakeView(AltchaGate())) + assert exc.value.code == "VALIDATION_ALTCHA_REQUIRED" + assert exc.value.status == 400 + + def test_valid_solution_passes(self): + from infrasynth.security.altcha.services import ALTCHAService + + svc = ALTCHAService() + challenge = svc.create_challenge() + solution, number = svc.compute_solution(challenge["salt"], challenge["difficulty"]) + request = FakeRequest( + data={"altcha": {"challenge_id": challenge["challenge_id"], "solution": solution, "number": number}} + ) + evaluate_gates(request, FakeView(AltchaGate())) + + def test_header_solution_passes(self): + from infrasynth.security.altcha.services import ALTCHAService + + svc = ALTCHAService() + challenge = svc.create_challenge() + solution, number = svc.compute_solution(challenge["salt"], challenge["difficulty"]) + token = f"{challenge['challenge_id']}:{solution}:{number}" + evaluate_gates(FakeRequest(headers={"X-Altcha": token}), FakeView(AltchaGate())) + + def test_bad_solution_denies(self): + from infrasynth.security.altcha.services import ALTCHAService + + challenge = ALTCHAService().create_challenge() + request = FakeRequest( + data={"altcha": {"challenge_id": challenge["challenge_id"], "solution": "deadbeef", "number": 1}} + ) + with pytest.raises(AppError) as exc: + evaluate_gates(request, FakeView(AltchaGate())) + assert exc.value.code == "VALIDATION_ALTCHA_INVALID" + + +class TestEntitlementGate: + @pytest.fixture + def entitled(self, tenant): + from infrasynth.billing.models import App, Entitlement, Plan + from infrasynth.shared.enums import EntitlementStatus, MonetizationModel + + app = App.objects.create(slug="messenger", name="Messenger", monetization=MonetizationModel.SUBSCRIPTION) + plan = Plan.objects.create(app=app, slug="pro", name="Pro", price_amount=0, features={"payouts": True}) + Entitlement.objects.create(tenant=tenant, app=app, plan=plan, status=EntitlementStatus.ACTIVE) + return app + + def test_entitled_passes(self, entitled): + evaluate_gates(FakeRequest(), FakeView(EntitlementGate("messenger", feature="payouts"))) + + def test_feature_not_in_plan_denies(self, entitled): + with pytest.raises(EntitlementError) as exc: + evaluate_gates(FakeRequest(), FakeView(EntitlementGate("messenger", feature="broadcast"))) + assert exc.value.code == "ENTITLEMENT_PLAN_UPGRADE_REQUIRED" + assert exc.value.status == 402 + + def test_no_entitlement_denies(self, tenant): + with pytest.raises(EntitlementError) as exc: + evaluate_gates(FakeRequest(), FakeView(EntitlementGate("messenger"))) + assert exc.value.code == "ENTITLEMENT_APP_NOT_OWNED" + + +class TestFeatureGate: + def test_disabled_hides_as_404(self): + from infrasynth.features.models import FeatureFlag + + FeatureFlag.objects.create(slug="ticketing", is_active=False) + with pytest.raises(NotFoundError) as exc: + evaluate_gates(FakeRequest(), FakeView(FeatureGate("ticketing"))) + assert exc.value.status == 404 + + def test_enabled_passes(self): + from infrasynth.features.models import FeatureFlag + + FeatureFlag.objects.create(slug="ticketing", is_active=True) + evaluate_gates(FakeRequest(), FakeView(FeatureGate("ticketing"))) + + +class TestPermissionGate: + def test_missing_permission_denies(self, user): + with pytest.raises(AuthError) as exc: + evaluate_gates(FakeRequest(user), FakeView(PermissionGate("billing.payout"))) + assert exc.value.code == "AUTH_FORBIDDEN" + + def test_grant_passes(self, user): + from infrasynth.security.models import Grant + + Grant.objects.create(user=user, codename="billing.payout") + evaluate_gates(FakeRequest(user), FakeView(PermissionGate("billing.payout"))) + + +class TestHybridPermissionIntegration: + def test_kit_permission_evaluates_declared_gates(self, user): + from infrasynth.features.models import FeatureFlag + from infrasynth.security.permissions import HybridPermission + + FeatureFlag.objects.create(slug="ticketing", is_active=False) + view = FakeView(FeatureGate("ticketing")) + with pytest.raises(NotFoundError): + HybridPermission().has_permission(FakeRequest(user), view) + + def test_kit_permission_allows_when_gate_passes(self, user): + from infrasynth.features.models import FeatureFlag + from infrasynth.security.permissions import HybridPermission + + FeatureFlag.objects.create(slug="ticketing", is_active=True) + assert HybridPermission().has_permission(FakeRequest(user), FakeView(FeatureGate("ticketing"))) is True diff --git a/tests/test_security/test_two_factor.py b/tests/test_security/test_two_factor.py index 5321ed4..5cf482d 100644 --- a/tests/test_security/test_two_factor.py +++ b/tests/test_security/test_two_factor.py @@ -155,6 +155,28 @@ class TestTwoFactorLoginFlow: assert resp.status_code == status.HTTP_200_OK assert "access_token" in resp.cookies + def test_verify_mints_second_factor_claim(self, db, user): + from rest_framework_simplejwt.tokens import AccessToken + + from infrasynth.shared.crypto import decrypt + + secret = self._enable(user) + client = APIClient() + client.post("/api/v1/auth/login/", {"username": "testuser", "password": "testpass123"}, format="json") + client.post("/api/v1/auth/2fa/verify/", {"code": pyotp.TOTP(secret).now()}, format="json") + token = AccessToken(decrypt(client.cookies["access_token"].value)) + assert token["2fa"] is True + + def test_plain_login_has_no_second_factor_claim(self, db, user): + from rest_framework_simplejwt.tokens import AccessToken + + from infrasynth.shared.crypto import decrypt + + client = APIClient() + client.post("/api/v1/auth/login/", {"username": "testuser", "password": "testpass123"}, format="json") + token = AccessToken(decrypt(client.cookies["access_token"].value)) + assert token["2fa"] is False + class TestTwoFactorSetup: def test_setup_returns_secret_and_qr(self, two_factor_client):