app/config.py
Rama main · réplica de lectura del repositorio original.
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlsplit
from dotenv import load_dotenv
def _env_bool(name: str, default: bool) -> bool:
value = os.getenv(name)
if value is None:
return default
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
raise ValueError(f"{name} must be a boolean value")
def _env_int(name: str, default: int) -> int:
value = os.getenv(name)
if value is None:
return default
try:
return int(value)
except ValueError as exc:
raise ValueError(f"{name} must be an integer") from exc
def _env_list(name: str) -> tuple[str, ...]:
return tuple(item.strip() for item in os.getenv(name, "").split(",") if item.strip())
def provider_key_is_configured(value: str | None) -> bool:
"""Return False for empty/example credentials so they never reach an upstream."""
normalized = (value or "").strip().casefold()
if not normalized:
return False
if normalized in {"replace_me", "replace-me", "changeme", "change_me", "your_key_here"}:
return False
return not normalized.startswith(("replace_", "replace-", "your_", "your-"))
@dataclass(frozen=True, slots=True)
class Settings:
app_env: str
app_host: str
app_port: int
llm_provider: str
llm_base_url: str
llm_api_key: str
llm_default_model: str
gemini_base_url: str
gemini_api_key: str
gemini_model: str
upstream_timeout_seconds: float
masking_mode: str
mapping_ttl_seconds: int
conversation_ttl_seconds: int
conversation_max_messages: int
conversation_max_chars: int
public_email_allowlist: tuple[str, ...]
public_email_domain_allowlist: tuple[str, ...]
public_person_allowlist: tuple[str, ...]
enable_debug_endpoints: bool
log_level: str
block_api_keys: bool
block_secrets: bool
block_credit_cards: bool
enable_playground: bool
policy_file: Path
api_keys: tuple[str, ...] = ()
max_request_body_bytes: int = 1_048_576
rate_limit_requests: int = 60
rate_limit_window_seconds: int = 60
require_auth: bool = False
trusted_hosts: tuple[str, ...] = ("localhost", "127.0.0.1", "testserver")
max_media_file_bytes: int = 10 * 1024 * 1024
media_max_concurrency: int = 2
conversation_max_count: int = 1_000
max_upstream_response_bytes: int = 8 * 1024 * 1024
conversation_max_mappings: int = 5_000
conversation_max_sensitive_bytes: int = 2 * 1024 * 1024
application_id: str = "maskgate"
jurisdiction: str = "unspecified"
default_purpose: str = "remote_llm_processing"
detector_profile: str = "strict"
policy_v2_file: Path | None = None
def __post_init__(self) -> None:
if self.is_production and not self.require_auth:
raise ValueError("REQUIRE_AUTH must be true in production")
if self.require_auth and not self.api_keys:
raise ValueError(
"MASKGATE_API_KEYS must contain at least one key when REQUIRE_AUTH is enabled"
)
if self.is_production and self.enable_debug_endpoints:
raise ValueError("ENABLE_DEBUG_ENDPOINTS must be false in production")
if self.is_production and self.enable_playground:
raise ValueError("ENABLE_PLAYGROUND must be false in production")
if self.is_production and "*" in self.trusted_hosts:
raise ValueError("TRUSTED_HOSTS cannot contain '*' in production")
if self.is_production:
provider_name = self.llm_provider.strip().casefold()
provider_secret = (
self.gemini_api_key or self.llm_api_key
if provider_name == "gemini"
else self.llm_api_key
)
provider_url = self.gemini_base_url if provider_name == "gemini" else self.llm_base_url
if not provider_key_is_configured(provider_secret):
raise ValueError("A real provider API key is required in production")
parsed = urlsplit(provider_url)
if (
parsed.scheme.casefold() != "https"
or not parsed.hostname
or parsed.username is not None
or parsed.password is not None
or parsed.query
or parsed.fragment
):
raise ValueError(
"The production provider URL must be HTTPS without credentials, "
"query, or fragment"
)
@property
def is_production(self) -> bool:
return self.app_env.strip().casefold() in {"prod", "production"}
@classmethod
def from_env(cls) -> "Settings":
# Loading .env is intentionally limited to configuration; no request data is
# ever persisted by this application.
load_dotenv(override=False)
mode = os.getenv("MASKING_MODE", "placeholder").strip().lower()
if mode not in {"placeholder", "semantic_placeholder", "surrogate", "redact"}:
mode = "placeholder"
app_env = os.getenv("APP_ENV", "local")
return cls(
app_env=app_env,
app_host=os.getenv("APP_HOST", "0.0.0.0"),
app_port=max(_env_int("APP_PORT", 8080), 1),
llm_provider=os.getenv("LLM_PROVIDER", "openai"),
llm_base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1").rstrip("/"),
llm_api_key=os.getenv("LLM_API_KEY", ""),
llm_default_model=os.getenv("LLM_DEFAULT_MODEL", "gpt-4.1-mini"),
gemini_base_url=os.getenv(
"GEMINI_BASE_URL", "https://generativelanguage.googleapis.com/v1beta"
).rstrip("/"),
gemini_api_key=os.getenv("GEMINI_API_KEY", ""),
gemini_model=os.getenv("GEMINI_MODEL", "gemini-2.5-flash"),
upstream_timeout_seconds=max(_env_int("UPSTREAM_TIMEOUT_SECONDS", 120), 1),
masking_mode=mode,
mapping_ttl_seconds=max(_env_int("MAPPING_TTL_SECONDS", 3600), 1),
conversation_ttl_seconds=max(_env_int("CONVERSATION_TTL_SECONDS", 1800), 1),
conversation_max_messages=max(_env_int("CONVERSATION_MAX_MESSAGES", 40), 4),
conversation_max_chars=max(_env_int("CONVERSATION_MAX_CHARS", 120000), 1000),
public_email_allowlist=_env_list("PUBLIC_EMAIL_ALLOWLIST"),
public_email_domain_allowlist=_env_list("PUBLIC_EMAIL_DOMAIN_ALLOWLIST"),
public_person_allowlist=_env_list("PUBLIC_PERSON_ALLOWLIST"),
enable_debug_endpoints=_env_bool("ENABLE_DEBUG_ENDPOINTS", False),
log_level=os.getenv("LOG_LEVEL", "INFO").upper(),
block_api_keys=_env_bool("BLOCK_API_KEYS", True),
block_secrets=_env_bool("BLOCK_SECRETS", True),
block_credit_cards=_env_bool("BLOCK_CREDIT_CARDS", False),
enable_playground=_env_bool("ENABLE_PLAYGROUND", app_env == "local"),
policy_file=Path(__file__).parent / "policies" / "default_policy.yaml",
api_keys=_env_list("MASKGATE_API_KEYS"),
max_request_body_bytes=max(_env_int("MAX_REQUEST_BODY_BYTES", 1_048_576), 1_024),
rate_limit_requests=max(_env_int("RATE_LIMIT_REQUESTS", 60), 1),
rate_limit_window_seconds=max(_env_int("RATE_LIMIT_WINDOW_SECONDS", 60), 1),
require_auth=_env_bool(
"REQUIRE_AUTH",
app_env.strip().casefold() in {"prod", "production"},
),
trusted_hosts=_env_list("TRUSTED_HOSTS") or ("localhost", "127.0.0.1", "testserver"),
max_media_file_bytes=max(_env_int("MAX_MEDIA_FILE_BYTES", 10 * 1024 * 1024), 1_024),
media_max_concurrency=max(_env_int("MEDIA_MAX_CONCURRENCY", 2), 1),
conversation_max_count=max(_env_int("CONVERSATION_MAX_COUNT", 1_000), 1),
max_upstream_response_bytes=max(
_env_int("MAX_UPSTREAM_RESPONSE_BYTES", 8 * 1024 * 1024),
1_024,
),
conversation_max_mappings=max(_env_int("CONVERSATION_MAX_MAPPINGS", 5_000), 1),
conversation_max_sensitive_bytes=max(
_env_int("CONVERSATION_MAX_SENSITIVE_BYTES", 2 * 1024 * 1024),
1_024,
),
application_id=os.getenv("MASKGATE_APPLICATION_ID", "maskgate").strip()
or "maskgate",
jurisdiction=os.getenv("MASKGATE_JURISDICTION", "unspecified").strip()
or "unspecified",
default_purpose=os.getenv(
"MASKGATE_DEFAULT_PURPOSE", "remote_llm_processing"
).strip()
or "remote_llm_processing",
detector_profile=(
os.getenv("DETECTOR_PROFILE", "strict").strip().casefold()
if os.getenv("DETECTOR_PROFILE", "strict").strip().casefold()
in {"fast", "balanced", "strict"}
else "strict"
),
policy_v2_file=(
Path(os.environ["POLICY_V2_FILE"])
if os.getenv("POLICY_V2_FILE", "").strip()
else None
),
)