Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Hardened evidence gateway + static server for the SZL Holographic Estate. | |
| Serves index.html + style.css on port 7860 (docker SDK) with security | |
| response headers. Additive / non-breaking: the CSP permits every resource | |
| this page actually uses (inline scripts + styles for the estate UI, data: | |
| favicon, and the live status/verify fetches to the a11oy flagship), so the | |
| estate and its live HUD keep working. Everything else is 'self' — zero CDN. | |
| """ | |
| import copy | |
| import functools | |
| import json | |
| import os | |
| import threading | |
| import time | |
| import urllib.error | |
| import urllib.parse | |
| import urllib.request | |
| from datetime import datetime, timezone | |
| from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer | |
| from szl_source_attestation import build_attestation | |
| PORT = 7860 | |
| DIRECTORY = os.environ.get("APP_DIRECTORY", "/app") | |
| UPSTREAM = "https://a-11-oy.com" | |
| SPACE_ID = "SZLHOLDINGS/holographic" | |
| SOURCE_OBSERVATION = { | |
| "repository": None, | |
| "commit": None, | |
| "path": None, | |
| "relation": "no-verifiable-direct-source-observed", | |
| "state": "UNKNOWN", | |
| "evidence_url": None, | |
| } | |
| UPSTREAM_TIMEOUT_SECONDS = 6 | |
| MAX_QUERY_CHARS = 240 | |
| MAX_CACHE_ENTRIES = 64 | |
| _CACHE = {} | |
| _CACHE_LOCK = threading.Lock() | |
| CONTENT_SECURITY_POLICY = ( | |
| "default-src 'self'; " | |
| "base-uri 'self'; " | |
| "object-src 'none'; " | |
| "script-src 'self' 'unsafe-inline'; " | |
| "style-src 'self' 'unsafe-inline'; " | |
| "img-src 'self' data: blob:; " | |
| "font-src 'self'; " | |
| "connect-src 'self' https://a-11-oy.com https://szlholdings-a11oy.hf.space " | |
| "https://szlholdings-killinchu.hf.space; " | |
| "frame-ancestors 'self' https://huggingface.co https://*.hf.space https://*.huggingface.co" | |
| ) | |
| def _now_iso(): | |
| return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") | |
| def _fetch_json(path): | |
| req = urllib.request.Request( | |
| UPSTREAM + path, | |
| headers={"Accept": "application/json", "User-Agent": "szl-holographic-estate/2"}, | |
| ) | |
| with urllib.request.urlopen(req, timeout=UPSTREAM_TIMEOUT_SECONDS) as response: | |
| if response.status != 200: | |
| raise RuntimeError("upstream HTTP %s" % response.status) | |
| return json.loads(response.read().decode("utf-8")) | |
| def _live_or_last_good(key, producer): | |
| try: | |
| payload = producer() | |
| with _CACHE_LOCK: | |
| _CACHE.pop(key, None) | |
| _CACHE[key] = {"stored_monotonic": time.monotonic(), "payload": copy.deepcopy(payload)} | |
| while len(_CACHE) > MAX_CACHE_ENTRIES: | |
| _CACHE.pop(next(iter(_CACHE))) | |
| return 200, payload | |
| except Exception as exc: | |
| with _CACHE_LOCK: | |
| cached = copy.deepcopy(_CACHE.get(key)) | |
| if cached: | |
| payload = cached["payload"] | |
| payload.update({ | |
| "transport_state": "DEGRADED", | |
| "evidence_state": "LAST_GOOD", | |
| "freshness_seconds": round(max(0.0, time.monotonic() - cached["stored_monotonic"]), 3), | |
| "upstream_error": "%s: %s" % (type(exc).__name__, str(exc)[:160]), | |
| }) | |
| return 200, payload | |
| return 503, { | |
| "schema": "szl.holographic-unavailable/v1", | |
| "observed_at": _now_iso(), | |
| "transport_state": "UNAVAILABLE", | |
| "evidence_state": "UNAVAILABLE", | |
| "error": "%s: %s" % (type(exc).__name__, str(exc)[:160]), | |
| "limits": ["No fabricated fallback values", "No live evidence was available for this request"], | |
| } | |
| def _atlas_live(): | |
| info = _fetch_json("/api/a11oy/v1/holographic/info") | |
| atlas = _fetch_json("/api/a11oy/v1/atlas/map") | |
| shell_ids = [str(row.get("id")) for row in info.get("surfaces", []) if row.get("id")] | |
| atlas_ids = [ | |
| str(row.get("id")) | |
| for cluster in atlas.get("clusters", []) | |
| for row in cluster.get("surfaces", []) | |
| if row.get("id") | |
| ] | |
| missing = sorted(set(shell_ids) - set(atlas_ids)) | |
| unexpected = sorted(set(atlas_ids) - set(shell_ids)) | |
| duplicate_ids = sorted({sid for sid in atlas_ids if atlas_ids.count(sid) > 1}) | |
| aligned = not missing and not unexpected and not duplicate_ids and len(shell_ids) == len(atlas_ids) | |
| return { | |
| "schema": "szl.holographic-atlas-alignment/v1", | |
| "observed_at": _now_iso(), | |
| "transport_state": "REACHABLE", | |
| "evidence_state": "LIVE", | |
| "state": "ALIGNED" if aligned else "DRIFT", | |
| "shell_surface_count": len(shell_ids), | |
| "atlas_surface_count": atlas.get("surface_count"), | |
| "atlas_total_classified": atlas.get("total_classified"), | |
| "atlas_coverage": atlas.get("coverage"), | |
| "missing_from_atlas": missing, | |
| "unexpected_in_atlas": unexpected, | |
| "duplicate_atlas_ids": duplicate_ids, | |
| "registry": atlas.get("registry"), | |
| "locked_core_count": atlas.get("locked_core_count"), | |
| "conjecture_cluster_gray": atlas.get("conjecture_cluster_gray"), | |
| "purpose": "Measure Atlas coverage against the canonical shell roster, not a stale internal denominator.", | |
| "limits": ["Cartography is MODELED", "Reachability is not readiness", "No maturity label is inferred here"], | |
| "sources": [ | |
| UPSTREAM + "/api/a11oy/v1/holographic/info", | |
| UPSTREAM + "/api/a11oy/v1/atlas/map", | |
| ], | |
| } | |
| def _brain_live(query): | |
| encoded = urllib.parse.urlencode({"q": query, "k": 12}) | |
| health = _fetch_json("/api/a11oy/v1/brain/health?" + encoded) | |
| gaps = _fetch_json("/api/a11oy/v1/brain/gaps?" + encoded) | |
| gap_payload = gaps.get("gaps") if isinstance(gaps.get("gaps"), dict) else gaps | |
| return { | |
| "schema": "szl.holographic-brain-trust/v1", | |
| "observed_at": _now_iso(), | |
| "transport_state": "REACHABLE", | |
| "evidence_state": "LIVE", | |
| "query": query, | |
| "verdict": health.get("verdict", "UNAVAILABLE"), | |
| "verdict_reason": health.get("verdict_reason"), | |
| "modeled_trust": health.get("modeled_trust"), | |
| "components": health.get("components", []), | |
| "component_summary": health.get("summary", {}), | |
| "gap_verdict": gaps.get("estate_verdict") or gap_payload.get("estate_verdict", "UNAVAILABLE"), | |
| "gap_metrics": gap_payload.get("metrics", {}), | |
| "purpose": "Expose whether the live knowledge graph can honestly support this query now.", | |
| "limits": [ | |
| "Read-only: no memory write, model call, or external action", | |
| "MODELED trust is not a probability or truth guarantee", | |
| "Unavailable honesty guards remain unavailable", | |
| "No consciousness or sentience claim", | |
| ], | |
| "sources": [ | |
| UPSTREAM + "/api/a11oy/v1/brain/health?" + encoded, | |
| UPSTREAM + "/api/a11oy/v1/brain/gaps?" + encoded, | |
| ], | |
| } | |
| def _manifest(): | |
| return { | |
| "schema": "szl.holographic-estate-manifest/v1", | |
| "contract_version": "2.0.0", | |
| "observed_at": _now_iso(), | |
| "transport_state": "REACHABLE", | |
| "evidence_state": "SNAPSHOT", | |
| "capabilities": { | |
| "atlas_alignment": "/api/holographic/v1/atlas", | |
| "brain_trust_probe": "/api/holographic/v1/brain/trust?query=<topic>", | |
| "live_3d_atlas": "https://a-11-oy.com/holographic#atlas", | |
| }, | |
| "interaction_contract": ["Purpose", "Try", "Evidence", "Limits", "Reproduce"], | |
| "write_policy": "READ_ONLY", | |
| "limits": ["Upstream live evidence may be unavailable", "LAST_GOOD is labelled with age", "RUNNING does not imply quality or readiness"], | |
| } | |
| class HardenedHandler(SimpleHTTPRequestHandler): | |
| def _send_json(self, status, payload): | |
| body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).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.send_header("X-SZL-Transport-State", str(payload.get("transport_state", "UNSPECIFIED"))) | |
| self.send_header("X-SZL-Evidence-State", str(payload.get("evidence_state", "UNSPECIFIED"))) | |
| if payload.get("verification_state"): | |
| self.send_header("X-SZL-Verification-State", str(payload["verification_state"])) | |
| if payload.get("authority_state"): | |
| self.send_header("X-SZL-Authority-State", str(payload["authority_state"])) | |
| self.end_headers() | |
| self.wfile.write(body) | |
| def do_GET(self): | |
| parsed = urllib.parse.urlsplit(self.path) | |
| if parsed.path == "/.well-known/szl-source.json": | |
| force = urllib.parse.parse_qs(parsed.query).get("refresh") == ["1"] | |
| self._send_json( | |
| 200, | |
| build_attestation( | |
| SPACE_ID, | |
| SOURCE_OBSERVATION, | |
| "UNKNOWN_SOURCE_RELATION", | |
| force=force, | |
| ), | |
| ) | |
| return | |
| if parsed.path == "/healthz": | |
| self._send_json(200, {"ok": True, "transport_state": "REACHABLE", "evidence_state": "LIVE"}) | |
| return | |
| if parsed.path == "/api/holographic/v1/manifest": | |
| self._send_json(200, _manifest()) | |
| return | |
| if parsed.path == "/api/holographic/v1/atlas": | |
| status, payload = _live_or_last_good("atlas", _atlas_live) | |
| self._send_json(status, payload) | |
| return | |
| if parsed.path == "/api/holographic/v1/brain/trust": | |
| query = urllib.parse.parse_qs(parsed.query).get("query", [""])[0].strip() | |
| if not query: | |
| self._send_json(400, { | |
| "schema": "szl.holographic-input-error/v1", | |
| "transport_state": "REACHABLE", | |
| "evidence_state": "UNAVAILABLE", | |
| "error": "query is required", | |
| }) | |
| return | |
| if len(query) > MAX_QUERY_CHARS: | |
| self._send_json(400, { | |
| "schema": "szl.holographic-input-error/v1", | |
| "transport_state": "REACHABLE", | |
| "evidence_state": "UNAVAILABLE", | |
| "error": "query exceeds %d characters" % MAX_QUERY_CHARS, | |
| }) | |
| return | |
| status, payload = _live_or_last_good("brain:" + query, lambda: _brain_live(query)) | |
| self._send_json(status, payload) | |
| return | |
| super().do_GET() | |
| def end_headers(self): | |
| self.send_header("Content-Security-Policy", CONTENT_SECURITY_POLICY) | |
| self.send_header("Strict-Transport-Security", "max-age=31536000; includeSubDomains") | |
| self.send_header("X-Content-Type-Options", "nosniff") | |
| self.send_header("Referrer-Policy", "strict-origin-when-cross-origin") | |
| self.send_header("Permissions-Policy", "camera=(), microphone=(), geolocation=()") | |
| super().end_headers() | |
| if __name__ == "__main__": | |
| handler = functools.partial(HardenedHandler, directory=DIRECTORY) | |
| with ThreadingHTTPServer(("0.0.0.0", PORT), handler) as httpd: | |
| print(f"[holographic] serving {DIRECTORY} on :{PORT}", flush=True) | |
| httpd.serve_forever() | |