Adom UI Design System

@@ -0,0 +1,290 @@+#!/usr/bin/env python3+"""proxy-fidelity — prove a dock dashboard forwards the control API FAITHFULLY.++The newbie-panel module calls two routes same-origin against your dashboard:++    GET  /updater/status        (version gate)+    POST /ai-threads/run-prompt (the send)++Your app forwards both to ah's control API. This checks that it forwards them+*unchanged*, which a "does it return 200" probe cannot see.++Three assertions, each against a real bug:++  1. response bytes identical (sha256), awkward whitespace and nested+     delivered_to included          -> catches parse-then-re-emit+  2. upstream's 207 comes back 207  -> catches a hand-built 200 {ok:true} wrapper+  3. an unproxied path under the same prefix returns YOUR 404, not upstream's+     418                            -> catches blanket-proxying the control API++Why 1 matters even when the JSON "looks the same": the module splits+updater.current_version on dots and compares NUMERICALLY, and reads+delivered_to as an ARRAY. A re-serialisation that retypes either one changes+behaviour while every status probe stays green.++USAGE++  # recommended: the script runs your app itself, in a throwaway HOME, so it+  # never touches a real workspace+  proxy-fidelity.py --app-cmd 'adom-project-manager --port $PORT'++  # or point it at an already-running app (needs --home so the stub URL can be+  # published where the app reads it; the file is backed up and restored)+  proxy-fidelity.py --base http://127.0.0.1:8785 --home ~++  # negative control: prove RED actually shows, using a deliberately+  # re-serialising proxy. Run this once before you trust a green run.+  proxy-fidelity.py --self-test++Exits non-zero if any assertion is RED, so it can sit in a build gate.+python3 stdlib only.+"""+import argparse, hashlib, json, os, random, shutil, signal, socket, subprocess, sys, time+import urllib.request, urllib.error+from http.server import BaseHTTPRequestHandler, HTTPServer+from threading import Thread++# Deliberately awkward: leading/trailing spaces that survive only if the bytes are+# untouched, a NUMERIC version, and delivered_to as an ARRAY.+STUB_BODY = (b'{"updater":{"current_version":"1.0.148","build":142},'+             b'"delivered_to":["tab-7"],"spacing":"  preserved  "}')+TEAPOT = b'{"teapot":true}'+++def free_port():+    s = socket.socket()+    s.bind(('127.0.0.1', 0))+    p = s.getsockname()[1]+    s.close()+    return p+++class Stub(BaseHTTPRequestHandler):+    """Stands in for ah's control API."""+    def _send(self, code, body):+        self.send_response(code)+        self.send_header('Content-Type', 'application/json')+        self.send_header('Content-Length', str(len(body)))+        self.end_headers()+        self.wfile.write(body)++    def do_GET(self):+        if self.path == '/updater/status':+            self._send(200, STUB_BODY)+        else:+            # anything else is a teapot, so a blanket proxy is visible+            self._send(418, TEAPOT)++    def do_POST(self):+        n = int(self.headers.get('Content-Length') or 0)+        raw = self.rfile.read(n)+        try:+            echo = json.loads(raw or b'{}')+        except Exception:+            echo = {'unparsed': raw.decode('utf-8', 'replace')}+        # 207 on purpose: a hand-built 200 wrapper cannot fake it+        self._send(207, json.dumps({'echo': echo, 'delivered_to': ['tab-9']}).encode())++    def log_message(self, *a):+        pass+++class BadProxy(BaseHTTPRequestHandler):+    """The NEGATIVE CONTROL: forwards correctly except it parses and re-emits,+    which is exactly what most hand-written forwarding routes do."""+    upstream = None++    def do_GET(self):+        if self.path == '/updater/status':+            with urllib.request.urlopen(self.upstream + '/updater/status') as r:+                data = json.loads(r.read())+            out = json.dumps(data).encode()   # <-- the bug, in one line+            self.send_response(200)+            self.send_header('Content-Type', 'application/json')+            self.send_header('Content-Length', str(len(out)))+            self.end_headers()+            self.wfile.write(out)+        else:+            self.send_response(404); self.end_headers(); self.wfile.write(b'not found')++    def do_POST(self):+        n = int(self.headers.get('Content-Length') or 0)+        raw = self.rfile.read(n)+        req = urllib.request.Request(self.upstream + self.path, data=raw,+                                     headers={'Content-Type': 'application/json'})+        try:+            with urllib.request.urlopen(req) as r:+                body = r.read()+        except urllib.error.HTTPError as e:+            body = e.read()+        out = json.dumps(json.loads(body)).encode()+        self.send_response(200)                # <-- collapses 207 to 200+        self.send_header('Content-Type', 'application/json')+        self.send_header('Content-Length', str(len(out)))+        self.end_headers()+        self.wfile.write(out)++    def log_message(self, *a):+        pass+++def serve(handler, port):+    srv = HTTPServer(('127.0.0.1', port), handler)+    Thread(target=srv.serve_forever, daemon=True).start()+    return srv+++def http(method, url, body=None, timeout=20):+    req = urllib.request.Request(url, method=method,+                                 data=body.encode() if isinstance(body, str) else body,+                                 headers={'Content-Type': 'application/json'} if body else {})+    try:+        with urllib.request.urlopen(req, timeout=timeout) as r:+            return r.status, r.read()+    except urllib.error.HTTPError as e:+        return e.code, e.read()+    except Exception as e:+        return 0, str(e).encode()+++def sha(b):+    return hashlib.sha256(b).hexdigest()+++def wait_ready(base, timeout=45):+    end = time.time() + timeout+    while time.time() < end:+        code, _ = http('GET', base + '/', timeout=3)+        if code:+            return True+        time.sleep(0.4)+    return False+++def run_checks(base, stub_port):+    rows = []++    code, body = http('GET', base + '/updater/status')+    ok = (code == 200 and body == STUB_BODY)+    rows.append(('GET /updater/status forwarded byte-for-byte', ok,+                 'status %s sha %s' % (code, sha(body)[:16]),+                 'status 200 sha %s' % sha(STUB_BODY)[:16],+                 None if ok else b'want: ' + STUB_BODY + b'\n   got: ' + body))++    sent = '{"new":true,"prompt":"proxy-fidelity probe","submit":false}'+    code, body = http('POST', base + '/ai-threads/run-prompt', sent)+    echoed = False+    try:+        echoed = json.loads(body).get('echo') == json.loads(sent)+    except Exception:+        pass+    ok2 = (code == 207 and echoed)+    rows.append(('POST /ai-threads/run-prompt: body intact, upstream 207 preserved', ok2,+                 'status %s, body echoed=%s' % (code, echoed), 'status 207, body echoed=True',+                 None if ok2 else b'got: ' + body))++    code, body = http('GET', base + '/updater/definitely-not-a-real-path')+    ok3 = (code != 418 and TEAPOT not in body)+    rows.append(('unproxied path is NOT blanket-forwarded', ok3,+                 'status %s' % code, 'anything but upstream 418',+                 None if ok3 else b'upstream teapot leaked through: ' + body))+    return rows+++def report(rows):+    print()+    worst = 0+    for what, ok, got, want, detail in rows:+        print(('  \033[32mPASS\033[0m  ' if ok else '  \033[31mRED \033[0m  ') + what)+        print('         want %s' % want)+        print('         got  %s' % got)+        if detail:+            worst = 1+            for line in detail.decode('utf-8', 'replace').splitlines():+                print('         %s' % line)+    print()+    return worst+++def main():+    ap = argparse.ArgumentParser(description='Prove a dashboard forwards the control API unchanged.')+    ap.add_argument('--base', help='base URL of an already-running dashboard, e.g. http://127.0.0.1:8785')+    ap.add_argument('--home', help='the HOME that running app reads (its .adom/hydrogen-control-url is '+                                   'briefly repointed at the stub, then restored)')+    ap.add_argument('--app-cmd', help='command to start your app; $PORT and $HOME are set for you. '+                                      'Safest: runs in a throwaway HOME.')+    ap.add_argument('--self-test', action='store_true',+                    help='negative control: run a deliberately re-serialising proxy and show it go RED')+    a = ap.parse_args()++    stub_port = free_port()+    serve(Stub, stub_port)+    stub_url = 'http://127.0.0.1:%d' % stub_port++    if a.self_test:+        print('NEGATIVE CONTROL: a proxy that parses and re-emits, which should go RED.')+        BadProxy.upstream = stub_url+        bad_port = free_port()+        serve(BadProxy, bad_port)+        rows = run_checks('http://127.0.0.1:%d' % bad_port, stub_port)+        rc = report(rows)+        if all(r[1] for r in rows):+            print('SELF-TEST FAILED: the negative control PASSED, so this harness proves nothing.')+            return 2+        print('Self-test OK: the harness detects a re-serialising proxy. A green run means something.')+        return 0++    proc = tmp_home = backup = ctrl = None+    try:+        if a.app_cmd:+            tmp_home = os.path.join('/tmp', 'proxy-fidelity-%d' % random.randint(1000, 9999))+            os.makedirs(os.path.join(tmp_home, '.adom'), exist_ok=True)+            open(os.path.join(tmp_home, '.adom', 'hydrogen-control-url'), 'w').write(stub_url)+            port = free_port()+            env = dict(os.environ, HOME=tmp_home, PORT=str(port))+            cmd = a.app_cmd.replace('$PORT', str(port)).replace('$HOME', tmp_home)+            proc = subprocess.Popen(cmd, shell=True, env=env,+                                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,+                                    preexec_fn=os.setsid)+            base = 'http://127.0.0.1:%d' % port+            if not wait_ready(base):+                print('RED: app did not answer on %s within 45s' % base)+                return 1+        elif a.base:+            base = a.base.rstrip('/')+            home = os.path.expanduser(a.home or os.environ.get('HOME', '~'))+            ctrl = os.path.join(home, '.adom', 'hydrogen-control-url')+            os.makedirs(os.path.dirname(ctrl), exist_ok=True)+            if os.path.exists(ctrl):+                backup = ctrl + '.proxy-fidelity.bak'+                shutil.copy2(ctrl, backup)+            open(ctrl, 'w').write(stub_url)+            # Relies on the published rule that the control URL is read at REQUEST+            # time. If your app caches it at startup, use --app-cmd instead.+        else:+            ap.error('give --app-cmd (recommended), --base, or --self-test')++        print('stub control API on %s' % stub_url)+        print('testing %s' % base)+        rows = run_checks(base, stub_port)+        rc = report(rows)+        if rc:+            print('One or more assertions are RED. Run --self-test once to confirm this harness')+            print('can detect a known-bad proxy, then fix the route rather than the harness.')+        return rc+    finally:+        if proc:+            try:+                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)+            except Exception:+                pass+        if backup:+            shutil.move(backup, ctrl)+        elif ctrl and os.path.exists(ctrl):+            os.remove(ctrl)+        if tmp_home:+            shutil.rmtree(tmp_home, ignore_errors=True)+++if __name__ == '__main__':+    sys.exit(main())