prompt_template/web/serve.py

#!/usr/bin/env python3
"""
Minimaler Entwicklungs-Server für die Prompt Templates Webansicht.
Startet auf Port 8081 und dient die statischen Dateien aus.
"""

import http.server
import socketserver
import socket
import os
import json
import logging
from urllib.parse import urlparse, unquote

logging.basicConfig(level=logging.INFO)

DIRECTORY = os.path.dirname(os.path.abspath(__file__))
ROOT_DIR = os.path.abspath(os.path.join(DIRECTORY, ".."))


def find_free_port(start_port=9000):
    """Finde einen freien Port ab start_port

    Für zukünftige Smoke-Test-Integration.
    """
    port = start_port
    while port < 10000:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.bind(("", port))
                return port
        except OSError:
            port += 1
    return None


class Handler(http.server.SimpleHTTPRequestHandler):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, directory=DIRECTORY, **kwargs)

    def _validate_template_path(self, path):
        """Validiert, dass der Pfad auf Schritte Beschränkt ist und innerhalb von ROOT_DIR/templates/ oder DIRECTORY liegt."""
        parsed = urlparse(path).path
        path = parsed

        # /templates.json ist ein Sonderfall - exakter Pfad im web/-Verzeichnis
        if path == "/templates.json":
            return os.path.join(DIRECTORY, "templates.json")

        # Nur /templates/* Pfade erlauben
        if not path.startswith("/templates/"):
            return None

        # Extrahiere den relativen Pfad nach /templates/
        rel_path = path[len("/templates/") :]  # 'system/test.json' oder '../etc/passwd'
        rel_path = unquote(rel_path)

        # Ablehnen bei leeren Pfaden oder absoluten Pfaden
        if not rel_path or os.path.isabs(rel_path):
            return None

        # Ablehnen bei '..' Sequenzen (vor der Normalisierung!)
        if ".." in rel_path.split(os.sep):
            return None

        # Pfad normalisieren
        normalized_rel = os.path.normpath(rel_path)

        # Nach Normalisierung nochmal prüfen
        if normalized_rel.startswith("..") or os.path.isabs(normalized_rel):
            return None

        # Vollständigen Pfad konstruieren
        full_path = os.path.join(ROOT_DIR, "templates", normalized_rel)

        # Explizite Prüfung, dass der Pfad innerhalb von ROOT_DIR/templates/ liegt
        templates_base = os.path.abspath(os.path.join(ROOT_DIR, "templates"))
        full_path_abs = os.path.abspath(full_path)

        if (
            not full_path_abs.startswith(templates_base + os.sep)
            and full_path_abs != templates_base
        ):
            return None

        return full_path_abs

    def do_PUT(self):
        # Nur PUT auf /templates/* Pfade erlauben
        file_path = self._validate_template_path(urlparse(self.path).path)
        if file_path is None:
            self.send_error(403, "Forbidden: Invalid path")
            return

        # Verzeichnis prüfen - muss existieren
        file_dir = os.path.dirname(file_path)
        if not os.path.exists(file_dir) or not os.path.isdir(file_dir):
            self.send_error(404, "Directory not found")
            return

        # Content-Type prüfen
        content_type = self.headers.get("Content-Type", "")
        if "text/plain" not in content_type and not content_type.startswith("text/"):
            self.send_error(400, "Unsupported content type")
            return

        # Inhalt lesen und speichern
        MAX_BODY = 10 * 1024 * 1024
        content_length = int(self.headers.get("Content-Length", 0))
        if content_length > MAX_BODY:
            self.send_error(413, "Request body too large")
            return
        if content_length <= 0:
            self.send_error(400, "No content provided")
            return

        try:
            file_content = self.rfile.read(content_length)

            # Datei schreiben
            with open(file_path, "wb") as f:
                f.write(file_content)

            # Response Content-Type je nach Dateityp setzen
            if file_path.endswith(".json"):
                response_content_type = "application/json"
            else:
                response_content_type = "text/plain"

            self.send_response(200)
            self.send_header("Content-type", response_content_type)
            self.end_headers()
            self.wfile.write("File saved successfully".encode())
        except Exception as e:
            self.send_error(500, f"Failed to save file: {e}")

    def do_GET(self):
        # Für Root-Pfad: index.html servieren
        parsed_path = urlparse(self.path).path
        if parsed_path == "/" or parsed_path == "/index.html":
            self.path = "/index.html"
            return super().do_GET()

        # Anfragen für /templates.json oder /templates/* umleiten
        file_path = self._validate_template_path(urlparse(self.path).path)
        if file_path is not None:
            # Content-Type je nach Dateiendung setzen
            if file_path.endswith(".md"):
                content_type = "text/plain"
            elif file_path.endswith(".json"):
                content_type = "application/json"
            else:
                content_type = "text/plain"

            try:
                with open(file_path, "rb") as f:
                    self.send_response(200)
                    self.send_header("Content-type", content_type)
                    self.end_headers()
                    self.wfile.write(f.read())
                return
            except FileNotFoundError:
                self.send_error(404, "File not found")
                return
            except Exception as e:
                self.send_error(500, f"Error serving file: {e}")
                return

        return super().do_GET()


def main():
    PORT = 8081
    logging.info("Serving on http://localhost:%s", PORT)

    socketserver.TCPServer.allow_reuse_address = True
    with socketserver.TCPServer(("", PORT), Handler) as httpd:
        logging.info("Serving Prompt Templates on http://localhost:%s", PORT)
        logging.info("Press Ctrl+C to stop")
        logging.info("Directory: %s", DIRECTORY)
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            logging.info("\nServer stopped")


if __name__ == "__main__":
    main()