backend/app.py - master - AlexGromer/local-stack-test-work

backend/app.py
107 строк5 KB
AlexGromer
Initial commit
24 июн 2026, 12:53
24 июн 2026, 12:53025dfc6
#!/usr/bin/env python3
# =====================================================================
#  backend/app.py — минимальный echo-сервер на стандартной библиотеке
#  Python (ZERO внешних зависимостей -> мгновенный старт, маленький
#  attack surface, не нужен pip install / Dockerfile).
#
#  Назначение: показать, что reverse proxy (nginx) корректно проставляет
#  заголовки X-Forwarded-For / X-Forwarded-Proto / X-Forwarded-Host.
#
#  Эндпоинты:
#    GET  /         -> HTML-таблица со всеми полученными заголовками
#    GET  /headers  -> JSON: {"headers": {...}}
#    GET  /get      -> JSON: метод, путь, заголовки, origin (httpbin-style)
#    POST /*        -> JSON: сколько байт тела получено (проверка лимита 10MB)
#    GET  /healthz  -> 200 "ok"  (для docker healthcheck)
# =====================================================================
import json
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# PORT берём из окружения (compose передаёт PORT=8080). Дефолт совпадает.
PORT = int(os.environ.get("PORT", "8080"))


class Handler(BaseHTTPRequestHandler):
    server_version = "local-backend/1.0"
    protocol_version = "HTTP/1.1"  # ЗАЧЕМ: keep-alive с upstream nginx (HTTP/1.1)

    # ---- helpers ----
    def _write(self, body: bytes, status: int = 200, ctype: str = "text/plain"):
        self.send_response(status)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _json(self, payload: dict, status: int = 200):
        body = json.dumps(payload, indent=2, ensure_ascii=False).encode("utf-8")
        self._write(body, status, "application/json; charset=utf-8")

    def _headers_dict(self) -> dict:
        return {k: v for k, v in self.headers.items()}

    # ---- routing ----
    def do_GET(self):
        if self.path == "/healthz":
            self._write(b"ok\n")
            return

        if self.path == "/headers":
            self._json({"headers": self._headers_dict()})
            return

        if self.path.startswith("/get"):
            self._json({
                "method": self.command,
                "path": self.path,
                "headers": self._headers_dict(),
                "origin": self.client_address[0],  # = IP nginx, не реального клиента
            })
            return

        # default: человекочитаемая HTML-страница
        hdrs = self._headers_dict()
        rows = "".join(
            f"<tr><td><code>{k}</code></td><td><code>{v}</code></td></tr>"
            for k, v in hdrs.items()
        )
        html = (
            "<!doctype html><html><head><meta charset='utf-8'>"
            "<title>Local Backend</title><style>"
            "body{font-family:system-ui,sans-serif;max-width:820px;margin:40px auto;padding:0 16px}"
            "table{border-collapse:collapse;width:100%}td,th{border:1px solid #ddd;padding:6px 10px}"
            "th{text-align:left;background:#f4f4f4}code{font-size:13px}"
            "</style></head><body>"
            "<h1>&#9989; Backend получил запрос</h1>"
            f"<p>TCP peer (кто реально подключился): <code>{self.client_address[0]}</code> — "
            "это IP nginx. Реальный клиент — в <code>X-Forwarded-For</code>.</p>"
            "<h2>Полученные заголовки</h2>"
            f"<table><tr><th>Header</th><th>Value</th></tr>{rows}</table>"
            "</body></html>"
        )
        self._write(html.encode("utf-8"), 200, "text/html; charset=utf-8")

    def do_POST(self):
        length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(length) if length else b""
        self._json({
            "method": "POST",
            "path": self.path,
            "received_bytes": len(body),
            "headers": self._headers_dict(),
        })

    # лог в stdout -> docker logs ls-backend
    def log_message(self, fmt, *args):
        print("[backend] " + (fmt % args), flush=True)


if __name__ == "__main__":
    # ThreadingHTTPServer: параллельная обработка (healthcheck + трафик одновременно)
    srv = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    print(f"[backend] listening on 0.0.0.0:{PORT}", flush=True)
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        srv.shutdown()