DataOps-platform/app/config/database_urls.py

"""Fail-closed validation for production PostgreSQL identities and templates."""

from __future__ import annotations

import argparse
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import unquote, urlsplit

_BAD_PERCENT_ESCAPE = re.compile(r"%(?![0-9a-fA-F]{2})")
_PLACEHOLDER_VALUES = {
    "change-me",
    "changeme",
    "database",
    "database_name",
    "dbname",
    "host",
    "hostname",
    "pass",
    "password",
    "replace",
    "user",
    "username",
}


@dataclass(frozen=True)
class DatabaseUrlParts:
    username: str
    password: str
    host: str
    database: str


def is_template_secret(value: str) -> bool:
    """Return true only for empty or explicit template values, after decoding."""
    normalized = unquote(str(value or "")).strip().lower()
    return (
        not normalized
        or normalized in _PLACEHOLDER_VALUES
        or normalized.startswith(("replace-", "your-"))
        or "${" in normalized
        or normalized.startswith("<")
        or normalized.endswith(">")
    )


def validate_postgresql_url(value: str, field_name: str) -> DatabaseUrlParts:
    """Parse a PostgreSQL URL and reject empty/template identity components."""
    raw = str(value or "").strip()
    if not raw or _BAD_PERCENT_ESCAPE.search(raw):
        raise RuntimeError(f"{field_name} is missing or malformed")
    try:
        parsed = urlsplit(raw)
        port = parsed.port
    except (TypeError, ValueError) as exc:
        raise RuntimeError(f"{field_name} is malformed") from exc
    if parsed.scheme.split("+", 1)[0].lower() not in {"postgres", "postgresql"}:
        raise RuntimeError(f"{field_name} must use PostgreSQL")
    values = {
        "username": unquote(parsed.username or ""),
        "password": unquote(parsed.password or ""),
        "host": unquote(parsed.hostname or ""),
        "database": unquote(parsed.path.lstrip("/")),
    }
    if port is not None and not 1 <= port <= 65535:
        raise RuntimeError(f"{field_name} has an invalid port")
    for component, component_value in values.items():
        if is_template_secret(component_value):
            raise RuntimeError(f"{field_name} has an invalid {component}")
    return DatabaseUrlParts(**values)


def _read_env_file(path: str | Path) -> dict[str, str]:
    values: dict[str, str] = {}
    for raw_line in Path(path).read_text(encoding="utf-8-sig").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        name, value = line.split("=", 1)
        values[name.strip()] = value.strip().strip('"').strip("'")
    return values


def validate_database_environment(values: dict[str, str]) -> None:
    runtime_parts = None
    for field_name in (
        "DB_ROLE_INIT_DATABASE_URL",
        "MIGRATION_DATABASE_URL",
        "DATABASE_URL",
    ):
        parts = validate_postgresql_url(values.get(field_name, ""), field_name)
        if field_name == "DATABASE_URL":
            runtime_parts = parts
    runtime_password = values.get("DATAOPS_RUNTIME_PASSWORD", "")
    if is_template_secret(runtime_password):
        raise RuntimeError("DATAOPS_RUNTIME_PASSWORD is missing or a template value")
    if runtime_parts is None or runtime_parts.password != runtime_password:
        raise RuntimeError("DATAOPS_RUNTIME_PASSWORD must match DATABASE_URL")


def load_and_validate_database_env(path: str | Path) -> None:
    validate_database_environment(_read_env_file(path))


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--env-file", required=True)
    args = parser.parse_args(argv)
    try:
        load_and_validate_database_env(args.env_file)
    except (OSError, RuntimeError) as exc:
        print(f"database configuration rejected: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())