diff --git a/solstone/apps/network/routes.py b/solstone/apps/network/routes.py index 64dd5dadc..22a115fe1 100644 --- a/solstone/apps/network/routes.py +++ b/solstone/apps/network/routes.py @@ -65,6 +65,7 @@ from solstone.convey.reasons import ( INVALID_CONFIG_VALUE, INVALID_OPERATION_FOR_STATE, INVALID_REQUEST_VALUE, + LOCAL_REQUEST_ONLY, MISSING_REQUIRED_FIELD, OPERATION_NO_LONGER_AVAILABLE, PAIRED_DEVICE_NOT_FOUND, @@ -184,8 +185,13 @@ def _rough_network(mode: str) -> str: return "anywhere" if mode == "pl-via-spl" else "network" -def _is_loopback_request() -> bool: - return request.remote_addr in {"127.0.0.1", "::1"} +def _is_hardened_loopback_request() -> bool: + if request.remote_addr not in {"127.0.0.1", "::1"}: + return False + return not any( + request.headers.get(header) + for header in ("X-Forwarded-For", "X-Real-IP", "X-Forwarded-Host") + ) def _read_link_health() -> dict[str, Any] | None: @@ -656,7 +662,7 @@ def set_home_address_route() -> Any: @network_bp.get("/local-endpoints") def local_endpoints() -> Any: - if not _is_loopback_request(): + if not _is_hardened_loopback_request(): abort(404) response = LocalEndpointsResponse( v=1, @@ -667,6 +673,15 @@ def local_endpoints() -> Any: return jsonify(response_to_dict(response)) +def _same_machine_requested(payload: dict[str, Any]) -> bool | None: + value = payload.get("same_machine") + if value is None: + return False + if isinstance(value, bool): + return value + return None + + # --------------------------------------------------------------------------- # pair ceremony # --------------------------------------------------------------------------- @@ -688,6 +703,33 @@ def pair_start() -> Any: if not isinstance(role, str) or role not in VALID_ROLES: return error_response(PAIRING_REQUEST_INVALID, detail="invalid role") + same_machine = _same_machine_requested(payload) + if same_machine is None: + return error_response( + PAIRING_REQUEST_INVALID, + detail="same_machine must be boolean", + ) + if same_machine: + if not _is_hardened_loopback_request(): + return error_response(LOCAL_REQUEST_ONLY) + ca_fp = _ca_fingerprint() + port = _secure_listener_port() + nonce = generate_nonce() + pair_link = _build_pair_link("127.0.0.1", port, nonce, ca_fp) + _nonces().add( + nonce, + device_label, + role=role, + ) + response = PairStartResponse( + nonce=nonce, + pair_link=pair_link, + expires_in=300, + device_label=device_label, + ca_fingerprint=ca_fp, + ) + return _jsonify_preserving_order(asdict(response)) + if read_posture() == "spl": service_token = load_service_token() if service_token is None: diff --git a/solstone/apps/network/tests/test_pair_start.py b/solstone/apps/network/tests/test_pair_start.py index dd5843ade..7aeb88d51 100644 --- a/solstone/apps/network/tests/test_pair_start.py +++ b/solstone/apps/network/tests/test_pair_start.py @@ -9,6 +9,8 @@ import ipaddress import json import re +import pytest + 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 @@ -26,6 +28,28 @@ PAIR_START_KEYS = [ ] +FIXED_NONCE = "11" * 16 +FIXED_SPL_NONCE = bytes.fromhex("0102030405060708") + + +def _set_home_address(env, value: str) -> None: + config_path = env.journal / "config" / "journal.json" + config = json.loads(config_path.read_text("utf-8")) + config["pairing"] = {"home_address": value} + config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") + + +def _decoded_pair_link_port(pair_link: str) -> int: + decoded = _decode_pair_link(pair_link) + return int.from_bytes(decoded[6:8], "big") + + +def _assert_single_pair_link_address(pair_link: str, address: str) -> None: + decoded = _decode_pair_link(pair_link) + assert decoded[0:2] == b"\x04\x01" + assert decoded[2:6] == ipaddress.IPv4Address(address).packed + + def test_pair_start_shape_and_locked_order(link_env) -> None: env = link_env() @@ -49,6 +73,116 @@ def test_pair_start_shape_and_locked_order(link_env) -> None: assert "qr_payload" not in payload +@pytest.mark.parametrize("remote_addr", ("127.0.0.1", "::1")) +def test_pair_start_same_machine_loopback_direct_link_wins_over_home_address( + link_env, + remote_addr, +) -> None: + env = link_env() + _set_home_address(env, "192.0.2.44:7657") + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone", "same_machine": True}, + environ_base={"REMOTE_ADDR": remote_addr}, + ) + + assert response.status_code == 200 + payload = response.get_json() + ca = load_or_generate_ca(ca_dir()) + assert list(payload.keys()) == PAIR_START_KEYS + assert payload["ca_fingerprint"] == ca.fingerprint_sha256() + _assert_single_pair_link_address(payload["pair_link"], "127.0.0.1") + assert _decoded_pair_link_port(payload["pair_link"]) == ( + link_routes._secure_listener_port() + ) + snap = link_routes._nonces().snapshot() + assert len(snap) == 1 + assert snap[0].value == payload["nonce"] + assert snap[0].device_label == "Test Phone" + + +@pytest.mark.parametrize( + "kwargs", + ( + pytest.param( + {"environ_base": {"REMOTE_ADDR": "192.168.1.5"}}, + id="non_loopback", + ), + pytest.param({"headers": {"X-Forwarded-For": "1.2.3.4"}}, id="forwarded_for"), + pytest.param({"headers": {"X-Real-IP": "1.2.3.4"}}, id="real_ip"), + pytest.param( + {"headers": {"X-Forwarded-Host": "example.test"}}, + id="forwarded_host", + ), + ), +) +def test_pair_start_same_machine_requires_hardened_loopback( + link_env, + kwargs, +) -> None: + env = link_env() + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone", "same_machine": True}, + **kwargs, + ) + + assert response.status_code == 403 + assert response.get_json()["reason_code"] == "local_request_only" + assert link_routes._nonces().snapshot() == [] + + +@pytest.mark.parametrize("same_machine", ("true", 1, [], {})) +def test_pair_start_same_machine_non_boolean_rejected_without_nonce( + link_env, + same_machine, +) -> None: + env = link_env() + + response = env.client.post( + "/app/network/pair-start", + json={"same_machine": same_machine}, + ) + + assert response.status_code == 400 + assert response.get_json()["reason_code"] == "pairing_request_invalid" + assert link_routes._nonces().snapshot() == [] + + +def test_pair_start_same_machine_returns_before_spl_posture_reads( + link_env, + monkeypatch, +) -> None: + env = link_env() + monkeypatch.setattr( + link_routes, + "read_posture", + lambda: (_ for _ in ()).throw(RuntimeError("posture read")), + ) + monkeypatch.setattr( + link_routes, + "load_service_token", + lambda: (_ for _ in ()).throw(RuntimeError("service token")), + ) + monkeypatch.setattr( + link_routes, + "start_pair_window", + lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("pair window")), + ) + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone", "same_machine": True}, + ) + + assert response.status_code == 200 + payload = response.get_json() + _assert_single_pair_link_address(payload["pair_link"], "127.0.0.1") + assert len(link_routes._nonces().snapshot()) == 1 + + def test_pair_start_omitted_assigned_label_stores_empty(link_env) -> None: env = link_env() @@ -279,6 +413,106 @@ def test_pair_start_detected_order_matches_api_status( assert pair_addresses == status_addresses +def test_pair_start_without_same_machine_from_loopback_uses_lan_candidate( + link_env, +) -> None: + env = link_env() + + response = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + + assert response.status_code == 200 + _assert_single_pair_link_address(response.get_json()["pair_link"], "192.168.1.50") + + +def test_pair_start_false_same_machine_matches_home_override_path( + link_env, + monkeypatch, +) -> None: + env = link_env() + _set_home_address(env, "192.0.2.44:7657") + monkeypatch.setattr(link_routes, "generate_nonce", lambda: FIXED_NONCE) + + absent = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + false = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone", "same_machine": False}, + ) + + assert absent.status_code == false.status_code == 200 + assert absent.get_json() == false.get_json() + + +def test_pair_start_false_same_machine_matches_lan_candidate_path( + link_env, + monkeypatch, +) -> None: + env = link_env() + monkeypatch.setattr(link_routes, "generate_nonce", lambda: FIXED_NONCE) + + absent = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + false = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone", "same_machine": False}, + ) + + assert absent.status_code == false.status_code == 200 + assert absent.get_json() == false.get_json() + + +def test_pair_start_false_same_machine_matches_no_candidates_refusal( + link_env, + monkeypatch, +) -> None: + env = link_env(local_endpoints=[]) + monkeypatch.setattr(link_routes, "_detect_lan_ip", lambda: None) + + absent = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + false = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone", "same_machine": False}, + ) + + assert absent.status_code == false.status_code == 400 + assert absent.get_json() == false.get_json() + assert absent.get_json()["reason_code"] == "pairing_request_invalid" + + +def test_pair_start_false_same_machine_matches_spl_path( + link_env, + monkeypatch, +) -> None: + env = link_env(posture="spl", service_token="svc") + monkeypatch.setattr( + link_routes, + "generate_pair_window_nonce", + lambda: FIXED_SPL_NONCE, + ) + + absent = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone"}, + ) + false = env.client.post( + "/app/network/pair-start", + json={"device_label": "Test Phone", "same_machine": False}, + ) + + assert absent.status_code == false.status_code == 200 + assert absent.get_json() == false.get_json() + + def _fragment(pair_link: str) -> str: return pair_link.rsplit("#", 1)[1] diff --git a/tests/link/test_local_endpoints_route.py b/tests/link/test_local_endpoints_route.py index 04a8e290a..d9c824e15 100644 --- a/tests/link/test_local_endpoints_route.py +++ b/tests/link/test_local_endpoints_route.py @@ -85,3 +85,16 @@ def test_local_endpoints_non_loopback_404(link_client) -> None: ) assert response.status_code == 404 + + +@pytest.mark.parametrize( + "header", + ("X-Forwarded-For", "X-Real-IP", "X-Forwarded-Host"), +) +def test_local_endpoints_proxy_headers_404(link_client, header: str) -> None: + response = link_client.get( + "/app/network/local-endpoints", + headers={header: "1.2.3.4"}, + ) + + assert response.status_code == 404 diff --git a/tests/link/test_network_alias_routes.py b/tests/link/test_network_alias_routes.py index 805bbd29c..55969a380 100644 --- a/tests/link/test_network_alias_routes.py +++ b/tests/link/test_network_alias_routes.py @@ -70,6 +70,23 @@ def test_legacy_link_prefix_serves_native_client_routes( assert client.get("/app/link/local-endpoints").status_code == 200 +@pytest.mark.parametrize( + "header", + ("X-Forwarded-For", "X-Real-IP", "X-Forwarded-Host"), +) +def test_legacy_link_local_endpoints_proxy_headers_404( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + header: str, +) -> None: + app, _journal = make_convey_app(tmp_path, monkeypatch, link={"posture": "spl"}) + client = app.test_client() + + response = client.get("/app/link/local-endpoints", headers={header: "1.2.3.4"}) + + assert response.status_code == 404 + + def _full_rule(prefix: str, suffix: str) -> str: return f"{prefix}/" if suffix == "/" else f"{prefix}{suffix}"