gendesign/tradein-mvp/backend/app/core/fdw.py

"""postgres_fdw USER MAPPING management for gendesign_remote server.

Password lives in env GENDESIGN_FDW_PASSWORD — never in SQL migration or git.
This helper:
  - validates password format (alphanumeric/hex, length >= 32) to prevent SQL
    injection via embedded quotes (the postgres CREATE USER MAPPING syntax does
    not accept bind parameters — password must be inlined into the DDL string),
  - applies idempotent CREATE or ALTER mapping on every backend startup so
    password rotation through .env.runtime is picked up after restart.
"""

from __future__ import annotations

import logging
import re

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings

logger = logging.getLogger(__name__)

# Strict whitelist — only hex/alphanumeric and a few separators. Refuses ANY
# quote, backslash, unicode whitespace, control char. Min length 32 forces
# rotation away from weak defaults.
_PASSWORD_RE = re.compile(r"^[A-Za-z0-9_\-]{32,256}$")


def ensure_fdw_user_mapping(db: Session) -> None:
    """Idempotent CREATE OR ALTER USER MAPPING for tradein → gendesign_remote.

    Skips silently if password not configured (dev / first deploy / forgotten env).
    Raises ValueError if password fails format whitelist — fail-fast on misconfig.
    """
    password = settings.gendesign_fdw_password
    if not password or not str(password).strip():
        logger.warning(
            "GENDESIGN_FDW_PASSWORD not set — skipping FDW user mapping "
            "(gendesign_cad_buildings queries will fail; cadastral lookups will "
            "fall back to Nominatim)"
        )
        return

    if not _PASSWORD_RE.fullmatch(password):
        # Do NOT echo the password into logs / exception even partially.
        raise ValueError(
            "GENDESIGN_FDW_PASSWORD failed format whitelist "
            "(allowed: 32-256 chars, [A-Za-z0-9_-]). Refusing to apply USER MAPPING "
            "— password rotation must use the same format."
        )

    # NB: even after whitelist validation we still cannot use SQLAlchemy bind
    # parameters here — postgres CREATE/ALTER USER MAPPING is DDL and treats
    # OPTIONS values as literal text. Whitelist guarantees the value is
    # injection-safe; we still wrap in single quotes for the DDL syntax.
    exists = db.execute(
        text(
            "SELECT 1 FROM pg_user_mappings "
            "WHERE srvname = 'gendesign_remote' "
            "AND (usename = current_user OR usename IS NULL)"
        )
    ).first()

    if exists is None:
        db.execute(
            text(
                f"CREATE USER MAPPING FOR CURRENT_USER SERVER gendesign_remote "
                f"OPTIONS (user 'tradein_fdw_reader', password '{password}')"
            )
        )
        logger.info("created FDW user mapping for gendesign_remote")
    else:
        db.execute(
            text(
                f"ALTER USER MAPPING FOR CURRENT_USER SERVER gendesign_remote "
                f"OPTIONS (SET password '{password}')"
            )
        )
        logger.info("refreshed FDW user mapping password for gendesign_remote")

    try:
        db.commit()
    except Exception:
        db.rollback()
        raise