diff --git a/libervia/cli/cmd_info.py b/libervia/cli/cmd_info.py index dc08eaea..3f2836ac 100644 --- a/libervia/cli/cmd_info.py +++ b/libervia/cli/cmd_info.py @@ -17,11 +17,17 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from pprint import pformat +from rich.align import Align +from rich.console import Group +from rich.panel import Panel +from rich.table import Table, Column +from rich.text import Text from libervia.backend.core.i18n import _ +from libervia.backend.models.tor import TorData, TorMode from libervia.backend.tools.common import data_format, date_utils from libervia.backend.tools.common.ansi import ANSI as A +from libervia.backend.tools.common.utils import get_human_size from libervia.cli import common from libervia.cli.constants import Const as C @@ -388,8 +394,97 @@ class Devices(base.CommandBase): self.host.quit() +class Tor(base.CommandBase): + + def __init__(self, host): + extra_outputs = {"default": self.default_output} + super().__init__( + host, + "tor", + use_output=C.OUTPUT_MODEL, + extra_outputs=extra_outputs, + help=_("Tor data"), + ) + + def add_parser_options(self): + pass + + def default_output(self, tor_data: TorData) -> None: + """Pretty print the TorData instance.""" + general_instruction = Text( + "General data on Tor daemon.", style="instructions", justify="center" + ) + general_table = Table.grid(Column(style="header"), Column(), padding=(0, 1)) + match tor_data.mode: + case TorMode.DISABLED: + mode_desc = _("Tor is disabled") + case TorMode.ENABLED_NO_WEBRTC: + mode_desc = _( + "Tor is enabled for all connections, WebRTC is disabled to avoid" + " leaking IPs" + ) + case _: + # This should never be reached. + raise exceptions.InternalError + general_table.add_row("mode", f"{tor_data.mode} ({mode_desc})") + general_table.add_row("version", tor_data.version) + general_table.add_row("uptime", date_utils.delta2human(0, tor_data.uptime)) + general_table.add_row("bytes read", get_human_size(tor_data.traffic.read)) + general_table.add_row("bytes written", get_human_size(tor_data.traffic.written)) + + self.console.print( + Panel( + Group(general_instruction, general_table), + title="Tor Daemon", + border_style="main_panel", + ) + ) + + circuit = tor_data.circuit + if circuit is not None: + circuit_instruction = Text( + "Information on the path used by your profile XMPP connection", + style="instructions", + justify="center", + ) + circuit_table = Table("id", "state", "purpose", box=None) + circuit_table.add_row(str(circuit.id), circuit.state, circuit.purpose) + relays_panels = [] + for idx, relay in enumerate(circuit.path): + relay_table = Table.grid(Column(style="header"), Column(), padding=(0, 1)) + relay_table.add_row("id", relay.id) + relay_table.add_row("name", relay.name) + relay_table.add_row("ip", relay.ip) + relay_table.add_row("country", relay.country) + relay_table.add_row("flags", ", ".join(relay.flags)) + relay_table.add_row("Details URL", relay.details_url) + + relays_panels.append(Panel(relay_table, title=str(idx))) + circuit_panel = Panel( + Group( + circuit_instruction, + Align(circuit_table, align="center"), + *relays_panels, + ), + title="Circuit", + border_style="main_panel", + ) + self.console.print(circuit_panel) + + async def start(self): + try: + data_s = await self.host.bridge.tor_infos_get("", self.host.profile) + except Exception as e: + self.disp(_("Error getting Tor infos: {e}").format(e=e), error=True) + self.host.quit(C.EXIT_BRIDGE_ERRBACK) + else: + tor_data = TorData.model_validate_json(data_s) + await self.output(tor_data) + self.host.quit() + + class Info(base.CommandBase): - subcommands = (Disco, Version, Session, Devices) + subcommands = (Disco, Version, Session, Devices, Tor) def __init__(self, host): super(Info, self).__init__(