# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
from datetime import timedelta
from urllib.parse import quote
from asgiref.sync import iscoroutinefunction, markcoroutinefunction
from django.conf import settings
from django.core.exceptions import ValidationError
from django.http import HttpRequest
from django.shortcuts import redirect
from django.urls import reverse
from ferrosoft.apps.ferrobase.models import Tenant, User, Organization
from ferrosoft.apps.ferrobase.tenant.context import get_current_tenant, active_tenant
[docs]
class BaseError(Exception):
"""Base class for all tenant-resolution errors."""
[docs]
class TenantNotFound(BaseError):
"""Raised when the requested tenant ID does not exist or is inactive."""
def __init__(self):
super().__init__("tenant not found")
[docs]
class TenantNotAllowed(BaseError):
"""Raised when the authenticated user is not authorised to access the requested tenant."""
def __init__(self):
super().__init__("user unauthorized to use requested tenant")
[docs]
class TenantChoiceRequired(BaseError):
"""Raised when no tenant has been chosen and user interaction is required to select one."""
[docs]
def get_validated_tenant(tenant_id: str, user: User | None) -> Tenant:
"""Look up and authorise a tenant synchronously.
Args:
tenant_id: Primary key string of the tenant to resolve (must be a
valid UUID; malformed values are treated as not-found).
user: Authenticated user to check against the tenant's organisation,
or ``None`` to skip the authorisation check.
Returns:
Tenant: The validated, active ``Tenant`` instance.
Raises:
TenantNotFound: If no active tenant with the given ID exists.
TenantNotAllowed: If the user is not permitted to use this tenant.
"""
try:
return _verify_tenant(Tenant.find(tenant_id), user)
except ValidationError:
# This error is raised if the value is not a valid UUID.
# In this case, assume a malformed request and treat it
# as TenantNotFound.
raise TenantNotFound()
[docs]
async def aget_validated_tenant(tenant_id: str, user: User | None) -> Tenant:
"""Async variant of ``get_validated_tenant``.
Args:
tenant_id: Primary key string of the tenant to resolve.
user: Authenticated user, or ``None`` to skip authorisation.
Returns:
Tenant: The validated, active ``Tenant`` instance.
Raises:
TenantNotFound: If no active tenant with the given ID exists.
TenantNotAllowed: If the user is not permitted to use this tenant.
"""
try:
return _verify_tenant(await Tenant.afind(tenant_id), user)
except ValidationError:
raise TenantNotFound()
def _verify_tenant(tenant: Tenant, user: User | None) -> Tenant:
if not tenant:
raise TenantNotFound()
if user and hasattr(user, "tenant_allowed") and not user.tenant_allowed(tenant):
raise TenantNotAllowed()
return tenant
[docs]
def validate_chosen_tenant(
tenant_id: str | None, user: User | None, organization: Organization
):
"""Resolve the tenant to use for an organisation, redirecting when no choice is made.
If ``tenant_id`` is provided, it is validated; otherwise the organisation's
single tenant is used automatically. When the organisation has multiple
tenants and none is specified, or when the requested tenant is invalid,
``TenantChoiceRequired`` is raised so the caller can redirect to the
selection page.
Args:
tenant_id: Tenant primary key string, or ``None``.
user: Currently authenticated user for authorisation checks.
organization: The user's organisation used to enumerate available tenants.
Returns:
Tenant: The resolved ``Tenant`` instance.
Raises:
TenantChoiceRequired: When the tenant cannot be resolved automatically.
"""
try:
if tenant_id:
return get_validated_tenant(tenant_id, user)
else:
if organization.tenants.count() == 1:
# Short circuit to not show superfluous selection page.
return organization.tenants.first()
else:
# No tenant has been chosen yet. Redirect to
# selection page.
raise TenantChoiceRequired("Organization has multiple tenants")
except TenantNotFound as exc:
# The tenant might not be found, because the database has
# been reset. In any case, redirect to selectio page.
raise TenantChoiceRequired("Tenant was not found") from exc
except TenantNotAllowed as exc:
# Let user choose a different tenant.
raise TenantChoiceRequired("Tenant is not allowed") from exc
[docs]
async def avalidate_chosen_tenant(
tenant_id: str | None, user: User | None, organization: Organization
):
"""Async variant of ``validate_chosen_tenant``.
Args:
tenant_id: Tenant primary key string, or ``None``.
user: Currently authenticated user for authorisation checks.
organization: The user's organisation.
Returns:
Tenant: The resolved ``Tenant`` instance.
Raises:
TenantChoiceRequired: When the tenant cannot be resolved automatically.
"""
try:
if tenant_id:
return await aget_validated_tenant(tenant_id, user)
else:
if await organization.tenants.acount() == 1:
# Short circuit to not show superfluous selection page.
return await organization.tenants.afirst()
else:
# No tenant has been chosen yet. Redirect to
# selection page.
raise TenantChoiceRequired("Organization has multiple tenants")
except TenantNotFound as exc:
# The tenant might not be found, because the database has
# been reset. In any case, redirect to selectio page.
raise TenantChoiceRequired("Tenant was not found") from exc
except TenantNotAllowed as exc:
# Let user choose a different tenant.
raise TenantChoiceRequired("Tenant is not allowed") from exc
[docs]
class TenantMiddleware:
"""ASGI-capable middleware that resolves and persists the active tenant per request.
Reads the tenant ID from POST → GET → cookie (in that order of priority).
When the ID arrives via POST or GET, it is written back to the cookie so
subsequent requests do not need to repeat it. The resolved ``Tenant``
instance is stored on ``request.current_tenant`` and in the
``active_tenant`` context variable so ORM queries are automatically routed
to the correct database.
Unauthenticated requests and the tenant-selection view itself bypass
resolution to avoid redirect loops.
"""
async_capable = True
sync_capable = False
TENANT_REQUEST_KEY = "fptid"
"""Name of the cookie and GET/POST parameter that carries the tenant ID."""
def __init__(self, get_response):
self.get_response = get_response
if iscoroutinefunction(self.get_response):
markcoroutinefunction(self)
async def __call__(self, request):
choose_url_path_only, choose_url = self.choose_url(request)
request.current_tenant = None
try:
tenant, tenant_change_requested = await self.adetermine_tenant(
request, choose_url_path_only
)
except TenantChoiceRequired:
return redirect(choose_url)
# Storing current tenant on request for convenience of views.
request.current_tenant = tenant
# Non-view code relies on tenant being set in context.
with active_tenant(tenant):
response = await self.get_response(request)
tenant_cookie_unset = self.TENANT_REQUEST_KEY not in request.COOKIES
if tenant is not None and (tenant_change_requested or tenant_cookie_unset):
response.set_cookie(
self.TENANT_REQUEST_KEY,
str(tenant.pk),
samesite="Strict",
httponly=True,
max_age=timedelta(weeks=2),
)
return response
[docs]
@staticmethod
def choose_url(request):
next_url = request.GET.get("next", None)
if next_url is None:
next_url = "/" if request.path == "/tenants/" else request.path
url = reverse("ferrobase:choose_tenant")
return url, "%s?next=%s" % (url, quote(next_url))
[docs]
@classmethod
async def adetermine_tenant(cls, request: HttpRequest, choose_url: str):
request.current_tenant = None
user: User = await request.auser()
organization: Organization = getattr(user, "organization", None)
# Avoid redirect loops by requiring: * A session must exist. * The user
# must be authenticated. * Request is not OIDC callback view. This is
# important to allow normal OIDC flow.
if await cls._is_enforced_request(request):
# Need to determine whether tenant ID was explicitly requested.
requested_tenant_id = cls._get_requested_tenant(request)
tenant_id = requested_tenant_id or request.COOKIES.get(
cls.TENANT_REQUEST_KEY, None
)
if request.path != choose_url:
return (
await avalidate_chosen_tenant(tenant_id, user, organization),
requested_tenant_id is not None,
)
return None, False
@classmethod
async def _is_enforced_request(cls, request):
# Avoid redirect loops by requiring: * A session must exist. * The user
# must be authenticated. * Request is not OIDC callback view. This is
# important to allow normal OIDC flow.
session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME, None)
user = await request.auser()
return (
session_key
and user.is_authenticated
and not request.path == "/oidc/callback/"
)
@classmethod
def _get_requested_tenant(cls, request):
return request.POST.get(
cls.TENANT_REQUEST_KEY,
request.GET.get(cls.TENANT_REQUEST_KEY, None),
)
[docs]
class GlobalRouter:
"""
Send queries for Django's built-in apps and some of Ferrobase to default database.
"""
_global_app_labels = {
"admin",
"auth",
"contenttypes",
"django_q",
"otp_email",
"otp_static",
"otp_totp",
"phonenumber",
"sessions",
}
_ferrobase_exempted_models = {
"apitoken",
"appfeature",
"featureassignment",
"language",
"organization",
"passwordresetrequest",
"subscriptionplan",
"tenant",
"timezone",
"accountactivationrequest",
"user",
"user_groups",
"user_user_permissions",
"user_tenants",
}
def _global_db(self, model):
if model._meta.app_label in self._global_app_labels:
return "default"
if (
model._meta.app_label == "ferrobase"
and model._meta.model_name in self._ferrobase_exempted_models
):
return "default"
return None
[docs]
def db_for_read(self, model, **hints):
return self._global_db(model)
[docs]
def db_for_write(self, model, **hints):
return self._global_db(model)
[docs]
class TenantRouter:
"""
Send queries to tenant database.
If no tenant is available, None is returned. TenantMiddleware makes sure
to plug all holes so that no user can accidentally write Ferrosoft Platform
objects to the default database.
"""
@staticmethod
def _tenant_db():
tenant = get_current_tenant()
if isinstance(tenant, Tenant):
return tenant.database_name
return None
[docs]
def db_for_read(self, model, **hints):
return self._tenant_db()
[docs]
def db_for_write(self, model, **hints):
return self._tenant_db()