Enforce tenant isolation on verified audit firm domains

app/core/middleware/context.py

@@ -59,6 +59,19 @@ def _normalise_session_int(value):
         return None
 
 
+_TENANT_BOUND_DOMAIN_TYPES = {"audit_firm_domain", "audit_firm_subdomain"}
+
+
+def _is_trusted_tenant_bound_domain(request: Request) -> bool:
+    return (
+        bool(getattr(request.state, "domain_resolved", False))
+        and bool(getattr(request.state, "domain_is_verified", False))
+        and (getattr(request.state, "domain_status", None) or "").strip().lower() == "active"
+        and (getattr(request.state, "domain_type", None) or "").strip() in _TENANT_BOUND_DOMAIN_TYPES
+        and _normalise_session_int(getattr(request.state, "domain_tenant_id", None)) is not None
+    )
+
+
 class ContextResolveMiddleware(BaseHTTPMiddleware):
     def __init__(self, app: ASGIApp) -> None:
         super().__init__(app)
@@ -101,16 +114,29 @@ class ContextResolveMiddleware(BaseHTTPMiddleware):
         session_tenant_code = (session.get("active_tenant_code") or session.get("tenant_code") or "").strip() or None
         session_branch_code = (session.get("active_branch_code") or session.get("branch_code") or "").strip() or None
 
+        domain_tenant_id = _normalise_session_int(getattr(request.state, "domain_tenant_id", None))
+        domain_branch_id = _normalise_session_int(getattr(request.state, "domain_branch_id", None))
         domain_tenant_code = getattr(request.state, "domain_tenant_code", None)
         domain_branch_code = getattr(request.state, "domain_branch_code", None)
+        tenant_bound_domain = _is_trusted_tenant_bound_domain(request)
 
+        # On a verified active audit-firm hostname, the domain mapping is the
+        # authoritative tenant/branch context. Session selection must never
+        # escape to another tenant while the user remains on that hostname.
+        if tenant_bound_domain:
+            active_tenant_id = domain_tenant_id
+            active_branch_id = domain_branch_id if domain_branch_id is not None else session_branch_id
+            tenant_code = domain_tenant_code or self.s.DEFAULT_TENANT_CODE
+            branch_code = domain_branch_code or session_branch_code or self.s.DEFAULT_BRANCH_CODE
+        else:
+            active_tenant_id = session_tenant_id
+            active_branch_id = session_branch_id
             tenant_code = (
                 session_tenant_code
                 or domain_tenant_code
                 or (request.headers.get(_TENANT_HEADER) if trust_headers else None)
                 or self.s.DEFAULT_TENANT_CODE
             )
-
             branch_code = (
                 session_branch_code
                 or domain_branch_code
@@ -124,8 +150,9 @@ class ContextResolveMiddleware(BaseHTTPMiddleware):
             or self.s.DEFAULT_YEAR_CODE
         )
 
-        request.state.active_tenant_id = session_tenant_id
-        request.state.active_branch_id = session_branch_id
+        request.state.active_tenant_id = active_tenant_id
+        request.state.active_branch_id = active_branch_id
+        request.state.tenant_bound_domain = tenant_bound_domain
         request.state.tenant_code = tenant_code
         request.state.branch_code = branch_code
         request.state.year_code = year_code

app/modules/system_settings/ui.py

@@ -114,6 +114,25 @@ def _store_active_branch_context(request: Request, branch: Branch | None) -> Non
     request.session["active_branch_code"] = branch.code
 
 
+_TENANT_BOUND_DOMAIN_TYPES = {"audit_firm_domain", "audit_firm_subdomain"}
+
+
+def _bound_domain_tenant_id(request: Request) -> int | None:
+    if not bool(getattr(request.state, "domain_resolved", False)):
+        return None
+    if not bool(getattr(request.state, "domain_is_verified", False)):
+        return None
+    if (getattr(request.state, "domain_status", None) or "").strip().lower() != "active":
+        return None
+    if (getattr(request.state, "domain_type", None) or "").strip() not in _TENANT_BOUND_DOMAIN_TYPES:
+        return None
+    value = getattr(request.state, "domain_tenant_id", None)
+    try:
+        return int(value) if value not in (None, "", 0, "0") else None
+    except (TypeError, ValueError):
+        return None
+
+
 def _can_manage_financial_years(db, user) -> bool:
     roles = set(get_user_roles(db, user.id))
     perms = set(get_user_permissions(db, user.id))
