linux-on-android-without-root/spike/bridge-mini.py

#!/usr/bin/env python3
"""HPOS Mini-Bridge — Prototyp des Intent-Handoffs für Beweis S4.

Läuft in Termux, lauscht auf 127.0.0.1 und übersetzt URIs vom Linux in
Android-Intents. Das ist die kleinste ehrliche Fassung dessen, was später die
Bridge-APK macht: sie speichert nichts, sie reicht durch.

Bewusst eng gehalten:
  * nur 127.0.0.1, niemals extern erreichbar
  * gemeinsames Token, damit nicht jede App auf dem Telefon Intents auslösen kann
  * feste Liste erlaubter Schemata statt beliebiger Intents

Start:  python spike/bridge-mini.py
"""

from __future__ import annotations

import os
import shlex
import socketserver
import subprocess
import sys
import time
import urllib.parse

HOST = "127.0.0.1"
PORT = int(os.environ.get("HPOS_BRIDGE_PORT", "8474"))

BASIS = os.path.dirname(os.path.abspath(__file__))
TOKEN_DATEI = os.path.join(BASIS, "out", "bridge-token")

# Paketname einer Begleit-App, die vor der Hauptanwendung starten soll — typisch
# eine App mit Overlay, das über der Navigation liegen muss. Leer = keine.
BEGLEIT_APP = os.environ.get("HPOS_BEGLEIT_APP", "").strip()
BEGLEIT_VORLAUF = float(os.environ.get("HPOS_BEGLEIT_VORLAUF", "2.5"))

# Was aus dem Linux kommen darf. Alles andere wird abgelehnt — eine offene
# Intent-Schnittstelle wäre ein Einfallstor für jede App auf dem Gerät.
ERLAUBTE_SCHEMATA = {"geo", "tel", "sms", "smsto", "mailto", "http", "https"}


def token_lesen() -> str:
    """Token holen, beim ersten Start erzeugen."""
    os.makedirs(os.path.dirname(TOKEN_DATEI), exist_ok=True)
    if not os.path.exists(TOKEN_DATEI):
        with open(os.open(TOKEN_DATEI, os.O_CREAT | os.O_WRONLY, 0o600), "w") as f:
            f.write(os.urandom(16).hex())
    with open(TOKEN_DATEI) as f:
        return f.read().strip()


TOKEN = token_lesen()


def am(*args: str) -> tuple[int, str]:
    """`am` aufrufen und Ergebnis zurückgeben."""
    befehl = ["am", *args]
    print(f"  → {' '.join(shlex.quote(a) for a in befehl)}", flush=True)
    try:
        p = subprocess.run(befehl, capture_output=True, text=True, timeout=20)
        return p.returncode, (p.stdout + p.stderr).strip()
    except FileNotFoundError:
        return 127, "am nicht gefunden — Paket termux-tools fehlt"
    except subprocess.TimeoutExpired:
        return 124, "am hat nicht geantwortet"


def app_starten(paket: str) -> None:
    am("start", "--user", "0", "-a", "android.intent.action.MAIN",
       "-c", "android.intent.category.LAUNCHER", "-p", paket)


def navigation(ziel: str) -> tuple[int, str]:
    """Turn-by-turn direkt starten statt nur die Karte zu zeigen.

    Vorher optional die Begleit-App, damit deren Overlay schon steht, wenn die
    Navigation in den Vordergrund kommt — andersherum legt sich die Navigation
    darüber und das Overlay ist verdeckt.
    """
    if BEGLEIT_APP:
        print(f"  Begleit-App vorab: {BEGLEIT_APP}", flush=True)
        app_starten(BEGLEIT_APP)
        time.sleep(BEGLEIT_VORLAUF)
    uri = "google.navigation:q=" + urllib.parse.quote(ziel) + "&mode=d"
    return am("start", "--user", "0", "-a", "android.intent.action.VIEW", "-d", uri)


def behandeln(zeile: str) -> str:
    """Eine Anfrage verarbeiten. Format: <token>\\t<uri>"""
    teile = zeile.split("\t", 1)
    if len(teile) != 2:
        return "FEHLER ungueltiges-format"
    token, uri = teile[0].strip(), teile[1].strip()

    if token != TOKEN:
        print("  ! Anfrage mit falschem Token abgelehnt", flush=True)
        return "FEHLER token"
    if not uri:
        return "FEHLER leere-uri"

    schema = uri.split(":", 1)[0].lower()

    # geo:-URIs mit Suchbegriff bedeuten in der Praxis fast immer "dahin navigieren",
    # nicht "Karte anschauen". Deshalb wird daraus Turn-by-turn.
    if schema == "geo":
        query = urllib.parse.parse_qs(urllib.parse.urlparse(uri).query).get("q", [""])[0]
        if query:
            rc, aus = navigation(query)
            return f"OK navigation {rc} {aus}" if rc == 0 else f"FEHLER am {rc} {aus}"

    if schema == "tel":
        # DIAL statt CALL: die Nummer erscheint im Wähler, gewählt wird bewusst.
        rc, aus = am("start", "--user", "0", "-a", "android.intent.action.DIAL", "-d", uri)
        return f"OK dial {rc} {aus}" if rc == 0 else f"FEHLER am {rc} {aus}"

    if schema in ERLAUBTE_SCHEMATA:
        rc, aus = am("start", "--user", "0", "-a", "android.intent.action.VIEW", "-d", uri)
        return f"OK view {rc} {aus}" if rc == 0 else f"FEHLER am {rc} {aus}"

    print(f"  ! Schema abgelehnt: {schema}", flush=True)
    return f"FEHLER schema-nicht-erlaubt {schema}"


class Handler(socketserver.StreamRequestHandler):
    timeout = 10

    def handle(self) -> None:
        try:
            zeile = self.rfile.readline(4096).decode("utf-8", "replace")
        except Exception as e:
            print(f"  ! Lesefehler: {e}", flush=True)
            return
        if not zeile:
            return
        antwort = behandeln(zeile)
        print(f"  ← {antwort}", flush=True)
        try:
            self.wfile.write((antwort + "\n").encode())
        except Exception:
            pass


class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True
    daemon_threads = True


def main() -> int:
    print("HPOS Mini-Bridge (Prototyp für Beweis S4)")
    print(f"  Lauscht auf   {HOST}:{PORT}")
    print(f"  Token-Datei   {TOKEN_DATEI}")
    print(f"  Begleit-App   {BEGLEIT_APP or '(keine)'}")
    print("  Beenden mit Strg+C")
    print()
    try:
        with Server((HOST, PORT), Handler) as srv:
            srv.serve_forever()
    except KeyboardInterrupt:
        print("\nBeendet.")
    except OSError as e:
        print(f"Start fehlgeschlagen: {e}", file=sys.stderr)
        print(f"Läuft schon etwas auf Port {PORT}?", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())