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
85
86
87
88
|
"""Opening a group: which of the nodes serving it to dial, and in which order."""
import asyncio
import logging
import time
from ..errors import NODE, CliError
from .account import Account
from .paths import handle
from .session import DeviceLinkPending, GroupSession, NodeKeyChanged, NodeRefusal, connect
# How often a machine waiting for its approval tries the node again. Each try is
# one offer: well inside what the hub admits per account and per node.
LINK_POLL = 10
# Join refusals about the account rather than the node: every node gives them.
ACCOUNT_REFUSALS = frozenset({"code_required", "code_invalid"})
log = logging.getLogger(__name__)
async def my_groups(account: Account) -> list[dict]:
return await account.call(account.hub.my_groups)
async def open_group(account: Account, group: dict, *, join_code: str | None = None,
node_pk: str | None = None) -> GroupSession:
"""
A session with a node serving `group`: the pinned one first, then the others
in the hub's order. A node whose key changed stops everything — that is an
alarm, not a reason to quietly try its neighbour.
`join_code` goes in the join; `node_pk` restricts the tries to that node — an
invitation link's code is for the node the link names and no other.
"""
nodes = await account.call(account.hub.group_nodes, group["id"])
if node_pk:
nodes = [n for n in nodes if n.get("pk_node") == node_pk]
if not nodes:
raise CliError(NODE, f"The node the invitation names for {handle(group)} "
"is not online.")
if not nodes:
raise CliError(NODE, f"No node serving {handle(group)} is online.")
nodes = sorted(nodes, key=lambda n: account.pin(n["node_id"]) is None)
failures = []
for node in nodes:
try:
return await _connect_linking(account, group, node, join_code)
except (NodeKeyChanged, DeviceLinkPending):
raise
except NodeRefusal as e:
# A refusal about this account — a code to give — is the same answer
# from every node; trying the next would only bury it.
if e.refusal_code in ACCOUNT_REFUSALS or len(nodes) == 1:
raise
log.info("node %s: %s", node["node_id"], e.message)
failures.append(f"{node['node_id']}: {e.message}")
except CliError as e:
if e.code != NODE or len(nodes) == 1:
raise
log.info("node %s: %s", node["node_id"], e.message)
failures.append(f"{node['node_id']}: {e.message}")
raise CliError(NODE, f"No node serving {handle(group)} could be reached:\n "
+ "\n ".join(failures))
async def _connect_linking(account: Account, group: dict, node: dict,
join_code: str | None = None) -> GroupSession:
"""
`connect`, and — when this machine is waiting to be approved as a device on
that node — the wait: the code is shown once (`account.on_device_code`), and
the node is tried again until the approval lands or the request expires. The
code names this node, so no other node is tried meanwhile.
"""
shown = False
while True:
try:
return await connect(account, group["id"], node, join_code)
except DeviceLinkPending as pending:
hook = account.on_device_code
wait = hook(group, pending, first=not shown) if hook else False
shown = True
if not wait:
raise
if pending.expires_at <= time.time():
raise CliError(NODE, "The device request expired before it was approved; "
"run the command again for a new code.") from None
await asyncio.sleep(LINK_POLL)
|