debug_webserver.py

"""Simple HTTP listener to be used with debugging various EDMC sends."""

from __future__ import annotations

from collections.abc import Callable
import gzip
from http import server
import json
import pathlib
import threading
from typing import Any, Literal
from urllib.parse import parse_qs
import zlib
from config import config
from EDMCLogging import get_main_logger

logger = get_main_logger()

output_lock = threading.Lock()
output_data_path = pathlib.Path(config.app_dir_path / "logs" / "http_debug")  # type: ignore
SAFE_TRANSLATE = str.maketrans(dict.fromkeys("!@#$%^&*()./\\\r\n[]-+='\";:?<>,~`", "_"))


class LoggingHandler(server.BaseHTTPRequestHandler):
    """HTTP Handler implementation that logs to EDMCs logger and writes data to files on disk."""

    def log_message(self, format: str, *args: Any) -> None:
        """Override default handler logger with EDMC logger."""
        logger.info(format % args)

    def do_POST(self) -> None:  # noqa: N802 # I cant change it
        """Handle POST."""
        logger.info(f"Received a POST for {self.path!r}!")

        content_length = int(self.headers.get("Content-Length", 0))
        data_raw: bytes = self.rfile.read(content_length)

        encoding = self.headers.get("Content-Encoding")
        data = self.get_printable(data_raw, encoding)
        to_save = data

        if len(self.path) > 1 and self.path.startswith("/"):
            target_path = self.path[1:]
        elif self.path == "/":
            target_path = "WEB_ROOT"
        else:
            target_path = self.path

        response_handler = DEFAULT_RESPONSES.get(target_path)
        response = (response_handler(to_save) if callable(response_handler) else response_handler)

        self.send_response_only(200, "OK")
        if response is not None:
            self.send_header("Content-Length", str(len(response)))
        self.end_headers()  # This is needed because send_response_only DOESN'T ACTUALLY SEND THE RESPONSE </rant>

        if response is not None:
            self.wfile.write(response.encode())
            self.wfile.flush()

        if target_path == "edsm":
            # attempt to extract data from urlencoded stream
            try:
                edsm_data = extract_edsm_data(data)
                to_save = f"{to_save}\n{json.dumps(edsm_data)}"
            except Exception:
                pass

        target_file = output_data_path / f"{safe_file_name(target_path)}.log"
        if target_file.parent != output_data_path:
            logger.warning(f"REFUSING TO WRITE FILE THAT ISN'T IN THE RIGHT PLACE! {target_file=}")
            logger.warning(f"DATA FOLLOWS\n{data}")
            return

        with output_lock, target_file.open("a", encoding="utf-8") as file:
            file.write(f"{to_save}\n\n")

    @staticmethod
    def get_printable(data: bytes, compression: Literal["deflate", "gzip"] | str | None = None) -> str:
        """Convert an incoming compressed or raw data stream into a string.

        :param data: The data to convert
        :param compression: The compression to remove, defaults to None
        :raises ValueError: If compression is unknown
        :return: printable strings
        """
        match compression:
            case None:
                ret = data
            case "deflate":
                ret = zlib.decompress(data)
            case "gzip":
                ret = gzip.decompress(data)
            case _:
                raise ValueError(f"Unknown encoding for data {compression!r}")

        return ret.decode("utf-8", errors="replace")


def safe_file_name(name: str) -> str:
    """Escape special characters out of a file name safely.

    This is a nicety. Don't rely on it to be ultra secure.
    """
    return name.translate(SAFE_TRANSLATE)


def generate_inara_response(raw_data: str) -> str:
    """Generate nonstatic data for inara plugin."""
    try:
        data = json.loads(raw_data)
    except json.JSONDecodeError:
        return "UNKNOWN REQUEST"

    events = data.get("events") or []
    out = {
        "header": {"eventStatus": 200},
        "events": [
            {
                "eventName": e["eventName"],
                "eventStatus": 200,
                "eventStatusText": "DEBUG STUFF",
            }
            for e in events
            if "eventName" in e
        ],
    }
    return json.dumps(out)


def extract_edsm_data(data: str) -> dict[str, Any]:
    """Extract relevant data from edsm data."""
    res = parse_qs(data)
    return {name: values[0] for name, values in res.items() if values}


def generate_edsm_response(raw_data: str) -> str:
    """Generate nonstatic data for edsm plugin."""
    try:
        data = extract_edsm_data(raw_data)
        events = json.loads(data["message"])
    except (json.JSONDecodeError, KeyError, Exception):
        logger.exception("Failed to parse EDSM debug request envelope payload")
        return "UNKNOWN REQUEST"

    out = {
        "msgnum": 100,  # OK Status Code
        "msg": "debug stuff",
        "events": [
            (
                {
                    "event": e["get"]("event", "unknown"),
                    "msgnum": 100,
                    "msg": "debug stuff",
                }
                if isinstance(e, dict)
                else {"event": "unknown", "msgnum": 100, "msg": "debug stuff"}
            )
            for e in events
        ],
    }
    return json.dumps(out)


DEFAULT_RESPONSES: dict[str, Callable[[str], str]] = {
    "inara": generate_inara_response,
    "edsm": generate_edsm_response,
}


def run_listener(host: str = "127.0.0.1", port: int = 9090) -> None:
    """Run a listener thread."""
    output_data_path.mkdir(exist_ok=True)
    logger.info(f"Starting HTTP listener on {host=} {port=}!")
    listener = server.HTTPServer((host, port), LoggingHandler)
    logger.info(listener)
    threading.Thread(target=listener.serve_forever, daemon=True).start()


if __name__ == "__main__":
    output_data_path.mkdir(exist_ok=True)
    server.HTTPServer(("127.0.0.1", 9090), LoggingHandler).serve_forever()