mrambossekandClaude Opus 5
52748ac853
The phone's wireless-debug port rotates and the bridge drops; this makes
adb reconnect automatically. Three pieces:
- adb-beacon (Android dev app): reads adbd's own mDNS advertisement
(_adb-tls-connect._tcp) via NsdManager for the live connect port — no
root, no Shizuku — plus the wlan0 IPv4, and POSTs {ip,port} to fmr every
time it changes (continuous NSD discovery catches rotation in seconds).
Foreground service (specialUse) so it survives backgrounding.
- tools/adb-beacon/receiver.py: ~30-line rendezvous on fmr:9099 (secret-
gated POST stores the latest endpoint; GET returns it). Deployed as
echolot-adb-beacon.service.
- PC connector polls the endpoint and keeps `adb connect` current.
Dev tooling, separate from the product. Bootstrap: sideload the beacon
APK once (no adb needed); thereafter adb self-heals for everything.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-FileCopyrightText: 2026 Echolot contributors
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
#
|
|
# Tiny rendezvous receiver for the wireless-adb beacon. The phone POSTs its
|
|
# current wireless-debug endpoint here; the PC-side connector reads the stored
|
|
# file (over SSH) and runs `adb connect`. Dev tooling — not part of the product.
|
|
#
|
|
# POST /beacon { "ip": "...", "port": N } header: X-Beacon-Secret: <secret>
|
|
# GET /beacon -> the last stored JSON (no secret; it's just an ip:port)
|
|
#
|
|
# Secret comes from ECHOLOT_BEACON_SECRET; stored state goes to STATE_PATH.
|
|
# Bind is 0.0.0.0:9099 so the phone can reach it over the internet/VPN.
|
|
|
|
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"))
|
|
|
|
|
|
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):
|
|
if self.path.rstrip("/") != "/beacon":
|
|
return self._send(404, b'{"error":"not found"}')
|
|
try:
|
|
with open(STATE_PATH, "rb") as f:
|
|
self._send(200, f.read())
|
|
except FileNotFoundError:
|
|
self._send(200, b'{"available":false}')
|
|
|
|
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"{}")
|
|
ip = str(data["ip"])
|
|
port = int(data["port"])
|
|
except Exception:
|
|
return self._send(400, b'{"error":"need {ip, port}"}')
|
|
record = {"ip": ip, "port": port, "seen_at": int(time.time())}
|
|
tmp = STATE_PATH + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(record, f)
|
|
os.replace(tmp, STATE_PATH)
|
|
self._send(200, json.dumps(record).encode())
|
|
|
|
def log_message(self, *_):
|
|
pass # quiet
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"adb-beacon receiver on {BIND}:{PORT} -> {STATE_PATH}", flush=True)
|
|
ThreadingHTTPServer((BIND, PORT), Handler).serve_forever()
|