Phone bridge: opt-in bearer-token HTTP transport for Termux

README.md

@@ -189,3 +189,7 @@ pytest
 ## License
 
 Apache-2.0.
+
+## Phone bridge (remote HTTP)
+
+Opt-in `superuser --http` serves MCP over streamable HTTP on 127.0.0.1 with a required bearer token. See [docs/PHONE_BRIDGE.md](docs/PHONE_BRIDGE.md).

docs/PHONE_BRIDGE.md

@@ -0,0 +1,38 @@
+# Phone bridge (remote MCP over HTTP)
+
+Run Super on an Android phone (Termux) and let a remote AI agent call its tools.
+stdio stays the default; the bridge is opt-in via `superuser --http`.
+
+## Security model
+- Binds to `127.0.0.1:8765` by default — nothing is exposed on the LAN.
+- **Bearer token required** on every request (missing/wrong → `401`). Generated on first
+  run at `~/.superuser/token` (mode `0600`, directory `0700`). It is never logged; the only
+  log line is `Bridge token saved to <path>`. Override with `--token-file` or `SUPERUSER_TOKEN_FILE`.
+- All tools still run through the Guard: approvals (`confirm=true` for HIGH risk),
+  denylist, AI Defense (if enabled) and the append-only audit log.
+- Rotate: delete `~/.superuser/token` and restart the bridge.
+
+## Termux
+```sh
+pkg install python git cloudflared
+pip install "git+https://github.com/fitzyracing1/Super@try/phone-bridge"
+# terminal 1: start the bridge (creates ~/.superuser/token on first run)
+AI_DEFENSE_ENABLED=false superuser --http            # --port 8765 --host 127.0.0.1
+# terminal 2: outbound tunnel (no inbound ports opened on the phone)
+cloudflared tunnel --url http://127.0.0.1:8765
+```
+cloudflared prints a `https://<random>.trycloudflare.com` URL. The MCP endpoint is
+`https://<random>.trycloudflare.com/mcp`. psutil is optional on Android; without it,
+system tools fall back to `/proc` / `shutil` and `list_processes` returns an empty list.
+
+## Client configuration
+- URL: `https://<random>.trycloudflare.com/mcp` (transport: streamable HTTP)
+- Header: `Authorization: Bearer <contents of ~/.superuser/token>`
+
+Example (JSON MCP client config):
+```json
+{"mcpServers": {"super-phone": {"type": "http",
+  "url": "https://<random>.trycloudflare.com/mcp",
+  "headers": {"Authorization": "Bearer ${SUPER_TOKEN}"}}}}
+```
+Share the token through a secret/secure field, never in chat or a committed file.

pyproject.toml

