Add current_keys()/current_mac_addresses() to RollingKeyPairSource

findmy/accessory.py

@@ -124,6 +124,79 @@ def keys_between(
                 yielded.add(key)
                 yield ind, key
 
+    def current_keys(
+        self,
+        now: datetime | None = None,
+        margin: timedelta | None = None,
+    ) -> dict[KeyPair, int]:
+        """
+        Get the keys the accessory might currently be advertising, each with its index.
+
+        Spans the full :meth:`get_min_index`-:meth:`get_max_index` range for `now`
+        (rather than a single index) to account for rollover uncertainty since the
+        last observed alignment -- see those methods for why that range can be wider
+        than one index.
+
+        Returns a mapping rather than a set so that a caller which matches an
+        advertisement can pass the matching index back to :meth:`update_alignment`.
+        That is what keeps the range narrow, and the difference is not small: on a
+        30-day-old accessory aligned at index 2880, a 12 hour margin derives 100 keys
+        where a fresh alignment needs 3, at roughly 100x the cost. Membership tests
+        and iteration behave as they would on a set, so ``key in acc.current_keys()``
+        still reads the same.
+
+        **A bare call assumes the alignment is trustworthy**, which is the common case
+        only for a caller that keeps it so. `margin` widens the range on *both* sides
+        and exists for when it is not: without it the range starts at the alignment
+        index, so an accessory whose true index has ended up below where alignment
+        believes it is can never be matched -- it is simply never recognized, with
+        nothing raising anywhere. This has been observed on a real accessory,
+        advertising steadily a metre from the scanner and absent from its own
+        candidate set. :meth:`findmy.scanner.NearbyOfflineFindingDevice.is_from` takes
+        the same precaution, with a 12 hour margin.
+
+        So the two work together: pass a margin to recover from drift, feed the index
+        of whatever matched back into :meth:`update_alignment`, and subsequent calls
+        collapse to the cheap case.
+        """
+        if now is None:
+            now = datetime.now(timezone.utc)
+        if margin is None:
+            margin = timedelta(0)
+
+        return {key: ind for ind, key in self.keys_between(now - margin, now + margin)}
+
+    def current_mac_addresses(
+        self,
+        now: datetime | None = None,
+        margin: timedelta | None = None,
+    ) -> dict[str, int]:
+        """
+        Get the BLE MAC addresses the accessory might currently be advertising.
+
+        Useful to recognize an owned accessory's own advertisement in a BLE scan,
+        e.g. to trigger it directly (playing a sound) without going through Apple's
+        Find My network.
+
+        Maps each address to the key index it came from, so a scanner can report a
+        match straight back to :meth:`update_alignment`::
+
+            candidates = accessory.current_mac_addresses(margin=timedelta(hours=12))
+            ...
+            index = candidates.get(seen_address)
+            if index is not None:
+                accessory.update_alignment(seen_at, index)
+
+        The index is the first one in the searched range at which that key is valid,
+        which for a secondary key is a lower bound: one covers 96 primary indices.
+        Handing it to :meth:`update_alignment` is still safe, since that ignores any
+        index below the one it already holds.
+
+        See :meth:`current_keys` for the underlying key selection, for why `margin`
+        is worth passing, and for what feeding the index back saves.
+        """
+        return {key.mac_address: ind for key, ind in self.current_keys(now, margin).items()}
+
 
 class FixedRollingKeyPairAccessory(
     RollingKeyPairSource, util.abc.Serializable[FixedRollingKeyPairAccessoryMapping]

findmy/keys.py

@@ -98,7 +98,9 @@ def hashed_adv_key_bytes(self) -> bytes:
     @property
     def mac_address(self) -> str:
         """Get the mac address from the public key."""
-        first_byte = (self.adv_key_bytes[0] | 0b11000000).to_bytes(1)
+        # Both arguments spelled out: int.to_bytes only gained defaults in 3.11, and
+        # this package supports 3.10.
+        first_byte = (self.adv_key_bytes[0] | 0b11000000).to_bytes(1, "big")
         return ":".join([parsers.format_hex_byte(x) for x in first_byte + self.adv_key_bytes[1:6]])
 
     def adv_data(self, status: int = 0, hint: int = 0) -> bytes:

tests/test_accessory.py

@@ -0,0 +1,183 @@
+"""Tests for rolling-key accessory current-key/current-MAC helpers."""
+
+import re
+import secrets
+from datetime import datetime, timedelta, timezone
+
+MAC_RE = re.compile(r"^[0-9A-F]{2}(:[0-9A-F]{2}){5}$")
+
+
+def test_fixed_rolling_current_keys() -> None:
+    """current_keys()/current_mac_addresses() on a fixed-key accessory return all its keys."""
+    import findmy
+
+    keys = [findmy.KeyPair.new() for _ in range(3)]
+    accessory = findmy.FixedRollingKeyPairAccessory(
+        private_keys=[key.private_key_bytes for key in keys],
+        name="test",
+        identifier=None,
+    )
+
+    current = accessory.current_keys()
+    assert {key.adv_key_bytes for key in current} == {key.adv_key_bytes for key in keys}
+
+    macs = accessory.current_mac_addresses()
+    assert set(macs) == {key.mac_address for key in keys}
+    for mac in macs:
+        assert MAC_RE.match(mac)
+
+
+def test_findmy_accessory_current_keys_matches_keys_at_alignment() -> None:
+    """At the alignment date itself, current_keys() must match keys_at(alignment_index)."""
+    import findmy
+
+    paired_at = datetime.now(timezone.utc)
+    accessory = findmy.FindMyAccessory(
+        master_key=secrets.token_bytes(28),
+        skn=secrets.token_bytes(32),
+        sks=secrets.token_bytes(32),
+        paired_at=paired_at,
+    )
+
+    expected = accessory.keys_at(0)
+    current = accessory.current_keys(paired_at)
+    assert {key.adv_key_bytes for key in current} == {key.adv_key_bytes for key in expected}
+
+    expected_macs = {key.mac_address for key in expected}
+    assert set(accessory.current_mac_addresses(paired_at)) == expected_macs
+    for mac in expected_macs:
+        assert MAC_RE.match(mac)
+
+
+def test_findmy_accessory_current_keys_defaults_to_now() -> None:
+    """Calling current_keys()/current_mac_addresses() without an explicit `now` must not raise."""
+    import findmy
+
+    accessory = findmy.FindMyAccessory(
+        master_key=secrets.token_bytes(28),
+        skn=secrets.token_bytes(32),
+        sks=secrets.token_bytes(32),
+        paired_at=datetime.now(timezone.utc),
+    )
+
+    assert len(accessory.current_keys()) > 0
+    assert len(accessory.current_mac_addresses()) > 0
+
+
+def test_a_margin_reaches_indices_below_the_alignment_point() -> None:
+    """
+    The backwards half, which is the half the margin exists for.
+
+    An accessory whose true index has ended up *below* where alignment believes it is
+    falls outside the range entirely and is never matched. That can only happen once
+    alignment is above zero, so it can only be tested there: with the default
+    `alignment_index == 0` every negative index yields nothing and the widening comes
+    entirely from the forward side.
+    """
+    import findmy
+
+    now = datetime.now(timezone.utc)
+    accessory = findmy.FindMyAccessory(
+        master_key=secrets.token_bytes(28),
+        skn=secrets.token_bytes(32),
+        sks=secrets.token_bytes(32),
+        paired_at=now - timedelta(days=30),
+    )
+    accessory.update_alignment(now, 2880)
+
+    behind = {k.adv_key_bytes for k in accessory.keys_at(2879)}
+    with_margin = {k.adv_key_bytes for k in accessory.current_keys(now, margin=timedelta(hours=12))}
+    forward_only = {
+        k.adv_key_bytes for _, k in accessory.keys_between(now, now + timedelta(hours=12))
+    }
+
+    assert behind, "the fixture should have keys at the index just behind alignment"
+    assert behind <= with_margin
+    assert not (behind <= forward_only), (
+        "if this passes, the forward half alone covers it and the test proves nothing"
+    )
+
+
+def test_no_margin_behaves_exactly_as_before() -> None:
+    """The parameter is additive: omitting it must not change what existing callers get."""
+    import findmy
+
+    paired_at = datetime.now(timezone.utc)
+    accessory = findmy.FindMyAccessory(
+        master_key=secrets.token_bytes(28),
+        skn=secrets.token_bytes(32),
+        sks=secrets.token_bytes(32),
+        paired_at=paired_at,
+    )
+
+    assert {k.adv_key_bytes for k in accessory.current_keys(paired_at)} == {
+        k.adv_key_bytes for k in accessory.current_keys(paired_at, margin=timedelta(0))
+    }
+
+
+def test_the_margin_reaches_mac_addresses_too() -> None:
+    import findmy
+
+    accessory = findmy.FindMyAccessory(
+        master_key=secrets.token_bytes(28),
+        skn=secrets.token_bytes(32),
+        sks=secrets.token_bytes(32),
+        paired_at=datetime.now(timezone.utc),
+    )
+
+    assert len(accessory.current_mac_addresses(margin=timedelta(hours=12))) > len(
+        accessory.current_mac_addresses()
+    )
+
+
+def test_the_index_comes_back_with_each_candidate() -> None:
+    """
+    The index is what makes the cheap path reachable.
+
+    A scanner that matches an advertisement has to be able to report *which* index it
+    matched, or it cannot call update_alignment and every later call pays for the wide
+    range again.
+    """
+    import findmy
+
+    now = datetime.now(timezone.utc)
+    accessory = findmy.FindMyAccessory(
+        master_key=secrets.token_bytes(28),
+        skn=secrets.token_bytes(32),
+        sks=secrets.token_bytes(32),
+        paired_at=now - timedelta(days=30),
+    )
+    accessory.update_alignment(now, 2880)
+
+    candidates = accessory.current_mac_addresses(now, margin=timedelta(hours=12))
+    known_mac = next(iter(accessory.keys_at(2879))).mac_address
+
+    # The index reported is the first one in the searched range at which that key is
+    # valid, not necessarily the only one: a secondary key covers 96 primary indices,
+    # so for those it is a lower bound. What has to hold is that the key really does
+    # occur there, which is what makes it safe to hand to update_alignment.
+    reported = candidates[known_mac]
+    assert known_mac in {k.mac_address for k in accessory.keys_at(reported)}
+
+
+def test_feeding_a_matched_index_back_collapses_the_next_call() -> None:
+    """The pattern the docstring recommends, pinned end to end."""
+    import findmy
+
+    now = datetime.now(timezone.utc)
+    accessory = findmy.FindMyAccessory(
+        master_key=secrets.token_bytes(28),
+        skn=secrets.token_bytes(32),
+        sks=secrets.token_bytes(32),
+        paired_at=now - timedelta(days=30),
+    )
+    accessory.update_alignment(now - timedelta(days=1), 2784)
+
+    wide = accessory.current_mac_addresses(now, margin=timedelta(hours=12))
+    seen_mac, seen_index = next(iter(wide.items()))
+
+    accessory.update_alignment(now, seen_index)
+    narrow = accessory.current_mac_addresses(now)
+
+    assert len(narrow) < len(wide)
+    assert seen_mac in wide