Initial gRPC server implementation

light_api/light_api/worker.py

@@ -0,0 +1,132 @@
+"""Run a `Light` session on a dedicated background thread.
+
+`Light` is not thread-safe, so any consumer with multiple threads must funnel every
+call through a single owner. `LightThread` is that owner: `submit(fn)` hands `fn(light)`
+to the worker thread and blocks for the result.
+"""
+
+from __future__ import annotations
+
+import queue
+import threading
+from concurrent.futures import Future
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from light_api.client import Light
+
+
+@dataclass
+class LightConfig:
+    """Everything needed to construct a `Light` session.
+
+    Mirrors `Light.__init__`; unset fields fall back to files/env/keyring exactly as they do there.
+    """
+
+    email: str | None = None
+    email_file: str | None = None
+    password: str | None = None
+    password_file: str | None = None
+    phone: str | None = None
+    phone_file: str | None = None
+    device_id: str | None = None
+    device_id_file: str | None = None
+    cache_enabled: bool = False
+    password_prompt: Callable[[], str] | None = None
+
+
+class LightThread:
+    """Owns a `Light` instance on a background thread. Work is submitted via `submit`."""
+
+    def __init__(self, config: LightConfig) -> None:
+        """Initialize the thread."""
+        self._config = config
+
+        # queue of work to do
+        self._queue: queue.Queue[tuple[Callable[[Light], Any], Future[Any]] | None] = (
+            queue.Queue()
+        )
+
+        self._thread = threading.Thread(target=self._run, daemon=True)
+        self._ready = threading.Event()
+        self._error: BaseException | None = None
+        self._lock = threading.Lock()
+        self._stopped = False
+
+    def start(self) -> None:
+        """Start the thread and block until the session finishes its auth attempt.
+
+        Raises:
+            Whatever the session setup raised (bad credentials, network, ...)
+        """
+        self._thread.start()
+        self._ready.wait()
+        if self._error is not None:
+            raise self._error
+
+    def _run(self) -> None:
+        try:
+            with Light(
+                email=self._config.email,
+                email_file=self._config.email_file,
+                password=self._config.password,
+                password_file=self._config.password_file,
+                phone=self._config.phone,
+                phone_file=self._config.phone_file,
+                device_id=self._config.device_id,
+                device_id_file=self._config.device_id_file,
+                cache_enabled=self._config.cache_enabled,
+                password_prompt=self._config.password_prompt,
+            ) as light:
+                self._ready.set()
+                while True:
+                    item = self._queue.get()
+                    if item is None:
+                        break
+                    func, future = item
+                    try:
+                        future.set_result(func(light))
+                    except BaseException as e:
+                        future.set_exception(e)
+        except BaseException as e:
+            self._error = e
+            self._ready.set()
+        finally:
+            self._fail_pending()
+
+    def _fail_pending(self) -> None:
+        """Empties queue (fails pending items) when worker is done to prevent callers from hanging."""
+        with self._lock:
+            self._stopped = True
+            while True:
+                try:
+                    item = self._queue.get_nowait()
+                except queue.Empty:
+                    break
+
+                if item is not None:
+                    _, future = item
+                    if not future.done():
+                        future.set_exception(
+                            RuntimeError("LightThread worker has stopped")
+                        )
+
+    def submit(self, func: Callable[[Light], Any], timeout: float | None = None) -> Any:
+        """Run `func(light)` on the worker thread and return its result (blocking).
+
+        Raises:
+            - RuntimeError if the worker is not running
+            - TimeoutError if `timeout` elapses first
+            - Or whatever `func` raised.
+        """
+        future: Future[Any] = Future()
+        with self._lock:
+            if self._stopped or not self._thread.is_alive():
+                raise RuntimeError("LightThread worker is not running")
+            self._queue.put((func, future))
+        return future.result(timeout)
+
+    def shutdown(self, timeout: float | None = 5.0) -> None:
+        """Ask the worker to stop and wait (up to `timeout` seconds) for it to exit."""
+        self._queue.put(None)
+        self._thread.join(timeout)

light_daemon/README.md

@@ -0,0 +1,12 @@
+# light-phone-daemon
+
+A local gRPC daemon that holds an authenticated Light Phone session and exposes
+it to other processes, so tools in any language can depend on it.
+
+**Status:** early scaffolding. Not usable - yet!
+
+## Layout
+
+- `proto/` - protobuf contract
+- `light_daemon/v1/` - generated protobuf / gRPC modules.
+    - Regenerate with `./scripts/generate.sh` after changing anything under `proto/`.

