#!/usr/bin/env python3 from __future__ import annotations import hashlib import json import os import re import secrets import sqlite3 import sys import threading import time import uuid from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import parse_qs, unquote, urlparse VERSION = "0.1.0-native-foundation" HOST = os.environ.get("SHIRE_GATEWAY_HOST", "127.0.0.1") PORT = int(os.environ.get("SHIRE_GATEWAY_PORT", "8786")) TOKEN = os.environ.get("SHIRE_GATEWAY_TOKEN", "") DB_PATH = Path( os.environ.get( "SHIRE_GATEWAY_DB", "/home/shire3d/ARMOR/data/agents/limbforge-canvas/" "limbforge-canvas.sqlite3", ) ) UPLOAD_ROOT = Path( os.environ.get( "SHIRE_GATEWAY_UPLOAD_ROOT", "/SHiREVault/SHiRELimbs", ) ) MAX_JSON_BYTES = 8 * 1024 * 1024 MAX_UPLOAD_BYTES = 150 * 1024 * 1024 COLLECTION_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,79}$") ALLOWED_COLLECTIONS = { # LimbForge "AuditEvent", "CompletedCover", "Component", "CoverProject", "Customer", "EnvelopeStation", "ImplementationReport", "Invitation", "LimbProfile", "ManufacturingJob", "MeasurementDefinition", "MeasurementSession", "MeasurementValue", "Order", "OssiguardProfile", "Prosthesis", "Quote", "Shipment", "SystemSettings", "WorkshopAccount", # CoverCanvas "ArtworkAsset", "ArtworkLayer", "CoverCanvasProject", "CoverDesignVariant", "CoverModel", "GeometryJob", "PromptConcept", "UniversalDesign", "UniversalDesignLayer", "ValidationReport", } WRITE_LOCK = threading.RLock() def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def connect() -> sqlite3.Connection: DB_PATH.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect( DB_PATH, timeout=30, ) connection.row_factory = sqlite3.Row connection.execute("PRAGMA journal_mode=WAL") connection.execute("PRAGMA foreign_keys=ON") connection.execute("PRAGMA busy_timeout=30000") return connection def initialise_database() -> None: with connect() as connection: connection.execute( """ CREATE TABLE IF NOT EXISTS records ( collection TEXT NOT NULL, id TEXT NOT NULL, uuid TEXT, record_state TEXT NOT NULL DEFAULT 'active', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY (collection, id) ) """ ) connection.execute( """ CREATE INDEX IF NOT EXISTS idx_records_collection_updated ON records(collection, updated_at DESC) """ ) connection.execute( """ CREATE INDEX IF NOT EXISTS idx_records_collection_uuid ON records(collection, uuid) """ ) connection.execute( """ CREATE TABLE IF NOT EXISTS audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, occurred_at TEXT NOT NULL, action TEXT NOT NULL, collection TEXT, record_id TEXT, detail TEXT NOT NULL ) """ ) def json_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") def normalise_collection(value: str) -> str: if not COLLECTION_RE.fullmatch(value): raise ValueError("Invalid collection name") if value not in ALLOWED_COLLECTIONS: raise ValueError( f"Collection is not enabled: {value}" ) return value def safe_relative_destination(value: str) -> Path: cleaned = unquote(value or "").strip().replace("\\", "/") cleaned = cleaned.lstrip("/") if cleaned.startswith("SHiRELimbs/"): cleaned = cleaned[len("SHiRELimbs/"):] parts = [ part for part in cleaned.split("/") if part not in ("", ".") ] if not parts: return Path("Uploads") if any(part == ".." for part in parts): raise ValueError("Unsafe upload destination") return Path(*parts) def parse_sort(sort_value: str | None) -> tuple[str, bool]: value = (sort_value or "-updated_date").strip() descending = value.startswith("-") field = value[1:] if descending else value if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", field): field = "updated_date" return field, descending def nested_get(data: dict[str, Any], key: str) -> Any: current: Any = data for part in key.split("."): if not isinstance(current, dict): return None current = current.get(part) return current def matches_query( payload: dict[str, Any], query: dict[str, Any], ) -> bool: for key, expected in query.items(): actual = nested_get(payload, key) if isinstance(expected, dict): if "$in" in expected: values = expected.get("$in") if not isinstance(values, list): return False if actual not in values: return False continue if "$ne" in expected: if actual == expected.get("$ne"): return False continue if actual != expected: return False return True def decode_record(row: sqlite3.Row) -> dict[str, Any]: payload = json.loads(row["payload"]) payload.setdefault("id", row["id"]) payload.setdefault("uuid", row["uuid"]) payload.setdefault("record_state", row["record_state"]) payload.setdefault("created_date", row["created_at"]) payload.setdefault("updated_date", row["updated_at"]) return payload def add_audit( connection: sqlite3.Connection, action: str, collection: str | None, record_id: str | None, detail: str, ) -> None: connection.execute( """ INSERT INTO audit ( occurred_at, action, collection, record_id, detail ) VALUES (?, ?, ?, ?, ?) """, ( utc_now(), action, collection, record_id, detail, ), ) class GatewayHandler(BaseHTTPRequestHandler): server_version = "SHiRELocalGateway/0.1" def log_message( self, format_string: str, *args: Any, ) -> None: sys.stdout.write( "%s - %s\n" % ( self.log_date_time_string(), format_string % args, ) ) sys.stdout.flush() def _send( self, status: int, payload: Any, ) -> None: body = json_bytes(payload) 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.send_header( "X-Content-Type-Options", "nosniff", ) self.send_header( "Content-Security-Policy", "default-src 'none'", ) self.end_headers() self.wfile.write(body) def _error( self, status: int, message: str, ) -> None: self._send( status, { "ok": False, "error": message, }, ) def _authorised(self) -> bool: if not TOKEN: self._error( 503, "Gateway token is not configured", ) return False supplied = self.headers.get( "X-SHiRE-Workspace-Token", "", ) if not secrets.compare_digest( supplied, TOKEN, ): self._error( 401, "Unauthorised", ) return False return True def _read_json(self) -> dict[str, Any]: try: length = int( self.headers.get( "Content-Length", "0", ) ) except ValueError as exc: raise ValueError( "Invalid Content-Length" ) from exc if length <= 0: return {} if length > MAX_JSON_BYTES: raise ValueError( "JSON request is too large" ) raw = self.rfile.read(length) try: value = json.loads( raw.decode("utf-8") ) except ( UnicodeDecodeError, json.JSONDecodeError, ) as exc: raise ValueError( "Invalid JSON" ) from exc if not isinstance(value, dict): raise ValueError( "JSON body must be an object" ) return value def _path_parts(self) -> list[str]: parsed = urlparse(self.path) return [ unquote(part) for part in parsed.path.split("/") if part ] def do_GET(self) -> None: try: parsed = urlparse(self.path) parts = self._path_parts() if parsed.path == "/health": self._send( 200, { "ok": True, "service": ( "shire-limbforge-canvas-gateway" ), "version": VERSION, "host": HOST, "port": PORT, "database": str(DB_PATH), "uploadRoot": str(UPLOAD_ROOT), "tokenRequired": True, "applications": [ "limbforge", "covercanvas", "ossiguard", ], "time": utc_now(), }, ) return if not self._authorised(): return if parts == ["api", "v1", "session"]: self._send( 200, { "ok": True, "authenticated": True, "user": { "id": "shire-local-owner", "email": "", "full_name": "Ray", "role": "owner_admin", "roles": [ "owner_admin", "admin", "builder", ], "local_only": True, }, }, ) return if ( len(parts) >= 4 and parts[:3] == [ "api", "v1", "collections", ] ): collection = normalise_collection( parts[3] ) if len(parts) == 4: query_args = parse_qs( parsed.query ) sort_value = query_args.get( "sort", ["-updated_date"], )[0] limit_value = query_args.get( "limit", ["500"], )[0] try: limit = max( 1, min( int(limit_value), 2000, ), ) except ValueError: limit = 500 with connect() as connection: rows = connection.execute( """ SELECT * FROM records WHERE collection = ? """, (collection,), ).fetchall() records = [ decode_record(row) for row in rows ] field, descending = parse_sort( sort_value ) records.sort( key=lambda item: ( nested_get( item, field, ) is None, str( nested_get( item, field, ) or "" ), ), reverse=descending, ) self._send( 200, { "ok": True, "records": records[:limit], }, ) return if len(parts) == 5: record_id = parts[4] with connect() as connection: row = connection.execute( """ SELECT * FROM records WHERE collection = ? AND ( id = ? OR uuid = ? ) LIMIT 1 """, ( collection, record_id, record_id, ), ).fetchone() if row is None: self._error( 404, "Record not found", ) return self._send( 200, { "ok": True, "record": decode_record( row ), }, ) return self._error( 404, "Route not found", ) except ValueError as exc: self._error( 400, str(exc), ) except Exception as exc: self._error( 500, f"Internal error: {exc}", ) def do_POST(self) -> None: try: parts = self._path_parts() if not self._authorised(): return if parts == [ "api", "v1", "files", ]: try: length = int( self.headers.get( "Content-Length", "0", ) ) except ValueError as exc: raise ValueError( "Invalid Content-Length" ) from exc if length <= 0: raise ValueError( "Upload is empty" ) if length > MAX_UPLOAD_BYTES: raise ValueError( "Upload exceeds 150 MB" ) filename = Path( self.headers.get( "X-SHiRE-Filename", "upload.bin", ) ).name destination = safe_relative_destination( self.headers.get( "X-SHiRE-Destination", "Uploads", ) ) destination_dir = ( UPLOAD_ROOT / destination ).resolve() root_resolved = UPLOAD_ROOT.resolve() if ( root_resolved not in destination_dir.parents and destination_dir != root_resolved ): raise ValueError( "Unsafe upload path" ) destination_dir.mkdir( parents=True, exist_ok=True, ) unique_name = ( f"{int(time.time())}-" f"{uuid.uuid4().hex[:12]}-" f"{filename}" ) final_path = destination_dir / unique_name temp_path = final_path.with_suffix( final_path.suffix + ".partial" ) digest = hashlib.sha256() remaining = length with temp_path.open("wb") as handle: while remaining: chunk = self.rfile.read( min( remaining, 1024 * 1024, ) ) if not chunk: raise ValueError( "Upload ended unexpectedly" ) handle.write(chunk) digest.update(chunk) remaining -= len(chunk) handle.flush() os.fsync(handle.fileno()) temp_path.replace(final_path) self._send( 201, { "ok": True, "file": { "location": str( final_path ), "destination": str( destination ), "filename": filename, "checksum": digest.hexdigest(), "verified": True, "storage": "SHiREVault", "size": length, }, }, ) return if ( len(parts) >= 5 and parts[:3] == [ "api", "v1", "collections", ] ): collection = normalise_collection( parts[3] ) operation = parts[4] body = self._read_json() if operation == "filter": query = body.get( "query", {}, ) if not isinstance(query, dict): raise ValueError( "Filter query must be an object" ) sort_value = str( body.get( "sort", "-updated_date", ) ) limit = max( 1, min( int( body.get( "limit", 500, ) ), 2000, ), ) with connect() as connection: rows = connection.execute( """ SELECT * FROM records WHERE collection = ? """, (collection,), ).fetchall() records = [ decode_record(row) for row in rows ] records = [ record for record in records if matches_query( record, query, ) ] field, descending = parse_sort( sort_value ) records.sort( key=lambda item: ( nested_get( item, field, ) is None, str( nested_get( item, field, ) or "" ), ), reverse=descending, ) self._send( 200, { "ok": True, "records": records[:limit], }, ) return if operation == "bulk-create": records = body.get( "records", [], ) if not isinstance(records, list): raise ValueError( "records must be a list" ) created: list[ dict[str, Any] ] = [] with WRITE_LOCK, connect() as connection: for item in records: if not isinstance(item, dict): raise ValueError( "Each record must be an object" ) created.append( self._create_record( connection, collection, item, ) ) connection.commit() self._send( 201, { "ok": True, "records": created, }, ) return if ( len(parts) == 4 and parts[:3] == [ "api", "v1", "collections", ] ): collection = normalise_collection( parts[3] ) body = self._read_json() with WRITE_LOCK, connect() as connection: record = self._create_record( connection, collection, body, ) connection.commit() self._send( 201, { "ok": True, "record": record, }, ) return self._error( 404, "Route not found", ) except ValueError as exc: self._error( 400, str(exc), ) except Exception as exc: self._error( 500, f"Internal error: {exc}", ) def _create_record( self, connection: sqlite3.Connection, collection: str, body: dict[str, Any], ) -> dict[str, Any]: now = utc_now() record = dict(body) record_id = str( record.get("id") or uuid.uuid4() ) record_uuid = str( record.get("uuid") or f"{collection.lower()}-{uuid.uuid4()}" ) record["id"] = record_id record["uuid"] = record_uuid record.setdefault( "record_state", "active", ) record.setdefault( "created_date", now, ) record["updated_date"] = now connection.execute( """ INSERT INTO records ( collection, id, uuid, record_state, created_at, updated_at, payload ) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( collection, record_id, record_uuid, str( record.get( "record_state", "active", ) ), str( record.get( "created_date", now, ) ), now, json.dumps( record, ensure_ascii=False, ), ), ) add_audit( connection, "create", collection, record_id, f"Created {collection}", ) return record def do_PATCH(self) -> None: try: parts = self._path_parts() if not self._authorised(): return if not ( len(parts) == 5 and parts[:3] == [ "api", "v1", "collections", ] ): self._error( 404, "Route not found", ) return collection = normalise_collection( parts[3] ) record_id = parts[4] changes = self._read_json() with WRITE_LOCK, connect() as connection: row = connection.execute( """ SELECT * FROM records WHERE collection = ? AND ( id = ? OR uuid = ? ) LIMIT 1 """, ( collection, record_id, record_id, ), ).fetchone() if row is None: self._error( 404, "Record not found", ) return record = decode_record(row) record.update(changes) record["updated_date"] = utc_now() connection.execute( """ UPDATE records SET uuid = ?, record_state = ?, updated_at = ?, payload = ? WHERE collection = ? AND id = ? """, ( str( record.get( "uuid", row["uuid"], ) ), str( record.get( "record_state", "active", ) ), record["updated_date"], json.dumps( record, ensure_ascii=False, ), collection, row["id"], ), ) add_audit( connection, "update", collection, row["id"], f"Updated {collection}", ) connection.commit() self._send( 200, { "ok": True, "record": record, }, ) except ValueError as exc: self._error( 400, str(exc), ) except Exception as exc: self._error( 500, f"Internal error: {exc}", ) def do_DELETE(self) -> None: try: parts = self._path_parts() if not self._authorised(): return if not ( len(parts) == 5 and parts[:3] == [ "api", "v1", "collections", ] ): self._error( 404, "Route not found", ) return collection = normalise_collection( parts[3] ) record_id = parts[4] with WRITE_LOCK, connect() as connection: row = connection.execute( """ SELECT id FROM records WHERE collection = ? AND ( id = ? OR uuid = ? ) LIMIT 1 """, ( collection, record_id, record_id, ), ).fetchone() if row is None: self._error( 404, "Record not found", ) return connection.execute( """ DELETE FROM records WHERE collection = ? AND id = ? """, ( collection, row["id"], ), ) add_audit( connection, "delete", collection, row["id"], f"Deleted {collection}", ) connection.commit() self._send( 200, { "ok": True, "deleted": record_id, }, ) except ValueError as exc: self._error( 400, str(exc), ) except Exception as exc: self._error( 500, f"Internal error: {exc}", ) def main() -> None: if HOST != "127.0.0.1": raise SystemExit( "Refusing to bind outside 127.0.0.1" ) if not TOKEN: raise SystemExit( "SHIRE_GATEWAY_TOKEN is required" ) UPLOAD_ROOT.mkdir( parents=True, exist_ok=True, ) initialise_database() server = ThreadingHTTPServer( (HOST, PORT), GatewayHandler, ) print( json.dumps( { "service": ( "shire-limbforge-canvas-gateway" ), "version": VERSION, "listen": f"{HOST}:{PORT}", "database": str(DB_PATH), "uploadRoot": str(UPLOAD_ROOT), } ), flush=True, ) server.serve_forever() if __name__ == "__main__": main()