libervia-backend/libervia/backend/models/tor.py

#!/usr/bin/env python3

# Libervia Tor models
# Copyright (C) 2009-2025 Jérôme Poisson (goffi@goffi.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# 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 enum import StrEnum, auto
from typing import TYPE_CHECKING, Self

from pydantic import BaseModel, Field

from libervia.backend.core import exceptions
from libervia.backend.core.log import getLogger

log = getLogger(__name__)

if TYPE_CHECKING:
    from libervia.backend.core.core_types import SatXMPPClient, SatXMPPEntity


class TorMode(StrEnum):
    #: Tor is completely disabled.
    DISABLED = auto()
    #: Tor is enable for all connections, but WebRTC is disabled (to avoid leaking IP).
    ENABLED_NO_WEBRTC = auto()


class Relay(BaseModel):
    id: str = Field(description="Hex-encoded router identity fingerprint")
    name: str = Field(description="Router nickname or name")
    ip: str = Field(description="IP address")
    flags: list = Field(description='Router flags (e.g., "Guard", "Fast")')
    bandwidth: int | None = Field(
        default=None, description="Reported bandwidth in bytes/sec"
    )
    country: str = Field(description="Country code (ISO 3166-1 alpha-2)")
    details_url: str = Field(
        default_factory=lambda data: f"https://metrics.torproject.org/rs.html#details/{data['id']}",
        description="URL of metrics website with more details on the relay.",
    )


class Circuit(BaseModel):
    id: int = Field(description="Unique identifier for the circuit")
    state: str = Field(
        description="Current state of the circuit (e.g., 'BUILT', 'FAILED')"
    )
    purpose: str = Field(
        description="Purpose of the circuit (e.g., 'GENERAL', 'HS_CLIENT_INTRO')"
    )
    path: list[Relay] = Field(description="List of relays in the circuit path")


class Traffic(BaseModel):
    read: int = Field(description="Bytes read (downloaded) since Tor started")
    written: int = Field(description="Bytes written (uploaded) since Tor started")


class TorData(BaseModel):
    mode: TorMode = Field(description="In which mode Tor active")
    version: str = Field(description="Tor version")
    uptime: int = Field(description="Uptime of the Tor daemon in seconds")
    traffic: Traffic = Field(description="Global traffic statistics")
    circuit: Circuit | None = Field(
        default=None, description="Circuit used for XMPP connection"
    )

    @classmethod
    async def from_client(cls, client: "SatXMPPClient") -> Self:
        """Built an instance from ``client.tor``.

        @return: Instance of TorData
        """
        if client.tor is None:
            raise exceptions.FeatureNotFound("Tor is not used for this profile.")
        host = client.jid.host

        # Initialize with basic version info
        infos = {
            "mode": client.tor_mode,
            "version": client.tor.version,
            "uptime": None,
            "traffic": None,
            "circuit": None,
        }

        state = await client.tor.create_state()

        for stream in state.streams.values():
            if (
                stream.target_host == host
                and stream.target_port == client.port
                and stream.circuit is not None
                and stream.state == "SUCCEEDED"
            ):
                circuit = stream.circuit
                # We have the active stream.
                router_infos = []
                for router in circuit.path:
                    try:
                        location = await router.get_location()
                        country = location.countrycode
                    except Exception:
                        log.exception("Can't get location.")
                        country = "UNKNOWN"

                    router_info = Relay(
                        id=router.id_hex.lstrip("$"),
                        name=router.name,
                        ip=router.ip,
                        flags=router.flags,
                        bandwidth=getattr(router, "bandwidth", None),
                        country=country,
                    )
                    router_infos.append(router_info)

                circuit_info = Circuit(
                    id=circuit.id,
                    state=circuit.state,
                    purpose=circuit.purpose,
                    path=router_infos,
                )
                infos["circuit"] = circuit_info

        try:
            infos_dict = await client.tor.protocol.get_info(
                "uptime", "traffic/read", "traffic/written"
            )
        except Exception:
            log.exception("Can't get Tor infos.")
            raise exceptions.DataError("Tor infos not available.")
        else:
            if "uptime" in infos_dict:
                infos["uptime"] = infos_dict["uptime"]
            if "traffic/read" in infos_dict and "traffic/written" in infos_dict:
                infos["traffic"] = Traffic(
                    read=infos_dict["traffic/read"].strip(),
                    written=infos_dict["traffic/written"].strip(),
                )

        return cls(**infos)