moravak/infrasynth-backend-kit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(gates): composable per-endpoint gating extension API

Browse source 
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
This commit is contained in:
jcv-dev 2026-09-24 10:49:44 -05:00
parent 551b42eab5
commit 21731b9887
9 changed files with 710 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
CHANGELOG.md
View file

@ -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

1
PLAN.md
View file

@ -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

73
README.md
View file

@ -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: <challenge_id>:<solution>:<number>` 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

2
config/settings/base.py
View file

@ -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,

381
infrasynth/gates.py Normal file
View file

@ -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: <challenge_id>:<solution>:<number>``
* 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

4
infrasynth/security/permissions.py
View file

@ -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):

32
infrasynth/security/views.py
View file

@ -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

196
tests/test_gates.py Normal file
View file

@ -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

22
tests/test_security/test_two_factor.py
View file

@ -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):