"""What every action needs: the hub, the account name, the passphrase, the store.""" import argparse import getpass import os import sys import time from pathlib import Path from ..client.account import Account, NoPassphrase from ..client.hub import DEFAULT_HUB, Hub from ..client.paths import handle from ..client.store import Store from ..errors import USAGE, CliError from ..output import safe class Context: def __init__(self, args: argparse.Namespace): self.args = args self.store = Store() stored = self.store.peek() explicit_hub = args.hub or os.environ.get("MESHBAY_HUB") self.hub_url = (explicit_hub or stored.get("hub") or DEFAULT_HUB).rstrip("/") user = args.user or os.environ.get("MESHBAY_USER") self.username = user.strip() if user else None self.stored_username = stored.get("username") if stored.get("hub") == self.hub_url \ else None self.hub = Hub(self.hub_url) self._passphrase: str | None = None def passphrase(self) -> str: """--password, --password-file, MESHBAY_PASSWORD, then a prompt on a terminal.""" if self._passphrase is not None: return self._passphrase a = self.args value = a.password if value is None and a.password_file: try: text = sys.stdin.read() if a.password_file == "-" \ else Path(a.password_file).read_text(encoding="utf-8") except OSError as e: raise CliError(USAGE, f"Cannot read {a.password_file}: {e.strerror}") from e value = text.splitlines()[0] if text else "" if value is None: value = os.environ.get("MESHBAY_PASSWORD") if value is None: if not sys.stdin.isatty(): raise NoPassphrase("A passphrase is needed and there is no terminal to ask " "on; pass --password-file or set MESHBAY_PASSWORD.") value = getpass.getpass(f"Passphrase for {self.username or self.stored_username}" f" on {self.hub_url}: ") if not value: raise NoPassphrase("No passphrase given.") self._passphrase = value return value def account(self) -> Account: account = Account.open(self.hub, self.username, self.passphrase, self.store) account.on_device_code = self._device_code return account def _device_code(self, group: dict, pending, *, first: bool) -> bool: """Show the code once. Wait only where somebody is there to approve it.""" waiting = sys.stderr.isatty() and not self.args.quiet if first: minutes = max(1, int((pending.expires_at - time.time()) // 60)) print(f"This machine is not yet a device of yours on the node hosting " f"{safe(handle(group))}.\n" "Approve it from the MeshBay application or a browser where you use " "this group:\n" " the group's Settings → \"Your devices on this node\" → enter " f"{pending.link_code} → Approve\n" "or from a machine already linked:\n" f" meshbay-cli devices approve {handle(group)} {pending.link_code}\n" + (f"Waiting for the approval (the code expires in {minutes} min; " "Ctrl-C to stop)..." if waiting else f"Run the command again once it is approved (the code expires in " f"{minutes} min)."), file=sys.stderr, flush=True) return waiting async def aclose(self) -> None: await self.hub.aclose()