cli (info): Add tor sub-command:

`tor` sub-command shows info on currently running Tor daemon (if any), and on the circuit
used to connect to XMPP server with current profile (if relevant).

rel 473

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 <http://www.gnu.org/licenses/>.
 
-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__(