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.")
)