Skip to content

Reference: Python API Reference

Automated API reference generated directly from Google-style Python docstrings in the codebase via mkdocstrings.

Firewall Package (ttp.firewall)

ttp.firewall.builder

Stateless Firewall Module - Pure ruleset string generator.

ttp.firewall.runner

Stateless Firewall Module - Low-level nftables execution engine.

apply_rules(tor_user, transport_port=9041, dns_port=9054, allow_root=False, lan_bypass=True, bypass_uids=None, bypass_gids=None, disable_ipv6=False)

Create the dedicated 'inet ttp' table and inject redirection rules.

Create Table -> Flush Table -> Apply Ruleset is submitted to nft as a single transaction, so the table is never observable in a half-applied state. If any step fails, triggers an automatic rollback (table destruction).

Parameters:

Name Type Description Default
tor_user str

Username or numeric UID string of the Tor daemon process.

 required
transport_port int

Local TCP port for Tor TransPort redirection.

 9041
dns_port int

Local UDP/TCP port for Tor DNSPort redirection.

 9054
allow_root bool

If True, allows processes running as root (UID 0) to bypass rules.

 False
lan_bypass bool

If True, excludes local LAN subnets from redirection.

 True
bypass_uids list[int] | None

Optional list of numeric UIDs exempted from redirection.

 None
bypass_gids list[int] | None

Optional list of numeric GIDs exempted from redirection.

 None
disable_ipv6 bool

If True, forces dropping all IPv6 traffic regardless of host availability.

 False

Raises:

Type Description
FirewallError

If the tor_user is invalid or rule injection fails.
Source code in ttp/firewall/runner.py 
 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
def apply_rules(
    tor_user: str,
    transport_port: int = 9041,
    dns_port: int = 9054,
    allow_root: bool = False,
    lan_bypass: bool = True,
    bypass_uids: list[int] | None = None,
    bypass_gids: list[int] | None = None,
    disable_ipv6: bool = False,
) -> None:
    """Create the dedicated 'inet ttp' table and inject redirection rules.

    Create Table -> Flush Table -> Apply Ruleset is submitted to nft as a single
    transaction, so the table is never observable in a half-applied state.
    If any step fails, triggers an automatic rollback (table destruction).

    Args:
        tor_user: Username or numeric UID string of the Tor daemon process.
        transport_port: Local TCP port for Tor TransPort redirection.
        dns_port: Local UDP/TCP port for Tor DNSPort redirection.
        allow_root: If True, allows processes running as root (UID 0) to bypass rules.
        lan_bypass: If True, excludes local LAN subnets from redirection.
        bypass_uids: Optional list of numeric UIDs exempted from redirection.
        bypass_gids: Optional list of numeric GIDs exempted from redirection.
        disable_ipv6: If True, forces dropping all IPv6 traffic regardless of host availability.

    Raises:
        FirewallError: If the tor_user is invalid or rule injection fails.
    """
    # Resolve numeric UID for the tor user to avoid nft resolution issues
    try:
        tor_uid = int(tor_user) if tor_user.isdigit() else pwd.getpwnam(tor_user).pw_uid
    except KeyError as e:
        raise FirewallError(f"Tor user '{tor_user}' not found on system.") from e

    from ttp.tor_detect import is_ipv6_supported

    ipv6_avail = is_ipv6_supported() and not disable_ipv6

    # Resolve systemd-resolved UID once, before building the ruleset
    resolved_uid: int | None = None
    for _user in ("systemd-resolve", "systemd-resolved"):
        try:
            resolved_uid = pwd.getpwnam(_user).pw_uid
            break
        except KeyError:
            continue

    ruleset = _build_ruleset(
        tor_uid=tor_uid,
        transport_port=transport_port,
        dns_port=dns_port,
        ipv6_avail=ipv6_avail,
        allow_root=allow_root,
        lan_bypass=lan_bypass,
        bypass_uids=bypass_uids,
        bypass_gids=bypass_gids,
        resolved_uid=resolved_uid,
        cgroup_bypass=_has_cgroup_bypass_support(),
    )

    try:
        _apply_table_atomically(ruleset)
        logger.info(f"Stateless rules applied. Tor user ({tor_user}, UID {tor_uid}) is exempt.")
    except Exception as e:
        logger.error(f"Firewall injection failed: {e}. Rolling back...")
        destroy_rules()
        if not isinstance(e, FirewallError):
            raise FirewallError(f"Failed to apply stateless rules: {e}") from e
        raise

destroy_rules()

Destroy the 'ttp' table and clean up firewall rules.

This is the atomic cleanup operation. It flushes the table, destroys it, and verifies that the table is no longer present in kernel state.

Returns:

Name Type Description
bool bool

True if the table was successfully destroyed or already gone.

Raises:

Type Description
FirewallError

If table destruction fails and the table remains active.
Source code in ttp/firewall/runner.py 
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def destroy_rules() -> bool:
    """Destroy the 'ttp' table and clean up firewall rules.

    This is the atomic cleanup operation. It flushes the table, destroys it,
    and verifies that the table is no longer present in kernel state.

    Returns:
        bool: True if the table was successfully destroyed or already gone.

    Raises:
        FirewallError: If table destruction fails and the table remains active.
    """
    # Flush the table first for absolute cleanup safety
    subprocess.run(
        [resolve("nft"), "flush", "table", "inet", "ttp"],
        capture_output=True,
        check=False,
        timeout=10,
    )
    result = subprocess.run(
        [resolve("nft"), "destroy", "table", "inet", "ttp"],
        capture_output=True,
        check=False,
        timeout=10,
    )
    # returncode 1 with table absent = already clean, not an error
    # to distinguish it, check if the table exists
    if result.returncode != 0:
        # Check: does the table still exist?
        check = subprocess.run(
            [resolve("nft"), "list", "table", "inet", "ttp"],
            capture_output=True,
            check=False,
            timeout=10,
        )
        if check.returncode != 0:
            # The table is gone - destroy "failed" because it was already clean
            return True
        # The table still exists - destroy actually failed
        err_msg = result.stderr.decode().strip() if result.stderr else "unknown error"
        logger.error(f"nft destroy failed: {err_msg}")
        raise FirewallError(f"Failed to destroy nftables ruleset: {err_msg}")

    RULES_TEMP_PATH.unlink(missing_ok=True)
    return True

ttp.firewall.emergency

Stateless Firewall Module - Emergency lockdown, killswitch, and socket slaughter mechanisms.

apply_active_socket_slaughter()

Inject temporary reject rules at the top of the filter_out chain.

Actively terminates pending local connections by sending immediate ICMP Port Unreachable for UDP sockets and TCP RST packets for open TCP streams.

Source code in ttp/firewall/emergency.py 
 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
def apply_active_socket_slaughter() -> None:
    """Inject temporary reject rules at the top of the filter_out chain.

    Actively terminates pending local connections by sending immediate ICMP Port Unreachable
    for UDP sockets and TCP RST packets for open TCP streams.
    """
    try:
        # 1. Kill pending UDP connections (sends ICMP Port Unreachable to the local process)
        _run_nft(
            [
                "insert",
                "rule",
                "inet",
                "ttp",
                "filter_out",
                "meta",
                "l4proto",
                "udp",
                "counter",
                "reject",
            ]
        )
        # 2. Kill pending TCP connections instantly (sends RST to the local process)
        _run_nft(
            [
                "insert",
                "rule",
                "inet",
                "ttp",
                "filter_out",
                "meta",
                "l4proto",
                "tcp",
                "counter",
                "reject",
                "with",
                "tcp",
                "reset",
            ]
        )
        logger.warning("Active socket slaughter rules applied: resetting pending connections.")
    except Exception as e:
        _log_teardown_failure("Active socket slaughter", e)

apply_emergency_killswitch()

Apply an emergency network killswitch.

Replaces the 'inet ttp' table with an ultra-restrictive ruleset that drops all inbound, outbound, and forwarded network traffic on physical interfaces, permitting only local loopback communication.

Raises:

Type Description
FirewallError

