reports: update real_airtag example

examples/_login.py

@@ -0,0 +1,101 @@
+import json
+from pathlib import Path
+
+from findmy.reports import (
+    AppleAccount,
+    AsyncAppleAccount,
+    BaseAnisetteProvider,
+    LoginState,
+    SmsSecondFactorMethod,
+    TrustedDeviceSecondFactorMethod,
+)
+
+ACCOUNT_STORE = "account.json"
+
+
+def _login_sync(account: AppleAccount) -> None:
+    email = input("email?  > ")
+    password = input("passwd? > ")
+
+    state = account.login(email, password)
+
+    if state == LoginState.REQUIRE_2FA:  # Account requires 2FA
+        # This only supports SMS methods for now
+        methods = account.get_2fa_methods()
+
+        # Print the (masked) phone numbers
+        for i, method in enumerate(methods):
+            if isinstance(method, TrustedDeviceSecondFactorMethod):
+                print(f"{i} - Trusted Device")
+            elif isinstance(method, SmsSecondFactorMethod):
+                print(f"{i} - SMS ({method.phone_number})")
+
+        ind = int(input("Method? > "))
+
+        method = methods[ind]
+        method.request()
+        code = input("Code? > ")
+
+        # This automatically finishes the post-2FA login flow
+        method.submit(code)
+
+
+async def _login_async(account: AsyncAppleAccount) -> None:
+    email = input("email?  > ")
+    password = input("passwd? > ")
+
+    state = await account.login(email, password)
+
+    if state == LoginState.REQUIRE_2FA:  # Account requires 2FA
+        # This only supports SMS methods for now
+        methods = await account.get_2fa_methods()
+
+        # Print the (masked) phone numbers
+        for i, method in enumerate(methods):
+            if isinstance(method, TrustedDeviceSecondFactorMethod):
+                print(f"{i} - Trusted Device")
+            elif isinstance(method, SmsSecondFactorMethod):
+                print(f"{i} - SMS ({method.phone_number})")
+
+        ind = int(input("Method? > "))
+
+        method = methods[ind]
+        await method.request()
+        code = input("Code? > ")
+
+        # This automatically finishes the post-2FA login flow
+        await method.submit(code)
+
+
+def get_account_sync(anisette: BaseAnisetteProvider) -> AppleAccount:
+    """Tries to restore a saved Apple account, or prompts the user for login otherwise. (sync)"""
+    acc = AppleAccount(anisette)
+
+    # Save / restore account logic
+    acc_store = Path("account.json")
+    try:
+        with acc_store.open() as f:
+            acc.restore(json.load(f))
+    except FileNotFoundError:
+        _login_sync(acc)
+        with acc_store.open("w+") as f:
+            json.dump(acc.export(), f)
+
+    return acc
+
+
+async def get_account_async(anisette: BaseAnisetteProvider) -> AsyncAppleAccount:
+    """Tries to restore a saved Apple account, or prompts the user for login otherwise. (async)"""
+    acc = AsyncAppleAccount(anisette)
+
+    # Save / restore account logic
+    acc_store = Path("account.json")
+    try:
+        with acc_store.open() as f:
+            acc.restore(json.load(f))
+    except FileNotFoundError:
+        await _login_async(acc)
+        with acc_store.open("w+") as f:
+            json.dump(acc.export(), f)
+
+    return acc

examples/fetch_reports.py

@@ -1,25 +1,15 @@
-import json
 import logging
-from pathlib import Path
+
+from _login import get_account_sync
 
 from findmy import KeyPair
-from findmy.reports import (
+from findmy.reports import RemoteAnisetteProvider
-    AppleAccount,
-    LoginState,
-    RemoteAnisetteProvider,
-    SmsSecondFactorMethod,
-    TrustedDeviceSecondFactorMethod,
-)
 
 # URL to (public or local) anisette server
 ANISETTE_SERVER = "http://localhost:6969"
 
-# Apple account details
-ACCOUNT_EMAIL = "test@test.com"
-ACCOUNT_PASS = ""
-
 # Private base64-encoded key to look up
-KEY_PRIV = ""
+KEY_PRIV = "Vq/RNibhblTitwb7hjPkZZj6gyJcAJSVMQ6Shg=="
 
 # Optional, to verify that advertisement key derivation works for your key
 KEY_ADV = ""
