Add Desktop OAuth client register + verify

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crdt-sync/crdt_sync/tests/test_tool_client_probe.py

@@ -0,0 +1,120 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Tests for ``tool/client_probe``: the click-free "does the client work" check."""
+
+from __future__ import annotations
+
+import base64
+from unittest.mock import MagicMock
+
+import requests
+
+from tool import client_probe as probe
+from tool.desktop_client import DesktopClient
+
+_HELD = "stored-client-credential"
+_CLIENT = DesktopClient(
+    client_id="1-abc.apps.googleusercontent.com",
+    client_secret=_HELD,
+    project_id="kuhy-syncs",
+)
+
+
+def _location(payload: bytes) -> str:
+    encoded = base64.urlsafe_b64encode(payload).decode().rstrip("=")
+    return f"https://accounts.google.com/signin/oauth/error?authError={encoded}&x=1"
+
+
+def test_auth_location_without_error_is_ok() -> None:
+    result = probe.classify_auth_location("https://accounts.google.com/v3/signin")
+    assert result.ok
+
+
+def test_auth_location_none_is_ok() -> None:
+    assert probe.classify_auth_location(None).ok
+
+
+def test_auth_error_is_decoded() -> None:
+    location = _location(b"\n\x0finvalid_request\x12\x02 Custom URI scheme is bad")
+    result = probe.classify_auth_location(location)
+    assert not result.ok
+    assert "invalid_request" in result.detail
+    assert "Custom URI scheme" in result.detail
+
+
+def test_undecodable_auth_error_is_still_a_failure() -> None:
+    result = probe.classify_auth_location(
+        "https://accounts.google.com/signin/oauth/error?authError=!!!"
+    )
+    assert not result.ok
+    assert "authError=!!!" in result.detail
+
+
+def test_token_invalid_grant_means_client_and_secret_are_good() -> None:
+    assert probe.classify_token(400, {"error": "invalid_grant"}).ok
+
+
+def test_token_client_not_found_means_still_propagating() -> None:
+    result = probe.classify_token(
+        401,
+        {
+            "error": "invalid_client",
+            "error_description": "The OAuth client was not found.",
+        },
+    )
+    assert not result.ok
+    assert "--verify-only" in result.detail
+
+
+def test_token_invalid_client_means_wrong_secret() -> None:
+    result = probe.classify_token(
+        401, {"error": "invalid_client", "error_description": "Unauthorized"}
+    )
+    assert not result.ok
+    assert "secret" in result.detail
+
+
+def test_token_other_error_is_reported_verbatim() -> None:
+    result = probe.classify_token(500, {"error": "backend_error"})
+    assert not result.ok
+    assert "backend_error" in result.detail
+
+
+def test_token_non_json_body_is_reported() -> None:
+    result = probe.classify_token(502, None)
+    assert not result.ok
+    assert "502" in result.detail
+
+
+def _fakes(location: str | None, status: int, body: object) -> tuple:
+    get = MagicMock(
+        return_value=MagicMock(headers={"Location": location} if location else {})
+    )
+    post_response = MagicMock(status_code=status)
+    if body is None:
+        post_response.json.side_effect = ValueError
+    else:
+        post_response.json.return_value = body
+    return get, MagicMock(return_value=post_response)
+
+
+def test_probe_client_runs_both_checks() -> None:
+    get, post = _fakes(None, 400, {"error": "invalid_grant"})
+    results = probe.probe_client(_CLIENT, get=get, post=post)
+    assert [r.ok for r in results] == [True, True]
+    assert get.call_args.kwargs["allow_redirects"] is False
+    assert post.call_args.kwargs["data"]["client_secret"] == _HELD
+
+
+def test_probe_client_turns_network_errors_into_failures() -> None:
+    get = MagicMock(side_effect=requests.ConnectionError("down"))
+    post = MagicMock(side_effect=requests.ConnectionError("down"))
+    results = probe.probe_client(_CLIENT, get=get, post=post)
+    assert [r.ok for r in results] == [False, False]
+    assert "down" in results[0].detail
+
+
+def test_probe_client_tolerates_a_non_json_token_reply() -> None:
+    get, post = _fakes(None, 502, None)
+    results = probe.probe_client(_CLIENT, get=get, post=post)
+    assert results[1].ok is False
+    assert "502" in results[1].detail

crdt-sync/crdt_sync/tests/test_tool_desktop_client.py

@@ -0,0 +1,186 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Tests for ``tool/desktop_client``: picking up and validating the client JSON.
+
+The Android client JSON Google offers for download has the same top-level
+``installed`` key and project as a Desktop one, so the negative fixture here is
+that exact shape: it must be refused, or the wrong client gets stored.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+from typing import TYPE_CHECKING
+
+import pytest
+
+from tool import desktop_client as dc
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+_CLIENT_ID = "1-abc.apps.googleusercontent.com"
+_HELD = "the-desktop-client-credential"
+
+
+def _installed(**overrides: object) -> dict[str, object]:
+    body: dict[str, object] = {
+        "client_id": _CLIENT_ID,
+        "project_id": "kuhy-syncs",
+        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+        "token_uri": "https://oauth2.googleapis.com/token",
+        "client_secret": _HELD,
+        "redirect_uris": ["http://localhost"],
+    }
+    body.update(overrides)
+    return body
+
+
+def _text(**overrides: object) -> str:
+    return json.dumps({"installed": _installed(**overrides)})
+
+
+def _android_text() -> str:
+    body = _installed()
+    del body["client_secret"]
+    del body["redirect_uris"]
+    return json.dumps({"installed": body})
+
+
+def test_valid_desktop_json_parses() -> None:
+    client = dc.parse_client_json(_text())
+    assert client == dc.DesktopClient(_CLIENT_ID, _HELD, "kuhy-syncs")
+
+
+@pytest.mark.parametrize(
+    ("text", "needle"),
+    [
+        ("{not json", "not valid JSON"),
+        ("[]", "JSON object"),
+        (json.dumps({"web": _installed()}), "'installed'"),
+        (_text(project_id="other"), "other"),
+        (_text(client_id="plain"), "client_id"),
+        (_android_text(), "client_secret"),
+        (_text(redirect_uris=["https://x.example"]), "http://localhost"),
+        (json.dumps({"installed": "nope"}), "'installed'"),
+    ],
+)
+def test_invalid_json_is_refused_with_a_reason(text: str, needle: str) -> None:
+    with pytest.raises(dc.ClientFileError, match=needle):
+        dc.parse_client_json(text)
+
+
+def _touch(directory: Path, name: str, text: str, mtime: float) -> Path:
+    path = directory / name
+    path.write_text(text, encoding="utf-8")
+    os.utime(path, (mtime, mtime))
+    return path
+
+
+def _wait(directory: Path, since: float, skipped: list[str]) -> dc.Pickup | None:
+    return dc.wait_for_client_json(
+        directory, since, attempts=3, pause=lambda: None, report=skipped.append
+    )
+
+
+def test_pickup_ignores_old_files_and_the_android_json(tmp_path: Path) -> None:
+    _touch(tmp_path, "client_secret_old.json", _text(), 50.0)
+    _touch(tmp_path, "client_secret_android.json", _android_text(), 150.0)
+    _touch(tmp_path, "notes.json", _text(), 150.0)
+    skipped: list[str] = []
+    assert _wait(tmp_path, 100.0, skipped) is None
+    assert len(skipped) == 1
+    assert "client_secret_android.json" in skipped[0]
+    assert "client_secret" in skipped[0]
+
+
+def test_pickup_takes_the_newest_valid_file(tmp_path: Path) -> None:
+    _touch(
+        tmp_path,
+        "client_secret_a.json",
+        _text(client_id="1-a.apps.googleusercontent.com"),
+        120.0,
+    )
+    newest = _touch(tmp_path, "client_secret_b.json", _text(), 130.0)
+    found = _wait(tmp_path, 100.0, [])
+    assert found is not None
+    assert found.path == newest
+    assert found.client.client_id == _CLIENT_ID
+
+
+def test_pickup_waits_for_the_file_to_appear(tmp_path: Path) -> None:
+    def pause() -> None:
+        _touch(tmp_path, "client_secret_late.json", _text(), 500.0)
+
+    found = dc.wait_for_client_json(tmp_path, 100.0, attempts=3, pause=pause)
+    assert found is not None
+
+
+def test_pickup_pauses_between_attempts_only(tmp_path: Path) -> None:
+    pauses: list[int] = []
+    result = dc.wait_for_client_json(
+        tmp_path, 0.0, attempts=3, pause=lambda: pauses.append(1)
+    )
+    assert result is None
+    assert len(pauses) == 2
+
+
+def test_store_client_is_private_and_consumes_the_download(tmp_path: Path) -> None:
+    source = _touch(tmp_path, "client_secret_x.json", _text(), 1.0)
+    dest = tmp_path / "conf" / "desktop_client.json"
+    dc.store_client(source, dest)
+    assert stat.S_IMODE(dest.stat().st_mode) == 0o600
+    assert dc.load_client(dest).client_id == _CLIENT_ID
+    assert not source.exists()
+
+
+def test_store_client_overwrites_and_tightens_an_existing_file(tmp_path: Path) -> None:
+    dest = tmp_path / "desktop_client.json"
+    dest.write_text("old", encoding="utf-8")
+    dest.chmod(0o644)
+    dc.store_client(_touch(tmp_path, "client_secret_y.json", _text(), 1.0), dest)
+    assert stat.S_IMODE(dest.stat().st_mode) == 0o600
+    assert dest.read_text(encoding="utf-8") == _text()
+
+
+def test_load_client_points_at_the_register_script_when_missing(tmp_path: Path) -> None:
+    with pytest.raises(dc.ClientFileError, match=r"register_desktop_client\.sh"):
+        dc.load_client(tmp_path / "absent.json")
+
+
+def test_main_pickup_stores_the_client(tmp_path: Path) -> None:
+    _touch(tmp_path, "client_secret_z.json", _text(), 200.0)
+    dest = tmp_path / "out" / "desktop_client.json"
+    code = dc.main(
+        [
+            "pickup",
+            "--since",
+            "100",
+            "--timeout",
+            "5",
+            "--downloads",
+            str(tmp_path),
+            "--dest",
+            str(dest),
+        ]
+    )
+    assert code == 0
+    assert dc.load_client(dest).project_id == "kuhy-syncs"
+
+
+def test_main_pickup_times_out(tmp_path: Path) -> None:
+    code = dc.main(
+        [
+            "pickup",
+            "--since",
+            "100",
+            "--timeout",
+            "0",
+            "--downloads",
+            str(tmp_path),
+            "--dest",
+            str(tmp_path / "d.json"),
+        ]
+    )
+    assert code == 1

crdt-sync/crdt_sync/tests/test_tool_pkce.py

@@ -0,0 +1,64 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Tests for ``tool/_pkce``: the verifier/challenge pair, consent URL, browser."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+import urllib.parse
+
+from tool import _pkce
+
+# The worked example from RFC 7636 appendix B.
+_RFC_VERIFIER = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
+_RFC_CHALLENGE = "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
+
+
+def test_challenge_matches_the_rfc_vector() -> None:
+    assert _pkce.challenge_for(_RFC_VERIFIER) == _RFC_CHALLENGE
+
+
+def test_make_pkce_is_random_and_self_consistent() -> None:
+    first, second = _pkce.make_pkce(), _pkce.make_pkce()
+    assert first.verifier != second.verifier
+    assert 43 <= len(first.verifier) <= 128
+    assert first.challenge == _pkce.challenge_for(first.verifier)
+
+
+def test_consent_url_carries_every_pkce_field() -> None:
+    url = _pkce.build_consent_url(
+        client_id="cid",
+        redirect_uri="http://127.0.0.1:5555",
+        challenge="chal",
+        state="st",
+        login_hint="me@example.com",
+    )
+    parsed = urllib.parse.urlparse(url)
+    query = {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query).items()}
+    assert parsed.netloc == "accounts.google.com"
+    assert query == {
+        "client_id": "cid",
+        "redirect_uri": "http://127.0.0.1:5555",
+        "response_type": "code",
+        "scope": "openid email profile",
+        "state": "st",
+        "code_challenge": "chal",
+        "code_challenge_method": "S256",
+        "login_hint": "me@example.com",
+        "prompt": "select_account",
+    }
+
+
+def test_open_detached_spawns_a_new_session() -> None:
+    popen = MagicMock()
+    which = MagicMock(side_effect=lambda name: f"/usr/bin/{name}")
+    assert _pkce.open_detached("https://x", popen=popen, which=which) is True
+    args, kwargs = popen.call_args
+    assert args[0] == ["/usr/bin/setsid", "/usr/bin/xdg-open", "https://x"]
+    assert kwargs["start_new_session"] is True
+
+
+def test_open_detached_reports_missing_tools() -> None:
+    popen = MagicMock()
+    which = MagicMock(return_value=None)
+    assert _pkce.open_detached("https://x", popen=popen, which=which) is False
+    popen.assert_not_called()

crdt-sync/crdt_sync/tests/test_tool_verify_desktop_client.py

@@ -0,0 +1,156 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Tests for ``tool/verify_desktop_client``: the PKCE loopback + Firebase check.
+
+No browser, no network, no socket: the consent step, the code exchange and
+Firebase's REST endpoint are all faked at the module boundary.
+"""
+
+from __future__ import annotations
+
+import base64
+import contextlib
+import json
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from crdt_sync._autherrors import FirebaseAuthError
+from crdt_sync._config import FirebaseConfig
+from tool import verify_desktop_client as vdc
+from tool._token_exchange import TokenError, _exchange_code
+from tool.desktop_client import DesktopClient
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+_HELD = "stored-client-credential"
+_CLIENT = DesktopClient("1-abc.apps.googleusercontent.com", _HELD, "kuhy-syncs")
+_CONFIG = FirebaseConfig("api", "https://db", "kuhy-syncs", "uid-1", "me@example.com")
+_ID_VALUE = (
+    "header."
+    + base64.urlsafe_b64encode(json.dumps({"aud": _CLIENT.client_id}).encode())
+    .decode()
+    .rstrip("=")
+    + ".sig"
+)
+
+
+def test_audience_of_reads_the_aud_claim() -> None:
+    assert vdc.audience_of(_ID_VALUE) == _CLIENT.client_id
+
+
+@pytest.mark.parametrize(
+    "bad", ["", "a.b", "a.!!!.c", "a." + base64.b64encode(b"[]").decode() + ".c"]
+)
+def test_audience_of_survives_garbage(bad: str) -> None:
+    assert vdc.audience_of(bad) == "?"
+
+
+@pytest.mark.parametrize(
+    ("message", "needle"),
+    [
+        (
+            "failed: HTTP 400 (INVALID_IDP_RESPONSE : id_token audience mismatch)",
+            "Safelist",
+        ),
+        ("signed in as the wrong account: Google resolved to uid 'x'", "login_hint"),
+        ("anything else", ""),
+    ],
+)
+def test_hint_for(message: str, needle: str) -> None:
+    hint = vdc.hint_for(message)
+    assert (needle in hint) if needle else hint == ""
+
+
+@contextlib.contextmanager
+def _flow(result: dict[str, str]) -> Iterator[MagicMock]:
+    """Fake the socket, the callback and the exchange; yield the exchange."""
+    with (
+        patch.object(vdc, "_free_port", return_value=5555),
+        patch.object(
+            vdc, "_start_callback_server", return_value=(MagicMock(), MagicMock())
+        ),
+        patch.object(vdc._CallbackHandler, "result", result),
+        patch.object(vdc, "_exchange_code", return_value=_ID_VALUE) as exchange,
+        patch.object(
+            vdc.secrets,
+            "token_urlsafe",
+            side_effect=lambda nbytes: "st" if nbytes == 16 else "v" * 86,
+        ),
+    ):
+        yield exchange
+
+
+def test_fetch_id_token_sends_pkce_and_the_secret() -> None:
+    opener = MagicMock(return_value=True)
+    with _flow({"code": "abc", "state": "st"}) as exchange:
+        token = vdc.fetch_id_token(_CLIENT, "me@example.com", opener=opener)
+    assert token == _ID_VALUE
+    url = opener.call_args.args[0]
+    assert "redirect_uri=http%3A%2F%2F127.0.0.1%3A5555" in url
+    assert "code_challenge_method=S256" in url
+    code, client_id, held, redirect, verifier = exchange.call_args.args
+    assert (code, client_id, held) == ("abc", _CLIENT.client_id, _HELD)
+    assert redirect == "http://127.0.0.1:5555"
+    assert len(verifier) >= 43
+
+
+def test_fetch_id_token_without_a_browser_still_waits() -> None:
+    opener = MagicMock(return_value=False)
+    with _flow({"code": "abc", "state": "st"}):
+        token = vdc.fetch_id_token(_CLIENT, "", open_browser=False, opener=opener)
+    assert token == _ID_VALUE
+    opener.assert_not_called()
+
+
+def test_fetch_id_token_reports_a_refused_consent() -> None:
+    with (
+        _flow({"error": "access_denied"}),
+        pytest.raises(TokenError, match="access_denied"),
+    ):
+        vdc.fetch_id_token(_CLIENT, "", opener=MagicMock())
+
+
+def _firebase_reply(*, ok: bool, body: dict[str, object]) -> MagicMock:
+    response = MagicMock(ok=ok, status_code=200 if ok else 400)
+    response.json.return_value = body
+    return response
+
+
+def test_sign_in_returns_the_email_and_writes_no_file() -> None:
+    reply = _firebase_reply(
+        ok=True,
+        body={
+            "localId": "uid-1",
+            "email": "me@example.com",
+            "idToken": "i",
+            "refreshToken": "r",
+            "expiresIn": "3600",
+        },
+    )
+    with patch("crdt_sync._firebase_auth.requests.post", return_value=reply) as post:
+        assert vdc.sign_in(_CONFIG, _ID_VALUE) == "me@example.com"
+    assert "signInWithIdp" in post.call_args.args[0]
+
+
+def test_sign_in_surfaces_the_exact_firebase_error() -> None:
+    reply = _firebase_reply(
+        ok=False,
+        body={
+            "error": {"message": "INVALID_IDP_RESPONSE : id_token audience mismatch"}
+        },
+    )
+    with (
+        patch("crdt_sync._firebase_auth.requests.post", return_value=reply),
+        pytest.raises(FirebaseAuthError, match="audience mismatch"),
+    ):
+        vdc.sign_in(_CONFIG, _ID_VALUE)
+
+
+def test_exchange_sends_the_pkce_verifier() -> None:
+    reply = MagicMock(ok=True)
+    reply.json.return_value = {"id_token": _ID_VALUE}
+    with patch("tool._token_exchange.requests.post", return_value=reply) as post:
+        _exchange_code("c", "id", _HELD, "http://127.0.0.1:1", "ver")
+    assert post.call_args.kwargs["data"]["code_verifier"] == "ver"