@@ -14,7 +14,8 @@ keywords = ["mcp", "agent", "sysadmin", "superuser", "cisco", "ai-defense", "gua
 
 dependencies = [
     "mcp>=1.2.0,<2",
-    "psutil>=5.9",
+    # psutil is installed everywhere except Android (Termux), where it is optional.
+    "psutil>=5.9; sys_platform != 'android' and platform_system != 'Android'",
     "PyYAML>=6.0",
     "requests>=2.31",
 ]
@@ -25,7 +26,8 @@ dependencies = [
 security = ["cisco-aidefense-sdk>=2.1.0"]
 # Stripe usage-based billing. Optional; metering degrades gracefully without it.
 billing = ["stripe>=11.0"]
-dev = ["pytest>=8.0", "ruff>=0.6"]
+dev = ["pytest>=8.0", "ruff>=0.6", "httpx>=0.27"]
+system = ["psutil>=5.9"]
 all = ["cisco-aidefense-sdk>=2.1.0", "stripe>=11.0"]
 
 [project.scripts]

src/superuser/__main__.py

@@ -20,6 +20,14 @@ def main(argv: list[str] | None = None) -> int:
         "--transport", default="stdio", choices=["stdio", "sse", "streamable-http"],
         help="MCP transport (default: stdio).",
     )
+    parser.add_argument(
+        "--http", action="store_true",
+        help="Opt-in remote bridge: MCP over streamable HTTP with bearer-token auth.",
+    )
+    parser.add_argument("--host", default=None, help="Bridge bind host (default 127.0.0.1).")
+    parser.add_argument("--port", type=int, default=None, help="Bridge port (default 8765).")
+    parser.add_argument("--token-file", default=None,
+                        help="Bridge token path (default ~/.superuser/token).")
     parser.add_argument("--log-level", default="INFO")
     parser.add_argument(
         "--check", action="store_true",
@@ -54,6 +62,19 @@ def main(argv: list[str] | None = None) -> int:
         print(f"  GitHub configured:    {bool(config.github.token)}")
         return 0
 
+    if args.http:
+        from pathlib import Path
+
+        from .bridge import DEFAULT_HOST, DEFAULT_PORT, run_bridge
+
+        run_bridge(
+            server,
+            host=args.host or DEFAULT_HOST,
+            port=args.port or DEFAULT_PORT,
+            token_path=Path(args.token_file).expanduser() if args.token_file else None,
+        )
+        return 0
+
     server.run(transport=args.transport)
     return 0
 

src/superuser/bridge.py

@@ -0,0 +1,127 @@
+"""Opt-in remote "phone bridge": MCP over streamable HTTP with bearer-token auth.
+
+Binds to 127.0.0.1 by default. Expose it off-device only through an outbound
+tunnel (e.g. ``cloudflared tunnel --url http://127.0.0.1:8765``). Every request
+must carry ``Authorization: Bearer <token>``; the token is generated on first
+run and stored at ``~/.superuser/token`` (mode 0600). The token is never logged.
+All tools still go through the same Guard (approvals, guardrails, audit log).
+"""
+
+from __future__ import annotations
+
+import hmac
+import logging
+import os
+import secrets
+from pathlib import Path
+from typing import Any, Awaitable, Callable
+
+logger = logging.getLogger("superuser.bridge")
+
+DEFAULT_HOST = "127.0.0.1"
+DEFAULT_PORT = 8765
+
+
+def default_token_path() -> Path:
+    env = os.environ.get("SUPERUSER_TOKEN_FILE")
+    return Path(env).expanduser() if env else Path.home() / ".superuser" / "token"
+
+
+def load_or_create_token(path: Path | None = None) -> str:
+    """Return the bridge token, creating it (0600) on first use."""
+    path = path or default_token_path()
+    if path.exists():
+        token = path.read_text(encoding="utf-8").strip()
+        if token:
+            try:
+                os.chmod(path, 0o600)
+            except OSError:
+                pass
+            return token
+    path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        os.chmod(path.parent, 0o700)
+    except OSError:
+        pass
+    token = secrets.token_urlsafe(32)
+    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(fd, "w", encoding="utf-8") as fh:
+        fh.write(token + "\n")
+    os.chmod(path, 0o600)
+    logger.warning("Bridge token saved to %s", path)
+    return token
+
+
+ASGIApp = Callable[[dict, Callable, Callable], Awaitable[None]]
+
+
+class BearerAuthMiddleware:
+    """Pure-ASGI middleware rejecting HTTP requests without the right bearer token."""
+
+    def __init__(self, app: ASGIApp, token: str) -> None:
+        if not token:
+            raise ValueError("bridge token must be non-empty")
+        self.app = app
+        self._token = token.encode()
+
+    def _authorized(self, scope: dict) -> bool:
+        for name, value in scope.get("headers") or []:
+            if name.lower() == b"authorization":
+                scheme, _, cred = value.partition(b" ")
+                if scheme.lower() == b"bearer" and hmac.compare_digest(cred.strip(), self._token):
+                    return True
+        return False
+
+    async def __call__(self, scope: dict, receive: Callable, send: Callable) -> None:
+        if scope["type"] == "lifespan":
+            await self.app(scope, receive, send)
+            return
+        if scope["type"] in ("http", "websocket") and not self._authorized(scope):
+            client = scope.get("client") or ("?", 0)
+            logger.warning("Rejected unauthenticated bridge request from %s", client[0])
+            if scope["type"] == "websocket":
+                await send({"type": "websocket.close", "code": 1008})
+                return
+            body = b'{"error":"unauthorized"}'
+            await send({
+                "type": "http.response.start", "status": 401,
+                "headers": [(b"content-type", b"application/json"),
+                            (b"www-authenticate", b'Bearer realm="superuser"'),
+                            (b"content-length", str(len(body)).encode())],
+            })
+            await send({"type": "http.response.body", "body": body})
+            return
+        await self.app(scope, receive, send)
+
+
+def build_bridge_app(server: Any, token: str) -> ASGIApp:
+    """Wrap a FastMCP server's streamable-HTTP app with bearer auth.
+
+    The MCP SDK's Host-header (DNS-rebinding) check is relaxed here because the
+    tunnel forwards a public hostname; the mandatory bearer token is what
+    protects the endpoint (a rebinding page cannot know the token).
+    """
+    settings = getattr(server, "settings", None)
+    if settings is not None and hasattr(settings, "transport_security"):
+        try:
+            from mcp.server.transport_security import TransportSecuritySettings
+
+            settings.transport_security = TransportSecuritySettings(
+                enable_dns_rebinding_protection=False
+            )
+        except ImportError:  # older mcp without the check
+            pass
+    return BearerAuthMiddleware(server.streamable_http_app(), token)
+
+
+def run_bridge(server: Any, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT,
+               token_path: Path | None = None) -> None:
+    import uvicorn
+
+    token = load_or_create_token(token_path)
+    if host not in ("127.0.0.1", "localhost", "::1"):
+        logger.warning("Bridge bound to %s (not loopback). Prefer 127.0.0.1 + a tunnel.", host)
+    app = build_bridge_app(server, token)
+    logger.info("Superuser bridge listening on http://%s:%d/mcp (bearer auth required)",
+                host, port)
+    uvicorn.run(app, host=host, port=port, log_level="info", access_log=False)

src/superuser/tools/system.py

@@ -8,10 +8,45 @@
 from datetime import datetime, timezone
 from typing import Any
 
-import psutil
+try:  # psutil is optional (it may not build/work on Android/Termux)
+    import psutil
+except Exception:  # pragma: no cover - platform dependent
+    psutil = None  # type: ignore[assignment]
+
+import os
+import shutil
+
+
+def _fallback_status() -> dict[str, Any]:
+    mem: dict[str, Any] = {}
+    try:
+        info = {}
+        with open("/proc/meminfo") as fh:
+            for line in fh:
+                k, _, v = line.partition(":")
+                info[k] = int(v.split()[0])
+        total, avail = info.get("MemTotal", 0), info.get("MemAvailable", 0)
+        mem = {"total_mb": round(total / 1024, 1), "available_mb": round(avail / 1024, 1),
+               "percent_used": round(100 * (1 - avail / total), 1) if total else None}
+    except Exception:
+        pass
+    try:
+        load_avg = os.getloadavg()
+    except (AttributeError, OSError):
+        load_avg = None
+    return {"cpu_count": os.cpu_count(), "cpu_percent": None, "load_average": load_avg,
+            "memory": mem, "boot_time_utc": None, "psutil": False}
 
 
 def system_status() -> dict[str, Any]:
+    if psutil is None:
+        try:
+            user = getpass.getuser()
+        except Exception:
+            user = None
+        return {"hostname": socket.gethostname(), "user": user,
+                "platform": platform.platform(), "python": platform.python_version(),
+                **_fallback_status()}
     vm = psutil.virtual_memory()
     boot = datetime.fromtimestamp(psutil.boot_time(), tz=timezone.utc)
     load_avg: tuple[float, float, float] | None
@@ -38,6 +73,18 @@ def system_status() -> dict[str, Any]:
 
 
 def disk_usage() -> dict[str, Any]:
+    if psutil is None:
+        out = []
+        for mp in ("/", os.path.expanduser("~")):
+            try:
+                u = shutil.disk_usage(mp)
+            except OSError:
+                continue
+            out.append({"device": None, "mountpoint": mp, "fstype": None,
+                        "total_gb": round(u.total / 1_073_741_824, 2),
+                        "used_gb": round(u.used / 1_073_741_824, 2),
+                        "percent_used": round(100 * u.used / u.total, 1) if u.total else None})
+        return {"partitions": out, "psutil": False}
     partitions = []
     for part in psutil.disk_partitions(all=False):
         try:
@@ -57,6 +104,9 @@ def disk_usage() -> dict[str, Any]:
 
 def list_processes(limit: int = 20, sort_by: str = "memory") -> dict[str, Any]:
     limit = max(1, min(limit, 200))
+    if psutil is None:
+        return {"count": 0, "sorted_by": sort_by, "processes": [], "psutil": False,
+                "note": "psutil unavailable on this platform (e.g. Android); install it to enable."}
     procs = []
     for proc in psutil.process_iter(["pid", "name", "username", "cpu_percent", "memory_percent"]):
         try:

tests/test_bridge.py

@@ -0,0 +1,61 @@
+import os
+import stat
+
+import pytest
+from starlette.testclient import TestClient
+
+from superuser.bridge import BearerAuthMiddleware, build_bridge_app, load_or_create_token
+
+TOKEN = "test-token-abc123"
+
+
+async def _ok_app(scope, receive, send):
+    await send({"type": "http.response.start", "status": 200, "headers": []})
+    await send({"type": "http.response.body", "body": b"ok"})
+
+
+@pytest.fixture
+def client():
+    return TestClient(BearerAuthMiddleware(_ok_app, TOKEN))
+
+
+def test_missing_token_rejected(client):
+    r = client.get("/mcp")
+    assert r.status_code == 401
+    assert TOKEN not in r.text
+
+
+def test_wrong_token_rejected(client):
+    assert client.get("/mcp", headers={"Authorization": "Bearer nope"}).status_code == 401
+    assert client.get("/mcp", headers={"Authorization": f"Basic {TOKEN}"}).status_code == 401
+
+
+def test_right_token_accepted(client):
+    r = client.get("/mcp", headers={"Authorization": f"Bearer {TOKEN}"})
+    assert r.status_code == 200 and r.text == "ok"
+
+
+def test_token_file_created_0600_and_stable(tmp_path):
+    p = tmp_path / "sub" / "token"
+    t1 = load_or_create_token(p)
+    assert len(t1) >= 32
+    if os.name == "posix":
+        assert stat.S_IMODE(p.stat().st_mode) == 0o600
+    assert load_or_create_token(p) == t1
+
+
+def test_real_mcp_app_requires_token(tmp_path, monkeypatch):
+    monkeypatch.setenv("AI_DEFENSE_ENABLED", "false")
+    monkeypatch.chdir(tmp_path)
+    from superuser.config import Config
+    from superuser.server import build_server
+
+    app = build_bridge_app(build_server(Config.load(None)), TOKEN)
+    init = {"jsonrpc": "2.0", "id": 1, "method": "initialize",
+            "params": {"protocolVersion": "2025-03-26", "capabilities": {},
+                       "clientInfo": {"name": "t", "version": "0"}}}
+    hdrs = {"Accept": "application/json, text/event-stream"}
+    with TestClient(app) as c:
+        assert c.post("/mcp", json=init, headers=hdrs).status_code == 401
+        r = c.post("/mcp", json=init, headers={**hdrs, "Authorization": f"Bearer {TOKEN}"})
+        assert r.status_code == 200