If table creation or killswitch ruleset injection fails.
Source code in ttp/firewall/emergency.py 
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
def apply_emergency_killswitch() -> None:
    """Apply an emergency network killswitch.

    Replaces the 'inet ttp' table with an ultra-restrictive ruleset that drops
    all inbound, outbound, and forwarded network traffic on physical interfaces,
    permitting only local loopback communication.

    Raises:
        FirewallError: If table creation or killswitch ruleset injection fails.
    """
    ruleset = """
    table inet ttp {
        chain filter_out {
            type filter hook output priority filter; policy drop;
            oifname "lo" accept
        }
        chain filter_forward {
            type filter hook forward priority filter; policy drop;
        }
        chain filter_input {
            type filter hook input priority filter; policy drop;
            iifname "lo" accept
        }
    }
    """
    try:
        # Table reset and drop-all ruleset go in as one transaction. Flushing in a
        # separate nft call would briefly leave the table empty, which is an open
        # network at the exact moment integrity has already been lost.
        _apply_table_atomically(ruleset)
        logger.warning("Emergency killswitch applied: network traffic isolated.")
    except Exception as e:
        logger.error(f"Failed to apply emergency killswitch: {e}")
        if not isinstance(e, FirewallError):
            raise FirewallError(f"Failed to apply emergency killswitch: {e}") from e
        raise

apply_teardown_lockdown(tor_uid=None)

Insert a lockdown drop rule at the top of the filter_out chain in table inet ttp.

Ensures all non-loopback outbound traffic is dropped during graceful session teardown, while permitting the Tor daemon UID to close control connections cleanly.

Parameters:

Name Type Description Default
tor_uid int | None

Optional numeric UID of the Tor daemon process to exempt from lockdown.

 None
Source code in ttp/firewall/emergency.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def apply_teardown_lockdown(tor_uid: int | None = None) -> None:
    """Insert a lockdown drop rule at the top of the filter_out chain in table inet ttp.

    Ensures all non-loopback outbound traffic is dropped during graceful session teardown,
    while permitting the Tor daemon UID to close control connections cleanly.

    Args:
        tor_uid: Optional numeric UID of the Tor daemon process to exempt from lockdown.
    """
    rule = ["insert", "rule", "inet", "ttp", "filter_out"]
    if tor_uid is not None:
        rule += ["meta", "skuid", "!=", str(tor_uid)]
    rule += ["oifname", "!=", "lo", "drop"]

    try:
        _run_nft(rule)
        logger.warning("Teardown lockdown applied: outbound traffic locked.")
    except Exception as e:
        _log_teardown_failure("Teardown lockdown", e)

Tor Configuration & Lifecycle (ttp.tor_*)

ttp.tor_config

Tor configuration and torrc generation module.

generate_torrc(tor_user, transport_port=9041, dns_port=9054, block_doh=True, use_bridges=False, bridges=None, disable_ipv6=False)

Generate a volatile torrc in /run/tor/ttp/torrc.

The DataDirectory points to the persistent cache (/var/lib/tor/ttp) so that Entry Guards are preserved across runs for fast bootstrapping.

Parameters:

Name Type Description Default
tor_user str

Username running the Tor process.

 required
transport_port int

Local TCP port for Tor TransPort redirection.

 9041
dns_port int

Local UDP/TCP port for Tor DNSPort redirection.

 9054
block_doh bool

If True, maps well-known DoH resolver domains to 0.0.0.0.

 True
use_bridges bool

If True, configures Tor to route via Pluggable Transports.

 False
bridges Optional[list[str]]

Optional list of bridge lines (e.g. ["obfs4 ...", "snowflake ..."]).

 None
disable_ipv6 bool

If True, forces IPv6 client routing off.

 False

Returns:

Name Type Description
Path Path

Absolute path to the generated torrc file.
Source code in ttp/tor_config.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def generate_torrc(
    tor_user: str,
    transport_port: int = 9041,
    dns_port: int = 9054,
    block_doh: bool = True,
    use_bridges: bool = False,
    bridges: Optional[list[str]] = None,
    disable_ipv6: bool = False,
) -> Path:
    """Generate a volatile ``torrc`` in ``/run/tor/ttp/torrc``.

    The ``DataDirectory`` points to the persistent cache (``/var/lib/tor/ttp``)
    so that Entry Guards are preserved across runs for fast bootstrapping.

    Args:
        tor_user: Username running the Tor process.
        transport_port: Local TCP port for Tor TransPort redirection.
        dns_port: Local UDP/TCP port for Tor DNSPort redirection.
        block_doh: If True, maps well-known DoH resolver domains to 0.0.0.0.
        use_bridges: If True, configures Tor to route via Pluggable Transports.
        bridges: Optional list of bridge lines (e.g. ``["obfs4 ...", "snowflake ..."]``).
        disable_ipv6: If True, forces IPv6 client routing off.

    Returns:
        Path: Absolute path to the generated ``torrc`` file.
    """
    TOR_RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
    try:
        shutil.chown(TOR_RUNTIME_DIR, user=tor_user, group=tor_user)
    except (KeyError, ValueError, OSError):
        pass
    os.chmod(TOR_RUNTIME_DIR, 0o700)

    # Ensure parent directory of cache is owned by the Tor user and has correct permissions
    parent_dir = TOR_CACHE_DIR.parent
    if parent_dir.exists():
        try:
            shutil.chown(parent_dir, user=tor_user, group=tor_user)
            os.chmod(parent_dir, 0o700)
        except (KeyError, ValueError, OSError):
            pass

    # Persistent cache directory must be fixed on every run
    data_dir = str(TOR_CACHE_DIR)
    os.makedirs(data_dir, exist_ok=True)
    try:
        shutil.chown(data_dir, user=tor_user)
    except (KeyError, ValueError, OSError):
        pass
    os.chmod(data_dir, 0o700)

    from ttp.tor_detect import is_ipv6_supported

    ipv6_avail = is_ipv6_supported() and not disable_ipv6

    torrc_content = _build_torrc_content(
        tor_user=tor_user,
        transport_port=transport_port,
        dns_port=dns_port,
        block_doh=block_doh,
        use_bridges=use_bridges,
        bridges=bridges,
        ipv6_avail=ipv6_avail,
    )

    torrc_path = TOR_RUNTIME_DIR / "torrc"
    _write_private(torrc_path, torrc_content)
    logger.info("Generated runtime torrc at %s", torrc_path)
    return torrc_path

ttp.tor_service

Tor systemd service lifecycle management module.

start_tor_service(tor_user, transport_port=9041, dns_port=9054, block_doh=True, use_bridges=False, bridges=None, disable_ipv6=False)

Generate the runtime torrc and start a dedicated TTP Tor systemd service.

Sequence
  1. Generate volatile torrc in /run/tor/ttp/torrc.
  2. Label SELinux ports if SELinux is enforcing.
  3. Write a volatile ttp-tor.service unit to /run/systemd/system/.
  4. Reload systemd daemon and start the service.

Parameters:

Name Type Description Default
tor_user str

System user designated to run Tor.

 required
transport_port int

Local TCP port for Tor TransPort redirection.

 9041
dns_port int

Local UDP/TCP port for Tor DNSPort redirection.

 9054
block_doh bool

If True, maps canary DoH domains to 0.0.0.0.

 True
use_bridges bool

If True, configures Tor to route via Pluggable Transports.

 False
bridges Optional[list[str]]

Optional list of bridge configuration strings.

 None
disable_ipv6 bool

If True, forces IPv6 client routing off.

 False

Raises:

Type Description
TorError

If systemd daemon reload or service restart fails.
Source code in ttp/tor_service.py 
 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