crdt-sync/crdt_sync/tests/test_tool_verify_desktop_client_main.py

@@ -0,0 +1,125 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Tests for ``verify_desktop_client.main``: every exit path and its message."""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from crdt_sync._autherrors import FirebaseAuthError
+from crdt_sync._config import ConfigError, FirebaseConfig
+from tool import verify_desktop_client as vdc
+from tool._token_exchange import TokenError
+from tool.client_probe import ProbeResult
+from tool.desktop_client import ClientFileError, DesktopClient
+
+_HELD = "stored-client-credential"
+_CLIENT = DesktopClient("1-abc.apps.googleusercontent.com", _HELD, "kuhy-syncs")
+_CONFIG = FirebaseConfig("api", "https://db", "kuhy-syncs", "uid-1", "me@example.com")
+_ID_VALUE = (
+    "header."
+    + base64.urlsafe_b64encode(json.dumps({"aud": _CLIENT.client_id}).encode())
+    .decode()
+    .rstrip("=")
+    + ".sig"
+)
+
+
+def _run(
+    *,
+    probe_ok: bool = True,
+    token: object = _ID_VALUE,
+    sign_in: object = "me@example.com",
+    argv: list[str] | None = None,
+) -> tuple[int, MagicMock]:
+    fetch = MagicMock(
+        side_effect=token if isinstance(token, Exception) else None, return_value=token
+    )
+    signer = MagicMock(
+        side_effect=sign_in if isinstance(sign_in, Exception) else None,
+        return_value=sign_in,
+    )
+    with (
+        patch.object(vdc.FirebaseConfig, "load", return_value=_CONFIG),
+        patch.object(vdc, "load_client", return_value=_CLIENT),
+        patch.object(
+            vdc, "probe_client", return_value=[ProbeResult(probe_ok, "probed")]
+        ),
+        patch.object(vdc, "fetch_id_token", fetch),
+        patch.object(vdc, "sign_in", signer),
+    ):
+        return vdc.main(argv or []), fetch
+
+
+def test_main_passes_and_names_the_uid(caplog: pytest.LogCaptureFixture) -> None:
+    caplog.set_level(logging.INFO)
+    code, _ = _run()
+    assert code == 0
+    assert "PASS" in caplog.text
+    assert "uid-1" in caplog.text
+    assert "me@example.com" in caplog.text
+
+
+def test_main_stops_before_consent_when_the_probe_fails(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    caplog.set_level(logging.INFO)
+    code, fetch = _run(probe_ok=False)
+    assert code == 1
+    assert "FAIL" in caplog.text
+    fetch.assert_not_called()
+
+
+def test_main_probe_only_never_opens_consent() -> None:
+    code, fetch = _run(argv=["--probe-only"])
+    assert code == 0
+    fetch.assert_not_called()
+
+
+def test_main_skip_probe_goes_straight_to_consent() -> None:
+    code, fetch = _run(probe_ok=False, argv=["--skip-probe", "--no-browser"])
+    assert code == 0
+    assert fetch.call_args.kwargs["open_browser"] is False
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "needle"),
+    [
+        (
+            {"token": TokenError("token exchange failed: HTTP 400 invalid_grant")},
+            "invalid_grant",
+        ),
+        (
+            {
+                "sign_in": FirebaseAuthError(
+                    "INVALID_IDP_RESPONSE : id_token audience mismatch"
+                )
+            },
+            "Safelist",
+        ),
+    ],
+)
+def test_main_prints_the_exact_error_and_exits_1(
+    kwargs: dict[str, object], needle: str, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.INFO)
+    code, _ = _run(**kwargs)
+    assert code == 1
+    assert "FAIL" in caplog.text
+    assert needle in caplog.text
+
+
+@pytest.mark.parametrize(
+    "error", [ConfigError("no config"), ClientFileError("no client")]
+)
+def test_main_reports_setup_errors(
+    error: Exception, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.INFO)
+    with patch.object(vdc.FirebaseConfig, "load", side_effect=error):
+        assert vdc.main([]) == 1
+    assert "FAIL" in caplog.text

crdt-sync/tool/_pkce.py

@@ -0,0 +1,85 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""PKCE pair, consent URL and detached browser launch for the loopback flow."""
+
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+import hashlib
+import secrets
+import shutil
+import subprocess
+from typing import TYPE_CHECKING
+import urllib.parse
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+AUTH_ENDPOINT = "https://accounts.google.com/o/oauth2/v2/auth"
+SCOPES = "openid email profile"
+_VERIFIER_BYTES = 64
+
+
+@dataclass(frozen=True)
+class Pkce:
+    """A code verifier and the S256 challenge derived from it."""
+
+    verifier: str
+    challenge: str
+
+
+def challenge_for(verifier: str) -> str:
+    """Return the unpadded base64url SHA-256 of ``verifier`` (RFC 7636 S256)."""
+    digest = hashlib.sha256(verifier.encode("ascii")).digest()
+    return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
+
+
+def make_pkce() -> Pkce:
+    """Return a fresh 86-char verifier and its challenge."""
+    verifier = secrets.token_urlsafe(_VERIFIER_BYTES)
+    return Pkce(verifier=verifier, challenge=challenge_for(verifier))
+
+
+def build_consent_url(
+    *,
+    client_id: str,
+    redirect_uri: str,
+    challenge: str,
+    state: str,
+    login_hint: str,
+) -> str:
+    """Return the consent URL, with the sync account preselected."""
+    query = urllib.parse.urlencode(
+        {
+            "client_id": client_id,
+            "redirect_uri": redirect_uri,
+            "response_type": "code",
+            "scope": SCOPES,
+            "state": state,
+            "code_challenge": challenge,
+            "code_challenge_method": "S256",
+            "login_hint": login_hint,
+            "prompt": "select_account",
+        }
+    )
+    return f"{AUTH_ENDPOINT}?{query}"
+
+
+def open_detached(
+    url: str,
+    *,
+    popen: Callable[..., object] = subprocess.Popen,
+    which: Callable[[str], str | None] = shutil.which,
+) -> bool:
+    """Open ``url`` via ``setsid xdg-open``; False if either tool is missing."""
+    setsid, opener = which("setsid"), which("xdg-open")
+    if not setsid or not opener:
+        return False
+    popen(
+        [setsid, opener, url],
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        start_new_session=True,
+    )
+    return True