@@ -27,43 +17,9 @@ KEY_ADV = ""
 logging.basicConfig(level=logging.DEBUG)
 
 
-def login(account: AppleAccount) -> None:
-    state = account.login(ACCOUNT_EMAIL, ACCOUNT_PASS)
-
-    if state == LoginState.REQUIRE_2FA:  # Account requires 2FA
-        # This only supports SMS methods for now
-        methods = account.get_2fa_methods()
-
-        # Print the (masked) phone numbers
-        for i, method in enumerate(methods):
-            if isinstance(method, TrustedDeviceSecondFactorMethod):
-                print(f"{i} - Trusted Device")
-            elif isinstance(method, SmsSecondFactorMethod):
-                print(f"{i} - SMS ({method.phone_number})")
-
-        ind = int(input("Method? > "))
-
-        method = methods[ind]
-        method.request()
-        code = input("Code? > ")
-
-        # This automatically finishes the post-2FA login flow
-        method.submit(code)
-
-
 def fetch_reports(lookup_key: KeyPair) -> None:
     anisette = RemoteAnisetteProvider(ANISETTE_SERVER)
-    acc = AppleAccount(anisette)
+    acc = get_account_sync(anisette)
-
-    # Save / restore account logic
-    acc_store = Path("account.json")
-    try:
-        with acc_store.open() as f:
-            acc.restore(json.load(f))
-    except FileNotFoundError:
-        login(acc)
-        with acc_store.open("w+") as f:
-            json.dump(acc.export(), f)
 
     print(f"Logged in as: {acc.account_name} ({acc.first_name} {acc.last_name})")
 

examples/fetch_reports_async.py

@@ -1,24 +1,14 @@
 import asyncio
-import json
 import logging
-from pathlib import Path
+
+from _login import get_account_async
 
 from findmy import KeyPair
-from findmy.reports import (
+from findmy.reports import RemoteAnisetteProvider
-    AsyncAppleAccount,
-    LoginState,
-    RemoteAnisetteProvider,
-    SmsSecondFactorMethod,
-    TrustedDeviceSecondFactorMethod,
-)
 
 # URL to (public or local) anisette server
 ANISETTE_SERVER = "http://localhost:6969"
 
-# Apple account details
-ACCOUNT_EMAIL = "test@test.com"
-ACCOUNT_PASS = ""
-
 # Private base64-encoded key to look up
 KEY_PRIV = ""
 
@@ -28,44 +18,12 @@ KEY_ADV = ""
 logging.basicConfig(level=logging.DEBUG)
 
 
-async def login(account: AsyncAppleAccount) -> None:
-    state = await account.login(ACCOUNT_EMAIL, ACCOUNT_PASS)
-
-    if state == LoginState.REQUIRE_2FA:  # Account requires 2FA
-        # This only supports SMS methods for now
-        methods = await account.get_2fa_methods()
-
-        # Print the (masked) phone numbers
-        for i, method in enumerate(methods):
-            if isinstance(method, TrustedDeviceSecondFactorMethod):
-                print(f"{i} - Trusted Device")
-            elif isinstance(method, SmsSecondFactorMethod):
-                print(f"{i} - SMS ({method.phone_number})")
-
-        ind = int(input("Method? > "))
-
-        method = methods[ind]
-        await method.request()
-        code = input("Code? > ")
-
-        # This automatically finishes the post-2FA login flow
-        await method.submit(code)
-
-
 async def fetch_reports(lookup_key: KeyPair) -> None:
     anisette = RemoteAnisetteProvider(ANISETTE_SERVER)
-    acc = AsyncAppleAccount(anisette)
+
+    acc = await get_account_async(anisette)
 
     try:
-        acc_store = Path("account.json")
-        try:
-            with acc_store.open() as f:
-                acc.restore(json.load(f))
-        except FileNotFoundError:
-            await login(acc)
-            with acc_store.open("w+") as f:
-                json.dump(acc.export(), f)
-
         print(f"Logged in as: {acc.account_name} ({acc.first_name} {acc.last_name})")
 
         # It's that simple!

examples/real_airtag.py

@@ -1,17 +1,20 @@
 """
-Example showing how to retrieve the primary key of your own AirTag, or any other FindMy-accessory.
+Example showing how to fetch locations of an AirTag, or any other FindMy accessory.
+"""
+from __future__ import annotations
 