light_daemon/light_daemon/__init__.py

@@ -0,0 +1 @@
+"""Local gRPC daemon for the Light Phone API."""

light_daemon/light_daemon/__main__.py

@@ -0,0 +1,88 @@
+"""Run the daemon: `python -m light_daemon` / `light-daemon`."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from light_api.worker import LightConfig, LightThread
+
+from light_daemon.auth import generate_token
+from light_daemon.server import serve
+
+
+def build_arg_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="light-daemon",
+        description="Local gRPC daemon for the Light Phone API (loopback only).",
+        allow_abbrev=False,
+    )
+    p.add_argument(
+        "--port",
+        type=int,
+        default=0,
+        help="TCP port on 127.0.0.1 (0 = OS-assigned, printed on start).",
+    )
+    p.add_argument("--email")
+    p.add_argument("--email-file")
+    p.add_argument("--password-file")
+    p.add_argument("--phone-number")
+    p.add_argument("--phone-number-file")
+    p.add_argument("--device-id")
+    p.add_argument("--device-id-file")
+    p.add_argument(
+        "--cache", action="store_true", help="Enable local response caching."
+    )
+    p.add_argument(
+        "--fake",
+        action="store_true",
+        help="Serve an in-memory fake session; no credentials needed (dev only).",
+    )
+    p.add_argument(
+        "--debug",
+        action="store_true",
+        help="Expose gRPC server reflection (still token-gated) for grpcurl etc.",
+    )
+    return p
+
+
+def config_from_args(args: argparse.Namespace) -> LightConfig:
+    return LightConfig(
+        email=args.email,
+        email_file=args.email_file,
+        password_file=args.password_file,
+        phone=args.phone_number,
+        phone_file=args.phone_number_file,
+        device_id=args.device_id,
+        device_id_file=args.device_id_file,
+        cache_enabled=args.cache,
+    )
+
+
+def main(argv: list[str] | None = None) -> None:
+    args = build_arg_parser().parse_args(argv)
+    token = generate_token()
+
+    if args.fake:
+        from light_daemon.testing import FakeLight, FakePw
+
+        serve(
+            FakePw(FakeLight()),
+            port=args.port,
+            token=token,
+            enable_reflection=args.debug,
+        )
+        return
+
+    worker = LightThread(config_from_args(args))
+    try:
+        worker.start()
+    except Exception as e:  # bad credentials, network, multi-device ambiguity, ...
+        print(f"light-daemon: could not start Light session: {e}", file=sys.stderr)
+        raise SystemExit(1)
+
+    serve(worker, port=args.port, token=token, enable_reflection=args.debug)
+
+
+if __name__ == "__main__":
+    main()

light_daemon/light_daemon/auth.py

@@ -0,0 +1,68 @@
+"""Per-instance bearer-token auth for the daemon.
+
+The daemon generates one random token per invocation. Every RPC must carry it as
+`authorization: bearer <token>` metadata. A serverside interceptor rejects the rest
+with UNAUTHENTICATED. The token reaches the client via a stdout handshake line.
+"""
+
+from __future__ import annotations
+
+import json
+import secrets
+
+import grpc
+
+_SCHEME = "bearer"
+_METADATA_KEY = "authorization"
+_DENY_MESSAGE = "missing or invalid token"
+
+
+def generate_token() -> str:
+    """A fresh, high-entropy token for one daemon run."""
+    return secrets.token_urlsafe(32)
+
+
+def bearer_metadata(token: str) -> list[tuple[str, str]]:
+    """Call metadata that a client attaches to every RPC: `metadata=bearer_metadata(tok)`."""
+    return [(_METADATA_KEY, f"{_SCHEME} {token}")]
+
+
+def handshake_line(host: str, port: int, token: str) -> str:
+    """The single JSON line the daemon prints to stdout on start for a parent process to read."""
+    return json.dumps({"host": host, "port": port, "token": token})
+
+
+def _deny_unary(request, context):
+    context.abort(grpc.StatusCode.UNAUTHENTICATED, _DENY_MESSAGE)
+
+
+def _deny_stream(request, context):
+    context.abort(grpc.StatusCode.UNAUTHENTICATED, _DENY_MESSAGE)
+    yield  # unreachable (abort raises) - present so this is a generator function
+
+
+def _deny_handler(handler: grpc.RpcMethodHandler) -> grpc.RpcMethodHandler:
+    """A handler of the same cardinality as `handler` that always aborts."""
+    if handler.request_streaming and handler.response_streaming:
+        return grpc.stream_stream_rpc_method_handler(_deny_stream)
+    if handler.request_streaming:
+        return grpc.stream_unary_rpc_method_handler(_deny_unary)
+    if handler.response_streaming:
+        return grpc.unary_stream_rpc_method_handler(_deny_stream)
+    return grpc.unary_unary_rpc_method_handler(_deny_unary)
+
+
+class AuthInterceptor(grpc.ServerInterceptor):
+    """Reject any call whose `authorization` metadata isn't `bearer <token>`."""
+
+    def __init__(self, token: str) -> None:
+        self._expected = f"{_SCHEME} {token}"
+
+    def intercept_service(self, continuation, handler_call_details):
+        handler = continuation(handler_call_details)
+        if handler is None:
+            return None  # unknown method - let gRPC return UNIMPLEMENTED
+        md = dict(handler_call_details.invocation_metadata or ())
+        if md.get(_METADATA_KEY) == self._expected:
+            return handler
+        return _deny_handler(handler)

