#!/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())