diff --git a/findmy/accessory.py b/findmy/accessory.py index 9e564bb..215131f 100644 --- a/findmy/accessory.py +++ b/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] diff --git a/findmy/keys.py b/findmy/keys.py index d5418e2..e6411a5 100644 --- a/findmy/keys.py +++ b/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: diff --git a/tests/test_accessory.py b/tests/test_accessory.py new file mode 100644 index 0000000..95ab225 --- /dev/null +++ b/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