diff --git a/crdt-sync/crdt_sync/tests/test_tool_link_google.py b/crdt-sync/crdt_sync/tests/test_tool_link_google.py new file mode 100644 index 0000000..1f4ca52 --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_tool_link_google.py @@ -0,0 +1,249 @@ +"""Tests for token inspection and linking in ``tool/link_google``. + +``main()`` and ``_read_token`` are covered in +:mod:`test_tool_link_google_main`, split to stay under the 250-line cap. + +The failure this whole script exists to prevent is silent: linking onto a +*new* account succeeds at the auth layer and then has every read and write +denied by rules that pin one uid. So the two assertions that matter most here +are that the existing account's ``idToken`` is always sent (omitting it turns +the call into "sign in or sign up") and that a mismatched uid is a hard error. +""" + +from __future__ import annotations + +import base64 +import json +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from crdt_sync import FirebaseConfig +from tool import link_google as lg + +_UID = "the-uid-pinned-in-the-rules" +_CONFIG = FirebaseConfig( + api_key="AIzaSyAthisIsTheWebApiKeyFormat", + database_url="https://project-default-rtdb.firebasedatabase.app", + project_id="the-project", + uid=_UID, + email="sync@example.com", +) + + +def _jwt(claims: dict[str, object]) -> str: + """Build an unsigned JWT carrying ``claims`` as its payload.""" + payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode() + return f"header.{payload.rstrip('=')}.signature" + + +def _google_token(**overrides: object) -> str: + """A plausible Google ID token, with individual claims overridden.""" + claims = { + "iss": "https://accounts.google.com", + "email": "person@gmail.com", + "aud": "the-web-client-id.apps.googleusercontent.com", + } + return _jwt({**claims, **overrides}) + + +def _response(*, ok: bool = True, status: int = 200, body: object = None) -> MagicMock: + """Fake a ``requests`` response from accounts:signInWithIdp.""" + response = MagicMock() + response.ok = ok + response.status_code = status + response.text = "" if ok else '{"error": {"message": "INVALID_IDP_RESPONSE"}}' + response.json.return_value = body if body is not None else {"localId": _UID} + return response + + +def test_a_token_that_is_not_three_parts_is_rejected() -> None: + """An access token or an auth code pasted by mistake. + + The message names both, because that is the actual confusion -- the + module docstring exists because the Firebase-side error does not. + """ + with pytest.raises(lg.LinkError, match="not an access token"): + lg.describe_google_token("not-a-jwt") + + +def test_a_token_with_an_unparseable_payload_is_rejected() -> None: + """Three parts, but the middle one is not base64url JSON.""" + with pytest.raises(lg.LinkError, match="not valid base64url JSON"): + lg.describe_google_token("header.!!!not-base64!!!.signature") + + +def test_a_payload_that_is_not_an_object_is_rejected() -> None: + """Valid base64url JSON, wrong shape.""" + payload = base64.urlsafe_b64encode(b'"just a string"').decode().rstrip("=") + + with pytest.raises(lg.LinkError, match="not a JSON object"): + lg.describe_google_token(f"header.{payload}.signature") + + +def test_a_token_from_another_issuer_is_rejected() -> None: + """The check that makes a misconfiguration report itself locally.""" + with pytest.raises(lg.LinkError, match="not Google"): + lg.describe_google_token(_google_token(iss="https://login.example.com")) + + +@pytest.mark.parametrize( + "issuer", ["https://accounts.google.com", "accounts.google.com"] +) +def test_both_google_issuer_spellings_are_accepted(issuer: str) -> None: + """Google uses both forms; rejecting either would be a false alarm.""" + assert "person@gmail.com" in lg.describe_google_token(_google_token(iss=issuer)) + + +def test_the_description_names_the_account_and_the_audience() -> None: + """The operator's only chance to notice they are linking the wrong one.""" + description = lg.describe_google_token(_google_token()) + + assert "person@gmail.com" in description + assert "the-web-client-id.apps.googleusercontent.com" in description + + +def test_missing_optional_claims_are_described_rather_than_crashing() -> None: + """A token without email/aud is odd but still describable.""" + token = _jwt({"iss": "accounts.google.com"}) + + description = lg.describe_google_token(token) + + assert "" in description + assert "" in description + + +def test_padding_is_restored_before_decoding() -> None: + """base64url in a JWT is unpadded; decoding without padding fails.""" + # A payload whose length is not a multiple of 4 once stripped. + claims = {"iss": "accounts.google.com", "email": "a@b.co"} + + assert "a@b.co" in lg.describe_google_token(_jwt(claims)) + + +def _auth_provider() -> MagicMock: + """A token provider that signs in and yields the account's ID token.""" + auth = MagicMock() + auth.id_token.return_value = "the-existing-account-id-token" + return auth + + +def test_linking_returns_the_unchanged_uid() -> None: + """The happy path: same uid before and after.""" + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object(lg.requests, "post", return_value=_response()) as posted, + ): + result = lg.link_google(_CONFIG, _google_token()) + + assert result == _UID + assert posted.call_count == 1 + + +def test_the_existing_id_token_is_always_sent() -> None: + """Omitting it makes the same endpoint create a brand-new account. + + That is the silent failure the whole script exists to prevent, so it is + asserted on the request body rather than inferred from the response. + """ + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object(lg.requests, "post", return_value=_response()) as posted, + ): + lg.link_google(_CONFIG, "the-google-token") + + sent = posted.call_args.kwargs["json"] + assert sent["idToken"] == "the-existing-account-id-token" + assert "id_token=the-google-token" in sent["postBody"] + assert "providerId=google.com" in sent["postBody"] + assert sent["returnSecureToken"] is True + + +def test_the_account_password_is_used_to_prove_ownership() -> None: + """Linking requires proving ownership of the account being linked to.""" + auth = _auth_provider() + config = MagicMock(spec=FirebaseConfig) + config.api_key = _CONFIG.api_key + config.email = _CONFIG.email + config.uid = _UID + config.read_password.return_value = "the-stored-password" + + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=auth), + patch.object(lg.requests, "post", return_value=_response()), + ): + lg.link_google(config, _google_token()) + + auth.sign_in.assert_called_once_with(_CONFIG.email, "the-stored-password") + + +def test_linking_onto_a_different_uid_is_a_hard_failure() -> None: + """The decisive check. + + A different uid authenticates perfectly and is then denied every read and + write, so the message has to say that outright and name the remedy. + """ + body = {"localId": "a-brand-new-uid"} + + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object(lg.requests, "post", return_value=_response(body=body)), + pytest.raises(lg.LinkError, match="LINKED THE WRONG ACCOUNT"), + ): + lg.link_google(_CONFIG, _google_token()) + + +def test_the_wrong_account_message_names_the_unlink_remedy() -> None: + """Knowing it failed is not enough; the console step is not obvious.""" + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object( + lg.requests, "post", return_value=_response(body={"localId": "x"}) + ), + pytest.raises(lg.LinkError, match="Unlink this Google identity"), + ): + lg.link_google(_CONFIG, _google_token()) + + +def test_a_response_without_a_local_id_is_rejected() -> None: + """No uid at all must not read as a successful link.""" + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object(lg.requests, "post", return_value=_response(body={})), + pytest.raises(lg.LinkError, match="no localId"), + ): + lg.link_google(_CONFIG, _google_token()) + + +def test_a_rejected_link_reports_the_status_and_body() -> None: + """Firebase's own message is the only clue to why it refused.""" + response = _response(ok=False, status=400) + + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object(lg.requests, "post", return_value=response), + pytest.raises(lg.LinkError, match="HTTP 400"), + ): + lg.link_google(_CONFIG, _google_token()) + + +def test_a_network_error_is_wrapped_as_a_link_error() -> None: + """A bare requests exception would escape main()'s handler.""" + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object(lg.requests, "post", side_effect=requests.Timeout("timed out")), + pytest.raises(lg.LinkError, match="network error"), + ): + lg.link_google(_CONFIG, _google_token()) + + +def test_the_timeout_is_passed_through() -> None: + """A hung link would otherwise block the run indefinitely.""" + with ( + patch.object(lg, "FirebaseTokenProvider", return_value=_auth_provider()), + patch.object(lg.requests, "post", return_value=_response()) as posted, + ): + lg.link_google(_CONFIG, _google_token(), timeout_seconds=5) + + assert posted.call_args.kwargs["timeout"] == 5 diff --git a/crdt-sync/crdt_sync/tests/test_tool_link_google_main.py b/crdt-sync/crdt_sync/tests/test_tool_link_google_main.py new file mode 100644 index 0000000..a65069c --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_tool_link_google_main.py @@ -0,0 +1,224 @@ +"""Tests for ``_read_token`` and ``main()`` in ``tool/link_google``. + +Split from :mod:`test_tool_link_google` to stay under the 250-line cap. + +``main()`` catches three exception types by name rather than with a bare +``except Exception``, deliberately, so that an unanticipated bug still +surfaces as a traceback instead of a tidy "FAIL" line. That distinction is +asserted here: the three named failures return 1, and anything else escapes. +""" + +from __future__ import annotations + +import argparse +import logging +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from crdt_sync._config import ConfigError +from crdt_sync._firebase_auth import FirebaseAuthError +from tool import link_google as lg + +from .test_tool_link_google import _CONFIG, _UID, _google_token + +if TYPE_CHECKING: + from pathlib import Path + + +def _args( + token: str | None = None, token_file: str | None = None +) -> argparse.Namespace: + """Build the namespace ``_read_token`` reads.""" + return argparse.Namespace( + google_id_token=token, + google_id_token_file=token_file, + ) + + +def test_a_token_is_read_from_the_command_line() -> None: + """The inline form.""" + assert lg._read_token(_args(token="the-token")) == "the-token" + + +def test_a_token_is_read_from_a_file(tmp_path: Path) -> None: + """The file form, which is what the paired minting tool writes.""" + path = tmp_path / "token.txt" + path.write_text("the-token", encoding="utf-8") + + assert lg._read_token(_args(token_file=str(path))) == "the-token" + + +def test_a_trailing_newline_is_stripped(tmp_path: Path) -> None: + """A token written by a file or heredoc picks one up. + + Firebase rejects it with an unhelpful message, so stripping here is what + keeps that from becoming a debugging session. + """ + path = tmp_path / "token.txt" + path.write_text("the-token\n", encoding="utf-8") + + assert lg._read_token(_args(token_file=str(path))) == "the-token" + + +def test_an_unreadable_token_file_is_reported_by_path(tmp_path: Path) -> None: + """Naming the path is the difference between a fix and a guess.""" + missing = tmp_path / "absent.txt" + + with pytest.raises(lg.LinkError, match="could not be read"): + lg._read_token(_args(token_file=str(missing))) + + +def test_an_empty_token_is_rejected(tmp_path: Path) -> None: + """An empty file would otherwise be sent to Firebase as a valid request.""" + path = tmp_path / "token.txt" + path.write_text(" \n", encoding="utf-8") + + with pytest.raises(lg.LinkError, match="is empty"): + lg._read_token(_args(token_file=str(path))) + + +def test_an_empty_inline_token_is_rejected() -> None: + """Same check, the other input route.""" + with pytest.raises(lg.LinkError, match="is empty"): + lg._read_token(_args(token="")) + + +def test_main_links_and_reports_the_unchanged_uid( + caplog: pytest.LogCaptureFixture, +) -> None: + """The happy path: exit 0, and the uid echoed so it can be eyeballed.""" + with ( + patch.object(lg.FirebaseConfig, "load", return_value=_CONFIG), + patch.object(lg, "link_google", return_value=_UID) as linked, + caplog.at_level(logging.INFO, logger="link_google"), + ): + code = lg.main(["--google-id-token", _google_token()]) + + assert code == 0 + assert linked.call_count == 1 + assert "uid unchanged" in caplog.text + assert _UID in caplog.text + + +def test_main_describes_the_token_before_linking( + caplog: pytest.LogCaptureFixture, +) -> None: + """The operator's chance to abort on the wrong Google account.""" + with ( + patch.object(lg.FirebaseConfig, "load", return_value=_CONFIG), + patch.object(lg, "link_google", return_value=_UID), + caplog.at_level(logging.INFO, logger="link_google"), + ): + lg.main(["--google-id-token", _google_token()]) + + assert "person@gmail.com" in caplog.text + assert "linking onto sync@example.com" in caplog.text + + +def test_main_points_at_the_verification_step( + caplog: pytest.LogCaptureFixture, +) -> None: + """Linking is only half the job; the preflight is what confirms it.""" + with ( + patch.object(lg.FirebaseConfig, "load", return_value=_CONFIG), + patch.object(lg, "link_google", return_value=_UID), + caplog.at_level(logging.INFO, logger="link_google"), + ): + lg.main(["--google-id-token", _google_token()]) + + assert "tool/preflight_firebase.py" in caplog.text + + +def test_main_reads_the_token_from_a_file(tmp_path: Path) -> None: + """The --google-id-token-file route reaches link_google intact.""" + path = tmp_path / "token.txt" + token = _google_token() + path.write_text(f"{token}\n", encoding="utf-8") + + with ( + patch.object(lg.FirebaseConfig, "load", return_value=_CONFIG), + patch.object(lg, "link_google", return_value=_UID) as linked, + ): + code = lg.main(["--google-id-token-file", str(path)]) + + assert code == 0 + assert linked.call_args.args[1] == token + + +def test_main_returns_1_on_a_link_error(caplog: pytest.LogCaptureFixture) -> None: + """The uid mismatch and every other LinkError.""" + with ( + patch.object(lg.FirebaseConfig, "load", return_value=_CONFIG), + patch.object(lg, "link_google", side_effect=lg.LinkError("wrong account")), + caplog.at_level(logging.INFO, logger="link_google"), + ): + code = lg.main(["--google-id-token", _google_token()]) + + assert code == 1 + assert "FAIL" in caplog.text + + +def test_main_returns_1_on_a_config_error(caplog: pytest.LogCaptureFixture) -> None: + """An unusable firebase.json is a reported failure, not a traceback.""" + with ( + patch.object(lg.FirebaseConfig, "load", side_effect=ConfigError("no config")), + caplog.at_level(logging.INFO, logger="link_google"), + ): + code = lg.main(["--google-id-token", _google_token()]) + + assert code == 1 + assert "FAIL" in caplog.text + + +def test_main_returns_1_on_an_auth_error(caplog: pytest.LogCaptureFixture) -> None: + """A rejected password, the third named failure.""" + with ( + patch.object(lg.FirebaseConfig, "load", return_value=_CONFIG), + patch.object(lg, "link_google", side_effect=FirebaseAuthError("bad password")), + caplog.at_level(logging.INFO, logger="link_google"), + ): + code = lg.main(["--google-id-token", _google_token()]) + + assert code == 1 + assert "FAIL" in caplog.text + + +def test_main_rejects_a_malformed_token_before_touching_the_network() -> None: + """The local shape check exists to save a round trip; prove it does.""" + with ( + patch.object(lg.FirebaseConfig, "load") as load, + patch.object(lg, "link_google") as linked, + ): + code = lg.main(["--google-id-token", "not-a-jwt"]) + + assert code == 1 + load.assert_not_called() + linked.assert_not_called() + + +def test_an_unexpected_error_is_not_swallowed() -> None: + """The reason the handler names its three types instead of Exception. + + A tidy "FAIL" line for an unanticipated bug would hide it; a traceback + is the point. + """ + with ( + patch.object(lg.FirebaseConfig, "load", return_value=_CONFIG), + patch.object(lg, "link_google", side_effect=RuntimeError("a real bug")), + pytest.raises(RuntimeError, match="a real bug"), + ): + lg.main(["--google-id-token", _google_token()]) + + +def test_the_two_token_sources_are_mutually_exclusive() -> None: + """Passing both is a user error argparse should catch, not a silent pick.""" + with pytest.raises(SystemExit): + lg.main(["--google-id-token", "a", "--google-id-token-file", "b"]) + + +def test_one_token_source_is_required() -> None: + """With neither, there is nothing to link.""" + with pytest.raises(SystemExit): + lg.main([])