From aecfd1a3a03e919535c07f7ffbe67c2e78ff9a01 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 24 Aug 2026 15:06:15 +0200 Subject: [PATCH] Split store_blocker along its one-way call boundaries Extract only functions that call nothing else in the module: the iptables/DNS layer and the /etc/hosts protection helpers. A two-way split here does not work -- block_store calls _reblock_hosts while _unblock_hosts calls is_store_blocked, so cutting between them makes the import circular. Choosing the leaf set instead keeps every import one-way and needed almost no test churn, where the earlier attempt cascaded into a dozen patch-target fixes. 1347 tests, 100% coverage. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016A1EYiv7e7HGhET84oa9M3 --- steam_backlog_enforcer/_hosts_protection.py | 114 ++++++++++ steam_backlog_enforcer/_store_iptables.py | 175 +++++++++++++++ steam_backlog_enforcer/store_blocker.py | 209 +----------------- .../tests/test_store_blocker_part2.py | 25 ++- 4 files changed, 317 insertions(+), 206 deletions(-) create mode 100644 steam_backlog_enforcer/_hosts_protection.py create mode 100644 steam_backlog_enforcer/_store_iptables.py diff --git a/steam_backlog_enforcer/_hosts_protection.py b/steam_backlog_enforcer/_hosts_protection.py new file mode 100644 index 0000000..3bc5653 --- /dev/null +++ b/steam_backlog_enforcer/_hosts_protection.py @@ -0,0 +1,114 @@ +"""Toggling the protected/immutable flag on /etc/hosts around edits. + +Leaf helpers: they call nothing else in the blocker, so extracting them +introduces no cycle. Split to keep both files under the 250-line cap. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +import shutil +import subprocess + +from steam_backlog_enforcer.config import ( + BLOCKED_DOMAINS, + HOSTS_FILE, +) + +logger = logging.getLogger(__name__) + +# Path to the hosts install script. _REPO_ROOT resolves to $HOME (this +# module lives two levels below it); the script itself is in the +# linux_configuration checkout under testsAndMisc, not directly under $HOME. +_REPO_ROOT = Path(__file__).resolve().parents[2] +HOSTS_INSTALL_SCRIPT = ( + _REPO_ROOT + / "testsAndMisc" + / "linux_configuration" + / "scripts" + / "periodic_background" + / "hosts" + / "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 _disable_hosts_protection() -> None: + """Temporarily unlock /etc/hosts so its content can be edited. + + Guard-lib: stop watcher, collapse bind mount, chattr -i. + """ + subprocess.run( + [_SUDO, _GUARDCTL, "file-guard", "pacman-unlock", "hosts"], + capture_output=True, + timeout=10, + check=False, + ) + + +def _enable_hosts_protection() -> None: + """Re-lock /etc/hosts, adopting its current content as the new canonical. + + Guard-lib: chattr +i, reapply bind mount, restart watcher. + """ + subprocess.run( + [_SUDO, _GUARDCTL, "file-guard", "sync", "hosts"], + capture_output=True, + timeout=10, + check=False, + ) + + +def _sudo_write_hosts(content: str) -> None: + """Write *content* to /etc/hosts via ``sudo tee``.""" + subprocess.run( + [_SUDO, _TEE, str(HOSTS_FILE)], + input=content.encode(), + stdout=subprocess.DEVNULL, + timeout=10, + check=True, + ) + + +def _reblock_hosts() -> bool: + """Uncomment Steam Store entries in /etc/hosts.""" + try: + _disable_hosts_protection() + content = HOSTS_FILE.read_text(encoding="utf-8") + new_lines = [] + changed = False + for line in content.splitlines(keepends=True): + stripped = line.strip() + if stripped.startswith("# ") and any( + d in stripped for d in BLOCKED_DOMAINS + ): + # Remove the '# ' prefix. + uncommented = line.replace("# ", "", 1) + new_lines.append(uncommented) + changed = True + else: + new_lines.append(line) + + if changed: + _sudo_write_hosts("".join(new_lines)) + logger.info("Re-enabled Steam Store entries in /etc/hosts.") + + _enable_hosts_protection() + except OSError: + logger.exception("Failed to modify /etc/hosts") + return False + else: + return True diff --git a/steam_backlog_enforcer/_store_iptables.py b/steam_backlog_enforcer/_store_iptables.py new file mode 100644 index 0000000..7729ccc --- /dev/null +++ b/steam_backlog_enforcer/_store_iptables.py @@ -0,0 +1,175 @@ +"""Steam-store blocking at the iptables/DNS layer. + +These call nothing else in the blocker, so they extract cleanly: the +one-way dependency is what keeps store_blocker importable. +Split to keep both files under the 250-line cap. +""" + +from __future__ import annotations + +import contextlib +import logging +from pathlib import Path +import shutil +import socket +import subprocess + +from steam_backlog_enforcer.config import ( + BLOCKED_DOMAINS, +) + +logger = logging.getLogger(__name__) + +# Path to the hosts install script. _REPO_ROOT resolves to $HOME (this +# module lives two levels below it); the script itself is in the +# linux_configuration checkout under testsAndMisc, not directly under $HOME. +_REPO_ROOT = Path(__file__).resolve().parents[2] +HOSTS_INSTALL_SCRIPT = ( + _REPO_ROOT + / "testsAndMisc" + / "linux_configuration" + / "scripts" + / "periodic_background" + / "hosts" + / "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 _is_iptables_blocked() -> bool: + """Check if our iptables chain exists and has rules.""" + try: + result = subprocess.run( + [_SUDO, _IPTABLES, "-L", IPTABLES_CHAIN, "-n"], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return False + else: + return result.returncode == 0 and "DROP" in result.stdout + + +def _block_store_iptables() -> bool: + """Block Steam Store domains using iptables (IP-based).""" + try: + # Create chain if it doesn't exist. + subprocess.run( + [_SUDO, _IPTABLES, "-N", IPTABLES_CHAIN], + capture_output=True, + timeout=5, + check=False, + ) + # Flush existing rules in our chain. + subprocess.run( + [_SUDO, _IPTABLES, "-F", IPTABLES_CHAIN], + capture_output=True, + timeout=5, + check=True, + ) + + # Resolve domains and block their IPs. + blocked_ips: set[str] = set() + for domain in BLOCKED_DOMAINS: + with contextlib.suppress(socket.gaierror): + ips = socket.getaddrinfo(domain, 443, socket.AF_INET) + for _, _, _, _, addr in ips: + blocked_ips.add(addr[0]) + + for ip in blocked_ips: + subprocess.run( + [ + _SUDO, + _IPTABLES, + "-A", + IPTABLES_CHAIN, + "-d", + ip, + "-j", + "DROP", + ], + capture_output=True, + timeout=5, + check=True, + ) + + # Hook our chain into OUTPUT if not already there. + result = subprocess.run( + [_SUDO, _IPTABLES, "-C", "OUTPUT", "-j", IPTABLES_CHAIN], + capture_output=True, + timeout=5, + check=False, + ) + if result.returncode != 0: + subprocess.run( + [_SUDO, _IPTABLES, "-I", "OUTPUT", "-j", IPTABLES_CHAIN], + capture_output=True, + timeout=5, + check=True, + ) + except (OSError, subprocess.SubprocessError): + logger.exception("Failed to block store via iptables") + return False + else: + logger.info("Steam Store blocked via iptables (%d IPs).", len(blocked_ips)) + return True + + +def _unblock_store_iptables() -> bool: + """Remove iptables-based block.""" + try: + subprocess.run( + [_SUDO, _IPTABLES, "-D", "OUTPUT", "-j", IPTABLES_CHAIN], + capture_output=True, + timeout=5, + check=False, + ) + subprocess.run( + [_SUDO, _IPTABLES, "-F", IPTABLES_CHAIN], + capture_output=True, + timeout=5, + check=False, + ) + subprocess.run( + [_SUDO, _IPTABLES, "-X", IPTABLES_CHAIN], + capture_output=True, + timeout=5, + check=False, + ) + except (OSError, subprocess.SubprocessError): + logger.exception("Failed to unblock iptables") + return False + else: + logger.info("Steam Store unblocked from iptables.") + return True + + +def flush_dns_cache() -> None: + """Flush the system DNS cache.""" + commands = [ + ["systemd-resolve", "--flush-caches"], + ["resolvectl", "flush-caches"], + ["nscd", "--invalidate=hosts"], + ] + for cmd in commands: + with contextlib.suppress(FileNotFoundError, OSError): + subprocess.run( + cmd, + capture_output=True, + timeout=5, + check=False, + ) diff --git a/steam_backlog_enforcer/store_blocker.py b/steam_backlog_enforcer/store_blocker.py index 3360691..2cc7901 100644 --- a/steam_backlog_enforcer/store_blocker.py +++ b/steam_backlog_enforcer/store_blocker.py @@ -11,13 +11,23 @@ blocks via iptables. from __future__ import annotations -import contextlib import logging from pathlib import Path import shutil -import socket import subprocess +from steam_backlog_enforcer._hosts_protection import ( + _disable_hosts_protection, + _enable_hosts_protection, + _reblock_hosts, + _sudo_write_hosts, +) +from steam_backlog_enforcer._store_iptables import ( + _block_store_iptables, + _is_iptables_blocked, + _unblock_store_iptables, + flush_dns_cache, +) from steam_backlog_enforcer.config import ( BLOCKED_DOMAINS, HOSTS_FILE, @@ -53,17 +63,6 @@ _TEE = shutil.which("tee") or "/usr/bin/tee" _HOSTS_REDIRECT_IP = ".".join(["0"] * 4) -def _sudo_write_hosts(content: str) -> None: - """Write *content* to /etc/hosts via ``sudo tee``.""" - subprocess.run( - [_SUDO, _TEE, str(HOSTS_FILE)], - input=content.encode(), - stdout=subprocess.DEVNULL, - timeout=10, - check=True, - ) - - def is_store_blocked() -> bool: """Check if Steam Store domains are blocked in /etc/hosts.""" try: @@ -150,87 +149,6 @@ def _block_via_hosts_install() -> bool: return False -def _is_iptables_blocked() -> bool: - """Check if our iptables chain exists and has rules.""" - try: - result = subprocess.run( - [_SUDO, _IPTABLES, "-L", IPTABLES_CHAIN, "-n"], - capture_output=True, - text=True, - timeout=5, - check=False, - ) - except (OSError, subprocess.SubprocessError): - return False - else: - return result.returncode == 0 and "DROP" in result.stdout - - -def _block_store_iptables() -> bool: - """Block Steam Store domains using iptables (IP-based).""" - try: - # Create chain if it doesn't exist. - subprocess.run( - [_SUDO, _IPTABLES, "-N", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=False, - ) - # Flush existing rules in our chain. - subprocess.run( - [_SUDO, _IPTABLES, "-F", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=True, - ) - - # Resolve domains and block their IPs. - blocked_ips: set[str] = set() - for domain in BLOCKED_DOMAINS: - with contextlib.suppress(socket.gaierror): - ips = socket.getaddrinfo(domain, 443, socket.AF_INET) - for _, _, _, _, addr in ips: - blocked_ips.add(addr[0]) - - for ip in blocked_ips: - subprocess.run( - [ - _SUDO, - _IPTABLES, - "-A", - IPTABLES_CHAIN, - "-d", - ip, - "-j", - "DROP", - ], - capture_output=True, - timeout=5, - check=True, - ) - - # Hook our chain into OUTPUT if not already there. - result = subprocess.run( - [_SUDO, _IPTABLES, "-C", "OUTPUT", "-j", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=False, - ) - if result.returncode != 0: - subprocess.run( - [_SUDO, _IPTABLES, "-I", "OUTPUT", "-j", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=True, - ) - except (OSError, subprocess.SubprocessError): - logger.exception("Failed to block store via iptables") - return False - else: - logger.info("Steam Store blocked via iptables (%d IPs).", len(blocked_ips)) - return True - - def unblock_store() -> bool: """Remove Steam Store blocks from both iptables and /etc/hosts.""" ipt_ok = _unblock_store_iptables() @@ -245,52 +163,6 @@ def unblock_store() -> bool: return ipt_ok or hosts_ok -def _unblock_store_iptables() -> bool: - """Remove iptables-based block.""" - try: - subprocess.run( - [_SUDO, _IPTABLES, "-D", "OUTPUT", "-j", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=False, - ) - subprocess.run( - [_SUDO, _IPTABLES, "-F", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=False, - ) - subprocess.run( - [_SUDO, _IPTABLES, "-X", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=False, - ) - except (OSError, subprocess.SubprocessError): - logger.exception("Failed to unblock iptables") - return False - else: - logger.info("Steam Store unblocked from iptables.") - return True - - -def flush_dns_cache() -> None: - """Flush the system DNS cache.""" - commands = [ - ["systemd-resolve", "--flush-caches"], - ["resolvectl", "flush-caches"], - ["nscd", "--invalidate=hosts"], - ] - for cmd in commands: - with contextlib.suppress(FileNotFoundError, OSError): - subprocess.run( - cmd, - capture_output=True, - timeout=5, - check=False, - ) - - # ────────────────────────────────────────────────────────────── # /etc/hosts protection helpers # @@ -307,32 +179,6 @@ def flush_dns_cache() -> None: # ────────────────────────────────────────────────────────────── -def _disable_hosts_protection() -> None: - """Temporarily unlock /etc/hosts so its content can be edited. - - Guard-lib: stop watcher, collapse bind mount, chattr -i. - """ - subprocess.run( - [_SUDO, _GUARDCTL, "file-guard", "pacman-unlock", "hosts"], - capture_output=True, - timeout=10, - check=False, - ) - - -def _enable_hosts_protection() -> None: - """Re-lock /etc/hosts, adopting its current content as the new canonical. - - Guard-lib: chattr +i, reapply bind mount, restart watcher. - """ - subprocess.run( - [_SUDO, _GUARDCTL, "file-guard", "sync", "hosts"], - capture_output=True, - timeout=10, - check=False, - ) - - def _unblock_hosts() -> bool: """Comment out Steam Store entries in /etc/hosts.""" if not is_store_blocked(): @@ -366,34 +212,3 @@ def _unblock_hosts() -> bool: return False else: return True - - -def _reblock_hosts() -> bool: - """Uncomment Steam Store entries in /etc/hosts.""" - try: - _disable_hosts_protection() - content = HOSTS_FILE.read_text(encoding="utf-8") - new_lines = [] - changed = False - for line in content.splitlines(keepends=True): - stripped = line.strip() - if stripped.startswith("# ") and any( - d in stripped for d in BLOCKED_DOMAINS - ): - # Remove the '# ' prefix. - uncommented = line.replace("# ", "", 1) - new_lines.append(uncommented) - changed = True - else: - new_lines.append(line) - - if changed: - _sudo_write_hosts("".join(new_lines)) - logger.info("Re-enabled Steam Store entries in /etc/hosts.") - - _enable_hosts_protection() - except OSError: - logger.exception("Failed to modify /etc/hosts") - return False - else: - return True diff --git a/steam_backlog_enforcer/tests/test_store_blocker_part2.py b/steam_backlog_enforcer/tests/test_store_blocker_part2.py index 6c3f4d4..f1eeb01 100644 --- a/steam_backlog_enforcer/tests/test_store_blocker_part2.py +++ b/steam_backlog_enforcer/tests/test_store_blocker_part2.py @@ -118,10 +118,12 @@ class TestReblockHosts: encoding="utf-8", ) with ( - patch(f"{PKG}.HOSTS_FILE", hosts_file), - patch(f"{PKG}._disable_hosts_protection"), - patch(f"{PKG}._enable_hosts_protection"), - patch(f"{PKG}._sudo_write_hosts") as mock_write, + patch("steam_backlog_enforcer._hosts_protection.HOSTS_FILE", hosts_file), + patch("steam_backlog_enforcer._hosts_protection._disable_hosts_protection"), + patch("steam_backlog_enforcer._hosts_protection._enable_hosts_protection"), + patch( + "steam_backlog_enforcer._hosts_protection._sudo_write_hosts" + ) as mock_write, ): result = _reblock_hosts() assert result is True @@ -134,17 +136,22 @@ class TestReblockHosts: hosts_file = tmp_path / "hosts" hosts_file.write_text("127.0.0.1 localhost\n", encoding="utf-8") with ( - patch(f"{PKG}.HOSTS_FILE", hosts_file), - patch(f"{PKG}._disable_hosts_protection"), - patch(f"{PKG}._enable_hosts_protection"), - patch(f"{PKG}._sudo_write_hosts") as mock_write, + patch("steam_backlog_enforcer._hosts_protection.HOSTS_FILE", hosts_file), + patch("steam_backlog_enforcer._hosts_protection._disable_hosts_protection"), + patch("steam_backlog_enforcer._hosts_protection._enable_hosts_protection"), + patch( + "steam_backlog_enforcer._hosts_protection._sudo_write_hosts" + ) as mock_write, ): result = _reblock_hosts() assert result is True mock_write.assert_not_called() def test_os_error(self) -> None: - with patch(f"{PKG}._disable_hosts_protection", side_effect=OSError): + with patch( + "steam_backlog_enforcer._hosts_protection._disable_hosts_protection", + side_effect=OSError, + ): result = _reblock_hosts() assert result is False