@@ -1435,6 +1454,10 @@ def switch_active_tenant(request: Request, tenant_id: int):
         if not user:
             return RedirectResponse(url="/login", status_code=303)
 
+        bound_tenant_id = _bound_domain_tenant_id(request)
+        if bound_tenant_id is not None and int(tenant_id) != int(bound_tenant_id):
+            return _redirect_denied()
+
         roles = get_user_roles(db, user.id)
         perms = set(get_user_permissions(db, user.id))
         if "System Admin" not in roles or "services.cross_tenant" not in perms:

app/ui/routes/auth.py

@@ -33,6 +33,54 @@ SAFE_POST_LOGIN_REDIRECTS = {
     "/employee/dashboard",
 }
 
+# Verified active audit-firm domains are tenant authentication boundaries.
+# Marketplace and consultant domain behaviour remains unchanged.
+TENANT_BOUND_DOMAIN_TYPES = {"audit_firm_domain", "audit_firm_subdomain"}
+
+
+def _bound_domain_tenant_id(request: Request) -> int | None:
+    """Return the trusted tenant id bound to the current audit-firm domain.
+
+    DomainResolverMiddleware only marks exact, active and verified mappings as
+    resolved. The additional checks here make the authentication boundary
+    explicit and safe if the middleware evolves later.
+    """
+    if not bool(getattr(request.state, "domain_resolved", False)):
+        return None
+    if not bool(getattr(request.state, "domain_is_verified", False)):
+        return None
+    if (getattr(request.state, "domain_status", None) or "").strip().lower() != "active":
+        return None
+    if (getattr(request.state, "domain_type", None) or "").strip() not in TENANT_BOUND_DOMAIN_TYPES:
+        return None
+    tenant_id = getattr(request.state, "domain_tenant_id", None)
+    try:
+        return int(tenant_id) if tenant_id not in (None, "", 0, "0") else None
+    except (TypeError, ValueError):
+        return None
+
+
+def _clear_login_session(request: Request) -> None:
+    """Remove authentication/context state without disturbing CSRF/session middleware."""
+    for key in (
+        SESSION_USER_ID_KEY,
+        SESSION_LOGIN_AT_KEY,
+        "user_email",
+        "tenant_id",
+        "branch_id",
+        "tenant_code",
+        "branch_code",
+        "active_tenant_id",
+        "active_branch_id",
+        "active_tenant_code",
+        "active_branch_code",
+        "active_financial_year",
+        "must_change_password",
+        "post_login_redirect",
+        "otp_verified",
+    ):
+        request.session.pop(key, None)
+
 
 def _consume_safe_post_login_redirect(request: Request) -> str | None:
     value = request.session.pop(PENDING_POST_LOGIN_REDIRECT_KEY, None)
@@ -447,6 +495,20 @@ def login_submit(
         permissions = _user_permissions(db, user_id)
         bs = _get_branch_security_policy(db, user)
 
+        # A verified active audit-firm domain is a hard tenant boundary.
+        # System Admin retains the existing platform-support capability, but all
+        # tenant users must belong to the tenant mapped to this hostname.
+        bound_tenant_id = _bound_domain_tenant_id(request)
+        is_system_admin = "System Admin" in set(roles or [])
+        user_tenant_id = int(tenant_id) if tenant_id not in (None, "", 0, "0") else None
+        if bound_tenant_id is not None and not is_system_admin and user_tenant_id != bound_tenant_id:
+            _clear_login_session(request)
+            return _render_login(
+                request,
+                flash="This account does not belong to the firm associated with this domain. Please use your firm's login URL.",
+                status_code=403,
+            )
+
         request.session[SESSION_USER_ID_KEY] = user_id
         request.session[SESSION_LOGIN_AT_KEY] = now.isoformat()
         request.session["user_email"] = user_email