diff --git a/solstone/apps/network/home_candidates.py b/solstone/apps/network/home_candidates.py new file mode 100644 index 000000000..4f60b7cd1 --- /dev/null +++ b/solstone/apps/network/home_candidates.py @@ -0,0 +1,48 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Pure home-address candidate resolution for network pairing.""" + +from __future__ import annotations + +from solstone.think.link.local_endpoints import LocalEndpoint +from solstone.think.pairing.config import is_usable_ipv4 + +VPN_SCOPES = {"vpn"} + + +def resolve_pair_link_candidates( + endpoints: list[LocalEndpoint], + route_ipv4: str | None, +) -> list[str]: + """Return up to four ordered usable IPv4 candidates for direct pairing.""" + + usable_route = ( + route_ipv4 if route_ipv4 is not None and is_usable_ipv4(route_ipv4) else None + ) + filtered = [endpoint for endpoint in endpoints if is_usable_ipv4(endpoint.ip)] + if not filtered: + return [usable_route] if usable_route is not None else [] + + non_vpn: list[str] = [] + vpn: list[str] = [] + for endpoint in filtered: + (vpn if endpoint.scope in VPN_SCOPES else non_vpn).append(endpoint.ip) + + if usable_route is not None: + for group in (non_vpn, vpn): + if usable_route in group: + group.remove(usable_route) + group.insert(0, usable_route) + break + + deduped: list[str] = [] + seen: set[str] = set() + for candidate in (*non_vpn, *vpn): + if candidate not in seen: + deduped.append(candidate) + seen.add(candidate) + return deduped[:4] + + +__all__ = ["VPN_SCOPES", "resolve_pair_link_candidates"] diff --git a/solstone/apps/network/routes.py b/solstone/apps/network/routes.py index 593ed5533..456894221 100644 --- a/solstone/apps/network/routes.py +++ b/solstone/apps/network/routes.py @@ -47,6 +47,10 @@ from solstone.apps.network.copy import ( link_copy_payload, ) from solstone.apps.network.crockford32 import encode as crockford_encode +from solstone.apps.network.home_candidates import ( + VPN_SCOPES, + resolve_pair_link_candidates, +) from solstone.apps.network.relay_link import ( derive_rk, encode_pair_window_link, @@ -98,11 +102,11 @@ from solstone.think.link.paths import ( ) from solstone.think.link.window import read_posture from solstone.think.pairing.config import ( - InvalidHostUrl, - clear_host_url, - override_host_port, - set_host_url, - validate_host_url, + InvalidHomeAddress, + clear_home_address, + get_home_address, + set_home_address, + validate_home_address, ) from solstone.think.services import operations, spl, spl_handoff from solstone.think.services import status as service_status @@ -113,9 +117,6 @@ from solstone.think.utils import get_journal, now_ms logger = logging.getLogger(__name__) _SENDER_INSTANCE_ID_RE = re.compile(r"^[A-Za-z0-9-]{1,256}$") VALID_ROLES = {"", "phone", "observer", "peer"} -# Overlay (Tailscale/CGNAT) endpoints the watcher scopes `vpn`; surfaced via -# /api/status and ordered after lan/ula in pair-link candidates. -VPN_SCOPES = {"vpn"} _HEALTH_FRESHNESS_MS = 90_000 journal_sources = import_module("solstone.apps.import.journal_sources") create_state_directory = journal_sources.create_state_directory @@ -190,35 +191,20 @@ def _current_local_endpoints() -> list[LocalEndpoint]: return watcher.snapshot() if watcher else [] -def _list_pair_link_candidates() -> list[str]: - """Return up to 4 watcher IPv4 candidates. +@dataclass(frozen=True) +class HomeAddressStatus: + home_address: str | None + lan_accessible: bool + candidates: list[str] + home_candidates: list[dict[str, Any]] + home_candidates_state: str + home_candidates_error: str | None + - vpn-scoped candidates always order after lan/ula ones, and the default-route - promotion happens only within the route IP's own scope group — a vpn route IP - never displaces an available lan candidate. Results are deduped and capped. - """ - non_vpn: list[str] = [] - vpn: list[str] = [] - for endpoint in _current_local_endpoints(): - address = ipaddress.ip_address(endpoint.ip) - if not isinstance(address, ipaddress.IPv4Address): - continue - (vpn if endpoint.scope in VPN_SCOPES else non_vpn).append(str(address)) - - route_ip = _detect_lan_ip() - for group in (non_vpn, vpn): - if route_ip in group: - group.remove(route_ip) - group.insert(0, route_ip) - break - - deduped: list[str] = [] - seen: set[str] = set() - for candidate in (*non_vpn, *vpn): - if candidate not in seen: - deduped.append(candidate) - seen.add(candidate) - return deduped[:4] +def _resolve_pair_link_candidates(endpoints: list[LocalEndpoint]) -> list[str]: + """Return resolved direct-pairing IPv4 candidates from local evidence.""" + + return resolve_pair_link_candidates(endpoints, _detect_lan_ip()) def _secure_listener_port() -> int: @@ -231,11 +217,13 @@ def _secure_listener_port() -> int: return interface_watcher.LINK_DIRECT_PORT -def _home_candidate_entries() -> list[dict[str, Any]]: +def _home_candidate_entries( + home_address: str | None, + candidates: list[str], +) -> list[dict[str, Any]]: port = _secure_listener_port() - detected = [f"{ip}:{port}" for ip in _list_pair_link_candidates()] - override = override_host_port() - selected = override or (detected[0] if detected else None) + detected = [f"{ip}:{port}" for ip in candidates] + selected = home_address or (detected[0] if detected else None) entries: list[dict[str, Any]] = [ { @@ -245,10 +233,10 @@ def _home_candidate_entries() -> list[dict[str, Any]]: } for address in detected ] - if override is not None and override not in detected: + if home_address is not None and home_address not in detected: entries.append( { - "address": override, + "address": home_address, "selected": True, "source": "override", } @@ -256,11 +244,53 @@ def _home_candidate_entries() -> list[dict[str, Any]]: return entries -def _effective_home_address() -> tuple[bool, str | None]: - override_addr = override_host_port() - if override_addr is not None: - return True, override_addr - return _is_lan_accessible(), None +def _home_address_host(home_address: str) -> str: + return home_address.partition(":")[0] + + +def _home_address_status() -> tuple[HomeAddressStatus, list[LocalEndpoint]]: + home_address = get_home_address() + try: + endpoints = _current_local_endpoints() + candidates = _resolve_pair_link_candidates(endpoints) + except Exception: + logger.exception("link home candidate collection failed") + if home_address is not None: + candidates = [_home_address_host(home_address)] + return ( + HomeAddressStatus( + home_address=home_address, + lan_accessible=True, + candidates=candidates, + home_candidates=_home_candidate_entries(home_address, []), + home_candidates_state="ready", + home_candidates_error=None, + ), + [], + ) + return ( + HomeAddressStatus( + home_address=None, + lan_accessible=False, + candidates=[], + home_candidates=[], + home_candidates_state="unavailable", + home_candidates_error=link_copy.HOME_CANDIDATES_ERROR, + ), + [], + ) + + return ( + HomeAddressStatus( + home_address=home_address, + lan_accessible=home_address is not None or bool(candidates), + candidates=candidates, + home_candidates=_home_candidate_entries(home_address, candidates), + home_candidates_state="ready", + home_candidates_error=None, + ), + endpoints, + ) def _detect_lan_ip() -> str | None: @@ -347,18 +377,6 @@ def _jsonify_preserving_order(payload: dict[str, Any]) -> Response: return Response(_json.dumps(payload), mimetype="application/json") -def _is_lan_accessible() -> bool: - """Check whether the journal's home address is reachable on the LAN. - - Feeds the home-address reachability status on /link. Best-effort: the - signal is the Host header the dashboard loaded under. - """ - hostname, _, _ = request.host.partition(":") - if hostname in ("localhost", "127.0.0.1", "::1"): - return bool(_detect_lan_ip()) - return True - - def _derive_relay_state(token_present: bool) -> str: """Return pre-mechanism relay attachment state. @@ -487,28 +505,23 @@ def api_status() -> Any: token = load_service_token() token_present = token is not None ca_fp = _ca_fingerprint() if ca_dir().exists() else None - lan_accessible, home_address = _effective_home_address() + home_status, local_endpoints = _home_address_status() posture = read_posture() relay_state = ( _derive_spl_relay_state(token_present, health, now_ms_val) if posture == "spl" else _derive_relay_state(token_present) ) - reachability = _derive_reachability(lan_accessible, posture, relay_state) + reachability = _derive_reachability( + home_status.lan_accessible, + posture, + relay_state, + ) vpn_candidates = [ {"label": ep.scope, "address": f"{ep.ip}:{ep.port}"} - for ep in _current_local_endpoints() + for ep in local_endpoints if ep.scope in VPN_SCOPES ] - try: - home_candidates = _home_candidate_entries() - home_candidates_state = "ready" - home_candidates_error = None - except Exception: - logger.exception("link home candidate collection failed") - home_candidates = [] - home_candidates_state = "unavailable" - home_candidates_error = link_copy.HOME_CANDIDATES_ERROR return jsonify( { "instance_id": state.instance_id if state else None, @@ -516,7 +529,7 @@ def api_status() -> Any: "enrolled": token_present, "relay_url": relay_url(), "ca_fingerprint": ca_fp, - "lan_accessible": lan_accessible, + "lan_accessible": home_status.lan_accessible, "posture": posture, "reachability": reachability, "relay_state": relay_state, @@ -531,11 +544,11 @@ def api_status() -> Any: "last_relay_tunnel_error_at": ( health["last_relay_tunnel_error_at"] if health else None ), - "home_address": home_address, + "home_address": home_status.home_address, "vpn": {"active": None, "candidates": vpn_candidates}, - "home_candidates": home_candidates, - "home_candidates_state": home_candidates_state, - "home_candidates_error": home_candidates_error, + "home_candidates": home_status.home_candidates, + "home_candidates_state": home_status.home_candidates_state, + "home_candidates_error": home_status.home_candidates_error, } ) @@ -617,21 +630,21 @@ def private_link_disable() -> tuple[Response, int]: @network_bp.route("/host-address", methods=["POST"]) -def set_host_address() -> Any: +def set_home_address_route() -> Any: payload = request.get_json(silent=True) or {} if not isinstance(payload, dict): payload = {} - raw_address = payload.get("address") - address = raw_address if isinstance(raw_address, str) else None + raw_address = payload.get("home_address") + home_address = raw_address if isinstance(raw_address, str) else None try: - if address is not None and address.strip(): - set_host_url(validate_host_url(address)) + if home_address is not None and home_address.strip(): + set_home_address(validate_home_address(home_address)) else: - clear_host_url() - except InvalidHostUrl as exc: + clear_home_address() + except InvalidHomeAddress as exc: return error_response(INVALID_CONFIG_VALUE, detail=str(exc)) - _, home_address = _effective_home_address() - return jsonify({"ok": True, "home_address": home_address}) + home_status, _ = _home_address_status() + return jsonify({"ok": True, "home_address": home_status.home_address}) @network_bp.get("/local-endpoints") @@ -701,11 +714,15 @@ def pair_start() -> Any: else: ca_fp = _ca_fingerprint() port = _secure_listener_port() - override = override_host_port() - if override is not None: - candidates = [override.partition(":")[0]] + home_address = get_home_address() + if home_address is not None: + candidates = [_home_address_host(home_address)] else: - candidates = _list_pair_link_candidates() + try: + candidates = _resolve_pair_link_candidates(_current_local_endpoints()) + except Exception: + logger.exception("link pair-start candidate collection failed") + candidates = [] if not candidates: return error_response( PAIRING_REQUEST_INVALID, diff --git a/solstone/apps/network/tests/test_api_status.py b/solstone/apps/network/tests/test_api_status.py index 2533206d1..07ca89809 100644 --- a/solstone/apps/network/tests/test_api_status.py +++ b/solstone/apps/network/tests/test_api_status.py @@ -99,10 +99,10 @@ def _write_service_token(env: Any, token: str = "secret-token-xyz") -> None: ) -def _write_host_override(env: Any, address: str) -> None: +def _write_home_address(env: Any, address: str) -> None: config_path = env.journal / "config" / "journal.json" config = json.loads(config_path.read_text("utf-8")) - config["pairing"] = {"host_url": f"http://{address}"} + config["pairing"] = {"home_address": address} config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") @@ -148,12 +148,9 @@ def test_direct_healthy_reports_online(link_env, monkeypatch) -> None: assert data["relay_state"] == "not-enrolled" -def test_direct_reports_host_address_override(link_env, monkeypatch) -> None: +def test_direct_reports_manual_home_address(link_env, monkeypatch) -> None: env = link_env() - config_path = env.journal / "config" / "journal.json" - config = json.loads(config_path.read_text("utf-8")) - config["pairing"] = {"host_url": "http://192.168.1.44:7657"} - config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") + _write_home_address(env, "192.168.1.44:7657") monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: "192.168.1.50") data = _get_status(env) @@ -165,10 +162,7 @@ def test_direct_reports_host_address_override(link_env, monkeypatch) -> None: def test_host_address_override_unblocks_lan_unreachable(link_env, monkeypatch) -> None: env = link_env() - config_path = env.journal / "config" / "journal.json" - config = json.loads(config_path.read_text("utf-8")) - config["pairing"] = {"host_url": "http://192.168.1.44:7657"} - config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") + _write_home_address(env, "192.168.1.44:7657") monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: None) data = _get_status(env) @@ -179,7 +173,7 @@ def test_host_address_override_unblocks_lan_unreachable(link_env, monkeypatch) - def test_loopback_only_is_lan_unreachable(link_env, monkeypatch) -> None: - env = link_env() + env = link_env(local_endpoints=[]) monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: None) data = _get_status(env) @@ -189,8 +183,24 @@ def test_loopback_only_is_lan_unreachable(link_env, monkeypatch) -> None: assert data["reachability"] == "lan-unreachable" +def test_empty_snapshot_route_fallback_reports_online(link_env, monkeypatch) -> None: + env = link_env(local_endpoints=[]) + monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: "192.168.1.50") + + data = _get_status(env) + + assert data["lan_accessible"] is True + assert data["home_address"] is None + assert data["reachability"] == "online" + assert data["home_candidates"] == [ + {"address": "192.168.1.50:7657", "selected": True, "source": "detected"} + ] + assert data["home_candidates_state"] == "ready" + assert data["home_candidates_error"] is None + + def test_lan_unreachable_precedence_over_spl(link_env, monkeypatch) -> None: - env = link_env() + env = link_env(local_endpoints=[]) _write_config(env, link={"posture": "spl"}) _write_service_token(env) monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: None) @@ -401,7 +411,7 @@ def test_home_candidates_override_in_detected_selects_detected( LocalEndpoint(ip="192.168.1.51", port=7657, scope="lan"), ] ) - _write_host_override(env, "192.168.1.51:7657") + _write_home_address(env, "192.168.1.51:7657") monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: "192.168.1.50") data = _get_status(env) @@ -419,7 +429,7 @@ def test_home_candidates_override_not_detected_appends_override( env = link_env( local_endpoints=[LocalEndpoint(ip="192.168.1.50", port=7657, scope="lan")] ) - _write_host_override(env, "192.168.1.44:7657") + _write_home_address(env, "192.168.1.44:7657") monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: "192.168.1.50") data = _get_status(env) @@ -437,17 +447,41 @@ def test_home_candidates_unavailable_keeps_status_200( env = link_env() monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: "192.168.1.50") - def fail_candidates() -> list[str]: + def fail_candidates(endpoints: list[LocalEndpoint]) -> list[str]: raise RuntimeError("watcher exploded") - monkeypatch.setattr(link_routes, "_list_pair_link_candidates", fail_candidates) + monkeypatch.setattr(link_routes, "_resolve_pair_link_candidates", fail_candidates) data = _get_status(env) assert data["home_candidates"] == [] assert data["home_candidates_state"] == "unavailable" assert data["home_candidates_error"] == link_copy.HOME_CANDIDATES_ERROR + assert data["reachability"] == "lan-unreachable" + + +def test_home_candidates_exception_with_override_stays_usable( + link_env, + monkeypatch, +) -> None: + env = link_env() + _write_home_address(env, "192.168.1.44:7657") + + def fail_candidates(endpoints: list[LocalEndpoint]) -> list[str]: + raise RuntimeError("watcher exploded") + + monkeypatch.setattr(link_routes, "_resolve_pair_link_candidates", fail_candidates) + + data = _get_status(env) + + assert data["lan_accessible"] is True + assert data["home_address"] == "192.168.1.44:7657" assert data["reachability"] == "online" + assert data["home_candidates"] == [ + {"address": "192.168.1.44:7657", "selected": True, "source": "override"} + ] + assert data["home_candidates_state"] == "ready" + assert data["home_candidates_error"] is None def test_api_status_does_not_mint_pairing_nonces(link_env, monkeypatch) -> None: diff --git a/solstone/apps/network/tests/test_home_candidates.py b/solstone/apps/network/tests/test_home_candidates.py new file mode 100644 index 000000000..d847b38ae --- /dev/null +++ b/solstone/apps/network/tests/test_home_candidates.py @@ -0,0 +1,65 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from solstone.apps.network.home_candidates import resolve_pair_link_candidates +from solstone.think.link.local_endpoints import LocalEndpoint + + +def _endpoint(ip: str, scope: str = "lan") -> LocalEndpoint: + return LocalEndpoint(ip=ip, port=7657, scope=scope) + + +def test_resolver_orders_non_vpn_before_vpn() -> None: + assert resolve_pair_link_candidates( + [ + _endpoint("100.64.0.5", "vpn"), + _endpoint("192.168.1.10"), + _endpoint("192.168.1.11"), + ], + None, + ) == ["192.168.1.10", "192.168.1.11", "100.64.0.5"] + + +def test_resolver_promotes_route_only_within_existing_bucket() -> None: + assert resolve_pair_link_candidates( + [ + _endpoint("192.168.1.10"), + _endpoint("100.64.0.5", "vpn"), + ], + "100.64.0.5", + ) == ["192.168.1.10", "100.64.0.5"] + + +def test_resolver_does_not_inject_distinct_route_into_non_empty_watcher() -> None: + assert resolve_pair_link_candidates( + [ + _endpoint("192.168.1.10"), + _endpoint("192.168.1.11"), + ], + "192.168.1.99", + ) == ["192.168.1.10", "192.168.1.11"] + + +def test_resolver_uses_route_fallback_when_watcher_empty() -> None: + assert resolve_pair_link_candidates([], "192.168.1.50") == ["192.168.1.50"] + + +def test_resolver_ignores_unusable_route() -> None: + assert resolve_pair_link_candidates([], "127.0.0.1") == [] + + +def test_resolver_dedupes_excludes_ipv6_and_caps() -> None: + assert resolve_pair_link_candidates( + [ + _endpoint("192.168.1.10"), + _endpoint("fd00::1", "ula"), + _endpoint("192.168.1.11"), + _endpoint("192.168.1.10"), + _endpoint("192.168.1.12"), + _endpoint("192.168.1.13"), + _endpoint("192.168.1.14"), + ], + "192.168.1.14", + ) == ["192.168.1.14", "192.168.1.10", "192.168.1.11", "192.168.1.12"] diff --git a/solstone/apps/network/tests/test_host_address_route.py b/solstone/apps/network/tests/test_host_address_route.py index b1d40b6c4..1bf2519bd 100644 --- a/solstone/apps/network/tests/test_host_address_route.py +++ b/solstone/apps/network/tests/test_host_address_route.py @@ -19,7 +19,7 @@ def test_host_address_sets_canonical_override(link_env) -> None: response = env.client.post( "/app/network/host-address", - json={"address": "http://192.168.1.44:7657"}, + json={"home_address": "192.168.1.44:7657"}, ) assert response.status_code == 200 @@ -27,9 +27,7 @@ def test_host_address_sets_canonical_override(link_env) -> None: "ok": True, "home_address": "192.168.1.44:7657", } - assert ( - _read_config(env.journal)["pairing"]["host_url"] == "http://192.168.1.44:7657" - ) + assert _read_config(env.journal)["pairing"]["home_address"] == "192.168.1.44:7657" def test_host_address_normalizes_bare_ipv4_port(link_env) -> None: @@ -37,27 +35,26 @@ def test_host_address_normalizes_bare_ipv4_port(link_env) -> None: response = env.client.post( "/app/network/host-address", - json={"address": "192.168.1.44:7657"}, + json={"home_address": " 192.168.1.44:7657 "}, ) assert response.status_code == 200 assert response.get_json()["home_address"] == "192.168.1.44:7657" - assert ( - _read_config(env.journal)["pairing"]["host_url"] == "http://192.168.1.44:7657" - ) + assert _read_config(env.journal)["pairing"]["home_address"] == "192.168.1.44:7657" def test_host_address_clears_on_empty_null_or_missing(link_env) -> None: env = link_env() - for body in ({"address": ""}, {"address": None}, {}): + for body in ({"home_address": ""}, {"home_address": None}, {}): env.client.post( - "/app/network/host-address", json={"address": "192.168.1.44:7657"} + "/app/network/host-address", + json={"home_address": "192.168.1.44:7657"}, ) response = env.client.post("/app/network/host-address", json=body) assert response.status_code == 200 - assert _read_config(env.journal)["pairing"]["host_url"] is None + assert _read_config(env.journal)["pairing"]["home_address"] is None def test_host_address_rejects_hostname_without_writing(link_env) -> None: @@ -66,13 +63,13 @@ def test_host_address_rejects_hostname_without_writing(link_env) -> None: response = env.client.post( "/app/network/host-address", - json={"address": "mylab.local:7657"}, + json={"home_address": "mylab.local:7657"}, ) assert response.status_code == 400 payload = response.get_json() assert payload["reason_code"] == "invalid_config_value" - assert payload["detail"] == pairing_config.HOST_URL_HOSTNAME_UNSUPPORTED + assert payload["detail"] == pairing_config.HOME_ADDRESS_HOSTNAME_UNSUPPORTED assert _read_config(env.journal) == before @@ -82,11 +79,11 @@ def test_host_address_rejects_malformed_without_writing(link_env) -> None: response = env.client.post( "/app/network/host-address", - json={"address": "http://192.168.1.44"}, + json={"home_address": "http://192.168.1.44"}, ) assert response.status_code == 400 payload = response.get_json() assert payload["reason_code"] == "invalid_config_value" - assert payload["detail"] == pairing_config.HOST_URL_INVALID + assert payload["detail"] == pairing_config.HOME_ADDRESS_INVALID assert _read_config(env.journal) == before diff --git a/solstone/apps/network/tests/test_pair_start.py b/solstone/apps/network/tests/test_pair_start.py index 8d4241aa9..dd5843ade 100644 --- a/solstone/apps/network/tests/test_pair_start.py +++ b/solstone/apps/network/tests/test_pair_start.py @@ -13,6 +13,7 @@ from solstone.apps.network import routes as link_routes from solstone.apps.network.crockford32 import decode as crockford_decode from solstone.apps.network.relay_link import decode_pair_window_link, derive_rk from solstone.think.link.ca import load_or_generate_ca +from solstone.think.link.local_endpoints import LocalEndpoint from solstone.think.link.nonces import NONCE_TTL_SECONDS from solstone.think.link.paths import ca_dir @@ -114,7 +115,7 @@ def test_pair_start_uses_host_address_override_for_direct_qr(link_env) -> None: env = link_env() config_path = env.journal / "config" / "journal.json" config = json.loads(config_path.read_text("utf-8")) - config["pairing"] = {"host_url": "http://192.0.2.44:7070"} + config["pairing"] = {"home_address": "192.0.2.44:7657"} config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") response = env.client.post( @@ -128,7 +129,6 @@ def test_pair_start_uses_host_address_override_for_direct_qr(link_env) -> None: assert decoded[0:2] == b"\x04\x01" assert decoded[2:6] == ipaddress.IPv4Address("192.0.2.44").packed assert int.from_bytes(decoded[6:8], "big") == link_routes._secure_listener_port() - assert int.from_bytes(decoded[6:8], "big") != 7070 def test_pair_start_direct_pair_link_port_uses_secure_listener_source( @@ -138,7 +138,7 @@ def test_pair_start_direct_pair_link_port_uses_secure_listener_source( env = link_env() config_path = env.journal / "config" / "journal.json" config = json.loads(config_path.read_text("utf-8")) - config["pairing"] = {"host_url": "http://192.0.2.44:7070"} + config["pairing"] = {"home_address": "192.0.2.44:7657"} config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") monkeypatch.setattr(link_routes.interface_watcher, "LINK_DIRECT_PORT", 8765) @@ -154,8 +154,55 @@ def test_pair_start_direct_pair_link_port_uses_secure_listener_source( assert int.from_bytes(decoded[6:8], "big") == 8765 -def test_pair_start_no_candidates_rejected_without_nonce(link_env) -> None: +def test_pair_start_no_candidates_rejected_without_nonce( + link_env, + monkeypatch, +) -> None: + env = link_env(local_endpoints=[]) + monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: None) + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + + assert response.status_code == 400 + payload = response.get_json() + assert payload["reason_code"] == "pairing_request_invalid" + assert payload["detail"] == "pair-link requires an IPv4 LAN address; none found" + assert link_routes._nonces().snapshot() == [] + + +def test_pair_start_uses_route_fallback_when_snapshot_empty( + link_env, + monkeypatch, +) -> None: env = link_env(local_endpoints=[]) + monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: "192.168.1.50") + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + + assert response.status_code == 200 + payload = response.get_json() + decoded = _decode_pair_link(payload["pair_link"]) + assert decoded[0:2] == b"\x04\x01" + assert decoded[2:6] == ipaddress.IPv4Address("192.168.1.50").packed + assert int.from_bytes(decoded[6:8], "big") == link_routes._secure_listener_port() + + +def test_pair_start_resolver_exception_rejected_without_nonce( + link_env, + monkeypatch, +) -> None: + env = link_env() + + def fail_candidates(endpoints): + raise RuntimeError("watcher exploded") + + monkeypatch.setattr(link_routes, "_resolve_pair_link_candidates", fail_candidates) response = env.client.post( "/app/network/pair-start", @@ -169,6 +216,69 @@ def test_pair_start_no_candidates_rejected_without_nonce(link_env) -> None: assert link_routes._nonces().snapshot() == [] +def test_pair_start_override_survives_resolver_exception( + link_env, + monkeypatch, +) -> None: + env = link_env() + config_path = env.journal / "config" / "journal.json" + config = json.loads(config_path.read_text("utf-8")) + config["pairing"] = {"home_address": "192.168.1.44:7657"} + config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") + + def fail_candidates(endpoints): + raise RuntimeError("watcher exploded") + + monkeypatch.setattr(link_routes, "_resolve_pair_link_candidates", fail_candidates) + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + + assert response.status_code == 200 + payload = response.get_json() + decoded = _decode_pair_link(payload["pair_link"]) + assert decoded[0:2] == b"\x04\x01" + assert decoded[2:6] == ipaddress.IPv4Address("192.168.1.44").packed + assert int.from_bytes(decoded[6:8], "big") == link_routes._secure_listener_port() + + +def test_pair_start_detected_order_matches_api_status( + link_env, + monkeypatch, +) -> None: + env = link_env( + local_endpoints=[ + LocalEndpoint(ip="192.168.1.51", port=1111, scope="lan"), + LocalEndpoint(ip="192.168.1.50", port=2222, scope="lan"), + LocalEndpoint(ip="192.168.1.52", port=3333, scope="lan"), + ] + ) + monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: "192.168.1.50") + + status_response = env.client.get( + "/app/network/api/status", + base_url="http://localhost:7657", + ) + assert status_response.status_code == 200 + status_payload = status_response.get_json() + status_addresses = [ + entry["address"].partition(":")[0] + for entry in status_payload["home_candidates"] + ] + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + assert response.status_code == 200 + pair_addresses = _decode_pair_link_addresses(response.get_json()["pair_link"]) + + assert status_addresses == ["192.168.1.50", "192.168.1.51", "192.168.1.52"] + assert pair_addresses == status_addresses + + def _fragment(pair_link: str) -> str: return pair_link.rsplit("#", 1)[1] @@ -177,6 +287,18 @@ def _decode_pair_link(pair_link: str) -> bytes: return crockford_decode(_fragment(pair_link)) +def _decode_pair_link_addresses(pair_link: str) -> list[str]: + decoded = _decode_pair_link(pair_link) + if decoded[0:2] == b"\x04\x01": + return [str(ipaddress.IPv4Address(decoded[2:6]))] + assert decoded[0:2] == b"\x05\x01" + count = decoded[2] + return [ + str(ipaddress.IPv4Address(decoded[offset : offset + 4])) + for offset in range(5, 5 + count * 4, 4) + ] + + def test_pair_start_spl_mints_relay_form_pair_link(link_env) -> None: env = link_env(posture="spl", service_token="svc-token-xyz") diff --git a/solstone/apps/network/tests/test_workspace_reach.py b/solstone/apps/network/tests/test_workspace_reach.py index 561f7d866..9f37158dd 100644 --- a/solstone/apps/network/tests/test_workspace_reach.py +++ b/solstone/apps/network/tests/test_workspace_reach.py @@ -267,6 +267,7 @@ def test_workspace_home_candidate_picker_js_paths(link_env) -> None: submit_end = body.index("async function applyHostAddressOverride", submit_start) submit_body = body[submit_start:submit_end] assert "'/app/network/host-address'" in submit_body + assert "JSON.stringify({ home_address: address })" in submit_body assert "setHostAddressError('');" in submit_body assert "return await refreshStatus();" in submit_body diff --git a/solstone/apps/network/workspace.html b/solstone/apps/network/workspace.html index cd5003139..530a055fd 100644 --- a/solstone/apps/network/workspace.html +++ b/solstone/apps/network/workspace.html @@ -956,7 +956,7 @@ await window.apiJson('/app/network/host-address', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ address }), + body: JSON.stringify({ home_address: address }), }); setHostAddressError(''); return await refreshStatus(); diff --git a/solstone/apps/settings/call.py b/solstone/apps/settings/call.py index 31620f3d9..09b9a7791 100644 --- a/solstone/apps/settings/call.py +++ b/solstone/apps/settings/call.py @@ -11,9 +11,7 @@ from typing import Any import typer -from solstone.convey.reasons import ( - INVALID_CONFIG_VALUE, -) +from solstone.convey.reasons import INVALID_CONFIG_VALUE from solstone.think.convey_client import ConveyClientError, convey_cli, get_client # Mirrors solstone.apps.settings.routes.API_KEY_ENV_VARS (the canonical order @@ -155,47 +153,10 @@ def processing_set( _echo_json(response.get("config", {}).get("processing", {})) -@convey_app.command("host-url") -@convey_cli -def convey_host_url( - url: str | None = typer.Argument( - None, help="Absolute URL to advertise to devices." - ), - auto: bool = typer.Option( - False, "--auto", help="Clear the manual host URL override." - ), - show: bool = typer.Option(False, "--show", help="Show the effective host URL."), -) -> None: - """Manage the host URL advertised to remote devices.""" - - if sum(bool(flag) for flag in (url is not None, auto, show)) != 1: - _exit_with("error: choose exactly one of , --auto, or --show") - if show: - result = _request("GET", "/app/settings/api/convey/host-url") - typer.echo(result["host_url"]) - return - if auto: - _request("POST", "/app/settings/api/convey/host-url", json_body={"auto": True}) - typer.echo("host url cleared. auto-detect is active.") - return - assert url is not None - try: - result = _request( - "POST", - "/app/settings/api/convey/host-url", - json_body={"url": url}, - ) - except ConveyClientError as err: - if err.reason_code == INVALID_CONFIG_VALUE.code and err.detail: - _exit_with(err.detail) - raise - typer.echo(f"host url set: {result['host_url']}") - - @convey_app.command("status") @convey_cli def convey_status() -> None: - """Show Convey network and host-URL status.""" + """Show Convey network and dashboard URL status.""" result = _request("GET", "/app/settings/api/convey/status") typer.echo(result["status_text"]) diff --git a/solstone/apps/settings/maint/008_migrate_pairing_home_address.py b/solstone/apps/settings/maint/008_migrate_pairing_home_address.py new file mode 100644 index 000000000..4a3d0ef08 --- /dev/null +++ b/solstone/apps/settings/maint/008_migrate_pairing_home_address.py @@ -0,0 +1,74 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Migrate legacy pairing host URLs to bare home addresses.""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import urlsplit + +from solstone.think.journal_config import ( + hold_config_lock, + read_journal_config, + write_journal_config, +) +from solstone.think.pairing.config import InvalidHomeAddress, validate_home_address +from solstone.think.utils import get_journal + + +def _legacy_host_url_to_home_address(value: Any) -> str | None: + if not isinstance(value, str): + return None + cleaned = value.strip() + if not cleaned: + return None + parsed = urlsplit(cleaned) + if parsed.scheme != "http" or not parsed.netloc: + return None + if parsed.username is not None or parsed.password is not None: + return None + if parsed.query or parsed.fragment or parsed.path not in ("", "/"): + return None + host = parsed.hostname + try: + port = parsed.port + except ValueError: + return None + if host is None or port is None: + return None + try: + return validate_home_address(f"{host}:{port}") + except InvalidHomeAddress: + return None + + +def migrate(config: dict[str, Any]) -> bool: + pairing = config.get("pairing") + if not isinstance(pairing, dict) or "host_url" not in pairing: + return False + + changed = False + home_address = _legacy_host_url_to_home_address(pairing.get("host_url")) + if home_address is not None and pairing.get("home_address") != home_address: + pairing["home_address"] = home_address + changed = True + + pairing.pop("host_url") + changed = True + return changed + + +def main() -> None: + journal = get_journal() + with hold_config_lock(journal): + config = read_journal_config(journal) + if not migrate(config): + print("Pairing home address already migrated.") + return + write_journal_config(config, journal) + print("Migrated pairing home address config.") + + +if __name__ == "__main__": + main() diff --git a/solstone/apps/settings/routes.py b/solstone/apps/settings/routes.py index 3be08b791..accc4d8af 100644 --- a/solstone/apps/settings/routes.py +++ b/solstone/apps/settings/routes.py @@ -496,64 +496,9 @@ def update_config() -> Any: return _settings_operation_failed() -def _host_url_status_value() -> str: - from solstone.think.pairing.config import get_host_url - - return get_host_url() - - -@settings_bp.route("/api/convey/host-url", methods=["GET", "POST"]) -def convey_host_url() -> Any: - """Read or update the host URL advertised to remote devices.""" - - from solstone.think.pairing.config import ( - InvalidHostUrl, - clear_host_url, - get_host_url, - set_host_url, - validate_host_url, - ) - - try: - if request.method == "GET": - return jsonify({"host_url": get_host_url()}) - - request_data = request.get_json() - if not isinstance(request_data, dict): - return error_response( - INVALID_REQUEST_VALUE, - detail="Expected JSON object with url or auto", - ) - - has_url = "url" in request_data and request_data.get("url") is not None - auto = bool(request_data.get("auto", False)) - if sum((has_url, auto)) != 1: - return error_response( - INVALID_REQUEST_VALUE, - detail="Provide exactly one of url or auto", - ) - - if auto: - clear_host_url() - return jsonify({"host_url": get_host_url(), "cleared": True}) - - raw_url = request_data.get("url") - if not isinstance(raw_url, str): - return error_response(INVALID_REQUEST_VALUE, detail="url must be a string") - try: - canonical = validate_host_url(raw_url) - except InvalidHostUrl as exc: - return error_response(INVALID_CONFIG_VALUE, detail=str(exc)) - set_host_url(canonical) - return jsonify({"host_url": canonical}) - except Exception: - logger.exception("error updating convey host url") - return _settings_operation_failed() - - @settings_bp.route("/api/convey/status") def convey_status() -> Any: - """Return formatted Convey bind and host URL status.""" + """Return formatted Convey bind and dashboard URL status.""" try: from solstone.convey.cli import _resolve_bind_host @@ -562,11 +507,12 @@ def convey_status() -> Any: bind_host = _resolve_bind_host() port = read_service_port("convey") or DEFAULT_SERVICE_PORT + dashboard_url = f"http://localhost:{port}" status_text = convey_copy.format_convey_status( bind=f"{bind_host}:{port}", - host_url=_host_url_status_value(), + dashboard_url=dashboard_url, ) - return jsonify({"status_text": status_text}) + return jsonify({"dashboard_url": dashboard_url, "status_text": status_text}) except Exception: logger.exception("error loading convey status") return _settings_operation_failed() diff --git a/solstone/apps/settings/tests/test_maint_008_migrate_pairing_home_address.py b/solstone/apps/settings/tests/test_maint_008_migrate_pairing_home_address.py new file mode 100644 index 000000000..2f7713eca --- /dev/null +++ b/solstone/apps/settings/tests/test_maint_008_migrate_pairing_home_address.py @@ -0,0 +1,67 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import importlib + +mod = importlib.import_module( + "solstone.apps.settings.maint.008_migrate_pairing_home_address" +) + + +def test_migration_moves_valid_legacy_host_url_to_home_address() -> None: + config = { + "pairing": { + "host_url": "http://192.168.1.44:7657", + "note": "preserve me", + }, + "unrelated": {"value": True}, + } + + assert mod.migrate(config) is True + + assert config == { + "pairing": { + "home_address": "192.168.1.44:7657", + "note": "preserve me", + }, + "unrelated": {"value": True}, + } + assert mod.migrate(config) is False + + +def test_migration_removes_invalid_legacy_values_without_home_address() -> None: + for legacy in ( + "http://localhost:7657", + "http://127.0.0.1:7657", + "http://192.168.1.44:5015", + "not a url", + None, + ): + config = {"pairing": {"host_url": legacy, "note": "preserve me"}} + + assert mod.migrate(config) is True + + assert config == {"pairing": {"note": "preserve me"}} + + +def test_migration_preserves_existing_new_key_when_legacy_invalid() -> None: + config = { + "pairing": { + "host_url": "http://127.0.0.1:7657", + "home_address": "192.168.1.44:7657", + } + } + + assert mod.migrate(config) is True + + assert config == {"pairing": {"home_address": "192.168.1.44:7657"}} + assert mod.migrate(config) is False + + +def test_migration_noops_without_legacy_key() -> None: + config = {"pairing": {"home_address": "192.168.1.44:7657"}} + + assert mod.migrate(config) is False + assert config == {"pairing": {"home_address": "192.168.1.44:7657"}} diff --git a/solstone/convey/copy.py b/solstone/convey/copy.py index 2db38f409..3333304b3 100644 --- a/solstone/convey/copy.py +++ b/solstone/convey/copy.py @@ -76,11 +76,11 @@ OBSERVER_CALLOSUM_LIVE_LABEL = "live" def format_convey_status( *, bind: str, - host_url: str, + dashboard_url: str, ) -> str: """Return the locked convey status block.""" - return f"convey\n bind: {bind}\n host url: {host_url}" + return f"convey\n bind: {bind}\n dashboard url: {dashboard_url}" __all__ = [ diff --git a/solstone/think/convey_client.py b/solstone/think/convey_client.py index 5edc6b6f9..ef1b19426 100644 --- a/solstone/think/convey_client.py +++ b/solstone/think/convey_client.py @@ -23,7 +23,6 @@ import requests import typer from requests.adapters import TimeoutSauce -from solstone.think.pairing.config import get_host_url_override from solstone.think.service import DEFAULT_SERVICE_PORT from solstone.think.utils import read_service_port, require_solstone @@ -81,9 +80,6 @@ class ConveyTimeoutError(ConveyClientError): def resolve_base_url() -> str: - override = get_host_url_override() - if override is not None: - return override port = read_service_port("convey") or DEFAULT_SERVICE_PORT return f"http://localhost:{port}" diff --git a/solstone/think/journal_default.json b/solstone/think/journal_default.json index fafc64eac..0a72e3f5f 100644 --- a/solstone/think/journal_default.json +++ b/solstone/think/journal_default.json @@ -73,7 +73,7 @@ } }, "pairing": { - "host_url": null + "home_address": null }, "backup": { "enabled": false, diff --git a/solstone/think/pairing/config.py b/solstone/think/pairing/config.py index 3b86ee7aa..9772daff7 100644 --- a/solstone/think/pairing/config.py +++ b/solstone/think/pairing/config.py @@ -1,27 +1,26 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc -"""Shared pairing host-address configuration.""" +"""Shared pairing home-address configuration.""" from __future__ import annotations import ipaddress from typing import Any -from urllib.parse import urlsplit from solstone.think.journal_config import write_journal_config -from solstone.think.service import DEFAULT_SERVICE_PORT -from solstone.think.utils import get_config, read_service_port +from solstone.think.link import interface_watcher +from solstone.think.utils import get_config -HOST_URL_INVALID = "enter an ipv4 address and port, like 192.168.1.44:7657" -HOST_URL_HOSTNAME_UNSUPPORTED = ( +HOME_ADDRESS_INVALID = "enter an ipv4 address and port, like 192.168.1.44:7657" +HOME_ADDRESS_HOSTNAME_UNSUPPORTED = ( "this needs an ip address — to reach home by name from anywhere, " "turn on your private network" ) -class InvalidHostUrl(Exception): - """Raised when a manual host URL cannot be normalized.""" +class InvalidHomeAddress(Exception): + """Raised when a manual home address cannot be normalized.""" def _pairing_config() -> dict[str, Any]: @@ -38,85 +37,79 @@ def _clean_str(value: Any) -> str | None: def _input_looks_like_hostname(host: str) -> bool: - return ":" not in host and any(char.isalpha() for char in host) + return any(char.isalpha() for char in host) and all( + char.isalnum() or char in ".-" for char in host + ) -def validate_host_url(value: str) -> str: - """Normalize a manual host URL to ``http://:``.""" +def is_usable_ipv4(value: Any) -> bool: + """Return whether value is a non-special IPv4 address usable for pairing.""" + + try: + ipv4 = ipaddress.IPv4Address(value) + except (TypeError, ValueError): + return False + return not ( + ipv4.is_loopback + or ipv4.is_unspecified + or ipv4.is_link_local + or ipv4.is_multicast + ) + + +def validate_home_address(value: str) -> str: + """Normalize a manual home address to ``:``.""" cleaned = value.strip() - if not cleaned: - raise InvalidHostUrl(HOST_URL_INVALID) - - candidate = cleaned if "://" in cleaned else f"http://{cleaned}" - parsed = urlsplit(candidate) - if parsed.scheme != "http" or not parsed.netloc: - raise InvalidHostUrl(HOST_URL_INVALID) - if parsed.username is not None or parsed.password is not None: - raise InvalidHostUrl(HOST_URL_INVALID) - if parsed.query or parsed.fragment: - raise InvalidHostUrl(HOST_URL_INVALID) - if parsed.path not in ("", "/"): - raise InvalidHostUrl(HOST_URL_INVALID) - - host = parsed.hostname or "" + if not cleaned or "://" in cleaned or "/" in cleaned: + raise InvalidHomeAddress(HOME_ADDRESS_INVALID) + + host, sep, port_text = cleaned.rpartition(":") + if sep != ":" or not host or not port_text: + if _input_looks_like_hostname(cleaned): + raise InvalidHomeAddress(HOME_ADDRESS_HOSTNAME_UNSUPPORTED) + raise InvalidHomeAddress(HOME_ADDRESS_INVALID) + try: ipv4 = ipaddress.IPv4Address(host) except ValueError as exc: if _input_looks_like_hostname(host): - raise InvalidHostUrl(HOST_URL_HOSTNAME_UNSUPPORTED) from exc - raise InvalidHostUrl(HOST_URL_INVALID) from exc + raise InvalidHomeAddress(HOME_ADDRESS_HOSTNAME_UNSUPPORTED) from exc + raise InvalidHomeAddress(HOME_ADDRESS_INVALID) from exc try: - port = parsed.port + port = int(port_text) except ValueError as exc: - raise InvalidHostUrl(HOST_URL_INVALID) from exc - if port is None or port < 1 or port > 65535: - raise InvalidHostUrl(HOST_URL_INVALID) + raise InvalidHomeAddress(HOME_ADDRESS_INVALID) from exc + if port != interface_watcher.LINK_DIRECT_PORT or not is_usable_ipv4(str(ipv4)): + raise InvalidHomeAddress(HOME_ADDRESS_INVALID) - return f"http://{ipv4}:{port}" - - -def get_host_url_override() -> str | None: - return _clean_str(_pairing_config().get("host_url")) - - -def override_host_port() -> str | None: - override = get_host_url_override() - if override is None: - return None - parsed = urlsplit(override) - return f"{parsed.hostname}:{parsed.port}" + return f"{ipv4}:{port}" -def get_host_url() -> str: - configured = get_host_url_override() - if configured is not None: - return configured - convey_port = read_service_port("convey") or DEFAULT_SERVICE_PORT - return f"http://localhost:{convey_port}" +def get_home_address() -> str | None: + return _clean_str(_pairing_config().get("home_address")) -def set_host_url(canonical: str) -> None: +def set_home_address(canonical: str) -> None: config = get_config() - config.setdefault("pairing", {})["host_url"] = canonical + config.setdefault("pairing", {})["home_address"] = canonical write_journal_config(config) -def clear_host_url() -> None: +def clear_home_address() -> None: config = get_config() - config.setdefault("pairing", {})["host_url"] = None + config.setdefault("pairing", {})["home_address"] = None write_journal_config(config) __all__ = [ - "HOST_URL_HOSTNAME_UNSUPPORTED", - "HOST_URL_INVALID", - "InvalidHostUrl", - "clear_host_url", - "get_host_url", - "get_host_url_override", - "override_host_port", - "set_host_url", - "validate_host_url", + "HOME_ADDRESS_HOSTNAME_UNSUPPORTED", + "HOME_ADDRESS_INVALID", + "InvalidHomeAddress", + "clear_home_address", + "get_home_address", + "is_usable_ipv4", + "set_home_address", + "validate_home_address", ] diff --git a/solstone/think/settings_cli.py b/solstone/think/settings_cli.py index edb16c428..982ebc374 100644 --- a/solstone/think/settings_cli.py +++ b/solstone/think/settings_cli.py @@ -11,7 +11,6 @@ import sys from solstone.convey.cli import _resolve_bind_host from solstone.convey.copy import format_convey_status -from solstone.think.pairing.config import get_host_url from solstone.think.service import DEFAULT_SERVICE_PORT from solstone.think.utils import ( read_service_port, @@ -19,17 +18,17 @@ from solstone.think.utils import ( ) -def _host_url_status_value() -> str: - return get_host_url() - - def _convey_port() -> int: return read_service_port("convey") or DEFAULT_SERVICE_PORT +def _dashboard_url() -> str: + return f"http://localhost:{_convey_port()}" + + def _status_payload() -> dict[str, str]: return { - "effective_host_url": get_host_url(), + "dashboard_url": _dashboard_url(), } @@ -43,7 +42,7 @@ def _print_status(*, as_json: bool) -> None: print( format_convey_status( bind=f"{bind_host}:{port}", - host_url=_host_url_status_value(), + dashboard_url=f"http://localhost:{port}", ) ) @@ -57,7 +56,7 @@ def main() -> None: status_parser = convey_subparsers.add_parser( "status", - help="Show convey bind and host-URL status", + help="Show convey bind and dashboard URL status", ) status_parser.add_argument( "--json", diff --git a/tests/test_convey_client.py b/tests/test_convey_client.py index ffd15d157..86c1b6b07 100644 --- a/tests/test_convey_client.py +++ b/tests/test_convey_client.py @@ -233,26 +233,21 @@ def test_non_envelope_error_raises_server_error_message() -> None: assert excinfo.value.error == SERVER_ERROR_MESSAGE -def test_resolve_base_url_uses_override(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr( - convey_client, - "get_host_url_override", - lambda: "http://192.168.1.44:5015", - ) +def test_resolve_base_url_ignores_pairing_home_address( + monkeypatch: pytest.MonkeyPatch, +) -> None: monkeypatch.setattr(convey_client, "read_service_port", lambda service: 5099) - assert resolve_base_url() == "http://192.168.1.44:5015" + assert resolve_base_url() == "http://localhost:5099" def test_resolve_base_url_uses_recorded_port(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(convey_client, "get_host_url_override", lambda: None) monkeypatch.setattr(convey_client, "read_service_port", lambda service: 5099) assert resolve_base_url() == "http://localhost:5099" def test_resolve_base_url_uses_default_port(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(convey_client, "get_host_url_override", lambda: None) monkeypatch.setattr(convey_client, "read_service_port", lambda service: None) assert resolve_base_url() == "http://localhost:5015" diff --git a/tests/test_link_call_parity.py b/tests/test_link_call_parity.py index d076c0fae..9135a7376 100644 --- a/tests/test_link_call_parity.py +++ b/tests/test_link_call_parity.py @@ -63,6 +63,8 @@ def _nonces() -> NonceStore: def _install_pair_watcher( monkeypatch: pytest.MonkeyPatch, endpoints: list[LocalEndpoint] | None = None, + *, + route_ipv4: str | None = "192.168.1.50", ) -> None: if endpoints is None: endpoints = [LocalEndpoint(ip="192.168.1.50", port=7657, scope="lan")] @@ -71,6 +73,7 @@ def _install_pair_watcher( "get_interface_watcher", lambda: _StubWatcher(endpoints), ) + monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: route_ipv4) def _add_device( @@ -347,7 +350,7 @@ def test_pair_reports_nonce_consumed_fallback(runner, monkeypatch): def test_pair_reports_no_lan_address_without_nonce(runner, monkeypatch): - _install_pair_watcher(monkeypatch, []) + _install_pair_watcher(monkeypatch, [], route_ipv4=None) result = runner.invoke( link_call.app, diff --git a/tests/test_pairing_config.py b/tests/test_pairing_config.py index 1bc1db846..70d875ef0 100644 --- a/tests/test_pairing_config.py +++ b/tests/test_pairing_config.py @@ -26,29 +26,31 @@ def test_pairing_config_defaults(journal_copy): payload["identity"] = {"name": "", "preferred": ""} _write_config(journal_copy, payload) - assert config.get_host_url() == "http://localhost:5015" + assert config.get_home_address() is None -def test_pairing_host_url_reads_trimmed_value(journal_copy): +def test_pairing_home_address_reads_trimmed_value(journal_copy): payload = _read_config(journal_copy) payload["pairing"] = { - "host_url": " http://192.168.1.44:6123 ", + "home_address": " 192.168.1.44:7657 ", } _write_config(journal_copy, payload) - assert config.get_host_url() == "http://192.168.1.44:6123" + assert config.get_home_address() == "192.168.1.44:7657" @pytest.mark.parametrize( ("raw", "expected"), [ - ("192.168.1.44:5015", "http://192.168.1.44:5015"), - (" http://192.168.1.44:5015 ", "http://192.168.1.44:5015"), - ("http://192.168.1.44:5015/", "http://192.168.1.44:5015"), + ("192.168.1.44:7657", "192.168.1.44:7657"), + (" 192.168.1.44:7657 ", "192.168.1.44:7657"), ], ) -def test_validate_host_url_accepts_ipv4_port(raw: str, expected: str) -> None: - assert config.validate_host_url(raw) == expected +def test_validate_home_address_accepts_ipv4_secure_port( + raw: str, + expected: str, +) -> None: + assert config.validate_home_address(raw) == expected @pytest.mark.parametrize( @@ -56,62 +58,79 @@ def test_validate_host_url_accepts_ipv4_port(raw: str, expected: str) -> None: [ "", " ", - "http://", + "http://192.168.1.44:7657", "192.168.1.44", "http://192.168.1.44", "192.168.1.44:0", + "192.168.1.44:5015", "192.168.1.44:65536", "192.168.1.44:notaport", - "https://192.168.1.44:5015", - "http://user@192.168.1.44:5015", - "http://192.168.1.44:5015/path", - "http://192.168.1.44:5015?x=1", - "http://192.168.1.44:5015#frag", - "http://[::1]:5015", - "http://[fe80::1]:5015", + "https://192.168.1.44:7657", + "user@192.168.1.44:7657", + "192.168.1.44:7657/path", + "192.168.1.44:7657?x=1", + "192.168.1.44:7657#frag", + "127.0.0.1:7657", + "0.0.0.0:7657", + "169.254.1.1:7657", + "224.0.0.1:7657", + "[::1]:7657", + "[fe80::1]:7657", ], ) -def test_validate_host_url_rejects_invalid_values(raw: str) -> None: - with pytest.raises(config.InvalidHostUrl) as excinfo: - config.validate_host_url(raw) +def test_validate_home_address_rejects_invalid_values(raw: str) -> None: + with pytest.raises(config.InvalidHomeAddress) as excinfo: + config.validate_home_address(raw) - assert str(excinfo.value) == config.HOST_URL_INVALID + assert str(excinfo.value) == config.HOME_ADDRESS_INVALID -@pytest.mark.parametrize("raw", ["mylab.local:5015", "http://home.local:5015"]) -def test_validate_host_url_rejects_hostname_with_sol_private_link_message( +@pytest.mark.parametrize("raw", ["mylab.local:7657", "home.local"]) +def test_validate_home_address_rejects_hostname_with_private_link_message( raw: str, ) -> None: - with pytest.raises(config.InvalidHostUrl) as excinfo: - config.validate_host_url(raw) + with pytest.raises(config.InvalidHomeAddress) as excinfo: + config.validate_home_address(raw) - assert str(excinfo.value) == config.HOST_URL_HOSTNAME_UNSUPPORTED + assert str(excinfo.value) == config.HOME_ADDRESS_HOSTNAME_UNSUPPORTED -def test_host_url_override_round_trip(journal_copy) -> None: - canonical = config.validate_host_url("192.168.1.44:5015") +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("192.168.1.44", True), + ("10.0.0.5", True), + ("127.0.0.1", False), + ("0.0.0.0", False), + ("169.254.1.1", False), + ("224.0.0.1", False), + ("::1", False), + ("not-an-ip", False), + (None, False), + ], +) +def test_is_usable_ipv4(value: object, expected: bool) -> None: + assert config.is_usable_ipv4(value) is expected - config.set_host_url(canonical) - assert _read_config(journal_copy)["pairing"]["host_url"] == canonical - assert config.get_host_url_override() == canonical - assert config.get_host_url() == canonical - assert config.override_host_port() == "192.168.1.44:5015" +def test_home_address_round_trip(journal_copy) -> None: + canonical = config.validate_home_address("192.168.1.44:7657") - config.clear_host_url() + config.set_home_address(canonical) - assert _read_config(journal_copy)["pairing"]["host_url"] is None - assert config.get_host_url_override() is None - assert config.override_host_port() is None + assert _read_config(journal_copy)["pairing"]["home_address"] == canonical + assert config.get_home_address() == canonical + config.clear_home_address() -def test_pairing_host_url_uses_localhost_without_manual_override(journal_copy): - payload = _read_config(journal_copy) - payload["pairing"] = {"host_url": None} - payload["convey"]["allow_network_access"] = True - _write_config(journal_copy, payload) - health_dir = journal_copy / "health" - health_dir.mkdir(parents=True, exist_ok=True) - (health_dir / "convey.port").write_text("6123", encoding="utf-8") + assert _read_config(journal_copy)["pairing"]["home_address"] is None + assert config.get_home_address() is None + + +def test_validate_home_address_rejects_without_writing(journal_copy) -> None: + before = _read_config(journal_copy) + + with pytest.raises(config.InvalidHomeAddress): + config.validate_home_address("192.168.1.44:5015") - assert config.get_host_url() == "http://localhost:6123" + assert _read_config(journal_copy) == before diff --git a/tests/test_settings_call_parity.py b/tests/test_settings_call_parity.py index 54733848f..e89a8948e 100644 --- a/tests/test_settings_call_parity.py +++ b/tests/test_settings_call_parity.py @@ -325,7 +325,7 @@ def test_identity_and_observer_setters(journal_copy: Path) -> None: } -def test_convey_status_host_url( +def test_convey_status_dashboard_url( journal_copy: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -334,49 +334,23 @@ def test_convey_status_host_url( assert status.stdout == ( "convey\n" " bind: 127.0.0.1:5015\n" - " host url: http://localhost:5015\n" + " dashboard url: http://localhost:5015\n" ) + config_before = _read_config(journal_copy) + manual_status = runner.invoke(settings_call.app, ["convey", "status"]) + assert manual_status.exit_code == 0 + assert "dashboard url: http://localhost:5015" in manual_status.stdout + assert _read_config(journal_copy) == config_before + calls = [] with monkeypatch.context() as m: m.setattr(settings_call, "get_client", lambda: calls.append("called")) - conflict = runner.invoke( - settings_call.app, - ["convey", "host-url", "--auto", "--show"], - ) - assert conflict.exit_code == 1 - assert conflict.stderr == "error: choose exactly one of , --auto, or --show\n" + missing = runner.invoke(settings_call.app, ["convey", "host-url"]) + assert missing.exit_code != 0 + assert "No such command" in missing.stderr assert calls == [] - set_url = runner.invoke( - settings_call.app, - ["convey", "host-url", "192.168.1.44:5015"], - ) - assert set_url.exit_code == 0 - assert set_url.stdout == "host url set: http://192.168.1.44:5015\n" - - show_url = runner.invoke(settings_call.app, ["convey", "host-url", "--show"]) - assert show_url.exit_code == 0 - assert show_url.stdout == "http://192.168.1.44:5015\n" - - manual_status = runner.invoke(settings_call.app, ["convey", "status"]) - assert manual_status.exit_code == 0 - assert "host url: http://192.168.1.44:5015" in manual_status.stdout - - auto = runner.invoke(settings_call.app, ["convey", "host-url", "--auto"]) - assert auto.exit_code == 0 - assert auto.stdout == "host url cleared. auto-detect is active.\n" - - bad_url = runner.invoke(settings_call.app, ["convey", "host-url", "/bad"]) - assert bad_url.exit_code == 1 - assert bad_url.stderr == "enter an ipv4 address and port, like 192.168.1.44:7657\n" - - bad_host = runner.invoke( - settings_call.app, - ["convey", "host-url", "mylab.local:5015"], - ) - assert bad_host.exit_code == 1 - assert bad_host.stderr == ( - "this needs an ip address — to reach home by name from anywhere, " - "turn on your private network\n" - ) + client = make_test_client(journal_copy) + response = client.get("/app/settings/api/convey/host-url") + assert response.status_code == 404 diff --git a/tests/test_settings_cli.py b/tests/test_settings_cli.py index fc346ebfb..3b65f319e 100644 --- a/tests/test_settings_cli.py +++ b/tests/test_settings_cli.py @@ -21,14 +21,14 @@ def test_status_json_does_not_require_running_stack( monkeypatch, capsys, ): - monkeypatch.setattr(settings_cli, "get_host_url", lambda: "http://localhost:5015") + monkeypatch.setattr(settings_cli, "read_service_port", lambda service: 5015) _run(monkeypatch, ["convey", "status", "--json"]) captured = capsys.readouterr() payload = json.loads(captured.out) assert payload == { - "effective_host_url": "http://localhost:5015", + "dashboard_url": "http://localhost:5015", } assert captured.err == ""