steam-backlog-enforcer/steam_backlog_enforcer/_store_tools.py
Krzysztof kuhy Rudnicki 7ea3e9e734 Exempt itch.io from the total block
kuhy 2026-11-11: itch.io unblocked, only it. A named
TOTAL_BLOCK_EXEMPT_DOMAINS drops itch from the hosts section and
its IPs from iptables (the exception wins on shared Cloudflare
IPs). New root command total-block-refresh applies the list to a
live block: hosts first, then per-rule iptables deletes, no flush.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWWzuCEg1m87jpKyRrw
2026-11-11 13:04:53 +01:00

106 lines
3.4 KiB
Python

"""The executables and paths every store-blocking layer shells out to.
One copy, imported by the hosts, iptables and orchestration modules alike --
each used to carry its own after the 250-line split, three blocks that had
to be kept identical by hand.
"""
from __future__ import annotations
from pathlib import Path
import shutil
import subprocess
# Path to the hosts install script. _SRC_ROOT is ~/src (this module lives two
# levels below it); the generator moved out of testsAndMisc into its own
# hosts-blocker repo, whose custom_entries.hosts carries the Steam Store block.
_SRC_ROOT = Path(__file__).resolve().parents[2]
HOSTS_INSTALL_SCRIPT = _SRC_ROOT / "hosts-blocker" / "install.sh"
# iptables chain name for our blocking rules.
IPTABLES_CHAIN = "STEAM_ENFORCER"
# Resolved absolute paths for executables (avoids S607 partial-path warnings).
SUDO = shutil.which("sudo") or "/usr/bin/sudo"
IPTABLES = shutil.which("iptables") or "/usr/sbin/iptables"
BASH = shutil.which("bash") or "/usr/bin/bash"
GUARDCTL = shutil.which("guardctl") or "/usr/local/bin/guardctl"
TEE = shutil.which("tee") or "/usr/bin/tee"
# IP address used in /etc/hosts for blocking domains.
HOSTS_REDIRECT_IP = ".".join(["0"] * 4)
def run_quiet(
cmd: list[str], *, timeout: float = 5
) -> subprocess.CompletedProcess[bytes]:
"""Run ``cmd`` with output captured and no exception on a non-zero exit."""
return subprocess.run(cmd, capture_output=True, timeout=timeout, check=False)
def ensure_chain_in_output(chain: str) -> None:
"""Hook ``chain`` into iptables' OUTPUT if it is not already there.
Raises:
subprocess.CalledProcessError: If the insert itself fails -- a chain
that exists but is not reachable from OUTPUT blocks nothing.
"""
probe = run_quiet([SUDO, IPTABLES, "-C", "OUTPUT", "-j", chain])
if probe.returncode != 0:
subprocess.run(
[SUDO, IPTABLES, "-I", "OUTPUT", "-j", chain],
capture_output=True,
timeout=5,
check=True,
)
def create_fresh_chain(chain: str) -> None:
"""Create ``chain`` if missing, then empty it, ready to be refilled.
``-N`` fails harmlessly when the chain already exists, so only the flush
is checked.
Raises:
subprocess.CalledProcessError: If the flush fails.
"""
run_quiet([SUDO, IPTABLES, "-N", chain])
subprocess.run(
[SUDO, IPTABLES, "-F", chain],
capture_output=True,
timeout=5,
check=True,
)
def add_drop_rule(chain: str, ip: str) -> None:
"""Append a rule to ``chain`` dropping all traffic to ``ip``.
Raises:
subprocess.CalledProcessError: If iptables rejects the rule.
"""
subprocess.run(
[SUDO, IPTABLES, "-A", chain, "-d", ip, "-j", "DROP"],
capture_output=True,
timeout=5,
check=True,
)
def delete_drop_rule(chain: str, ip: str) -> bool:
"""Remove the rule dropping ``ip`` from ``chain``; False if iptables refuses."""
return (
run_quiet([SUDO, IPTABLES, "-D", chain, "-d", ip, "-j", "DROP"]).returncode == 0
)
def drop_chain(chain: str) -> None:
"""Unhook ``chain`` from OUTPUT, flush it and delete it.
Each step tolerates failure, so removing an absent or half-built chain
is a no-op rather than an error.
"""
run_quiet([SUDO, IPTABLES, "-D", "OUTPUT", "-j", chain])
run_quiet([SUDO, IPTABLES, "-F", chain])
run_quiet([SUDO, IPTABLES, "-X", chain])