#!/usr/bin/env python3 import argparse import json import mimetypes import os import re import secrets import shutil import subprocess import threading import urllib.request import uuid from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlsplit VERSION = "0.1.0-private-workspace" AGENT_ID = "scanforge" STATIC_ROOT = Path( os.environ["SHIRE_SCANFORGE_STATIC_ROOT"] ).resolve() DATA_ROOT = Path( os.environ["SHIRE_SCANFORGE_DATA_ROOT"] ).resolve() WORKSPACE_TOKEN = os.environ.get( "SHIRE_WORKSPACE_TOKEN", "", ) RUNTIME_BASE = os.environ.get( "SHIRE_SCANFORGE_RUNTIME_BASE", "http://127.0.0.1:8784", ).rstrip("/") PROJECTS_FILE = DATA_ROOT / "projects.json" MAX_BODY = 2 * 1024 * 1024 STARTED_AT = datetime.now(timezone.utc) LOCK = threading.RLock() def now_iso(): return datetime.now(timezone.utc).isoformat() def runtime_health(): try: with urllib.request.urlopen( f"{RUNTIME_BASE}/health", timeout=3, ) as response: payload = json.loads( response.read().decode("utf-8") ) return { "online": ( response.status == 200 and payload.get("ok") is True ), "payload": payload, "error": None, } except Exception as exc: return { "online": False, "payload": {}, "error": str(exc), } def kinect_hardware(): try: result = subprocess.run( ["lsusb"], capture_output=True, text=True, timeout=3, check=False, ) lines = [ line.strip() for line in result.stdout.splitlines() if line.strip() ] except Exception as exc: return { "detected": False, "motor": False, "camera": False, "audio": False, "devices": [], "error": str(exc), } patterns = { "motor": r"\b045e:02b0\b", "camera": r"\b045e:02ae\b", "audio": r"\b045e:02bb\b", } matches = { name: [ line for line in lines if re.search(pattern, line, re.I) ] for name, pattern in patterns.items() } devices = [] for group in matches.values(): for device in group: if device not in devices: devices.append(device) return { "detected": bool(devices), "motor": bool(matches["motor"]), "camera": bool(matches["camera"]), "audio": bool(matches["audio"]), "devices": devices, "error": ( None if result.returncode == 0 else result.stderr.strip() or "lsusb failed" ), } def load_projects(): with LOCK: if not PROJECTS_FILE.exists(): return [] try: value = json.loads( PROJECTS_FILE.read_text( encoding="utf-8" ) ) except Exception: return [] return value if isinstance(value, list) else [] def save_projects(projects): with LOCK: DATA_ROOT.mkdir( parents=True, exist_ok=True, ) temporary = PROJECTS_FILE.with_suffix( ".json.new" ) temporary.write_text( json.dumps(projects, indent=2) + "\n", encoding="utf-8", ) os.replace( temporary, PROJECTS_FILE, ) def create_project(payload): timestamp = now_iso() project_id = str(uuid.uuid4()) name = str( payload.get("name") or payload.get("projectName") or payload.get("objectName") or "Untitled Scan Project" ).strip()[:160] project = { **payload, "id": project_id, "uuid": project_id, "name": name, "status": "planned", "workflow_stage": "capture_planning", "workflowStage": "capture_planning", "scale_status": "unscaled", "scaleStatus": "unscaled", "capture_ready": False, "captureReady": False, "bridge_required": True, "bridgeRequired": True, "createdAt": timestamp, "created_date": timestamp, "updatedAt": timestamp, "updated_date": timestamp, } projects = load_projects() projects.insert(0, project) save_projects(projects) return project def current_status(): runtime = runtime_health() hardware = kinect_hardware() disk = shutil.disk_usage(DATA_ROOT) warning = None if not hardware["detected"]: warning = "KINECT NOT DETECTED" elif not runtime["online"]: warning = "SCANFORGE RUNTIME OFFLINE" else: warning = "SCAN BRIDGE REQUIRED" return { "agentId": AGENT_ID, "agentState": ( "online" if runtime["online"] else "offline" ), "bridgeState": "offline", "kinectState": ( "detected" if hardware["detected"] else "not_detected" ), "operation": "idle", "progress": 0, "activeProjectId": None, "activeProjectName": None, "lastSuccessfulScan": None, "warning": warning, "error": runtime["error"], "captureBridgeConnected": False, "calibrationVerified": False, "readyForCapture": False, "health": { "runtime_online": runtime["online"], "rgb_stream": False, "depth_stream": False, "capture_bridge_connected": False, "ready_for_capture": False, "available_disk": disk.free, "workspace_version": VERSION, }, "device": { "provider": "Xbox 360 Kinect", "hardwareDetected": hardware["detected"], "motorDetected": hardware["motor"], "cameraDetected": hardware["camera"], "audioDetected": hardware["audio"], "devices": hardware["devices"], }, "capabilities": { "scanner.status": True, "scan.plan": True, "capture.coordinate": False, "mesh.handoff": True, "scanner.kinect.v1": False, }, } class Handler(BaseHTTPRequestHandler): server_version = f"SHiREScanForge/{VERSION}" def security_headers(self): 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( "Content-Security-Policy", "default-src 'self'; " "script-src 'self'; " "style-src 'self' 'unsafe-inline'; " "img-src 'self' data: blob:; " "font-src 'self' data:; " "connect-src 'self' ws: wss:; " "object-src 'none'; " "base-uri 'self'; " "form-action 'self'", ) def json_response(self, status, payload): body = json.dumps( payload, indent=2, ).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 read_json(self): length = int( self.headers.get( "Content-Length", "0", ) or 0 ) if length > MAX_BODY: raise ValueError( "request_body_too_large" ) if length == 0: return {} payload = json.loads( self.rfile.read(length).decode("utf-8") ) if not isinstance(payload, dict): raise ValueError( "request_body_must_be_an_object" ) return payload def write_authorised(self): supplied = self.headers.get( "X-SHiRE-Workspace-Token", "", ) return ( bool(WORKSPACE_TOKEN) and secrets.compare_digest( supplied, WORKSPACE_TOKEN, ) ) def require_write_authorisation(self): if self.write_authorised(): return True self.json_response( HTTPStatus.UNAUTHORIZED, { "ok": False, "error": ( "workspace_authentication_required" ), }, ) return False def serve_frontend(self, request_path): prefix = "/agents/scanforge" relative = request_path[len(prefix):].lstrip("/") requested = ( STATIC_ROOT / relative if relative else STATIC_ROOT / "index.html" ).resolve() try: permitted = ( os.path.commonpath( [ str(requested), str(STATIC_ROOT), ] ) == str(STATIC_ROOT) ) except ValueError: permitted = False if not permitted: self.send_error( HTTPStatus.FORBIDDEN ) return candidate = requested if not candidate.is_file(): candidate = STATIC_ROOT / "index.html" if not candidate.is_file(): self.send_error( HTTPStatus.NOT_FOUND ) return body = candidate.read_bytes() self.send_response(HTTPStatus.OK) self.send_header( "Content-Type", mimetypes.guess_type(candidate.name)[0] or "application/octet-stream", ) self.send_header( "Content-Length", str(len(body)), ) self.send_header( "Cache-Control", ( "no-store" if candidate.name == "index.html" else "public, max-age=31536000, immutable" ), ) self.security_headers() self.end_headers() self.wfile.write(body) def do_GET(self): parsed = urlsplit(self.path) path = parsed.path if path == "/health": runtime = runtime_health() self.json_response( HTTPStatus.OK, { "ok": True, "service": ( "shire-scanforge-workspace" ), "version": VERSION, "boundTo": ( f"{self.server.server_address[0]}:" f"{self.server.server_address[1]}" ), "frontendBuilt": ( STATIC_ROOT.joinpath( "index.html" ).is_file() ), "runtimeOnline": runtime["online"], "publicInboundPorts": False, "startedAt": STARTED_AT.isoformat(), }, ) return if path == "/api/v1/modules/scanforge/context": self.json_response( HTTPStatus.OK, { "context": { "sessionId": "armor-local-session", "userId": "ray", "userRole": "owner_admin", "deviceId": "Work", "apiBaseUrl": "", "agentId": AGENT_ID, "permissions": [ "scanforge.read", "scanforge.plan", "scanforge.approve", "scanforge.cancel", ], "accessToken": None, } }, ) return if path == "/api/v1/agents/scanforge": self.json_response( HTTPStatus.OK, { "id": AGENT_ID, "name": "ScanForge Scanner Agent", "version": VERSION, "route": "/agents/scanforge", "workspaceMode": "SHIRE_MODULE", "dedicatedFrontend": True, "base44RuntimeDependency": False, "publicInboundPorts": False, }, ) return if path == "/api/v1/agents/scanforge/status": self.json_response( HTTPStatus.OK, current_status(), ) return if path == "/api/v1/agents/scanforge/capabilities": self.json_response( HTTPStatus.OK, { "agentId": AGENT_ID, "capabilities": current_status()[ "capabilities" ], "blockedActions": [ "fake_hardware_ready", "medical_certification", "socket_or_implant_geometry", "forge_auto_unlock", ], }, ) return if path == "/api/v1/agents/scanforge/projects": query = parse_qs(parsed.query) try: limit = min( max( int( query.get( "limit", ["100"], )[0] ), 1, ), 500, ) except Exception: limit = 100 self.json_response( HTTPStatus.OK, { "projects": load_projects()[:limit] }, ) return if path == "/api/v1/agents/scanforge/jobs": self.json_response( HTTPStatus.OK, { "jobs": [] }, ) return if path == "/": self.send_response(HTTPStatus.FOUND) self.send_header( "Location", "/agents/scanforge/", ) self.end_headers() return if path == "/agents/scanforge": self.send_response(HTTPStatus.FOUND) self.send_header( "Location", "/agents/scanforge/", ) self.end_headers() return if path.startswith("/agents/scanforge/"): self.serve_frontend(path) return self.json_response( HTTPStatus.NOT_FOUND, { "ok": False, "error": "not_found", "path": path, }, ) def do_POST(self): path = urlsplit(self.path).path if not self.require_write_authorisation(): return try: payload = self.read_json() except Exception as exc: self.json_response( HTTPStatus.BAD_REQUEST, { "ok": False, "error": str(exc), }, ) return if path == "/api/v1/agents/scanforge/projects": self.json_response( HTTPStatus.CREATED, { "ok": True, "project": create_project(payload), }, ) return if path == "/api/v1/agents/scanforge/actions": action = str( payload.get("action") or "" ).strip() if action == "open_dashboard": self.json_response( HTTPStatus.OK, { "ok": True, "action": action, }, ) return if action == "new_scan": parameters = payload.get( "parameters" ) if not isinstance(parameters, dict): parameters = {} self.json_response( HTTPStatus.CREATED, { "ok": True, "action": action, "project": create_project( parameters ), }, ) return if action in { "start_capture", "pause_capture", "resume_capture", "finish_capture", "cancel_capture", "process_scan", "open_mesh", "run_print_check", "prepare_export", "send_to_forge", "save_to_shirevault", }: self.json_response( HTTPStatus.CONFLICT, { "ok": False, "error": "SCAN_BRIDGE_REQUIRED", "action": action, "operationStarted": False, }, ) return self.json_response( HTTPStatus.BAD_REQUEST, { "ok": False, "error": ( "unsupported_scanforge_action" ), "action": action, }, ) return if path == "/api/v1/agents/scanforge/start": self.json_response( HTTPStatus.OK, { "ok": True, "state": "already_running", }, ) return if path in { "/api/v1/agents/scanforge/stop", "/api/v1/agents/scanforge/restart", }: self.json_response( HTTPStatus.CONFLICT, { "ok": False, "error": ( "armor_approval_required" ), }, ) return self.json_response( HTTPStatus.NOT_FOUND, { "ok": False, "error": "not_found", "path": path, }, ) def log_message(self, format_string, *args): print( f"{self.client_address[0]} " f"[{self.log_date_time_string()}] " f"{format_string % args}", flush=True, ) def main(): parser = argparse.ArgumentParser() parser.add_argument( "--host", default="127.0.0.1", ) parser.add_argument( "--port", type=int, default=8785, ) args = parser.parse_args() if not WORKSPACE_TOKEN: raise SystemExit( "SHIRE_WORKSPACE_TOKEN is missing." ) if not STATIC_ROOT.joinpath( "index.html" ).is_file(): raise SystemExit( "ScanForge frontend is missing." ) DATA_ROOT.mkdir( parents=True, exist_ok=True, ) server = ThreadingHTTPServer( (args.host, args.port), Handler, ) print( f"ScanForge workspace {VERSION} listening on " f"http://{args.host}:{args.port}", flush=True, ) server.serve_forever() if __name__ == "__main__": main()