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
89
90
91
92
93
94
95
96
|
"""
`devices`: this account's devices on the node serving a group — what the
application's "Your devices on this node" shows and does.
Identities are per node, so a group names the node. `approve` admits the device
showing a code; `revoke` retires one, asked first (`--yes` for a script). A node
where this machine is the only linked device admits no other without this.
"""
import json
import sys
from ..client import devices
from ..client.groups import my_groups, open_group
from ..client.paths import group_arg, handle, resolve_group
from ..errors import REFUSED, USAGE, CliError
from ..output import confirm, err, safe, sentence
from .context import Context
HELP = "your devices on the node serving a group: list, approve a new one, revoke one"
def add_parser(sub, parent) -> None:
p = sub.add_parser("devices", parents=[parent], help=HELP, description=sentence(HELP)
+ ". A new device shows a code; approve it here, from a machine "
"already linked to that node.")
p.add_argument("verb", choices=("list", "approve", "revoke"))
p.add_argument("group", metavar="GROUP", help="the group, as `groups` lists it")
p.add_argument("arg", nargs="?", metavar="CODE|DEVICE",
help="approve: the code the new device shows; revoke: the device, by "
"the first characters of its key as `devices list` shows it")
p.add_argument("-y", "--yes", action="store_true",
help="revoke without asking for confirmation")
def _when(iso: str) -> str:
return (iso or "")[:16].replace("T", " ")
async def run(ctx: Context) -> int:
a = ctx.args
if a.verb != "list" and not a.arg:
raise CliError(USAGE, f"Name the {'code' if a.verb == 'approve' else 'device'}: "
f"meshbay-cli devices {a.verb} GROUP "
f"{'CODE' if a.verb == 'approve' else 'DEVICE'}")
await ctx.hub.check_version()
account = ctx.account()
session = None
try:
group = resolve_group(await my_groups(account), group_arg(a.group))
name = handle(group)
session = await open_group(account, group)
if a.verb == "list":
return _print(ctx, name, await devices.list_devices(session))
if a.verb == "approve":
pk = await devices.approve(session, account, a.arg)
print(json.dumps({"approved": pk}) if a.json else
f"Approved the device {pk[:16]} on the node serving {safe(name)}.")
return 0
device = devices.find_device((await devices.list_devices(session))["devices"],
a.arg)
if device.get("is_this_one"):
raise CliError(REFUSED, "That is this machine. Revoke it from another of your "
"devices.")
label = device.get("label") or "no label"
confirm(f"Revoke the device {device['pk_ed25519'][:16]} ({safe(label)}) on the "
f"node serving {safe(name)}? It will no longer open this node's groups.",
a.yes)
await devices.revoke(session, account, device["pk_ed25519"])
print(json.dumps({"revoked": device["pk_ed25519"]}) if a.json else
f"Revoked the device {device['pk_ed25519'][:16]}.")
return 0
finally:
if session is not None:
await session.close()
await account.close()
def _print(ctx: Context, name: str, listed: dict) -> int:
rows = listed["devices"]
if ctx.args.json:
json.dump(listed, sys.stdout, indent=1)
print()
return 0
print(f" {'KEY':<18}{'ADDED':<18}{'VIA':<12}LABEL")
for d in rows:
mark = "*" if d.get("is_this_one") else " "
print(f"{mark} {d['pk_ed25519'][:16]:<18}{_when(d.get('pinned_at')):<18}"
f"{safe(d.get('pinned_via') or ''):<12}{safe(d.get('label') or '')}")
if not ctx.args.quiet:
err("* this machine")
if listed["pending"]:
n = listed["pending"]
err(f"{n} device{'s' if n != 1 else ''} waiting for approval: "
f"meshbay-cli devices approve {name} CODE")
return 0
|