loyaly-catalogue/backend/app/infrastructure/settings.py

"""
Centralized configuration for the Electronics Catalog backend.

Every credential is read ONLY from the environment (backend/.env via
python-dotenv, or real OS variables). Non-secret values keep safe local
defaults.

LOCAL-ONLY GUARD
----------------
This project is a copy of the grocery catalogue, whose .env files pointed at a
remote production database. To make it impossible to write electronics data
there by accident, settings refuse to load unless DB_HOST is a local host and
DB_NAME is the dedicated electronics database. See _guard_local_database().
The one exception is an explicit production opt-in (ELEC_ALLOW_REMOTE_DB plus
an exact host/database allowlist), used only by the production deployment.
"""
from __future__ import annotations

import os
from pathlib import Path

# Snapshotted BEFORE load_dotenv, and that ordering is the entire point.
# load_dotenv() is called without override=True, so a variable already in the
# process environment silently beats the .env file and keeps beating it no
# matter how many times the file is corrected. That is not hypothetical here:
# the deployment platform injects its Environment tab into the container, so a
# stale value left in that tab overrides the credentials baked into the image
# (backend/Dockerfile copies .env.production to /app/.env) and the only symptom
# is a 401 that nothing explains. Comparing a name against this set answers
# "which of the two won?" - see config_source() below.
_PREEXISTING_ENV = frozenset(os.environ)

try:
    from dotenv import load_dotenv

    # backend/.env  (one level up from this file: app/infrastructure/settings.py)
    _env_path = Path(__file__).resolve().parents[2] / ".env"
    load_dotenv(_env_path)
except ImportError:
    # python-dotenv not installed - fall back to whatever is already in the
    # process environment (e.g. set by the shell, Docker, systemd, CI, etc.)
    pass


# Names whose raw value arrived wrapped in quotes or padded with whitespace.
# Recorded rather than merely fixed: stripping keeps the login working, but the
# only place the original shape is still visible is right here, before the value
# is normalised. A quoted hash is the signature of a value pasted into a web
# form, so surfacing it at startup is what stops the next person rediscovering
# it from a 401. See DB_PASSWORD in .env.production for the counter-case where
# the quotes ARE part of the secret - which is why this warns, and does not fail.
_ENV_NEEDED_CLEANUP = set()


def _clean(name: str, raw: str) -> str:
    """Strip surrounding quotes/whitespace off an env value, remembering if it mattered."""
    cleaned = raw.strip().strip("'\"")
    if cleaned != raw:
        _ENV_NEEDED_CLEANUP.add(name)
    return cleaned


def cleaned_env_names() -> list:
    """Which settings needed quote/whitespace stripping. Reported at startup."""
    return sorted(_ENV_NEEDED_CLEANUP)


def config_source(name: str) -> str:
    """
    Where a setting's value actually came from: the process environment, the
    .env file, or this module's own default.

    Reported at startup for the AUTH_* values (see app/main.py) so that an
    override arriving from outside the image is visible in the logs instead of
    being inferred from a failing login.
    """
    if name in _PREEXISTING_ENV:
        return "process-env"
    if name in os.environ:
        return "env-file"
    return "default"


def _bool(name: str, default: str) -> bool:
    return os.getenv(name, default).strip().lower() in {"1", "true", "yes"}


def _require(name: str, *, feature_flag: str) -> str:
    """Read a required secret. Raises if missing and the owning feature is enabled."""
    value = os.getenv(name)
    if not value:
        raise RuntimeError(
            f"Missing required environment variable '{name}'. It is required because "
            f"'{feature_flag}' is enabled. Set it in backend/.env (copy from "
            f".env.example) or disable the feature by setting {feature_flag}=false."
        )
    return value


# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
_BACKEND_ROOT = Path(__file__).resolve().parents[2]
DATA_DIR = Path(os.getenv("DATA_DIR", "").strip() or _BACKEND_ROOT / "data")

# ---------------------------------------------------------------------------
# Ollama (local LLM) - only ever used to read text we fetched, never to invent
# ---------------------------------------------------------------------------
USE_OLLAMA = _bool("USE_OLLAMA", "true")
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
OLLAMA_MODEL_NAME = os.getenv("OLLAMA_MODEL_NAME", "qwen2.5:1.5b")
OLLAMA_TIMEOUT_SECONDS = int(os.getenv("OLLAMA_TIMEOUT_SECONDS", "120"))

