feat(core): enforce tenant isolation guard outside production
Non-production environments (development/testing) now force the effective tenant to the reserved test tenant "llm_tester" (or a test_-prefixed override) regardless of DEFAULT_USER misconfiguration: - Config.effective_default_user only honors DEFAULT_USER outside production when it is llm_tester or test_-prefixed; anything else is forced to llm_tester (tenant_forced flags the override) - Config refuses startup (validation error) when a non-production environment is explicitly configured with the production tenant jpmschweitzer - get_user() applies the same guard at request-context resolution, so an explicit request for the production tenant in dev/test is forced to llm_tester with a warning log - initialize_application() emits one loud startup log line (tenant_guard_active / tenant_guard_production) stating the effective tenant Unit tests cover the dev/test/prod x default/explicit-user matrix. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
+55
-8
@@ -6,9 +6,17 @@ from enum import Enum
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
|
||||
from pydantic import Field, HttpUrl
|
||||
from pydantic import Field, HttpUrl, model_validator
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
# Tenant isolation constants (see docs: tenant-based isolation, no separate
|
||||
# test infrastructure). The production tenant owns real data in the shared
|
||||
# services (Qdrant/Neo4j/Wiki.js/Redis); everything non-production must run
|
||||
# under the reserved test tenant or an explicit test_-prefixed namespace.
|
||||
PRODUCTION_TENANT = "jpmschweitzer"
|
||||
TEST_TENANT = "llm_tester"
|
||||
TEST_TENANT_PREFIX = "test_"
|
||||
|
||||
|
||||
def _get_version_from_pyproject() -> str:
|
||||
"""
|
||||
@@ -207,6 +215,29 @@ class Config(BaseSettings):
|
||||
CORS_ALLOW_METHODS: list[str] = ["*"]
|
||||
CORS_ALLOW_HEADERS: list[str] = ["*"]
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _refuse_production_tenant_outside_production(self) -> "Config":
|
||||
"""
|
||||
Refuse startup when a non-production environment is explicitly
|
||||
configured with the production tenant.
|
||||
|
||||
This is the hard stop of the tenant isolation guard: a dev/test
|
||||
instance must never be able to read or write the production
|
||||
tenant's data in the shared services.
|
||||
"""
|
||||
if (
|
||||
self.ENVIRONMENT != Environment.PRODUCTION
|
||||
and self.DEFAULT_USER == PRODUCTION_TENANT
|
||||
):
|
||||
raise ValueError(
|
||||
f"Refusing to start: ENVIRONMENT={self.ENVIRONMENT.value} is "
|
||||
f"explicitly configured with the production tenant "
|
||||
f"'{PRODUCTION_TENANT}'. Non-production environments must use "
|
||||
f"'{TEST_TENANT}' or a '{TEST_TENANT_PREFIX}'-prefixed tenant. "
|
||||
f"Unset DEFAULT_USER or set ENVIRONMENT=production."
|
||||
)
|
||||
return self
|
||||
|
||||
@property
|
||||
def redis_memory_url(self) -> str:
|
||||
"""Construct Redis connection URL for memory cache."""
|
||||
@@ -247,16 +278,32 @@ class Config(BaseSettings):
|
||||
@property
|
||||
def effective_default_user(self) -> str:
|
||||
"""
|
||||
Get effective default user, auto-determining from environment if not set.
|
||||
Get effective default user (tenant), enforcing tenant isolation.
|
||||
|
||||
- development/testing: llm_tester (isolated test scope)
|
||||
- production: jpmschweitzer (real user)
|
||||
- production: DEFAULT_USER if set, else the production tenant
|
||||
- development/testing: FORCED to the reserved test tenant
|
||||
("llm_tester") - the only accepted overrides are the test tenant
|
||||
itself or a "test_"-prefixed namespace. Any other DEFAULT_USER
|
||||
value is treated as misconfiguration and ignored.
|
||||
"""
|
||||
if self.DEFAULT_USER is not None:
|
||||
return self.DEFAULT_USER
|
||||
if self.ENVIRONMENT == Environment.PRODUCTION:
|
||||
return "jpmschweitzer"
|
||||
return "llm_tester"
|
||||
return self.DEFAULT_USER or PRODUCTION_TENANT
|
||||
|
||||
if self.DEFAULT_USER is not None and (
|
||||
self.DEFAULT_USER == TEST_TENANT
|
||||
or self.DEFAULT_USER.startswith(TEST_TENANT_PREFIX)
|
||||
):
|
||||
return self.DEFAULT_USER
|
||||
return TEST_TENANT
|
||||
|
||||
@property
|
||||
def tenant_forced(self) -> bool:
|
||||
"""Whether the tenant guard overrode a misconfigured DEFAULT_USER."""
|
||||
return (
|
||||
self.ENVIRONMENT != Environment.PRODUCTION
|
||||
and self.DEFAULT_USER is not None
|
||||
and self.effective_default_user != self.DEFAULT_USER
|
||||
)
|
||||
|
||||
|
||||
@lru_cache
|
||||
|
||||
Reference in New Issue
Block a user