Adom UI Design System

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