feat: extract proxy server from lfs-proxy.py into module

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

src/jianda_proxy/proxy.py

@@ -0,0 +1,96 @@
+import http.client
+import http.server
+import logging
+from urllib.parse import urlparse
+
+logger = logging.getLogger("jianda_proxy")
+
+
+class ProxyHandler(http.server.BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    REMOTE_DOMAIN = ""
+    UPSTREAM_IP = ""
+    UPSTREAM_PORT = 0
+
+    def _proxy_request(self, target_host, target_port, target_path, body=None):
+        body_len = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(body_len) if body_len else None
+
+        conn = http.client.HTTPConnection(target_host, target_port, timeout=120)
+        try:
+            headers = {}
+            skip = {"host", "transfer-encoding", "connection"}
+            for k in self.headers:
+                if k.lower() not in skip:
+                    headers[k] = self.headers[k]
+            headers["Host"] = f"{ProxyHandler.UPSTREAM_IP}:{ProxyHandler.UPSTREAM_PORT}"
+            if body_len:
+                headers["Content-Length"] = str(body_len)
+
+            conn.request(self.command, target_path, body=body, headers=headers)
+            resp = conn.getresponse()
+            resp_body = resp.read()
+
+            self.send_response(resp.status)
+            sent_cl = False
+            for k, v in resp.getheaders():
+                kl = k.lower()
+                if kl in ("transfer-encoding", "connection"):
+                    continue
+                if kl == "content-length":
+                    self.send_header(k, len(resp_body))
+                    sent_cl = True
+                    continue
+                self.send_header(k, v)
+            if not sent_cl:
+                self.send_header("Content-Length", len(resp_body))
+            self.end_headers()
+            self.wfile.write(resp_body)
+        finally:
+            conn.close()
+
+    def _do(self):
+        raw_path = self.path
+        parsed = urlparse(raw_path)
+
+        if (
+            parsed.hostname
+            and parsed.hostname != "127.0.0.1"
+            and parsed.hostname != "localhost"
+        ):
+            if parsed.hostname == ProxyHandler.REMOTE_DOMAIN:
+                target_host = ProxyHandler.UPSTREAM_IP
+            else:
+                target_host = parsed.hostname
+            target_port = parsed.port or 80
+            target_path = parsed.path + ("?" + parsed.query if parsed.query else "")
+            self._proxy_request(target_host, target_port, target_path)
+        else:
+            self._proxy_request(
+                ProxyHandler.UPSTREAM_IP, ProxyHandler.UPSTREAM_PORT, raw_path
+            )
+
+    do_GET = do_POST = do_PUT = do_DELETE = do_HEAD = do_PATCH = _do
+
+    def log_message(self, fmt, *args):
+        logger.info("%s - %s", self.address_string(), fmt % args)
+
+
+def create_server(config):
+    ProxyHandler.REMOTE_DOMAIN = config["remote_domain"]
+    ProxyHandler.UPSTREAM_IP = config["upstream_host"]
+    ProxyHandler.UPSTREAM_PORT = int(config["upstream_port"])
+
+    listen_host = config.get("listen_host", "127.0.0.1")
+    listen_port = int(config["listen_port"])
+
+    server = http.server.ThreadingHTTPServer((listen_host, listen_port), ProxyHandler)
+    logger.info(
+        "Proxy: %s:%s -> %s:%s",
+        listen_host,
+        listen_port,
+        ProxyHandler.UPSTREAM_IP,
+        ProxyHandler.UPSTREAM_PORT,
+    )
+    return server

tests/test_proxy.py

@@ -0,0 +1,50 @@
+import threading
+import http.client
+import time
+import pytest
+from unittest.mock import MagicMock, patch
+from jianda_proxy.proxy import create_server
+
+
+class TestCreateServer:
+    def test_creates_server_with_config(self):
+        config = {
+            "remote_domain": "git.example.com",
+            "upstream_host": "1.2.3.4",
+            "upstream_port": "9000",
+            "listen_host": "127.0.0.1",
+            "listen_port": "19999",
+        }
+        server = create_server(config)
+        assert server is not None
+        server.server_close()
+
+    def test_binds_to_correct_port(self):
+        config = {
+            "remote_domain": "git.example.com",
+            "upstream_host": "1.2.3.4",
+            "upstream_port": "9000",
+            "listen_host": "127.0.0.1",
+            "listen_port": "19998",
+        }
+        server = create_server(config)
+        assert server.server_address[1] == 19998
+        server.server_close()
+
+
+class TestProxyHandler:
+    def test_forward_proxy_mode(self):
+        from jianda_proxy.proxy import ProxyHandler
+
+        config = {
+            "remote_domain": "git.example.com",
+            "upstream_host": "1.2.3.4",
+            "upstream_port": "9000",
+            "listen_host": "127.0.0.1",
+            "listen_port": "19997",
+        }
+        server = create_server(config)
+        # Verify the handler has the correct config set as class attributes
+        assert hasattr(ProxyHandler, "REMOTE_DOMAIN")
+        assert ProxyHandler.REMOTE_DOMAIN == "git.example.com"
+        server.server_close()