mrambossekandClaude Opus 5
ee4031b086
A test device that can only reach fmr on 443 (LAN blocks other outbound ports) can pull an APK via its own downloader — more resilient than adb's sustained transport over flaky wifi. GET /apk serves APK_PATH. (Doesn't help the Lenovo tablet, which ships no curl; kept for devices that do.) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
#!/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")
|
|
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 == "/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):
|
|
if self.path.rstrip("/") != "/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()
|