harley/hermes-relay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dev
hermes-relay/plugin/pair.py
T

2241 lines
83 KiB
Python
Raw Permalink Blame History

"""Hermes Android pairing — generate QR codes and connection info.
Replaces the standalone bash script `skills/hermes-pairing-qr/hermes-pair`.
Exposed as the `hermes pair` CLI sub-command via plugin/cli.py.
The payload format matches what QrPairingScanner.kt in the Android app expects.
v2 added optional TTL / per-channel grants / transport hint / HMAC signature.
v3 (ADR 24) adds an optional ordered ``endpoints`` array for multi-network
operators (LAN + Tailscale + public reverse-proxy in a single QR). v1/v2
remain valid — phones synthesize a priority-0 candidate from the top-level
fields when ``endpoints`` is absent, and the pair CLI only bumps the
version bit when it's actually emitting endpoints::
v3 (multi-endpoint — ADR 24):
{
"hermes": 3,
"host": "<ip>", "port": <port>, "key": "<token>", "tls": <bool>,
"dashboard_url": "https://dashboard.example.com",
"relay": { ... same as v2 ... },
"endpoints": [
{ "role": "lan", "priority": 0,
"api": {"host": "<ip>", "port": <port>, "tls": <bool>},
"relay": {"url": "ws://<ip>:<port>", "transport_hint": "ws"} },
{ "role": "tailscale", "priority": 1, "api": {...}, "relay": {...} },
{ "role": "public", "priority": 2, "api": {...}, "relay": {...} }
],
"sig": "<base64-hmac-sha256>"
}
v2 (extended):
{
"hermes": 2,
"host": "<ip>", "port": <port>, "key": "<token>", "tls": <bool>,
"relay": {
"url": "ws://<ip>:<port>",
"code": "<6-char>",
"ttl_seconds": 2592000, // 0 = never expire
"grants": {"terminal": ..., "bridge": ...},
"transport_hint": "ws" // "wss" or "ws"
},
"sig": "<base64-hmac-sha256>"
}
v1 (legacy fallback):
{
"hermes": 1,
"host": "<ip>", "port": <port>, "key": "<token>", "tls": <bool>,
"relay": { "url": "ws://<ip>:<port>", "code": "<6-char>" }
}
Top-level fields configure the optional direct-chat Hermes API server (port
8642 by default) when it is enabled. New Dashboard-first payloads may omit
them and advertise independently optional Dashboard, Relay, and API surfaces
inside ``endpoints``.
The optional ``relay`` block configures the Hermes-Relay WSS connection
used by the terminal and bridge channels. The normal CLI flow obtains the
fully signed invite from the local Relay via ``POST /pairing/mint`` so
server-owned Secure Link and Reach candidates are included. Older Relays may
use ``POST /pairing/register`` only when Reach is not configured.
"""
from __future__ import annotations
import io
import ipaddress
import json
import math
import os
import random
import socket
import string
import sys
import tempfile
import urllib.error
import urllib.request
import base64
from pathlib import Path
from typing import Any, Optional
from urllib.parse import urlparse
from .relay.qr_sign import load_or_create_secret, sign_payload
def _get_hermes_home() -> Path:
return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
def _parse_env_file(path: Path) -> dict[str, str]:
"""Parse a simple KEY=value env file. Ignores comments and blank lines."""
result: dict[str, str] = {}
if not path.exists():
return result
try:
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
# Strip matching surrounding quotes
value = value.strip()
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
value = value[1:-1]
result[key.strip()] = value
except OSError:
pass
return result
def read_server_config() -> dict:
"""Read API server config with fallback chain.
Priority: hermes config.yaml → ~/.hermes/.env → environment vars → defaults.
Returns dict with keys: host, port, key, tls, enabled. ``enabled`` follows
Hermes' API-server rule: an explicit boolean wins; otherwise a configured
API key enables the compatibility surface.
"""
host: Optional[str] = None
port: Optional[int] = None
key: Optional[str] = None
enabled: Optional[bool] = None
tls = False
# 1. Try hermes_cli.config.load_config()
try:
from hermes_cli.config import load_config # type: ignore
config = load_config()
api = config.get("platforms", {}).get("api_server", {}) or {}
extra = api.get("extra", {}) or {}
enabled = _parse_optional_bool(api.get("enabled"))
key = extra.get("key") or api.get("api_key")
port_val = extra.get("port") or api.get("port")
if port_val is not None:
port = int(port_val)
host = extra.get("host") or api.get("host")
except Exception:
pass
# 2. Fall back to ~/.hermes/.env
if key is None or host is None or port is None:
env_vals = _parse_env_file(_get_hermes_home() / ".env")
if key is None:
key = env_vals.get("API_SERVER_KEY")
if host is None:
host = env_vals.get("API_SERVER_HOST")
if port is None and env_vals.get("API_SERVER_PORT"):
try:
port = int(env_vals["API_SERVER_PORT"])
except ValueError:
pass
if enabled is None:
enabled = _parse_optional_bool(env_vals.get("API_SERVER_ENABLED"))
# 3. Fall back to process environment
if key is None:
key = os.getenv("API_SERVER_KEY", "")
if enabled is None:
enabled = _parse_optional_bool(os.getenv("API_SERVER_ENABLED"))
if host is None:
host = os.getenv("API_SERVER_HOST", "127.0.0.1")
if port is None:
try:
port = int(os.getenv("API_SERVER_PORT", "8642"))
except ValueError:
port = 8642
if enabled is None:
enabled = bool(key)
return {
"host": host,
"port": port,
"key": key or "",
"tls": tls,
"enabled": enabled,
}
def _parse_optional_bool(value: Any) -> Optional[bool]:
if isinstance(value, bool):
return value
if value is None:
return None
normalized = str(value).strip().lower()
if normalized in ("1", "true", "yes", "on"):
return True
if normalized in ("0", "false", "no", "off"):
return False
return None
def _resolve_lan_ip(host: str) -> str:
"""Auto-detect LAN IP when host is loopback or bind-all.
Cross-platform: uses a UDP socket connect() trick to find the outbound
interface IP without sending any packets.
"""
if host not in ("0.0.0.0", "127.0.0.1", "localhost", "::", "::1"):
return host
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(2)
# Doesn't actually send; just picks the outbound interface
s.connect(("1.1.1.1", 80))
ip = s.getsockname()[0]
s.close()
return ip
except OSError:
return host
def _relay_has_v2_fields(relay: Optional[dict]) -> bool:
"""True if the relay block uses any v2-only field."""
if relay is None:
return False
return any(
k in relay for k in ("ttl_seconds", "grants", "transport_hint")
)
def build_payload(
host: Optional[str],
port: Optional[int],
key: Optional[str],
tls: Optional[bool],
relay: Optional[dict] = None,
sign: bool = True,
endpoints: Optional[list[dict]] = None,
dashboard_url: Optional[str] = None,
legacy_direct_relay: bool = False,
) -> str:
"""Build compact JSON payload matching HermesPairingPayload.kt format.
If ``relay`` is provided it's embedded as a nested ``"relay"`` object.
When any v2-only field (``ttl_seconds``, ``grants``, ``transport_hint``)
is present the top-level ``hermes`` field is bumped to ``2``; otherwise
it stays at ``1`` so old phones can still parse.
If ``endpoints`` is provided and non-empty it's embedded as a top-level
``"endpoints"`` array and the version bumps to ``3`` (ADR 24 —
multi-endpoint pairing payload). List order is preserved verbatim
through canonicalization + signing so strict priority semantics
survive the HMAC. Callers are responsible for sorting by
``priority`` before passing the list in; ``build_endpoint_candidates``
does this for the CLI path.
Configured top-level ``host`` / ``port`` / ``key`` / ``tls`` fields are
retained for API compatibility. API-less Dashboard-first payloads omit
them; direct Relay remains top-level for v1/v2 clients.
``dashboard_url`` is optional. When present, Android stores it as the
Manage/dashboard URL instead of deriving the conventional same-host
``:9119`` URL from the API server.
If ``sign`` is true the payload is signed with the host-local QR
secret and the base64 HMAC is added as a top-level ``sig`` field.
"""
for candidate in endpoints or []:
if isinstance(candidate, dict):
validate_endpoint_candidate_security(candidate)
normalized_dashboard_url = (
normalize_dashboard_url(dashboard_url) if dashboard_url else None
)
endpoints = add_dashboard_ingress_candidate(
endpoints=endpoints,
dashboard_url=normalized_dashboard_url,
relay=relay,
api_host=host,
api_port=port,
api_tls=tls,
legacy_direct_relay=legacy_direct_relay,
)
if endpoints:
version = 3
elif _relay_has_v2_fields(relay):
version = 2
else:
version = 1
payload: dict = {"hermes": version}
if host is not None and str(host).strip():
payload.update(
{
"host": str(host).strip(),
"port": int(port) if port is not None else 8642,
"key": key or "",
"tls": bool(tls),
}
)
if relay is not None:
advertised_relay = dict(relay)
if not legacy_direct_relay:
ingress_relay: Optional[dict] = None
if normalized_dashboard_url:
expected_dashboard_key = _dashboard_origin_key(normalized_dashboard_url)
expected_relay_key = _relay_ingress_key(
dashboard_relay_ingress_url(normalized_dashboard_url)
)
matching_ingresses: list[dict] = []
for candidate in endpoints or []:
if not isinstance(candidate, dict):
continue
candidate_dashboard = candidate.get("dashboard")
candidate_relay = candidate.get("relay")
if not isinstance(candidate_dashboard, dict) or not isinstance(candidate_relay, dict):
continue
try:
dashboard_matches = _dashboard_origin_key(
str(candidate_dashboard.get("url") or "")
) == expected_dashboard_key
relay_matches = _relay_ingress_key(
str(candidate_relay.get("url") or "")
) == expected_relay_key
except ValueError:
continue
if dashboard_matches and relay_matches:
matching_ingresses.append(candidate_relay)
if len(matching_ingresses) != 1:
raise ValueError(
"dashboard_url must have exactly one exact same-origin "
"Relay ingress candidate"
)
ingress_relay = matching_ingresses[0]
else:
ingress_relay = next(
(
candidate.get("relay")
for candidate in (endpoints or [])
if isinstance(candidate, dict)
and isinstance(candidate.get("dashboard"), dict)
and isinstance(candidate.get("relay"), dict)
),
None,
)
if isinstance(ingress_relay, dict) and ingress_relay.get("url"):
advertised_relay["url"] = ingress_relay["url"]
if ingress_relay.get("transport_hint"):
advertised_relay["transport_hint"] = ingress_relay["transport_hint"]
payload["relay"] = advertised_relay
if endpoints:
payload["endpoints"] = endpoints
if normalized_dashboard_url:
payload["dashboard_url"] = normalized_dashboard_url
if sign:
try:
secret = load_or_create_secret()
payload["sig"] = sign_payload(payload, secret)
except Exception as exc:
# Signing is best-effort: if we can't read/write the secret
# file we still want to emit a usable QR rather than crashing
# the `hermes pair` flow. Log to stderr so the operator sees
# it but doesn't lose the QR.
print(
f" [warn] QR signing failed ({exc}) — payload will be unsigned.",
file=sys.stderr,
)
return json.dumps(payload, separators=(",", ":"))
def endpoint_role_for_url(url: str) -> str:
"""Return the existing route role that best describes an origin."""
parsed = urlparse(url)
host = (parsed.hostname or "").lower()
if host.endswith(".ts.net"):
return "tailscale"
try:
address = ipaddress.ip_address(host)
except ValueError:
address = None
if address is not None:
tailnet_v4 = ipaddress.ip_network("100.64.0.0/10")
tailnet_v6 = ipaddress.ip_network("fd7a:115c:a1e0::/48")
if address in tailnet_v4 or address in tailnet_v6:
return "tailscale"
if address.is_private or address.is_loopback or address.is_link_local:
return "lan"
if (
host in {"localhost", "localhost.localdomain"}
or host.endswith(".local")
or (host and "." not in host)
):
return "lan"
return "public"
def normalize_dashboard_url(dashboard_url: str) -> str:
"""Return a credential-free canonical Dashboard HTTP(S) base URL."""
trimmed = dashboard_url.strip().rstrip("/")
if any(character.isspace() for character in trimmed):
raise ValueError("dashboard_url must not include whitespace")
parsed = urlparse(trimmed)
if parsed.scheme.lower() not in ("http", "https") or not parsed.netloc:
raise ValueError("dashboard_url must be an absolute http(s) URL")
if parsed.username is not None or parsed.password is not None:
raise ValueError("dashboard_url must not include credentials")
if parsed.query or parsed.fragment:
raise ValueError("dashboard_url must not include a query or fragment")
try:
parsed.port
except ValueError as exc:
raise ValueError("dashboard_url contains an invalid port") from exc
normalized = parsed._replace(scheme=parsed.scheme.lower()).geturl().rstrip("/")
if endpoint_role_for_url(normalized) == "public" and parsed.scheme.lower() != "https":
raise ValueError("public Dashboard URLs must use https://")
return normalized
def configured_dashboard_url(explicit: Optional[str] = None) -> Optional[str]:
"""Resolve Dashboard identity from explicit input or trusted URL envs."""
if explicit and explicit.strip():
return normalize_dashboard_url(explicit)
for key in (
"HERMES_DASHBOARD_PUBLIC_URL",
"HERMES_DASHBOARD_URL",
"HERMES_WEB_URL",
"DASHBOARD_URL",
):
value = os.environ.get(key, "").strip()
if value:
return normalize_dashboard_url(value)
return None
def normalize_public_url(public_url: str) -> str:
"""Return a canonical HTTPS public Dashboard or explicit Relay URL."""
normalized = normalize_dashboard_url(public_url)
if urlparse(normalized).scheme.lower() != "https":
raise ValueError("public_url must use https://")
return normalized
def dashboard_relay_ingress_url(dashboard_url: str) -> str:
"""Build the same-origin Relay transport base mounted by Hermes Dashboard."""
parsed = urlparse(normalize_dashboard_url(dashboard_url))
scheme = "wss" if parsed.scheme == "https" else "ws"
prefix = (parsed.path or "").rstrip("/")
return (
f"{scheme}://{parsed.netloc}{prefix}"
"/api/plugins/hermes-relay/transport"
)
def _route_url_key(url: str, *, expected_schemes: set[str]) -> tuple[str, str, int, str]:
"""Canonical scheme/host/effective-port/path identity for route matching."""
parsed = urlparse(url.strip().rstrip("/"))
scheme = parsed.scheme.lower()
if scheme not in expected_schemes or not parsed.hostname:
raise ValueError(f"route URL must use {sorted(expected_schemes)}")
try:
port = parsed.port
except ValueError as exc:
raise ValueError("route URL contains an invalid port") from exc
effective_port = port or (443 if scheme in {"https", "wss"} else 80)
return scheme, parsed.hostname.lower().rstrip("."), effective_port, parsed.path.rstrip("/")
def _dashboard_origin_key(dashboard_url: str) -> tuple[str, str, int, str]:
return _route_url_key(
normalize_dashboard_url(dashboard_url), expected_schemes={"http", "https"}
)
def _relay_ingress_key(relay_url: str) -> tuple[str, str, int, str]:
return _route_url_key(relay_url, expected_schemes={"ws", "wss"})
def add_dashboard_ingress_candidate(
*,
endpoints: Optional[list[dict]],
dashboard_url: Optional[str],
relay: Optional[dict],
api_host: Optional[str] = None,
api_port: Optional[int] = None,
api_tls: Optional[bool] = None,
legacy_direct_relay: bool = False,
) -> Optional[list[dict]]:
"""Advertise Dashboard's same-origin Relay ingress with direct fallbacks.
The candidate uses the normal network role (``https``, ``tailscale``, or
``lan``), not a service-specific role. API is intentionally optional.
Direct Relay is included only after an explicit legacy opt-in. New
Dashboard-origin pairing must not leak the loopback Relay port into a
public or tailnet route.
"""
dashboard = normalize_dashboard_url(dashboard_url) if dashboard_url else ""
if not dashboard:
return endpoints
relay_code = str((relay or {}).get("code") or "").strip()
if relay is None or not relay_code:
return endpoints
ingress_url = dashboard_relay_ingress_url(dashboard)
transport_hint = "wss" if ingress_url.startswith("wss://") else "ws"
ingress_role = endpoint_role_for_url(dashboard)
ingress: dict[str, Any] = {
"role": ingress_role,
"priority": 0,
"recommended": ingress_role == "tailscale",
"dashboard": {"url": dashboard},
}
ingress["relay"] = {
"url": ingress_url,
"transport_hint": transport_hint,
}
existing = [dict(candidate) for candidate in (endpoints or []) if isinstance(candidate, dict)]
relay_urls = {
str(candidate.get("relay", {}).get("url") or "").rstrip("/")
for candidate in existing
if isinstance(candidate.get("relay"), dict)
}
dashboard_origins: set[tuple[str, str, int, str]] = set()
for candidate in existing:
candidate_dashboard = candidate.get("dashboard")
if not isinstance(candidate_dashboard, dict) or not candidate_dashboard.get("url"):
continue
try:
dashboard_origins.add(_dashboard_origin_key(str(candidate_dashboard["url"])))
except ValueError:
continue
combined: list[dict] = []
if (
ingress_url.rstrip("/") not in relay_urls
and _dashboard_origin_key(dashboard) not in dashboard_origins
):
rank = {"tailscale": 0, "public": 1, "lan": 2}.get(ingress_role, 1)
inserted = False
for candidate in existing:
candidate_rank = {"tailscale": 0, "public": 1, "lan": 2}.get(
str(candidate.get("role") or "").lower(), 1
)
if not inserted and candidate_rank > rank:
combined.append(ingress)
inserted = True
combined.append(candidate)
if not inserted:
combined.append(ingress)
else:
combined.extend(existing)
direct_url = str((relay or {}).get("url") or "").strip()
if (
legacy_direct_relay
and direct_url
and direct_url.rstrip("/") not in relay_urls
and direct_url.rstrip("/") != ingress_url.rstrip("/")
):
direct: dict[str, Any] = {
"role": "legacy_direct",
"priority": len(combined),
"recommended": False,
"legacy": True,
"relay": {
"url": direct_url,
**(
{"transport_hint": relay["transport_hint"]}
if isinstance(relay, dict) and relay.get("transport_hint")
else {}
),
},
}
if api_host is not None and str(api_host).strip():
direct["api"] = {
"host": str(api_host).strip(),
"port": int(api_port) if api_port is not None else 8642,
"tls": bool(api_tls),
}
combined.append(direct)
for priority, candidate in enumerate(combined):
candidate["priority"] = priority
return combined or None
def build_relay_pairing_block(
*,
relay_url: str,
code: str,
ttl_seconds: Any = None,
grants: Optional[dict] = None,
transport_hint: Optional[str] = None,
) -> dict[str, Any]:
"""Build the nested ``relay`` block used by all QR emitters."""
def wire_seconds(value: Any) -> Any:
if isinstance(value, float) and math.isfinite(value) and value.is_integer():
return int(value)
return value
relay_block: dict[str, Any] = {
"url": relay_url,
"code": code,
}
if ttl_seconds is not None:
relay_block["ttl_seconds"] = wire_seconds(ttl_seconds)
if grants is not None:
relay_block["grants"] = {
channel: wire_seconds(duration)
for channel, duration in grants.items()
}
if transport_hint is not None:
relay_block["transport_hint"] = transport_hint
return relay_block
def build_pairing_qr_payload(
*,
host: Optional[str],
port: Optional[int],
key: Optional[str],
tls: Optional[bool],
relay: Optional[dict] = None,
endpoints: Optional[list[dict]] = None,
dashboard_url: Optional[str] = None,
legacy_direct_relay: bool = False,
sign: bool = True,
) -> str:
"""Build the Android pairing QR payload shared by CLI and dashboard mint."""
return build_payload(
host=host,
port=port,
key=key,
tls=tls,
relay=relay,
endpoints=endpoints,
dashboard_url=dashboard_url,
legacy_direct_relay=legacy_direct_relay,
sign=sign,
)
def build_pairing_invite_url(payload: str) -> str:
"""Wrap a QR payload in a paste-friendly Hermes Relay invite URL.
The QR remains compact JSON for Android scanner compatibility. The URL is
for terminals, dashboard copy buttons, and desktop paste flows where a
one-line string is easier than raw JSON.
"""
encoded = base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii")
return f"hermes-relay://pair?payload={encoded.rstrip('=')}"
# ── Endpoint candidate discovery (ADR 24) ────────────────────────────────────
_VALID_MODES = ("auto", "lan", "tailscale", "public")
_DEFAULT_DASHBOARD_PORT = 9119
_RECOMMENDED_DASHBOARD_LISTENER_PORT = 10443
_LEGACY_DASHBOARD_LISTENER_PORT = 443
def _lan_endpoint(
api_host: str,
api_port: int,
api_tls: bool,
relay_host: str,
relay_port: int,
relay_tls: bool,
priority: int = 0,
dashboard_port: int = _DEFAULT_DASHBOARD_PORT,
) -> dict[str, Any]:
"""Build a ``role: lan`` endpoint candidate using the LAN-resolved host.
Reuses the same :func:`_resolve_lan_ip` helper the single-endpoint
flow uses, so a bind-all / loopback host surfaces as a routable LAN
IP in the candidate.
"""
lan_host = _resolve_lan_ip(api_host)
dashboard_host = _resolve_lan_ip(relay_host)
url_host = (
f"[{dashboard_host}]"
if ":" in dashboard_host and not dashboard_host.startswith("[")
else dashboard_host
)
dashboard_url = f"http://{url_host}:{dashboard_port}"
return {
"role": "lan",
"priority": priority,
"api": {"host": lan_host, "port": api_port, "tls": api_tls},
"dashboard": {"url": dashboard_url},
"relay": {
"url": dashboard_relay_ingress_url(dashboard_url),
"transport_hint": "ws",
},
}
def is_explicit_relay_url(url: str) -> bool:
"""Whether an HTTP(S) URL names a Relay path rather than Dashboard base."""
path = urlparse(url).path.rstrip("/").lower()
return path.endswith(("/relay", "/ws", "/transport", "/transport/ws"))
def _tailscale_status() -> Optional[dict[str, Any]]:
"""Probe the optional Tailscale helper. Returns None on any failure.
The helper at ``plugin.relay.tailscale`` is owned by a sibling track
(ADR 25). We import it lazily and treat both ``ImportError`` and any
runtime exception as "no Tailscale endpoint available" so this code
works today on a vanilla install where the helper hasn't landed
yet. When the helper is present and returns a usable ``.ts.net``
hostname we surface it as a ``role: tailscale`` candidate.
"""
try:
from .relay import tailscale # type: ignore
except ImportError:
return None
try:
status = tailscale.status()
except Exception:
return None
if not isinstance(status, dict):
return None
return status
def _recommended_tailscale_listener(status: Optional[dict[str, Any]]) -> int:
"""Return the helper-owned recommended listener, with a safe new default."""
value = status.get("recommended_listener_port") if isinstance(status, dict) else None
if isinstance(value, int) and not isinstance(value, bool) and 1 <= value <= 65535:
return value
return _RECOMMENDED_DASHBOARD_LISTENER_PORT
def _tailscale_endpoint(
status: dict[str, Any],
api_port: int,
relay_port: int,
api_tls: bool,
relay_tls: bool,
priority: int,
dashboard_port: int = _DEFAULT_DASHBOARD_PORT,
) -> Optional[dict[str, Any]]:
"""Materialize a ``role: tailscale`` candidate from a helper status dict.
The helper contract (ADR 25) hands back something shaped roughly
like ``{"hostname": "hermes.tail-scale.ts.net", "tailscale_ip":
"100.64.0.1", "serve_ports": [...]}``. When Tailscale Serve is active
for Dashboard, emit the MagicDNS hostname with TLS. Otherwise use the
raw tailnet IP and Dashboard's normal HTTP port. Relay always rides the
Dashboard same-origin ingress; direct 8767 is legacy-only.
"""
hostname = status.get("hostname") or status.get("dns_name") or status.get("host")
if isinstance(hostname, str):
hostname = hostname.strip().rstrip(".")
else:
hostname = None
tailscale_ip = status.get("tailscale_ip") or status.get("ip") or status.get("address")
if isinstance(tailscale_ip, str):
tailscale_ip = tailscale_ip.strip()
else:
tailscale_ip = None
serve_ports_raw = status.get("serve_ports")
serve_ports: set[int] = set()
if isinstance(serve_ports_raw, list):
for port in serve_ports_raw:
try:
serve_ports.add(int(port))
except (TypeError, ValueError):
continue
serve_services = status.get("serve_services")
serve_services_dict = serve_services if isinstance(serve_services, dict) else {}
recommended_listener = _recommended_tailscale_listener(status)
def _service_listener(name: str, default_port: int) -> Optional[int]:
service = serve_services_dict.get(name)
if not isinstance(service, dict) or service.get("active") is not True:
return None
ports = service.get("listen_ports")
valid_ports = sorted({
int(port)
for port in (ports if isinstance(ports, list) else [])
if isinstance(port, int) and not isinstance(port, bool) and 1 <= port <= 65535
})
if name == "dashboard" and recommended_listener in valid_ports:
return recommended_listener
if name == "dashboard" and _LEGACY_DASHBOARD_LISTENER_PORT in valid_ports:
return _LEGACY_DASHBOARD_LISTENER_PORT
if default_port in valid_ports:
return default_port
return valid_ports[0] if valid_ports else None
dashboard_listener = _service_listener("dashboard", dashboard_port)
api_listener = _service_listener("api", api_port)
if dashboard_listener is None and not serve_services_dict:
dashboard_listener = next(
(
port
for port in (
recommended_listener,
_LEGACY_DASHBOARD_LISTENER_PORT,
dashboard_port,
)
if port in serve_ports
),
None,
)
# Compatibility with pre-classification helpers: only the well-known
# Dashboard listener proves Dashboard Serve. Legacy Relay 8767 must not.
dashboard_serve_tls = (
dashboard_listener is not None
)
if dashboard_serve_tls:
if not hostname or not hostname.endswith(".ts.net"):
return None
dashboard_host = hostname
dashboard_scheme = "https"
else:
dashboard_host = tailscale_ip or hostname
if not isinstance(dashboard_host, str) or not dashboard_host.strip():
return None
dashboard_scheme = "http"
dashboard_host = dashboard_host.strip().rstrip(".")
if not dashboard_host:
return None
dashboard_url_host = (
f"[{dashboard_host}]"
if ":" in dashboard_host and not dashboard_host.startswith("[")
else dashboard_host
)
advertised_dashboard_port = dashboard_listener or dashboard_port
dashboard_port_suffix = (
"" if dashboard_scheme == "https" and advertised_dashboard_port == 443
else f":{advertised_dashboard_port}"
)
dashboard_url = f"{dashboard_scheme}://{dashboard_url_host}{dashboard_port_suffix}"
# A classified helper status is authoritative about which optional
# services are actually published. Do not advertise a raw 100.x API
# fallback when the helper explicitly reports API Serve inactive: the
# API process may be bound only to LAN, making that tailnet address both
# misleading and unreachable. Pre-classification helpers keep the old
# direct-tailnet fallback for compatibility because they cannot express
# per-service activity.
include_api = api_listener is not None or not serve_services_dict
api_endpoint: Optional[dict[str, Any]] = None
if include_api:
api_serve_tls = bool(
(api_listener is not None or api_port in serve_ports)
and hostname
and hostname.endswith(".ts.net")
)
api_host = hostname if api_serve_tls else (tailscale_ip or hostname)
if not isinstance(api_host, str) or not api_host.strip():
return None
api_endpoint = {
"host": api_host.strip().rstrip("."),
"port": api_listener or api_port,
"tls": True if api_serve_tls else api_tls,
}
candidate: dict[str, Any] = {
"role": "tailscale",
"priority": priority,
"recommended": True,
"dashboard": {"url": dashboard_url},
"relay": {
"url": dashboard_relay_ingress_url(dashboard_url),
"transport_hint": "wss" if dashboard_serve_tls else "ws",
},
}
if api_endpoint is not None:
candidate["api"] = api_endpoint
return candidate
def validate_endpoint_candidate_security(candidate: dict[str, Any]) -> None:
"""Reject plaintext on candidates explicitly labeled as public."""
role = str(candidate.get("role") or "").strip().lower()
if role not in {"public", "public_legacy"}:
return
candidate_dashboard = candidate.get("dashboard")
if isinstance(candidate_dashboard, dict) and candidate_dashboard.get("url"):
public_dashboard = normalize_public_url(str(candidate_dashboard["url"]))
if urlparse(public_dashboard).scheme.lower() != "https": # pragma: no cover
raise ValueError("role=public Dashboard routes must use https://")
candidate_relay = candidate.get("relay")
if isinstance(candidate_relay, dict) and candidate_relay.get("url"):
relay_scheme = urlparse(str(candidate_relay["url"])).scheme.lower()
if relay_scheme != "wss":
raise ValueError("role=public Relay routes must use wss://")
candidate_api = candidate.get("api")
if isinstance(candidate_api, dict) and candidate_api.get("tls") is not True:
raise ValueError("role=public API routes must enable TLS")
def normalize_endpoint_candidates(
endpoints: Optional[list[Any]],
*,
api_port: Optional[int] = None,
relay_port: Optional[int] = None,
api_tls: Optional[bool] = None,
relay_tls: Optional[bool] = None,
) -> Optional[list[Any]]:
"""Normalize server-owned endpoint candidates before QR signing.
Dashboard/plugin callers may be long-running processes with an older
``plugin.pair`` module cached. If they hand the relay a stale Tailscale
MagicDNS+TLS candidate while Tailscale Serve is not active, rewrite that
candidate to the direct 100.x tailnet route before the relay signs the QR.
Non-Tailscale roles and malformed records pass through unchanged.
"""
if endpoints is None:
return None
normalized: list[Any] = []
status_loaded = False
status: Optional[dict[str, Any]] = None
def _int_or(value: Any, fallback: int) -> int:
try:
return int(value)
except (TypeError, ValueError):
return fallback
for index, candidate in enumerate(endpoints):
if not isinstance(candidate, dict):
normalized.append(candidate)
continue
candidate = dict(candidate)
role = str(candidate.get("role") or "").strip().lower()
validate_endpoint_candidate_security(candidate)
if candidate.get("legacy") is True:
normalized.append(candidate)
continue
candidate_dashboard = candidate.get("dashboard")
if isinstance(candidate_dashboard, dict) and candidate_dashboard.get("url"):
dashboard_dict = dict(candidate_dashboard)
dashboard_url = normalize_dashboard_url(str(dashboard_dict["url"]))
dashboard_dict["url"] = dashboard_url
candidate["dashboard"] = dashboard_dict
if candidate.get("legacy") is not True:
existing_relay = candidate.get("relay")
existing_relay_url = (
str(existing_relay.get("url") or "")
if isinstance(existing_relay, dict)
else ""
)
if existing_relay_url:
dashboard_host = (urlparse(dashboard_url).hostname or "").lower().rstrip(".")
relay_host = (urlparse(existing_relay_url).hostname or "").lower().rstrip(".")
if not relay_host or relay_host != dashboard_host:
raise ValueError(
"dashboard_url has no exact same-origin Relay ingress candidate"
)
ingress_url = dashboard_relay_ingress_url(dashboard_url)
relay_dict = (
dict(candidate["relay"])
if isinstance(candidate.get("relay"), dict)
else {}
)
relay_dict.update({
"url": ingress_url,
"transport_hint": "wss" if ingress_url.startswith("wss://") else "ws",
})
candidate["relay"] = relay_dict
if not isinstance(candidate.get("dashboard"), dict) and role in {"lan", "public", "tailscale"}:
relay_dict = candidate.get("relay")
api_dict = candidate.get("api")
relay_url = (
str(relay_dict.get("url") or "")
if isinstance(relay_dict, dict)
else ""
)
parsed_relay = urlparse(relay_url) if relay_url else None
host = parsed_relay.hostname if parsed_relay is not None else None
if not host and isinstance(api_dict, dict):
host = str(api_dict.get("host") or "").strip() or None
if host:
url_host = f"[{host}]" if ":" in host and not host.startswith("[") else host
if role == "public":
dashboard_url = f"https://{url_host}"
elif role == "tailscale":
dashboard_scheme = (
"https"
if relay_url.startswith("wss://")
or (isinstance(api_dict, dict) and api_dict.get("tls") is True)
else "http"
)
dashboard_url = f"{dashboard_scheme}://{url_host}:{_DEFAULT_DASHBOARD_PORT}"
else:
dashboard_url = f"http://{url_host}:{_DEFAULT_DASHBOARD_PORT}"
ingress_url = dashboard_relay_ingress_url(dashboard_url)
candidate["dashboard"] = {"url": dashboard_url}
candidate["relay"] = {
**(dict(relay_dict) if isinstance(relay_dict, dict) else {}),
"url": ingress_url,
"transport_hint": "wss" if ingress_url.startswith("wss://") else "ws",
}
if role != "tailscale":
normalized.append(candidate)
continue
if not status_loaded:
status = _tailscale_status()
status_loaded = True
if status is None:
normalized.append(candidate)
continue
api = candidate.get("api")
api_dict = api if isinstance(api, dict) else {}
relay = candidate.get("relay")
relay_dict = relay if isinstance(relay, dict) else {}
relay_url = str(relay_dict.get("url") or "")
parsed_relay = urlparse(relay_url) if relay_url else None
effective_api_port = _int_or(
api_dict.get("port"),
api_port if api_port is not None else 8642,
)
try:
parsed_relay_port = parsed_relay.port if parsed_relay is not None else None
except ValueError:
parsed_relay_port = None
effective_relay_port = _int_or(
parsed_relay_port, relay_port if relay_port is not None else 8767
)
effective_api_tls = (
bool(api_tls) if api_tls is not None else bool(api_dict.get("tls"))
)
effective_relay_tls = (
bool(relay_tls)
if relay_tls is not None
else relay_url.startswith("wss://")
or str(relay_dict.get("transport_hint") or "").lower() == "wss"
)
priority = _int_or(candidate.get("priority"), index)
replacement = _tailscale_endpoint(
status,
api_port=effective_api_port,
relay_port=effective_relay_port,
api_tls=effective_api_tls,
relay_tls=effective_relay_tls,
priority=priority,
dashboard_port=_DEFAULT_DASHBOARD_PORT,
)
if replacement is None:
normalized.append(candidate)
continue
merged = dict(candidate)
merged_relay = dict(relay_dict)
merged_relay.update(replacement["relay"])
merged["priority"] = replacement["priority"]
merged["recommended"] = replacement["recommended"]
if "api" in replacement:
merged_api = dict(api_dict)
merged_api.update(replacement["api"])
merged["api"] = merged_api
else:
merged.pop("api", None)
merged["dashboard"] = replacement["dashboard"]
merged["relay"] = merged_relay
normalized.append(merged)
return normalized
def _public_endpoint(
public_url: str,
relay_port: int,
priority: int,
) -> dict[str, Any]:
"""Parse ``--public-url`` as a public Dashboard-origin candidate.
Relay uses Dashboard's same-origin plugin transport. The Relay process's
private 8767 listener is never inferred from a public origin.
Raises :class:`ValueError` when ``public_url`` is empty or its
scheme isn't ``http`` / ``https``.
"""
if not public_url:
raise ValueError("public_url is required for role=public")
dashboard_url = normalize_public_url(public_url)
relay_url = dashboard_relay_ingress_url(dashboard_url)
relay_scheme = "wss" if relay_url.startswith("wss://") else "ws"
return {
"role": "public",
"priority": priority,
"recommended": False,
"dashboard": {"url": dashboard_url},
"relay": {
"url": relay_url,
"transport_hint": relay_scheme,
},
}
def _legacy_public_relay_endpoint(
public_url: str,
relay_port: int,
priority: int,
) -> dict[str, Any]:
"""Build the old public direct-Relay route after explicit opt-in."""
dashboard_url = normalize_public_url(public_url)
parsed = urlparse(dashboard_url)
scheme = "wss" if parsed.scheme == "https" else "ws"
host = parsed.hostname or ""
url_host = f"[{host}]" if ":" in host and not host.startswith("[") else host
relay_url = (
f"{scheme}://{parsed.netloc}{parsed.path.rstrip('/')}"
if is_explicit_relay_url(dashboard_url)
else f"{scheme}://{url_host}:{relay_port}"
)
return {
"role": "public_legacy",
"priority": priority,
"recommended": False,
"legacy": True,
"relay": {
"url": relay_url,
"transport_hint": scheme,
},
}
def build_endpoint_candidates(
mode: str,
api_host: str,
api_port: int,
api_tls: bool,
relay_host: str,
relay_port: int,
relay_tls: bool,
public_url: Optional[str] = None,
prefer: Optional[str] = None,
legacy_direct_relay: bool = False,
) -> list[dict[str, Any]]:
"""Build the ordered ``endpoints`` array for a v3 QR payload.
``mode`` is one of ``auto`` / ``lan`` / ``tailscale`` / ``public``
(see ADR 24). Priority is strictly increasing by role, starting at
0 for secure routes and going up — matching DNS SRV semantics (lower
number = higher priority). Tailscale and public TLS precede plain LAN,
which remains the final fallback. An empty list is returned when no candidates
could be detected (e.g. ``--mode tailscale`` on a host without
Tailscale installed); callers should treat that as "stay on the
single-endpoint v2 payload".
``prefer`` (optional) names a role that should be promoted to
priority 0, with all other roles shifted down by one. Useful for
testing a specific path end-to-end ("force Tailscale even though
I'm on LAN") without re-ordering defaults globally. When the named
role is not present in the detected candidates a warning is printed
to stderr and the list is emitted in its natural order — callers
asking for something that isn't there should see it, not silently
get the default.
Raises :class:`ValueError` if ``mode`` is unknown or ``mode=public``
is requested without a ``public_url``.
"""
if mode not in _VALID_MODES:
raise ValueError(
f"invalid --mode {mode!r}; expected one of {_VALID_MODES}"
)
candidates: list[dict[str, Any]] = []
next_priority = 0
def _emit(candidate: Optional[dict[str, Any]]) -> None:
nonlocal next_priority
if candidate is None:
return
# Respect the priority we assigned above; _lan_endpoint /
# _tailscale_endpoint / _public_endpoint honor it.
candidates.append(candidate)
next_priority += 1
want_lan = mode in ("auto", "lan")
want_tailscale = mode in ("auto", "tailscale")
want_public = mode in ("auto", "public")
effective_public_url = public_url
tailscale_status: Optional[dict[str, Any]] = None
if want_tailscale:
tailscale_status = _tailscale_status()
if tailscale_status is not None:
_emit(
_tailscale_endpoint(
tailscale_status,
api_port=api_port,
relay_port=relay_port,
api_tls=api_tls,
relay_tls=relay_tls,
priority=next_priority,
)
)
elif mode == "tailscale":
# Explicit mode but no helper / no hostname — fail soft
# rather than emit a half-formed QR. Caller falls back
# to the single-endpoint path and logs.
pass
if want_public:
# Auto-detect Tailscale Funnel URL when mode=auto and caller
# didn't pin one explicitly. Saves operators the "set public
# URL" step on the Remote Access tab when Funnel is already
# publishing the relay port. Fails soft — funnel_url returns
# None when the CLI is absent, nothing is funneled on this
# port, or the JSON parse hits an unexpected shape.
if not effective_public_url:
detected = None
_ts_helper = None
try:
from .relay import tailscale as _ts_helper # type: ignore
except Exception: # noqa: BLE001 — any failure = no helper
_ts_helper = None
if _ts_helper is not None:
probe_ports = dict.fromkeys((
_recommended_tailscale_listener(tailscale_status),
_LEGACY_DASHBOARD_LISTENER_PORT,
_DEFAULT_DASHBOARD_PORT,
))
for funnel_port in probe_ports:
try:
detected = _ts_helper.funnel_url(port=funnel_port)
except Exception: # noqa: BLE001 — each probe fails soft
detected = None
if detected:
break
if detected:
effective_public_url = detected
if effective_public_url:
explicit_relay_url = is_explicit_relay_url(
normalize_public_url(effective_public_url)
)
if explicit_relay_url and not legacy_direct_relay:
raise ValueError(
"an explicit Relay public path requires "
"--legacy-direct-relay"
)
if not explicit_relay_url:
_emit(
_public_endpoint(
effective_public_url,
relay_port=relay_port,
priority=next_priority,
)
)
elif mode == "public":
raise ValueError(
"--mode public requires --public-url <url> "
"(no Tailscale Funnel detected for this port)"
)
# Plain LAN is deliberately last in auto mode. It remains available as
# an explicit fallback, while Android and desktop receive the same signed
# secure-first ordering instead of applying divergent client heuristics.
if want_lan:
_emit(
_lan_endpoint(
api_host,
api_port,
api_tls,
relay_host,
relay_port,
relay_tls,
priority=next_priority,
)
)
if legacy_direct_relay:
if effective_public_url:
_emit(
_legacy_public_relay_endpoint(
effective_public_url,
relay_port=relay_port,
priority=next_priority,
)
)
relay_host_resolved = _resolve_lan_ip(relay_host)
url_host = (
f"[{relay_host_resolved}]"
if ":" in relay_host_resolved and not relay_host_resolved.startswith("[")
else relay_host_resolved
)
relay_scheme = "wss" if relay_tls else "ws"
_emit({
"role": "legacy_direct",
"priority": next_priority,
"recommended": False,
"legacy": True,
"relay": {
"url": f"{relay_scheme}://{url_host}:{relay_port}",
"transport_hint": relay_scheme,
},
})
# Priority override — promote the named role to priority 0 and
# renumber the rest in their existing relative order. Role string
# matches are case-insensitive + whitespace-trimmed for
# operator-ergonomics, but the candidate's original ``role`` value
# is preserved verbatim because the HMAC canonical form requires it.
if prefer:
wanted = prefer.strip().lower()
idx = next(
(i for i, c in enumerate(candidates) if str(c.get("role", "")).lower() == wanted),
-1,
)
if idx < 0:
print(
f" [warn] --prefer {prefer!r}: role not in candidates "
f"{[c.get('role') for c in candidates]}; emitting natural order",
file=sys.stderr,
)
elif idx > 0:
promoted = candidates.pop(idx)
candidates.insert(0, promoted)
for new_priority, c in enumerate(candidates):
c["priority"] = new_priority
return candidates
# ── TTL / grants parsing ─────────────────────────────────────────────────────
_TTL_PRESETS: dict[str, int] = {
# 0 => never expire
"never": 0,
"1d": 1 * 24 * 3600,
"7d": 7 * 24 * 3600,
"30d": 30 * 24 * 3600,
"90d": 90 * 24 * 3600,
# 1y ≈ 365 days. Not a leap-year-aware calendar year, just a round
# duration; "never" covers the "really long" case.
"1y": 365 * 24 * 3600,
}
def parse_duration(spec: str) -> int:
"""Parse a duration spec like ``"30d"`` / ``"1y"`` / ``"never"``.
Returns the duration in seconds. ``"never"`` returns 0, which the
relay interprets as ``math.inf`` (session never expires).
Also accepts an explicit number of seconds (e.g. ``"3600"``) for
power users.
"""
normalized = spec.strip().lower()
if not normalized:
raise ValueError("empty duration")
if normalized in _TTL_PRESETS:
return _TTL_PRESETS[normalized]
# Explicit numeric seconds.
if normalized.isdigit():
return int(normalized)
# Loose suffix form: <int>[smhdwy]
unit = normalized[-1]
head = normalized[:-1]
if not head.isdigit():
raise ValueError(f"cannot parse duration {spec!r}")
n = int(head)
if unit == "s":
return n
if unit == "m":
return n * 60
if unit == "h":
return n * 3600
if unit == "d":
return n * 24 * 3600
if unit == "w":
return n * 7 * 24 * 3600
if unit == "y":
return n * 365 * 24 * 3600
raise ValueError(f"unknown duration unit {unit!r} in {spec!r}")
def parse_grants(spec: str) -> dict[str, int]:
"""Parse a ``--grants`` spec like ``"terminal=7d,bridge=1d"``.
Returns a dict ``{channel: duration_seconds}``. Unknown channels are
accepted — the server applies its own whitelist.
"""
out: dict[str, int] = {}
if not spec.strip():
return out
for pair_str in spec.split(","):
pair_str = pair_str.strip()
if not pair_str:
continue
if "=" not in pair_str:
raise ValueError(
f"invalid grant {pair_str!r} — expected channel=duration"
)
channel, _, duration = pair_str.partition("=")
channel = channel.strip()
if not channel:
raise ValueError(f"empty channel in grant {pair_str!r}")
out[channel] = parse_duration(duration)
return out
def format_duration_label(ttl_seconds: int) -> str:
"""Return a human-readable label for a TTL like ``'30 days'`` or ``'indefinitely'``."""
if ttl_seconds == 0:
return "indefinitely"
day = 24 * 3600
if ttl_seconds % (365 * day) == 0:
n = ttl_seconds // (365 * day)
return f"{n} year{'s' if n != 1 else ''}"
if ttl_seconds % day == 0:
n = ttl_seconds // day
return f"{n} day{'s' if n != 1 else ''}"
if ttl_seconds % 3600 == 0:
n = ttl_seconds // 3600
return f"{n} hour{'s' if n != 1 else ''}"
return f"{ttl_seconds} seconds"
# ── Relay pre-pairing ────────────────────────────────────────────────────────
# Mirrors the relay's PAIRING_ALPHABET and the app's AuthManager generator.
_RELAY_CODE_ALPHABET = string.ascii_uppercase + string.digits
_RELAY_CODE_LENGTH = 6
class InvalidPairingCodeError(ValueError):
"""Raised when a user-supplied pairing code fails format validation."""
def _generate_relay_code() -> str:
"""Generate a fresh 6-char pairing code (A-Z / 0-9)."""
rng = random.SystemRandom()
return "".join(rng.choice(_RELAY_CODE_ALPHABET) for _ in range(_RELAY_CODE_LENGTH))
def normalize_pairing_code(code: str) -> str:
"""Validate a user-supplied pairing code and return its canonical form.
The relay's :class:`PairingManager` upper-cases codes internally and
accepts only characters from ``PAIRING_ALPHABET`` at exactly
``PAIRING_CODE_LENGTH`` chars (6). We mirror that here so the CLI can
fail fast with a clear message instead of letting the operator find
out via an HTTP 400.
Raises :class:`InvalidPairingCodeError` on length or alphabet
mismatches. Returns the upper-cased code on success.
"""
if code is None:
raise InvalidPairingCodeError("pairing code is required")
normalized = code.strip().upper()
if not normalized:
raise InvalidPairingCodeError("pairing code is empty")
if len(normalized) != _RELAY_CODE_LENGTH:
raise InvalidPairingCodeError(
f"pairing code must be exactly {_RELAY_CODE_LENGTH} characters "
f"(got {len(normalized)}: {code!r})"
)
bad = [c for c in normalized if c not in _RELAY_CODE_ALPHABET]
if bad:
raise InvalidPairingCodeError(
f"pairing code contains invalid characters {bad!r} — "
f"only A-Z and 0-9 are allowed"
)
return normalized
def _relay_lan_base_url(relay_host: str, relay_port: int, tls: bool = False) -> str:
"""Build the ws[s]://host:port URL the phone should connect to.
Always resolves loopback/bind-all to a routable LAN IP so the QR payload
contains a URL the phone can actually reach across the network.
"""
lan_host = _resolve_lan_ip(relay_host)
scheme = "wss" if tls else "ws"
return f"{scheme}://{lan_host}:{relay_port}"
def register_relay_code(
localhost_port: int,
code: str,
timeout_s: float = 2.0,
ttl_seconds: int | None = None,
grants: dict[str, int] | None = None,
transport_hint: str | None = None,
) -> bool:
"""Pre-register ``code`` with the running relay via loopback HTTP.
The relay's ``/pairing/register`` endpoint is gated to loopback callers,
which matches the trust model: only a process running on the same host
as the relay (operator shell) can inject pairing codes. A phone on the
LAN cannot register codes.
Optional ``ttl_seconds`` / ``grants`` / ``transport_hint`` are passed
through verbatim so the operator's choices at QR generation time are
applied to the freshly-minted session when the phone claims the code.
Returns ``True`` on success, ``False`` on any failure (relay not running,
timeout, HTTP error). Callers should treat failure as "relay pairing
unavailable" and render an API-only QR.
"""
url = f"http://127.0.0.1:{localhost_port}/pairing/register"
body_dict: dict = {"code": code}
if ttl_seconds is not None:
body_dict["ttl_seconds"] = ttl_seconds
if grants:
body_dict["grants"] = grants
if transport_hint:
body_dict["transport_hint"] = transport_hint
body = json.dumps(body_dict).encode("utf-8")
req = urllib.request.Request(
url,
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout_s) as resp:
if resp.status != 200:
return False
data = json.loads(resp.read().decode("utf-8"))
return bool(data.get("ok"))
except (urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError):
return False
def mint_relay_pairing(
localhost_port: int,
*,
host: str,
port: int,
api_key: str,
tls: bool,
ttl_seconds: int,
grants: dict[str, int] | None,
transport_hint: str,
endpoints: list[dict] | None,
dashboard_url: str | None,
api_enabled: bool = True,
legacy_direct_relay: bool = False,
timeout_s: float = 5.0,
) -> dict[str, Any] | None:
"""Ask the running Relay to mint the authoritative signed invite.
Reach bootstrap credentials are created server-side and exist only in the
returned payload. This function deliberately never logs the request or
response because both may contain pairing and route credentials.
"""
body: dict[str, Any] = {
"api_enabled": api_enabled,
"ttl_seconds": ttl_seconds,
"transport_hint": transport_hint,
}
if api_enabled:
body.update({
"host": host,
"port": port,
"api_key": api_key,
"tls": tls,
})
if grants:
body["grants"] = grants
if endpoints:
body["endpoints"] = endpoints
if dashboard_url:
body["dashboard_url"] = dashboard_url
if legacy_direct_relay:
body["legacy_direct_relay"] = True
request = urllib.request.Request(
f"http://127.0.0.1:{localhost_port}/pairing/mint",
data=json.dumps(body).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout_s) as response:
if response.status != 200:
return None
result = json.loads(response.read().decode("utf-8"))
except (urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError):
return None
if not isinstance(result, dict) or result.get("ok") is not True:
return None
qr_payload = result.get("qr_payload")
pairing_url = result.get("pairing_url")
if not isinstance(qr_payload, str) or not isinstance(pairing_url, str):
return None
try:
parsed = json.loads(qr_payload)
except (TypeError, json.JSONDecodeError):
return None
if not isinstance(parsed, dict) or not isinstance(parsed.get("relay"), dict):
return None
return result
def probe_relay(localhost_port: int, timeout_s: float = 1.0) -> Optional[dict]:
"""Check if a relay is listening on ``localhost:<port>``.
Returns the parsed /health JSON on success, or None if the relay isn't
reachable. Used to decide whether to embed a relay block in the QR.
"""
url = f"http://127.0.0.1:{localhost_port}/health"
try:
with urllib.request.urlopen(url, timeout=timeout_s) as resp:
if resp.status != 200:
return None
return json.loads(resp.read().decode("utf-8"))
except (urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError):
return None
def read_relay_config() -> dict:
"""Resolve relay host/port from env vars + defaults.
Mirrors ``plugin/relay/config.py`` — uses ``RELAY_HOST`` / ``RELAY_PORT``
if set, otherwise falls back to ``0.0.0.0:8767`` which the LAN resolver
will turn into a routable address. The ``tls`` flag is True whenever
``RELAY_SSL_CERT`` is set in the environment; ``transport_hint`` gets
set accordingly for the QR payload.
"""
host = os.getenv("RELAY_HOST") or "0.0.0.0"
try:
port = int(os.getenv("RELAY_PORT") or "8767")
except ValueError:
port = 8767
tls = bool(os.getenv("RELAY_SSL_CERT"))
return {"host": host, "port": port, "tls": tls}
def _mask_key(key: str) -> str:
"""Return a redacted preview of the API key."""
if not key:
return "(none - open access)"
if len(key) <= 8:
return "*" * len(key) + f" ({len(key)} chars)"
return f"{key[:4]}...{key[-3:]} ({len(key)} chars)"
def render_text_block(
host: str,
port: int,
key: str,
tls: bool,
relay: Optional[dict] = None,
invite_url: Optional[str] = None,
dashboard_url: Optional[str] = None,
api_enabled: bool = True,
) -> str:
"""Return formatted connection details — always shown (works in any terminal).
When a ``relay`` block is provided, adds a second section showing the
WebSocket URL and the pre-registered pairing code so the operator can
enter them manually if QR scanning fails.
"""
scheme = "https" if tls else "http"
url = f"{scheme}://{host}:{port}"
auth_status = "Bearer token configured" if key else "NO AUTH (open access)"
lines: list[Optional[str]] = [
"",
" Hermes Android Pairing",
" " + "-" * 40,
"",
f" Dashboard: {dashboard_url}" if dashboard_url else None,
f" API Server: {url}" if api_enabled else None,
f" API Key: {_mask_key(key)}" if api_enabled else None,
f" API Auth: {auth_status}" if api_enabled else None,
"",
" Enter manually in the app if QR won't scan:",
f" Dashboard: {dashboard_url}" if dashboard_url else None,
f" API: {url}" if api_enabled else None,
]
lines = [line for line in lines if line is not None]
if api_enabled and key:
lines.append(f" Key: {key}")
if relay is not None:
lines.extend([
"",
" Relay (terminal + bridge)",
" " + "-" * 40,
f" URL : {relay['url']}",
f" Code : {relay['code']} (expires in 10 min, one-shot)",
])
ttl = relay.get("ttl_seconds")
if ttl is not None:
label = format_duration_label(int(ttl))
if ttl == 0:
lines.append(f" Pair : {label} (never expires)")
else:
lines.append(f" Pair : for {label}")
grants = relay.get("grants")
if grants:
grant_parts = []
for channel, duration in grants.items():
grant_parts.append(
f"{channel}={format_duration_label(int(duration))}"
)
lines.append(f" Grants: {', '.join(grant_parts)}")
if invite_url:
lines.extend([
"",
" Copy/paste pairing invite",
" " + "-" * 40,
f" URL : {invite_url}",
" Use : Hermes Relay Desktop -> Pair -> Paste invite",
" or: hermes-relay pair --pair-qr '<URL>'",
])
lines.append("")
return "\n".join(lines)
def render_qr_terminal(payload: str) -> str:
"""Render QR as Unicode half-block string for terminal output.
Returns a helpful message if segno is not installed (graceful fallback).
"""
try:
import segno # type: ignore
except ImportError:
return (
" (QR rendering unavailable — install segno: pip install segno)\n"
" Use the text details above to enter connection info manually.\n"
)
try:
# error="l" (low ~7% redundancy) keeps the QR version as small as
# possible given the signed payload length. Keep the standard
# four-module quiet zone; dense multi-route invites become unreliable
# when terminal chrome or adjacent text touches the symbol. compact=True
# packs two modules per character
# vertically via ▀ / ▄ half-blocks, halving the visual height vs the
# full-block renderer. Together these produce the smallest terminal
# QR segno can emit without dropping features.
qr = segno.make(payload, error="l")
buf = io.StringIO()
qr.terminal(out=buf, compact=True, border=4)
return buf.getvalue()
except Exception as e:
return f" (QR render failed: {e})\n"
def render_qr_png(payload: str, path: Optional[str] = None) -> Optional[str]:
"""Save QR as PNG file. Returns the file path, or None on failure."""
try:
import segno # type: ignore
except ImportError:
return None
if path is None:
path = str(Path(tempfile.gettempdir()) / "hermes-pairing-qr.png")
try:
qr = segno.make(payload, error="l")
qr.save(path, scale=8, border=4)
return path
except Exception:
return None
def register_code_command(args) -> int:
"""CLI entry point for ``hermes-pair --register-code <code>``.
Skips QR rendering entirely. Validates the user-supplied code against
``PAIRING_ALPHABET`` + length, probes the local relay, then calls the
same loopback ``/pairing/register`` HTTP endpoint that the QR flow
uses — including any ``--ttl`` / ``--grants`` / ``--transport-hint``
options so a manual-paired session has the same gating as a QR-paired
one.
Use case: the operator can't render a QR or there's no second device
to scan with (SSH-only / camera unavailable / phone displaying a
locally-generated code in Settings → Connection → Manual pairing
code). The phone displays a code, the operator types it into a host
shell, this command pre-registers it with the relay, the phone taps
Connect.
Returns a process exit code (0 on success, non-zero on failure).
"""
raw_code = getattr(args, "register_code", None)
try:
code = normalize_pairing_code(raw_code or "")
except InvalidPairingCodeError as exc:
print(f" [error] --register-code: {exc}", file=sys.stderr)
return 2
# ── TTL + grants from CLI flags (same shape as the QR flow) ───────────
ttl_spec = getattr(args, "ttl", None) or "30d"
try:
ttl_seconds = parse_duration(ttl_spec)
except ValueError as exc:
print(f" [error] --ttl: {exc}", file=sys.stderr)
return 2
grants_spec = getattr(args, "grants", None)
grants_dict: Optional[dict[str, int]] = None
if grants_spec:
try:
grants_dict = parse_grants(grants_spec)
except ValueError as exc:
print(f" [error] --grants: {exc}", file=sys.stderr)
return 2
# Resolve relay host/port + transport hint. The CLI flag overrides
# the env-derived default so the operator can register against a
# non-default relay or pretend the transport is wss (e.g. when
# running behind an external reverse proxy that terminates TLS).
relay_cfg = read_relay_config()
relay_port = relay_cfg["port"]
relay_tls = bool(relay_cfg.get("tls"))
transport_hint = getattr(args, "transport_hint", None) or (
"wss" if relay_tls else "ws"
)
# Probe the relay first so we can give a precise error message
# instead of a generic "post failed".
health = probe_relay(relay_port)
if health is None:
print(
f" [error] No relay reachable at http://127.0.0.1:{relay_port}.\n"
f" Start the relay first: hermes relay start "
f"(or python -m plugin.relay --no-ssl)",
file=sys.stderr,
)
return 1
ok = register_relay_code(
relay_port,
code,
ttl_seconds=ttl_seconds,
grants=grants_dict,
transport_hint=transport_hint,
)
if not ok:
print(
" [error] Relay rejected the pairing code. The relay's "
"/pairing/register endpoint is loopback-only — make sure "
"you're running this command on the same host as the relay.",
file=sys.stderr,
)
return 1
# ── Success — tell the operator exactly what to do next ──────────────
ttl_label = format_duration_label(ttl_seconds)
print()
print(" Hermes-Relay manual pairing")
print(" " + "-" * 40)
print(f" Code : {code}")
print(f" Relay : http://127.0.0.1:{relay_port}")
print(f" Transport : {transport_hint}")
if ttl_seconds == 0:
print(f" Session TTL : {ttl_label} (never expires)")
else:
print(f" Session TTL : {ttl_label}")
if grants_dict:
grant_parts = [
f"{ch}={format_duration_label(int(d))}" for ch, d in grants_dict.items()
]
print(f" Grants : {', '.join(grant_parts)}")
print()
print(" Code registered. The pairing code is single-use and expires")
print(" in 10 minutes.")
print()
print(" In the Hermes-Relay app:")
print(" 1. Open Settings -> Connection -> Manual pairing code (fallback).")
print(f" 2. Confirm the displayed code matches: {code}")
print(" 3. Tap Connect.")
print()
return 0
def pair_command(args) -> None:
"""CLI entry point for `hermes pair`. Called by argparse dispatch."""
# Manual-fallback flow: skip QR entirely and just pre-register the code.
if getattr(args, "register_code", None):
sys.exit(register_code_command(args))
config = read_server_config()
# Apply CLI overrides
if getattr(args, "host", None):
config["host"] = args.host
if getattr(args, "port", None):
config["port"] = int(args.port)
host = _resolve_lan_ip(config["host"])
port = config["port"]
key = config["key"]
tls = config["tls"]
api_enabled = bool(config.get("enabled", True))
try:
dashboard_url = configured_dashboard_url(
str(getattr(args, "dashboard_url", "") or "") or None
)
except ValueError as exc:
print(f" [error] --dashboard-url: {exc}", file=sys.stderr)
sys.exit(2)
legacy_direct_relay = bool(getattr(args, "legacy_direct_relay", False))
# ── Relay pre-pairing ────────────────────────────────────────────────
#
# If a relay is running locally, ask its loopback-only /pairing/mint
# endpoint for the authoritative signed payload. This is essential for
# server-owned one-use Reach credentials. A legacy /pairing/register
# fallback is allowed only when Reach is not configured.
# ── TTL + grants from CLI flags ──────────────────────────────────────
ttl_spec = getattr(args, "ttl", None) or "30d"
try:
ttl_seconds = parse_duration(ttl_spec)
except ValueError as exc:
print(f" [error] --ttl: {exc}", file=sys.stderr)
sys.exit(2)
grants_spec = getattr(args, "grants", None)
grants_dict: Optional[dict[str, int]] = None
if grants_spec:
try:
grants_dict = parse_grants(grants_spec)
except ValueError as exc:
print(f" [error] --grants: {exc}", file=sys.stderr)
sys.exit(2)
relay_block: Optional[dict] = None
relay_health: Optional[dict] = None
skip_relay = getattr(args, "no_relay", False)
if not skip_relay:
relay_cfg = read_relay_config()
relay_port = relay_cfg["port"]
relay_tls = bool(relay_cfg.get("tls"))
transport_hint = "wss" if relay_tls else "ws"
relay_health = probe_relay(relay_port)
if relay_health is None:
print(
" [info] Relay not running at localhost:"
f"{relay_port} — QR will configure chat only."
)
print(
" Start the relay with: hermes relay start "
"(or: python -m plugin.relay --no-ssl)\n"
)
else:
# A provisional block gates endpoint discovery. The authoritative
# code and signed payload come from /pairing/mint below.
relay_block = build_relay_pairing_block(
relay_url=_relay_lan_base_url(
relay_cfg["host"], relay_port, tls=relay_tls
),
code="PENDING",
ttl_seconds=ttl_seconds,
grants=grants_dict,
transport_hint=transport_hint,
)
# ── Multi-endpoint detection (ADR 24) ────────────────────────────────
# Build an ``endpoints`` array whenever the operator asked for multi-
# network support (non-default mode OR --public-url). ``mode=auto``
# silently probes LAN + Tailscale + (if passed) --public-url; explicit
# modes emit just that role (or empty list on detection failure —
# caller falls back to the single-endpoint payload).
mode = (getattr(args, "mode", None) or "auto").strip().lower()
public_url = getattr(args, "public_url", None)
prefer = getattr(args, "prefer", None)
endpoints: list[dict] = []
# Only emit endpoints when the relay is present — the phone needs a
# relay URL per candidate to drive reconnect on network switch. If
# the operator ran with --no-relay we stay on single-endpoint.
if relay_block is not None:
_relay_cfg = read_relay_config()
try:
endpoints = build_endpoint_candidates(
mode=mode,
api_host=host,
api_port=port,
api_tls=tls,
relay_host=_relay_cfg["host"],
relay_port=_relay_cfg["port"],
relay_tls=bool(_relay_cfg.get("tls")),
public_url=public_url,
prefer=prefer,
legacy_direct_relay=legacy_direct_relay,
)
if not api_enabled:
endpoints = [
{k: v for k, v in candidate.items() if k != "api"}
for candidate in endpoints
]
except ValueError as exc:
print(f" [error] --mode/--public-url: {exc}", file=sys.stderr)
sys.exit(2)
payload: str
invite_url: str
if relay_block is not None:
broker_status = (
relay_health.get("secure_link", {}).get("broker", {})
if isinstance(relay_health, dict)
and isinstance(relay_health.get("secure_link"), dict)
else {}
)
reach_requested = bool(
isinstance(broker_status, dict) and broker_status.get("enabled")
)
minted = mint_relay_pairing(
relay_port,
host=host,
port=port,
api_key=key,
tls=tls,
ttl_seconds=ttl_seconds,
grants=grants_dict,
transport_hint=transport_hint,
endpoints=endpoints or None,
dashboard_url=dashboard_url,
api_enabled=api_enabled,
legacy_direct_relay=legacy_direct_relay,
)
if minted is not None:
payload = str(minted["qr_payload"])
invite_url = str(minted["pairing_url"])
parsed_payload = json.loads(payload)
relay_block = parsed_payload["relay"]
elif not reach_requested:
# Compatibility with pre-/pairing/mint Relay versions only. Never
# downgrade a configured Reach host to a locally constructed QR,
# because that would silently omit its one-use route credential.
relay_code = _generate_relay_code()
if register_relay_code(
relay_port,
relay_code,
ttl_seconds=ttl_seconds,
grants=grants_dict,
transport_hint=transport_hint,
):
relay_block = build_relay_pairing_block(
relay_url=_relay_lan_base_url(
relay_cfg["host"], relay_port, tls=relay_tls
),
code=relay_code,
ttl_seconds=ttl_seconds,
grants=grants_dict,
transport_hint=transport_hint,
)
payload = build_pairing_qr_payload(
host=host if api_enabled else None,
port=port if api_enabled else None,
key=key if api_enabled else None,
tls=tls if api_enabled else None,
relay=relay_block, endpoints=endpoints or None,
dashboard_url=dashboard_url,
legacy_direct_relay=legacy_direct_relay,
)
invite_url = build_pairing_invite_url(payload)
else:
relay_block = None
print(" [warn] Relay pairing mint was rejected — QR will configure chat only.\n")
payload = build_pairing_qr_payload(
host=host if api_enabled else None,
port=port if api_enabled else None,
key=key if api_enabled else None,
tls=tls if api_enabled else None,
dashboard_url=dashboard_url,
)
invite_url = build_pairing_invite_url(payload)
else:
relay_block = None
print(
" [error] Hermes Reach is configured, but the Relay could not "
"mint its secure pairing invite. No downgraded relay QR was created.\n",
file=sys.stderr,
)
payload = build_pairing_qr_payload(
host=host if api_enabled else None,
port=port if api_enabled else None,
key=key if api_enabled else None,
tls=tls if api_enabled else None,
dashboard_url=dashboard_url,
)
invite_url = build_pairing_invite_url(payload)
else:
payload = build_pairing_qr_payload(
host=host if api_enabled else None,
port=port if api_enabled else None,
key=key if api_enabled else None,
tls=tls if api_enabled else None,
dashboard_url=dashboard_url,
)
invite_url = build_pairing_invite_url(payload)
# Always show text block — works in any terminal including Hermes TUI
print(
render_text_block(
host,
port,
key,
tls,
relay=relay_block,
invite_url=invite_url,
dashboard_url=dashboard_url,
)
)
png_only = getattr(args, "png", False)
no_qr = getattr(args, "no_qr", False)
if png_only:
path = render_qr_png(payload)
if path:
print(f" PNG saved: {path}")
else:
print(" PNG render failed — install segno: pip install segno")
elif not no_qr and sys.stdout.isatty():
print(render_qr_terminal(payload))
png_path = render_qr_png(payload)
if png_path:
print(f" PNG: {png_path}")
print(" Scan with the Hermes-Relay Android app.")
if (api_enabled and key) or relay_block is not None:
print(
" WARNING: This QR contains credentials "
"(API key and/or relay pairing code). Do not share screenshots.\n"
)
else:
print()
if __name__ == "__main__":
# Allow running as `python -m plugin.pair` for manual testing
import argparse
parser = argparse.ArgumentParser(description="Hermes Android pairing")
parser.add_argument("--png", action="store_true", help="Save PNG only")
parser.add_argument("--no-qr", action="store_true", dest="no_qr", help="Text only")
parser.add_argument(
"--no-relay",
action="store_true",
dest="no_relay",
help="Skip relay pre-pairing (render API-only QR)",
)
parser.add_argument("--host", help="Override API server host")
parser.add_argument("--port", type=int, help="Override API server port")
parser.add_argument(
"--dashboard-url",
help="Embed an explicit Hermes dashboard URL for Manage/standard voice",
)
parser.add_argument(
"--legacy-direct-relay",
action="store_true",
help=(
"Also advertise the direct Relay listener (normally port 8767). "
"Use only for older Desktop clients without Dashboard ticket auth."
),
)
parser.add_argument(
"--ttl",
default="30d",
help=(
"Session TTL — one of 1d/7d/30d/90d/1y/never, or an explicit "
"<N><unit> like 12h/4w. Default: 30d."
),
)
parser.add_argument(
"--grants",
default=None,
help=(
"Per-channel grants, comma-separated channel=duration pairs, "
"e.g. 'terminal=7d,bridge=1d'. Unspecified channels get server "
"defaults."
),
)
parser.add_argument(
"--register-code",
dest="register_code",
default=None,
help=(
"Manual-fallback flow: pre-register a 6-char pairing code "
"(A-Z / 0-9) supplied by the phone and exit. Skips QR "
"rendering entirely. Composes with --ttl / --grants / "
"--transport-hint. Use this when you can't scan a QR "
"(camera unavailable, SSH-only access, second-device pair "
"impossible) — the phone displays a code in Settings -> "
"Connection -> Manual pairing code (fallback), you type it "
"into this command, and then tap Connect in the app."
),
)
parser.add_argument(
"--transport-hint",
dest="transport_hint",
default=None,
choices=["ws", "wss"],
help=(
"Override the transport hint stored alongside the session "
"(only meaningful with --register-code). Defaults to 'wss' "
"when RELAY_SSL_CERT is set, otherwise 'ws'."
),
)
parser.add_argument(
"--mode",
dest="mode",
default="auto",
choices=list(_VALID_MODES),
help=(
"Endpoint discovery mode (ADR 24). 'auto' (default) probes "
"LAN + Tailscale + --public-url (when passed) and emits an "
"ordered ``endpoints`` array in the QR. 'lan' / 'tailscale' "
"/ 'public' emit just that role. 'public' requires "
"--public-url."
),
)
parser.add_argument(
"--public-url",
dest="public_url",
default=None,
help=(
"Public hostname for a reverse proxy / Cloudflare Tunnel "
"(e.g. https://hermes.example.com). Added as a role=public "
"endpoint candidate in the QR. Must be http:// or https://."
),
)
parser.add_argument(
"--prefer",
dest="prefer",
default=None,
help=(
"Promote a named role to priority 0 (highest). Any open-vocab "
"role string accepted — commonly 'lan' / 'tailscale' / 'public'. "
"Role matching is case-insensitive. Example: "
"'--mode auto --prefer tailscale' emits all detected modes but "
"with Tailscale as the first-probed endpoint. Warns to stderr "
"if the named role isn't in the detected candidates."
),
)
pair_command(parser.parse_args())
Reference in New Issue View Git Blame Copy Permalink