diff --git a/crdt-sync/crdt_sync/_autherrors.py b/crdt-sync/crdt_sync/_autherrors.py new file mode 100644 index 0000000..1270118 --- /dev/null +++ b/crdt-sync/crdt_sync/_autherrors.py @@ -0,0 +1,69 @@ +"""Turning a Firebase auth response into a message, and reading its verdict. + +Split from :mod:`crdt_sync._firebase_auth`, which keeps the token provider. +identitytoolkit reports failures as a nested JSON error whose useful part is +buried; these turn that into something a log line can carry, and decide +whether the failure is permanent (the refresh token is dead and the user must +sign in again) or worth retrying. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from crdt_sync._remote import RemoteSyncError + +if TYPE_CHECKING: + import requests as _requests + + +class FirebaseAuthError(RemoteSyncError): + """Raised for an authentication failure the caller must not ignore. + + A :class:`crdt_sync.RemoteSyncError` so callers that only care about + "sync is broken" can catch one type, while a settings screen can single + this out to say "your password is wrong" rather than "the network is + down". + """ + +def _reason(response: _requests.Response) -> str: + """Return Google's machine-readable error reason, in parentheses. + + Pulls ``INVALID_PASSWORD`` / ``TOKEN_EXPIRED`` / ``USER_DISABLED`` out of + the body so the raised message says what actually went wrong rather than + just reporting a status code. + """ + try: + data = response.json() + except ValueError: + # A non-JSON body (a proxy error page, say): the status code is all + # the detail there is. + return "" + if not isinstance(data, dict): + return "" + error = data.get("error") + if isinstance(error, dict): + return f" ({error.get('message')})" + if isinstance(error, str): + return f" ({error})" + return "" + + +#: Reasons that mean the refresh token is permanently dead. Only these clear +#: the session: a network error or a 5xx must not, because a device that signs +#: itself out whenever the wifi drops needs a manual sign-in to recover, which +#: is a worse failure than the stale banner this guards against. +_TERMINAL_AUTH_REASONS = ( + "TOKEN_EXPIRED", + "USER_DISABLED", + "USER_NOT_FOUND", + "INVALID_REFRESH_TOKEN", + "INVALID_GRANT_TYPE", + "MISSING_REFRESH_TOKEN", +) + + +def _is_revoked(error: FirebaseAuthError) -> bool: + """Return whether ``error`` means the refresh token is permanently dead.""" + message = str(error) + return any(reason in message for reason in _TERMINAL_AUTH_REASONS) diff --git a/crdt-sync/crdt_sync/_credentials.py b/crdt-sync/crdt_sync/_credentials.py new file mode 100644 index 0000000..af1b0df --- /dev/null +++ b/crdt-sync/crdt_sync/_credentials.py @@ -0,0 +1,146 @@ +"""Firebase credentials and the stores that persist them. + +Split from :mod:`crdt_sync._firebase_auth`, which keeps the token provider. +A credential set is an ID token, a refresh token and an expiry; the stores +decide where it survives a restart -- memory for tests, a mode-0600 file for +real installs. + +Re-exported from :mod:`crdt_sync._firebase_auth`, so existing imports work. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import datetime as dt +import json +import logging +import os +from pathlib import Path +from typing import Protocol + +_logger = logging.getLogger(__name__) + +# Refresh this long before the token actually expires. An ID token lives ~1h; +# refreshing early means a tick that starts just under the wire still finishes +# with a valid token rather than 401ing halfway through a multi-file push. +_REFRESH_SKEW = dt.timedelta(minutes=5) + + +@dataclass(frozen=True) +class FirebaseCredentials: + """One device's session: a short-lived ID token plus its refresh token. + + Attributes: + ---------- + id_token: + The bearer credential for Realtime Database requests. Expires fast. + refresh_token: + The long-lived credential. **This is the secret worth protecting.** + expires_at: + When ``id_token`` stops being accepted, in UTC. + """ + + id_token: str + refresh_token: str + expires_at: dt.datetime + + def is_expired_at(self, now: dt.datetime) -> bool: + """Return whether the token is expired, or close enough to it.""" + return now + _REFRESH_SKEW >= self.expires_at + + def to_json(self) -> dict[str, str]: + """Return a JSON-serializable form.""" + return { + "id_token": self.id_token, + "refresh_token": self.refresh_token, + "expires_at": self.expires_at.isoformat(), + } + + @classmethod + def from_json(cls, data: dict[str, str]) -> FirebaseCredentials: + """Rebuild credentials from :meth:`to_json` output.""" + return cls( + id_token=data["id_token"], + refresh_token=data["refresh_token"], + expires_at=dt.datetime.fromisoformat(data["expires_at"]), + ) + + +class CredentialStore(Protocol): + """Where a device keeps its credentials between runs.""" + + def load(self) -> FirebaseCredentials | None: + """Return the stored credentials, or ``None`` if not signed in.""" + + def save(self, credentials: FirebaseCredentials) -> None: + """Persist ``credentials``.""" + + def clear(self) -> None: + """Forget any stored credentials.""" + + +class MemoryCredentialStore: + """A :class:`CredentialStore` for tests and one-shot scripts.""" + + def __init__(self, credentials: FirebaseCredentials | None = None) -> None: + """Start holding ``credentials`` (or nothing).""" + self._credentials = credentials + + def load(self) -> FirebaseCredentials | None: + """Return the held credentials.""" + return self._credentials + + def save(self, credentials: FirebaseCredentials) -> None: + """Replace the held credentials.""" + self._credentials = credentials + + def clear(self) -> None: + """Drop the held credentials.""" + self._credentials = None + + +class FileCredentialStore: + """A :class:`CredentialStore` backed by a ``0600`` JSON file. + + Persistence is not an optimisation here: ``wake_alarm``'s PC side is a + fresh process every minute and ``diet_guard``'s every 15 minutes, so an + in-memory store would re-authenticate thousands of times a day. + """ + + def __init__(self, path: Path) -> None: + """Store credentials at ``path``, creating parent dirs as needed.""" + self._path = path + + def load(self) -> FirebaseCredentials | None: + """Return the stored credentials, or ``None`` if the file is absent. + + A corrupt or truncated file (an interrupted write) reads as "not + signed in" rather than raising: the caller's next step is to sign in + again, which repairs it. + """ + try: + data = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + try: + return FirebaseCredentials.from_json(data) + except (KeyError, TypeError, ValueError): + return None + + def save(self, credentials: FirebaseCredentials) -> None: + """Write ``credentials`` atomically, readable only by this user.""" + self._path.parent.mkdir(parents=True, exist_ok=True) + temp = self._path.with_suffix(f"{self._path.suffix}.tmp") + # Create with 0600 from the outset -- writing then chmod'ing would + # leave the refresh token world-readable for the gap in between. + temp.touch(mode=0o600) + temp.write_text(json.dumps(credentials.to_json()), encoding="utf-8") + temp.replace(self._path) + + def clear(self) -> None: + """Delete the credentials file if it exists.""" + self._path.unlink(missing_ok=True) + + +def _utcnow() -> dt.datetime: + return dt.datetime.now(dt.timezone.utc) diff --git a/crdt-sync/crdt_sync/_fbkeys.py b/crdt-sync/crdt_sync/_fbkeys.py new file mode 100644 index 0000000..c287903 --- /dev/null +++ b/crdt-sync/crdt_sync/_fbkeys.py @@ -0,0 +1,100 @@ +"""Escaping arbitrary strings into Firebase Realtime Database keys. + +Split from :mod:`crdt_sync._firebase`, which keeps the client. RTDB forbids +``.``, ``$``, ``#``, ``[``, ``]`` and ``/`` in a key, but the app stores +record ids and file paths that contain them, so every key is escaped on the +way in and unescaped on the way out. The pair has to round-trip exactly or a +device silently stops seeing its own writes. + +Re-exported from :mod:`crdt_sync._firebase`, so existing imports keep working. +""" + +from __future__ import annotations + +# Characters Realtime Database forbids in a key, plus ``~`` itself because it +# is this module's escape character. ``/`` is absent deliberately: it is the +# path separator, handled by splitting into segments before escaping. +_ESCAPES = { + "~": "~7E", + ".": "~2E", + "$": "~24", + "#": "~23", + "[": "~5B", + "]": "~5D", +} +_UNESCAPES = {escape: char for char, escape in _ESCAPES.items()} +_ESCAPE_LENGTH = 3 + + +def encode_key(segment: str) -> str: + """Escape one path segment into a legal Realtime Database key. + + RTDB rejects ``. $ # [ ] /`` in keys, and the REST API's trailing + ``.json`` is a *format suffix* rather than part of the path -- so a + filename like ``log.json`` cannot be stored verbatim. The mapping is a + reversible ``~XX`` escape (``log.json`` -> ``log~2Ejson``) rather than a + lossy "strip the extension", because callers see these names again: + todo-app lists *filenames*, not device directories, and must keep getting + ``.json`` back. + + ``~`` is the escape character because it is legal in RTDB keys and + unreserved in URLs, so the escaped form needs no percent-encoding -- + which the server would otherwise decode back into the illegal character. + + Parameters + ---------- + segment : str + One path segment, with no ``/`` in it. + + Returns: + ------- + str + The escaped key. + + """ + return "".join(_ESCAPES.get(char, char) for char in segment) + + +def decode_key(key: str) -> str: + """Reverse :func:`encode_key`. + + Parameters + ---------- + key : str + An escaped Realtime Database key. + + Returns: + ------- + str + The original segment. + + """ + out: list[str] = [] + index = 0 + while index < len(key): + escape = key[index : index + _ESCAPE_LENGTH] + char = _UNESCAPES.get(escape) + if char is None: + out.append(key[index]) + index += 1 + else: + out.append(char) + index += _ESCAPE_LENGTH + return "".join(out) + + +def encode_path(path: str) -> str: + """Escape every segment of a ``/``-separated logical path. + + Parameters + ---------- + path : str + A logical path such as ``"todo-sync/notes/abc.json"``. + + Returns: + ------- + str + The path with each segment escaped, empty segments dropped. + + """ + return "/".join(encode_key(part) for part in path.split("/") if part) diff --git a/crdt-sync/crdt_sync/_firebase.py b/crdt-sync/crdt_sync/_firebase.py index f44c315..a5a57a3 100644 --- a/crdt-sync/crdt_sync/_firebase.py +++ b/crdt-sync/crdt_sync/_firebase.py @@ -19,6 +19,7 @@ from typing import TYPE_CHECKING import requests as _requests +from crdt_sync._fbkeys import decode_key, encode_key, encode_path from crdt_sync._http import new_session from crdt_sync._remote import RemoteNotFoundError, RemoteSyncError @@ -34,19 +35,6 @@ _DEFAULT_TIMEOUT_SECONDS = 15 _HTTP_UNAUTHORIZED = 401 _HTTP_FORBIDDEN = 403 -# Characters Realtime Database forbids in a key, plus ``~`` itself because it -# is this module's escape character. ``/`` is absent deliberately: it is the -# path separator, handled by splitting into segments before escaping. -_ESCAPES = { - "~": "~7E", - ".": "~2E", - "$": "~24", - "#": "~23", - "[": "~5B", - "]": "~5D", -} -_UNESCAPES = {escape: char for char, escape in _ESCAPES.items()} -_ESCAPE_LENGTH = 3 class FirebaseSyncError(RemoteSyncError): @@ -63,80 +51,6 @@ class DatabaseNotFoundError(FirebaseSyncError, RemoteNotFoundError): """ -def encode_key(segment: str) -> str: - """Escape one path segment into a legal Realtime Database key. - - RTDB rejects ``. $ # [ ] /`` in keys, and the REST API's trailing - ``.json`` is a *format suffix* rather than part of the path -- so a - filename like ``log.json`` cannot be stored verbatim. The mapping is a - reversible ``~XX`` escape (``log.json`` -> ``log~2Ejson``) rather than a - lossy "strip the extension", because callers see these names again: - todo-app lists *filenames*, not device directories, and must keep getting - ``.json`` back. - - ``~`` is the escape character because it is legal in RTDB keys and - unreserved in URLs, so the escaped form needs no percent-encoding -- - which the server would otherwise decode back into the illegal character. - - Parameters - ---------- - segment : str - One path segment, with no ``/`` in it. - - Returns: - ------- - str - The escaped key. - - """ - return "".join(_ESCAPES.get(char, char) for char in segment) - - -def decode_key(key: str) -> str: - """Reverse :func:`encode_key`. - - Parameters - ---------- - key : str - An escaped Realtime Database key. - - Returns: - ------- - str - The original segment. - - """ - out: list[str] = [] - index = 0 - while index < len(key): - escape = key[index : index + _ESCAPE_LENGTH] - char = _UNESCAPES.get(escape) - if char is None: - out.append(key[index]) - index += 1 - else: - out.append(char) - index += _ESCAPE_LENGTH - return "".join(out) - - -def encode_path(path: str) -> str: - """Escape every segment of a ``/``-separated logical path. - - Parameters - ---------- - path : str - A logical path such as ``"todo-sync/notes/abc.json"``. - - Returns: - ------- - str - The path with each segment escaped, empty segments dropped. - - """ - return "/".join(encode_key(part) for part in path.split("/") if part) - - class FirebaseSyncClient: """Realtime Database seen through the same contract as GitHub. diff --git a/crdt-sync/crdt_sync/_firebase_auth.py b/crdt-sync/crdt_sync/_firebase_auth.py index 808f3b1..a3d2b14 100644 --- a/crdt-sync/crdt_sync/_firebase_auth.py +++ b/crdt-sync/crdt_sync/_firebase_auth.py @@ -19,8 +19,15 @@ from typing import TYPE_CHECKING, Protocol import requests as _requests +from crdt_sync._autherrors import FirebaseAuthError, _is_revoked, _reason +from crdt_sync._credentials import ( + CredentialStore, + FileCredentialStore, + FirebaseCredentials, + MemoryCredentialStore, + _utcnow, +) from crdt_sync._http import new_session -from crdt_sync._remote import RemoteSyncError # Bound to the name ``requests`` so the call sites below -- and the tests # that patch ``.requests.`` -- are unchanged by pooling. @@ -34,160 +41,6 @@ _SIGN_IN_BASE = "https://identitytoolkit.googleapis.com/v1" _REFRESH_BASE = "https://securetoken.googleapis.com/v1" _DEFAULT_TIMEOUT_SECONDS = 15 -# Refresh this long before the token actually expires. An ID token lives ~1h; -# refreshing early means a tick that starts just under the wire still finishes -# with a valid token rather than 401ing halfway through a multi-file push. -_REFRESH_SKEW = dt.timedelta(minutes=5) - - -class FirebaseAuthError(RemoteSyncError): - """Raised for an authentication failure the caller must not ignore. - - A :class:`crdt_sync.RemoteSyncError` so callers that only care about - "sync is broken" can catch one type, while a settings screen can single - this out to say "your password is wrong" rather than "the network is - down". - """ - - -@dataclass(frozen=True) -class FirebaseCredentials: - """One device's session: a short-lived ID token plus its refresh token. - - Attributes: - ---------- - id_token: - The bearer credential for Realtime Database requests. Expires fast. - refresh_token: - The long-lived credential. **This is the secret worth protecting.** - expires_at: - When ``id_token`` stops being accepted, in UTC. - """ - - id_token: str - refresh_token: str - expires_at: dt.datetime - - def is_expired_at(self, now: dt.datetime) -> bool: - """Return whether the token is expired, or close enough to it.""" - return now + _REFRESH_SKEW >= self.expires_at - - def to_json(self) -> dict[str, str]: - """Return a JSON-serializable form.""" - return { - "id_token": self.id_token, - "refresh_token": self.refresh_token, - "expires_at": self.expires_at.isoformat(), - } - - @classmethod - def from_json(cls, data: dict[str, str]) -> FirebaseCredentials: - """Rebuild credentials from :meth:`to_json` output.""" - return cls( - id_token=data["id_token"], - refresh_token=data["refresh_token"], - expires_at=dt.datetime.fromisoformat(data["expires_at"]), - ) - - -class CredentialStore(Protocol): - """Where a device keeps its credentials between runs.""" - - def load(self) -> FirebaseCredentials | None: - """Return the stored credentials, or ``None`` if not signed in.""" - - def save(self, credentials: FirebaseCredentials) -> None: - """Persist ``credentials``.""" - - def clear(self) -> None: - """Forget any stored credentials.""" - - -class MemoryCredentialStore: - """A :class:`CredentialStore` for tests and one-shot scripts.""" - - def __init__(self, credentials: FirebaseCredentials | None = None) -> None: - """Start holding ``credentials`` (or nothing).""" - self._credentials = credentials - - def load(self) -> FirebaseCredentials | None: - """Return the held credentials.""" - return self._credentials - - def save(self, credentials: FirebaseCredentials) -> None: - """Replace the held credentials.""" - self._credentials = credentials - - def clear(self) -> None: - """Drop the held credentials.""" - self._credentials = None - - -class FileCredentialStore: - """A :class:`CredentialStore` backed by a ``0600`` JSON file. - - Persistence is not an optimisation here: ``wake_alarm``'s PC side is a - fresh process every minute and ``diet_guard``'s every 15 minutes, so an - in-memory store would re-authenticate thousands of times a day. - """ - - def __init__(self, path: Path) -> None: - """Store credentials at ``path``, creating parent dirs as needed.""" - self._path = path - - def load(self) -> FirebaseCredentials | None: - """Return the stored credentials, or ``None`` if the file is absent. - - A corrupt or truncated file (an interrupted write) reads as "not - signed in" rather than raising: the caller's next step is to sign in - again, which repairs it. - """ - try: - data = json.loads(self._path.read_text(encoding="utf-8")) - except (OSError, ValueError): - return None - try: - return FirebaseCredentials.from_json(data) - except (KeyError, TypeError, ValueError): - return None - - def save(self, credentials: FirebaseCredentials) -> None: - """Write ``credentials`` atomically, readable only by this user.""" - self._path.parent.mkdir(parents=True, exist_ok=True) - temp = self._path.with_suffix(f"{self._path.suffix}.tmp") - # Create with 0600 from the outset -- writing then chmod'ing would - # leave the refresh token world-readable for the gap in between. - temp.touch(mode=0o600) - temp.write_text(json.dumps(credentials.to_json()), encoding="utf-8") - temp.replace(self._path) - - def clear(self) -> None: - """Delete the credentials file if it exists.""" - self._path.unlink(missing_ok=True) - - -def _utcnow() -> dt.datetime: - return dt.datetime.now(dt.timezone.utc) - - -#: Reasons that mean the refresh token is permanently dead. Only these clear -#: the session: a network error or a 5xx must not, because a device that signs -#: itself out whenever the wifi drops needs a manual sign-in to recover, which -#: is a worse failure than the stale banner this guards against. -_TERMINAL_AUTH_REASONS = ( - "TOKEN_EXPIRED", - "USER_DISABLED", - "USER_NOT_FOUND", - "INVALID_REFRESH_TOKEN", - "INVALID_GRANT_TYPE", - "MISSING_REFRESH_TOKEN", -) - - -def _is_revoked(error: FirebaseAuthError) -> bool: - """Return whether ``error`` means the refresh token is permanently dead.""" - message = str(error) - return any(reason in message for reason in _TERMINAL_AUTH_REASONS) class FirebaseTokenProvider: @@ -387,26 +240,3 @@ class FirebaseTokenProvider: msg = f"failed to {what}: HTTP {response.status_code}{reason}" raise FirebaseAuthError(msg) return response.json() - - -def _reason(response: _requests.Response) -> str: - """Return Google's machine-readable error reason, in parentheses. - - Pulls ``INVALID_PASSWORD`` / ``TOKEN_EXPIRED`` / ``USER_DISABLED`` out of - the body so the raised message says what actually went wrong rather than - just reporting a status code. - """ - try: - data = response.json() - except ValueError: - # A non-JSON body (a proxy error page, say): the status code is all - # the detail there is. - return "" - if not isinstance(data, dict): - return "" - error = data.get("error") - if isinstance(error, dict): - return f" ({error.get('message')})" - if isinstance(error, str): - return f" ({error})" - return "" diff --git a/crdt-sync/crdt_sync/_gherrors.py b/crdt-sync/crdt_sync/_gherrors.py new file mode 100644 index 0000000..e11667c --- /dev/null +++ b/crdt-sync/crdt_sync/_gherrors.py @@ -0,0 +1,30 @@ +"""The errors the GitHub sync backend raises. + +Split from :mod:`crdt_sync._github` for the 250-line cap. Kept separate so a +caller can catch them without importing the client -- and so the client file +is the client, not its taxonomy. + +Re-exported from :mod:`crdt_sync._github`, so existing imports keep working. +""" + +from __future__ import annotations + +from crdt_sync._remote import RemoteNotFoundError, RemoteSyncError + + +class GitHubSyncError(RemoteSyncError): + """Raised for a GitHub API failure the caller must not silently ignore.""" + + +class RepoNotFoundError(GitHubSyncError, RemoteNotFoundError): + """Raised when the configured repo itself is unreachable. + + Distinguished from a path-404 (nothing pushed to that path yet, which is + benign -- it just means no other device has synced before) so the caller + can tell "the repo name is wrong or the token isn't scoped to it" apart + from "no other device has synced yet". + + Derives from both so existing ``except GitHubSyncError`` handlers still + cover it *and* backend-neutral callers can catch "the remote itself is + missing" without naming GitHub. + """ diff --git a/crdt-sync/crdt_sync/_github.py b/crdt-sync/crdt_sync/_github.py index 3c3ad47..1ca24b8 100644 --- a/crdt-sync/crdt_sync/_github.py +++ b/crdt-sync/crdt_sync/_github.py @@ -13,6 +13,7 @@ import base64 import requests as _requests +from crdt_sync._gherrors import GitHubSyncError, RepoNotFoundError from crdt_sync._http import new_session from crdt_sync._remote import RemoteNotFoundError, RemoteSyncError @@ -26,24 +27,6 @@ _HTTP_NOT_FOUND = 404 _DEFAULT_TIMEOUT_SECONDS = 15 -class GitHubSyncError(RemoteSyncError): - """Raised for a GitHub API failure the caller must not silently ignore.""" - - -class RepoNotFoundError(GitHubSyncError, RemoteNotFoundError): - """Raised when the configured repo itself is unreachable. - - Distinguished from a path-404 (nothing pushed to that path yet, which is - benign -- it just means no other device has synced before) so the caller - can tell "the repo name is wrong or the token isn't scoped to it" apart - from "no other device has synced yet". - - Derives from both so existing ``except GitHubSyncError`` handlers still - cover it *and* backend-neutral callers can catch "the remote itself is - missing" without naming GitHub. - """ - - class GitHubSyncClient: """Thin wrapper around the subset of the Contents API sync needs.""" diff --git a/crdt-sync/crdt_sync/_pull.py b/crdt-sync/crdt_sync/_pull.py new file mode 100644 index 0000000..9b85de2 --- /dev/null +++ b/crdt-sync/crdt_sync/_pull.py @@ -0,0 +1,109 @@ +"""Downloading peers' logs, skipping the ones we already have. + +Split from :mod:`crdt_sync._sync`, which keeps the push half and the +orchestration. The revision map is the optimisation that makes a tick cheap: +a peer whose published revision matches what we last merged is not fetched at +all, so a quiet device costs one small read instead of one read per peer. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + + from crdt_sync._log import Log + from crdt_sync._remote import RemoteStore + +from crdt_sync._revisions import revision_of +from crdt_sync._syncstate import SyncState + +_logger = logging.getLogger(__name__) + + +def _remote_revs(client: RemoteStore, revs_path: str) -> dict[str, str]: + """Return every peer's published revision, cheaply where possible. + + Degrades to an empty map -- meaning "fetch everything", the old behaviour + -- on any backend without a bulk-map read, so correctness never depends + on the optimisation being available. + """ + get_string_map = getattr(client, "get_string_map", None) + if get_string_map is None: + return {} + return get_string_map(revs_path) + + +@dataclass(frozen=True) +class _PullContext: + """Everything :func:`_pull_remote_logs` needs for one tick's pull. + + A bundle rather than eight positional parameters: they are always passed + together, from one call site, and several share a type -- exactly the + shape where a positional mix-up is silent. + """ + + client: RemoteStore + own_ids: frozenset[str] + path_prefix: str + filename: str + decode: Callable[[str], Log] + remote_revs: dict[str, str] + state: SyncState + + +def _pull_remote_logs( + ctx: _PullContext, + seen_revs: dict[str, str], +) -> list[Log]: + """Return every other device's last-pushed log, skipping this one. + + A device whose pushed file is corrupt or unparsable (e.g. an interrupted + push) is logged and skipped, same as one that has never pushed at all -- + the remote is an external system boundary, and one bad device's file must + not stall merging in every other device's. + + ``own_ids`` is a *set* rather than a single id because a device that has + migrated from a role constant to a persisted uuid still owns the log it + pushed under the old id. Matching only the current id would pull that + file back and re-merge this device's own pre-migration history as though + a peer had written it -- idempotent, but pure wasted transfer. + + Records into ``seen_revs`` which peers are now merged in, so the next + tick can skip them. + """ + remote_logs: list[Log] = [] + for other_device_id in ctx.client.list_directory(ctx.path_prefix): + if other_device_id in ctx.own_ids: + continue + remote_rev = ctx.remote_revs.get(other_device_id) + if remote_rev is not None and remote_rev == ctx.state.peer_revs.get( + other_device_id + ): + # Unchanged since we last merged it, and that merge is already + # part of local_log -- so the (potentially hundreds of KB) + # download is pure waste. Carry the revision forward so it stays + # skipped next tick. + seen_revs[other_device_id] = remote_rev + continue + text = ctx.client.get_file_text( + f"{ctx.path_prefix}/{other_device_id}/{ctx.filename}" + ) + if text is None: + continue + try: + remote_logs.append(ctx.decode(text)) + except (ValueError, KeyError, TypeError): + _logger.warning( + "Unparsable log pushed by device %r, skipping", + other_device_id, + ) + # Deliberately not recorded as seen: a corrupt push must be + # retried next tick, not remembered as merged. + continue + seen_revs[other_device_id] = remote_rev or revision_of(text) + return remote_logs diff --git a/crdt-sync/crdt_sync/_revisions.py b/crdt-sync/crdt_sync/_revisions.py new file mode 100644 index 0000000..e14cc7d --- /dev/null +++ b/crdt-sync/crdt_sync/_revisions.py @@ -0,0 +1,36 @@ +"""The revision string a device publishes alongside its log. + +Split from :mod:`crdt_sync._sync` so both the push and pull halves can import +it without importing each other. Pure: a revision is a function of the encoded +log and nothing else, which is what lets two devices agree on whether they are +looking at the same bytes. + +Re-exported from :mod:`crdt_sync._sync`, so existing imports keep working. +""" + +from __future__ import annotations + +import hashlib + +def revision_of(encoded_log: str) -> str: + """Return the revision of a serialized log: a content hash. + + A hash rather than the log's maximum HLC because merging a peer's *older* + record changes the content without raising the maximum -- with three + devices in one namespace (diet-guard has ``pc``, ``phone`` and + ``desktop``) a clock-based revision can miss a peer's merged state. It is + also the same value used to suppress no-op pushes, so the two + optimisations share one mechanism. + + Parameters + ---------- + encoded_log : str + The serialized log, exactly as it would be pushed. + + Returns: + ------- + str + A hex SHA-256 digest. + + """ + return hashlib.sha256(encoded_log.encode("utf-8")).hexdigest() diff --git a/crdt-sync/crdt_sync/_sync.py b/crdt-sync/crdt_sync/_sync.py index 06d6af1..d749bbe 100644 --- a/crdt-sync/crdt_sync/_sync.py +++ b/crdt-sync/crdt_sync/_sync.py @@ -26,147 +26,20 @@ if TYPE_CHECKING: from crdt_sync._log import Log from crdt_sync._remote import RemoteStore +from crdt_sync._pull import _PullContext, _pull_remote_logs, _remote_revs +from crdt_sync._revisions import revision_of +from crdt_sync._syncstate import ( + FileSyncStateStore, + MemorySyncStateStore, + SyncState, + SyncStateStore, +) + _logger = logging.getLogger(__name__) _DEFAULT_FILENAME = "log.json" -@dataclass(frozen=True) -class SyncState: - """What one device remembers between ticks to skip needless traffic. - - Attributes: - ---------- - pushed_rev: - Hash of what this device last pushed, so an unchanged log is not - re-uploaded. 88% of the pushes in the GitHub-backed history this - replaces were byte-identical no-ops. - peer_revs: - The hash each peer had when we last downloaded it, so an unchanged - peer is not re-downloaded. - - Notes: - ----- - **Must be stored next to the log itself and cleared with it.** Skipping - an unchanged peer is only sound because that peer's records are already - merged into the local log; a cache that outlived its log would skip peers - whose data had been lost. - - """ - - pushed_rev: str | None = None - peer_revs: dict[str, str] = field(default_factory=dict) - - def to_json(self) -> dict[str, object]: - """Return a JSON-serializable form.""" - return {"pushed_rev": self.pushed_rev, "peer_revs": self.peer_revs} - - @classmethod - def from_json(cls, data: dict[str, object]) -> SyncState: - """Rebuild state from :meth:`to_json` output, ignoring junk. - - A malformed cache degrades into "fetch everything" rather than - raising: it is an optimisation, never a source of failure. - """ - pushed = data.get("pushed_rev") - peers = data.get("peer_revs") - return cls( - pushed_rev=pushed if isinstance(pushed, str) else None, - peer_revs={ - key: value - for key, value in (peers or {}).items() - if isinstance(value, str) - } - if isinstance(peers, dict) - else {}, - ) - - -class SyncStateStore(Protocol): - """Where :func:`sync_log` persists its :class:`SyncState` between runs.""" - - def load(self) -> SyncState: - """Return the stored state, or a default one.""" - - def save(self, state: SyncState) -> None: - """Persist ``state``.""" - - -class MemorySyncStateStore: - """A :class:`SyncStateStore` that forgets on exit, for tests. - - Correct but pessimistic: every process re-downloads every peer and - re-pushes the local log, which is exactly the old behaviour. - """ - - def __init__(self) -> None: - """Start with empty state.""" - self._state = SyncState() - - def load(self) -> SyncState: - """Return the held state.""" - return self._state - - def save(self, state: SyncState) -> None: - """Replace the held state.""" - self._state = state - - -class FileSyncStateStore: - """A :class:`SyncStateStore` backed by a JSON file. - - Persistence is what makes the saving real for short-lived callers: - ``wake_alarm``'s PC side is a fresh process every minute and - ``diet_guard``'s every 15 minutes, so an in-memory store would save - nothing at all. - """ - - def __init__(self, path: Path) -> None: - """Keep state in the file at ``path``.""" - self._path = path - - def load(self) -> SyncState: - """Return the stored state, or a default one if unreadable.""" - try: - data = json.loads(self._path.read_text(encoding="utf-8")) - except (OSError, ValueError): - return SyncState() - if not isinstance(data, dict): - return SyncState() - return SyncState.from_json(data) - - def save(self, state: SyncState) -> None: - """Write ``state`` atomically, so a crash cannot truncate it.""" - self._path.parent.mkdir(parents=True, exist_ok=True) - temp = self._path.with_suffix(f"{self._path.suffix}.tmp") - temp.write_text(json.dumps(state.to_json()), encoding="utf-8") - temp.replace(self._path) - - -def revision_of(encoded_log: str) -> str: - """Return the revision of a serialized log: a content hash. - - A hash rather than the log's maximum HLC because merging a peer's *older* - record changes the content without raising the maximum -- with three - devices in one namespace (diet-guard has ``pc``, ``phone`` and - ``desktop``) a clock-based revision can miss a peer's merged state. It is - also the same value used to suppress no-op pushes, so the two - optimisations share one mechanism. - - Parameters - ---------- - encoded_log : str - The serialized log, exactly as it would be pushed. - - Returns: - ------- - str - A hex SHA-256 digest. - - """ - return hashlib.sha256(encoded_log.encode("utf-8")).hexdigest() - - def default_revs_path(path_prefix: str) -> str: """Return where revisions live for ``path_prefix``. @@ -189,90 +62,6 @@ def default_revs_path(path_prefix: str) -> str: return f"{head}/revs" if separator else f"{path_prefix}/revs" -def _remote_revs(client: RemoteStore, revs_path: str) -> dict[str, str]: - """Return every peer's published revision, cheaply where possible. - - Degrades to an empty map -- meaning "fetch everything", the old behaviour - -- on any backend without a bulk-map read, so correctness never depends - on the optimisation being available. - """ - get_string_map = getattr(client, "get_string_map", None) - if get_string_map is None: - return {} - return get_string_map(revs_path) - - -@dataclass(frozen=True) -class _PullContext: - """Everything :func:`_pull_remote_logs` needs for one tick's pull. - - A bundle rather than eight positional parameters: they are always passed - together, from one call site, and several share a type -- exactly the - shape where a positional mix-up is silent. - """ - - client: RemoteStore - own_ids: frozenset[str] - path_prefix: str - filename: str - decode: Callable[[str], Log] - remote_revs: dict[str, str] - state: SyncState - - -def _pull_remote_logs( - ctx: _PullContext, - seen_revs: dict[str, str], -) -> list[Log]: - """Return every other device's last-pushed log, skipping this one. - - A device whose pushed file is corrupt or unparsable (e.g. an interrupted - push) is logged and skipped, same as one that has never pushed at all -- - the remote is an external system boundary, and one bad device's file must - not stall merging in every other device's. - - ``own_ids`` is a *set* rather than a single id because a device that has - migrated from a role constant to a persisted uuid still owns the log it - pushed under the old id. Matching only the current id would pull that - file back and re-merge this device's own pre-migration history as though - a peer had written it -- idempotent, but pure wasted transfer. - - Records into ``seen_revs`` which peers are now merged in, so the next - tick can skip them. - """ - remote_logs: list[Log] = [] - for other_device_id in ctx.client.list_directory(ctx.path_prefix): - if other_device_id in ctx.own_ids: - continue - remote_rev = ctx.remote_revs.get(other_device_id) - if remote_rev is not None and remote_rev == ctx.state.peer_revs.get( - other_device_id - ): - # Unchanged since we last merged it, and that merge is already - # part of local_log -- so the (potentially hundreds of KB) - # download is pure waste. Carry the revision forward so it stays - # skipped next tick. - seen_revs[other_device_id] = remote_rev - continue - text = ctx.client.get_file_text( - f"{ctx.path_prefix}/{other_device_id}/{ctx.filename}" - ) - if text is None: - continue - try: - remote_logs.append(ctx.decode(text)) - except (ValueError, KeyError, TypeError): - _logger.warning( - "Unparsable log pushed by device %r, skipping", - other_device_id, - ) - # Deliberately not recorded as seen: a corrupt push must be - # retried next tick, not remembered as merged. - continue - seen_revs[other_device_id] = remote_rev or revision_of(text) - return remote_logs - - @dataclass(frozen=True) class SyncTarget: """Which remote, which device, and where its logs live. diff --git a/crdt-sync/crdt_sync/_syncstate.py b/crdt-sync/crdt_sync/_syncstate.py new file mode 100644 index 0000000..159791e --- /dev/null +++ b/crdt-sync/crdt_sync/_syncstate.py @@ -0,0 +1,131 @@ +"""Where a device records what it has already pushed and pulled. + +Split from :mod:`crdt_sync._sync`, which keeps the sync itself. The state is +what makes a tick cheap: without it every run re-downloads every peer, and +with a stale one a device silently stops seeing updates. Two stores ship -- +in-memory for tests, and a JSON file for real installs. + +Re-exported from :mod:`crdt_sync._sync`, so existing imports keep working. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import json +import logging +from pathlib import Path +from typing import Protocol + +_logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class SyncState: + """What one device remembers between ticks to skip needless traffic. + + Attributes: + ---------- + pushed_rev: + Hash of what this device last pushed, so an unchanged log is not + re-uploaded. 88% of the pushes in the GitHub-backed history this + replaces were byte-identical no-ops. + peer_revs: + The hash each peer had when we last downloaded it, so an unchanged + peer is not re-downloaded. + + Notes: + ----- + **Must be stored next to the log itself and cleared with it.** Skipping + an unchanged peer is only sound because that peer's records are already + merged into the local log; a cache that outlived its log would skip peers + whose data had been lost. + + """ + + pushed_rev: str | None = None + peer_revs: dict[str, str] = field(default_factory=dict) + + def to_json(self) -> dict[str, object]: + """Return a JSON-serializable form.""" + return {"pushed_rev": self.pushed_rev, "peer_revs": self.peer_revs} + + @classmethod + def from_json(cls, data: dict[str, object]) -> SyncState: + """Rebuild state from :meth:`to_json` output, ignoring junk. + + A malformed cache degrades into "fetch everything" rather than + raising: it is an optimisation, never a source of failure. + """ + pushed = data.get("pushed_rev") + peers = data.get("peer_revs") + return cls( + pushed_rev=pushed if isinstance(pushed, str) else None, + peer_revs={ + key: value + for key, value in (peers or {}).items() + if isinstance(value, str) + } + if isinstance(peers, dict) + else {}, + ) + + +class SyncStateStore(Protocol): + """Where :func:`sync_log` persists its :class:`SyncState` between runs.""" + + def load(self) -> SyncState: + """Return the stored state, or a default one.""" + + def save(self, state: SyncState) -> None: + """Persist ``state``.""" + + +class MemorySyncStateStore: + """A :class:`SyncStateStore` that forgets on exit, for tests. + + Correct but pessimistic: every process re-downloads every peer and + re-pushes the local log, which is exactly the old behaviour. + """ + + def __init__(self) -> None: + """Start with empty state.""" + self._state = SyncState() + + def load(self) -> SyncState: + """Return the held state.""" + return self._state + + def save(self, state: SyncState) -> None: + """Replace the held state.""" + self._state = state + + +class FileSyncStateStore: + """A :class:`SyncStateStore` backed by a JSON file. + + Persistence is what makes the saving real for short-lived callers: + ``wake_alarm``'s PC side is a fresh process every minute and + ``diet_guard``'s every 15 minutes, so an in-memory store would save + nothing at all. + """ + + def __init__(self, path: Path) -> None: + """Keep state in the file at ``path``.""" + self._path = path + + def load(self) -> SyncState: + """Return the stored state, or a default one if unreadable.""" + try: + data = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return SyncState() + if not isinstance(data, dict): + return SyncState() + return SyncState.from_json(data) + + def save(self, state: SyncState) -> None: + """Write ``state`` atomically, so a crash cannot truncate it.""" + self._path.parent.mkdir(parents=True, exist_ok=True) + temp = self._path.with_suffix(f"{self._path.suffix}.tmp") + temp.write_text(json.dumps(state.to_json()), encoding="utf-8") + temp.replace(self._path) diff --git a/crdt-sync/tool/_oauth_callback.py b/crdt-sync/tool/_oauth_callback.py new file mode 100644 index 0000000..326c513 --- /dev/null +++ b/crdt-sync/tool/_oauth_callback.py @@ -0,0 +1,109 @@ +"""The one-shot local HTTP server that catches Google's OAuth redirect. + +Split from :mod:`google_id_token`, which keeps the token exchange. Google +redirects the browser back to ``http://localhost:`` with the code in the +query string, so something has to be listening on a free port for exactly one +request and then stop. +""" + +from __future__ import annotations + +import http.server +import time +import urllib.parse +import logging +import socket +import threading +from typing import TYPE_CHECKING, ClassVar +from urllib.parse import parse_qs, urlparse + +if TYPE_CHECKING: + from collections.abc import Callable + +_logger = logging.getLogger(__name__) + +_CONSENT_TIMEOUT_SECONDS = 300 + + +class _CallbackHandler(http.server.BaseHTTPRequestHandler): + """Captures the ``code`` (or ``error``) Google redirects back with.""" + + # Set by the server owner before serving. Class attributes because + # BaseHTTPRequestHandler is instantiated per request by the server, so + # there is no instance to hand them to. + expected_state: ClassVar[str] = "" + result: ClassVar[dict[str, str]] = {} + + def do_GET(self) -> None: + """Record the query parameters and show a closable page. + + Requests that carry neither a ``code`` nor an ``error`` are answered + but *not* recorded: browsers fetch ``/favicon.ico`` against the + redirect origin, and letting that count as the callback ends the flow + before Google ever redirects, which surfaces as a consent timeout. + """ + query = urllib.parse.urlparse(self.path).query + params = {k: v[0] for k, v in urllib.parse.parse_qs(query).items()} + if "code" not in params and "error" not in params: + self.send_response(204) + self.send_header("Content-Length", "0") + self.end_headers() + return + type(self).result = params + + ok = "code" in params and params.get("state") == self.expected_state + body = ( + b"