# ---------------------------------------------------------------------------
# Embeddings (sentence-transformers, CPU-friendly)
# ---------------------------------------------------------------------------
USE_EMBEDDINGS = _bool("USE_EMBEDDINGS", "true")
EMBEDDINGS_MODEL = os.getenv("EMBEDDINGS_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
EMBEDDINGS_DIM = int(os.getenv("EMBEDDINGS_DIM", "384"))

# ---------------------------------------------------------------------------
# Postgres / pgvector - the LOCAL electronics database only
# ---------------------------------------------------------------------------
ELECTRONICS_DB_NAME = "electronics_catalog"
# The test suite uses its own database on the same local server.
ALLOWED_DB_NAMES = frozenset({ELECTRONICS_DB_NAME, ELECTRONICS_DB_NAME + "_test"})
LOCAL_DB_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "host.docker.internal", "postgres"})

DB_HOST = os.getenv("DB_HOST", "127.0.0.1").strip()
DB_PORT = os.getenv("DB_PORT", "5433").strip()
DB_NAME = os.getenv("DB_NAME", ELECTRONICS_DB_NAME).strip()
DB_USER = os.getenv("DB_USER", "postgres").strip()
DB_PASSWORD = _require("DB_PASSWORD", feature_flag="the electronics database")
DB_CONNECT_TIMEOUT_SECONDS = int(os.getenv("DB_CONNECT_TIMEOUT_SECONDS", "5"))


def _csv_set(name: str) -> frozenset:
    return frozenset(v.strip() for v in os.getenv(name, "").split(",") if v.strip())


# Production opt-in. Off by default: without ELEC_ALLOW_REMOTE_DB=true the
# guard below behaves exactly as it always has. With it on, only the host(s)
# and database name(s) listed here are accepted - never "any remote host".
ELEC_ALLOW_REMOTE_DB = _bool("ELEC_ALLOW_REMOTE_DB", "false")
ELEC_REMOTE_DB_HOSTS = _csv_set("ELEC_REMOTE_DB_HOSTS")
ELEC_REMOTE_DB_NAMES = _csv_set("ELEC_REMOTE_DB_NAMES")


def _guard_local_database(host: str, name: str, *, allow_remote: bool = False,
                          remote_hosts: frozenset = frozenset(), remote_names: frozenset = frozenset()) -> None:
    """Refuse to run against anything but the local electronics database,
    unless the production opt-in names this exact host and database."""
    if allow_remote and host in remote_hosts:
        if name not in remote_names:
            raise RuntimeError(
                f"DB_NAME={name!r} is not in ELEC_REMOTE_DB_NAMES for remote host {host!r}."
            )
        return
    if host not in LOCAL_DB_HOSTS:
        raise RuntimeError(
            f"DB_HOST={host!r} is not a local host. This project only runs against the "
            f"local Docker database (see docker-compose.yml); it must never touch the "
            f"remote catalogue database."
        )
    if name not in ALLOWED_DB_NAMES:
        raise RuntimeError(
            f"DB_NAME={name!r}; expected {ELECTRONICS_DB_NAME!r}. The electronics data "
            f"lives in its own database so existing databases are never modified."
        )


_guard_local_database(DB_HOST, DB_NAME, allow_remote=ELEC_ALLOW_REMOTE_DB,
                      remote_hosts=ELEC_REMOTE_DB_HOSTS, remote_names=ELEC_REMOTE_DB_NAMES)

# ---------------------------------------------------------------------------
# Web search (discovery, prices, images)
# ---------------------------------------------------------------------------
# DuckDuckGo needs no key. Google Programmable Search is used in addition when
# both GOOGLE_API_KEY and GOOGLE_CSE_ID are set (100 free queries/day).
USE_DDG_SEARCH = _bool("USE_DDG_SEARCH", "true")
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY", "").strip()
GOOGLE_CSE_ID = os.getenv("GOOGLE_CSE_ID", "").strip()
USE_GOOGLE_CSE = bool(GOOGLE_API_KEY and GOOGLE_CSE_ID) and _bool("USE_GOOGLE_CSE", "true")
GOOGLE_CSE_DAILY_QUOTA = int(os.getenv("GOOGLE_CSE_DAILY_QUOTA", "100"))
SEARCH_REGION = os.getenv("SEARCH_REGION", "in-en")
# Minimum pause between two search queries to the same provider.
SEARCH_MIN_INTERVAL_SECONDS = float(os.getenv("SEARCH_MIN_INTERVAL_SECONDS", "2.5"))
SEARCH_CACHE_TTL_HOURS = int(os.getenv("SEARCH_CACHE_TTL_HOURS", "24"))

