428th-website-2026/api/contact_server.py

#!/usr/bin/env python3
"""Minimal /api/contact endpoint — forwards form submissions to Telegram."""

from __future__ import annotations

import json
import os
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import defaultdict, deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

# ---------- config ----------

ROOT = Path(__file__).resolve().parent
MAX_NAME = 200
MAX_CONTACT = 200
MAX_MESSAGE = 4000
MAX_BODY = 16_384
RATE_LIMIT = 5  # requests
RATE_WINDOW = 60  # seconds


def load_dotenv(path: Path) -> None:
    if not path.is_file():
        return
    for raw in path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        key = key.strip()
        value = value.strip().strip("'").strip('"')
        if key and key not in os.environ:
            os.environ[key] = value


load_dotenv(ROOT / ".env")

BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip()
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "").strip()
HOST = os.environ.get("HOST", "127.0.0.1").strip() or "127.0.0.1"
PORT = int(os.environ.get("PORT", "8787"))
_backup_raw = os.environ.get("MESSAGES_FILE", "messages.jsonl").strip() or "messages.jsonl"
MESSAGES_FILE = Path(_backup_raw)
if not MESSAGES_FILE.is_absolute():
    MESSAGES_FILE = ROOT / MESSAGES_FILE

_rate_lock = threading.Lock()
_file_lock = threading.Lock()
_rate_hits: dict[str, deque[float]] = defaultdict(deque)


def client_ip(handler: BaseHTTPRequestHandler) -> str:
    forwarded = handler.headers.get("X-Real-IP") or handler.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return handler.client_address[0]


def rate_limited(ip: str) -> bool:
    now = time.monotonic()
    with _rate_lock:
        q = _rate_hits[ip]
        while q and now - q[0] > RATE_WINDOW:
            q.popleft()
        if len(q) >= RATE_LIMIT:
            return True
        q.append(now)
        return False


def send_telegram(text: str) -> None:
    if not BOT_TOKEN or not CHAT_ID:
        raise RuntimeError("Telegram is not configured")
    url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
    payload = urllib.parse.urlencode(
        {
            "chat_id": CHAT_ID,
            "text": text,
            "disable_web_page_preview": "1",
        }
    ).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=payload,
        method="POST",
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    with urllib.request.urlopen(req, timeout=15) as resp:
        body = json.loads(resp.read().decode("utf-8"))
    if not body.get("ok"):
        raise RuntimeError("Telegram API rejected the message")


def append_message(record: dict) -> None:
    """Append one JSON line to the backup file (best-effort durable log)."""
    line = json.dumps(record, ensure_ascii=False) + "\n"
    with _file_lock:
        MESSAGES_FILE.parent.mkdir(parents=True, exist_ok=True)
        with MESSAGES_FILE.open("a", encoding="utf-8") as f:
            f.write(line)
            f.flush()
            os.fsync(f.fileno())


def validate_fields(data: dict) -> tuple[str, str, str] | str:
    """Return (name, contact, message) or an error code string."""
    if not isinstance(data, dict):
        return "invalid"

    # Honeypot — treat as success upstream; caller checks separately.
    # Accept legacy "website" (autofill magnets) and current "hp_company".
    for pot in ("hp_company", "website"):
        raw_pot = data.get(pot, "")
        if isinstance(raw_pot, str) and raw_pot.strip():
            return "honeypot"

    def field(key: str, max_len: int) -> str | None:
        raw = data.get(key, "")
        if not isinstance(raw, str):
            return None
        value = raw.strip()
        if not value or len(value) > max_len:
            return None
        return value

    name = field("name", MAX_NAME)
    contact = field("contact", MAX_CONTACT)
    message = field("message", MAX_MESSAGE)
    if name is None or contact is None or message is None:
        return "invalid"
    return name, contact, message


class ContactHandler(BaseHTTPRequestHandler):
    server_version = "428thContact/1.0"

    def log_message(self, fmt: str, *args) -> None:
        # Avoid logging bodies; keep a short access line.
        sys_stderr = __import__("sys").stderr
        print(
            f"{client_ip(self)} - [{self.log_date_time_string()}] {fmt % args}",
            file=sys_stderr,
        )

    def _json(self, status: int, payload: dict) -> None:
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self) -> None:
        # Same-origin from the site; allow preflight just in case.
        if self.path.rstrip("/") != "/api/contact":
            self.send_error(404)
            return
        self.send_response(204)
        self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Content-Length", "0")
        self.end_headers()

    def do_POST(self) -> None:
        if self.path.rstrip("/") != "/api/contact":
            self._json(404, {"ok": False, "error": "not_found"})
            return

        ip = client_ip(self)
        if rate_limited(ip):
            self._json(429, {"ok": False, "error": "rate_limited"})
            return

        length_raw = self.headers.get("Content-Length", "0")
        try:
            length = int(length_raw)
        except ValueError:
            self._json(400, {"ok": False, "error": "invalid"})
            return
        if length < 0 or length > MAX_BODY:
            self._json(400, {"ok": False, "error": "invalid"})
            return

        raw = self.rfile.read(length)
        try:
            data = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            self._json(400, {"ok": False, "error": "invalid"})
            return

        result = validate_fields(data)
        if result == "honeypot":
            self._json(200, {"ok": True})
            return
        if isinstance(result, str):
            self._json(400, {"ok": False, "error": "invalid"})
            return

        name, contact, message = result
        text = (
            "New contact form\n"
            f"Name: {name}\n"
            f"Contact: {contact}\n"
            f"Message:\n{message}"
        )

        telegram_ok = False
        telegram_error = None
        try:
            send_telegram(text)
            telegram_ok = True
        except (urllib.error.URLError, TimeoutError, RuntimeError, OSError) as err:
            telegram_error = type(err).__name__

        try:
            append_message(
                {
                    "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                    "name": name,
                    "contact": contact,
                    "message": message,
                    "ip": ip,
                    "telegram_ok": telegram_ok,
                    "telegram_error": telegram_error,
                }
            )
        except OSError:
            # Disk failure after Telegram: still report based on Telegram.
            if not telegram_ok:
                self._json(502, {"ok": False, "error": "upstream"})
                return
            # Telegram delivered; backup failed — accept but log.
            print(f"WARNING: failed to append backup for {ip}", flush=True)

        if not telegram_ok:
            self._json(502, {"ok": False, "error": "upstream"})
            return

        self._json(200, {"ok": True})

    def do_GET(self) -> None:
        if self.path.rstrip("/") == "/api/health":
            configured = bool(BOT_TOKEN and CHAT_ID)
            self._json(200, {"ok": True, "configured": configured})
            return
        self._json(404, {"ok": False, "error": "not_found"})


def main() -> None:
    if not BOT_TOKEN or not CHAT_ID:
        print(
            "WARNING: TELEGRAM_BOT_TOKEN / TELEGRAM_CHAT_ID not set — "
            "POSTs will fail with 502 until configured.",
            flush=True,
        )
    print(f"message backup file: {MESSAGES_FILE}", flush=True)
    httpd = ThreadingHTTPServer((HOST, PORT), ContactHandler)
    print(f"contact API listening on http://{HOST}:{PORT}/api/contact", flush=True)
    httpd.serve_forever()


if __name__ == "__main__":
    main()