-This key can be used to retrieve the device's location for a single day.
-"""
 import plistlib
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
-from findmy import FindMyAccessory
+from _login import get_account_sync
+
+from findmy import FindMyAccessory, KeyPair
+from findmy.reports import RemoteAnisetteProvider
+
+# URL to (public or local) anisette server
+ANISETTE_SERVER = "http://localhost:6969"
 
-# PUBLIC key that the accessory is broadcasting or has previously broadcast.
-# For nearby devices, you can use `device_scanner.py` to find it.
-PUBLIC_KEY = ""
 # Path to a .plist dumped from the Find My app.
 PLIST_PATH = Path("airtag.plist")
 
@@ -29,34 +32,50 @@ SKN = device_data["sharedSecret"]["key"]["data"]
 # "Secondary" shared secret. 32 bytes.
 SKS = device_data["secondarySharedSecret"]["key"]["data"]
 
+# "Paired at" timestamp (UTC)
+PAIRED_AT = device_data["pairingDate"].replace(tzinfo=timezone.utc)
+
+
+def _gen_keys(airtag: FindMyAccessory, _from: datetime, to: datetime) -> set[KeyPair]:
+    keys = set()
+    while _from < to:
+        keys.update(airtag.keys_at(_from))
+
+        _from += timedelta(minutes=15)
+
+    return keys
+
 
 def main() -> None:
-    paired_at = device_data["pairingDate"].replace(tzinfo=timezone.utc)
+    # Step 0: create an accessory key generator
-    airtag = FindMyAccessory(MASTER_KEY, SKN, SKS, paired_at)
+    airtag = FindMyAccessory(MASTER_KEY, SKN, SKS, PAIRED_AT)
 
-    now = datetime.now(tz=timezone.utc)
+    # Step 1: Generate the accessory's private keys,
-    lookup_time = paired_at.replace(
+    # starting from 7 days ago until now (12 hour margin)
-        minute=paired_at.minute // 15 * 15,
+    fetch_to = datetime.now(tz=timezone.utc).astimezone() + timedelta(hours=12)
-        second=0,
+    fetch_from = fetch_to - timedelta(days=8)
-        microsecond=0,
-    ) + timedelta(minutes=15)
 
-    while lookup_time < now:
+    print(f"Generating keys from {fetch_from} to {fetch_to} ...")
-        keys = airtag.keys_at(lookup_time)
+    lookup_keys = _gen_keys(airtag, fetch_from, fetch_to)
-        for key in keys:
-            if key.adv_key_b64 != PUBLIC_KEY:
-                continue
 
-            print("KEY FOUND!!")
+    print(f"Generated {len(lookup_keys)} keys")
-            print("KEEP THE BELOW KEY SECRET! IT CAN BE USED TO RETRIEVE THE DEVICE'S LOCATION!")
-            print(f"  - Key:           {key.private_key_b64}")
-            print(f"  - Approx. Time:  {lookup_time}")
-            print(f"  - Type:          {key.key_type}")
-            return
 
-        lookup_time += timedelta(minutes=15)
+    # Step 2: log into an Apple account
+    print("Logging into account")
+    anisette = RemoteAnisetteProvider(ANISETTE_SERVER)
+    acc = get_account_sync(anisette)
 
-    print("No match found! :(")
+    # step 3: fetch reports!
+    print("Fetching reports")
+    reports = acc.fetch_reports(list(lookup_keys), fetch_from, fetch_to)
+
+    # step 4: print 'em
+    # reports are in {key: [report]} format, but we only really care about the reports
+    print()
+    print("Location reports:")
+    reports = sorted([r for rs in reports.values() for r in rs])
+    for report in reports:
+        print(f" - {report}")
 
 
 if __name__ == "__main__":

findmy/reports/__init__.py

@@ -1,6 +1,6 @@
 """Code related to fetching location reports."""
 from .account import AppleAccount, AsyncAppleAccount
-from .anisette import RemoteAnisetteProvider
+from .anisette import BaseAnisetteProvider, RemoteAnisetteProvider
 from .state import LoginState
 from .twofactor import SmsSecondFactorMethod, TrustedDeviceSecondFactorMethod
 
@@ -8,6 +8,7 @@ __all__ = (
     "AppleAccount",
     "AsyncAppleAccount",
     "LoginState",
+    "BaseAnisetteProvider",
     "RemoteAnisetteProvider",
     "SmsSecondFactorMethod",
     "TrustedDeviceSecondFactorMethod",