tegwick
b428ffecb1
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
71 lines
2.8 KiB
Python
71 lines
2.8 KiB
Python
"""Isolated fixed-origin HTTP transport. No credentials in argv or diagnostics.
|
|
|
|
Invoked by the SDK through a bounded subprocess, not as an application command.
|
|
Isolation allows the parent to terminate DNS, reads and exceptional close without
|
|
signals in an application's main thread or mandatory multiprocessing main guards.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from urllib.error import HTTPError
|
|
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
|
|
|
|
MAX_BODY = 256 * 1024
|
|
|
|
|
|
class NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
raise HTTPError(req.full_url, code, "redirect refused", headers, fp)
|
|
|
|
|
|
def fetch(data):
|
|
request = Request(data["url"], data=data["body"].encode("utf-8") if data["body"] is not None else None,
|
|
method=data["method"], headers={"Authorization": "Bearer " + data["token"],
|
|
"Content-Type": "application/json", "Accept": "application/json"})
|
|
opener = build_opener(ProxyHandler({}), NoRedirect())
|
|
response = None
|
|
try:
|
|
response = opener.open(request, timeout=data["timeout"])
|
|
if response.geturl() != data["url"] or response.getcode() != 200:
|
|
return {"error": "status", "status": response.getcode()}
|
|
types = response.headers.get_all("Content-Type", [])
|
|
if len(types) != 1 or response.headers.get_content_type() != "application/json":
|
|
return {"error": "content"}
|
|
charset = response.headers.get_content_charset()
|
|
if charset not in (None, "utf-8") or response.headers.get("Content-Encoding", "identity").lower() != "identity":
|
|
return {"error": "content"}
|
|
lengths = response.headers.get_all("Content-Length", [])
|
|
if len(lengths) > 1 or (lengths and (not lengths[0].isdigit() or int(lengths[0]) > MAX_BODY)):
|
|
return {"error": "content"}
|
|
raw = response.read(MAX_BODY + 1)
|
|
if len(raw) > MAX_BODY or (lengths and len(raw) != int(lengths[0])):
|
|
return {"error": "content"}
|
|
return {"body": raw.decode("utf-8")}
|
|
except HTTPError as exc:
|
|
# Never read a refused/redirect/error body; even close is parent-bounded.
|
|
try:
|
|
exc.close()
|
|
except Exception:
|
|
pass
|
|
return {"error": "status", "status": int(exc.code)}
|
|
except Exception:
|
|
return {"error": "unavailable"}
|
|
finally:
|
|
if response is not None:
|
|
try:
|
|
response.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def main():
|
|
try:
|
|
data = json.loads(sys.stdin.buffer.read(MAX_BODY + 1))
|
|
result = fetch(data)
|
|
except Exception:
|
|
result = {"error": "unavailable"}
|
|
sys.stdout.write(json.dumps(result, allow_nan=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|