Merge pull request #147 from malmeloo/feat/local-anisette

feat(reports): implement local Anisette provider

Author: malmeloo

findmy/reports/account.py

@@ -33,7 +33,7 @@
     UnhandledProtocolError,
 )
 from findmy.util import crypto
-from findmy.util.closable import Closable
+from findmy.util.abc import Closable
 from findmy.util.http import HttpResponse, HttpSession, decode_plist
 
 from .reports import LocationReport, LocationReportsFetcher

findmy/reports/anisette.py

@@ -8,16 +8,20 @@
 import time
 from abc import ABC, abstractmethod
 from datetime import datetime, timezone
+from io import BytesIO
+from pathlib import Path
+from typing import BinaryIO
 
+from anisette import Anisette, AnisetteHeaders
 from typing_extensions import override
 
-from findmy.util.closable import Closable
+from findmy.util.abc import Closable, Serializable
 from findmy.util.http import HttpSession
 
 logger = logging.getLogger(__name__)
 
 
-class BaseAnisetteProvider(Closable, ABC):
+class BaseAnisetteProvider(Closable, Serializable, ABC):
     """
     Abstract base class for Anisette providers.
 
@@ -27,22 +31,13 @@ class BaseAnisetteProvider(Closable, ABC):
     @property
     @abstractmethod
     def otp(self) -> str:
-        """
-        A seemingly random base64 string containing 28 bytes.
-
-        TODO: Figure out how to generate this.
-        """
+        """A seemingly random base64 string containing 28 bytes."""
         raise NotImplementedError
 
     @property
     @abstractmethod
     def machine(self) -> str:
-        """
-        A base64 encoded string of 60 'random' bytes.
-
-        We're not sure how this is generated, we have to rely on the server.
-        TODO: Figure out how to generate this.
-        """
+        """A base64 encoded string of 60 'random' bytes."""
         raise NotImplementedError
 
     @property
@@ -177,6 +172,24 @@ def __init__(self, server_url: str) -> None:
         self._anisette_data: dict[str, str] | None = None
         self._anisette_data_expires_at: float = 0
 
+    @override
+    def serialize(self) -> dict:
+        """See `BaseAnisetteProvider.serialize`."""
+        return {
+            "type": "aniRemote",
+            "url": self._server_url,
+        }
+
+    @classmethod
+    @override
+    def deserialize(cls, data: dict) -> RemoteAnisetteProvider:
+        """See `BaseAnisetteProvider.deserialize`."""
+        assert data["type"] == "aniRemote"
+
+        server_url = data["url"]
+
+        return cls(server_url)
+
     @property
     @override
     def otp(self) -> str:
@@ -219,22 +232,101 @@ async def close(self) -> None:
         await self._http.close()
 
 
-# TODO(malmeloo): implement using pyprovision
-# https://github.com/malmeloo/FindMy.py/issues/2
 class LocalAnisetteProvider(BaseAnisetteProvider):
-    """Anisette provider. Generates headers without a remote server using pyprovision."""
+    """Anisette provider. Generates headers without a remote server using the `anisette` library."""
+
+    def __init__(
+        self,
+        *,
+        state_blob: BinaryIO | None = None,
+        libs_path: str | Path | None = None,
+    ) -> None:
+        """Initialize the provider."""
+        super().__init__()
+
+        if isinstance(libs_path, str):
+            libs_path = Path(libs_path)
+
+        if libs_path is None or not libs_path.is_file():
+            logger.info(
+                "The Anisette engine will download libraries required for operation, "
+                "this may take a few seconds...",
+            )
+            logger.info(
+                "To speed up future local Anisette initializations, "
+                "provide a filesystem path to load the libraries from.",
+            )
+
+        files: list[BinaryIO | Path] = []
+        if state_blob is not None:
+            files.append(state_blob)
+        if libs_path is not None and libs_path.exists():
+            files.append(libs_path)
+
+        self._ani = Anisette.load(*files)
+        self._ani_data: AnisetteHeaders | None = None
+        self._libs_path: Path | None = libs_path
+
+        if libs_path is not None:
+            self._ani.save_libs(libs_path)
+        if state_blob is not None and not self._ani.is_provisioned:
+            logger.warning(
+                "The Anisette state that was loaded has not yet been provisioned. "
+                "Was the previous session saved properly?",
+            )
+
+    @override
+    def serialize(self) -> dict:
+        """See `BaseAnisetteProvider.serialize`."""
+        with BytesIO() as buf:
+            self._ani.save_provisioning(buf)
+            prov_data = base64.b64encode(buf.getvalue()).decode("utf-8")
+
+        return {
+            "type": "aniLocal",
+            "prov_data": prov_data,
+        }
+
+    @classmethod
+    @override
+    def deserialize(cls, data: dict, libs_path: str | Path | None = None) -> LocalAnisetteProvider:
+        """See `BaseAnisetteProvider.deserialize`."""
+        assert data["type"] == "aniLocal"
+
+        state_blob = BytesIO(base64.b64decode(data["prov_data"]))
+
+        return cls(state_blob=state_blob, libs_path=libs_path)
+
+    @override
+    async def get_headers(
+        self,
+        user_id: str,
+        device_id: str,
+        serial: str = "0",
+        with_client_info: bool = False,
+    ) -> dict[str, str]:
+        """See `BaseAnisetteProvider.get_headers`_."""
+        self._ani_data = self._ani.get_data()
+
+        return await super().get_headers(user_id, device_id, serial, with_client_info)
 
     @property
     @override
     def otp(self) -> str:
         """See `BaseAnisetteProvider.otp`_."""
-        raise NotImplementedError
+        machine = (self._ani_data or {}).get("X-Apple-I-MD")
+        if machine is None:
+            logger.warning("X-Apple-I-MD header not found! Returning fallback...")
+        return machine or ""
 
     @property
     @override
     def machine(self) -> str:
         """See `BaseAnisetteProvider.machine`_."""
-        raise NotImplementedError
+        machine = (self._ani_data or {}).get("X-Apple-I-MD-M")
+        if machine is None:
+            logger.warning("X-Apple-I-MD-M header not found! Returning fallback...")
+        return machine or ""
 
     @override
     async def close(self) -> None:

findmy/util/closable.py findmy/util/abc.pyfindmy/util/closable.py renamed to findmy/util/abc.py

@@ -1,4 +1,4 @@
-"""ABC for async classes that need to be cleaned up before exiting."""
+"""Various utility ABCs for internal and external classes."""
 
 from __future__ import annotations
 
@@ -36,3 +36,18 @@ def __del__(self) -> None:
                 loop.run_until_complete(self.close())
         except RuntimeError:
             pass
+
+
+class Serializable(ABC):
+    """ABC for serializable classes."""
+
+    @abstractmethod
+    def serialize(self) -> dict:
+        """Serialize the object to a JSON-serializable dictionary."""
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    def deserialize(cls, data: dict) -> Serializable:
+        """Deserialize the object from a JSON-serializable dictionary."""
+        raise NotImplementedError

findmy/util/http.py

@@ -11,7 +11,7 @@
 from aiohttp import BasicAuth, ClientSession, ClientTimeout
 from typing_extensions import Unpack, override
 
-from .closable import Closable
+from .abc import Closable
 from .parsers import decode_plist
 
 logger = logging.getLogger(__name__)

pyproject.toml

@@ -6,6 +6,7 @@ readme = "README.md"
 authors = [
     {name = "Mike Almeloo", email = "git@mikealmel.ooo"},
 ]
+license-files = ["LICENSE.md"]
 requires-python = ">=3.9,<3.14"
 dependencies = [
     "srp>=1.0.21,<2.0.0",
@@ -14,6 +15,7 @@ dependencies = [
     "aiohttp>=3.9.5,<4.0.0",
     "bleak>=0.22.2,<1.0.0",
     "typing-extensions>=4.12.2,<5.0.0",
+    "anisette>=1.2.1",
 ]
 
 [dependency-groups]
@@ -78,9 +80,6 @@ ignore = [
     "D",  # documentation
 ]
 
-[tool.setuptools]
-license-files = []
-
 [build-system]
 requires = ["setuptools", "setuptools-scm"]
 build-backend = "setuptools.build_meta"