NikkeDoy/CORSProxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ | refactor: clean up code, add systemd service, and write comprehensive README

Browse Source
This commit is contained in:
NikkeDoy
2026-05-19 17:44:14 +03:00
parent 0519ba4908
commit 9660cfa41b
3 changed files with 338 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

223
main.py
View File

@@ -1,136 +1,219 @@
#!/usr/bin/env python3
"""
CORS proxy for Unsloth Studio API.
"""CORS proxy for Unsloth Studio API.
A lightweight reverse-proxy that adds Cross-Origin Resource Sharing (CORS)
headers, allowing browser-based frontend apps to call the Unsloth Studio API
running on localhost without being blocked by the Same-Origin Policy.
Usage:
python cors_proxy.py # proxies to http://127.0.0.1:8888 on port 8080
python cors_proxy.py --target 8000 # proxies to http://127.0.0.1:8000
python cors_proxy.py --listen 9090 # listens on port 9090
python cors_proxy.py --target 10.0.0.5:8000 --listen 9000
python main.py # listen :8080, forward to :8888
python main.py --target 8000 # forward to http://127.0.0.1:8000
python main.py --listen 9090 # listen on port 9090
python main.py --target 10.0.0.5 # forward to http://10.0.0.5:8888
python main.py --target 10.0.0.5:8000 --listen 9000
Systemd:
sudo cp services/corsproxy.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable --now corsproxy.service
"""
from __future__ import annotations
import argparse
import json
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.request import Request, urlopen, URLError
from urllib.request import Request, urlopen
from urllib.error import URLError
CORS_HEADERS = {
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
CORS_HEADERS: dict[str, str] = {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "POST, OPTIONS, GET",
"Access-Control-Allow-Headers": "Content-Type, Authorization",
"Access-Control-Max-Age": "86400",
}
# ---------------------------------------------------------------------------
# Utilities
# ---------------------------------------------------------------------------
def parse_target(target: str) -> tuple[str, int]:
"""Split a ``host:port`` string into ``(host, port)``.
If no colon is present the default port ``8888`` is assumed.
"""
if ":" in target:
host, port_str = target.rsplit(":", 1)
else:
host = target
port_str = "8888"
try:
port: int = int(port_str)
except ValueError:
print(f"Error: Invalid target port: {port_str!r}")
sys.exit(1)
return host, port
# ---------------------------------------------------------------------------
# Request handler
# ---------------------------------------------------------------------------
class ProxyHandler(BaseHTTPRequestHandler):
target_host = "127.0.0.1"
target_port = 8888
"""Handle individual proxy requests.
def _target_url(self):
Class attributes can be mutated at runtime (e.g. by ``main``) to change
the upstream target before the server starts listening.
"""
target_host: str = "127.0.0.1"
target_port: int = 8888
# ---- helpers ----------------------------------------------------------
def _target_url(self) -> str:
return f"http://{self.target_host}:{self.target_port}{self.path}"
def _set_cors(self, status=200):
def _set_cors(self, status: int = 200) -> None:
"""Send a 200 OK (or other *status*) with standard CORS headers."""
self.send_response(status)
for k, v in CORS_HEADERS.items():
self.send_header(k, v)
for key, value in CORS_HEADERS.items():
self.send_header(key, value)
def do_OPTIONS(self):
def _json_response(self, status: int, payload: dict) -> None:
"""Write a JSON-encoded response body."""
self._set_cors(status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(payload).encode())
# ---- HTTP methods -----------------------------------------------------
def do_OPTIONS(self) -> None:
"""Respond to CORS preflight requests with a 204 No Content."""
self._set_cors(204)
self.end_headers()
def do_POST(self):
def do_POST(self) -> None:
"""Forward a POST request to the upstream server."""
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length)
target = self._target_url()
req = Request(target, data=body, method="POST")
for h in ("Content-Type", "Authorization"):
val = self.headers.get(h)
if val:
req.add_header(h, val)
# Forward Content-Type and Authorization headers
for header_name in ("Content-Type", "Authorization"):
value = self.headers.get(header_name)
if value:
req.add_header(header_name, value)
try:
resp = urlopen(req, timeout=60)
resp_body = resp.read()
status = resp.status
except URLError as e:
detail = f"Upstream server at {self.target_host}:{self.target_port} is unreachable."
if hasattr(e, "reason"):
detail += f" Reason: {e.reason}"
self._set_cors(502)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({
"error": {"message": detail, "type": "proxy_error"}
}).encode())
except URLError as exc:
detail = (
f"Upstream server at {self.target_host}:{self.target_port} "
"is unreachable."
)
if hasattr(exc, "reason"):
detail += f" Reason: {exc.reason}"
self._json_response(502, {"error": {"message": detail, "type": "proxy_error"}})
return
except Exception as e:
self._set_cors(502)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({
"error": {"message": f"Proxy error: {e}", "type": "proxy_error"}
}).encode())
except Exception as exc:
self._json_response(
502, {"error": {"message": f"Proxy error: {exc}", "type": "proxy_error"}}
)
return
self._set_cors(status)
ctype = resp.headers.get("Content-Type", "application/json")
self.send_header("Content-Type", ctype)
content_type = resp.headers.get("Content-Type", "application/json")
self.send_header("Content-Type", content_type)
self.end_headers()
self.wfile.write(resp_body)
def log_message(self, fmt, *args):
method = self.command
path = self.path
print(f" [{method}] {path} -> {self._target_url()}")
def do_GET(self):
def do_GET(self) -> None:
"""Serve a health-check endpoint; everything else returns 405."""
if self.path == "/health":
self._set_cors(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({
self._json_response(200, {
"status": "ok",
"proxy_to": f"{self.target_host}:{self.target_port}",
}).encode())
})
return
self._set_cors(405)
self.end_headers()
# ---- logging ----------------------------------------------------------
def main():
parser = argparse.ArgumentParser(description="CORS proxy for Unsloth Studio")
parser.add_argument(
"--target", default="127.0.0.1:8888",
help="Unsloth Studio address (default: 127.0.0.1:8888)",
def log_message(self, format: str, *args) -> None:
"""Print a human-readable log line for every request."""
method = self.command
path = self.path
target = self._target_url()
print(f" [{method}] {path} -> {target}")
# Suppress the default server stderr output from BaseHTTPRequestHandler
def log_error(self, format: str, *args) -> None:
pass
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main() -> None:
"""Parse CLI arguments, start the HTTP server, and block."""
parser = argparse.ArgumentParser(
description="CORS proxy for Unsloth Studio",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Examples:\n"
" python main.py # :8080 -> :8888\n"
" python main.py --target 10.0.0.5:8000 --listen 9090\n"
),
)
parser.add_argument(
"--listen", default=8080, type=int,
"--target",
default="127.0.0.1:8888",
help="Upstream Unsloth Studio address (default: 127.0.0.1:8888)",
)
parser.add_argument(
"--listen",
default=8080,
type=int,
help="Port to listen on (default: 8080)",
)
args = parser.parse_args()
host, port_str = args.target, "8888"
if ":" in args.target:
host, port_str = args.target.rsplit(":", 1)
try:
port = int(port_str)
except ValueError:
print(f"Invalid target port: {port_str}")
sys.exit(1)
# Resolve upstream host and port
host, port = parse_target(args.target)
# Configure handler and start the server
ProxyHandler.target_host = host
ProxyHandler.target_port = port
server = HTTPServer(("0.0.0.0", args.listen), ProxyHandler)
print(f" Unsloth Studio CORS Proxy")
print(f" ─────────────────────────")
print(f" Listening on: http://127.0.0.1:{args.listen}")
print(f" Forwarding to: http://{host}:{port}")
print(f" Plugin API URL: http://127.0.0.1:{args.listen}")
print(f" Health check: http://127.0.0.1:{args.listen}/health")
banner = (
" Unsloth Studio CORS Proxy\n"
" ─────────────────────────\n"
f" Listening on: http://127.0.0.1:{args.listen}\n"
f" Forwarding to: http://{host}:{port}\n"
f" Plugin API URL: http://127.0.0.1:{args.listen}\n"
f" Health check: http://127.0.0.1:{args.listen}/health"
)
print(banner)
print()
try: