d17462edc4
- llm.py: _create() wrapper retries APIConnectionError (incl. timeout) up to LLM_MAX_ATTEMPTS=3 with LLM_RETRY_DELAY_SECONDS=5s backoff; 4xx still fails fast. The local LLM server is shared with Hermes and serializes requests, so a request can be dropped by the client timeout while queued — chat/send was 500ing on that. - helpers.py: internal_error() now logs full exception detail server-side (was only the type name), so 500s are diagnosable; client message unchanged. - config.py: LLM_MAX_ATTEMPTS / LLM_RETRY_DELAY_SECONDS env knobs. - tests: test_llm_retry_queue (transient retried, 4xx not retried, no retry on success), test_internal_error_logging (detail in log, hidden from client, order-independent via explicit handler + logger re-enable).
240 lines
10 KiB
Python
240 lines
10 KiB
Python
"""Configuration from environment / .env."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
|
|
# Load .env from backend/ (project root for this app)
|
|
_BACKEND_DIR = Path(__file__).resolve().parent.parent
|
|
load_dotenv(_BACKEND_DIR / ".env", override=False)
|
|
|
|
|
|
def _get_bool(name: str, default: bool = False) -> bool:
|
|
raw = os.environ.get(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _get_int(
|
|
name: str,
|
|
default: int,
|
|
*,
|
|
minimum: int | None = None,
|
|
maximum: int | None = None,
|
|
) -> int:
|
|
"""Read an optional integer without breaking import on bad environment input."""
|
|
try:
|
|
value = int(os.environ.get(name, str(default)))
|
|
except (TypeError, ValueError):
|
|
value = default
|
|
if minimum is not None and value < minimum:
|
|
return default
|
|
if maximum is not None and value > maximum:
|
|
return default
|
|
return value
|
|
|
|
|
|
def _get_float(name: str, default: float, *, minimum: float | None = None) -> float:
|
|
"""Read an optional float with a safe fallback for malformed deployment config."""
|
|
try:
|
|
value = float(os.environ.get(name, str(default)))
|
|
except (TypeError, ValueError):
|
|
value = default
|
|
if minimum is not None and value < minimum:
|
|
return default
|
|
return value
|
|
|
|
|
|
def resolve_llm() -> tuple[str, str, str, str | None]:
|
|
"""Return (base_url, model, api_key, provider_name)."""
|
|
provider = os.environ.get("LLM_PROVIDER", "").strip()
|
|
explicit_base = os.environ.get("LLM_BASE_URL", "").strip()
|
|
explicit_model = os.environ.get("LLM_MODEL_NAME", "").strip()
|
|
api_key = os.environ.get("LLM_API_KEY", "").strip() or None
|
|
|
|
presets = {
|
|
"deepseek": ("https://api.deepseek.com/v1", "deepseek-chat"),
|
|
"openai": ("https://api.openai.com/v1", "gpt-4o-mini"),
|
|
"custom": ("", ""),
|
|
}
|
|
if provider and provider in presets:
|
|
base_url, model = presets[provider]
|
|
if explicit_base:
|
|
base_url = explicit_base
|
|
if explicit_model:
|
|
model = explicit_model
|
|
return base_url, model, api_key or "", provider
|
|
# No/unknown provider: fall back to explicit config
|
|
return (
|
|
explicit_base or "https://api.openai.com/v1",
|
|
explicit_model or "gpt-4o-mini",
|
|
api_key or "",
|
|
provider or None,
|
|
)
|
|
|
|
|
|
class Config:
|
|
APP_NAME = "Sales Trainer"
|
|
# Secure-by-default: deployments must opt into development/test explicitly.
|
|
APP_ENV = os.environ.get("APP_ENV", "production").strip().lower()
|
|
BOOTSTRAP_ADMIN_PASSWORD = os.environ.get("BOOTSTRAP_ADMIN_PASSWORD", "").strip()
|
|
SECRET_KEY = os.environ.get("JWT_SECRET", "").strip()
|
|
JWT_ALGO = "HS256"
|
|
JWT_MIN_EXPIRES_HOURS = 1
|
|
JWT_MAX_EXPIRES_HOURS = 24 * 30
|
|
JWT_EXPIRES_HOURS = _get_int(
|
|
"JWT_EXPIRES_HOURS",
|
|
24,
|
|
minimum=JWT_MIN_EXPIRES_HOURS,
|
|
maximum=JWT_MAX_EXPIRES_HOURS,
|
|
)
|
|
MIN_PASSWORD_LENGTH = 12
|
|
|
|
# Demo accounts are isolated in their own tenant and get a fixed UTC trial
|
|
# that starts on first successful password login. Keep the defaults safe for
|
|
# local development; production should set the tenant id explicitly.
|
|
DEMO_ORG_ID = os.environ.get("DEMO_ORG_ID", "org-demo").strip() or "org-demo"
|
|
DEMO_ORG_NAME = os.environ.get("DEMO_ORG_NAME", "Sales Trainer Demo").strip() or "Sales Trainer Demo"
|
|
DEMO_ORG_SEATS = _get_int("DEMO_ORG_SEATS", 10000, minimum=1)
|
|
DEMO_TRIAL_DAYS = 30
|
|
|
|
# Demo credential delivery is fail-closed and opt-in. Tests/local installs
|
|
# remain usable when neither SES nor the webhook is configured.
|
|
AWS_REGION = os.environ.get("AWS_REGION", "").strip()
|
|
SES_FROM_EMAIL = os.environ.get("SES_FROM_EMAIL", "").strip()
|
|
SES_REPLY_TO = os.environ.get("SES_REPLY_TO", "").strip()
|
|
DEMO_WEBHOOK_URL = os.environ.get("DEMO_WEBHOOK_URL", "").strip()
|
|
DEMO_WEBHOOK_SECRET = os.environ.get("DEMO_WEBHOOK_SECRET", "").strip()
|
|
DEMO_WEBHOOK_TIMEOUT_SECONDS = _get_float(
|
|
"DEMO_WEBHOOK_TIMEOUT_SECONDS", 5.0, minimum=0.1
|
|
)
|
|
DEMO_WEBHOOK_MAX_RETRIES = _get_int("DEMO_WEBHOOK_MAX_RETRIES", 3, minimum=1)
|
|
DEMO_WEBHOOK_RETRY_BACKOFF_SECONDS = _get_float(
|
|
"DEMO_WEBHOOK_RETRY_BACKOFF_SECONDS", 0.25, minimum=0.0
|
|
)
|
|
|
|
DATA_DIR = Path(
|
|
os.environ.get("DATA_DIR", str(_BACKEND_DIR / "data"))
|
|
).resolve()
|
|
DATABASE_URL = os.environ.get("DATABASE_URL", "").strip()
|
|
|
|
FLASK_HOST = os.environ.get("FLASK_HOST", "0.0.0.0")
|
|
FLASK_PORT = int(os.environ.get("FLASK_PORT", "5001"))
|
|
FLASK_DEBUG = _get_bool("FLASK_DEBUG", False)
|
|
CORS_ORIGINS = tuple(
|
|
origin.strip()
|
|
for origin in os.environ.get("CORS_ORIGINS", "").split(",")
|
|
if origin.strip()
|
|
)
|
|
|
|
UPLOAD_MAX_MB = int(os.environ.get("UPLOAD_MAX_MB", "15"))
|
|
ALLOWED_UPLOAD_EXTS = {"pdf", "md", "txt", "docx", "xlsx"}
|
|
UPLOAD_TEXT_MAX_KB = int(os.environ.get("UPLOAD_TEXT_MAX_KB", "256"))
|
|
UPLOAD_TEXT_MAX_BYTES = UPLOAD_TEXT_MAX_KB * 1024
|
|
UPLOAD_MAX_PDF_PAGES = int(os.environ.get("UPLOAD_MAX_PDF_PAGES", "100"))
|
|
UPLOAD_MAX_EXTRACTED_CHARS = int(os.environ.get("UPLOAD_MAX_EXTRACTED_CHARS", "60000"))
|
|
UPLOAD_MAX_FILES = int(os.environ.get("UPLOAD_MAX_FILES", "10"))
|
|
UPLOAD_MAX_PDF_CHUNK_CHARS = int(os.environ.get("UPLOAD_MAX_PDF_CHUNK_CHARS", "8192"))
|
|
UPLOAD_MAX_XLSX_SHEETS = int(os.environ.get("UPLOAD_MAX_XLSX_SHEETS", "20"))
|
|
UPLOAD_MAX_XLSX_ROWS = int(os.environ.get("UPLOAD_MAX_XLSX_ROWS", "5000"))
|
|
|
|
ANALYTICS_EXPORT_MAX_ROWS = int(os.environ.get("ANALYTICS_EXPORT_MAX_ROWS", "10000"))
|
|
ANALYTICS_EXPORT_MAX_BYTES = int(
|
|
os.environ.get("ANALYTICS_EXPORT_MAX_BYTES", str(4 * 1024 * 1024))
|
|
)
|
|
ANALYTICS_EXPORT_MAX_CELL_CHARS = int(
|
|
os.environ.get("ANALYTICS_EXPORT_MAX_CELL_CHARS", "10000")
|
|
)
|
|
ANALYTICS_EXPORT_MAX_SCAN_RECORDS = int(
|
|
os.environ.get("ANALYTICS_EXPORT_MAX_SCAN_RECORDS", "100000")
|
|
)
|
|
|
|
# LLM
|
|
LLM_BASE_URL, LLM_MODEL, LLM_API_KEY, LLM_PROVIDER = resolve_llm()
|
|
# The local LLM server is shared (also serves Hermes): a concurrent long
|
|
# generation serializes requests, so a request can sit in the server queue
|
|
# well beyond a single attempt. Retry transient connection/timeout failures
|
|
# a few times so chat survives the queue instead of failing with 500.
|
|
LLM_MAX_ATTEMPTS = _get_int("LLM_MAX_ATTEMPTS", 3, minimum=1)
|
|
LLM_RETRY_DELAY_SECONDS = _get_float(
|
|
"LLM_RETRY_DELAY_SECONDS", 5.0, minimum=0.0
|
|
)
|
|
|
|
# OAuth (Google + Facebook) — all optional; OAuth is disabled unless creds
|
|
# are fully configured (fail closed). Client IDs / app IDs are public and
|
|
# may be exposed to the frontend; the *secrets* must never be.
|
|
OAUTH_GOOGLE_CLIENT_ID = os.environ.get("OAUTH_GOOGLE_CLIENT_ID", "").strip()
|
|
OAUTH_GOOGLE_CLIENT_SECRET = os.environ.get("OAUTH_GOOGLE_CLIENT_SECRET", "").strip()
|
|
OAUTH_FACEBOOK_APP_ID = os.environ.get("OAUTH_FACEBOOK_APP_ID", "").strip()
|
|
OAUTH_FACEBOOK_APP_SECRET = os.environ.get("OAUTH_FACEBOOK_APP_SECRET", "").strip()
|
|
# Tenant id that public social signups land in. Missing/placeholder disables OAuth.
|
|
OAUTH_DEFAULT_ORG = os.environ.get("OAUTH_DEFAULT_ORG", "").strip()
|
|
|
|
_OAUTH_PROVIDER_CREDS = {
|
|
"google": ("OAUTH_GOOGLE_CLIENT_ID", "OAUTH_GOOGLE_CLIENT_SECRET"),
|
|
"facebook": ("OAUTH_FACEBOOK_APP_ID", "OAUTH_FACEBOOK_APP_SECRET"),
|
|
}
|
|
|
|
ROLES = ("super_admin", "admin", "user", "demo")
|
|
|
|
@classmethod
|
|
def oauth_provider_enabled(cls, provider: str) -> bool:
|
|
"""A provider is enabled only when every one of its creds + the default
|
|
org are configured and not a placeholder. Fail closed otherwise."""
|
|
names = cls._OAUTH_PROVIDER_CREDS.get(provider)
|
|
if names is None or cls._is_placeholder(cls.OAUTH_DEFAULT_ORG):
|
|
return False
|
|
for name in names:
|
|
if cls._is_placeholder(getattr(cls, name)):
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def _is_placeholder(value: str) -> bool:
|
|
normalized = (value or "").strip().lower()
|
|
return (
|
|
not normalized
|
|
or normalized in {"replace_me", "changeme", "change_me", "default", "password"}
|
|
or normalized.startswith("replace_with_")
|
|
or normalized.startswith("your_")
|
|
)
|
|
|
|
@staticmethod
|
|
def _has_secret_diversity(value: str) -> bool:
|
|
"""Reject trivially repeated secrets that satisfy length checks only."""
|
|
if not isinstance(value, str) or len(set(value)) < 4:
|
|
return False
|
|
for period in range(1, len(value) // 2 + 1):
|
|
if len(value) % period == 0 and value == value[:period] * (len(value) // period):
|
|
return False
|
|
return True
|
|
|
|
@classmethod
|
|
def validate_runtime_security(cls, *, require_bootstrap: bool = False) -> None:
|
|
"""Fail closed for production secrets and first-run initialization."""
|
|
secure_runtime = cls.APP_ENV not in {"development", "test"} or not cls.FLASK_DEBUG
|
|
if secure_runtime and (
|
|
cls._is_placeholder(cls.SECRET_KEY)
|
|
or len(cls.SECRET_KEY) < 32
|
|
or not cls._has_secret_diversity(cls.SECRET_KEY)
|
|
):
|
|
raise RuntimeError("JWT_SECRET must be configured with at least 32 characters")
|
|
if require_bootstrap:
|
|
if cls._is_placeholder(cls.BOOTSTRAP_ADMIN_PASSWORD):
|
|
raise RuntimeError("BOOTSTRAP_ADMIN_PASSWORD is required to initialize the first admin")
|
|
if (
|
|
len(cls.BOOTSTRAP_ADMIN_PASSWORD) < cls.MIN_PASSWORD_LENGTH
|
|
or not cls._has_secret_diversity(cls.BOOTSTRAP_ADMIN_PASSWORD)
|
|
):
|
|
raise RuntimeError(
|
|
"BOOTSTRAP_ADMIN_PASSWORD must be at least "
|
|
f"{cls.MIN_PASSWORD_LENGTH} characters"
|
|
)
|
|
|
|
@classmethod
|
|
def ensure_dirs(cls) -> None:
|
|
for name in ("users", "orgs", "groups", "sessions", "demo_deliveries"):
|
|
(cls.DATA_DIR / name).mkdir(parents=True, exist_ok=True)
|