Source code for ferrosoft.apps.webapi.auth

from django.utils.translation import gettext as _
from rest_framework import exceptions
from rest_framework.authentication import BasicAuthentication

from ferrosoft.apps.ferrobase.middleware import (
    TenantMiddleware,
    TenantNotFound,
    TenantNotAllowed,
)
from ferrosoft.apps.ferrobase.middleware.tenant import get_validated_tenant
from ferrosoft.apps.ferrobase.models import User
from ferrosoft.apps.ferrobase.tenant.context import set_current_tenant


[docs] class TenantAuthentication(BasicAuthentication):
[docs] def authenticate(self, request): ret = super().authenticate(request) if ret is None: raise exceptions.NotAuthenticated() # A tenant ID is always required, with query param having precedence over cookie. tenant_id = request.query_params.get( TenantMiddleware.TENANT_REQUEST_KEY, request.COOKIES.get(TenantMiddleware.TENANT_REQUEST_KEY, None), ) if not tenant_id: raise exceptions.AuthenticationFailed("tenant ID is required") user, _ = ret self._check_feature(user) try: tenant = get_validated_tenant(tenant_id, user) set_current_tenant(tenant) request.current_tenant = tenant except TenantNotFound as e: raise exceptions.AuthenticationFailed(str(e)) except TenantNotAllowed as e: raise exceptions.AuthenticationFailed(str(e)) return ret
@staticmethod def _check_feature(user: User): if not hasattr(user, "get_assigned_feature"): # User class is different for anonymous users, in which case feature # need not be checked. return assignment, plan = user.get_assigned_feature("api_access") assignment_required = plan is not None and not plan.enable_all_features if assignment_required and assignment is None: raise exceptions.AuthenticationFailed( _("API access isn’t included in your current subscription.") )