Revision of Claude proxy

depau revised this gist 1 month ago. Go to revision

2 files changed, 135 insertions

0-README.md(file created)

		@@ -0,0 +1,7 @@
1	+	```bash
2	+	python3 proxy.py
3	+	```
4	+
5	+	```
6	+	ANTHROPIC_BASE_URL=http://127.0.0.1:8787 claude
7	+	```

proxy.py(file created)

		@@ -0,0 +1,128 @@
1	+	#!/usr/bin/env python3
2	+	"""Logging HTTP origin proxy for Claude Code. LOCAL test aid only.
3	+
4	+	Claude is pointed at this proxy via ANTHROPIC_BASE_URL; every request is
5	+	logged and forwarded to the real Anthropic backend untouched. Optionally the
6	+	remote-settings endpoint is answered with a local mock payload (--mock) —
7	+	note a custom base URL makes real remote settings ineligible, so the mock is
8	+	mainly for manual curl testing, not the live settings fetch.
9	+
10	+	Usage:
11	+	python3 proxy.py # listen 127.0.0.1:8787
12	+	python3 proxy.py --mock --port 8788 # answer settings GET with mock
13	+	python3 proxy.py --upstream http://localhost:9000
14	+
15	+	Then: ANTHROPIC_BASE_URL=http://127.0.0.1:8787 claude -p ...
16	+	"""
17	+	import argparse, hashlib, json, re, time
18	+	from urllib.parse import urlparse
19	+	import http.client
20	+	from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
21	+
22	+	LOG = lambda a: print(f"[{time.strftime('%H:%M:%S')}]", a, flush=True)
23	+
24	+	MOCK_SETTINGS = {
25	+	"$schema": "https://json.schemastore.org/claude-code-settings.json",
26	+	"companyAnnouncements": [
27	+	"PROXY-INJECTED announcement - this came from your local passthrough proxy, not Anthropic."
28	+	],
29	+	"permissions": {
30	+	"allow": ["Bash(echo )", "Read(./*)"],
31	+	"deny": [],
32	+	"ask": [],
33	+	"disableBypassPermissionsMode": "off"
34	+	}
35	+	}
36	+
37	+	HOP_HEADERS = {"connect", "keep-alive", "proxy-authenticate", "proxy-authorization",
38	+	"te", "trailers", "transfer-encoding", "upgrade", "host", "proxy-connection"}
39	+
40	+
41	+	def route(server, method: str, path: str, headers, body: bytes, wfile):
42	+	upstream = server.upstream
43	+	LOG(f"{method} {path} Host={headers.get('Host', '?')} -> {upstream}")
44	+
45	+	# Mock the settings endpoint.
46	+	if server.mock and method == "GET" and re.search(r"/api/claude_code/settings\b", path):
47	+	payload = server.mock_body
48	+	wfile.write(b"HTTP/1.1 200 OK\r\n")
49	+	wfile.write(b"Content-Type: application/json\r\n")
50	+	wfile.write(f'ETag: "{hashlib.sha256(payload).hexdigest()}"\r\n'.encode())
51	+	wfile.write(f"Content-Length: {len(payload)}\r\n".encode())
52	+	wfile.write(b"\r\n")
53	+	wfile.write(payload)
54	+	wfile.flush()
55	+	LOG(f" -> MOCK 200 (settings) len={len(payload)}")
56	+	return
57	+
58	+	# Forward to upstream.
59	+	u = urlparse(upstream)
60	+	conn = (http.client.HTTPSConnection(u.netloc, timeout=30) if u.scheme == "https"
61	+	else http.client.HTTPConnection(u.netloc, timeout=30))
62	+	hs = {k: v for k, v in headers.items() if k.lower() not in HOP_HEADERS}
63	+	try:
64	+	conn.request(method, path, body=body or None, headers=hs)
65	+	r = conn.getresponse()
66	+	rbody = r.read()
67	+	LOG(f" -> {r.status} {r.reason} len={len(rbody)}")
68	+	wfile.write(f"HTTP/1.1 {r.status} {r.reason}\r\n".encode())
69	+	for k, v in r.getheaders():
70	+	if k.lower() not in HOP_HEADERS:
71	+	wfile.write(f"{k}: {v}\r\n".encode())
72	+	# Delimit the body ourselves so the client never hangs on keep-alive.
73	+	wfile.write(f"Content-Length: {len(rbody)}\r\n".encode())
74	+	wfile.write(b"Connection: close\r\n\r\n")
75	+	wfile.write(rbody)
76	+	wfile.flush()
77	+	except Exception as e:
78	+	LOG(f" !! forward error: {e}")
79	+	wfile.write(b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\n\r\n")
80	+	wfile.flush()
81	+	finally:
82	+	conn.close()
83	+
84	+
85	+	class Handler(BaseHTTPRequestHandler):
86	+	protocol_version = "HTTP/1.1"
87	+
88	+	def log_message(self, fmt, *args):
89	+	pass
90	+
91	+	def do_GET(self):
92	+	route(self.server, "GET", self.path, self.headers, None, self.wfile)
93	+
94	+	def do_HEAD(self):
95	+	route(self.server, "HEAD", self.path, self.headers, None, self.wfile)
96	+
97	+	def do_POST(self):
98	+	route(self.server, "POST", self.path, self.headers, self._read_body(), self.wfile)
99	+
100	+	def _read_body(self):
101	+	cl = int(self.headers.get("Content-Length") or 0)
102	+	return self.rfile.read(cl) if cl else b""
103	+
104	+
105	+	def main():
106	+	ap = argparse.ArgumentParser()
107	+	ap.add_argument("--host", default="127.0.0.1")
108	+	ap.add_argument("--port", type=int, default=8787)
109	+	ap.add_argument("--upstream", default="https://api.anthropic.com")
110	+	ap.add_argument("--mock", action="store_true",
111	+	help="answer GET /api/claude_code/settings with a mock payload")
112	+	args = ap.parse_args()
113	+
114	+	srv = ThreadingHTTPServer((args.host, args.port), Handler)
115	+	srv.upstream = args.upstream
116	+	srv.mock = args.mock
117	+	srv.mock_body = json.dumps(MOCK_SETTINGS).encode()
118	+
119	+	LOG(f"listening on http://{args.host}:{args.port} upstream={args.upstream} mock={args.mock}")
120	+	try:
121	+	srv.serve_forever()
122	+	except KeyboardInterrupt:
123	+	print("\nbye")
124	+	srv.server_close()
125	+
126	+
127	+	if __name__ == "__main__":
128	+	main()

Newer Older