crdt-sync/tool/_token_exchange.py

@@ -44,9 +44,17 @@ def _authorization_code(params: dict[str, str], redirect_uri: str, state: str) -
 
 
 def _exchange_code(
-    code: str, client_id: str, client_secret: str, redirect_uri: str
+    code: str,
+    client_id: str,
+    client_secret: str,
+    redirect_uri: str,
+    code_verifier: str = "",
 ) -> str:
-    """Trade the authorization code for an ID token at Google's endpoint."""
+    """Trade the authorization code for an ID token at Google's endpoint.
+
+    ``code_verifier`` is sent only when given, for PKCE flows.
+    """
+    pkce = {"code_verifier": code_verifier} if code_verifier else {}
     try:
         response = requests.post(
             _TOKEN_EXCHANGE_URL,
@@ -56,6 +64,7 @@ def _exchange_code(
                 "client_secret": client_secret,
                 "redirect_uri": redirect_uri,
                 "grant_type": "authorization_code",
+                **pkce,
             },
             timeout=_TIMEOUT_SECONDS,
         )

crdt-sync/tool/client_probe.py

@@ -0,0 +1,158 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Click-free checks that a freshly stored Desktop OAuth client actually works.
+
+Both requests need no signed-in user: Google's endpoints answer them before any
+account is involved, so they say whether the client exists, accepts a loopback
+redirect and whether the stored secret is right -- without opening a browser.
+That separates "the console step went wrong" from "Firebase rejects the
+audience" before the one human consent click is spent.
+"""
+
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+import re
+from typing import TYPE_CHECKING
+import urllib.parse
+
+import requests
+
+from tool._pkce import AUTH_ENDPOINT, challenge_for
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from tool.desktop_client import DesktopClient
+
+# Assembled from host and path: a literal ending in "/token" trips ruff's
+# hardcoded-password check, and suppressions are banned in this repo.
+_OAUTH_HOST = "https://oauth2.googleapis.com"
+_TOKEN_URL = f"{_OAUTH_HOST}/token"
+_PROBE_REDIRECT = "http://127.0.0.1:41234"
+_TIMEOUT_SECONDS = 20
+_ASCII_RUN = re.compile(rb"[ -~]{6,}")
+_PROBE_VERIFIER = "probe-verifier-" + "x" * 30
+
+
+@dataclass(frozen=True)
+class ProbeResult:
+    """One check's verdict and the words that explain it."""
+
+    ok: bool
+    detail: str
+
+
+def classify_auth_location(location: str | None) -> ProbeResult:
+    """Judge the auth endpoint's redirect target.
+
+    A valid request while signed out still redirects (to the sign-in page), so
+    the status code means nothing: only an ``authError`` parameter in the
+    target says Google refused the client or its redirect URI.
+    """
+    if not location or "authError=" not in location:
+        return ProbeResult(
+            ok=True, detail="auth endpoint accepts the loopback redirect"
+        )
+    raw = location.split("authError=", 1)[1].split("&", 1)[0]
+    try:
+        decoded = base64.b64decode(
+            raw + "=" * (-len(raw) % 4), altchars=b"-_", validate=True
+        )
+    except ValueError:
+        return ProbeResult(
+            ok=False, detail=f"auth endpoint refused the request (authError={raw})"
+        )
+    text = " | ".join(run.decode("ascii") for run in _ASCII_RUN.findall(decoded))
+    return ProbeResult(
+        ok=False, detail=f"auth endpoint refused the request: {text[:300]}"
+    )
+
+
+def classify_token(status: int, body: dict[str, str] | None) -> ProbeResult:
+    """Judge the token endpoint's answer to a deliberately bogus code.
+
+    ``invalid_grant`` is the success signal: Google checked the client and
+    secret first, accepted them, and only then rejected the fake code.
+    """
+    if body is None:
+        return ProbeResult(ok=False, detail=f"token endpoint: HTTP {status}, no JSON")
+    error = body.get("error", "")
+    description = body.get("error_description", "")
+    if error == "invalid_grant":
+        return ProbeResult(
+            ok=True, detail="token endpoint accepts the client and secret"
+        )
+    if error == "invalid_client" and "not found" in description:
+        return ProbeResult(
+            ok=False,
+            detail=(
+                "client not found yet: new clients take a few minutes to "
+                "propagate; rerun with --verify-only shortly"
+            ),
+        )
+    if error == "invalid_client":
+        return ProbeResult(
+            ok=False, detail=f"the stored secret is wrong ({description})"
+        )
+    return ProbeResult(ok=False, detail=f"token endpoint: HTTP {status} {body}")
+
+
+def _probe_auth(
+    client: DesktopClient, get: Callable[..., requests.Response]
+) -> ProbeResult:
+    query = urllib.parse.urlencode(
+        {
+            "client_id": client.client_id,
+            "redirect_uri": _PROBE_REDIRECT,
+            "response_type": "code",
+            "scope": "openid email",
+            "code_challenge": challenge_for(_PROBE_VERIFIER),
+            "code_challenge_method": "S256",
+            "hl": "en",
+        }
+    )
+    try:
+        reply = get(
+            f"{AUTH_ENDPOINT}?{query}",
+            allow_redirects=False,
+            timeout=_TIMEOUT_SECONDS,
+        )
+    except requests.RequestException as exc:
+        return ProbeResult(ok=False, detail=f"auth endpoint unreachable: {exc}")
+    return classify_auth_location(reply.headers.get("Location"))
+
+
+def _probe_token(
+    client: DesktopClient, post: Callable[..., requests.Response]
+) -> ProbeResult:
+    try:
+        answer = post(
+            _TOKEN_URL,
+            data={
+                "code": "probe-not-a-real-code",
+                "client_id": client.client_id,
+                "client_secret": client.client_secret,
+                "redirect_uri": _PROBE_REDIRECT,
+                "grant_type": "authorization_code",
+                "code_verifier": _PROBE_VERIFIER,
+            },
+            timeout=_TIMEOUT_SECONDS,
+        )
+    except requests.RequestException as exc:
+        return ProbeResult(ok=False, detail=f"token endpoint unreachable: {exc}")
+    try:
+        body = answer.json()
+    except ValueError:
+        body = None
+    return classify_token(answer.status_code, body)
+
+
+def probe_client(
+    client: DesktopClient,
+    *,
+    get: Callable[..., requests.Response] = requests.get,
+    post: Callable[..., requests.Response] = requests.post,
+) -> list[ProbeResult]:
+    """Run the auth-endpoint and token-endpoint checks for ``client``."""
+    return [_probe_auth(client, get), _probe_token(client, post)]