def start_tor_service(
    tor_user: str,
    transport_port: int = 9041,
    dns_port: int = 9054,
    block_doh: bool = True,
    use_bridges: bool = False,
    bridges: Optional[list[str]] = None,
    disable_ipv6: bool = False,
) -> None:
    """Generate the runtime torrc and start a dedicated TTP Tor systemd service.

    Sequence:
        1. Generate volatile torrc in ``/run/tor/ttp/torrc``.
        2. Label SELinux ports if SELinux is enforcing.
        3. Write a volatile ``ttp-tor.service`` unit to ``/run/systemd/system/``.
        4. Reload systemd daemon and start the service.

    Args:
        tor_user: System user designated to run Tor.
        transport_port: Local TCP port for Tor TransPort redirection.
        dns_port: Local UDP/TCP port for Tor DNSPort redirection.
        block_doh: If True, maps canary DoH domains to 0.0.0.0.
        use_bridges: If True, configures Tor to route via Pluggable Transports.
        bridges: Optional list of bridge configuration strings.
        disable_ipv6: If True, forces IPv6 client routing off.

    Raises:
        TorError: If systemd daemon reload or service restart fails.
    """
    generate_torrc(
        tor_user,
        transport_port=transport_port,
        dns_port=dns_port,
        block_doh=block_doh,
        use_bridges=use_bridges,
        bridges=bridges,
        disable_ipv6=disable_ipv6,
    )
    label_ports_selinux(transport_port, dns_port)
    _write_service_unit(tor_user)

    try:
        subprocess.run(
            [resolve("systemctl"), "daemon-reload"],
            capture_output=True,
            text=True,
            check=True,
        )
        subprocess.run(
            [resolve("systemctl"), "restart", TTP_SERVICE_NAME],
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        raise TorError(f"Failed to start '{TTP_SERVICE_NAME}': {e.stderr.strip()}") from e
    logger.info("TTP Tor service started with dedicated config.")

stop_tor_service()

Stop the dedicated TTP Tor service and remove the volatile systemd unit.

Source code in ttp/tor_service.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def stop_tor_service() -> None:
    """Stop the dedicated TTP Tor service and remove the volatile systemd unit."""
    subprocess.run(
        [resolve("systemctl"), "stop", TTP_SERVICE_NAME],
        capture_output=True,
        text=True,
        check=False,
    )
    # Clean up the volatile unit
    TTP_SERVICE_PATH.unlink(missing_ok=True)
    subprocess.run(
        [resolve("systemctl"), "daemon-reload"],
        capture_output=True,
        text=True,
        check=False,
    )
    logger.info("TTP Tor service stopped and unit removed.")

ttp.tor_install

Tor readiness checking and orchestration module.

This module enforces a strict NO AUTO-INSTALL policy. TTP will never attempt to install system packages automatically. If Tor or required pluggable transport helpers are missing, TTP displays distro-specific package guidance and official documentation links, then exits gracefully with status code 0.

ensure_pluggable_transports(required_transports)

Verify that required pluggable transport helper binaries are installed.

If any required binary is missing, displays distro package guidance and official Tor documentation links, then exits gracefully with status code 0.

Parameters:

Name Type Description Default
required_transports list[str]

List of pluggable transport names (e.g. ["obfs4", "snowflake"]).

 required

Raises:

Type Description
Exit

With exit code 0 if any transport binary is missing or unsupported.
Source code in ttp/tor_install.py 
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
def ensure_pluggable_transports(required_transports: list[str]) -> None:
    """Verify that required pluggable transport helper binaries are installed.

    If any required binary is missing, displays distro package guidance and official
    Tor documentation links, then exits gracefully with status code 0.

    Args:
        required_transports: List of pluggable transport names (e.g. ``["obfs4", "snowflake"]``).

    Raises:
        typer.Exit: With exit code 0 if any transport binary is missing or unsupported.
    """
    for pt in required_transports:
        pt = pt.lower()
        if pt not in PT_MAP:
            logger.error("Unsupported pluggable transport: '%s'", pt)
            from ttp.commands._common import _PREFIX, console

            console.print(f"{_PREFIX} [bold red]Unsupported pluggable transport: '{pt}'[/bold red]")
            raise typer.Exit(code=0)

        pt_info = PT_MAP[pt]
        binary = pt_info["binary"]

        if not resolve_optional(binary):
            cmd = _get_distro_install_command(
                pkg_debian=pt_info["apt-get"],
                pkg_fedora=pt_info["dnf"],
                pkg_arch=pt_info["pacman"],
                pkg_suse=pt_info["zypper"],
            )
            doc_url = "https://tb-manual.torproject.org/bridges/"

            msg = (
                f"[bold red]Pluggable transport helper binary '{binary}' (required for '{pt}') is missing.[/bold red]\n\n"
                f"[bold cyan]Recommended installation command:[/bold cyan]\n"
                f"  [bold yellow]{cmd}[/bold yellow]\n\n"
                f"[bold cyan]Official Tor Bridges Documentation:[/bold cyan]\n"
                f"  {doc_url}"
            )
            from ttp.commands._common import console

            console.print(Panel(msg, title="[bold red]Missing Dependency[/bold red]", expand=False))
            raise typer.Exit(code=0)

ensure_tor_ready(transport_port=9041, dns_port=9054, block_doh=True, use_bridges=False, bridges=None, disable_ipv6=False)

Ensure Tor is installed and start it via the dedicated systemd service.

Enforces a strict NO AUTO-INSTALL policy. If Tor or required helper binaries are missing, displays distro-aware installation guidance and exits gracefully with status code 0.

Parameters:

Name Type Description Default
transport_port int

Local TCP port for Tor TransPort redirection.

 9041
dns_port int

Local UDP/TCP port for Tor DNSPort redirection.

 9054
block_doh bool

If True, blocks known DoH resolvers in firewall and torrc.

 True
use_bridges bool

If True, enables Pluggable Transport bridges.

 False
bridges Optional[list[str]]

Optional list of bridge configuration strings.

 None
disable_ipv6 bool

If True, forces IPv6 client routing off.

 False

Returns:

Type Description
dict[str, Any]

dict[str, Any]: Dictionary containing detected Tor installation metadata.

Raises:

Type Description
Exit

With exit code 0 if Tor binary is not installed on the system.
Source code in ttp/tor_install.py 
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def ensure_tor_ready(
    transport_port: int = 9041,
    dns_port: int = 9054,
    block_doh: bool = True,
    use_bridges: bool = False,
    bridges: Optional[list[str]] = None,
    disable_ipv6: bool = False,
) -> dict[str, Any]:
    """Ensure Tor is installed and start it via the dedicated systemd service.

    Enforces a strict NO AUTO-INSTALL policy. If Tor or required helper binaries are missing,
    displays distro-aware installation guidance and exits gracefully with status code 0.

    Args:
        transport_port: Local TCP port for Tor TransPort redirection.
        dns_port: Local UDP/TCP port for Tor DNSPort redirection.
        block_doh: If True, blocks known DoH resolvers in firewall and torrc.
        use_bridges: If True, enables Pluggable Transport bridges.
        bridges: Optional list of bridge configuration strings.
        disable_ipv6: If True, forces IPv6 client routing off.

    Returns:
        dict[str, Any]: Dictionary containing detected Tor installation metadata.

    Raises:
        typer.Exit: With exit code 0 if Tor binary is not installed on the system.
    """
    info = detect_tor(transport_port=transport_port, dns_port=dns_port)

    if not info["is_installed"]:
        cmd = _get_distro_install_command(pkg_debian="tor", pkg_fedora="tor")
        doc_url = "https://community.torproject.org/onion-services/setup/install/"

        msg = (
            "[bold red]Tor daemon ('tor') is not installed on this system.[/bold red]\n\n"
            "[bold cyan]Recommended installation command:[/bold cyan]\n"
            f"  [bold yellow]{cmd}[/bold yellow]\n\n"
            "[bold cyan]Official Tor Project Installation Guide:[/bold cyan]\n"
            f"  {doc_url}"
        )
        from ttp.commands._common import console

        console.print(Panel(msg, title="[bold red]Missing Tor Dependency[/bold red]", expand=False))
        raise typer.Exit(code=0)

    tor_user = info.get("tor_user", "debian-tor")

    # If bridges are requested, ensure the corresponding pluggable transports are installed
    if use_bridges and bridges:
        required_transports = []
        for b in bridges:
            parts = b.split()
            if parts:
                first_word = parts[0].lower()
                if first_word in PT_MAP and first_word not in required_transports:
                    required_transports.append(first_word)
        if required_transports:
            ensure_pluggable_transports(required_transports)

    # Start Tor via the dedicated ttp-tor service
    start_tor_service(
        tor_user,
        transport_port=transport_port,
        dns_port=dns_port,
        block_doh=block_doh,
        use_bridges=use_bridges,
        bridges=bridges,
        disable_ipv6=disable_ipv6,
    )

    return info

ttp.tor_control

Tor daemon control and circuit verification.

This module is the "voice" of TTP. It communicates directly with the Tor daemon using the Stem library. It handles authentication, bootstrap monitoring, circuit rotation (Signal.NEWNYM), and external API verification to confirm that traffic is actually being routed through Tor.

KEY RESPONSIBILITIES: 1. Connect to Tor via Unix Socket or TCP ControlPort. 2. Monitor bootstrap progress until 100%. 3. Request new exit IPs (circuits). 4. Verify the current exit IP via multiple endpoints for resilience.

get_controller()

Connect to the dedicated ttp-tor control interface only.

Uses exclusively ControlSocket /run/tor/ttp/control.sock so bootstrap queries, NEWNYM, and shutdown signals always target TTP's isolated instance - never tor.service or TCP ControlPort. Returns an authenticated :class:stem.control.Controller or None.

Source code in ttp/tor_control.py 
 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
def get_controller():
    """Connect to the dedicated ``ttp-tor`` control interface only.

    Uses exclusively ``ControlSocket /run/tor/ttp/control.sock`` so
    bootstrap queries, NEWNYM, and shutdown signals always target TTP's
    isolated instance - never ``tor.service`` or TCP ControlPort.
    Returns an authenticated :class:`stem.control.Controller` or ``None``.
    """
    if Controller is None:
        return None

    if not os.path.exists(_TTP_CONTROL_SOCKET):
        return None

    import stem
    import stem.connection

    try:
        ctrl = Controller.from_socket_file(_TTP_CONTROL_SOCKET)
        ctrl.authenticate()
        return ctrl
    except (OSError, stem.SocketError) as e:
        logger.debug("Tor control socket not reachable: %s", e)
        return None
    except stem.connection.AuthenticationFailure as e:
        logger.error("Tor control socket authentication failed: %s", e)
        return None
    except stem.ControllerError as e:
        logger.warning("Tor controller error: %s", e)
        return None

get_exit_ip()

Fetch the current Tor exit IP, trying multiple endpoints for resilience.

Uses urllib.request from the stdlib so we don't need to add requests as a dependency.

Source code in ttp/tor_control.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_exit_ip() -> str:
    """Fetch the current Tor exit IP, trying multiple endpoints for resilience.

    Uses ``urllib.request`` from the stdlib so we don't need to add
    ``requests`` as a dependency.
    """
    for endpoint in VERIFY_ENDPOINTS:
        data = _fetch_endpoint(endpoint)
        if data is not None:
            # check.torproject.org uses "IP", ipify uses "ip", ifconfig.me uses "ip_addr"
            ip = data.get("IP") or data.get("ip") or data.get("ip_addr")
            if isinstance(ip, str) and ip:
                return ip
    return "unknown"

graceful_shutdown(timeout=10)

Send SHUTDOWN signal to Tor for clean circuit teardown.

This MUST be called before firewall teardown to avoid leaking cleartext RST packets on the physical interface. Tor will close all circuits cryptographically, then exit.

Parameters

timeout: Maximum seconds to wait for Tor to finish closing circuits.

Returns

bool True if the shutdown signal was sent successfully.

Source code in ttp/tor_control.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def graceful_shutdown(timeout: int = 10) -> bool:
    """Send ``SHUTDOWN`` signal to Tor for clean circuit teardown.

    This MUST be called **before** firewall teardown to avoid leaking
    cleartext RST packets on the physical interface.  Tor will close
    all circuits cryptographically, then exit.

    Parameters
    ----------
    timeout:
        Maximum seconds to wait for Tor to finish closing circuits.

    Returns
    -------
    bool
        ``True`` if the shutdown signal was sent successfully.
    """
    if Signal is None:
        return False

    ctrl = get_controller()
    if ctrl is None:
        return False

    import stem

    try:
        with ctrl:
            ctrl.signal(Signal.SHUTDOWN)

        # Wait for Tor to finish closing circuits
        for _ in range(timeout):
            ctrl_check = get_controller()
            if ctrl_check is None:
                return True
            try:
                ctrl_check.close()
            except (OSError, stem.ControllerError):
                pass
            time.sleep(1)
        return True
    except (OSError, stem.ControllerError):
        return False

request_new_circuit()

Request a new Tor circuit (new exit IP) and wait for it to change.

Returns

tuple[bool, str] (ip_changed, current_ip)

Source code in ttp/tor_control.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def request_new_circuit() -> tuple[bool, str]:
    """Request a new Tor circuit (new exit IP) and wait for it to change.

    Returns
    -------
    tuple[bool, str]
        ``(ip_changed, current_ip)``
    """
    old_ip = get_exit_ip()

    ctrl = get_controller()
    if ctrl is None:
        raise TorError("Cannot connect to Tor control interface. Check that Tor is running.")

    import stem

    try:
        with ctrl:
            ctrl.signal(Signal.NEWNYM)
    except (OSError, stem.ControllerError) as e:
        raise TorError(f"Failed to request new circuit from Tor controller: {e}") from e

    new_ip = old_ip

    # Poll for IP change instead of fixed sleep
    for _ in range(12):  # max ~60s
        time.sleep(5)
        new_ip = get_exit_ip()
        if new_ip != old_ip and new_ip != "unknown":
            return True, new_ip

    return False, new_ip

verify_tor()

Verify that traffic is actually routed through Tor.

Tries multiple endpoints for resilience. The Tor Project's API is authoritative (it returns IsTor); the fallback endpoints only confirm we can reach the internet through some exit node.

Returns

tuple[bool, str] (is_tor, exit_ip) - whether we confirmed Tor routing, and the exit IP address.

Source code in ttp/tor_control.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def verify_tor() -> tuple[bool, str]:
    """Verify that traffic is actually routed through Tor.

    Tries multiple endpoints for resilience. The Tor Project's API is
    authoritative (it returns ``IsTor``); the fallback endpoints only
    confirm we can reach the internet through *some* exit node.

    Returns
    -------
    tuple[bool, str]
        ``(is_tor, exit_ip)`` - whether we confirmed Tor routing,
        and the exit IP address.
    """
    for _attempt in range(1, 6):  # 5 attempts
        for endpoint in VERIFY_ENDPOINTS:
            data = _fetch_endpoint(endpoint)
            if data is None:
                continue

            # The Tor Project API is the only one that returns IsTor.
            if "IsTor" in data:
                return data.get("IsTor", False), data.get("IP", "unknown")

            # Fallback endpoints: we got a response, so traffic is routed
            # through *something*. We can't confirm it's Tor, but we have an IP.
            ip = data.get("ip") or data.get("ip_addr") or "unknown"
            return False, ip
        time.sleep(3)

    return False, "unknown"

wait_for_bootstrap(progress_callback=None, timeout=180)

Wait for Tor to reach 100% bootstrap status via ControlPort.

Parameters

progress_callback: Optional callable that takes an integer (0-100) representing the bootstrap percentage.

Source code in ttp/tor_control.py 
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
def wait_for_bootstrap(progress_callback: Optional[Callable[[int], None]] = None, timeout: int = 180) -> bool:
    """Wait for Tor to reach 100% bootstrap status via ControlPort.

    Parameters
    ----------
    progress_callback:
        Optional callable that takes an integer (0-100) representing
        the bootstrap percentage.
    """
    # 1. Wait for the control interface to be available (up to 60s).
    controller = None
    bootstrap_conn_timeout = 60
    for _ in range(bootstrap_conn_timeout // 2):
        controller = get_controller()
        if controller is not None:
            break
        time.sleep(2)

    if not controller:
        raise TorError(f"Could not connect to Tor control interface after {bootstrap_conn_timeout}s.")

    # 2. Monitor bootstrap progress.
    import stem

    try:
        with controller:
            for _ in range(timeout):  # Use the provided timeout (default 90)
                status = controller.get_info("status/bootstrap-phase")
                match = re.search(r"PROGRESS=(\d+)", status)
                progress_val = int(match.group(1)) if match else 0

                if progress_callback:
                    progress_callback(progress_val)

                if "PROGRESS=100" in status:
                    return True

                time.sleep(1)

            raise TorError("Tor bootstrap timed out.")
    except (OSError, stem.ControllerError) as e:
        raise TorError(f"Tor control socket communication failed during bootstrap: {e}") from e

DNS & State Management (ttp.dns, ttp.state)

ttp.dns

DNS Management Module - Handles routing DNS queries through Tor.

This module implements a stateless, Kernel-level DNS redirection strategy using a mount --bind overlay on /etc/resolv.conf.

apply_dns(interface, disable_ipv6=False, dns_port=9054)

Apply Tor DNS settings using a Kernel-level overlay (mount --bind).

If systemd-resolved is active, also writes a volatile drop-in configuration and restarts it via the ttp.dns_resolved module.

Returns a dictionary containing backup data for restoration.

Source code in ttp/dns.py 
 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
def apply_dns(interface: str, disable_ipv6: bool = False, dns_port: int = 9054) -> dict[str, Any]:
    """Apply Tor DNS settings using a Kernel-level overlay (mount --bind).

    If systemd-resolved is active, also writes a volatile drop-in configuration
    and restarts it via the ttp.dns_resolved module.

    Returns a dictionary containing backup data for restoration.
    """
    resolved_active = False
    try:
        from ttp import dns_resolved
        from ttp.system_info import is_ipv6_supported

        # Configure systemd-resolved if active
        resolved_active = dns_resolved.apply_resolved(dns_port=dns_port, disable_ipv6=disable_ipv6)

        nameservers = "nameserver 127.0.0.1\n"
        if is_ipv6_supported() and not disable_ipv6:
            nameservers += "nameserver ::1\n"

        # 1. Write the Tor resolver to /run/ttp/resolv.conf (volatile)
        RUNTIME_RESOLV.parent.mkdir(parents=True, exist_ok=True)
        RUNTIME_RESOLV.write_text(f"# Generated by TTP\n{nameservers}", encoding="utf-8")

        # 2. Symlink check: resolve the real target for mount --bind
        target = RESOLV_CONF
        if os.path.islink(str(RESOLV_CONF)):
            target = Path(os.path.realpath(str(RESOLV_CONF)))

        # 3. Clear any stale mount stacks (idempotency guard)
        _clear_stale_mounts(str(target))

        # 4. Overlay via mount --bind (non-destructive)
        subprocess.run(
            [resolve("mount"), "--bind", str(RUNTIME_RESOLV), str(target)],
            capture_output=True,
            text=True,
            check=True,
            timeout=10,
        )

        return {
            "mode": "overlay",
            "mount_target": str(target),
            "systemd_resolved": resolved_active,
        }
    except Exception as e:
        # Clean up any systemd-resolved drop-in if we failed during overlay setup
        if resolved_active:
            try:
                from ttp import dns_resolved

                dns_resolved.restore_resolved()
            except Exception:
                pass

        if isinstance(e, subprocess.CalledProcessError):
            raise DNSError(f"Command failed: {e.cmd} -> {e.stderr.strip()}") from e
        raise DNSError(f"Failed to apply DNS configuration: {e}") from e

detect_active_interface()

Detect the primary network interface using 'ip route'.

Source code in ttp/dns.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def detect_active_interface() -> str:
    """Detect the primary network interface using 'ip route'."""
    try:
        result = subprocess.run(
            [resolve("ip"), "route", "show", "default"],
            capture_output=True,
            text=True,
            check=True,
            timeout=10,
        )
        # Output example: "default via 192.168.1.1 dev eth0 proto dhcp..."
        parts = result.stdout.split()
        if "dev" in parts:
            return parts[parts.index("dev") + 1]
    except (subprocess.CalledProcessError, IndexError):
        pass
    return "eth0"  # Sane fallback if detection fails

restore_dns(backup)

Restore original system DNS settings by unmounting the overlay.

If systemd-resolved was active on startup, also removes the volatile drop-in configuration and restarts it.

Source code in ttp/dns.py 
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
187
188
189
190
191
192
193
194
195
196
def restore_dns(backup: dict[str, Any] | None) -> None:
    """Restore original system DNS settings by unmounting the overlay.

    If systemd-resolved was active on startup, also removes the volatile
    drop-in configuration and restarts it.
    """
    if not backup:
        return

    # 1. Unmount TTP DNS overlay first to restore the base /etc/resolv.conf file
    mount_target = backup.get("mount_target", str(RESOLV_CONF))

    if _is_ttp_mount(mount_target):
        try:
            # Lazy unmount ensures immediate release even if busy
            subprocess.run(
                [resolve("umount"), "-l", mount_target],
                capture_output=True,
                text=True,
                check=True,
                timeout=10,
            )
            logger.info("Successfully unmounted DNS overlay on %s", mount_target)
        except subprocess.CalledProcessError as e:
            err_msg = e.stderr.strip() if e.stderr else str(e)
            logger.warning("Failed to unmount DNS overlay on %s: %s", mount_target, err_msg)
    else:
        logger.debug("DNS target %s is not mounted, skipping unmount", mount_target)

    # 2. Handle systemd-resolved teardown second (so it reads the restored base resolv.conf)
    if backup.get("systemd_resolved"):
        try:
            from ttp import dns_resolved

            dns_resolved.restore_resolved()
        except Exception as e:
            logger.warning("Failed to restore systemd-resolved: %s", e)

    # 3. Cleanup the ephemeral file to free tmpfs space
    try:
        if RUNTIME_RESOLV.exists():
            RUNTIME_RESOLV.unlink()
    except OSError as e:
        logger.debug("Failed to remove runtime resolv.conf: %s", e)

ttp.state

State management - Volatile lock file for crash-safe operations.

This module acts as the "memory" of TTP. It tracks active sessions using a JSON-formatted lock file stored in /run/ttp/ (a tmpfs mount). Because the lock lives on a volatile filesystem, it vanishes on reboot, eliminating stale-lock issues after power loss.

The only persistent path used by TTP is /var/lib/ttp/ which is managed by ttp.ux (one-time UX engagement features).

CORE CONCEPTS: - Lock File: Located at /run/ttp/ttp.lock (volatile - tmpfs). - Orphans: A lock exists but the recorded PID is dead. - Recovery: The process of reading an orphan lock and calling rollback logic.

attempt_recovery(destroy_firewall, restore_dns)

Attempt automatic recovery from an orphaned lock.

Reads the lock, invokes the firewall and DNS restoration callbacks, then deletes the lock. Returns True on success.

Parameters

destroy_firewall: firewall.destroy_rules() restore_dns: dns.restore_dns(backup)

Source code in ttp/state.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def attempt_recovery(
    destroy_firewall: Callable[[], Any],
    restore_dns: Callable[[Any], Any],
) -> bool:
    """Attempt automatic recovery from an orphaned lock.

    Reads the lock, invokes the firewall and DNS restoration
    callbacks, then deletes the lock.  Returns ``True`` on success.

    Parameters
    ----------
    destroy_firewall:
        ``firewall.destroy_rules()``
    restore_dns:
        ``dns.restore_dns(backup)``
    """
    data = read_lock()
    if data is None:
        return False

    try:
        destroy_firewall()
        restore_dns(data.get("dns_backup"))
    finally:
        delete_lock()

    return True

check_tmpfs_space(min_bytes=MIN_TMPFS_BYTES)

Abort if /run (tmpfs) has insufficient free space.

Must be called before any I/O to /run so that TTP fails fast instead of crashing mid-setup with ENOSPC.

Raises

StateError If free space on /run is below min_bytes.

Source code in ttp/state.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def check_tmpfs_space(min_bytes: int = MIN_TMPFS_BYTES) -> None:
    """Abort if ``/run`` (tmpfs) has insufficient free space.

    Must be called **before** any I/O to ``/run`` so that TTP
    fails fast instead of crashing mid-setup with ``ENOSPC``.

    Raises
    ------
    StateError
        If free space on ``/run`` is below *min_bytes*.
    """
    try:
        usage = shutil.disk_usage("/run")
        if usage.free < min_bytes:
            free_mb = usage.free / (1024 * 1024)
            min_mb = min_bytes / (1024 * 1024)
            raise StateError(
                f"Insufficient space on /run (tmpfs): {free_mb:.1f}MB free, "
                f"minimum {min_mb:.1f}MB required. "
                f"Free space before starting TTP."
            )
    except OSError as e:
        if isinstance(e, StateError):
            raise

delete_lock()

Remove the lock file if it exists.

Source code in ttp/state.py 
229
230
231
def delete_lock() -> None:
    """Remove the lock file if it exists."""
    LOCK_PATH.unlink(missing_ok=True)

ensure_runtime_dir()

Create /run/ttp with mode 0700.

Must be called early in the CLI startup before any I/O that targets the runtime directory (lock file, log file, torrc, etc.).

Owned by ttp-watchdog if the user exists, otherwise owned by root.

Source code in ttp/state.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def ensure_runtime_dir() -> None:
    """Create ``/run/ttp`` with mode 0700.

    Must be called early in the CLI startup before any I/O that targets
    the runtime directory (lock file, log file, torrc, etc.).

    Owned by ttp-watchdog if the user exists, otherwise owned by root.
    """
    import pwd

    LOCK_DIR.mkdir(parents=True, exist_ok=True)
    os.chmod(LOCK_DIR, 0o700)

    uid = 0
    gid = 0
    try:
        pw = pwd.getpwnam("ttp-watchdog")
        uid = pw.pw_uid
        gid = pw.pw_gid
    except KeyError:
        pass

    os.chown(LOCK_DIR, uid, gid)

is_orphan()

Return True if the lock file exists but its PID is dead or recycled.

Uses os.kill(pid, 0) which sends no signal but raises OSError when the target process does not exist.

Source code in ttp/state.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def is_orphan() -> bool:
    """Return ``True`` if the lock file exists but its PID is dead or recycled.

    Uses ``os.kill(pid, 0)`` which sends no signal but raises
    ``OSError`` when the target process does not exist.
    """
    data = read_lock()
    if data is None:
        return False

    pid = data.get("pid")
    if pid is None:
        return True  # corrupt lock -> treat as orphan

    try:
        os.kill(pid, 0)
        if not _is_pid_ttp(pid):
            return True  # PID is alive but not TTP -> recycled (orphan)
    except OSError:
        return True  # process not running -> orphan
    return False

read_lock()

Read and return the lock data, or None if no lock exists.

Source code in ttp/state.py 
216
217
218
219
220
221
222
223
224
225
226
def read_lock() -> dict[str, Any] | None:
    """Read and return the lock data, or ``None`` if no lock exists."""
    if not LOCK_PATH.exists():
        return None
    try:
        data = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
        if isinstance(data, dict):
            return data
        return None
    except (json.JSONDecodeError, OSError):
        return None

update_lock_keys(**kwargs)

Update specific keys in the existing lock file, preserving other keys.

If no lock file exists, this raises a StateError.

Source code in ttp/state.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
def update_lock_keys(**kwargs: Any) -> None:
    """Update specific keys in the existing lock file, preserving other keys.

    If no lock file exists, this raises a StateError.
    """
    data = read_lock()
    if data is None:
        raise StateError("No active TTP session found to update.")
    data.update(kwargs)
    try:
        _write_lock_file(data)
    except OSError as e:
        raise StateError(f"Failed to update session lock file: {e}")

write_lock(*, pid=None, dns_backup=None, transport_port=9041, dns_port=9054, allow_root=False, lan_bypass=True, watchdog_active=False, watchdog_pid=None, interface=None, bypass_users=None, bypass_groups=None, use_bridges=False, bridge_file=None, bridges=None, external_daemon=False, no_ipv6=False, tor_uid=None)

Write the session lock file with the current state.

Parameters

pid: PID to record. Defaults to the current process. dns_backup: Original DNS data (resolv.conf mount target dictionary). transport_port: The customized or default TransPort port. dns_port: The customized or default DNSPort port. allow_root: True to allow root processes to bypass Tor. lan_bypass: True to exempt LAN subnet traffic from Tor routing. watchdog_active: True if the watchdog background daemon is active. watchdog_pid: PID of the active watchdog daemon if running. interface: The name of the primary active interface being proxyed. bypass_users: List of system users bypassed from Tor routing. bypass_groups: List of system groups bypassed from Tor routing. use_bridges: True if Tor bridges support is enabled. bridge_file: Path to the bridge file, if provided. bridges: List of configured bridge lines. tor_uid: The resolved UID of the Tor daemon process.

Source code in ttp/state.py 
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
def write_lock(
    *,
    pid: int | None = None,
    dns_backup: Any = None,
    transport_port: int = 9041,
    dns_port: int = 9054,
    allow_root: bool = False,
    lan_bypass: bool = True,
    watchdog_active: bool = False,
    watchdog_pid: int | None = None,
    interface: str | None = None,
    bypass_users: list[str] | None = None,
    bypass_groups: list[str] | None = None,
    use_bridges: bool = False,
    bridge_file: str | None = None,
    bridges: list[str] | None = None,
    external_daemon: bool = False,
    no_ipv6: bool = False,
    tor_uid: int | None = None,
) -> None:
    """Write the session lock file with the current state.

    Parameters
    ----------
    pid:
        PID to record.  Defaults to the current process.
    dns_backup:
        Original DNS data (resolv.conf mount target dictionary).
    transport_port:
        The customized or default TransPort port.
    dns_port:
        The customized or default DNSPort port.
    allow_root:
        True to allow root processes to bypass Tor.
    lan_bypass:
        True to exempt LAN subnet traffic from Tor routing.
    watchdog_active:
        True if the watchdog background daemon is active.
    watchdog_pid:
        PID of the active watchdog daemon if running.
    interface:
        The name of the primary active interface being proxyed.
    bypass_users:
        List of system users bypassed from Tor routing.
    bypass_groups:
        List of system groups bypassed from Tor routing.
    use_bridges:
        True if Tor bridges support is enabled.
    bridge_file:
        Path to the bridge file, if provided.
    bridges:
        List of configured bridge lines.
    tor_uid:
        The resolved UID of the Tor daemon process.
    """
    try:
        data = {
            "pid": pid if pid is not None else os.getpid(),
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "dns_backup": dns_backup,
            "transport_port": transport_port,
            "dns_port": dns_port,
            "allow_root": allow_root,
            "lan_bypass": lan_bypass,
            "watchdog_active": watchdog_active,
            "watchdog_pid": watchdog_pid,
            "interface": interface,
            "bypass_users": bypass_users if bypass_users is not None else [],
            "bypass_groups": bypass_groups if bypass_groups is not None else [],
            "use_bridges": use_bridges,
            "bridge_file": bridge_file,
            "bridges": bridges if bridges is not None else [],
            "external_daemon": external_daemon,
            "no_ipv6": no_ipv6,
            "tor_uid": tor_uid,
        }
        _write_lock_file(data)
    except OSError as e:
        raise StateError(f"Failed to write session lock file: {e}")

Watchdog Engine (ttp.watchdog)

ttp.watchdog.integrity

Session integrity checks and auto-healing logic.

attempt_auto_healing(failed_component)

Attempt to dynamically repair a failed session component.

Returns:

bool True if the healing commands succeeded, False otherwise.

Source code in ttp/watchdog/integrity.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def attempt_auto_healing(failed_component: str) -> bool:
    """Attempt to dynamically repair a failed session component.

    Returns:
    --------
    bool
        True if the healing commands succeeded, False otherwise.
    """
    lock = state.read_lock()
    if not lock:
        return False

    logger.warning(
        "Watchdog: Initiating auto-healing for failed component '%s'...",
        failed_component,
    )
    try:
        if failed_component == "tor":
            logger.info("Watchdog: Restarting Tor service via systemctl...")
            res = subprocess.run(
                [resolve("systemctl"), "restart", "ttp-tor.service"],
                capture_output=True,
                text=True,
                check=False,
                timeout=10,
            )
            if res.returncode == 0:
                logger.info("Watchdog: Restarted Tor service successfully.")
                return True
            else:
                logger.error(
                    "Watchdog: Failed to restart Tor service: %s",
                    res.stderr.strip() if res.stderr else f"Exit code {res.returncode}",
                )
                return False
        else:
            # dns and firewall tampering: fail closed immediately.
            logger.error(
                "Watchdog: Tampering or failure detected on critical component '%s'. "
                "Fail-closed policy active: auto-healing skipped.",
                failed_component,
            )
            return False
    except Exception as e:
        logger.error("Watchdog: Auto-healing failed for '%s': %s", failed_component, e)
        return False

check_system_integrity()

Verify Tor connection, firewall rules and DNS overlay.

Returns:

tuple[Optional[str], Optional[str]] (failed_component, error_message) e.g., ("dns", "overlay unmounted") or (None, None) if all is healthy.

Source code in ttp/watchdog/integrity.py 
 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
def check_system_integrity() -> tuple[Optional[str], Optional[str]]:
    """Verify Tor connection, firewall rules and DNS overlay.

    Returns:
    --------
    tuple[Optional[str], Optional[str]]
        (failed_component, error_message)
        e.g., ("dns", "overlay unmounted") or (None, None) if all is healthy.
    """
    lock = state.read_lock()

    # 1. DNS Overlay mount check
    target = dns.RESOLV_CONF
    if os.path.islink(str(dns.RESOLV_CONF)):
        target = Path(os.path.realpath(str(dns.RESOLV_CONF)))
    if not dns._is_mount_point(str(target)):
        return "dns", "resolv.conf overlay mount has been unmounted"

    # Verify content points to localhost nameservers only
    try:
        content = Path("/etc/resolv.conf").read_text(encoding="utf-8")
        nameservers = []
        for line in content.splitlines():
            line = line.strip()
            if line.startswith("nameserver"):
                parts = line.split()
                if len(parts) >= 2:
                    nameservers.append(parts[1])
        if not nameservers:
            return "dns", "resolv.conf has no nameservers configured"
        for ns in nameservers:
            if ns not in ("127.0.0.1", "::1"):
                return (
                    "dns",
                    f"resolv.conf nameserver points to non-local resolver: {ns}",
                )
    except Exception as e:
        return "dns", f"Failed to read/verify resolv.conf: {e}"

    # 1b. Check systemd-resolved if it was active on startup
    if lock:
        dns_backup = lock.get("dns_backup")
        if dns_backup and dns_backup.get("systemd_resolved"):
            resolved_config = Path("/run/systemd/resolved.conf.d/ttp.conf")
            if not resolved_config.exists():
                return (
                    "dns",
                    "systemd-resolved drop-in configuration file has been deleted",
                )
            res_resolved = subprocess.run(
                [resolve("systemctl"), "is-active", "systemd-resolved"],
                capture_output=True,
                text=True,
                check=False,
                timeout=10,
            )
            if res_resolved.stdout.strip() != "active":
                return "dns", "systemd-resolved systemd service is inactive/stopped"

    # 2. Firewall Ruleset check
    res = subprocess.run(
        [resolve("nft"), "list", "table", "inet", "ttp"],
        capture_output=True,
        text=True,
        check=False,
        timeout=10,
    )
    if res.returncode != 0:
        return "firewall", "nftables 'inet ttp' table is missing"
    if "chain filter_out" not in res.stdout:
        return (
            "firewall",
            "nftables 'inet ttp' table is incomplete (missing filter_out)",
        )

    # Verify bypass rules if configured in state lock
    if lock:
        import grp
        import pwd

        for u in lock.get("bypass_users", []):
            try:
                uid = int(u) if u.isdigit() else pwd.getpwnam(u).pw_uid
                if f"meta skuid {uid} accept" not in res.stdout:
                    return (
                        "firewall",
                        f"bypass rule for user '{u}' (UID {uid}) is missing",
                    )
            except KeyError:
                return "firewall", f"bypass user '{u}' cannot be resolved on system"

        for g in lock.get("bypass_groups", []):
            try:
                gid = int(g) if g.isdigit() else grp.getgrnam(g).gr_gid
                if f"meta skgid {gid} accept" not in res.stdout:
                    return (
                        "firewall",
                        f"bypass rule for group '{g}' (GID {gid}) is missing",
                    )
            except KeyError:
                return "firewall", f"bypass group '{g}' cannot be resolved on system"

    # 3. Tor Connection check: perform an *active* query to the control socket
    ctrl = tor_control.get_controller()
    if ctrl is not None:
        try:
            with ctrl:
                ctrl.get_info("status/bootstrap-phase")
        except Exception as e:
            return "tor", f"Tor control interface unresponsive: {e}"
    else:
        # Control socket unavailable - fall back to systemd service status
        res_tor = subprocess.run(
            [resolve("systemctl"), "is-active", "ttp-tor"],
            capture_output=True,
            text=True,
            check=False,
            timeout=10,
        )
        if res_tor.stdout.strip() != "active":
            return "tor", "Tor systemd service is inactive/stopped"

    return None, None

has_default_route()

Return True if a default gateway route exists in the system.

Source code in ttp/watchdog/integrity.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def has_default_route() -> bool:
    """Return True if a default gateway route exists in the system."""
    try:
        route_path = Path("/proc/net/route")
        if not route_path.exists():
            return False
        with open(route_path) as f:
            for line in f:
                parts = line.split()
                if len(parts) >= 8:
                    # Destination is 2nd column, Mask is 8th column
                    dest = parts[1]
                    mask = parts[7]
                    if dest == "00000000" and mask == "00000000":
                        return True
    except OSError:
        pass
    return False

is_interface_online(interface)

Check if a network interface is physically online (has carrier and is up).

Source code in ttp/watchdog/integrity.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def is_interface_online(interface: str) -> bool:
    """Check if a network interface is physically online (has carrier and is up)."""
    sys_path = Path(f"/sys/class/net/{interface}")
    if not sys_path.exists():
        return False
    try:
        # Check operstate
        operstate_file = sys_path / "operstate"
        if operstate_file.exists():
            operstate = operstate_file.read_text().strip().lower()
            # If state is explicitly down, it is offline
            if operstate == "down":
                return False

        # Check carrier
        carrier_file = sys_path / "carrier"
        if carrier_file.exists():
            carrier = carrier_file.read_text().strip()
            if carrier == "0":
                return False
        return True
    except OSError:
        return False

ttp.watchdog.fsm

Finite State Machine (FSM) implementation for TTP watchdog.

WatchdogFSM

Watchdog Finite State Machine managing transitions, sockets, and recovery logic.

Source code in ttp/watchdog/fsm.py 
 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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
class WatchdogFSM:
    """Watchdog Finite State Machine managing transitions, sockets, and recovery logic."""

    states = ["stopped", "healthy", "suspended", "healing", "killswitch"]

    def __init__(self) -> None:
        self.netlink_socket: socket.socket | None = None
        self.inotify_fd: int = -1
        self.wd_real: int = -1
        self.wd_link: int = -1
        self.interface: str | None = None
        self.interval_seconds: int = 15
        self.last_heal_time: float = 0.0
        self.last_check_time: float = 0.0
        self.COOLDOWN_SECONDS: float = 2.0
        self._libc: Any = None

        # Initialize Transitions Machine
        self.machine = Machine(
            model=self,
            states=WatchdogFSM.states,
            initial="stopped",
            send_event=True,
        )

        # Transition Rules
        self.machine.add_transition(
            trigger="initialize",
            source="stopped",
            dest="healthy",
            before="_on_initialize",
        )
        self.machine.add_transition(
            trigger="disconnect",
            source="healthy",
            dest="suspended",
            before="_on_disconnect",
        )
        self.machine.add_transition(
            trigger="reconnect",
            source="suspended",
            dest="healthy",
            before="_on_reconnect",
        )
        self.machine.add_transition(
            trigger="integrity_fail",
            source="healthy",
            dest="healing",
            after="_on_integrity_fail",
        )
        self.machine.add_transition(
            trigger="heal_success",
            source="healing",
            dest="healthy",
            before="_on_heal_success",
        )
        self.machine.add_transition(
            trigger="heal_fail",
            source="healing",
            dest="killswitch",
            before="_on_heal_fail",
        )
        self.machine.add_transition(
            trigger="tamper",
            source="healthy",
            dest="killswitch",
            before="_on_tamper",
        )
        self.machine.add_transition(
            trigger="shutdown",
            source="*",
            dest="stopped",
            before="_on_shutdown",
        )

    def _load_libc(self) -> None:
        """Helper to dynamically load libc for inotify functions."""
        if self._libc is None:
            libc_name = ctypes.util.find_library("c")
            self._libc = ctypes.CDLL(libc_name, use_errno=True)

            self._libc.inotify_init.argtypes = []
            self._libc.inotify_init.restype = ctypes.c_int
            self._libc.inotify_add_watch.argtypes = [
                ctypes.c_int,
                ctypes.c_char_p,
                ctypes.c_uint32,
            ]
            self._libc.inotify_add_watch.restype = ctypes.c_int
            self._libc.inotify_rm_watch.argtypes = [ctypes.c_int, ctypes.c_int]
            self._libc.inotify_rm_watch.restype = ctypes.c_int

    def readd_watch(self) -> None:
        """Re-register inotify watches on the target file and symlink."""
        self._load_libc()
        # Remove old watches
        for wd_val in (self.wd_real, self.wd_link):
            if wd_val >= 0:
                try:
                    self._libc.inotify_rm_watch(self.inotify_fd, wd_val)
                except Exception:
                    pass
        self.wd_real = -1
        self.wd_link = -1

        # 1. Watch the real target path of resolv.conf
        try:
            resolv_real_path = os.path.realpath("/etc/resolv.conf")
            self.wd_real = self._libc.inotify_add_watch(self.inotify_fd, resolv_real_path.encode("utf-8"), WATCH_MASK)
            if self.wd_real >= 0:
                logger.info(
                    "Watchdog: Inotify watch established on real target %s",
                    resolv_real_path,
                )
            else:
                logger.warning(
                    "Watchdog: Failed to add inotify watch on real target %s",
                    resolv_real_path,
                )
        except Exception as e:
            logger.warning("Watchdog: Exception when adding watch on real target: %s", e)

        # 2. Watch the symlink itself (without following) to detect link target swapping
        try:
            self.wd_link = self._libc.inotify_add_watch(
                self.inotify_fd, b"/etc/resolv.conf", WATCH_MASK | IN_DONT_FOLLOW
            )
            if self.wd_link >= 0:
                logger.info("Watchdog: Inotify watch established on symlink /etc/resolv.conf")
            else:
                logger.warning("Watchdog: Failed to add inotify watch on symlink /etc/resolv.conf")
        except Exception as e:
            logger.warning("Watchdog: Exception when adding watch on symlink: %s", e)

    def flush_event_buffers(self, fds: list[Any]) -> None:
        """Discard any accumulated events in netlink or inotify queues."""
        sock = self.netlink_socket
        if sock is not None and sock in fds:
            try:
                while True:
                    data = sock.recv(65535)
                    if not isinstance(data, (bytes, bytearray)) or len(data) == 0:
                        break
            except BlockingIOError:
                pass
            except Exception:
                pass

        if self.inotify_fd in fds and self.inotify_fd >= 0:
            try:
                while True:
                    data = os.read(self.inotify_fd, 4096)
                    if not isinstance(data, (bytes, bytearray)) or len(data) == 0:
                        break
            except BlockingIOError:
                pass
            except Exception:
                pass

    # Transition Callbacks
    def _on_initialize(self, event: Any) -> None:
        self.interface = event.kwargs.get("interface")
        self.interval_seconds = event.kwargs.get("interval_seconds", 15)

        logger.info(
            "Watchdog FSM: Initializing monitoring. Interface: %s. Heartbeat: %d",
            self.interface,
            self.interval_seconds,
        )

        # Setup Netlink Socket
        try:
            self.netlink_socket = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_NETFILTER)
            self.netlink_socket.setsockopt(SOL_NETLINK, NETLINK_ADD_MEMBERSHIP, NFNLGRP_NFTABLES)
            self.netlink_socket.bind((0, 0))
            self.netlink_socket.setblocking(False)
        except Exception as e:
            logger.critical("Watchdog FSM failed to setup Netlink socket: %s", e)
            trigger_emergency_killswitch("firewall", f"Netlink setup failure: {e}")
            raise e

        # Setup Inotify
        try:
            self._load_libc()
            self.inotify_fd = self._libc.inotify_init()
            if self.inotify_fd < 0:
                raise OSError("inotify_init failed")
            os.set_blocking(self.inotify_fd, False)
            self.readd_watch()
        except Exception as e:
            logger.critical("Watchdog FSM failed to setup Inotify: %s", e)
            if self.netlink_socket:
                try:
                    self.netlink_socket.close()
                except Exception:
                    pass
            trigger_emergency_killswitch("dns", f"Inotify setup failure: {e}")
            raise e

    def _on_disconnect(self, event: Any) -> None:
        logger.warning(
            "Watchdog FSM: Network link went offline. Suspending monitoring on %s.",
            self.interface,
        )
        self.flush_event_buffers([self.netlink_socket, self.inotify_fd])

    def _on_reconnect(self, event: Any) -> None:
        logger.info(
            "Watchdog FSM: Network link restored on '%s'. Waiting 10 seconds for Tor circuit stabilization...",
            self.interface,
        )
        time.sleep(10)
        self.readd_watch()
        self.flush_event_buffers([self.netlink_socket, self.inotify_fd])

    def _on_integrity_fail(self, event: Any) -> None:
        failed_comp = event.kwargs.get("failed_comp")
        err_msg = event.kwargs.get("err_msg")

        logger.warning(
            "Watchdog FSM: Integrity check failed! Component: %s. Error: %s",
            failed_comp,
            err_msg,
        )
        logger.info(
            "Watchdog FSM: Initiating auto-healing for failed component '%s'...",
            failed_comp,
        )

        healed = attempt_auto_healing(failed_comp)
        self.last_heal_time = time.time()

        if not healed:
            logger.error(
                "Watchdog FSM: Auto-healing command failed for '%s'. Triggering emergency killswitch.",
                failed_comp,
            )
            # Fail immediately by triggering state change to killswitch
            self.heal_fail(failed_comp=failed_comp, err_msg=err_msg)

    def _on_heal_success(self, event: Any) -> None:
        logger.info("Watchdog FSM: Auto-healing was successful. Session integrity restored.")

    def _on_heal_fail(self, event: Any) -> None:
        failed_comp = event.kwargs.get("failed_comp")
        err_msg = event.kwargs.get("err_msg")
        trigger_emergency_killswitch(failed_comp, err_msg)

    def _on_tamper(self, event: Any) -> None:
        failed_comp = event.kwargs.get("failed_comp")
        err_msg = event.kwargs.get("err_msg")
        logger.critical(
            "Watchdog FSM: Tampering detected on critical component '%s'! Activating emergency killswitch.",
            failed_comp,
        )
        trigger_emergency_killswitch(failed_comp, err_msg)

    def _on_shutdown(self, event: Any) -> None:
        logger.info("Watchdog FSM: Stopping watchdog. Cleaning up resources.")
        if self.netlink_socket:
            try:
                self.netlink_socket.close()
            except Exception:
                pass
            self.netlink_socket = None

        if self.inotify_fd >= 0:
            # Try to remove watches
            for wd_val in (self.wd_real, self.wd_link):
                if wd_val >= 0:
                    try:
                        self._libc.inotify_rm_watch(self.inotify_fd, wd_val)
                    except Exception:
                        pass
            try:
                os.close(self.inotify_fd)
            except Exception:
                pass
            self.inotify_fd = -1
            self.wd_real = -1
            self.wd_link = -1

flush_event_buffers(fds)

Discard any accumulated events in netlink or inotify queues.

Source code in ttp/watchdog/fsm.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def flush_event_buffers(self, fds: list[Any]) -> None:
    """Discard any accumulated events in netlink or inotify queues."""
    sock = self.netlink_socket
    if sock is not None and sock in fds:
        try:
            while True:
                data = sock.recv(65535)
                if not isinstance(data, (bytes, bytearray)) or len(data) == 0:
                    break
        except BlockingIOError:
            pass
        except Exception:
            pass

    if self.inotify_fd in fds and self.inotify_fd >= 0:
        try:
            while True:
                data = os.read(self.inotify_fd, 4096)
                if not isinstance(data, (bytes, bytearray)) or len(data) == 0:
                    break
        except BlockingIOError:
            pass
        except Exception:
            pass

readd_watch()

Re-register inotify watches on the target file and symlink.

Source code in ttp/watchdog/fsm.py 
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
def readd_watch(self) -> None:
    """Re-register inotify watches on the target file and symlink."""
    self._load_libc()
    # Remove old watches
    for wd_val in (self.wd_real, self.wd_link):
        if wd_val >= 0:
            try:
                self._libc.inotify_rm_watch(self.inotify_fd, wd_val)
            except Exception:
                pass
    self.wd_real = -1
    self.wd_link = -1

    # 1. Watch the real target path of resolv.conf
    try:
        resolv_real_path = os.path.realpath("/etc/resolv.conf")
        self.wd_real = self._libc.inotify_add_watch(self.inotify_fd, resolv_real_path.encode("utf-8"), WATCH_MASK)
        if self.wd_real >= 0:
            logger.info(
                "Watchdog: Inotify watch established on real target %s",
                resolv_real_path,
            )
        else:
            logger.warning(
                "Watchdog: Failed to add inotify watch on real target %s",
                resolv_real_path,
            )
    except Exception as e:
        logger.warning("Watchdog: Exception when adding watch on real target: %s", e)

    # 2. Watch the symlink itself (without following) to detect link target swapping
    try:
        self.wd_link = self._libc.inotify_add_watch(
            self.inotify_fd, b"/etc/resolv.conf", WATCH_MASK | IN_DONT_FOLLOW
        )
        if self.wd_link >= 0:
            logger.info("Watchdog: Inotify watch established on symlink /etc/resolv.conf")
        else:
            logger.warning("Watchdog: Failed to add inotify watch on symlink /etc/resolv.conf")
    except Exception as e:
        logger.warning("Watchdog: Exception when adding watch on symlink: %s", e)