kuhyx/steam-backlog-enforcer
1
0
Fork 0
mirror of https://github.com/kuhyx/steam-backlog-enforcer.git synced 2026-11-10 20:47:04 +01:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Split store_blocker along its one-way call boundaries

Browse Source 
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016A1EYiv7e7HGhET84oa9M3
This commit is contained in:
Krzysztof Rudnicki 2026-08-24 15:06:15 +02:00
parent d8ebbf978b
commit aecfd1a3a0
4 changed files with 317 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
steam_backlog_enforcer/_hosts_protection.py Normal file
View File

@ -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

175
steam_backlog_enforcer/_store_iptables.py Normal file
View File

@ -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,
)

209
steam_backlog_enforcer/store_blocker.py
View File

@ -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

25
steam_backlog_enforcer/tests/test_store_blocker_part2.py
View File

@ -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