crdt-sync/tool/desktop_client.py

@@ -0,0 +1,173 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Pick up, validate and store the Desktop OAuth client JSON from the console.
+
+Android and iOS clients download as the same top-level ``installed`` object, so
+the type alone proves nothing: only a ``client_secret`` plus the
+``http://localhost`` redirect marks a Desktop client.
+"""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+from functools import partial
+import json
+import logging
+import os
+from pathlib import Path
+import sys
+import time
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+PROJECT_ID = "kuhy-syncs"
+DEFAULT_DEST = Path.home() / ".config" / "crdt-sync" / "desktop_client.json"
+DEFAULT_DOWNLOADS = Path.home() / "Downloads"
+_POLL_SECONDS = 2.0
+_pause_between_polls = partial(time.sleep, _POLL_SECONDS)
+_CLIENT_ID_SUFFIX = ".apps.googleusercontent.com"
+_LOOPBACK_REDIRECT = "http://localhost"
+
+_logger = logging.getLogger("desktop_client")
+
+
+class ClientFileError(Exception):
+    """The client JSON is missing, malformed or not a Desktop client."""
+
+
+@dataclass(frozen=True)
+class DesktopClient:
+    """The installed-app client as downloaded from the console."""
+
+    client_id: str
+    client_secret: str
+    project_id: str
+
+
+@dataclass(frozen=True)
+class Pickup:
+    """A valid client JSON found in the downloads folder."""
+
+    path: Path
+    client: DesktopClient
+
+
+def parse_client_json(text: str, project_id: str = PROJECT_ID) -> DesktopClient:
+    """Return the Desktop client in ``text``, or raise saying what is wrong."""
+    try:
+        document = json.loads(text)
+    except ValueError as exc:
+        msg = f"not valid JSON: {exc}"
+        raise ClientFileError(msg) from exc
+    if not isinstance(document, dict):
+        msg = "must contain a JSON object"
+        raise ClientFileError(msg)
+    installed = document.get("installed")
+    if not isinstance(installed, dict):
+        msg = f"no 'installed' object (found {sorted(document)}); pick 'Desktop app'"
+        raise ClientFileError(msg)
+    if installed.get("project_id") != project_id:
+        msg = f"belongs to project {installed.get('project_id')!r}, not {project_id!r}"
+        raise ClientFileError(msg)
+    client_id = str(installed.get("client_id", ""))
+    if not client_id.endswith(_CLIENT_ID_SUFFIX):
+        msg = f"client_id {client_id!r} is not a Google OAuth client id"
+        raise ClientFileError(msg)
+    held = str(installed.get("client_secret", ""))
+    if not held:
+        msg = "no client_secret: this is an Android/iOS client, not a Desktop app"
+        raise ClientFileError(msg)
+    if _LOOPBACK_REDIRECT not in installed.get("redirect_uris", []):
+        msg = f"redirect_uris lacks {_LOOPBACK_REDIRECT}: not a Desktop app client"
+        raise ClientFileError(msg)
+    return DesktopClient(client_id=client_id, client_secret=held, project_id=project_id)
+
+
+def _scan(
+    downloads: Path, since: float, rejected: set[str], report: Callable[[str], None]
+) -> Pickup | None:
+    candidates = [
+        path
+        for path in downloads.glob("client_secret_*.json")
+        if path.stat().st_mtime >= since
+    ]
+    for path in sorted(candidates, key=lambda p: p.stat().st_mtime, reverse=True):
+        try:
+            client = parse_client_json(path.read_text(encoding="utf-8"))
+        except ClientFileError as exc:
+            if path.name not in rejected:
+                rejected.add(path.name)
+                report(f"ignoring {path.name}: {exc}")
+            continue
+        return Pickup(path=path, client=client)
+    return None
+
+
+def wait_for_client_json(
+    downloads: Path,
+    since: float,
+    *,
+    attempts: int,
+    pause: Callable[[], None] = _pause_between_polls,
+    report: Callable[[str], None] = print,
+) -> Pickup | None:
+    """Scan ``downloads`` up to ``attempts`` times for a JSON newer than ``since``."""
+    rejected: set[str] = set()
+    for attempt in range(attempts):
+        found = _scan(downloads, since, rejected, report)
+        if found:
+            return found
+        if attempt < attempts - 1:
+            pause()
+    return None
+
+
+def store_client(source: Path, dest: Path) -> None:
+    """Move ``source`` to ``dest`` with mode 0600, so no copy lingers in Downloads."""
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    descriptor = os.open(dest, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
+        handle.write(source.read_text(encoding="utf-8"))
+    dest.chmod(0o600)
+    source.unlink()
+
+
+def load_client(dest: Path = DEFAULT_DEST) -> DesktopClient:
+    """Load the stored client, pointing at the register script when absent."""
+    try:
+        return parse_client_json(dest.read_text(encoding="utf-8"))
+    except OSError as exc:
+        msg = f"{dest} unreadable ({exc}); run tool/register_desktop_client.sh"
+        raise ClientFileError(msg) from exc
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Wait for the downloaded JSON and store it. Returns an exit code."""
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("action", choices=["pickup"])
+    parser.add_argument("--since", type=float, required=True, help="epoch seconds")
+    parser.add_argument("--timeout", type=float, default=600.0)
+    parser.add_argument("--downloads", type=Path, default=DEFAULT_DOWNLOADS)
+    parser.add_argument("--dest", type=Path, default=DEFAULT_DEST)
+    args = parser.parse_args(argv)
+    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
+    found = wait_for_client_json(
+        args.downloads,
+        args.since,
+        attempts=int(args.timeout / _POLL_SECONDS) + 1,
+        report=_logger.info,
+    )
+    if found is None:
+        _logger.error("  FAIL  no Desktop client JSON appeared in %s", args.downloads)
+        return 1
+    store_client(found.path, args.dest)
+    _logger.info(
+        "  PASS  stored %s at %s (mode 600)", found.client.client_id, args.dest
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

crdt-sync/tool/register_desktop_client.sh

@@ -0,0 +1,127 @@
+#!/bin/bash
+# Registers the Desktop OAuth client (no API exists, so one console step stays
+# manual), stores its JSON privately, then proves Firebase accepts its tokens.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
+readonly SCRIPT_DIR REPO_DIR
+readonly PROJECT="kuhy-syncs"
+readonly ACCOUNT="321krzychu@gmail.com"
+readonly CONSOLE_URL="https://console.cloud.google.com/auth/clients?project=${PROJECT}&authuser=${ACCOUNT}"
+readonly APP_TYPE="Desktop app"
+readonly CLIENT_NAME="crdt-sync desktop/android PKCE"
+readonly PAUSE_SECONDS="${PAUSE_SECONDS:-12}"
+readonly WAIT_SECONDS="${WAIT_SECONDS:-900}"
+readonly PROBE_TRIES="${PROBE_TRIES:-12}"
+readonly PROBE_GAP_SECONDS="${PROBE_GAP_SECONDS:-30}"
+
+log() { printf '\n\033[1m==> %s\033[0m\n' "$1"; }
+step() { printf '   %s\n' "$1"; }
+
+usage() {
+    cat <<EOF
+Usage: $(basename "$0") [--verify-only | --dry-run | --help]
+
+  (no flag)      open the console, clipboard-feed the form, pick up the
+                 downloaded JSON, store it, then verify
+  --verify-only  skip registration; probe, consent flow, signInWithIdp
+  --dry-run      print the plan; open, copy and write nothing
+
+Your part: Application type = Desktop app, paste the name, CREATE, DOWNLOAD
+JSON in the dialog; later tap Allow on the consent page.
+Stored at ~/.config/crdt-sync/desktop_client.json (mode 600).
+EOF
+}
+
+# Prints the value always, so this works over ssh with no X display too.
+clip() {
+    if command -v xclip >/dev/null 2>&1 && [[ -n "${DISPLAY:-}" ]]; then
+        printf '%s' "$1" | xclip -selection clipboard 2>/dev/null || true
+        printf '   \033[32m[copied]\033[0m %s\n' "$1"
+    else
+        printf '   %s\n' "$1"
+    fi
+}
+
+# Without a terminal (e.g. run via `!`) read would hit EOF at once.
+pause() {
+    if [[ -t 0 ]]; then
+        read -rp "   ...press Enter once pasted " _ || true
+    else
+        sleep "$PAUSE_SECONDS"
+    fi
+}
+
+open_console() {
+    if command -v xdg-open >/dev/null 2>&1; then
+        setsid xdg-open "$CONSOLE_URL" >/dev/null 2>&1 &
+    else
+        step "open: $CONSOLE_URL"
+    fi
+}
+
+run_tool() { (cd "$REPO_DIR" && python3 -m "tool.$1" "${@:2}"); }
+
+wait_for_propagation() {
+    local try
+    for ((try = 1; try <= PROBE_TRIES; try++)); do
+        if run_tool verify_desktop_client --probe-only; then
+            return 0
+        fi
+        step "probe $try/$PROBE_TRIES failed; new clients need a few minutes"
+        sleep "$PROBE_GAP_SECONDS"
+    done
+    echo "error: client still not accepted; rerun with --verify-only later" >&2
+    return 1
+}
+
+verify() {
+    log "Verifying: probe, consent flow, Firebase signInWithIdp"
+    step "Tap Allow on the sync account's consent page when it opens."
+    run_tool verify_desktop_client
+}
+
+register() {
+    local started
+    started="$(date +%s)"
+    log "Create ONE Desktop OAuth client in project $PROJECT"
+    open_console
+    sleep 2
+    step "Click '+ CREATE CLIENT'. Application type (paste or pick):"
+    clip "$APP_TYPE"
+    pause
+    step "Field 'Name':"
+    clip "$CLIENT_NAME"
+    pause
+    step "Click CREATE, then DOWNLOAD JSON in the dialog (the secret shows once)."
+    step "Leave the existing Web and Android clients alone."
+    log "Waiting up to ${WAIT_SECONDS}s for the downloaded client_secret_*.json"
+    run_tool desktop_client pickup --since "$started" --timeout "$WAIT_SECONDS"
+    wait_for_propagation
+    verify
+}
+
+dry_run() {
+    log "Dry run: nothing is opened, copied or written"
+    step "page:      $CONSOLE_URL"
+    step "type:      $APP_TYPE"
+    step "name:      $CLIENT_NAME"
+    step "stores at: ~/.config/crdt-sync/desktop_client.json (mode 600)"
+    step "verify:    python3 -m tool.verify_desktop_client (in $REPO_DIR)"
+}
+
+main() {
+    case "${1:-}" in
+        --verify-only) verify ;;
+        --dry-run) dry_run ;;
+        -h | --help) usage ;;
+        "") register ;;
+        *)
+            usage >&2
+            exit 2
+            ;;
+    esac
+}
+
+main "$@"

