diff --git a/.gitea/workflows/testing.yml b/.gitea/workflows/testing.yml index ba0f40c..de50963 100644 --- a/.gitea/workflows/testing.yml +++ b/.gitea/workflows/testing.yml @@ -21,3 +21,27 @@ jobs: - name: Run tests run: uv run pytest + + remote-protocol-compat: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + browser-cli-client-version: + - "0.9.3" + - "0.9.5" + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Install dependencies + run: uv sync --group dev --managed-python + + - name: Run remote protocol compatibility matrix + env: + BROWSER_CLI_COMPAT_CLIENT_VERSION: ${{ matrix.browser-cli-client-version }} + run: uv run pytest tests/test_remote_protocol_matrix.py -v diff --git a/tests/test_remote_protocol_matrix.py b/tests/test_remote_protocol_matrix.py new file mode 100644 index 0000000..08c1f6f --- /dev/null +++ b/tests/test_remote_protocol_matrix.py @@ -0,0 +1,213 @@ +"""Remote protocol compatibility matrix for post-quantum transport. + +These tests exercise the wire-level combinations that matter for mixed +browser-cli versions without requiring a real browser. The native-host lookup is +mocked so successful auth/transport reaches the proxy layer and then returns the +expected "browser not connected" error. +""" +import contextlib +import io +import json +import os +import socket +import struct +import threading + +import pytest + +from browser_cli.auth import ( + generate_keypair, + load_private_key, + pq_decrypt, + pq_encrypt, + pq_kex_client_encapsulate, + pq_kex_server_decapsulate, + pq_kex_server_keypair, + sign, +) +from browser_cli.client import BrowserNotConnected, send_command +from browser_cli.commands.serve import _handle_client + +def _send_framed(sock: socket.socket, msg: dict) -> None: + payload = json.dumps(msg).encode("utf-8") + sock.sendall(struct.pack(" dict: + raw_len = b"" + while len(raw_len) < 4: + chunk = sock.recv(4 - len(raw_len)) + if not chunk: + raise ConnectionError("socket closed before response header") + raw_len += chunk + msg_len = struct.unpack(" tuple[dict, bytes]: + if "pq_kex" not in challenge: + pytest.skip("ML-KEM backend not available") + + ciphertext_hex, shared_secret = pq_kex_client_encapsulate(challenge["pq_kex"]["public_key"]) + clean_msg = { + **command_msg, + "pq_kex": {"alg": "ML-KEM-768", "ciphertext": ciphertext_hex}, + } + sig = sign(priv, bytes.fromhex(nonce_hex), clean_msg, shared_secret).hex() + if not encrypted: + return {**clean_msg, "pubkey": pub, "sig": sig}, shared_secret + + envelope = { + "id": clean_msg["id"], + "user_agent": clean_msg["user_agent"], + "pubkey": pub, + "sig": sig, + "pq_kex": clean_msg["pq_kex"], + "encrypted": pq_encrypt(shared_secret, "request", json.dumps(clean_msg).encode("utf-8")), + } + return envelope, shared_secret + +def _assert_browser_not_connected(resp: dict) -> None: + assert resp.get("success") is False + assert "browser" in resp.get("error", "").lower() or "connected" in resp.get("error", "").lower() + +def test_real_mlkem_primitive_roundtrip(): + keypair = pq_kex_server_keypair() + if keypair is None: + pytest.skip("ML-KEM backend not available") + private_key, public_key = keypair + + ciphertext_hex, client_secret = pq_kex_client_encapsulate(public_key.hex()) + server_secret = pq_kex_server_decapsulate(private_key, ciphertext_hex) + + assert server_secret == client_secret + +@pytest.mark.parametrize( + ("client_version", "encrypted", "expect_encrypted_response"), + [ + ("0.9.3", False, False), # legacy client stays compatible + ("0.9.5", True, True), # current client must use encrypted transport + ], +) +def test_remote_protocol_version_matrix(auth_material, client_version, encrypted, expect_encrypted_response): + selected_version = os.environ.get("BROWSER_CLI_COMPAT_CLIENT_VERSION") + if selected_version and selected_version != client_version: + pytest.skip(f"compat matrix selected {selected_version}") + + _key_path, auth_path, priv, pub = auth_material + client, thread, challenge = _connect(auth_path) + + msg = { + "id": f"tabs-{client_version}", + "command": "tabs.list", + "args": {}, + "user_agent": f"browser-cli/{client_version}", + } + wire_msg, shared_secret = _pq_auth_message(priv, pub, challenge["nonce"], msg, challenge, encrypted=encrypted) + _send_framed(client, wire_msg) + resp = _recv_framed(client) + + if expect_encrypted_response: + assert set(resp) == {"encrypted"} + resp = json.loads(pq_decrypt(shared_secret, "response", resp["encrypted"])) + else: + assert "encrypted" not in resp + + _assert_browser_not_connected(resp) + client.close() + thread.join(timeout=2) + +def test_current_client_plaintext_transport_is_rejected(auth_material): + _key_path, auth_path, priv, pub = auth_material + client, thread, challenge = _connect(auth_path) + + msg = { + "id": "new-plain", + "command": "tabs.list", + "args": {}, + "user_agent": "browser-cli/0.9.5", + } + wire_msg, _shared_secret = _pq_auth_message(priv, pub, challenge["nonce"], msg, challenge, encrypted=False) + _send_framed(client, wire_msg) + resp = _recv_framed(client) + + assert resp.get("success") is False + assert "encrypted transport" in resp.get("error", "").lower() + client.close() + thread.join(timeout=2) + +def test_send_command_uses_encrypted_remote_transport(auth_material): + key_path, auth_path, _priv, _pub = auth_material + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen(1) + host, port = server.getsockname() + + def _accept_once(): + conn, addr = server.accept() + _handle_client(conn, addr, None, auth_path) + server.close() + + thread = threading.Thread(target=_accept_once, daemon=True) + thread.start() + + with pytest.raises(RuntimeError, match="browser|connected"): + send_command("tabs.list", remote=f"{host}:{port}", profile="default", key=key_path) + + thread.join(timeout=2) + +def test_no_mlkem_backend_falls_back_and_client_warns(auth_material, monkeypatch): + key_path, auth_path, _priv, _pub = auth_material + monkeypatch.setattr("browser_cli.auth.pq_kex_server_keypair", lambda: None) + + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen(1) + host, port = server.getsockname() + + def _accept_once(): + conn, addr = server.accept() + _handle_client(conn, addr, None, auth_path) + server.close() + + thread = threading.Thread(target=_accept_once, daemon=True) + thread.start() + + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + with pytest.raises(RuntimeError, match="browser|connected"): + send_command("tabs.list", remote=f"{host}:{port}", profile="default", key=key_path) + + assert "not using a post-quantum key exchange" in stderr.getvalue() + thread.join(timeout=2)