VacatureRadar/scripts/configure_public_url.py

#!/usr/bin/env python
from __future__ import annotations

import argparse
import os
import tempfile
from pathlib import Path
from urllib.parse import urlparse


class ConfigurationError(ValueError):
    pass


def normalize_public_url(value: str) -> tuple[str, str]:
    raw = value.strip().rstrip("/")
    parsed = urlparse(raw)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        raise ConfigurationError("Gebruik een volledige http(s)-URL, bijvoorbeeld https://jobs.example.be")
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ConfigurationError("De publieke URL mag geen credentials, query of fragment bevatten.")
    if parsed.path not in {"", "/"}:
        raise ConfigurationError("De applicatie moet op de root van het domein gepubliceerd worden.")
    try:
        port = parsed.port
    except ValueError as exc:
        raise ConfigurationError("De publieke URL bevat een ongeldige poort.") from exc
    host = parsed.hostname.casefold()
    display_host = f"[{host}]" if ":" in host else host
    default_port = 443 if parsed.scheme == "https" else 80
    port_suffix = f":{port}" if port and port != default_port else ""
    return f"{parsed.scheme}://{display_host}{port_suffix}", host


def parse_env(text: str) -> tuple[list[str], dict[str, str]]:
    lines = text.splitlines()
    values: dict[str, str] = {}
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith("#") or "=" not in stripped:
            continue
        key, value = stripped.split("=", 1)
        values[key.strip()] = value.strip()
    return lines, values


def merge_csv(existing: str, additions: list[str]) -> str:
    result: list[str] = []
    seen: set[str] = set()
    for item in [*existing.split(","), *additions]:
        value = item.strip()
        if value and value.casefold() not in seen:
            seen.add(value.casefold())
            result.append(value)
    return ",".join(result)


def render_env(lines: list[str], updates: dict[str, str]) -> str:
    rendered: list[str] = []
    written: set[str] = set()
    for line in lines:
        stripped = line.strip()
        if stripped and not stripped.startswith("#") and "=" in stripped:
            key = stripped.split("=", 1)[0].strip()
            if key in updates:
                rendered.append(f"{key}={updates[key]}")
                written.add(key)
                continue
        rendered.append(line)
    missing = [key for key in updates if key not in written]
    if missing:
        if rendered and rendered[-1].strip():
            rendered.append("")
        rendered.append("# Publieke URL en reverse-proxyhardening")
        rendered.extend(f"{key}={updates[key]}" for key in missing)
    return "\n".join(rendered).rstrip() + "\n"


def build_updates(
    current: dict[str, str],
    public_url: str,
    *,
    hsts_seconds: int,
    cache_url: str | None,
    trusted_proxy_cidrs: list[str],
) -> dict[str, str]:
    origin, host = normalize_public_url(public_url)
    if not origin.startswith("https://"):
        raise ConfigurationError("Een publieke productie-URL moet HTTPS gebruiken.")
    updates = {
        "PUBLIC_BASE_URL": origin,
        "DJANGO_DEBUG": "0",
        "DJANGO_ALLOWED_HOSTS": merge_csv(
            current.get("DJANGO_ALLOWED_HOSTS", ""), [host, "127.0.0.1", "localhost"]
        ),
        "DJANGO_CSRF_TRUSTED_ORIGINS": merge_csv(
            current.get("DJANGO_CSRF_TRUSTED_ORIGINS", ""), [origin]
        ),
        "SESSION_COOKIE_SECURE": "1",
        "CSRF_COOKIE_SECURE": "1",
        "SECURE_SSL_REDIRECT": "1",
        "SECURE_HSTS_SECONDS": str(hsts_seconds),
        "TRUST_PROXY_HEADERS": "1",
        "DEMO_READ_ONLY": "1",
        "SEARCH_ENGINE_INDEXING_ENABLED": "0",
        "HEALTHCHECK_REQUIRE_CACHE": "1",
    }
    if cache_url:
        updates["CACHE_URL"] = cache_url
    if trusted_proxy_cidrs:
        updates["TRUSTED_PROXY_CIDRS"] = merge_csv(
            current.get("TRUSTED_PROXY_CIDRS", ""), trusted_proxy_cidrs
        )
    return updates


def configure_env(
    env_file: Path,
    public_url: str,
    *,
    hsts_seconds: int = 300,
    cache_url: str | None = None,
    trusted_proxy_cidrs: list[str] | None = None,
) -> list[str]:
    if hsts_seconds < 0:
        raise ConfigurationError("HSTS-seconden mogen niet negatief zijn.")
    original = env_file.read_text(encoding="utf-8") if env_file.exists() else ""
    lines, current = parse_env(original)
    updates = build_updates(
        current,
        public_url,
        hsts_seconds=hsts_seconds,
        cache_url=cache_url,
        trusted_proxy_cidrs=trusted_proxy_cidrs or [],
    )
    output = render_env(lines, updates)
    env_file.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary_name = tempfile.mkstemp(prefix=f".{env_file.name}.", dir=env_file.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(output)
        os.chmod(temporary_name, 0o600)
        os.replace(temporary_name, env_file)
    finally:
        if os.path.exists(temporary_name):
            os.unlink(temporary_name)
    return list(updates)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Configureer VacatureRadar veilig voor een publieke reverse-proxy-URL."
    )
    parser.add_argument("public_url", help="Bijvoorbeeld https://vacatureradar.example.be")
    parser.add_argument("--env-file", type=Path, default=Path(".env"))
    parser.add_argument(
        "--hsts-seconds",
        type=int,
        default=300,
        help="Start conservatief; verhoog na succesvolle HTTPS-validatie.",
    )
    parser.add_argument("--cache-url", help="Bij Unraid AIO: redis://127.0.0.1:6379/1")
    parser.add_argument(
        "--trusted-proxy-cidr",
        action="append",
        default=[],
        help="Optioneel vertrouwd proxy-IP/CIDR voor correcte client-IP-rate-limiting.",
    )
    args = parser.parse_args()
    try:
        changed = configure_env(
            args.env_file,
            args.public_url,
            hsts_seconds=args.hsts_seconds,
            cache_url=args.cache_url,
            trusted_proxy_cidrs=args.trusted_proxy_cidr,
        )
    except (OSError, ConfigurationError) as exc:
        parser.error(str(exc))
    print(f"Bijgewerkt: {args.env_file}")
    print("Veilig ingestelde sleutels: " + ", ".join(changed))
    print("Hermaak de container; een gewone restart laadt gewijzigde environment niet opnieuw.")
    return 0


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