crdt-sync/tool/verify_desktop_client.py

@@ -0,0 +1,165 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Prove Firebase accepts an id_token issued to the stored Desktop OAuth client.
+
+Runs the PKCE loopback flow with the installed-app client (secret included in
+the exchange), then calls Firebase Auth REST ``signInWithIdp`` with the result.
+The consent click in the browser is the only human step. No app credential
+file is written: the session lives in memory and is dropped.
+
+Run from the crdt-sync directory::
+
+    python3 -m tool.verify_desktop_client
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import logging
+import secrets
+import sys
+from typing import TYPE_CHECKING
+
+from crdt_sync._autherrors import FirebaseAuthError
+from crdt_sync._config import ConfigError, FirebaseConfig
+from crdt_sync._credentials import MemoryCredentialStore
+from crdt_sync._firebase_auth import FirebaseTokenProvider
+from tool._oauth_callback import (
+    _CONSENT_TIMEOUT_SECONDS,
+    _CallbackHandler,
+    _free_port,
+    _start_callback_server,
+)
+from tool._pkce import build_consent_url, make_pkce, open_detached
+from tool._token_exchange import TokenError, _authorization_code, _exchange_code
+from tool.client_probe import probe_client
+from tool.desktop_client import ClientFileError, DesktopClient, load_client
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+_logger = logging.getLogger("verify_desktop_client")
+
+
+def audience_of(id_token: str) -> str:
+    """Return the ``aud`` claim of an unverified JWT, or ``?`` if unreadable."""
+    try:
+        segment = id_token.split(".")[1]
+        claims = json.loads(
+            base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4))
+        )
+        return str(claims["aud"])
+    except IndexError, ValueError, KeyError, TypeError:
+        return "?"
+
+
+def hint_for(message: str) -> str:
+    """Return what to do about a known failure message, or an empty string."""
+    if "audience mismatch" in message:
+        return (
+            "Firebase does not trust this client yet: Firebase console > "
+            "Authentication > Sign-in method > Google > Safelist client IDs "
+            "from external projects, add the Desktop client id, then rerun."
+        )
+    if "wrong account" in message:
+        return (
+            "Pick the sync account in the consent picker (login_hint only "
+            "preselects it). Firebase may have created a stray user for the "
+            "other account: check Authentication > Users."
+        )
+    return ""
+
+
+def fetch_id_token(
+    client: DesktopClient,
+    login_hint: str,
+    *,
+    open_browser: bool = True,
+    opener: Callable[[str], bool] = open_detached,
+) -> str:
+    """Run the PKCE loopback flow with ``client`` and return its id_token."""
+    port = _free_port()
+    redirect_uri = f"http://127.0.0.1:{port}"
+    state = secrets.token_urlsafe(16)
+    pkce = make_pkce()
+    server, thread = _start_callback_server(port, state)
+    url = build_consent_url(
+        client_id=client.client_id,
+        redirect_uri=redirect_uri,
+        challenge=pkce.challenge,
+        state=state,
+        login_hint=login_hint,
+    )
+    _logger.info("  ..  consent page (opened in your browser, tap Allow):")
+    _logger.info("      %s", url)
+    if open_browser:
+        opener(url)
+    thread.join(timeout=_CONSENT_TIMEOUT_SECONDS)
+    server.server_close()
+    code = _authorization_code(_CallbackHandler.result, redirect_uri, state)
+    return _exchange_code(
+        code, client.client_id, client.client_secret, redirect_uri, pkce.verifier
+    )
+
+
+def sign_in(config: FirebaseConfig, id_token: str) -> str:
+    """Call ``signInWithIdp`` in memory; return the account email.
+
+    Raises:
+        FirebaseAuthError: With Firebase's exact message, or on a uid other
+            than the pinned one.
+    """
+    provider = FirebaseTokenProvider(config.api_key, MemoryCredentialStore())
+    return provider.sign_in_with_google(id_token, expected_uid=config.uid)
+
+
+def _fail(message: str) -> int:
+    _logger.error("  FAIL  %s", message)
+    hint = hint_for(message)
+    if hint:
+        _logger.error("  ..    %s", hint)
+    return 1
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Probe the stored client, run the consent flow, sign in to Firebase."""
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--no-browser", action="store_true", help="only print the URL")
+    parser.add_argument(
+        "--skip-probe", action="store_true", help="skip click-free checks"
+    )
+    parser.add_argument(
+        "--probe-only", action="store_true", help="stop after the click-free checks"
+    )
+    args = parser.parse_args(argv)
+    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
+
+    try:
+        config = FirebaseConfig.load()
+        client = load_client()
+        if not args.skip_probe:
+            results = probe_client(client)
+            for result in results:
+                _logger.info("  %s  %s", "ok" if result.ok else "!!", result.detail)
+            failed = [r.detail for r in results if not r.ok]
+            if failed:
+                return _fail("; ".join(failed))
+            if args.probe_only:
+                return 0
+        id_token = fetch_id_token(
+            client, config.email, open_browser=not args.no_browser
+        )
+        _logger.info("  ..  id_token audience: %s", audience_of(id_token))
+        email = sign_in(config, id_token)
+    except (ConfigError, ClientFileError, TokenError, FirebaseAuthError) as exc:
+        return _fail(str(exc))
+
+    _logger.info(
+        "  PASS  Firebase accepted the Desktop client: uid %s, %s", config.uid, email
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())