diff --git a/steam_backlog_enforcer/_cmd_total_block_refresh.py b/steam_backlog_enforcer/_cmd_total_block_refresh.py new file mode 100644 index 0000000..078341c --- /dev/null +++ b/steam_backlog_enforcer/_cmd_total_block_refresh.py @@ -0,0 +1,50 @@ +"""``total-block-refresh``: apply the current domain list to a live total block. + +The domain list (and the itch.io exception in it) is read at block start, so a +change does not reach an already-active block by itself. This command reconciles +/etc/hosts and the iptables chain to the list without ending the lock or +touching the guard-lib package-block. It is root-only, idempotent, and +fail-closed: the hosts step runs first (so the exempt names resolve again for +the firewall step) and a failure stops the run; nothing here ever adds a +domain to the block's exceptions except TOTAL_BLOCK_EXEMPT_DOMAINS. +""" + +from __future__ import annotations + +import os + +from steam_backlog_enforcer._total_block import is_total_block_active +from steam_backlog_enforcer._total_block_hosts import refresh_total_block_hosts +from steam_backlog_enforcer._total_block_iptables import refresh_total_block_iptables +from steam_backlog_enforcer.game_install import _echo +from steam_backlog_enforcer.store_blocker import flush_dns_cache + + +def cmd_total_block_refresh(args: list[str]) -> int: + """Reconcile the live total block with the current domain list. + + Args: + args: Arguments after the command name (none are accepted). + + Returns: + Process exit code: 0 done or nothing to do, 1 failed, 2 bad usage. + """ + if args: + _echo(f"Unknown argument(s): {' '.join(args)}") + return 2 + if os.geteuid() != 0: + _echo("total-block-refresh requires root (it edits /etc/hosts, iptables).") + return 1 + if not is_total_block_active(): + _echo("No active total block; nothing to refresh.") + return 0 + + if not refresh_total_block_hosts(): + _echo("Failed to refresh /etc/hosts; block left as it was.") + return 1 + flush_dns_cache() + if not refresh_total_block_iptables(): + _echo("Failed to refresh iptables (hosts done); rerun is safe.") + return 1 + _echo("Total block refreshed.") + return 0 diff --git a/steam_backlog_enforcer/_command_gate.py b/steam_backlog_enforcer/_command_gate.py index 39810eb..cb0324b 100644 --- a/steam_backlog_enforcer/_command_gate.py +++ b/steam_backlog_enforcer/_command_gate.py @@ -29,7 +29,9 @@ if TYPE_CHECKING: # Usable before an API key exists: setup creates one, add-exception never # talks to Steam, and serve hosts the web Setup page that asks for the key # (refusing it left a fresh install with no way into the UI's setup). -UNCONFIGURED_OK: Final = frozenset({"setup", "add-exception", "serve"}) +UNCONFIGURED_OK: Final = frozenset( + {"setup", "add-exception", "serve", "total-block-refresh"} +) NOT_CONFIGURED_MSG: Final = "Not configured. Run 'setup' first." diff --git a/steam_backlog_enforcer/_command_help.py b/steam_backlog_enforcer/_command_help.py index eda6c34..3e8b8c5 100644 --- a/steam_backlog_enforcer/_command_help.py +++ b/steam_backlog_enforcer/_command_help.py @@ -43,6 +43,7 @@ EXTRA_COMMAND_DESCRIPTIONS: Final[dict[str, str]] = { "block-gaming": "Block ALL gaming for days, no in-app undo", "enforce": "Run enforcer: block, uninstall, kill, hide (--demo for a 60s budget)", "gaming-unblock": "Force-release playtime bind mounts (root; recovery hatch)", + "total-block-refresh": "Apply the current domain list to a live total block (root)", } ALL_COMMAND_DESCRIPTIONS: Final[dict[str, str]] = ( diff --git a/steam_backlog_enforcer/_command_locks.py b/steam_backlog_enforcer/_command_locks.py index b198d0d..75ee48f 100644 --- a/steam_backlog_enforcer/_command_locks.py +++ b/steam_backlog_enforcer/_command_locks.py @@ -30,6 +30,8 @@ _MANUAL_LOCK_EXEMPT_COMMANDS = frozenset( "gaming-status", "gaming-unblock", "gaming-reset", + # Reconciles the block's domain list; cannot shorten or end a lock. + "total-block-refresh", } ) @@ -44,5 +46,12 @@ _MANUAL_LOCK_EXEMPT_COMMANDS = frozenset( # and refusing it took the web UI down for the whole block, crash-looping its # systemd unit, the first time anything restarted it. _TOTAL_BLOCK_EXEMPT_COMMANDS = frozenset( - {"status", "enforce", "gaming-status", "gaming-unblock", "serve"} + { + "status", + "enforce", + "gaming-status", + "gaming-unblock", + "serve", + "total-block-refresh", + } ) diff --git a/steam_backlog_enforcer/_store_tools.py b/steam_backlog_enforcer/_store_tools.py index 22f9a17..d91477a 100644 --- a/steam_backlog_enforcer/_store_tools.py +++ b/steam_backlog_enforcer/_store_tools.py @@ -87,6 +87,13 @@ def add_drop_rule(chain: str, ip: str) -> None: ) +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. diff --git a/steam_backlog_enforcer/_total_block_domains.py b/steam_backlog_enforcer/_total_block_domains.py index 8620659..b23c9e3 100644 --- a/steam_backlog_enforcer/_total_block_domains.py +++ b/steam_backlog_enforcer/_total_block_domains.py @@ -7,6 +7,8 @@ storefront here should not mean touching blocking code. from __future__ import annotations +from typing import Final + from steam_backlog_enforcer.config import BLOCKED_DOMAINS TOTAL_BLOCK_DOMAINS = [ @@ -21,8 +23,8 @@ TOTAL_BLOCK_DOMAINS = [ "cdn.steamstatic.com", ] -# Browser/flash game sites. Note itch.io overlaps with the "itch" desktop -# app process kill above (web storefront vs. desktop client). +# Browser/flash game sites. itch.io is deliberately absent: see +# TOTAL_BLOCK_EXEMPT_DOMAINS below. GAME_WEBSITE_DOMAINS = [ "newgrounds.com", "armorgames.com", @@ -33,9 +35,27 @@ GAME_WEBSITE_DOMAINS = [ "addictinggames.com", "y8.com", "coolmathgames.com", - "itch.io", ] +# EXCEPTION (kuhy, 2026-11-11: "itch io should be unblocked but only it"): +# a permanent, narrow hole so butler can publish his game and he can manage the +# itch page, while every other game site and Steam stay blocked. Nothing here is +# ever null-routed in /etc/hosts and its IPs are never dropped in iptables. +# Hosts: itch.io + www.itch.io (the page), api.itch.io and static.itch.io (both +# strings in the butler binary), kuhyx.itch.io (his account's page). butler +# uploads go to Google Cloud Storage via a URL the API returns (resumable +# upload), which is not blocked anywhere, so it needs no entry. +# TRADEOFF: itch is on shared Cloudflare IPs, and the iptables layer works per +# IP. When a still-blocked site sits on the same IP, the exception wins for +# that IP; /etc/hosts stays the primary block for the other site. +TOTAL_BLOCK_EXEMPT_DOMAINS: Final = ( + "itch.io", + "www.itch.io", + "api.itch.io", + "static.itch.io", + "kuhyx.itch.io", +) + def _expand_with_www(domains: list[str]) -> list[str]: """Add a ``www.`` variant for each bare second-level domain. @@ -54,9 +74,11 @@ def _expand_with_www(domains: list[str]) -> list[str]: return expanded -_ALL_TOTAL_BLOCK_DOMAINS = _expand_with_www( - [*TOTAL_BLOCK_DOMAINS, *GAME_WEBSITE_DOMAINS] -) +_ALL_TOTAL_BLOCK_DOMAINS = [ + domain + for domain in _expand_with_www([*TOTAL_BLOCK_DOMAINS, *GAME_WEBSITE_DOMAINS]) + if domain not in TOTAL_BLOCK_EXEMPT_DOMAINS +] # The null-route redirect target used to make a blocked domain resolve # nowhere - built from parts rather than the literal so linters don't diff --git a/steam_backlog_enforcer/_total_block_hosts.py b/steam_backlog_enforcer/_total_block_hosts.py index 9cdc594..b0bf95b 100644 --- a/steam_backlog_enforcer/_total_block_hosts.py +++ b/steam_backlog_enforcer/_total_block_hosts.py @@ -28,6 +28,14 @@ _HOSTS_BLOCK_BEGIN = "# BEGIN steam-backlog-enforcer total-block\n" _HOSTS_BLOCK_END = "# END steam-backlog-enforcer total-block\n" +def _expected_block() -> str: + """The enforcer's /etc/hosts section as it should currently read.""" + lines = [_HOSTS_BLOCK_BEGIN] + lines += [f"{NULL_ROUTE_IP} {domain}\n" for domain in _ALL_TOTAL_BLOCK_DOMAINS] + lines.append(_HOSTS_BLOCK_END) + return "".join(lines) + + def apply_total_block_hosts() -> bool: """Append the total-block domain block to /etc/hosts, if not present.""" try: @@ -39,14 +47,8 @@ def apply_total_block_hosts() -> bool: if _HOSTS_BLOCK_BEGIN in content: return True - block_lines = [_HOSTS_BLOCK_BEGIN] - block_lines += [ - f"{NULL_ROUTE_IP} {domain}\n" for domain in _ALL_TOTAL_BLOCK_DOMAINS - ] - block_lines.append(_HOSTS_BLOCK_END) - new_content = content if content.endswith("\n") else content + "\n" - new_content += "".join(block_lines) + new_content += _expected_block() try: _disable_hosts_protection() @@ -59,6 +61,45 @@ def apply_total_block_hosts() -> bool: return True +def refresh_total_block_hosts() -> bool: + """Bring an existing section in line with the current domain list. + + Unlike :func:`apply_total_block_hosts` (which leaves a present section + alone, so the 3 s tick stays cheap), this replaces the section when its + text differs -- how a list change reaches an already-active block. Only the + text between the markers changes; a missing section is installed; a + BEGIN without END, or an unreadable file, is left untouched (False). + """ + try: + content = HOSTS_FILE.read_text(encoding="utf-8") + except OSError: + logger.exception("Failed to read /etc/hosts") + return False + + start = content.find(_HOSTS_BLOCK_BEGIN) + if start == -1: + return apply_total_block_hosts() + end_at = content.find(_HOSTS_BLOCK_END, start) + if end_at == -1: + logger.error("total-block section has no END marker; not touching hosts") + return False + + end = end_at + len(_HOSTS_BLOCK_END) + expected = _expected_block() + if content[start:end] == expected: + return True + + try: + _disable_hosts_protection() + _sudo_write_hosts(content[:start] + expected + content[end:]) + except OSError, subprocess.SubprocessError: + logger.exception("Failed to refresh total-block hosts entries") + return False + finally: + _enable_hosts_protection() + return True + + def remove_total_block_hosts() -> bool: """Remove the total-block domain block from /etc/hosts, if present.""" try: diff --git a/steam_backlog_enforcer/_total_block_iptables.py b/steam_backlog_enforcer/_total_block_iptables.py index 0736428..3fc14bf 100644 --- a/steam_backlog_enforcer/_total_block_iptables.py +++ b/steam_backlog_enforcer/_total_block_iptables.py @@ -10,6 +10,7 @@ their real addresses and could not be re-derived. from __future__ import annotations import contextlib +import ipaddress import json import logging import socket @@ -20,12 +21,14 @@ from steam_backlog_enforcer._store_tools import ( SUDO, add_drop_rule, create_fresh_chain, + delete_drop_rule, drop_chain, ensure_chain_in_output, ) from steam_backlog_enforcer._total_block_domains import ( _ALL_TOTAL_BLOCK_DOMAINS, NULL_ROUTE_IP, + TOTAL_BLOCK_EXEMPT_DOMAINS, ) from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write @@ -54,14 +57,8 @@ def _save_cached_ips(ips: set[str]) -> None: _atomic_write(_IPTABLES_IP_CACHE_FILE, json.dumps(sorted(ips)) + "\n") -def _iptables_chain_intact(expected_ips: set[str]) -> bool: - """Cheap check for whether the chain and its OUTPUT hook are intact. - - One `-S` + one `-C` call (two forks), versus the ~30 forks a full - rebuild costs - this is what keeps :func:`_apply_total_block_iptables` - from re-resolving DNS and re-forking a subprocess per IP on every - 3-second enforce tick. - """ +def _chain_ips() -> set[str] | None: + """IPs the chain currently drops, or None when the chain does not exist.""" listing = subprocess.run( [SUDO, IPTABLES, "-S", IPTABLES_CHAIN], capture_output=True, @@ -70,7 +67,7 @@ def _iptables_chain_intact(expected_ips: set[str]) -> bool: check=False, ) if listing.returncode != 0: - return False + return None current_ips: set[str] = set() for line in listing.stdout.splitlines(): @@ -79,8 +76,29 @@ def _iptables_chain_intact(expected_ips: set[str]) -> bool: idx = parts.index("-d") if idx + 1 < len(parts): current_ips.add(parts[idx + 1].split("/")[0]) + return current_ips - if not expected_ips.issubset(current_ips): + +def _resolve_exempt_ips() -> set[str]: + """Real public IPs of the exempt domains (null-routed answers do not count).""" + ips: set[str] = set() + for domain in TOTAL_BLOCK_EXEMPT_DOMAINS: + with contextlib.suppress(OSError): + for _, _, _, _, addr in socket.getaddrinfo(domain, 443, socket.AF_INET): + ips.add(str(addr[0])) + return {ip for ip in ips if ipaddress.ip_address(ip).is_global} + + +def _iptables_chain_intact(expected_ips: set[str]) -> bool: + """Cheap check for whether the chain and its OUTPUT hook are intact. + + One `-S` + one `-C` call (two forks), versus the ~30 forks a full + rebuild costs - this is what keeps :func:`_apply_total_block_iptables` + from re-resolving DNS and re-forking a subprocess per IP on every + 3-second enforce tick. + """ + current_ips = _chain_ips() + if current_ips is None or not expected_ips.issubset(current_ips): return False hook = subprocess.run( @@ -125,7 +143,8 @@ def apply_total_block_iptables() -> bool: for _, _, _, _, addr in socket.getaddrinfo(domain, 443, socket.AF_INET): resolved_ips.add(str(addr[0])) - blocked_ips = (cached | resolved_ips) - {NULL_ROUTE_IP} + # The itch exception wins over a shared IP (TOTAL_BLOCK_EXEMPT_DOMAINS). + blocked_ips = (cached | resolved_ips) - {NULL_ROUTE_IP} - _resolve_exempt_ips() _save_cached_ips(blocked_ips) for ip in blocked_ips: @@ -139,6 +158,42 @@ def apply_total_block_iptables() -> bool: return True +def refresh_total_block_iptables() -> bool: + """Release the exempt domains' IPs from a live chain, never opening a gap. + + Rules are deleted one by one (no flush), so every other IP stays dropped + throughout. The cache is saved first: a half-done run then still looks + intact to the 3 s tick (cache is a subset of the chain) and the next run + finishes the deletes. Fails closed: when the exempt names resolve to no + public IP (hosts still null-route them, or DNS is down) nothing changes. + A missing chain is rebuilt by :func:`apply_total_block_iptables`. + """ + exempt = _resolve_exempt_ips() + if not exempt: + logger.error("Exempt domains resolve to no public IP; iptables unchanged") + return False + try: + in_chain = _chain_ips() + cached = _load_cached_ips() + if in_chain is None: + _save_cached_ips(cached - exempt) + return apply_total_block_iptables() + if not (cached | in_chain) & exempt: + return True + _save_cached_ips(cached - exempt) + failed = [ + ip for ip in in_chain & exempt if not delete_drop_rule(IPTABLES_CHAIN, ip) + ] + except OSError, subprocess.SubprocessError: + logger.exception("Failed to refresh total-block iptables rules") + return False + if failed: + logger.error("Could not delete iptables rule(s) for %s", sorted(failed)) + return False + logger.info("Total block: released %d exempt IP(s).", len(exempt)) + return True + + def remove_total_block_iptables() -> bool: """Remove the total-block iptables chain and its OUTPUT hook.""" try: diff --git a/steam_backlog_enforcer/main/__init__.py b/steam_backlog_enforcer/main/__init__.py index 39ee2f5..87f5163 100644 --- a/steam_backlog_enforcer/main/__init__.py +++ b/steam_backlog_enforcer/main/__init__.py @@ -23,6 +23,7 @@ from steam_backlog_enforcer._cmd_playtime import ( cmd_gaming_status, cmd_gaming_unblock, ) +from steam_backlog_enforcer._cmd_total_block_refresh import cmd_total_block_refresh from steam_backlog_enforcer._command_help import ( COMMAND_DESCRIPTIONS, EXTRA_COMMAND_DESCRIPTIONS, @@ -182,6 +183,8 @@ def _dispatch_extra_command(command: str, config: Config, state: State) -> bool: sys.exit(cmd_enforce(config, state, args)) if command == "gaming-unblock": sys.exit(cmd_gaming_unblock(args)) + if command == "total-block-refresh": + sys.exit(cmd_total_block_refresh(args)) handlers: dict[str, Callable[[], object]] = { "add-exception": lambda: cmd_add_exception(args), "unblock": lambda: cmd_unblock(config, state, args), diff --git a/steam_backlog_enforcer/tests/test_cmd_total_block_refresh.py b/steam_backlog_enforcer/tests/test_cmd_total_block_refresh.py new file mode 100644 index 0000000..0b961e4 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_cmd_total_block_refresh.py @@ -0,0 +1,100 @@ +"""The `total-block-refresh` command: root-only, lock-exempt, fail-closed.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer import _cmd_total_block_refresh as cmd +from steam_backlog_enforcer._command_gate import UNCONFIGURED_OK +from steam_backlog_enforcer._command_help import ALL_COMMAND_DESCRIPTIONS +from steam_backlog_enforcer._command_locks import ( + _MANUAL_LOCK_EXEMPT_COMMANDS, + _TOTAL_BLOCK_EXEMPT_COMMANDS, +) + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + +PKG = "steam_backlog_enforcer._cmd_total_block_refresh" +NAME = "total-block-refresh" + + +def _step( + calls: list[str], name: str, *, result: bool | None +) -> Callable[[], bool | None]: + """A stand-in step that logs its name and returns *result*.""" + + def run() -> bool | None: + calls.append(name) + return result + + return run + + +@pytest.fixture(name="steps") +def _steps() -> Iterator[list[str]]: + """Root, active block, every step succeeding; yields the call log.""" + calls: list[str] = [] + with ( + patch(f"{PKG}.os.geteuid", return_value=0), + patch(f"{PKG}.is_total_block_active", return_value=True), + patch(f"{PKG}.refresh_total_block_hosts", _step(calls, "hosts", result=True)), + patch(f"{PKG}.flush_dns_cache", _step(calls, "flush", result=None)), + patch(f"{PKG}.refresh_total_block_iptables", _step(calls, "ipt", result=True)), + ): + yield calls + + +class TestCmd: + """Order, gating and failure behaviour.""" + + def test_runs_hosts_then_flush_then_iptables(self, steps: list[str]) -> None: + """Test that hosts unblock first so exempt names resolve for iptables.""" + assert cmd.cmd_total_block_refresh([]) == 0 + assert steps == ["hosts", "flush", "ipt"] + + def test_requires_root(self, steps: list[str]) -> None: + """Test that non-root exits 1 and touches nothing.""" + with patch(f"{PKG}.os.geteuid", return_value=1000): + assert cmd.cmd_total_block_refresh([]) == 1 + assert steps == [] + + def test_no_active_block_is_noop(self, steps: list[str]) -> None: + """Test that without a lock there is nothing to reconcile.""" + with patch(f"{PKG}.is_total_block_active", return_value=False): + assert cmd.cmd_total_block_refresh([]) == 0 + assert steps == [] + + def test_unknown_args_rejected(self, steps: list[str]) -> None: + """Test that stray arguments are an error.""" + assert cmd.cmd_total_block_refresh(["--now"]) == 2 + assert steps == [] + + def test_hosts_failure_stops_before_iptables(self, steps: list[str]) -> None: + """Test that a failed hosts step never reaches the firewall.""" + with patch(f"{PKG}.refresh_total_block_hosts", return_value=False): + assert cmd.cmd_total_block_refresh([]) == 1 + assert steps == [] + + def test_iptables_failure_exits_nonzero(self, steps: list[str]) -> None: + """Test that an iptables failure is reported.""" + with patch(f"{PKG}.refresh_total_block_iptables", return_value=False): + assert cmd.cmd_total_block_refresh([]) == 1 + assert steps == ["hosts", "flush"] + + +class TestRegistration: + """The command is reachable during the lock it reconciles.""" + + def test_lock_exempt_and_unconfigured_ok(self) -> None: + """Test that neither lock, nor a missing API key, refuses it.""" + assert NAME in _TOTAL_BLOCK_EXEMPT_COMMANDS + assert NAME in _MANUAL_LOCK_EXEMPT_COMMANDS + assert NAME in UNCONFIGURED_OK + + def test_has_help_text(self) -> None: + """Test that help lists it.""" + assert NAME in ALL_COMMAND_DESCRIPTIONS diff --git a/steam_backlog_enforcer/tests/test_itch_exception_hosts.py b/steam_backlog_enforcer/tests/test_itch_exception_hosts.py new file mode 100644 index 0000000..8624617 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_itch_exception_hosts.py @@ -0,0 +1,166 @@ +"""itch.io exception (kuhy, 2026-11-11): domain table and /etc/hosts half.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer._total_block_domains import ( + _ALL_TOTAL_BLOCK_DOMAINS, + GAME_WEBSITE_DOMAINS, + TOTAL_BLOCK_DOMAINS, + TOTAL_BLOCK_EXEMPT_DOMAINS, +) +from steam_backlog_enforcer._total_block_hosts import ( + _HOSTS_BLOCK_BEGIN, + _HOSTS_BLOCK_END, + apply_total_block_hosts, + refresh_total_block_hosts, +) + +if TYPE_CHECKING: + from steam_backlog_enforcer.tests._total_block_paths import Paths + +PKG = "steam_backlog_enforcer._total_block_hosts" +STILL_BLOCKED = ( + "store.steampowered.com", + "steamcommunity.com", + "newgrounds.com", + "www.newgrounds.com", + "crazygames.com", + "poki.com", + "www.y8.com", + "coolmathgames.com", +) +# What the live /etc/hosts section looked like before the exception. +OLD_SECTION = ( + _HOSTS_BLOCK_BEGIN + + "".join(f"0.0.0.0 {d}\n" for d in (*STILL_BLOCKED, "itch.io", "www.itch.io")) + + _HOSTS_BLOCK_END +) + + +class _Hosts: + """Captures what the refresh would write, then plays it back as the file.""" + + def __init__(self, paths: Paths) -> None: + self.paths = paths + self.writes: list[str] = [] + + def write(self, content: str) -> None: + """Stand-in for the sudo-tee write.""" + self.writes.append(content) + self.paths.hosts_file.write_text(content, encoding="utf-8") + + +@pytest.fixture(name="hosts") +def _hosts_fixture(total_block_paths: Paths) -> _Hosts: + """A tmp hosts file holding the pre-exception section between real lines.""" + total_block_paths.hosts_file.write_text( + "127.0.0.1 localhost\n" + OLD_SECTION + "0.0.0.0 evil.example\n", + encoding="utf-8", + ) + return _Hosts(total_block_paths) + + +def _refresh(hosts: _Hosts) -> bool: + with ( + patch(f"{PKG}._disable_hosts_protection"), + patch(f"{PKG}._enable_hosts_protection"), + patch(f"{PKG}._sudo_write_hosts", side_effect=hosts.write), + ): + return refresh_total_block_hosts() + + +class TestExemptTable: + """The exception is one named constant the block lists cannot contain.""" + + def test_confirmed_hosts_are_exempt(self) -> None: + """Test that itch.io and the hosts butler talks to are exempt.""" + for host in ("itch.io", "www.itch.io", "api.itch.io"): + assert host in TOTAL_BLOCK_EXEMPT_DOMAINS + + def test_exempt_never_in_block_list(self) -> None: + """Test that no exempt domain is in the computed block list.""" + assert not set(TOTAL_BLOCK_EXEMPT_DOMAINS) & set(_ALL_TOTAL_BLOCK_DOMAINS) + + def test_exempt_wins_even_if_readded_to_source_lists(self) -> None: + """Test that the filter, not the source list, enforces the exception.""" + assert "itch.io" not in GAME_WEBSITE_DOMAINS + assert "itch.io" not in TOTAL_BLOCK_DOMAINS + + @pytest.mark.parametrize("domain", STILL_BLOCKED) + def test_other_sites_still_blocked(self, domain: str) -> None: + """Test that every other game site and Steam stay in the block list.""" + assert domain in _ALL_TOTAL_BLOCK_DOMAINS + + +class TestApplyOmitsItch: + """A fresh block never writes itch.io.""" + + def test_fresh_block_has_no_itch_line(self, total_block_paths: Paths) -> None: + """Test that apply on a clean file writes the others but not itch.io.""" + total_block_paths.hosts_file.write_text("127.0.0.1 localhost\n") + with ( + patch(f"{PKG}._disable_hosts_protection"), + patch(f"{PKG}._enable_hosts_protection"), + patch(f"{PKG}._sudo_write_hosts") as write, + ): + assert apply_total_block_hosts() is True + written = write.call_args.args[0] + assert "itch.io" not in written + assert "0.0.0.0 newgrounds.com\n" in written + + +class TestRefreshHosts: + """refresh rewrites only the enforcer section of a live block.""" + + def test_removes_itch_keeps_rest(self, hosts: _Hosts) -> None: + """Test that itch lines go and every other line is untouched.""" + assert _refresh(hosts) is True + new = hosts.paths.hosts_file.read_text(encoding="utf-8") + assert "itch.io" not in new + assert new.startswith("127.0.0.1 localhost\n" + _HOSTS_BLOCK_BEGIN) + assert new.endswith(_HOSTS_BLOCK_END + "0.0.0.0 evil.example\n") + for domain in STILL_BLOCKED: + assert f"0.0.0.0 {domain}\n" in new + + def test_idempotent(self, hosts: _Hosts) -> None: + """Test that a second refresh writes nothing.""" + assert _refresh(hosts) is True + assert _refresh(hosts) is True + assert len(hosts.writes) == 1 + + def test_no_section_applies_block(self, total_block_paths: Paths) -> None: + """Test that refresh on an active block with no section installs it.""" + total_block_paths.hosts_file.write_text("127.0.0.1 localhost\n") + h = _Hosts(total_block_paths) + assert _refresh(h) is True + assert "newgrounds.com" in h.writes[0] + + def test_broken_section_fails_closed(self, total_block_paths: Paths) -> None: + """Test that a BEGIN without END is left alone and reported.""" + total_block_paths.hosts_file.write_text(_HOSTS_BLOCK_BEGIN + "0.0.0.0 x\n") + h = _Hosts(total_block_paths) + assert _refresh(h) is False + assert h.writes == [] + + def test_unreadable_hosts_fails_closed(self, hosts: _Hosts) -> None: + """Test that an unreadable hosts file is reported, not written.""" + hosts.paths.hosts_file.unlink() + assert _refresh(hosts) is False + assert hosts.writes == [] + + def test_write_error_relocks_and_fails(self, hosts: _Hosts) -> None: + """Test that a failed write returns False and re-enables protection.""" + before = hosts.paths.hosts_file.read_text(encoding="utf-8") + with ( + patch(f"{PKG}._disable_hosts_protection"), + patch(f"{PKG}._enable_hosts_protection") as enable, + patch(f"{PKG}._sudo_write_hosts", side_effect=OSError), + ): + assert refresh_total_block_hosts() is False + enable.assert_called_once() + assert hosts.paths.hosts_file.read_text(encoding="utf-8") == before diff --git a/steam_backlog_enforcer/tests/test_itch_exception_iptables.py b/steam_backlog_enforcer/tests/test_itch_exception_iptables.py new file mode 100644 index 0000000..f2bf2c4 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_itch_exception_iptables.py @@ -0,0 +1,204 @@ +"""itch.io exception (kuhy, 2026-11-11): iptables half, against a fake chain.""" + +from __future__ import annotations + +import socket +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from steam_backlog_enforcer._total_block_domains import NULL_ROUTE_IP +from steam_backlog_enforcer._total_block_iptables import ( + _load_cached_ips, + _resolve_exempt_ips, + _save_cached_ips, + apply_total_block_iptables, + refresh_total_block_iptables, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + +PKG = "steam_backlog_enforcer._total_block_iptables" +ITCH_IP = "104.26.9.198" # Cloudflare, shared by itch.io and api.itch.io +STEAM_IP = "23.1.1.1" +OTHER_CF_IP = "172.67.1.1" # a still-blocked site + + +class FakeIptables: + """A fake `iptables` that keeps one chain's rules in a list.""" + + def __init__(self, ips: list[str], *, chain_exists: bool = True) -> None: + self.rules = list(ips) + self.chain_exists = chain_exists + self.calls: list[list[str]] = [] + + def run(self, cmd: list[str], **_kw: object) -> MagicMock: + """Answer one iptables call against the fake chain.""" + self.calls.append(cmd) + args = ( + cmd[cmd.index("/usr/sbin/iptables") + 1 :] + if "/usr/sbin/iptables" in cmd + else cmd[2:] + ) + verb = args[0] + if verb == "-S": + if not self.chain_exists: + return MagicMock(returncode=1, stdout="") + lines = ["-N STEAM_TOTAL_BLOCK"] + [ + f"-A STEAM_TOTAL_BLOCK -d {ip}/32 -j DROP" for ip in self.rules + ] + return MagicMock(returncode=0, stdout="\n".join(lines) + "\n") + if verb == "-N": + self.chain_exists = True + elif verb == "-F": + self.rules.clear() + elif verb == "-A": + self.rules.append(args[args.index("-d") + 1]) + elif verb == "-D" and args[1] != "OUTPUT": + ip = args[args.index("-d") + 1].split("/")[0] + if ip not in self.rules: + return MagicMock(returncode=1) + self.rules.remove(ip) + return MagicMock(returncode=0) + + def verbs(self) -> list[str]: + """The iptables verb (-S, -D, ...) of every call so far.""" + return [ + next(a for a in c if a.startswith("-") and len(a) == 2) for c in self.calls + ] + + +def _dns(mapping: dict[str, list[str]]) -> Callable[..., list[tuple[object, ...]]]: + def getaddrinfo(host: str, *_a: object, **_k: object) -> list[tuple[object, ...]]: + """A getaddrinfo that answers from *mapping*, NXDOMAIN otherwise.""" + if host not in mapping: + raise socket.gaierror + return [(None, None, None, None, (ip, 443)) for ip in mapping[host]] + + return getaddrinfo + + +ITCH_DNS = {h: [ITCH_IP] for h in ("itch.io", "www.itch.io", "api.itch.io")} + + +def _run(fake: FakeIptables, dns: dict[str, list[str]]) -> bool: + with ( + patch(f"{PKG}.subprocess.run", side_effect=fake.run), + patch(f"{PKG}.socket.getaddrinfo", side_effect=_dns(dns)), + ): + return refresh_total_block_iptables() + + +class TestRefreshIptables: + """Reconciling a live chain to the exception, without a gap.""" + + def test_removes_itch_ip_keeps_rest(self) -> None: + """Test that the shared Cloudflare IP leaves chain and cache.""" + _save_cached_ips({ITCH_IP, STEAM_IP, OTHER_CF_IP}) + fake = FakeIptables([ITCH_IP, STEAM_IP, OTHER_CF_IP]) + assert _run(fake, ITCH_DNS) is True + assert sorted(fake.rules) == sorted([STEAM_IP, OTHER_CF_IP]) + assert _load_cached_ips() == {STEAM_IP, OTHER_CF_IP} + + def test_never_flushes_the_chain(self) -> None: + """Test that other IPs stay dropped throughout (no -F, no -X).""" + _save_cached_ips({ITCH_IP, STEAM_IP}) + fake = FakeIptables([ITCH_IP, STEAM_IP]) + _run(fake, ITCH_DNS) + assert "-F" not in fake.verbs() + assert "-X" not in fake.verbs() + + def test_idempotent(self) -> None: + """Test that a second refresh changes nothing and runs no -D.""" + _save_cached_ips({ITCH_IP, STEAM_IP}) + fake = FakeIptables([ITCH_IP, STEAM_IP]) + assert _run(fake, ITCH_DNS) is True + fake.calls.clear() + assert _run(fake, ITCH_DNS) is True + assert "-D" not in fake.verbs() + assert fake.rules == [STEAM_IP] + + def test_exception_wins_on_shared_ip(self) -> None: + """Test that an IP a blocked site also uses is still released.""" + _save_cached_ips({ITCH_IP}) + fake = FakeIptables([ITCH_IP]) + dns = {**ITCH_DNS, "newgrounds.com": [ITCH_IP]} + assert _run(fake, dns) is True + assert fake.rules == [] + + def test_resumes_after_cache_was_saved_but_chain_not(self) -> None: + """Test that a chain still holding the IP is cleaned although cache is not.""" + _save_cached_ips({STEAM_IP}) + fake = FakeIptables([ITCH_IP, STEAM_IP]) + assert _run(fake, ITCH_DNS) is True + assert fake.rules == [STEAM_IP] + + def test_unresolvable_exempt_hosts_change_nothing(self) -> None: + """Test that no DNS answer leaves cache and chain exactly as they were.""" + _save_cached_ips({ITCH_IP, STEAM_IP}) + fake = FakeIptables([ITCH_IP, STEAM_IP]) + assert _run(fake, {}) is False + assert _run(fake, {"itch.io": [NULL_ROUTE_IP]}) is False # still null-routed + assert sorted(fake.rules) == sorted([ITCH_IP, STEAM_IP]) + assert _load_cached_ips() == {ITCH_IP, STEAM_IP} + + def test_delete_failure_reports_false(self) -> None: + """Test that a rule that cannot be deleted is reported.""" + _save_cached_ips({ITCH_IP}) + fake = FakeIptables([ITCH_IP]) + real = fake.run + + def failing(cmd: list[str], **kw: object) -> MagicMock: + if "-D" in cmd: + return MagicMock(returncode=1) + return real(cmd, **kw) + + with ( + patch(f"{PKG}.subprocess.run", side_effect=failing), + patch(f"{PKG}.socket.getaddrinfo", side_effect=_dns(ITCH_DNS)), + ): + assert refresh_total_block_iptables() is False + + def test_subprocess_error_reports_false(self) -> None: + """Test that an iptables that cannot run fails closed.""" + _save_cached_ips({ITCH_IP}) + with ( + patch(f"{PKG}.subprocess.run", side_effect=OSError), + patch(f"{PKG}.socket.getaddrinfo", side_effect=_dns(ITCH_DNS)), + ): + assert refresh_total_block_iptables() is False + + def test_missing_chain_is_rebuilt_without_itch(self) -> None: + """Test that a vanished chain comes back, minus the itch IP.""" + _save_cached_ips({ITCH_IP, STEAM_IP}) + fake = FakeIptables([], chain_exists=False) + dns = {**ITCH_DNS, "store.steampowered.com": [STEAM_IP]} + assert _run(fake, dns) is True + assert fake.rules == [STEAM_IP] + + +class TestApplyExcludesItch: + """A first build or rebuild never drops the exempt IPs.""" + + def test_rebuild_skips_exempt_ip(self) -> None: + """Test that a blocked site sharing itch's IP does not re-block it.""" + fake = FakeIptables([], chain_exists=False) + dns = {**ITCH_DNS, "newgrounds.com": [ITCH_IP, OTHER_CF_IP]} + with ( + patch(f"{PKG}.subprocess.run", side_effect=fake.run), + patch(f"{PKG}.socket.getaddrinfo", side_effect=_dns(dns)), + ): + assert apply_total_block_iptables() is True + assert fake.rules == [OTHER_CF_IP] + assert ITCH_IP not in _load_cached_ips() + + +class TestResolveExempt: + """Only real, public answers count as an exempt IP.""" + + def test_filters_null_and_private(self) -> None: + """Test that 0.0.0.0, loopback and LAN answers are ignored.""" + dns = {"itch.io": [NULL_ROUTE_IP, "127.0.0.1", "192.168.1.1", ITCH_IP]} + with patch(f"{PKG}.socket.getaddrinfo", side_effect=_dns(dns)): + assert _resolve_exempt_ips() == {ITCH_IP} diff --git a/steam_backlog_enforcer/tests/test_main_shared.py b/steam_backlog_enforcer/tests/test_main_shared.py index 1774f30..2088891 100644 --- a/steam_backlog_enforcer/tests/test_main_shared.py +++ b/steam_backlog_enforcer/tests/test_main_shared.py @@ -233,11 +233,11 @@ class TestEnforceTotalBlockLock: total block's own `pacman -R steam` fail EBUSY - it has to stay reachable exactly when the two collide. gaming-reset is deliberately absent: it would shorten enforcement. serve changes nothing itself - (each job it starts is lock-checked) and refusing it took the web UI - down for the whole block. + (each job it starts is lock-checked); refusing it took the web UI down. """ assert ( frozenset({"status", "enforce", "gaming-status", "gaming-unblock", "serve"}) + | {"total-block-refresh"} # reconciles the list; cannot shorten the lock == _TOTAL_BLOCK_EXEMPT_COMMANDS ) assert "gaming-reset" not in _TOTAL_BLOCK_EXEMPT_COMMANDS diff --git a/steam_backlog_enforcer/tests/test_total_block_iptables.py b/steam_backlog_enforcer/tests/test_total_block_iptables.py index e2c71f7..96641a1 100644 --- a/steam_backlog_enforcer/tests/test_total_block_iptables.py +++ b/steam_backlog_enforcer/tests/test_total_block_iptables.py @@ -7,6 +7,7 @@ import socket from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch +from steam_backlog_enforcer._total_block_domains import TOTAL_BLOCK_EXEMPT_DOMAINS from steam_backlog_enforcer._total_block_iptables import ( _iptables_chain_intact, _load_cached_ips, @@ -122,12 +123,15 @@ class TestApplyTotalBlockIptables: def test_rebuilds_when_not_intact(self) -> None: """Test that rebuilds when not intact.""" + + def resolve(host: str, *_a: object, **_k: object) -> list[tuple[object, ...]]: + if host in TOTAL_BLOCK_EXEMPT_DOMAINS: + raise OSError + return [(None, None, None, None, ("9.9.9.9", 443))] + with ( patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0)), - patch( - f"{PKG}.socket.getaddrinfo", - return_value=[(None, None, None, None, ("9.9.9.9", 443))], - ), + patch(f"{PKG}.socket.getaddrinfo", side_effect=resolve), ): assert apply_total_block_iptables() is True assert "9.9.9.9" in _load_cached_ips()