#!/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=", "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()