#!/usr/bin/env python3 from __future__ import annotations import http.client import json import mimetypes import os import posixpath import sys from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import unquote, urlsplit VERSION = "0.1.0-native-workspace" HOST = os.environ.get("SHIRE_WORKSPACE_HOST", "127.0.0.1") PORT = int(os.environ["SHIRE_WORKSPACE_PORT"]) AGENT_ID = os.environ["SHIRE_WORKSPACE_AGENT_ID"] TITLE = os.environ["SHIRE_WORKSPACE_TITLE"] BUILD_ROOT = Path( os.environ["SHIRE_WORKSPACE_BUILD_ROOT"] ).resolve() ROUTE_ROOT = os.environ["SHIRE_WORKSPACE_ROUTE_ROOT"].rstrip("/") GATEWAY_HOST = os.environ.get( "SHIRE_GATEWAY_HOST", "127.0.0.1", ) GATEWAY_PORT = int( os.environ.get( "SHIRE_GATEWAY_PORT", "8786", ) ) GATEWAY_TOKEN = os.environ["SHIRE_GATEWAY_TOKEN"] MAX_PROXY_BYTES = 160 * 1024 * 1024 mimetypes.add_type( "application/javascript", ".js", ) mimetypes.add_type( "text/css", ".css", ) mimetypes.add_type( "image/svg+xml", ".svg", ) mimetypes.add_type( "model/gltf-binary", ".glb", ) mimetypes.add_type( "model/gltf+json", ".gltf", ) class WorkspaceHandler(BaseHTTPRequestHandler): server_version = "SHiREWorkspace/0.1" def log_message( self, format_string: str, *args: Any, ) -> None: sys.stdout.write( "%s %s %s\n" % ( AGENT_ID, self.log_date_time_string(), format_string % args, ) ) sys.stdout.flush() def _security_headers(self) -> None: self.send_header( "X-Content-Type-Options", "nosniff", ) self.send_header( "X-Frame-Options", "SAMEORIGIN", ) self.send_header( "Referrer-Policy", "no-referrer", ) self.send_header( "Permissions-Policy", "camera=(), microphone=(), geolocation=()", ) def _send_json( self, status: int, payload: Any, ) -> None: body = json.dumps( payload, ensure_ascii=False, ).encode("utf-8") self.send_response(status) self.send_header( "Content-Type", "application/json; charset=utf-8", ) self.send_header( "Content-Length", str(len(body)), ) self.send_header( "Cache-Control", "no-store", ) self._security_headers() self.end_headers() self.wfile.write(body) def _send_file( self, path: Path, ) -> None: try: data = path.read_bytes() except OSError: self.send_error(404) return content_type, _ = mimetypes.guess_type( path.name ) self.send_response(200) self.send_header( "Content-Type", content_type or "application/octet-stream", ) self.send_header( "Content-Length", str(len(data)), ) if path.name == "shire-index.html": self.send_header( "Cache-Control", "no-store", ) else: self.send_header( "Cache-Control", "public, max-age=31536000, immutable", ) self._security_headers() self.end_headers() self.wfile.write(data) def _read_body(self) -> bytes: raw_length = self.headers.get( "Content-Length", "0", ) try: length = int(raw_length) except ValueError as exc: raise ValueError( "Invalid Content-Length" ) from exc if length < 0: raise ValueError( "Invalid Content-Length" ) if length > MAX_PROXY_BYTES: raise ValueError( "Request exceeds workspace limit" ) if length == 0: return b"" return self.rfile.read(length) def _proxy(self) -> None: split = urlsplit(self.path) proxy_path = split.path if split.query: proxy_path += "?" + split.query try: body = self._read_body() headers = { "X-SHiRE-Workspace-Token": ( GATEWAY_TOKEN ), "Accept": self.headers.get( "Accept", "application/json", ), } content_type = self.headers.get( "Content-Type" ) if content_type: headers["Content-Type"] = content_type for name in ( "X-SHiRE-Filename", "X-SHiRE-Destination", ): value = self.headers.get(name) if value: headers[name] = value connection = http.client.HTTPConnection( GATEWAY_HOST, GATEWAY_PORT, timeout=120, ) connection.request( self.command, proxy_path, body=body or None, headers=headers, ) response = connection.getresponse() response_body = response.read() self.send_response(response.status) response_content_type = response.getheader( "Content-Type", "application/json", ) self.send_header( "Content-Type", response_content_type, ) self.send_header( "Content-Length", str(len(response_body)), ) self.send_header( "Cache-Control", "no-store", ) self._security_headers() self.end_headers() self.wfile.write(response_body) connection.close() except ValueError as exc: self._send_json( 400, { "ok": False, "error": str(exc), }, ) except Exception as exc: self._send_json( 502, { "ok": False, "error": ( "Native gateway unavailable: " f"{exc}" ), }, ) def _static_path(self) -> Path | None: split = urlsplit(self.path) path = unquote(split.path) if not path.startswith(ROUTE_ROOT): return None relative = path[len(ROUTE_ROOT):].lstrip("/") if not relative: return BUILD_ROOT / "shire-index.html" normalised = posixpath.normpath(relative) if ( normalised == ".." or normalised.startswith("../") ): return None candidate = ( BUILD_ROOT / Path(normalised) ).resolve() if ( candidate != BUILD_ROOT and BUILD_ROOT not in candidate.parents ): return None if candidate.is_file(): return candidate return BUILD_ROOT / "shire-index.html" def do_GET(self) -> None: split = urlsplit(self.path) if split.path == "/health": self._send_json( 200, { "ok": True, "service": ( "shire-specialist-workspace" ), "version": VERSION, "agentId": AGENT_ID, "title": TITLE, "host": HOST, "port": PORT, "routeRoot": ROUTE_ROOT, "buildRoot": str(BUILD_ROOT), "gateway": ( f"{GATEWAY_HOST}:" f"{GATEWAY_PORT}" ), "gatewayTokenExposed": False, }, ) return if split.path.startswith("/api/v1/"): self._proxy() return path = self._static_path() if path is None: self.send_error(404) return self._send_file(path) def do_POST(self) -> None: if urlsplit(self.path).path.startswith( "/api/v1/" ): self._proxy() return self.send_error(404) def do_PATCH(self) -> None: if urlsplit(self.path).path.startswith( "/api/v1/" ): self._proxy() return self.send_error(404) def do_DELETE(self) -> None: if urlsplit(self.path).path.startswith( "/api/v1/" ): self._proxy() return self.send_error(404) def main() -> None: if HOST != "127.0.0.1": raise SystemExit( "Refusing to bind outside 127.0.0.1" ) if not BUILD_ROOT.is_dir(): raise SystemExit( f"Build root missing: {BUILD_ROOT}" ) if not ( BUILD_ROOT / "shire-index.html" ).is_file(): raise SystemExit( "shire-index.html is missing" ) if not GATEWAY_TOKEN: raise SystemExit( "Gateway token is required" ) server = ThreadingHTTPServer( (HOST, PORT), WorkspaceHandler, ) print( json.dumps( { "service": ( "shire-specialist-workspace" ), "version": VERSION, "agentId": AGENT_ID, "listen": f"{HOST}:{PORT}", "routeRoot": ROUTE_ROOT, "buildRoot": str(BUILD_ROOT), } ), flush=True, ) server.serve_forever() if __name__ == "__main__": main()