1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
|
"""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()
|