From f1822f154f82e79f41aa447fc19b0ad34d371ba4 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Tue, 10 Nov 2026 15:57:53 +0100 Subject: [PATCH] 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 --- .pre-commit-config.yaml | 4 +- pyproject.toml | 10 +++- steam_backlog_enforcer/_store_iptables.py | 53 +++---------------- steam_backlog_enforcer/_store_tools.py | 43 +++++++++++++++ .../_total_block_iptables.py | 48 ++++------------- 5 files changed, 72 insertions(+), 86 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 93b68ef..09ab164 100644 --- a/.pre-commit-config.yaml +++ b/.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 diff --git a/pyproject.toml b/pyproject.toml index f1fb5b8..f58c19c 100644 --- a/pyproject.toml +++ b/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 diff --git a/steam_backlog_enforcer/_store_iptables.py b/steam_backlog_enforcer/_store_iptables.py index d834441..3694f96 100644 --- a/steam_backlog_enforcer/_store_iptables.py +++ b/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 diff --git a/steam_backlog_enforcer/_store_tools.py b/steam_backlog_enforcer/_store_tools.py index 5de8e9e..22f9a17 100644 --- a/steam_backlog_enforcer/_store_tools.py +++ b/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]) diff --git a/steam_backlog_enforcer/_total_block_iptables.py b/steam_backlog_enforcer/_total_block_iptables.py index 9424667..0736428 100644 --- a/steam_backlog_enforcer/_total_block_iptables.py +++ b/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