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
86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
from django.db.models import Q
|
|
from django.utils import timezone
|
|
|
|
from .models import Grant, Revoke
|
|
|
|
|
|
class AuthorizationService:
|
|
"""Singleton service for permission resolution."""
|
|
|
|
def _get_system_user_scopes(self, user) -> set[str] | None:
|
|
from .auth.api_keys import SystemUser
|
|
|
|
if isinstance(user, SystemUser):
|
|
return set(user.scopes)
|
|
return None
|
|
|
|
def has_permission(self, user, codename: str) -> bool:
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
if user.is_superuser:
|
|
return True
|
|
system_scopes = self._get_system_user_scopes(user)
|
|
if system_scopes is not None:
|
|
return codename in system_scopes
|
|
if Revoke.objects.filter(user=user, codename=codename).exists():
|
|
return False
|
|
if (
|
|
Grant.objects.filter(user=user, codename=codename)
|
|
.filter(Q(expires_at__isnull=True) | Q(expires_at__gt=timezone.now()))
|
|
.exists()
|
|
):
|
|
return True
|
|
user_roles = self._get_role_permission_lists(user)
|
|
for perm_list in user_roles:
|
|
if codename in (perm_list or []):
|
|
return True
|
|
return False
|
|
|
|
def get_effective_permissions(self, user) -> set[str]:
|
|
if not user or not user.is_authenticated:
|
|
return set()
|
|
if user.is_superuser:
|
|
return {"*"}
|
|
system_scopes = self._get_system_user_scopes(user)
|
|
if system_scopes is not None:
|
|
return system_scopes
|
|
revoked = set(Revoke.objects.filter(user=user).values_list("codename", flat=True))
|
|
granted = set(
|
|
Grant.objects.filter(user=user)
|
|
.filter(Q(expires_at__isnull=True) | Q(expires_at__gt=timezone.now()))
|
|
.values_list("codename", flat=True)
|
|
)
|
|
role_perms = set()
|
|
for perm_list in self._get_role_permission_lists(user):
|
|
role_perms.update(perm_list or [])
|
|
return (granted | role_perms) - revoked
|
|
|
|
@staticmethod
|
|
def _get_role_permission_lists(user) -> list[list[str]]:
|
|
from infrasynth.tenancy.context import get_current_tenant
|
|
from infrasynth.tenancy.models import TenantMembership
|
|
|
|
from .models import Role
|
|
|
|
role_perms: list[list[str]] = []
|
|
roles = getattr(user, "roles", None)
|
|
if roles is not None:
|
|
role_perms.extend(list(roles.values_list("permissions", flat=True)))
|
|
|
|
# Roles assigned via membership in the current tenant (TENANCY.md §6).
|
|
tenant = get_current_tenant()
|
|
if tenant is not None:
|
|
slugs = TenantMembership.objects.filter(user=user, tenant=tenant, is_active=True).values_list(
|
|
"role", flat=True
|
|
)
|
|
for slug in set(slugs):
|
|
role = Role.objects.filter(slug=slug).first()
|
|
if role is not None:
|
|
role_perms.append(role.permissions or [])
|
|
return role_perms
|
|
|
|
def has_all_permissions(self, user, codenames: list[str]) -> bool:
|
|
return all(self.has_permission(user, c) for c in codenames)
|
|
|
|
def has_any_permission(self, user, codenames: list[str]) -> bool:
|
|
return any(self.has_permission(user, c) for c in codenames)
|