109 lines
4.6 KiB
Python
109 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate CALLBACK_PROVIDER and render its safe Nginx proxy locations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ipaddress
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
class ProviderValidationError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CallbackProvider:
|
|
origin: str
|
|
path_template: str
|
|
|
|
|
|
def validate_callback_provider(value: str | None) -> CallbackProvider | None:
|
|
if value is None or value == "":
|
|
return None
|
|
if value != value.strip() or len(value) > 2048 or any(character.isspace() for character in value):
|
|
raise ProviderValidationError("CALLBACK_PROVIDER não pode conter espaços e deve ter até 2048 caracteres")
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password:
|
|
raise ProviderValidationError("CALLBACK_PROVIDER deve ser uma URL HTTPS pública, sem usuário ou senha")
|
|
if parsed.query or parsed.fragment or not parsed.path.startswith("/"):
|
|
raise ProviderValidationError("CALLBACK_PROVIDER não aceita query, fragmento e exige um caminho absoluto")
|
|
try:
|
|
port = parsed.port
|
|
except ValueError as error:
|
|
raise ProviderValidationError("CALLBACK_PROVIDER possui porta inválida") from error
|
|
host = parsed.hostname
|
|
if not host or not re.fullmatch(r"[A-Za-z0-9.-]+", host) or ".." in host or host.startswith(".") or host.endswith("."):
|
|
raise ProviderValidationError("CALLBACK_PROVIDER possui host inválido")
|
|
try:
|
|
address = ipaddress.ip_address(host)
|
|
except ValueError:
|
|
address = None
|
|
if address is not None and not address.is_global:
|
|
raise ProviderValidationError("CALLBACK_PROVIDER não pode apontar para endereço IP privado ou reservado")
|
|
if port is not None and not 1 <= port <= 65535:
|
|
raise ProviderValidationError("CALLBACK_PROVIDER possui porta inválida")
|
|
if not re.fullmatch(r"/[A-Za-z0-9._~/%{}-]+", parsed.path) or "/../" in f"{parsed.path}/" or any(token in parsed.path for token in ("$", ";", "\\")):
|
|
raise ProviderValidationError("CALLBACK_PROVIDER possui caminho não permitido")
|
|
for token in ("{x}", "{y}", "{z}"):
|
|
if parsed.path.count(token) != 1:
|
|
raise ProviderValidationError("CALLBACK_PROVIDER deve conter exatamente uma ocorrência de {x}, {y} e {z}")
|
|
remaining_placeholders = parsed.path
|
|
for token in ("{x}", "{y}", "{z}"):
|
|
remaining_placeholders = remaining_placeholders.replace(token, "")
|
|
if "{" in remaining_placeholders or "}" in remaining_placeholders:
|
|
raise ProviderValidationError("CALLBACK_PROVIDER contém placeholder desconhecido")
|
|
return CallbackProvider(origin=f"https://{parsed.netloc}", path_template=parsed.path)
|
|
|
|
|
|
def render_location(name: str, provider: CallbackProvider | None, x: str, y: str, z: str) -> str:
|
|
if provider is None:
|
|
return f"""location @{name} {{
|
|
try_files /tiles/empty.png =500;
|
|
}}
|
|
"""
|
|
rendered_path = provider.path_template.replace("{x}", f"${x}").replace("{y}", f"${y}").replace("{z}", f"${z}")
|
|
return f"""location @{name} {{
|
|
rewrite ^ {rendered_path} break;
|
|
proxy_pass {provider.origin};
|
|
proxy_set_header Host $proxy_host;
|
|
proxy_ssl_server_name on;
|
|
proxy_cache callback_provider;
|
|
proxy_cache_valid 200 7d;
|
|
proxy_cache_use_stale error timeout http_500 http_502 http_503 http_504;
|
|
proxy_hide_header Cache-Control;
|
|
proxy_hide_header Access-Control-Allow-Origin;
|
|
proxy_hide_header Access-Control-Allow-Methods;
|
|
proxy_hide_header Access-Control-Allow-Headers;
|
|
}}
|
|
"""
|
|
|
|
|
|
def render_nginx(provider: CallbackProvider | None) -> str:
|
|
return (
|
|
"# Generated at container startup. Do not edit.\n"
|
|
+ render_location("callback_provider_tile", provider, "tile_x", "tile_y", "tile_z")
|
|
+ "\n"
|
|
+ render_location("callback_provider_base", provider, "base_x", "base_y", "base_z")
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--nginx-output", type=Path)
|
|
arguments = parser.parse_args()
|
|
provider = validate_callback_provider(__import__("os").environ.get("CALLBACK_PROVIDER"))
|
|
if arguments.nginx_output:
|
|
arguments.nginx_output.parent.mkdir(parents=True, exist_ok=True)
|
|
arguments.nginx_output.write_text(render_nginx(provider), encoding="utf-8")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except ProviderValidationError as error:
|
|
raise SystemExit(f"CALLBACK_PROVIDER inválido: {error}")
|