perf(whitelist): only shell out to chattr when the immutable flag differs
Some checks are pending
pre-commit / pre-commit (push) Waiting to run
Tests / test (3.10) (push) Waiting to run
Tests / test (3.11) (push) Waiting to run
Tests / test (3.12) (push) Waiting to run

lock_enforcement_files() runs at the end of every enforce-loop iteration and
called _try_set_immutable() on both config files unconditionally, so the
daemon forked `chattr +i` twice per iteration whether or not the flag was
already set. Measured with bpftrace: 20 chattr execs per 30 seconds, ~56,000
per day, all of them re-applying a flag that was already there.

Read the flag first with FS_IOC_GETFLAGS, which is an ioctl and costs no
process, and only spawn chattr when the current state differs.

_immutable_flag_is() fails closed on purpose: an unsupported filesystem, a
refused ioctl, or an unreadable file all report False so the caller still
runs chattr. A wrong True would silently stop enforcing immutability, which
is much worse than a redundant fork.

Verified live, not just in tests: `chattr` execs over 30s went 20 -> 0 after
restarting the service, and clearing the flag by hand
(`sudo chattr -i ~/.config/steam_backlog_enforcer/config.json`) still got it
re-applied within ~10 seconds, so enforcement is unchanged.

Full suite: 1091 passed, coverage 100.00%.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-07-25 20:15:12 +02:00
parent e5d0627a30
commit 243e11bc00
2 changed files with 116 additions and 1 deletions

View File

@ -2,8 +2,10 @@
from __future__ import annotations
import array
from collections import Counter
import contextlib
import fcntl
import json
import logging
import math
@ -112,13 +114,48 @@ def validate_reason(reason: str) -> str | None:
# Immutability helpers
# ──────────────────────────────────────────────────────────────
# From <linux/fs.h>. FS_IOC_GETFLAGS is the 64-bit encoding of _IOR('f', 1,
# long); FS_IMMUTABLE_FL is the "immutable" bit that `chattr +i` sets.
_FS_IOC_GETFLAGS = 0x80086601
_FS_IMMUTABLE_FL = 0x00000010
def _immutable_flag_is(path: Path, *, immutable: bool) -> bool:
"""Report whether *path*'s immutable bit already equals *immutable*.
Reading the flag is an ioctl, which costs no process; ``chattr`` costs a
fork and an exec. Since ``lock_enforcement_files`` runs on every
enforce-loop iteration, asking first turns a steady ~0.65 execs/second
into none at all for a flag that essentially never changes.
Fails closed: any uncertainty at all — filesystem without flag support,
ioctl refused, file unreadable — reports False so the caller still shells
out to chattr. A wrong True would silently stop enforcing immutability.
Args:
path: File to inspect.
immutable: The state to compare against.
Returns:
True only if the flag could be read and already matches.
"""
buf = array.array("L", [0])
try:
with path.open("rb") as handle:
# mutate_flag defaults to True, so buf receives the flags in place.
fcntl.ioctl(handle.fileno(), _FS_IOC_GETFLAGS, buf)
except (OSError, ValueError):
return False
return bool(buf[0] & _FS_IMMUTABLE_FL) is immutable
def _try_set_immutable(path: Path, *, immutable: bool) -> None:
"""Silently attempt to set or clear the immutable flag on *path*.
This is a best-effort operation — it fails silently if chattr is not
available, the process lacks the required capability, or the filesystem
does not support the flag.
does not support the flag. When the flag already holds the requested
value, no subprocess is spawned at all.
Args:
path: File to modify.
@ -126,6 +163,8 @@ def _try_set_immutable(path: Path, *, immutable: bool) -> None:
"""
if not path.exists():
return
if _immutable_flag_is(path, immutable=immutable):
return
chattr = shutil.which("chattr")
if chattr is None:
return

View File

