mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-10 19:27:04 +01:00
Krzysztof kuhy Rudnicki
f1822f154f
Some checks are pending
dependency freshness / dependencies on newest stable (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (mypy, mypy, ) (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (pylint, pylint, ) (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (rest, , mypy,pylint) (push) Waiting to run
Tests / test (push) Waiting to run
Tests / web-coverage (push) Waiting to run
repo-contract / repo contract (push) Waiting to run
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>
99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""The executables and paths every store-blocking layer shells out to.
|
|
|
|
One copy, imported by the hosts, iptables and orchestration modules alike --
|
|
each used to carry its own after the 250-line split, three blocks that had
|
|
to be kept identical by hand.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
|
|
# Path to the hosts install script. _SRC_ROOT is ~/src (this module lives two
|
|
# levels below it); the generator moved out of testsAndMisc into its own
|
|
# hosts-blocker repo, whose custom_entries.hosts carries the Steam Store block.
|
|
_SRC_ROOT = Path(__file__).resolve().parents[2]
|
|
HOSTS_INSTALL_SCRIPT = _SRC_ROOT / "hosts-blocker" / "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 run_quiet(
|
|
cmd: list[str], *, timeout: float = 5
|
|
) -> subprocess.CompletedProcess[bytes]:
|
|
"""Run ``cmd`` with output captured and no exception on a non-zero exit."""
|
|
return subprocess.run(cmd, capture_output=True, timeout=timeout, check=False)
|
|
|
|
|
|
def ensure_chain_in_output(chain: str) -> None:
|
|
"""Hook ``chain`` into iptables' OUTPUT if it is not already there.
|
|
|
|
Raises:
|
|
subprocess.CalledProcessError: If the insert itself fails -- a chain
|
|
that exists but is not reachable from OUTPUT blocks nothing.
|
|
"""
|
|
probe = run_quiet([SUDO, IPTABLES, "-C", "OUTPUT", "-j", chain])
|
|
if probe.returncode != 0:
|
|
subprocess.run(
|
|
[SUDO, IPTABLES, "-I", "OUTPUT", "-j", chain],
|
|
capture_output=True,
|
|
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])
|