light_daemon/light_daemon/errors.py

@@ -0,0 +1,83 @@
+"""Map API failures to gRPC status codes.
+
+The API only raises bare RuntimeErrors and httpx errors.
+
+httpx errors -> DEADLINE_EXCEEDED / UNAVAILABLE.
+
+For RuntimeErrors: disambiguate based on the error message.
+1. trailing HTTP status     -> follow _STATUS_MAP (5xx: UNAVAILABLE, 404: NOT_FOUND, ...)
+2. auth keywords detected   -> UNAUTHENTICATED   (only when there's no status)
+3. "no <x> found" keyword   -> NOT_FOUND
+4. else                     -> FAILED_PRECONDITION
+
+Anything else that isn't RuntimeError -> INTERNAL.
+"""
+
+from __future__ import annotations
+
+import re
+from contextlib import contextmanager
+from typing import Any, Iterator
+
+import grpc
+import httpx
+
+_AUTH_HINTS = ("credential", "cached session", "log in", "login", "password")
+_NOT_FOUND_HINTS = ("no device found", "no tool found", "no installed tool")
+
+# _ensure_ok(...) -> RuntimeError("Get tracks: 503")
+_TRAILING_STATUS = re.compile(r":\s*(\d{3})\s*$")
+
+_STATUS_MAP = {
+    400: grpc.StatusCode.INVALID_ARGUMENT,
+    401: grpc.StatusCode.UNAUTHENTICATED,
+    403: grpc.StatusCode.PERMISSION_DENIED,
+    404: grpc.StatusCode.NOT_FOUND,
+    409: grpc.StatusCode.ABORTED,
+    429: grpc.StatusCode.UNAVAILABLE,
+}
+
+
+def _looks_like_auth(message: str) -> bool:
+    lowered = message.lower()
+    return any(hint in lowered for hint in _AUTH_HINTS)
+
+
+def _code_from_http_status(message: str) -> grpc.StatusCode | None:
+    m = _TRAILING_STATUS.search(message)
+    if m is None:
+        return None
+    http = int(m.group(1))
+    if http in _STATUS_MAP:
+        return _STATUS_MAP[http]
+    if 500 <= http <= 599:
+        return grpc.StatusCode.UNAVAILABLE  # transient upstream failure; safe to retry
+    if 400 <= http <= 499:
+        return grpc.StatusCode.FAILED_PRECONDITION
+    return None
+
+
+def _runtime_error_code(message: str) -> grpc.StatusCode:
+    mapped = _code_from_http_status(message)
+    if mapped is not None:
+        return mapped
+    if _looks_like_auth(message):
+        return grpc.StatusCode.UNAUTHENTICATED
+    if any(hint in message.lower() for hint in _NOT_FOUND_HINTS):
+        return grpc.StatusCode.NOT_FOUND
+    return grpc.StatusCode.FAILED_PRECONDITION
+
+
+@contextmanager
+def grpc_errors(context: Any) -> Iterator[None]:
+    """Run a servicer body; translate any `light_api` failure into `context.abort`."""
+    try:
+        yield
+    except httpx.TimeoutException as e:
+        context.abort(grpc.StatusCode.DEADLINE_EXCEEDED, f"Light API timed out: {e}")
+    except httpx.HTTPError as e:
+        context.abort(grpc.StatusCode.UNAVAILABLE, f"Light API unreachable: {e}")
+    except RuntimeError as e:
+        context.abort(_runtime_error_code(str(e)), str(e))
+    except Exception as e:  # noqa: BLE001 - last-resort catch-all
+        context.abort(grpc.StatusCode.INTERNAL, f"{type(e).__name__}: {e}")