@ -9,6 +9,7 @@ import pytest
from steam_backlog_enforcer._whitelist import (
_append_audit_log,
_immutable_flag_is,
_load_approved,
_save_approved,
_shannon_entropy,
@ -21,6 +22,7 @@ from steam_backlog_enforcer._whitelist import (
)
if TYPE_CHECKING:
import array
from pathlib import Path
# ──────────────────────────────────────────────────────────────
@ -125,12 +127,73 @@ class TestTrySetImmutable:
fake_chattr = tmp_path / "chattr"
with (
patch("shutil.which", return_value=str(fake_chattr)),
# A fresh tmp file is already mutable, so without this the call
# would (correctly) short-circuit before reaching chattr.
patch(
"steam_backlog_enforcer._whitelist._immutable_flag_is",
return_value=False,
),
patch("subprocess.run") as mock_run,
):
_try_set_immutable(target, immutable=False)
args = mock_run.call_args[0][0]
assert "-i" in args
def test_skips_chattr_when_flag_already_correct(self, tmp_path: Path) -> None:
# The hot path: lock_enforcement_files runs every loop iteration, and
# the flag is already what it should be, so nothing should be spawned.
target = tmp_path / "file.txt"
target.write_text("data", encoding="utf-8")
with (
patch(
"steam_backlog_enforcer._whitelist._immutable_flag_is",
return_value=True,
),
patch("shutil.which") as mock_which,
patch("subprocess.run") as mock_run,
):
_try_set_immutable(target, immutable=True)
mock_run.assert_not_called()
mock_which.assert_not_called()
class TestImmutableFlagIs:
def test_reads_clear_flag_on_real_file(self, tmp_path: Path) -> None:
target = tmp_path / "file.txt"
target.write_text("data", encoding="utf-8")
assert _immutable_flag_is(target, immutable=False) is True
assert _immutable_flag_is(target, immutable=True) is False
def test_fails_closed_when_ioctl_refuses(self, tmp_path: Path) -> None:
# Filesystems without flag support raise OSError; we must report False
# so the caller still shells out rather than silently skipping.
target = tmp_path / "file.txt"
target.write_text("data", encoding="utf-8")
with patch("fcntl.ioctl", side_effect=OSError("unsupported")):
assert _immutable_flag_is(target, immutable=False) is False
assert _immutable_flag_is(target, immutable=True) is False
def test_fails_closed_on_value_error(self, tmp_path: Path) -> None:
target = tmp_path / "file.txt"
target.write_text("data", encoding="utf-8")
with patch("fcntl.ioctl", side_effect=ValueError("bad buffer")):
assert _immutable_flag_is(target, immutable=True) is False
def test_fails_closed_when_file_cannot_be_opened(self, tmp_path: Path) -> None:
assert _immutable_flag_is(tmp_path / "missing.txt", immutable=True) is False
def test_reports_set_flag(self, tmp_path: Path) -> None:
target = tmp_path / "file.txt"
target.write_text("data", encoding="utf-8")
def _fake_ioctl(_fd: int, _req: int, buf: array.array[int]) -> int:
buf[0] = 0x00000010 # FS_IMMUTABLE_FL
return 0
with patch("fcntl.ioctl", side_effect=_fake_ioctl):
assert _immutable_flag_is(target, immutable=True) is True
assert _immutable_flag_is(target, immutable=False) is False
def test_oserror_swallowed(self, tmp_path: Path) -> None:
target = tmp_path / "file.txt"
target.write_text("data", encoding="utf-8")
@ -180,12 +243,25 @@ class TestLockAndUnlock:
target.write_text("data", encoding="utf-8")
with (
patch("shutil.which", return_value="/usr/bin/chattr"),
# Pretend the file really is immutable; a fresh tmp file is not,
# and unlocking an already-mutable file is now a no-op.
patch(
"steam_backlog_enforcer._whitelist._immutable_flag_is",
return_value=False,
),
patch("subprocess.run") as mock_run,
):
unlock_for_write(target)
args = mock_run.call_args[0][0]
assert "-i" in args
def test_unlock_for_write_skips_already_mutable_file(self, tmp_path: Path) -> None:
target = tmp_path / "file.txt"
target.write_text("data", encoding="utf-8")
with patch("subprocess.run") as mock_run:
unlock_for_write(target)
mock_run.assert_not_called()
# ──────────────────────────────────────────────────────────────
# Persistence helpers (_load_approved, _save_approved)