A R R R Associates
134 lines
4.9 KiB
Python
134 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
from ipaddress import ip_address, ip_network
|
|
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.requests import Request
|
|
from starlette.types import ASGIApp
|
|
|
|
from app.core.settings import get_settings
|
|
|
|
|
|
_CONTEXT_SECRET_HEADER = "X-AuditFirm-Context-Secret"
|
|
_TENANT_HEADER = "X-Tenant-Code"
|
|
_BRANCH_HEADER = "X-Branch-Code"
|
|
_YEAR_HEADER = "X-Year-Code"
|
|
|
|
|
|
def _csv_values(value: str | None) -> list[str]:
|
|
return [item.strip() for item in (value or "").split(",") if item.strip()]
|
|
|
|
|
|
def _safe_env(value: str | None) -> str:
|
|
return (value or "").strip().lower()
|
|
|
|
|
|
def _host_matches_trusted_entry(client_host: str, trusted_entry: str) -> bool:
|
|
"""Return True when client_host matches a trusted host/IP/CIDR entry.
|
|
|
|
Deliberately does not support '*' wildcard. For Docker/Coolify internal
|
|
networks, use an explicit CIDR such as 172.16.0.0/12.
|
|
"""
|
|
client_host = (client_host or "").strip().lower()
|
|
trusted_entry = (trusted_entry or "").strip().lower()
|
|
if not client_host or not trusted_entry:
|
|
return False
|
|
|
|
if client_host == trusted_entry:
|
|
return True
|
|
|
|
try:
|
|
client_ip = ip_address(client_host)
|
|
except ValueError:
|
|
return False
|
|
|
|
try:
|
|
if "/" in trusted_entry:
|
|
return client_ip in ip_network(trusted_entry, strict=False)
|
|
return client_ip == ip_address(trusted_entry)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _normalise_session_int(value):
|
|
if value in (None, "", 0, "0"):
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
class ContextResolveMiddleware(BaseHTTPMiddleware):
|
|
def __init__(self, app: ASGIApp) -> None:
|
|
super().__init__(app)
|
|
self.s = get_settings()
|
|
|
|
def _context_headers_are_trusted(self, request: Request) -> bool:
|
|
"""Permit context headers only from trusted internal callers.
|
|
|
|
Public users must not be able to switch tenant/branch/FY by adding
|
|
X-Tenant-Code, X-Branch-Code or X-Year-Code headers. The production-safe
|
|
default is TRUST_CONTEXT_HEADERS=false.
|
|
"""
|
|
if not bool(getattr(self.s, "TRUST_CONTEXT_HEADERS", False)):
|
|
return False
|
|
|
|
required_secret = (getattr(self.s, "CONTEXT_HEADER_SECRET", "") or "").strip()
|
|
if required_secret:
|
|
supplied_secret = (request.headers.get(_CONTEXT_SECRET_HEADER) or "").strip()
|
|
if supplied_secret != required_secret:
|
|
return False
|
|
elif _safe_env(getattr(self.s, "ENV", "")) in {"prod", "production"}:
|
|
return False
|
|
|
|
client_host = request.client.host if request.client else ""
|
|
trusted_entries = _csv_values(getattr(self.s, "TRUST_CONTEXT_HEADER_HOSTS", ""))
|
|
return any(_host_matches_trusted_entry(client_host, item) for item in trusted_entries)
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
# Trusted production context priority:
|
|
# 1) Authenticated UI session selected tenant/branch/FY.
|
|
# 2) Domain resolver mapping for pre-login/domain-routed requests.
|
|
# 3) Trusted internal headers only when explicitly enabled with secret/host.
|
|
# 4) Application defaults.
|
|
session = request.scope.get("session") or {}
|
|
trust_headers = self._context_headers_are_trusted(request)
|
|
|
|
session_tenant_id = _normalise_session_int(session.get("active_tenant_id") or session.get("tenant_id"))
|
|
session_branch_id = _normalise_session_int(session.get("active_branch_id") or session.get("branch_id"))
|
|
|
|
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_code = getattr(request.state, "domain_tenant_code", None)
|
|
domain_branch_code = getattr(request.state, "domain_branch_code", None)
|
|
|
|
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")
|
|
or (request.headers.get(_YEAR_HEADER) if trust_headers else None)
|
|
or self.s.DEFAULT_YEAR_CODE
|
|
)
|
|
|
|
request.state.active_tenant_id = session_tenant_id
|
|
request.state.active_branch_id = session_branch_id
|
|
request.state.tenant_code = tenant_code
|
|
request.state.branch_code = branch_code
|
|
request.state.year_code = year_code
|
|
request.state.context_headers_trusted = trust_headers
|
|
return await call_next(request)
|