diff --git a/docs/findmy-py-announce-401-report.md b/docs/findmy-py-announce-401-report.md new file mode 100644 index 00000000..cf397763 --- /dev/null +++ b/docs/findmy-py-announce-401-report.md @@ -0,0 +1,156 @@ +# Report to FindMy.py: `announce_device` returns HTTP 401 + +> ## Answered: Apple wants the one field this flow refuses to send +> +> With the body now reported, the refusal explains itself: +> +> ``` +> HTTP 401, body {'ec': -800012, 'em': 'Push token is invalid.'} +> ``` +> +> Neither header divergence was the cause. Both were real and both are fixed; this is something +> else, and it is not a bug in the implementation. +> +> **`ptkn` is omitted on purpose.** Stage 2 §7 says "never send `ptkn` — omit the key entirely +> rather than sending it empty", and Stage 1 §13 argues at length why: a push token is the most +> likely reason a registered device becomes trusted for verification codes, and this client must +> never become a second factor for somebody's Apple ID. The library implements that, with a +> comment saying there is deliberately no parameter for it. +> +> Apple appears to treat *absent* as *invalid*. That is a reading of one error string rather than +> a proven mechanism — Apple's codes are frequently imprecise, and `-800012` may cover more than +> its text suggests — but it is the obvious one and it is consistent with the endpoint's purpose: +> `postdata` announces a live client, and liveness to Apple means reachable by push. +> +> ### This is a decision, not a fix +> +> Making the call succeed means sending a real push token, which means **registering with APNs and +> becoming addressable** — the precise thing §13 says not to do. It is also not a small change: a +> token has to be obtained from Apple's push service, so there is nothing to "just pass through". +> +> The safety property currently holds and the user can see it. The device list entry reads: +> +> > This device cannot be used to receive Apple Account verification codes. +> +> That line is what a client with no push token earns, and it is worth more than a nicer name. +> +> **So this consumer is not asking for `ptkn` support**, and would not use it if it appeared. The +> open question belongs to whoever owns the specification: whether §7's device naming is reachable +> at all by a client that declines to be push-addressable, or whether the section should record +> that it is not. Either answer is useful; the current text implies the call works as described, +> and on this account it does not. +> +> Everything below is the original report, kept because the two divergences it names were real. + +--- + + +> **A bug report from a consumer.** Addressed to whoever works on FindMy.py. Delete it from this +> repository once the library has an answer. +> +> Against `feat/icloud-keychain-export` at `534867d` — the commit that added the feature. Follow-up +> to [the device-name request](./findmy-py-device-name-request.md), which this implements. +> +> **The specification is not at fault here.** Stage 2 §7 states both of the values below +> explicitly. This is a divergence between the implementation and the document it was written +> from, which is why it is a report rather than a question. + +## What happens + +``` +findmy.errors.UnhandledProtocolError: Announcing the device failed with HTTP 401 + File ".../findmy/reports/account.py", line 1554, in announce_device +``` + +On a live account, immediately after a successful login, on a machine Apple already knows (no +second factor was requested). The export itself is unaffected — the consumer treats a failed +announce as cosmetic — but the device list entry keeps its hardware name, which is the entire +point of the call. + +Not a missing-token failure: `announce_device` raises `InvalidStateError` when `adsid` or +`idms_hb` are absent, and it did not. The token was present and the request was rejected. + +## Two headers disagree with the specification, and with each other + +Stage 2 §7 gives the header set for `POST gsas.apple.com/grandslam/GsService2/postdata`: + +| Header | §7 requires | `account.py` sends | +| --- | --- | --- | +| `X-MMe-Client-Info` | `mmeClientInfoAkd` | `self._anisette.client` | +| `User-Agent` | `akdUserAgent` | `"akd/1.0 CFNetwork/978.0.7 Darwin/18.7.0"` | + +### 1. `X-MMe-Client-Info` is the wrong variant + +Stage 1 §2.2 defines three composite strings, and the difference between two of them is the +trailing bundle — *which Apple daemon is speaking*: + +``` +mmeClientInfo = +mmeClientInfoAkd = +``` + +`AnisetteProvider.client` produces neither. Observed value: + +``` + +``` + +So the request tells a Grand Slam endpoint that **Xcode** is speaking, while its `User-Agent` +says **akd**. §7 asks for the akd variant specifically, and §1 (Stage 1 §2.2) is explicit that +the bundle identifies the speaker. + +There is currently no way for a consumer to fix this from outside: the provider exposes `client` +and nothing else. + +### 2. The `User-Agent` describes a different macOS from the client info + +``` +akd/1.0 CFNetwork/978.0.7 Darwin/18.7.0 <- hardcoded +<... Mac OS X;13.4.1;22F8 ...> <- what the client info claims +``` + +Darwin 18 is macOS 10.14; the client info says 13.4.1, which is Darwin 22. Stage 1 §2.2 requires +that the OS version, build, CFNetwork version and Darwin version correspond to one another, +because they describe one real release and Apple's own clients never contradict themselves. This +request contradicts itself. + +The specification's `akdUserAgent` example (`akd/1.0 CFNetwork/1494.0.7 Darwin/23.4.0`) is the +matching pair for *its* example device, iOS 17.4 — so it is an illustration of the rule, not a +constant to hardcode. The value has to be derived from whatever release the identity claims. + +## What is not ruled out + +The 401 has not been traced to either header — both are stated as divergences from §7, not as +proven causes. Two other candidates are untested here: + +- **The `X-Apple-HB-Token` contents.** Stage 1 §8 observes the heartbeat token arrives as + `service:token` — two fields, no lifetime — and `account.py:1323` reads + `tokens["com.apple.gs.idms.hb"]["token"]`. Whether that yields the token alone or something + else has not been checked here. +- **Whether the token survives the 2FA route.** `_set_login_state(REQUIRE_2FA, …)` stores only + `adsid` and `idms_token`; the heartbeat is read in the `au is None` branch. Whether a + post-2FA re-authentication reaches that branch is not something this consumer can observe. + +## Suggested shape + +The client-info variant is the one a consumer cannot work around: + +```python +provider.client # what exists: the AOSKit/Xcode form +provider.client_akd # the com.apple.akd/1.0 form, per Stage 1 §2.2 +``` + +with `announce_device` sending `client_akd`, and the `User-Agent` built from the same identity +rather than fixed — so the CFNetwork and Darwin versions match the OS the client info claims. + +## How to reproduce + +Any account, on a machine that has signed in before so no second factor is requested: + +```python +account = AsyncAppleAccount(anisette, device_name="Anything") +await account.login(email, password) +await account.announce_device() # HTTP 401 +``` + +The consumer's own call site is `exporter/icloud.py::_announce`, which logs and continues. diff --git a/docs/how-to-export-with-the-cli.md b/docs/how-to-export-with-the-cli.md index 4d7b77c6..3054867a 100644 --- a/docs/how-to-export-with-the-cli.md +++ b/docs/how-to-export-with-the-cli.md @@ -247,7 +247,90 @@ would fall in seconds. | `--anisette-url` | Use a remote Anisette server instead of running one locally | | `--anisette-libs` | Cache Apple's ADI libraries in a file, so a later run does not download them | | `-v`, `--verbose` | Show what the library is doing. Worth having on the first run | -| `-vv` | Everything, including the CloudKit and keychain protocol steps. This is what to send with a bug report | +| `-vv` | Everything, including the CloudKit and keychain protocol steps. This is what to send with a bug report — **read the warning below first** | + +> [!CAUTION] +> **Read `-v` and `-vv` output before you paste it anywhere, and strip anything that identifies +> you.** It is written for debugging, not for publishing, and it names real things: +> +> - **Your devices, by name, model and serial number.** Escrow records are described the way a +> person recognises them, so "Paula's iPhone, iPhone14,2, serial F2L…" is a normal line. +> - **Keychain item attributes, verbatim.** The account and label fields are printed as they are +> stored, and what is in them is Apple's choice rather than this program's. +> - **Peer identifiers** for every device in your account's trust circle. Pseudonymous, but stable +> and unique to you. +> +> No key material, passcode or password is logged, and payloads appear only as byte counts. But +> **this cannot promise a given identifier never reaches the output** — the text comes from a +> library reading Apple's structures, and a field that is innocuous on one account may not be on +> another. So the check is yours to make, every time, before it goes in an issue. + +--- + +## Running it from a script + +**Sign in by hand once first.** The first run registers this machine as a device on your Apple +account and remembers it, so later runs usually go straight through without a verification code. +Until that has happened, no amount of flags will make it unattended — a code is sent to your +phone and something has to type it. + +After that, every other answer can be given up front: + +```shell +export OPENTAGVIEWER_APPLE_PASSWORD='...' +export OPENTAGVIEWER_DEVICE_PASSCODE='...' + +uv run python -m exporter.cli \ + --non-interactive \ + --apple-id me@example.com \ + --device F2LX9Q \ + --all-tags \ + --no-password \ + --output tags.zip +``` + +`--non-interactive` is the one that matters. Without it a missing answer is *prompted for*, and a +scheduled job sits waiting on a keystroke that never arrives — no error, no output, no end. With +it, the run fails and names the question it wanted to ask. + +### Where to put the password + +**There is no `--password` flag and there will not be one.** Anything on a command line is +readable by every other user on the machine through `ps`, and it lands in your shell history. + +Worst last: + +| | | +| --- | --- | +| **A file, or a pipe** | `--password-file secret.txt`, or `--password-file -` to read one line from standard input. Nothing in `ps`, nothing in the environment, nothing in history. The file must not be readable by other users — `chmod 600` it, or the run refuses it | +| **An environment variable** | `OPENTAGVIEWER_APPLE_PASSWORD` and `OPENTAGVIEWER_DEVICE_PASSCODE`. Fine in a CI secret store. Weaker than a file: on Linux another process of yours can read `/proc//environ`, it is inherited by every child, and setting it inline on a command puts it in history anyway | +| **Being asked** | What happens when neither is given | + +If you use a variable, set it from something that is not itself the secret — a secret store, or +`read -s`, not a literal in a checked-in script. In most shells, prefixing a command with a space +also keeps it out of history. + +### The other flags + +| Option | | +| --- | --- | +| `--apple-id` | Sign in as this, instead of being asked | +| `--password-file` | Read the Apple ID password from a file, or `-` for standard input | +| `--passcode-file` | The same, for the device screen-lock passcode | +| `--device SERIAL` | Unlock with this device's escrow record instead of choosing from a list. Fails, listing what is available, if nothing matches | +| `--all-tags` | Export everything found, instead of choosing | +| `--include-my-devices` | Let `--all-tags` include your own iPhones, iPads and Macs. **Never implied** | +| `--non-interactive` | Fail rather than ask | + +> [!WARNING] +> **`--all-tags` leaves your own devices out on purpose.** "All my tags" and "all my tags plus the +> laptop I am sitting at" are different requests, and a bundle holding your MacBook lets whoever +> receives it locate *you*. Interactively the CLI names each device and asks again; a scheduled run +> has nobody to ask, so it takes the safe half and prints what it left out. + +Exit codes: `0` if everything asked for was exported, `1` if not — including the case where a +bundle *was* written but the account could not be read, which is worth checking for rather than +assuming a file on disk means success. --- diff --git a/python/exporter/cli.py b/python/exporter/cli.py index 55ccd2f3..1dde0c17 100644 --- a/python/exporter/cli.py +++ b/python/exporter/cli.py @@ -23,16 +23,19 @@ import asyncio import getpass import logging +import os import pydoc import sys import time +import traceback from pathlib import Path from typing import Sequence from findmy import InvalidCredentialsError, LoginState, MobileMeDelegateError, TermsError from findmy.errors import UnhandledProtocolError +from findmy.keychain.recovery import RecoveryError -from exporter import icloud, localsource, prompts, source, terms +from exporter import icloud, localsource, prompts, secrets, source, terms from exporter.codes import ( VERIFICATION_CODE_LENGTH, is_verification_code, @@ -69,12 +72,23 @@ def build_parser() -> argparse.ArgumentParser: "Nothing is saved: your Apple ID password and your device passcode are used once and" " dropped, and no account file is written." ), + # **Off because it silently recreates the flag this deliberately does not have.** argparse + # accepts any unambiguous prefix by default, so `--password hunter2` is taken as + # `--password-file hunter2` - the exact spelling somebody reaches for when they want to put + # a secret on the command line, quietly accepted. It then fails looking for a file named + # `hunter2`, by which point the password is already in `ps` output and shell history. + # + # Abbreviations worth having are spelled out as aliases instead, which is the same + # convenience without a rule that invents flag names nobody wrote down. + allow_abbrev=False, ) parser.add_argument("--version", action="version", version=VERSION) parser.add_argument( "-o", "--output", + # Spelled out rather than left to prefix matching, which is off - see allow_abbrev above. + "--out", type=Path, help="Where to write the bundle. Defaults to a timestamped zip in the current directory.", ) @@ -125,6 +139,68 @@ def build_parser() -> argparse.ArgumentParser: " account at all, for a bundle of nothing but --add-keys tags." ), ) + scripted = parser.add_argument_group( + "Running without a person", + "Answers given up front are not asked for. Sign in once by hand first: this machine is" + " then a device Apple knows, and later runs usually skip the verification code.", + ) + scripted.add_argument( + "--apple-id", + help="The Apple ID to sign in as, instead of being asked for it.", + ) + scripted.add_argument( + "--password-file", + type=Path, + metavar="PATH", + help=( + f"Read the Apple ID password from this file, or from standard input if it is '-'." + f" The file must not be readable by other users. ${secrets.APPLE_PASSWORD_VAR} is" + f" also read, and is second best. There is deliberately no --password flag: a" + f" command line is visible to everyone on the machine." + ), + ) + scripted.add_argument( + "--passcode-file", + type=Path, + metavar="PATH", + help=( + f"Read the device screen-lock passcode the same way." + f" ${secrets.DEVICE_PASSCODE_VAR} is also read." + ), + ) + scripted.add_argument( + "--device", + metavar="SERIAL", + help=( + "Unlock using the escrow record of the device with this serial, instead of asking" + " which one. Match is case-insensitive." + ), + ) + scripted.add_argument( + "--all-tags", + action="store_true", + help=( + "Export every accessory found, instead of asking which. Your own iPhones, iPads and" + " Macs are left out unless --include-my-devices is given as well." + ), + ) + scripted.add_argument( + "--include-my-devices", + action="store_true", + help=( + "Let --all-tags include your own devices. A bundle holding your MacBook lets whoever" + " receives it locate you, so this is never implied." + ), + ) + scripted.add_argument( + "--non-interactive", + action="store_true", + help=( + "Fail rather than ask anything. Without it, a missing answer is prompted for and a" + " scripted run hangs waiting for a keystroke that never comes." + ), + ) + parser.add_argument( "--anisette-url", help=( @@ -166,9 +242,66 @@ def configure_logging(verbosity: int) -> None: ) if verbosity >= 2: - for name in ("findmy.cloudkit", "findmy.keychain", "findmy.icloud", "exporter"): + # **The package, not a list of its subpackages.** This used to name `findmy.cloudkit`, + # `findmy.keychain` and `findmy.icloud`, which is every part anybody had needed so far and + # therefore wrong the first time a new one mattered: `announce_device` logs under + # `findmy.reports`, so the diagnostic written specifically to explain a failure was + # discarded by the flag turned on to see it. + # + # The reason it was a list is still real - the *root* logger at DEBUG carries aiohttp and + # asyncio, which drown everything about Apple in socket chatter. `findmy` is the level that + # excludes those without having to predict which of its modules will matter next. + for name in ("findmy", "exporter"): logging.getLogger(name).setLevel(logging.DEBUG) + if verbosity: + warn_about_sharing_logs() + + +def warn_about_sharing_logs() -> None: + """ + Say what is about to be printed, before it is printed. + + **Said here rather than only in the docs**, because the person who turns this on is usually + about to paste the result into an issue, and by then it has scrolled past. The docs are read + before a first run; this is read at the moment it matters. + + The list is specific on purpose. "Contains personal information" is easy to skim past and + tells nobody what to look for; a device serial and a keychain `acct` field are things somebody + can actually find and delete. + """ + lines = [ + " Verbose output is for debugging, not for publishing. It names your devices by", + " name, model and serial, prints keychain item attributes as Apple stores them,", + " and identifies every device in your trust circle.", + "", + " No key, password or passcode is logged. But this cannot promise a given", + " identifier never appears, because the text comes from a library reading", + " Apple's own structures.", + "", + " Read it and strip anything that identifies you before pasting it anywhere.", + ] + + print("", file=sys.stderr) + for line in lines: + print(_red(line), file=sys.stderr) + print("", file=sys.stderr) + + +def _red(text: str) -> str: + """ + Colour, when there is a terminal to colour and the reader has not asked otherwise. + + **Guarded rather than unconditional.** The likely next step after reading this is piping the + run to a file to attach it, and escape codes written into that file are noise in the issue - + which is the opposite of what the warning is for. `NO_COLOR` is honoured because it costs one + lookup and somebody has already decided. + """ + if not sys.stderr.isatty() or os.environ.get("NO_COLOR"): + return text + + return f"\033[31m{text}\033[0m" + # --------------------------------------------------------------------------------------------- # Asking @@ -183,6 +316,46 @@ async def ask_choice(question: str, options: Sequence[str]) -> int: ) +async def _retry_credentials(error, attempt: int): + """ + Offer the Apple ID and password again after Apple rejects them. + + Both are asked for again rather than only the password: the Apple ID may be the wrong one, and + a screen that will not let you correct it is worse than one extra field to press enter on. + """ + print(f"\nApple would not accept that sign-in:\n\n {error}\n", file=sys.stderr) + print(f"Attempt {attempt} of {icloud.MAX_LOGIN_ATTEMPTS}. Apple locks an account after enough", + file=sys.stderr) + print("failed sign-ins, so it is worth being sure rather than guessing.\n", file=sys.stderr) + + if not await prompts.confirm("Try again?"): + return None + + email = await prompts.text("Apple ID") + + return email, await prompts.password("Password") + + +async def _retry_code(error, attempt: int): + """ + Offer the verification code again, and only send a new one if asked. + + **Re-typing is the default and sending a new code is not**, because a resend invalidates the + code Apple already sent - so somebody who simply mistyped the code in front of them would have + it taken away by the recovery step. + """ + print(f"\nApple would not accept that code:\n\n {error}\n", file=sys.stderr) + print(f"Attempt {attempt} of {icloud.MAX_CODE_ATTEMPTS}.", file=sys.stderr) + + choice = await ask_choice("What now?", [ + "Type the code again (the one Apple sent is still valid)", + "Send a new code (this cancels the one already sent)", + "Stop", + ]) + + return [icloud.CODE_AGAIN, icloud.CODE_RESEND, None][choice] + + async def _ask_verification_code() -> str: """ Ask for the code Apple sent, and keep asking until it could be one. @@ -292,8 +465,11 @@ async def sign_in(arguments: argparse.Namespace): print("program, not a Mac you own. Remove it any time at account.apple.com > Devices.\n", file=sys.stderr) - email = await prompts.text("Apple ID") - password = await prompts.password("Password") + email = arguments.apple_id or await prompts.text("Apple ID") + password = ( + secrets.read(arguments.password_file, secrets.APPLE_PASSWORD_VAR) + or await prompts.password("Password") + ) try: # Nested rather than another `except` clause beside the one below, and that is not a style @@ -307,6 +483,8 @@ async def sign_in(arguments: argparse.Namespace): password, choose_second_factor=lambda methods: ask_choice("How should Apple send the code?", methods), get_code=_ask_verification_code, + retry_credentials=_retry_credentials, + retry_code=_retry_code, ) except MobileMeDelegateError as e: # Authentication itself worked; the exchange that follows it did not. Unaccepted terms @@ -340,7 +518,7 @@ async def sign_in(arguments: argparse.Namespace): return account -async def unlock(client) -> bool: +async def unlock(client, arguments: argparse.Namespace) -> bool: """ Recover the keychain keys, which needs the passcode of one of the account's devices. @@ -359,20 +537,79 @@ async def unlock(client) -> bool: print("\nUnlocking needs the screen-lock passcode of one of your Apple devices -", file=sys.stderr) print("its PIN or login password, not your Apple ID password.\n", file=sys.stderr) - chosen = options.recoverable[await ask_choice( - "Which device's passcode do you have?", - [record.describe() for record in options.recoverable], - )] + chosen = _pick_device(options.recoverable, arguments.device) or options.recoverable[ + await ask_choice( + "Which device's passcode do you have?", + [record.describe() for record in options.recoverable], + ) + ] + + # Read once, outside the loop: re-reading a file or the environment on every attempt would + # retry the identical value three times and report it as three rejections. + supplied = secrets.read(arguments.passcode_file, secrets.DEVICE_PASSCODE_VAR) - passcode = await prompts.password(f"Screen-lock passcode for {chosen.serial}") - try: - await client.unlock(chosen, passcode) - finally: - del passcode + async def ask(attempt: int) -> str: + if attempt == 1 and supplied is not None: + return supplied + + again = " (try again)" if attempt > 1 else "" + return await prompts.password(f"Screen-lock passcode for {chosen.serial}{again}") + + async def rejected(error, attempt: int) -> bool: + # The library's own text, printed whole. It says what was rejected and then three things + # worth doing about it, in the order worth doing them - the first being to try the same + # passcode again, because this call has been seen to fail intermittently. + print(f"\nThat was not accepted:\n\n{error}\n", file=sys.stderr) + print(f"Attempt {attempt} of {icloud.MAX_UNLOCK_ATTEMPTS}.", file=sys.stderr) + + return await prompts.confirm("Try again?") + + await icloud.unlock(client, chosen, ask, rejected) return True +def _pick_device(recoverable, serial: str | None): + """ + Find the escrow record `--device` names, so a scripted run does not have to answer a menu. + + :raises ExportSourceError: If nothing matches. Naming what is available, because a serial is + easy to mistype and the alternative is a run that silently unlocks with the wrong device. + """ + if serial is None: + return None + + wanted = serial.strip().casefold() + for record in recoverable: + if (record.serial or "").strip().casefold() == wanted: + return record + + available = ", ".join(sorted(r.serial for r in recoverable if r.serial)) or "none" + msg = f"No recoverable device on this account has the serial {serial!r}. Available: {available}" + raise ExportSourceError(msg) + + +def _take_all(candidates: list[Candidate], *, include_my_devices: bool) -> list[Candidate]: + """ + Everything, for `--all-tags`, with the account's own hardware left out by default. + + **"All my tags" and "all my tags plus the laptop I am sitting at" are different requests**, and + only one of them is what somebody writing a cron job meant. The interactive path names each + device and asks again before including one; a scripted run has nobody to ask, so the safe half + is the default and the other half has its own flag. + """ + if include_my_devices: + return list(candidates) + + keeping = [c for c in candidates if not is_own_device(c.owned_beacon)] + + for left_out in [c for c in candidates if is_own_device(c.owned_beacon)]: + print(f"Leaving out {left_out.label}: it is one of your own devices." + " --include-my-devices exports it anyway.", file=sys.stderr) + + return keeping + + async def choose(candidates: list[Candidate], *, or_nothing: bool = False) -> list[Candidate]: """ Ask which accessories to export. @@ -549,7 +786,7 @@ async def read_icloud(arguments: argparse.Namespace): try: async with await icloud.open_client(account) as client: - if not await unlock(client): + if not await unlock(client, arguments): return None return await icloud.fetch(client) @@ -591,7 +828,12 @@ async def run(arguments: argparse.Namespace) -> int: print(f"Not exportable: {skipped.beacon_id} {skipped.reason}", file=sys.stderr) if fetched.candidates: - exports = await name_the_nameless(await choose(fetched.candidates, or_nothing=bool(prepared))) + picked = ( + _take_all(fetched.candidates, include_my_devices=arguments.include_my_devices) + if arguments.all_tags + else await choose(fetched.candidates, or_nothing=bool(prepared)) + ) + exports = await name_the_nameless(picked) else: print(f"\nNothing on {where} can be exported.", file=sys.stderr) @@ -633,6 +875,20 @@ def main(argv: Sequence[str] | None = None) -> int: configure_logging(arguments.verbose) + if arguments.non_interactive: + prompts.forbid_prompting() + + try: + return _run_and_return(arguments) + finally: + # **Again, at the end.** The copy printed when logging was turned on is thousands of + # lines up by now, and the end of the run is both where somebody is looking and where + # they start selecting from. A warning that has scrolled away is one that was not given. + if arguments.verbose: + warn_about_sharing_logs() + + +def _run_and_return(arguments: argparse.Namespace) -> int: try: return asyncio.run(run(arguments)) except prompts.Abandoned: @@ -647,13 +903,42 @@ def main(argv: Sequence[str] | None = None) -> int: print("password are both worth checking - a code expires quickly, and Apple reports", file=sys.stderr) print("either as a credentials failure at this point. Nothing was written.", file=sys.stderr) return 1 - except (ExportError, ExportSourceError, KeyFileError, CustomTagError, TermsError) as e: + except ( + ExportError, ExportSourceError, KeyFileError, CustomTagError, TermsError, + secrets.SecretError, prompts.PromptForbidden, + ) as e: print(f"\n{e}", file=sys.stderr) return 1 + except RecoveryError as e: + # **Before the handler below, and deliberately not through it.** RecoveryError is an + # UnhandledProtocolError by inheritance, but it is not an unmodelled shape: it is a + # documented rejection that carries its own advice, and "Apple returned something + # unexpected" over the top of that is both wrong and in the way. A stack says nothing here + # either, so it does not print one. + print(f"\nThe keychain could not be unlocked:\n\n{e}\n", file=sys.stderr) + print("Nothing was written.", file=sys.stderr) + return 1 except UnhandledProtocolError as e: # Apple said something this library does not model. Worth its own message: it is not the # user's mistake and there is nothing for them to correct. print(f"\nApple returned something unexpected: {e}", file=sys.stderr) + + # **And the traceback, because this is the one error where it is the whole diagnosis.** + # The message names a symptom - a length that disagrees with the bytes around it - and + # says nothing about which structure was being read. There are a dozen places that parse + # DER here and the message is identical from all of them, so without the stack a report + # of this cannot be acted on at all. That is not hypothetical: issue #89 arrived with + # exactly this line and nothing else, and it could not be placed. + if arguments.verbose: + traceback.print_exc() + else: + print("\nRe-run with -vv and include the output if you report this. Without the", + file=sys.stderr) + print("stack this message names a symptom and not a place. That run will say what", + file=sys.stderr) + print("it prints about you, and it is worth reading before pasting it anywhere.", + file=sys.stderr) + return 1 except KeyboardInterrupt: print("\nStopped. Nothing was written.", file=sys.stderr) diff --git a/python/exporter/icloud.py b/python/exporter/icloud.py index c5453171..4fa7b396 100644 --- a/python/exporter/icloud.py +++ b/python/exporter/icloud.py @@ -27,6 +27,7 @@ from findmy import ( AsyncAppleAccount, + InvalidCredentialsError, LocalAnisetteProvider, LoginState, RemoteAnisetteProvider, @@ -46,10 +47,11 @@ ) from findmy.cloudkit.client import AsyncCloudKitClient from findmy.icloud import AsyncFindMyClient +from findmy.keychain.recovery import RecoveryError from findmy.keychain.session import AsyncKeychainSession from exporter import device -from exporter.identity import CLOUDKIT_DEVICE_NAME, EXPORTER_SERIAL +from exporter.identity import DEVICE_NAME, EXPORTER_SERIAL from opentagviewer_export import AccessoryExport from opentagviewer_export.hardware import identify @@ -150,7 +152,7 @@ def make_account(anisette_url: str | None = None, libs_path: str | None = None) provider = _make_provider(anisette_url, libs_path, stored) if stored is None: - return AsyncAppleAccount(provider) + return AsyncAppleAccount(provider, device_name=DEVICE_NAME) # Only the ids are restored. The rest of the shape has to be there because the library reads # it, and every field of it is empty on purpose - this is a logged-out account that happens to @@ -165,7 +167,7 @@ def make_account(anisette_url: str | None = None, libs_path: str | None = None) logger.info("Reusing the stored device identity, so this is not a new device to Apple") - return AsyncAppleAccount(provider, state_info=state) + return AsyncAppleAccount(provider, state_info=state, device_name=DEVICE_NAME) def _make_provider( @@ -197,8 +199,24 @@ def remember(account: AsyncAppleAccount) -> None: Called once signing in has worked, rather than at the end: the device is registered by then, and an export that is abandoned afterwards should not leave an entry nothing can reuse. + + **This does not name the entry, and nothing here can.** Signing in registers the device by + itself (findmy-export 01-authentication §13) and Apple synthesises its row from the client + identity, so the name defaults to the claimed hardware - `MacBookPro`. The only call that sets + a name is the `postdata` announce of Stage 2 §7, and Apple refuses it without a push token: + + HTTP 401, body {'ec': -800012, 'em': 'Push token is invalid.'} + + A push token is what makes a registered device trusted for verification codes, and §13 is + explicit that this is "a boundary to hold rather than a gap to close" - an exporter that became + a second factor for somebody's Apple ID would be a much larger thing to have compromised. So + the announce was tried, refused, and removed rather than left failing on every fresh install. + + **The serial is the label instead**, which is what §13 designed it for: `X-Apple-I-SRL-NO` is + sent during sign-in, needs no announce, and `0PENTAGXPORT` is the one field in that row a + person can actually read. See :mod:`exporter.identity`. """ - # The account's own serialisation, with only the two harmless parts taken out of it. It also + # The account's own serialisation, with only the harmless parts taken out of it. It also # carries the username, the password and the login state - which is exactly why this picks # fields rather than handing the whole mapping to `device.save`. state = account.to_json() @@ -206,6 +224,31 @@ def remember(account: AsyncAppleAccount) -> None: device.save(state["ids"]["uid"], state["ids"]["devid"], state["anisette"]) +MAX_LOGIN_ATTEMPTS = 3 +""" +How many times an Apple ID and password may be offered in one run. + +Small on purpose. **Apple locks an account after enough failed sign-ins**, and how many is not a +thing to establish by experiment on somebody's real account - the cost of being wrong is not a +failed export, it is an account they have to go and recover. +""" + +MAX_CODE_ATTEMPTS = 3 +"""How many times a verification code may be submitted before the sign-in is given up on.""" + +CODE_AGAIN = "again" +"""Ask for the code again. The one Apple already sent is still valid, so this is the usual answer.""" + +CODE_RESEND = "resend" +""" +Have Apple send a new code first. + +**This invalidates the code already sent** (findmy-export 01-authentication §5), so it is the right +answer only when that one is gone or expired - never a default, and never something to do on the +user's behalf. +""" + + async def log_in( account: AsyncAppleAccount, email: str, @@ -213,16 +256,27 @@ async def log_in( *, choose_second_factor: Callable[[Sequence[str]], Awaitable[int]], get_code: Callable[[], Awaitable[str]], + retry_credentials: Callable[[Exception, int], Awaitable[tuple[str, str] | None]] | None = None, + retry_code: Callable[[Exception, int], Awaitable[str | None]] | None = None, ) -> None: """ Sign in, dealing with two-factor authentication if the account asks for it. + Both things the user types can be got wrong, and neither ends the run when it is. See + :func:`unlock` for why that matters: everything before a failure has to be done again, and the + verification code in particular is not a thing that can simply be re-entered once it is gone. + :param choose_second_factor: Given the methods as text, returns which one to use. Awaited, because asking is drawn on a terminal and drawing happens inside this same event loop. :param get_code: Returns the code the user received. Awaited, as above. + :param retry_credentials: Awaited with the error and the attempt number when Apple rejects the + Apple ID or password; returns a replacement pair, or None to stop. None means one attempt. + :param retry_code: Awaited with the error and the attempt number when Apple rejects the + verification code; returns :data:`CODE_AGAIN`, :data:`CODE_RESEND`, or None to stop. :raises ExportSourceError: If signing in does not end signed in. + :raises InvalidCredentialsError: If the last attempt is rejected, or the user stops. """ - state = await account.login(email, password) + state = await _log_in_with_retries(account, email, password, retry_credentials) if state == LoginState.REQUIRE_2FA: methods = await account.get_2fa_methods() @@ -233,12 +287,70 @@ async def log_in( chosen = methods[await choose_second_factor([_describe_factor(m) for m in methods])] await chosen.request() - state = await chosen.submit(await get_code()) + state = await _submit_code_with_retries(chosen, get_code, retry_code) if state != LoginState.LOGGED_IN: raise ExportSourceError(f"Signing in ended at {state} rather than signed in.") +async def _log_in_with_retries(account, email, password, retry_credentials): + """ + Offer the Apple ID and password, letting a rejected pair be corrected in place. + + **Capped, and never automatic.** Apple locks an account after enough failed sign-ins, and how + many is not this project's to discover on somebody's real account - so `retry_credentials` is + asked every time and the count is small. A caller that passes None gets exactly one attempt, + which is what this did before. + """ + for attempt in range(1, MAX_LOGIN_ATTEMPTS + 1): + try: + return await account.login(email, password) + except InvalidCredentialsError as e: + logger.info("Apple rejected the credentials on attempt %d", attempt) + + replacement = ( + None + if retry_credentials is None or attempt == MAX_LOGIN_ATTEMPTS + else await retry_credentials(e, attempt) + ) + if replacement is None: + raise + + email, password = replacement + + raise AssertionError("unreachable") # pragma: no cover + + +async def _submit_code_with_retries(chosen, get_code, retry_code): + """ + Submit the verification code, letting a mistyped one be corrected. + + **The default retry re-types the code rather than asking for a new one**, and the difference is + not cosmetic: requesting delivery again sends a second code *and invalidates the first* + (findmy-export 01-authentication §5). So somebody who mistyped a code they are still holding + would lose it by being "helpfully" sent another, and a resend has to be a thing they choose. + """ + for attempt in range(1, MAX_CODE_ATTEMPTS + 1): + try: + return await chosen.submit(await get_code()) + except InvalidCredentialsError as e: + logger.info("Apple rejected the verification code on attempt %d", attempt) + + choice = ( + None + if retry_code is None or attempt == MAX_CODE_ATTEMPTS + else await retry_code(e, attempt) + ) + if choice is None: + raise + + if choice == CODE_RESEND: + # Deliberate: this is what invalidates whatever they were holding. + await chosen.request() + + raise AssertionError("unreachable") # pragma: no cover + + def _describe_factor(method: object) -> str: """Name one second-factor method for a person choosing between them.""" if isinstance(method, TrustedDeviceSecondFactorMethod): @@ -262,7 +374,7 @@ async def open_client(account: AsyncAppleAccount) -> AsyncFindMyClient: account, client=AsyncCloudKitClient( account, - device_name=CLOUDKIT_DEVICE_NAME, + device_name=DEVICE_NAME, device_serial=EXPORTER_SERIAL, ), ) @@ -273,6 +385,57 @@ async def open_client(account: AsyncAppleAccount) -> AsyncFindMyClient: return AsyncFindMyClient(account, session, store) +MAX_UNLOCK_ATTEMPTS = 3 +""" +How many times a passcode may be offered in one run. + +A bound rather than a free retry, because **attempts are probably a limited resource**. FindMy.py +says Apple's escrow services generally cap them and that what this one allows is not established, +which is a good reason not to find out on somebody's real account. +""" + + +async def unlock(client, record, ask_passcode, on_rejection) -> None: + """ + Recover the keychain keys, letting a rejected passcode be retried in place. + + **The retry is the point.** Both entry points used to make exactly one attempt, and a rejection + propagated out of the whole export - so a typo in a hidden field cost the sign-in, the + verification code and everything after it, and the user started again from the Apple ID. + + That was the wrong shape for this particular failure. The library's first advice on a rejection + is to try again *with the same passcode*, because the call has been observed to fail + intermittently and then succeed; the second is that a wrong passcode and a mis-parameterised + exchange are indistinguishable here, so a rejection is not proof the passcode was wrong. Both + say retry, and only this loop makes retrying cheap. + + **Nothing retries on its own.** :func:`on_rejection` is asked each time and a false answer + stops, so the decision to spend another attempt is always the user's - which matters because + the cap is unknown and the cost of exhausting it is not recoverable from here. + + :param client: An open Find My client. + :param record: The escrow record chosen from `recovery_options()`. + :param ask_passcode: Awaited for a passcode, given the attempt number, starting at 1. + :param on_rejection: Awaited with the error and the attempt number; returns whether to ask + again. Not called after the last attempt, where there is nothing to decide. + :raises RecoveryError: If the last attempt is rejected, or the user declines another. + """ + for attempt in range(1, MAX_UNLOCK_ATTEMPTS + 1): + passcode = await ask_passcode(attempt) + try: + await client.unlock(record, passcode) + return + except RecoveryError as e: + logger.info("Escrow recovery rejected attempt %d of %d", attempt, MAX_UNLOCK_ATTEMPTS) + + if attempt == MAX_UNLOCK_ATTEMPTS or not await on_rejection(e, attempt): + raise + finally: + # The passcode is not this program's to keep a moment longer than the call needs it, + # and a retry loop holds one across an interaction that waits on a person. + del passcode + + async def fetch(client: AsyncFindMyClient) -> Fetched: """ Fetch and decrypt the account's accessories, as the records a bundle carries. diff --git a/python/exporter/identity.py b/python/exporter/identity.py index dc58f9ae..b020c94f 100644 --- a/python/exporter/identity.py +++ b/python/exporter/identity.py @@ -38,10 +38,29 @@ Android app's `0PENTAGVIEWR` so a user seeing both recognises them as the same project. """ -CLOUDKIT_DEVICE_NAME = "OpenTagViewer Exporter" +DEVICE_NAME = "OpenTagViewer Exporter" """ -What CloudKit is told this client is called. +What this client tells CloudKit it is called. -FindMy.py defaults it to `FindMy.py`, which would name the library rather than the program in the -one place this is reported. +FindMy.py otherwise defaults it to `FindMy.py`, which names the library rather than the program in +the one place this is reported. + +**It is not the name in the account's device list, and nothing here can set that.** Signing in +registers the device by itself and Apple builds its row from the client identity, so the name +defaults to the claimed hardware - `MacBookPro`. The only call that sets one is the `postdata` +announce of findmy-export 02-mobileme-delegate §7, and Apple refuses it without a push token +(`ec -800012`, "Push token is invalid."). + +That is a boundary rather than a gap: a push token is what makes a registered device trusted for +verification codes, and 01-authentication §13 argues at length that this program must never become +a second factor for somebody's Apple ID. The row says so, and it is worth more than a better name: + + This device cannot be used to receive Apple Account verification codes. + +**So the serial is the label.** §13's three display fields are three independent levers, and the +serial is the only one sign-in controls that a person actually reads - which is why +:data:`EXPORTER_SERIAL` is legible rather than plausible. """ + +CLOUDKIT_DEVICE_NAME = DEVICE_NAME +"""Older spelling, kept for anything still importing it.""" diff --git a/python/exporter/prompts.py b/python/exporter/prompts.py index fcf9a56a..1c43c07e 100644 --- a/python/exporter/prompts.py +++ b/python/exporter/prompts.py @@ -87,6 +87,43 @@ class Abandoned(Exception): """Raised when the user pressed Ctrl-C at a prompt rather than answering it.""" +class PromptForbidden(Exception): + """ + Raised instead of asking, when the run was told there is nobody to ask. + + Carries the question so the failure names what was missing rather than only that something + was. "It wanted to ask: Screen-lock passcode for F2L…" is actionable; "cannot prompt" is not. + """ + + def __init__(self, question: str) -> None: + super().__init__( + f"This run cannot ask questions (--non-interactive), and it needed to ask:" + f"\n\n {question}\n\n" + f"Supply it up front, or drop --non-interactive. See --help.", + ) + self.question = question + + +_forbidden = False + + +def forbid_prompting(forbidden: bool = True) -> None: + """ + Make every prompt raise rather than ask. + + **Here rather than at each call site**, because a scripted run that hangs on a question is the + exact failure this prevents, and a check that has to be remembered at a dozen prompts will be + missed at the thirteenth. A prompt added later is covered without anybody thinking about it. + """ + global _forbidden + _forbidden = forbidden + + +def _refuse_to_ask(question: str) -> None: + if _forbidden: + raise PromptForbidden(question) + + def interactive() -> bool: """ Whether there is a terminal to draw on, at both ends. @@ -99,6 +136,8 @@ def interactive() -> bool: async def select(question: str, options: Sequence[Option]) -> Any: """Pick exactly one. Arrow keys where possible, a numbered list where not.""" + _refuse_to_ask(question) + if not options: raise ValueError("Nothing to choose between.") @@ -127,6 +166,8 @@ async def checkbox(question: str, options: Sequence[Option], instruction: str = list that arrives pre-ticked turns "hand over one tag" and "hand over everything I own" into the same keystroke. """ + _refuse_to_ask(question) + if not options: raise ValueError("Nothing to choose between.") @@ -149,6 +190,8 @@ async def checkbox(question: str, options: Sequence[Option], instruction: str = async def text(question: str, default: str = "") -> str: """Ask for a line of text.""" + _refuse_to_ask(question) + if not interactive(): print(f"{question}{f' [{default}]' if default else ''} ", end="", file=sys.stderr, flush=True) return input().strip() or default @@ -165,6 +208,8 @@ async def password(question: str) -> str: getpass either way: it is the one prompt where falling back to `input()` would echo a password into a terminal's scrollback, so the non-interactive path has to hide it too. """ + _refuse_to_ask(question) + if not interactive(): return getpass.getpass(f"{question} ") @@ -175,6 +220,8 @@ async def password(question: str) -> str: async def confirm(question: str, default: bool = False) -> bool: """Ask a yes or no question.""" + _refuse_to_ask(question) + if not interactive(): print(f"{question} [{'Y/n' if default else 'y/N'}] ", end="", file=sys.stderr, flush=True) typed = input().strip().lower() diff --git a/python/exporter/secrets.py b/python/exporter/secrets.py new file mode 100644 index 00000000..f146b9cb --- /dev/null +++ b/python/exporter/secrets.py @@ -0,0 +1,101 @@ +""" +Getting a secret into a scripted run without leaving it somewhere. + +**There is deliberately no `--password` flag, and there will not be one.** Anything on a command +line is readable by every other user on the machine through `ps`, ends up in shell history, and is +copied into any process listing a crash reporter or a CI log happens to capture. Offering the flag +guarantees somebody uses it, and the failure is silent - nothing goes wrong at the time. + +So the ways in, worst last: + +1. **A file, or standard input.** `--password-file secret.txt`, or `--password-file -` to read one + line from a pipe. Nothing appears in `ps`, nothing in the environment, nothing in history. A + file is checked for being readable by other users and refused if it is. +2. **An environment variable.** `OPENTAGVIEWER_APPLE_PASSWORD`. Better than a flag and worse than a + file: on Linux the environment of a running process is readable by the same user through + `/proc`, it is inherited by every child process, and setting it inline on a shell command puts + it in history anyway. Fine in a CI secret store, which is where most people will want it. +3. **Being asked**, which is what happens when neither is given and there is a terminal. + +The docs say the same thing, in the same order - see `docs/how-to-export-with-the-cli.md`. +""" + +from __future__ import annotations + +import os +import stat +import sys +from pathlib import Path + +APPLE_PASSWORD_VAR = "OPENTAGVIEWER_APPLE_PASSWORD" +DEVICE_PASSCODE_VAR = "OPENTAGVIEWER_DEVICE_PASSCODE" + +STDIN = "-" + + +class SecretError(Exception): + """A secret was asked for by file or environment and could not be read safely.""" + + +def read(path: Path | None, variable: str) -> str | None: + """ + Read a secret from a file or the environment, in that order of preference. + + :param path: Where to read it from. `-` reads one line from standard input. None skips to the + environment. + :param variable: The environment variable to fall back to. + :returns: The secret, or None if neither was given - in which case the caller should ask. + :raises SecretError: If a file was named and could not be read, or is readable by others. + """ + if path is not None: + return _from_file(path) + + # Not stripped of anything but the newline a file or a pipe adds. A password may legitimately + # begin or end with a space, and quietly removing one produces a rejected sign-in that looks + # like a wrong password - which is the hardest kind of bug to see, because the user is certain + # they typed it right and they did. + value = os.environ.get(variable) + + return value if value is None else value.removesuffix("\n") + + +def _from_file(path: Path) -> str: + if str(path) == STDIN: + line = sys.stdin.readline() + if not line: + msg = "Nothing arrived on standard input, so there is no secret to read." + raise SecretError(msg) + + return line.removesuffix("\n").removesuffix("\r") + + try: + mode = path.stat().st_mode + except OSError as e: + msg = f"Could not read {path}: {e}" + raise SecretError(msg) from None + + # **Refused rather than warned about.** A warning on a file anybody can read is advice nobody + # acts on, and the whole reason to prefer a file over an environment variable is that a file + # can have permissions. One that does not is worse than the option it was chosen over. + if mode & (stat.S_IRGRP | stat.S_IROTH): + msg = ( + f"{path} is readable by other users on this machine, so it is not a safe place for a" + f" secret. Run: chmod 600 {path}" + ) + raise SecretError(msg) + + try: + content = path.read_text(encoding="utf-8") + except OSError as e: + msg = f"Could not read {path}: {e}" + raise SecretError(msg) from None + + # The first line, so a file ending in a newline works and a file with a trailing comment line + # does not silently become part of the password. + secret = content.split("\n", 1)[0].removesuffix("\r") + + if not secret: + msg = f"{path} is empty, so there is no secret in it." + raise SecretError(msg) + + return secret diff --git a/python/exporter/version.py b/python/exporter/version.py index 2382785f..b08a8a11 100644 --- a/python/exporter/version.py +++ b/python/exporter/version.py @@ -19,7 +19,7 @@ from __future__ import annotations -VERSION = "1.1.0" +VERSION = "1.2.0" APP_TITLE = f"OpenTagViewer AirTag Exporter {VERSION}" diff --git a/python/exporter/wizard.py b/python/exporter/wizard.py index 06810f5b..e76de52c 100644 --- a/python/exporter/wizard.py +++ b/python/exporter/wizard.py @@ -17,6 +17,7 @@ from __future__ import annotations +import atexit import datetime import getpass import logging @@ -46,6 +47,9 @@ suggested_identifier, suggested_name, ) +from findmy import InvalidCredentialsError +from findmy.keychain.recovery import RecoveryError + from exporter.icloud import Candidate, ExportSourceError from exporter.version import APP_TITLE, EXPORT_VIA_WIZARD, VERSION from opentagviewer_export import ( @@ -230,6 +234,25 @@ def _load(self) -> None: except (ExportSourceError, ExportError) as e: messagebox.showerror("Could not read your accessories", str(e)) return + except InvalidCredentialsError as e: + # **Not the handler below, which asks for a bug report.** A password Apple rejected is + # the most ordinary thing that can happen here, and being told to go and file an issue + # about your own typo is both wrong and slightly insulting. Nothing was changed and the + # button is still there, so this says that and stops. + messagebox.showerror( + "Apple did not accept that sign-in", + f"{e}\n\nNothing was changed. You can try again whenever you like.", + ) + return + except RecoveryError as e: + # Same reasoning, for the passcode. Its message already carries FindMy.py's advice - + # starting with trying again, because this call fails intermittently - so it is shown + # whole rather than summarised. + messagebox.showerror( + "Could not unlock your keychain", + f"{e}\n\nNothing was changed. You can try again whenever you like.", + ) + return except Exception as e: # noqa: BLE001 - anything else is still the user's problem to see logger.exception("Reading accessories failed") messagebox.showerror( @@ -296,6 +319,12 @@ async def _read_icloud(self, asker: Asker): transform=verification_code, )), ), + retry_credentials=lambda error, attempt: _async( + _ask_again_for_credentials(self, asker, error, attempt), + ), + retry_code=lambda error, attempt: _async( + _ask_again_for_code(self, asker, error, attempt), + ), ) # Registered a device by now; remembering it stops the next export registering another. @@ -318,11 +347,30 @@ async def _read_icloud(self, asker: Asker): )) chosen = options.recoverable[index] - passcode = asker.ask(lambda: _ask_string( - self, "Unlock", f"Screen-lock passcode for {chosen.serial}:", secret=True, - )) + async def ask_passcode(attempt: int, chosen=chosen) -> str: + again = "\n\nThat last one was not accepted." if attempt > 1 else "" + return asker.ask(lambda: _ask_string( + self, + "Unlock", + f"Screen-lock passcode for {chosen.serial}:{again}", + secret=True, + )) + + async def rejected(error, attempt: int) -> bool: + # Offered rather than taken: the attempt cap is Apple's and unknown, so + # spending another one is the user's call. The library's text says why the + # first thing to try is the same passcode over again. + if asker.ask(lambda: messagebox.askyesno( + "Unlock", + f"That passcode was not accepted (attempt {attempt} of" + f" {icloud.MAX_UNLOCK_ATTEMPTS}).\n\n{error}\n\nTry again?", + parent=self, + )): + return True - await client.unlock(chosen, passcode) + raise _stopped() + + await icloud.unlock(client, chosen, ask_passcode, rejected) return await icloud.fetch(client) finally: @@ -757,6 +805,68 @@ async def _async(value): return value +def _ask_again_for_credentials(parent: tk.Tk, asker: Asker, error, attempt: int): + """ + Offer the Apple ID and password again after Apple rejects them. + + Both fields, not only the password: the Apple ID may be the one that is wrong, and a retry that + will not let it be corrected is a dead end with a text box in it. + """ + keep_going = asker.ask(lambda: messagebox.askretrycancel( + "Sign in", + f"Apple would not accept that sign-in (attempt {attempt} of" + f" {icloud.MAX_LOGIN_ATTEMPTS}).\n\n{error}\n\n" + "Apple locks an account after enough failed sign-ins, so it is worth being sure" + " rather than guessing.", + parent=parent, + )) + if not keep_going: + raise _stopped() + + email, password = asker.ask(lambda: _ask_credentials(parent)) + if not email or not password: + raise _stopped() + + return email, password + + +def _stopped() -> ExportSourceError: + """ + What a user pressing Cancel means, as distinct from what Apple said. + + **Returning None here would re-raise Apple's rejection**, and the window would then report + "Password authentication failed" to somebody whose last action was to deliberately stop. Worse, + it used to arrive through the generic handler, so choosing to give up on a typo ended in a + dialog asking them to file a bug with a log attached. + """ + return ExportSourceError("Signing in was stopped. Nothing was changed.") + + +def _ask_again_for_code(parent: tk.Tk, asker: Asker, error, attempt: int): + """ + Offer the verification code again, and send a new one only if asked. + + **Yes re-types, No sends a new one, Cancel stops**, and that order is deliberate: a resend + invalidates the code Apple already sent, so the common case - a mistyped code that is still + sitting on somebody's phone - must be the answer that does not destroy it. The legend is in the + message because a three-button dialog cannot label its own buttons here. + """ + answer = asker.ask(lambda: messagebox.askyesnocancel( + "Verification", + f"Apple would not accept that code (attempt {attempt} of" + f" {icloud.MAX_CODE_ATTEMPTS}).\n\n{error}\n\n" + "Yes - type the code again. The one Apple sent is still valid.\n" + "No - send a new code. This cancels the one already sent.\n" + "Cancel - stop signing in.", + parent=parent, + )) + + if answer is None: + raise _stopped() + + return icloud.CODE_AGAIN if answer else icloud.CODE_RESEND + + def log_file() -> Path: """ Where this writes its log. @@ -778,6 +888,70 @@ def log_file() -> Path: return directory / "exporter.log" +_WARNED_AT_THE_BOTTOM = False + +_CAUTION = ( + "=" * 88, + " This log is for debugging and is NOT safe to publish as-is.", + "", + " It names your devices by name, model and serial, records keychain item attributes", + " as Apple stores them, and identifies every device in your account's trust circle.", + " No key, password or passcode is written here, and payloads appear only as byte", + " counts - but nothing can promise a given identifier never appears, because the text", + " comes from a library reading Apple's own structures.", + "", + " READ THIS FILE AND REMOVE ANYTHING THAT IDENTIFIES YOU BEFORE SENDING IT ANYWHERE.", + "=" * 88, +) + + +def _warn_at_the_top_of_the_log() -> None: + """ + Put the caution in the file, at the start of every run. + + **The CLI can warn the person before they turn logging on. The wizard cannot**: it logs at + INFO unconditionally, because a windowed build has no console and a log that has to be + enabled is a log nobody has when it is needed. So the file exists on every machine that has + ever run this, and it holds what INFO holds - escrow records described by device name, model + and serial, keychain item attributes as Apple stores them, an identifier per peer in the + trust circle. + + The warning goes *in the file* rather than only in the window because that is what travels. + Somebody asked to "attach exporter.log" sends the file and never sees the window again, so + the caution has to be in what they open. + + **At both ends, for different reasons.** The file is appended to across runs and the end is + where anybody looks first, because that is where the failure they came for is - a banner only + at the top of a run is buried under that run's own output within seconds. But the end can only + be written on the way out, and the interesting runs are the ones that die. So the head copy is + the one that is always there, and the tail copy is the one that is actually read. + """ + for line in _CAUTION: + logging.getLogger("exporter.privacy").warning(line) + + atexit.register(_warn_at_the_bottom_of_the_log) + + +def _warn_at_the_bottom_of_the_log() -> None: + """ + Leave the caution as the last thing in the file, since that is where a reader starts. + + Guarded against running twice: `configure_logging` is called again by `--self-test`, and two + registrations would print the banner twice at exit. Harmless, and it reads like a bug in the + thing whose whole job is to be believed. + """ + global _WARNED_AT_THE_BOTTOM + if _WARNED_AT_THE_BOTTOM: + return + + _WARNED_AT_THE_BOTTOM = True + + for line in _CAUTION: + logging.getLogger("exporter.privacy").warning(line) + + logging.shutdown() + + def configure_logging() -> None: """Log to a file always, and to the console as well when there is one.""" handlers: list[logging.Handler] = [logging.FileHandler(log_file(), encoding="utf-8")] @@ -794,6 +968,8 @@ def configure_logging() -> None: force=True, ) + _warn_at_the_top_of_the_log() + # Anything that escapes Tk's callback handling as well, which otherwise vanishes the same way. def _report(exc_type, value, tb) -> None: logger.error("Unhandled: %s", "".join(traceback.format_exception(exc_type, value, tb))) diff --git a/python/test/test_retries.py b/python/test/test_retries.py new file mode 100644 index 00000000..8cc1b67a --- /dev/null +++ b/python/test/test_retries.py @@ -0,0 +1,416 @@ +""" +Retrying the three things a person types, instead of ending the run on a typo. + +**The behaviour under test is what a typo costs.** Every one of these made exactly one attempt, and +a rejection propagated out of the whole export - so one wrong character threw away everything +before it and the user began again from their Apple ID. + +Each stage has its own reason for retrying, and its own reason for being careful: + +- **The Apple ID and password.** Capped low, because Apple locks an account after enough failures + and how many is not a thing to establish by experiment on somebody's real account. +- **The verification code.** Re-typing must not request a new one: a resend invalidates the code + Apple already sent, so the recovery step would destroy what the user is holding. +- **The screen-lock passcode.** FindMy.py's own advice leads with trying again *with the same + passcode*, because the call fails intermittently and then succeeds. Attempts are probably capped + too, so nothing here retries without being asked. + +`asyncio.run` rather than pytest-asyncio, which this project does not depend on - the same way +`test_cli.py` drives the flow's other coroutines. +""" + +from __future__ import annotations + +import asyncio +import inspect + +import pytest +from findmy import InvalidCredentialsError, LoginState +from findmy.keychain.recovery import RecoveryError + +from exporter import icloud + +RECORD = object() + +REJECTION = ( + "Escrow proxy rejected recover: HTTP 409, status -6015 -- CLUBH ERROR: Credential is" + " not verified." +) + + +class FakeClient: + """ + A client that accepts one passcode and rejects every other. + + It records what it was given, so a test can assert on the passcodes tried and not only on how + many times - "it retried" and "it retried with the right thing" are different claims. + """ + + def __init__(self, accepts: str | None = None, *, reject_first: int = 0) -> None: + self.accepts = accepts + self.reject_first = reject_first + self.tried: list[str] = [] + + async def unlock(self, record, passcode: str) -> None: + assert record is RECORD + self.tried.append(passcode) + + # `reject_first` models the intermittent failure: the same passcode is refused and then + # accepted, which is the case the library says to expect. + if len(self.tried) <= self.reject_first: + raise RecoveryError(REJECTION) + + if self.accepts is not None and passcode != self.accepts: + raise RecoveryError(REJECTION) + + +class Typist: + """Types the given passcodes in order, and records the attempt numbers it was asked for.""" + + def __init__(self, *passcodes: str) -> None: + self.passcodes = passcodes + self.asked: list[int] = [] + + async def __call__(self, attempt: int) -> str: + self.asked.append(attempt) + + return self.passcodes[attempt - 1] + + +class Answer: + """Answers "try again?" the same way every time, recording which attempts asked.""" + + def __init__(self, reply: bool) -> None: + self.reply = reply + self.asked: list[int] = [] + self.messages: list[str] = [] + + async def __call__(self, error, attempt: int) -> bool: + self.asked.append(attempt) + self.messages.append(str(error)) + + return self.reply + + +def run(client, typist, answer): + return asyncio.run(icloud.unlock(client, RECORD, typist, answer)) + + +class TestARejectedPasscodeCanBeRetried: + def test_a_typo_does_not_end_the_run(self): + client = FakeClient(accepts="1234") + + run(client, Typist("12e4", "1234"), Answer(True)) + + assert client.tried == ["12e4", "1234"] + + def test_the_same_passcode_again_is_a_normal_thing_to_do(self): + # The library's first advice on a rejection, because the call fails intermittently and the + # identical passcode then works. Nothing may require a retry to differ from what preceded + # it, which is easy to break by "helpfully" rejecting a repeat. + client = FakeClient(reject_first=1) + + run(client, Typist("1234", "1234"), Answer(True)) + + assert client.tried == ["1234", "1234"] + + def test_a_passcode_that_works_first_time_asks_nothing_further(self): + client, answer = FakeClient(accepts="1234"), Answer(True) + + run(client, Typist("1234"), answer) + + assert client.tried == ["1234"] + assert answer.asked == [] + + def test_the_attempt_number_counts_from_one(self): + # It is shown to the user as "attempt N of M", so an off-by-one is visible and alarming. + client, typist = FakeClient(accepts="1234"), Typist("wrong", "1234") + + run(client, typist, Answer(True)) + + assert typist.asked == [1, 2] + + +class TestNothingIsSpentWithoutBeingAsked: + """ + The cap is Apple's and its size is not established here, so an attempt is not this program's to + spend on its own. That is the whole reason the rejection callback returns a value. + """ + + def test_declining_stops_immediately(self): + client = FakeClient(accepts="1234") + + with pytest.raises(RecoveryError): + run(client, Typist("wrong"), Answer(False)) + + assert client.tried == ["wrong"], "declining must not try again" + + def test_it_gives_up_at_the_cap_rather_than_looping(self): + client = FakeClient(accepts="never matches") + typist = Typist(*["wrong"] * icloud.MAX_UNLOCK_ATTEMPTS) + + with pytest.raises(RecoveryError): + run(client, typist, Answer(True)) + + assert len(client.tried) == icloud.MAX_UNLOCK_ATTEMPTS + + def test_the_last_attempt_does_not_ask_whether_to_try_again(self): + # There is nothing to decide: the answer cannot be acted on either way, and asking implies + # an attempt is available when it is not. + client = FakeClient(accepts="never matches") + answer = Answer(True) + + with pytest.raises(RecoveryError): + run(client, Typist(*["wrong"] * icloud.MAX_UNLOCK_ATTEMPTS), answer) + + assert answer.asked == list(range(1, icloud.MAX_UNLOCK_ATTEMPTS)) + + def test_the_cap_is_more_than_one(self): + # Guards the point of the change: a cap of 1 is the old behaviour wearing a loop. + assert icloud.MAX_UNLOCK_ATTEMPTS > 1 + + +class TestWhatTheUserIsTold: + def test_the_librarys_own_advice_reaches_the_caller(self): + # The exporter writes no advice about escrow recovery of its own: FindMy.py's message + # carries the rejection and three ordered things to do about it, and a re-wording here + # would be a second version to keep in step with the first. + client, answer = FakeClient(accepts="1234"), Answer(True) + + run(client, Typist("wrong", "1234"), answer) + + assert len(answer.messages) == 1 + assert "-6015" in answer.messages[0] + + def test_the_error_that_escapes_is_a_recovery_error(self): + # `main` distinguishes this from an unmodelled protocol shape and reports it without a + # traceback, which only works if the type survives the loop. + client = FakeClient(accepts="never matches") + + with pytest.raises(RecoveryError, match="-6015"): + run(client, Typist("a", "b", "c"), Answer(True)) + + +class TestThePasscodeIsNotKept: + def test_it_is_released_after_every_attempt(self): + """ + The loop holds a passcode across a question that waits on a person, which is longer than + the one-shot version ever did. The `del` in the `finally` is what stops that becoming a + name that outlives the call, and it is invisible unless something says so. + """ + source = inspect.getsource(icloud.unlock) + + assert "del passcode" in source, "the passcode must not outlive the attempt" + assert source.index("finally") < source.index("del passcode") + + +class FakeAccount: + """An account that accepts one credential pair and rejects every other.""" + + def __init__(self, accepts=("me@example.com", "hunter2"), state=None) -> None: + self.accepts = accepts + self.state = state + self.tried: list[tuple[str, str]] = [] + + async def login(self, email: str, password: str): + self.tried.append((email, password)) + + if (email, password) != self.accepts: + raise InvalidCredentialsError("Apple rejected that.") + + return self.state + + +class FakeFactor: + """A second factor that accepts one code, and counts how often a new one was requested.""" + + def __init__(self, accepts: str = "123456") -> None: + self.accepts = accepts + self.submitted: list[str] = [] + self.requests = 0 + + async def request(self) -> None: + self.requests += 1 + + async def submit(self, code: str): + self.submitted.append(code) + + if code != self.accepts: + raise InvalidCredentialsError("Apple rejected that code.") + + return LoginState.LOGGED_IN + + +class TestAWrongPasswordCanBeCorrected: + def test_a_typo_does_not_end_the_run(self): + account = FakeAccount(state=LoginState.LOGGED_IN) + replacements = iter([("me@example.com", "hunter2")]) + + async def retry(_error, _attempt): + return next(replacements) + + asyncio.run(icloud.log_in( + account, "me@example.com", "wrong", + choose_second_factor=_unused, get_code=_unused, retry_credentials=retry, + )) + + assert account.tried == [("me@example.com", "wrong"), ("me@example.com", "hunter2")] + + def test_the_apple_id_can_be_corrected_too(self): + # Not only the password: the address may be the wrong one, and a retry that cannot fix it + # is a dead end. + account = FakeAccount(accepts=("right@example.com", "pw"), state=LoginState.LOGGED_IN) + + async def retry(_error, _attempt): + return ("right@example.com", "pw") + + asyncio.run(icloud.log_in( + account, "wrong@example.com", "pw", + choose_second_factor=_unused, get_code=_unused, retry_credentials=retry, + )) + + assert account.tried[-1] == ("right@example.com", "pw") + + def test_declining_stops_and_raises(self): + account = FakeAccount() + + async def decline(_error, _attempt): + return None + + with pytest.raises(InvalidCredentialsError): + asyncio.run(icloud.log_in( + account, "me@example.com", "wrong", + choose_second_factor=_unused, get_code=_unused, retry_credentials=decline, + )) + + assert len(account.tried) == 1 + + def test_it_stops_at_the_cap(self): + # Apple locks an account after enough failures, so this bound is the one that matters most. + account = FakeAccount() + + async def retry(_error, _attempt): + return ("me@example.com", "still wrong") + + with pytest.raises(InvalidCredentialsError): + asyncio.run(icloud.log_in( + account, "me@example.com", "wrong", + choose_second_factor=_unused, get_code=_unused, retry_credentials=retry, + )) + + assert len(account.tried) == icloud.MAX_LOGIN_ATTEMPTS + + def test_no_callback_means_one_attempt(self): + account = FakeAccount() + + with pytest.raises(InvalidCredentialsError): + asyncio.run(icloud.log_in( + account, "me@example.com", "wrong", + choose_second_factor=_unused, get_code=_unused, + )) + + assert len(account.tried) == 1 + + +class TestAWrongCodeCanBeCorrected: + def test_retyping_does_not_request_a_new_code(self): + """ + The property this whole design exists for. + + Requesting delivery again invalidates the code Apple already sent, so somebody who mistyped + a code still sitting on their phone must not have it taken away by the recovery step. + """ + factor = FakeFactor(accepts="123456") + codes = iter(["123455", "123456"]) + + async def retry(_error, _attempt): + return icloud.CODE_AGAIN + + state = asyncio.run(icloud._submit_code_with_retries( + factor, lambda: _next(codes), retry, + )) + + assert state == LoginState.LOGGED_IN + assert factor.submitted == ["123455", "123456"] + assert factor.requests == 0, "re-typing must not invalidate the code they hold" + + def test_asking_for_a_new_code_requests_one(self): + factor = FakeFactor(accepts="123456") + codes = iter(["999999", "123456"]) + + async def retry(_error, _attempt): + return icloud.CODE_RESEND + + asyncio.run(icloud._submit_code_with_retries(factor, lambda: _next(codes), retry)) + + assert factor.requests == 1 + + def test_stopping_raises_without_requesting_anything(self): + factor = FakeFactor(accepts="123456") + + async def stop(_error, _attempt): + return None + + with pytest.raises(InvalidCredentialsError): + asyncio.run(icloud._submit_code_with_retries( + factor, lambda: _next(iter(["000000"])), stop, + )) + + assert factor.requests == 0 + + def test_it_stops_at_the_cap(self): + factor = FakeFactor(accepts="never") + codes = iter(["1"] * icloud.MAX_CODE_ATTEMPTS) + + async def retry(_error, _attempt): + return icloud.CODE_AGAIN + + with pytest.raises(InvalidCredentialsError): + asyncio.run(icloud._submit_code_with_retries(factor, lambda: _next(codes), retry)) + + assert len(factor.submitted) == icloud.MAX_CODE_ATTEMPTS + + +async def _next(codes): + return next(codes) + + +async def _unused(*_args, **_kwargs): + raise AssertionError("should not have been asked") + + +class TestTheDiagnosticsSurvive: + """ + `-vv` has to cover the module that explains a failure, not the modules that were interesting + when the flag was written. + """ + + def test_verbose_covers_every_part_of_findmy(self): + """ + The regression this exists for. + + `configure_logging` used to enumerate `findmy.cloudkit`, `findmy.keychain` and + `findmy.icloud` - every part anybody had needed, and therefore wrong the first time a new + one mattered. `announce_device` logs under `findmy.reports`, so the DEBUG line written + specifically to explain an HTTP 401 was discarded by the flag turned on to read it. + """ + import logging + + from exporter import cli + + cli.configure_logging(2) + + assert logging.getLogger("findmy.reports.account").isEnabledFor(logging.DEBUG) + + def test_but_not_the_socket_chatter(self): + # Why it was a list rather than the root logger in the first place: aiohttp and asyncio at + # DEBUG bury everything about Apple. `findmy` is the level that excludes them without + # having to predict which of its modules matters next. + import logging + + from exporter import cli + + cli.configure_logging(2) + + assert not logging.getLogger("aiohttp").isEnabledFor(logging.DEBUG) + assert not logging.getLogger("asyncio").isEnabledFor(logging.DEBUG) diff --git a/python/test/test_scriptable.py b/python/test/test_scriptable.py new file mode 100644 index 00000000..c8c62210 --- /dev/null +++ b/python/test/test_scriptable.py @@ -0,0 +1,303 @@ +""" +Running the CLI with nobody at the keyboard. + +Two things are being protected, and the second is the reason this file is long. + +**That a scripted run does not hang.** The failure it replaces is not an error - it is a cron job +sitting forever on a question nobody will answer, which reports nothing and finishes never. So +`--non-interactive` has to cover *every* prompt, including ones added later, and the test for that +is written against the guard rather than against a list of prompts. + +**That a secret does not end up somewhere it can be read.** There is deliberately no `--password` +flag; a file must not be world-readable; and nothing is silently trimmed, because a password that +legitimately ends in a space would otherwise fail as though it were wrong. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from exporter import cli, prompts, secrets + + +@pytest.fixture(autouse=True) +def _prompting_allowed(): + """Every test starts able to prompt, whatever the last one did to the module-level flag.""" + prompts.forbid_prompting(False) + yield + prompts.forbid_prompting(False) + + +@pytest.fixture +def secret_file(tmp_path): + def write(content: str, mode: int = 0o600): + path = tmp_path / "secret" + path.write_text(content, encoding="utf-8") + path.chmod(mode) + return path + + return write + + +class TestThereIsNoPasswordFlag: + """ + The single most important property here, and the easiest to undo by being helpful. + + A command line is readable by every user on the machine through `ps`, and it lands in shell + history. Adding `--password` would be one line and would quietly undo the whole design, so a + test says it is absent rather than a comment asking people not to. + """ + + @pytest.mark.parametrize("flag", ["--password", "--passcode", "--apple-password"]) + def test_no_secret_can_be_given_on_the_command_line(self, flag, capsys): + with pytest.raises(SystemExit): + cli.build_parser().parse_args([flag, "hunter2"]) + + assert "unrecognized arguments" in capsys.readouterr().err + + def test_the_file_and_environment_routes_do_exist(self): + arguments = cli.build_parser().parse_args(["--password-file", "/dev/null"]) + + assert arguments.password_file is not None + assert secrets.APPLE_PASSWORD_VAR.startswith("OPENTAGVIEWER_") + + +class TestReadingASecretFromAFile: + def test_it_reads_the_first_line(self, secret_file): + assert secrets.read(secret_file("hunter2\n"), "UNSET_VAR") == "hunter2" + + def test_a_file_with_no_trailing_newline_works(self, secret_file): + assert secrets.read(secret_file("hunter2"), "UNSET_VAR") == "hunter2" + + def test_leading_and_trailing_spaces_are_kept(self, secret_file): + # A password may legitimately begin or end with a space. Trimming one produces a rejected + # sign-in that looks exactly like a wrong password, which is the worst kind to debug: the + # user is certain they typed it correctly, and they did. + assert secrets.read(secret_file(" hunter2 \n"), "UNSET_VAR") == " hunter2 " + + def test_a_second_line_is_not_part_of_the_secret(self, secret_file): + assert secrets.read(secret_file("hunter2\n# a note\n"), "UNSET_VAR") == "hunter2" + + def test_a_world_readable_file_is_refused(self, secret_file): + # Refused rather than warned about. The reason to prefer a file over an environment + # variable is that a file can have permissions; one that does not is worse than what it + # was chosen over, and a warning is advice nobody acts on. + path = secret_file("hunter2\n", mode=0o644) + + with pytest.raises(secrets.SecretError, match="readable by other users"): + secrets.read(path, "UNSET_VAR") + + def test_a_group_readable_file_is_refused_too(self, secret_file): + path = secret_file("hunter2\n", mode=0o640) + + with pytest.raises(secrets.SecretError, match="readable by other users"): + secrets.read(path, "UNSET_VAR") + + def test_the_advice_names_the_fix(self, secret_file): + path = secret_file("hunter2\n", mode=0o644) + + with pytest.raises(secrets.SecretError, match=f"chmod 600 {path}"): + secrets.read(path, "UNSET_VAR") + + def test_an_empty_file_is_an_error_rather_than_an_empty_password(self, secret_file): + with pytest.raises(secrets.SecretError, match="is empty"): + secrets.read(secret_file(""), "UNSET_VAR") + + def test_a_missing_file_says_so(self, tmp_path): + with pytest.raises(secrets.SecretError, match="Could not read"): + secrets.read(tmp_path / "nope", "UNSET_VAR") + + +class TestReadingASecretFromTheEnvironment: + def test_it_is_used_when_no_file_is_given(self, monkeypatch): + monkeypatch.setenv("A_TEST_VAR", "hunter2") + + assert secrets.read(None, "A_TEST_VAR") == "hunter2" + + def test_a_file_wins_over_the_environment(self, secret_file, monkeypatch): + # The better mechanism takes precedence, so a leftover variable cannot quietly override a + # deliberate file. + monkeypatch.setenv("A_TEST_VAR", "from the environment") + + assert secrets.read(secret_file("from the file\n"), "A_TEST_VAR") == "from the file" + + def test_neither_is_not_an_error(self, monkeypatch): + # It means "ask", and only the caller knows whether it may. + monkeypatch.delenv("A_TEST_VAR", raising=False) + + assert secrets.read(None, "A_TEST_VAR") is None + + def test_an_empty_variable_is_not_treated_as_a_password(self, monkeypatch): + monkeypatch.setenv("A_TEST_VAR", "") + + assert secrets.read(None, "A_TEST_VAR") == "" + + +class TestRefusingToAsk: + """ + `--non-interactive` covers every prompt, not a remembered list of them. + + The guard lives in `prompts` so a prompt added later is covered without anybody deciding to + cover it - which is the only version of this that stays true. + """ + + @pytest.mark.parametrize("call", [ + lambda: prompts.text("Apple ID"), + lambda: prompts.password("Password"), + lambda: prompts.confirm("Try again?"), + lambda: prompts.select("Which device?", [prompts.Option(label="a", value=1)]), + lambda: prompts.checkbox("What should go in?", [prompts.Option(label="a", value=1)]), + ]) + def test_every_prompt_raises_instead_of_asking(self, call): + prompts.forbid_prompting() + + with pytest.raises(prompts.PromptForbidden): + asyncio.run(call()) + + def test_the_failure_names_the_question_it_wanted_to_ask(self): + # "Cannot prompt" tells somebody reading a CI log nothing. The question tells them which + # value to supply. + prompts.forbid_prompting() + + with pytest.raises(prompts.PromptForbidden, match="Screen-lock passcode for F2LX"): + asyncio.run(prompts.password("Screen-lock passcode for F2LX")) + + def test_it_can_be_turned_back_off(self): + prompts.forbid_prompting() + prompts.forbid_prompting(False) + + # No exception: the flag is not one-way, which matters because it is module-level state + # and the tests share a process. + assert prompts._forbidden is False + + def test_every_public_prompt_is_guarded(self): + """ + The one that survives a refactor. + + Written against the module rather than a list, so a prompt added next year fails this + instead of quietly hanging a cron job. + """ + import inspect + + asked = [ + name for name, value in vars(prompts).items() + if inspect.iscoroutinefunction(value) and not name.startswith("_") + ] + + assert asked, "no prompts found - this test is checking nothing" + + for name in asked: + source = inspect.getsource(getattr(prompts, name)) + assert "_refuse_to_ask(" in source, f"prompts.{name} can hang a scripted run" + + +class FakeRecord: + def __init__(self, serial: str) -> None: + self.serial = serial + + def describe(self) -> str: + return f"A device, serial {self.serial}" + + +class TestChoosingTheDeviceByserial: + def test_it_finds_the_named_record(self): + records = [FakeRecord("AAAA"), FakeRecord("BBBB")] + + assert cli._pick_device(records, "BBBB").serial == "BBBB" + + def test_the_match_ignores_case_and_spaces(self): + # Serials get copied out of a device list, and a stray space is not worth a failed run. + records = [FakeRecord("F2LX9Q")] + + assert cli._pick_device(records, " f2lx9q ").serial == "F2LX9Q" + + def test_no_serial_means_ask(self): + assert cli._pick_device([FakeRecord("AAAA")], None) is None + + def test_an_unknown_serial_is_an_error_that_lists_what_there_is(self): + # Silently falling back to asking would hang a scripted run; silently picking the first + # would unlock with the wrong device. Neither is acceptable, so it says what exists. + records = [FakeRecord("AAAA"), FakeRecord("BBBB")] + + with pytest.raises(Exception, match="AAAA, BBBB"): + cli._pick_device(records, "CCCC") + + +class FakeCandidate: + def __init__(self, label: str, owned_beacon: dict) -> None: + self.label = label + self.owned_beacon = owned_beacon + + +AIRTAG = {"productId": 21760, "vendorId": 76} +A_MAC = {"model": "Mac16,1", "productId": 1, "vendorId": 76} + + +class TestTakingEverything: + def test_it_takes_the_tags(self): + candidates = [FakeCandidate("Wallet", AIRTAG), FakeCandidate("Backpack", AIRTAG)] + + taken = cli._take_all(candidates, include_my_devices=False) + + assert [c.label for c in taken] == ["Wallet", "Backpack"] + + def test_your_own_devices_are_left_out_by_default(self): + """ + The safety property of `--all-tags`. + + "All my tags" and "all my tags plus the laptop I am sitting at" are different requests, and + only one of them is what somebody writing a cron job meant. The interactive path names each + device and asks again; a scripted run has nobody to ask. + """ + candidates = [FakeCandidate("Wallet", AIRTAG), FakeCandidate("My MacBook", A_MAC)] + + taken = cli._take_all(candidates, include_my_devices=False) + + assert [c.label for c in taken] == ["Wallet"] + + def test_and_can_be_asked_for_explicitly(self): + candidates = [FakeCandidate("Wallet", AIRTAG), FakeCandidate("My MacBook", A_MAC)] + + taken = cli._take_all(candidates, include_my_devices=True) + + assert [c.label for c in taken] == ["Wallet", "My MacBook"] + + def test_leaving_one_out_is_said_rather_than_done_quietly(self, capsys): + # A bundle with fewer things in it than expected, and no line saying why, reads as a bug. + candidates = [FakeCandidate("My MacBook", A_MAC)] + + cli._take_all(candidates, include_my_devices=False) + + assert "My MacBook" in capsys.readouterr().err + + +class TestTheFlagsParse: + def test_the_scripted_run_needs_no_prompt_for_anything_but_the_code(self, tmp_path): + # The documented shape of a scripted invocation, asserted so the docs cannot drift from it. + password = tmp_path / "pw" + password.write_text("hunter2") + password.chmod(0o600) + + arguments = cli.build_parser().parse_args([ + "--non-interactive", + "--apple-id", "me@example.com", + "--password-file", str(password), + "--passcode-file", str(password), + "--device", "F2LX9Q", + "--all-tags", + "--no-password", + "--output", str(tmp_path / "out.zip"), + ]) + + assert arguments.non_interactive + assert arguments.apple_id == "me@example.com" + assert arguments.device == "F2LX9Q" + assert arguments.all_tags + assert not arguments.include_my_devices + + def test_including_your_own_devices_is_never_implied(self): + arguments = cli.build_parser().parse_args(["--all-tags"]) + + assert not arguments.include_my_devices diff --git a/python/uv.lock b/python/uv.lock index a04e92af..46b599be 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -659,7 +659,7 @@ wheels = [ [[package]] name = "findmy" version = "0.10.1" -source = { git = "https://github.com/parawanderer/FindMy.py?branch=feat%2Ficloud-keychain-export#dbed340006d05ff78a8af1a76d9dddd17e00f71f" } +source = { git = "https://github.com/parawanderer/FindMy.py?branch=feat%2Ficloud-keychain-export#acd17c507b63e8bd0e56d8b4c5a92c096050bfa3" } dependencies = [ { name = "aiohttp" }, { name = "anisette" },