XFESystem

"""Controlled host HTTP peer for independent QEMU native-network acceptance.

The guest must obtain these bytes through its NIC/TCP adapter. This server never
reads guest files or fetches another URL on behalf of the guest.
"""

import argparse
import html
import json
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--Port", type=int, default=18080)
    parser.add_argument("--Nonce", required=True)
    parser.add_argument("--Log", type=Path, required=True)
    options = parser.parse_args()
    if not 0 <= options.Port <= 65535 or not options.Nonce or len(options.Nonce) > 128:
        parser.error("Port or nonce is outside the fixture limit")
    options.Log.parent.mkdir(parents=True, exist_ok=True)
    lock = threading.Lock()
    escaped_nonce = html.escape(options.Nonce, quote=True)
    document = (
        "<!doctype html><html><head><title>XFE Native HTTP</title></head><body>"
        "<h1>XFESystem native TCP browser</h1>"
        f"<p id='Nonce'>Server nonce: {escaped_nonce}</p>"
        "<p id='Result'>Script pending</p><p>中文 HTTP 正文</p>"
        "<a href='/Chunked'>Chunked response</a>"
        "<script>document.getElementById('Result').textContent = 'Native network JS: ' + (6 * 7);</script>"
        "</body></html>"
    ).encode("utf-8")

    class Handler(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def log_message(self, *_):
            pass

        def do_GET(self):
            path = urlsplit(self.path).path
            with lock:
                with options.Log.open("a", encoding="utf-8") as log:
                    log.write(json.dumps({
                        "Utc": datetime.now(timezone.utc).isoformat(),
                        "Path": self.path,
                        "Peer": self.client_address[0],
                        "Host": self.headers.get("Host"),
                        "UserAgent": self.headers.get("User-Agent"),
                        "Nonce": options.Nonce,
                    }, ensure_ascii=False) + "\n")
            self.close_connection = True
            try:
                if path == "/Redirect":
                    self.send_response(302)
                    self.send_header("Location", "/Index")
                    self.send_header("Content-Length", "0")
                    self.send_header("Connection", "close")
                    self.end_headers()
                    return
                if path == "/Drop":
                    return
                if path == "/Oversized":
                    self.send_response(200)
                    self.send_header("Content-Length", "999999999")
                    self.send_header("Connection", "close")
                    self.end_headers()
                    return
                if path == "/InvalidFraming":
                    self.send_response(200)
                    self.send_header("Content-Length", "0")
                    self.send_header("Transfer-Encoding", "chunked")
                    self.send_header("Connection", "close")
                    self.end_headers()
                    self.wfile.write(b"0\r\n\r\n")
                    return
                body = document
                if path == "/Csp":
                    body = (
                        "<title>CSP native HTTP</title><p id='Value'>CSP original</p>"
                        "<script>document.getElementById('Value').textContent='CSP should block this';</script>"
                    ).encode("utf-8")
                if path not in ("/", "/Index", "/Chunked", "/Truncated", "/Csp", "/Slow"):
                    self.send_error(404)
                    return
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Connection", "close")
                if path == "/Csp":
                    self.send_header("Content-Security-Policy", "script-src 'none'")
                if path == "/Chunked":
                    self.send_header("Transfer-Encoding", "chunked")
                    self.end_headers()
                    offset = 0
                    for size in (1, 3, 7, 11, 31, 61, len(body)):
                        chunk = body[offset:offset + size]
                        if not chunk:
                            break
                        self.wfile.write(f"{len(chunk):x};Fixture=True\r\n".encode("ascii") + chunk + b"\r\n")
                        self.wfile.flush()
                        offset += len(chunk)
                    self.wfile.write(b"0\r\nX-Fixture: Complete\r\n\r\n")
                else:
                    self.send_header("Content-Length", str(len(body) + (5 if path == "/Truncated" else 0)))
                    self.end_headers()
                    if path == "/Slow":
                        self.wfile.flush()
                        time.sleep(30)
                    self.wfile.write(body)
                self.wfile.flush()
            except (BrokenPipeError, ConnectionResetError):
                pass

    server = ThreadingHTTPServer(("127.0.0.1", options.Port), Handler)
    server.daemon_threads = True
    print(json.dumps({"HostOnlyFixture": True, "Port": server.server_port,
                      "GuestAddress": f"http://10.0.2.2:{server.server_port}/Index",
                      "Nonce": options.Nonce, "Log": str(options.Log.resolve())}), flush=True)
    try:
        server.serve_forever(poll_interval=0.1)
    finally:
        server.server_close()


if __name__ == "__main__":
    main()