From 2e87013563e4479f41c21bfd1129d1f13a871d70 Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Fri, 10 Jul 2026 15:04:54 +0530 Subject: [PATCH] Enforce tenant isolation on verified audit firm domains --- app/core/middleware/context.py | 57 ++++++++++++++++++++-------- app/modules/system_settings/ui.py | 23 ++++++++++++ app/ui/routes/auth.py | 62 +++++++++++++++++++++++++++++++ 3 files changed, 127 insertions(+), 15 deletions(-) diff --git a/app/core/middleware/context.py b/app/core/middleware/context.py index 71cfabd..00a0790 100644 --- a/app/core/middleware/context.py +++ b/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,22 +114,35 @@ 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) - 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 - or (request.headers.get(_BRANCH_HEADER) if trust_headers else None) - or self.s.DEFAULT_BRANCH_CODE - ) + # 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 + or (request.headers.get(_BRANCH_HEADER) if trust_headers else None) + or self.s.DEFAULT_BRANCH_CODE + ) year_code = ( session.get("active_financial_year") @@ -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 diff --git a/app/modules/system_settings/ui.py b/app/modules/system_settings/ui.py index 19e163d..bdaab8e 100644 --- a/app/modules/system_settings/ui.py +++ b/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: diff --git a/app/ui/routes/auth.py b/app/ui/routes/auth.py index 7232d53..1b81c3a 100644 --- a/app/ui/routes/auth.py +++ b/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