# ---------------------------------------------------------------------------
# Polite fetching of retailer / brand pages
# ---------------------------------------------------------------------------
# An honest User-Agent with a contact address. Set ELEC_CONTACT to a real
# address before running a crawl.
ELEC_CONTACT = os.getenv("ELEC_CONTACT", "admin@example.com").strip()
USER_AGENT = os.getenv(
    "USER_AGENT", f"ElectronicsCatalogBot/0.1 (+mailto:{ELEC_CONTACT}; local research)"
)
REQUEST_TIMEOUT_SECONDS = int(os.getenv("REQUEST_TIMEOUT_SECONDS", "20"))
ELEC_SITE_MIN_INTERVAL_SECONDS = float(os.getenv("ELEC_SITE_MIN_INTERVAL_SECONDS", "3"))
ELEC_BREAKER_COOLDOWN_HOURS = float(os.getenv("ELEC_BREAKER_COOLDOWN_HOURS", "24"))
# Brand stores embed their reviews in the page: samsung.com/in pages run ~3.1 MB.
ELEC_MAX_PAGE_BYTES = int(os.getenv("ELEC_MAX_PAGE_BYTES", str(6 * 1024 * 1024)))
ELEC_PROBE_TTL_DAYS = int(os.getenv("ELEC_PROBE_TTL_DAYS", "7"))
MIN_IMAGE_BYTES = int(os.getenv("MIN_IMAGE_BYTES", "3000"))

# Reference pincodes (Tamil Nadu). "pincode:City" pairs, comma-separated. The
# first one is the default. A pincode is stored against a price only when the
# site actually accepted it.
ELEC_REFERENCE_PINCODES = [
    tuple(p.split(":", 1)) if ":" in p else (p, "")
    for p in (x.strip() for x in os.getenv("ELEC_REFERENCE_PINCODES", "641001:Coimbatore,600001:Chennai").split(","))
    if p
]

# The LLM may only fill spec gaps from text we fetched; set false to run fully
# deterministic.
ELEC_USE_LLM = _bool("ELEC_USE_LLM", "true")

# ---------------------------------------------------------------------------
# FastAPI / web server
# ---------------------------------------------------------------------------
API_CORS_ORIGINS = [
    origin.strip()
    for origin in os.getenv("API_CORS_ORIGINS", "http://localhost:5173,http://127.0.0.1:5173").split(",")
    if origin.strip()
]

# ---------------------------------------------------------------------------
# Authentication
# ---------------------------------------------------------------------------
# CORS above is not access control - browsers enforce it, and curl ignores it
# entirely. These settings are what actually guards the write/compute endpoints
# (catalog generation, ML training, uploads, chat).
#
# AUTH_ENABLED=false turns every guard off, restoring the old behaviour where
# any caller could reach any endpoint. It exists so a fresh checkout still runs
# without generating secrets first; app/infrastructure/security.py logs a
# warning at import when it is off. Never deploy with it off.
AUTH_ENABLED = _bool("AUTH_ENABLED", "true")

# Signs and verifies access tokens. Changing it invalidates every issued token,
# which is the intended way to force everyone to sign in again. Generate with:
#   python scripts/make_auth_secrets.py
AUTH_SECRET_KEY = (
    _require("AUTH_SECRET_KEY", feature_flag="AUTH_ENABLED")
    if AUTH_ENABLED
    else os.getenv("AUTH_SECRET_KEY", "")
)

# How long an issued token stays valid. 12h by default: long enough that a
# working day needs one sign-in, short enough that a leaked token expires.
AUTH_TOKEN_TTL_MINUTES = int(os.getenv("AUTH_TOKEN_TTL_MINUTES", "720"))

# The interactive accounts. Only PBKDF2 digests are stored - never a password.
# `make_auth_secrets.py` prints the lines ready to paste.
#
# `admin` is required whenever auth is on: without it nobody could sign in.
AUTH_ADMIN_USERNAME = _clean(
    "AUTH_ADMIN_USERNAME", os.getenv("AUTH_ADMIN_USERNAME", "admin")
)
AUTH_ADMIN_PASSWORD_HASH = _clean(
    "AUTH_ADMIN_PASSWORD_HASH",
    (
        _require("AUTH_ADMIN_PASSWORD_HASH", feature_flag="AUTH_ENABLED")
        if AUTH_ENABLED
        else os.getenv("AUTH_ADMIN_PASSWORD_HASH", "")
    ),
)

# The second `user` account is OPTIONAL, and left unset in this deployment.
# An empty hash is how the account is switched off: auth.py builds its account
# table from these values and omits any entry whose hash is blank, so there is
# nothing to sign in to. Setting the hash again re-enables it with no code
# change - which is exactly what the test suite does in tests/conftest.py.
AUTH_USER_USERNAME = _clean(
    "AUTH_USER_USERNAME", os.getenv("AUTH_USER_USERNAME", "user")
)
AUTH_USER_PASSWORD_HASH = _clean(
    "AUTH_USER_PASSWORD_HASH", os.getenv("AUTH_USER_PASSWORD_HASH", "")
)

