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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
|
"""
meshbay-cli [global options] <action> [action options] [arguments]
With no argument at all, the interactive mode opens.
"""
import argparse
import asyncio
import logging
import os
import signal
import sys
import threading
from . import __version__
from .commands import (
account,
browser_access,
chat,
completion,
decline,
devices,
get,
groups,
help,
invitations,
join,
leave,
login,
logout,
ls,
mkdir,
put,
rm,
search,
stream,
trust,
)
from .commands.context import Context
from .errors import INTERRUPTED, USAGE, CliError
ACTIONS = {"login": login, "logout": logout, "account": account, "groups": groups,
"invitations": invitations, "join": join, "decline": decline, "leave": leave,
"ls": ls, "search": search, "get": get,
"stream": stream, "put": put, "mkdir": mkdir, "rm": rm, "chat": chat, "devices": devices,
"trust": trust, "browser-access": browser_access,
"completion": completion, "help": help}
class _Parser(argparse.ArgumentParser):
"""argparse exits 2 on a usage error, which here means a refused sign-in."""
def error(self, message: str):
self.print_usage(sys.stderr)
self.exit(USAGE, f"{self.prog}: error: {message}\n")
def _global_options(p: argparse.ArgumentParser, default) -> None:
p.add_argument("--hub", default=default, help="the hub (default: the one signed in to, "
"else https://meshbay.org); also MESHBAY_HUB")
p.add_argument("-u", "--user", default=default, help="account name; also MESHBAY_USER")
p.add_argument("-p", "--password", default=default,
help="the passphrase; visible to other users in the process list and "
"kept in shell history — --password-file - or MESHBAY_PASSWORD "
"are not")
p.add_argument("--password-file", metavar="FILE", default=default,
help="read the passphrase from FILE's first line (- for stdin)")
p.add_argument("--json", action="store_true", default=default or False,
help="machine-readable output")
p.add_argument("-q", "--quiet", action="store_true", default=default or False,
help="errors only")
p.add_argument("-v", "--verbose", action="count", default=default or 0,
help="-v: connection steps, -vv: protocol messages")
def build_parser() -> argparse.ArgumentParser:
"""Global options are accepted before the action and after it."""
p = _Parser(
prog="meshbay-cli",
description="A member's MeshBay groups and files, from a terminal. Run with no "
"argument for the interactive mode.")
_global_options(p, None)
p.add_argument("--version", action="version", version=f"meshbay-cli {__version__}")
# SUPPRESS: an option the action's own parser did not see must not
# overwrite the value given before the action.
late = _Parser(add_help=False)
_global_options(late, argparse.SUPPRESS)
sub = p.add_subparsers(dest="action", metavar="<action>")
for module in ACTIONS.values():
module.add_parser(sub, late)
return p
def _logging(verbose: int, quiet: bool) -> None:
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter("meshbay-cli: %(levelname)s: %(message)s"))
root = logging.getLogger()
root.handlers[:] = [handler]
root.setLevel(logging.ERROR if quiet else logging.WARNING)
level = {0: logging.WARNING, 1: logging.INFO}.get(verbose, logging.DEBUG)
logging.getLogger("meshbay_cli").setLevel(logging.ERROR if quiet else level)
# aiortc and aioice are chatty at INFO; their story is told from -vv on.
for name in ("aiortc", "aioice", "httpx", "httpcore"):
logging.getLogger(name).setLevel(logging.DEBUG if verbose >= 3 else logging.WARNING)
async def _run(args: argparse.Namespace) -> int:
ctx = Context(args)
try:
if args.action is None:
from .tui.app import run_interactive
return await run_interactive(ctx)
return await ACTIONS[args.action].run(ctx)
finally:
await ctx.aclose()
class _WindowsSelectorLoop(asyncio.SelectorEventLoop):
"""
A selector loop that Ctrl+C wakes. Windows' select() is not interrupted by
a console event, and unlike the proactor (`windows_events.ProactorEventLoop.
run_forever`) this loop never points the signal wakeup at its self-pipe: Ctrl+C
waited for the next timer, which in `chat --follow` or `stream` can be half
a minute away (measured: 29 s on a loop whose next timer was 30 s off).
"""
def run_forever(self):
if threading.current_thread() is not threading.main_thread():
return super().run_forever() # signals reach the main thread only
old = signal.set_wakeup_fd(self._csock.fileno())
try:
super().run_forever()
finally:
signal.set_wakeup_fd(old)
def loop_factory():
"""aiortc's ICE needs a selector loop: Windows' default proactor never
completes a DataChannel handshake."""
return _WindowsSelectorLoop if sys.platform == "win32" else None
def main(argv: list[str] | None = None) -> int:
argv = sys.argv[1:] if argv is None else argv
if argv[:1] == ["__complete"]:
_utf8_output() # what the completion scripts read
from .complete import main as complete
return complete(argv[1:])
if not argv and os.environ.get("MESHBAY_CLI_SELFTEST") == "1":
from .selftest import run
return run()
parser = build_parser()
interactive = not argv
argv = chat.rewrite_count(argv, ACTIONS)
try:
args = parser.parse_args(argv)
except SystemExit as e: # --help, --version, or a usage error
return int(e.code or 0)
if args.action is None and not interactive:
parser.print_usage(sys.stderr)
return USAGE
_logging(args.verbose, args.quiet)
_utf8_output()
try:
return asyncio.run(_run(args), loop_factory=loop_factory())
except CliError as e:
print(f"meshbay-cli: {e.message}", file=sys.stderr)
return e.code
except KeyboardInterrupt:
print("meshbay-cli: interrupted", file=sys.stderr)
return INTERRUPTED
def _utf8_output() -> None:
"""Names are printed as they are; a Windows console's code page cannot hold most."""
for out in (sys.stdout, sys.stderr):
reconfigure = getattr(out, "reconfigure", None)
if reconfigure is not None:
try:
reconfigure(encoding="utf-8", errors="replace")
except (ValueError, OSError):
pass
if __name__ == "__main__":
sys.exit(main())
|