echolot/tools/adb-beacon/receiver.py

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 Echolot contributors
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Rendezvous receiver for the wireless-adb beacon (dev tool). Phones/tablets
# POST their current wireless-debug endpoint here, keyed by device; the PC-side
# connector reads them all and keeps `adb connect` current for each.
#
#   POST /beacon  { "device": "...", "ip": "...", "port": N }
#                 header: X-Beacon-Secret: <secret>
#   GET  /beacon  -> { "<device>": {ip, port, seen_at}, ... }
#
# ECHOLOT_BEACON_SECRET gates POST; STATE_PATH holds the device map.

import json
import os
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

STATE_PATH = os.environ.get("STATE_PATH", "/run/echolot-adb-beacon.json")
REPORT_DIR = os.environ.get("REPORT_DIR", "/tmp/echolot-reports")
SECRET = os.environ.get("ECHOLOT_BEACON_SECRET", "")
BIND = os.environ.get("BEACON_BIND", "0.0.0.0")
PORT = int(os.environ.get("BEACON_PORT", "9099"))


def load():
    try:
        with open(STATE_PATH) as f:
            d = json.load(f)
            return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def store(d):
    tmp = STATE_PATH + ".tmp"
    with open(tmp, "w") as f:
        json.dump(d, f)
    os.replace(tmp, STATE_PATH)


class Handler(BaseHTTPRequestHandler):
    def _send(self, code, body=b"", ctype="application/json"):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        if body:
            self.wfile.write(body)

    def do_GET(self):
        p = self.path.rstrip("/")
        if p == "/beacon":
            return self._send(200, json.dumps(load()).encode())
        # Dev convenience: serve a staged APK so a device (which can only reach
        # this host on 443) can pull it via its own curl — more resilient than
        # adb's sustained transport over a flaky wifi. Path from APK_PATH.
        if p == "/reports":
            try:
                names = sorted(os.listdir(REPORT_DIR), reverse=True)
            except FileNotFoundError:
                names = []
            return self._send(200, json.dumps(names).encode())
        if p.startswith("/report/"):
            name = os.path.basename(p[len("/report/"):])
            try:
                with open(os.path.join(REPORT_DIR, name), "rb") as f:
                    return self._send(200, f.read())
            except FileNotFoundError:
                return self._send(404, b'{"error":"no such report"}')
        if p == "/apk":
            apk = os.environ.get("APK_PATH", "/tmp/echolot-app.apk")
            try:
                with open(apk, "rb") as f:
                    return self._send(200, f.read(), "application/vnd.android.package-archive")
            except FileNotFoundError:
                return self._send(404, b'{"error":"no apk staged"}')
        return self._send(404, b'{"error":"not found"}')

    def do_POST(self):
        p = self.path.rstrip("/")
        # Measurement-report drop box: the app's autorun mode POSTs its run JSON here so a
        # test run needs no adb at all (start it once, collect the result over HTTP).
        if p == "/report":
            if SECRET and self.headers.get("X-Beacon-Secret") != SECRET:
                return self._send(403, b'{"error":"bad secret"}')
            n = int(self.headers.get("Content-Length", "0"))
            body = self.rfile.read(n)
            try:
                doc = json.loads(body)
                dev = doc.get("run", {}).get("device", {}).get("model", "device")
                rid = doc.get("run", {}).get("id", "unknown")[:8]
            except Exception:
                dev, rid = "device", "unparsed"
            os.makedirs(REPORT_DIR, exist_ok=True)
            name = f"{int(time.time())}-{dev}-{rid}.json"
            with open(os.path.join(REPORT_DIR, name), "wb") as f:
                f.write(body)
            return self._send(200, json.dumps({"stored": name, "bytes": len(body)}).encode())

        if p != "/beacon":
            return self._send(404, b'{"error":"not found"}')
        if SECRET and self.headers.get("X-Beacon-Secret") != SECRET:
            return self._send(403, b'{"error":"bad secret"}')
        n = int(self.headers.get("Content-Length", "0"))
        try:
            data = json.loads(self.rfile.read(n) or b"{}")
            device = str(data.get("device", "default"))
            ip = str(data["ip"])
            port = int(data["port"])
        except Exception:
            return self._send(400, b'{"error":"need {device, ip, port}"}')
        d = load()
        d[device] = {"ip": ip, "port": port, "seen_at": int(time.time())}
        store(d)
        self._send(200, json.dumps(d[device]).encode())

    def log_message(self, *_):
        pass


if __name__ == "__main__":
    print(f"adb-beacon receiver on {BIND}:{PORT} -> {STATE_PATH}", flush=True)
    ThreadingHTTPServer((BIND, PORT), Handler).serve_forever()