# Failed-login throttle, applied per username+client-IP. Prevents an exposed
# login endpoint from being a free password oracle.
AUTH_MAX_LOGIN_ATTEMPTS = int(os.getenv("AUTH_MAX_LOGIN_ATTEMPTS", "10"))
AUTH_LOCKOUT_SECONDS = int(os.getenv("AUTH_LOCKOUT_SECONDS", "300"))

# Local-development escape hatch: accept ANY password at /api/auth/login, so a
# developer who does not have the configured passwords to hand can still reach
# the admin and user pages. The username still selects the role, and the token
# issued is a normal signed one - so every downstream guard, /api/auth/me, and
# the React route gating all behave exactly as they do in production. What is
# skipped is only the password check.
#
# This is NOT the same as AUTH_ENABLED=false. That disables every guard *and*
# makes /api/auth/login return 503, which breaks the login page outright. This
# flag keeps the whole auth machinery running and unlocks just the front door.
#
# Anyone who can reach the API can sign in as admin while it is on. Keep it
# false anywhere the port is reachable by someone you would not hand the admin
# password to.
AUTH_ALLOW_ANY_LOGIN = _bool("AUTH_ALLOW_ANY_LOGIN", "false")


# Shortest acceptable API key secret. token_urlsafe(32) yields 43 characters, so
# this rejects hand-typed values without rejecting anything the documented
# generator produces.
API_KEY_MIN_LENGTH = 32


def _parse_api_keys(raw: str) -> dict:
    """
    Parse ``API_KEYS`` - ``name:role:secret`` triples, comma-separated.

    Keyed by secret because that is what an inbound request presents. One entry
    per consumer is the point: a shared key cannot be revoked for one caller
    without breaking all of them.

    Secrets must be at least API_KEY_MIN_LENGTH characters. That is not about
    guessing the key over the network - the lockout and the network itself make
    online brute force impractical - but about what /api/health publishes. It
    reports a truncated digest of every configured key so a deployment can be
    checked against the config it was built from, and a digest of a *raw* secret
    is only safe when the secret is unguessable offline. An admin password hash
    embeds a random salt, so its fingerprint discloses nothing; an API key has no
    salt, and a hand-picked "changeme" would fall to a wordlist in seconds.
    Generate one with: python -c "import secrets; print(secrets.token_urlsafe(32))"
    """
    parsed: dict = {}
    for entry in raw.split(","):
        entry = entry.strip()
        if not entry:
            continue
        parts = entry.split(":")
        if len(parts) != 3:
            raise RuntimeError(
                f"Malformed API_KEYS entry {entry!r}. Expected 'name:role:secret', "
                f"comma-separated between entries."
            )
        name, role, secret = (p.strip() for p in parts)
        # MUST stay in step with ROLE_PERMISSIONS in app/infrastructure/security.py,
        # which is the source of truth. It is duplicated rather than imported
        # because security.py imports THIS module, so importing it back here
        # would be a cycle. A role added there but not here is rejected at boot
        # with the message below - loud, and before any request is served.
        if role not in {"admin", "user", "uploader"}:
            raise RuntimeError(
                f"API_KEYS entry {name!r} has role {role!r}; expected 'admin', 'user' "
                f"or 'uploader'."
            )
        if not secret:
            raise RuntimeError(f"API_KEYS entry {name!r} has an empty secret.")
        if len(secret) < API_KEY_MIN_LENGTH:
            raise RuntimeError(
                f"API_KEYS entry {name!r} has a {len(secret)}-character secret; at least "
                f"{API_KEY_MIN_LENGTH} are required, because /api/health publishes a digest "
                f"of it. Generate one with: "
                f"python -c \"import secrets; print(secrets.token_urlsafe(32))\""
            )
        parsed[secret] = (name, role)
    return parsed


# Machine consumers of api.<domain>. Empty by default - browser sessions go
# through /api/auth/login instead, and a key that nobody needs is only risk.
#
# NAME THE KEY FOR ITS FUNCTION, NOT THE PERSON HOLDING IT.
# /api/health is public and reports {name, role, fingerprint} for every
# configured key (describe_api_keys in security.py). The secret is never
# exposed, but the NAME is - so `catalog-drop:uploader:...` is right and
# `priya-laptop:uploader:...` publishes a colleague's name to anyone who
# curls the health endpoint.
API_KEYS = _parse_api_keys(os.getenv("API_KEYS", ""))