Relay endpoint — receives a JSON relay payload, fetches the URL.
(self)
| 240 | ) |
| 241 | |
| 242 | def do_POST(self): # noqa: N802 |
| 243 | """Relay endpoint — receives a JSON relay payload, fetches the URL.""" |
| 244 | content_length = int(self.headers.get("Content-Length") or 0) |
| 245 | if content_length <= 0: |
| 246 | self._send_json(400, {"e": "empty_body"}) |
| 247 | return |
| 248 | if content_length > _MAX_REQUEST_BODY: |
| 249 | self._send_json(413, {"e": "request_too_large"}) |
| 250 | return |
| 251 | |
| 252 | raw = self.rfile.read(content_length) |
| 253 | try: |
| 254 | body = json.loads(raw) |
| 255 | except Exception: |
| 256 | self._send_json(400, {"e": "bad_json"}) |
| 257 | return |
| 258 | |
| 259 | if not isinstance(body, dict): |
| 260 | self._send_json(400, {"e": "bad_json"}) |
| 261 | return |
| 262 | |
| 263 | k = str(body.get("k") or "") |
| 264 | u = str(body.get("u") or "") |
| 265 | m = str(body.get("m") or "GET").upper() |
| 266 | h = _sanitize_headers(body.get("h")) |
| 267 | b64 = body.get("b") |
| 268 | |
| 269 | if not _PSK: |
| 270 | self._send_json(500, {"e": "server_psk_missing"}) |
| 271 | return |
| 272 | |
| 273 | if k != _PSK: |
| 274 | log.warning("Rejected unauthorized request from %s", self.client_address[0]) |
| 275 | self._send_json(401, {"e": "unauthorized"}) |
| 276 | return |
| 277 | |
| 278 | if not _safe_url(u): |
| 279 | self._send_json(400, {"e": "bad_url"}) |
| 280 | return |
| 281 | |
| 282 | payload_bytes = b"" |
| 283 | if isinstance(b64, str) and b64: |
| 284 | try: |
| 285 | payload_bytes = base64.b64decode(b64) |
| 286 | except Exception: |
| 287 | self._send_json(400, {"e": "bad_base64"}) |
| 288 | return |
| 289 | |
| 290 | log.info("Relaying %s %s", m, u[:100]) |
| 291 | try: |
| 292 | result = _relay_request(u, m, h, payload_bytes) |
| 293 | except Exception as exc: |
| 294 | log.warning("Relay error for %s: %s", u[:80], exc) |
| 295 | self._send_json(500, {"e": str(exc) or type(exc).__name__}) |
| 296 | return |
| 297 | |
| 298 | log.info("Relay OK %s → HTTP %d (%d B)", u[:80], result["s"], len(result.get("b", ""))) |
| 299 | self._send_json(200, result) |
nothing calls this directly
no test coverage detected