Dedupe iptables chain code; pylint jobs=1

Set [tool.pylint.similarities] min-similarity-lines = 8 (as diet-guard)
and refactor the remaining R0801 hits: the store and total-block
iptables modules now share create_fresh_chain / add_drop_rule /
drop_chain in _store_tools. No duplicate-code disable or exclude.

Run pylint serially (pyproject jobs = 1, hook --jobs=1): parallel mode
pickles PyLinter with dill, which crashes on Python 3.15 (co_lnotab).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.pre-commit-config.yaml

@@ -128,7 +128,9 @@ repos:
         args:
           - --rcfile=pyproject.toml
           - --fail-under=8.0
-          - --jobs=4
+          # Serial: parallel pylint pickles PyLinter with dill, which crashes on
+          # Python 3.15 (removed co_lnotab). Same reason as jobs = 1 in pyproject.
+          - --jobs=1
         # pylint resolves imports in its OWN isolated env, so anything the
         # package imports has to be listed here or it reports import-error and
         # drags the score under --fail-under. `mcp` is a real runtime import

pyproject.toml

@@ -132,7 +132,10 @@ exclude = [".venv/"]
 init-hook = "import sys; sys.path.insert(0, '.')"
 analyse-fallback-blocks = true
 persistent = true
-jobs = 0
+# Serial on purpose: with jobs != 1 pylint ships the linter to its workers via
+# dill.dumps(PyLinter), which crashes on Python 3.15 (dill does not handle the
+# removed code.co_lnotab yet). Restore parallel runs once dill supports 3.15.
+jobs = 1
 py-version = "3.14"
 ignore = [".venv", "__pycache__"]
 # Test suites intentionally use patterns (protected-access, fixture-arg
@@ -166,6 +169,11 @@ disable = [
     "use-implicit-booleaness-not-comparison-to-zero",
 ]
 
+[tool.pylint.similarities]
+# Same duplicate-code threshold as diet-guard: shorter matches are mostly
+# import blocks and subprocess.run keyword lists, not real copy-paste.
+min-similarity-lines = 8
+
 [tool.pylint.design]
 min-public-methods = 0
 max-module-lines = 250

steam_backlog_enforcer/_store_iptables.py

@@ -16,6 +16,9 @@ from steam_backlog_enforcer._store_tools import (
     IPTABLES,
     IPTABLES_CHAIN,
     SUDO,
+    add_drop_rule,
+    create_fresh_chain,
+    drop_chain,
     ensure_chain_in_output,
     run_quiet,
 )
@@ -44,20 +47,7 @@ def _is_iptables_blocked() -> bool:
 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,
-        )
+        create_fresh_chain(IPTABLES_CHAIN)
 
         # Resolve domains and block their IPs.
         blocked_ips: set[str] = set()
@@ -68,21 +58,7 @@ def _block_store_iptables() -> bool:
                     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,
-            )
+            add_drop_rule(IPTABLES_CHAIN, ip)
 
         ensure_chain_in_output(IPTABLES_CHAIN)
     except OSError, subprocess.SubprocessError:
@@ -95,24 +71,7 @@ def _block_store_iptables() -> bool:
 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,
-        )
+        drop_chain(IPTABLES_CHAIN)
     except OSError, subprocess.SubprocessError:
         logger.exception("Failed to unblock iptables")
         return False

steam_backlog_enforcer/_store_tools.py

@@ -53,3 +53,46 @@ def ensure_chain_in_output(chain: str) -> None:
             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 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])

steam_backlog_enforcer/_total_block_iptables.py

@@ -15,7 +15,14 @@ import logging
 import socket
 import subprocess
 
-from steam_backlog_enforcer._store_tools import IPTABLES, SUDO, ensure_chain_in_output
+from steam_backlog_enforcer._store_tools import (
+    IPTABLES,
+    SUDO,
+    add_drop_rule,
+    create_fresh_chain,
+    drop_chain,
+    ensure_chain_in_output,
+)
 from steam_backlog_enforcer._total_block_domains import (
     _ALL_TOTAL_BLOCK_DOMAINS,
     NULL_ROUTE_IP,
@@ -111,18 +118,7 @@ def apply_total_block_iptables() -> bool:
 
     resolved_ips: set[str] = set()
     try:
-        subprocess.run(
-            [SUDO, IPTABLES, "-N", IPTABLES_CHAIN],
-            capture_output=True,
-            timeout=5,
-            check=False,
-        )
-        subprocess.run(
-            [SUDO, IPTABLES, "-F", IPTABLES_CHAIN],
-            capture_output=True,
-            timeout=5,
-            check=True,
-        )
+        create_fresh_chain(IPTABLES_CHAIN)
 
         for domain in _ALL_TOTAL_BLOCK_DOMAINS:
             with contextlib.suppress(socket.gaierror):
@@ -133,12 +129,7 @@ def apply_total_block_iptables() -> bool:
         _save_cached_ips(blocked_ips)
 
         for ip in blocked_ips:
-            subprocess.run(
-                [SUDO, IPTABLES, "-A", IPTABLES_CHAIN, "-d", ip, "-j", "DROP"],
-                capture_output=True,
-                timeout=5,
-                check=True,
-            )
+            add_drop_rule(IPTABLES_CHAIN, ip)
 
         ensure_chain_in_output(IPTABLES_CHAIN)
     except OSError, subprocess.SubprocessError:
@@ -151,24 +142,7 @@ def apply_total_block_iptables() -> bool:
 def remove_total_block_iptables() -> bool:
     """Remove the total-block iptables chain and its OUTPUT hook."""
     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,
-        )
+        drop_chain(IPTABLES_CHAIN)
     except OSError, subprocess.SubprocessError:
         logger.exception("Failed to remove total-block iptables rules")
         return False