Done - you can close this tab.

" + if ok + else b"

Authorization failed.

" + ) + self.send_response(200 if ok else 400) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args: object) -> None: + """Silence the default stderr request logging.""" + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def _start_callback_server( + port: int, + state: str, +) -> tuple[http.server.HTTPServer, threading.Thread]: + """Listen on ``port`` until the OAuth callback arrives. + + Serves in a loop rather than handling exactly one request: the first + request to reach this port is often a favicon fetch or a speculative + preconnect, and retiring the server on it leaves nothing listening when + Google finally redirects -- which surfaces as a consent timeout. + + Args: + port: The loopback port to bind. + state: The value the callback must echo back. + + Returns: + The server and the thread serving it, so the caller can join and close. + """ + _CallbackHandler.expected_state = state + _CallbackHandler.result = {} + server = http.server.HTTPServer(("127.0.0.1", port), _CallbackHandler) + # Bounded so handle_request() returns periodically instead of blocking + # forever; without it the loop could not notice the deadline and the + # daemon thread would sit on the socket after the caller gave up. + server.timeout = 1.0 + deadline = time.monotonic() + _CONSENT_TIMEOUT_SECONDS + + def _serve_until_callback() -> None: + while not _CallbackHandler.result and time.monotonic() < deadline: + server.handle_request() + + thread = threading.Thread(target=_serve_until_callback, daemon=True) + thread.start() + return server, thread diff --git a/crdt-sync/tool/google_id_token.py b/crdt-sync/tool/google_id_token.py index 70e85c5..9c92737 100644 --- a/crdt-sync/tool/google_id_token.py +++ b/crdt-sync/tool/google_id_token.py @@ -62,99 +62,22 @@ _TOKEN_EXCHANGE_URL = f"{_OAUTH_HOST}/token" _SCOPES = "openid email profile" _TIMEOUT_SECONDS = 30 -_CONSENT_TIMEOUT_SECONDS = 300 _logger = logging.getLogger("google_id_token") +from tool._oauth_callback import ( + _CONSENT_TIMEOUT_SECONDS, + _CallbackHandler, + _free_port, + _start_callback_server, +) + + class TokenError(Exception): """The authorization flow did not produce an ID token.""" -class _CallbackHandler(http.server.BaseHTTPRequestHandler): - """Captures the ``code`` (or ``error``) Google redirects back with.""" - - # Set by the server owner before serving. Class attributes because - # BaseHTTPRequestHandler is instantiated per request by the server, so - # there is no instance to hand them to. - expected_state: ClassVar[str] = "" - result: ClassVar[dict[str, str]] = {} - - def do_GET(self) -> None: - """Record the query parameters and show a closable page. - - Requests that carry neither a ``code`` nor an ``error`` are answered - but *not* recorded: browsers fetch ``/favicon.ico`` against the - redirect origin, and letting that count as the callback ends the flow - before Google ever redirects, which surfaces as a consent timeout. - """ - query = urllib.parse.urlparse(self.path).query - params = {k: v[0] for k, v in urllib.parse.parse_qs(query).items()} - if "code" not in params and "error" not in params: - self.send_response(204) - self.send_header("Content-Length", "0") - self.end_headers() - return - type(self).result = params - - ok = "code" in params and params.get("state") == self.expected_state - body = ( - b"

Done - you can close this tab.

" - if ok - else b"

Authorization failed.

" - ) - self.send_response(200 if ok else 400) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def log_message(self, *args: object) -> None: - """Silence the default stderr request logging.""" - - -def _free_port() -> int: - with socket.socket() as sock: - sock.bind(("127.0.0.1", 0)) - return int(sock.getsockname()[1]) - - -def _start_callback_server( - port: int, - state: str, -) -> tuple[http.server.HTTPServer, threading.Thread]: - """Listen on ``port`` until the OAuth callback arrives. - - Serves in a loop rather than handling exactly one request: the first - request to reach this port is often a favicon fetch or a speculative - preconnect, and retiring the server on it leaves nothing listening when - Google finally redirects -- which surfaces as a consent timeout. - - Args: - port: The loopback port to bind. - state: The value the callback must echo back. - - Returns: - The server and the thread serving it, so the caller can join and close. - """ - _CallbackHandler.expected_state = state - _CallbackHandler.result = {} - server = http.server.HTTPServer(("127.0.0.1", port), _CallbackHandler) - # Bounded so handle_request() returns periodically instead of blocking - # forever; without it the loop could not notice the deadline and the - # daemon thread would sit on the socket after the caller gave up. - server.timeout = 1.0 - deadline = time.monotonic() + _CONSENT_TIMEOUT_SECONDS - - def _serve_until_callback() -> None: - while not _CallbackHandler.result and time.monotonic() < deadline: - server.handle_request() - - thread = threading.Thread(target=_serve_until_callback, daemon=True) - thread.start() - return server, thread - - def fetch_id_token( client_id: str, client_secret: str,