Remove whitelist exception cooldown; approve immediately
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

Whitelist exceptions previously entered a 24h pending queue and were
promoted to approved by the enforce loop once the cooldown elapsed.
add_pending_exception now approves immediately, so the entire pending/
promote/cooldown mechanism is dead: drop WHITELIST_COOLDOWN_SECONDS,
PENDING_EXCEPTIONS_FILE, _load_pending/_save_pending,
promote_pending_exceptions, and list_pending_exceptions, plus the
enforce-loop promotion step and the CLI's now-empty pending display.
Usage text and docstrings updated to state exceptions are active
immediately. Tests for the removed machinery deleted; add-exception
tests assert immediate approval. Coverage remains 100%.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017RRCdR4D8SaVrN2xraMQxM
This commit is contained in:
Krzysztof Rudnicki 2026-07-10 22:57:39 +02:00
parent 26c4771192
commit de76c6541e
7 changed files with 27 additions and 341 deletions

View File

@ -15,7 +15,6 @@ from steam_backlog_enforcer._total_block import (
) )
from steam_backlog_enforcer._whitelist import ( from steam_backlog_enforcer._whitelist import (
lock_enforcement_files, lock_enforcement_files,
promote_pending_exceptions,
) )
from steam_backlog_enforcer.config import ( from steam_backlog_enforcer.config import (
CONFIG_DIR, CONFIG_DIR,
@ -296,12 +295,7 @@ def _enforce_loop_iteration(config: Config, state: State) -> None:
config.steam_id, config.steam_id,
) )
# D) Promote any cooldown-elapsed pending exceptions to approved. # D) Re-apply immutable flag so config cannot be edited without root.
newly_approved = promote_pending_exceptions()
for aid in newly_approved:
logger.info("Exception approved: AppID=%d", aid)
# E) Re-apply immutable flag so config cannot be edited without root.
lock_enforcement_files(CONFIG_FILE) lock_enforcement_files(CONFIG_FILE)

View File

@ -24,7 +24,6 @@ logger = logging.getLogger(__name__)
# File paths (patched in tests via conftest) # File paths (patched in tests via conftest)
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
PENDING_EXCEPTIONS_FILE: Path = CONFIG_DIR / "pending_exceptions.json"
APPROVED_EXCEPTIONS_FILE: Path = CONFIG_DIR / "approved_exceptions.json" APPROVED_EXCEPTIONS_FILE: Path = CONFIG_DIR / "approved_exceptions.json"
EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log" EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log"
@ -32,8 +31,6 @@ EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log"
# Constants # Constants
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
WHITELIST_COOLDOWN_SECONDS: int = 86400 # 24 hours
_MIN_REASON_WORDS: int = 5 _MIN_REASON_WORDS: int = 5
_MIN_REASON_LENGTH: int = 25 _MIN_REASON_LENGTH: int = 25
_MIN_ENTROPY: float = 3.0 _MIN_ENTROPY: float = 3.0
@ -169,24 +166,6 @@ def unlock_for_write(path: Path) -> None:
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
def _load_pending() -> list[dict[str, object]]:
"""Load pending exception entries from disk."""
if not PENDING_EXCEPTIONS_FILE.exists():
return []
try:
data: object = json.loads(PENDING_EXCEPTIONS_FILE.read_text(encoding="utf-8"))
if isinstance(data, list):
return cast("list[dict[str, object]]", data)
except (json.JSONDecodeError, OSError, ValueError):
pass
return []
def _save_pending(entries: list[dict[str, object]]) -> None:
"""Persist pending exception entries to disk."""
_atomic_write(PENDING_EXCEPTIONS_FILE, json.dumps(entries, indent=2) + "\n")
def _load_approved() -> list[dict[str, object]]: def _load_approved() -> list[dict[str, object]]:
"""Load approved exception entries from disk.""" """Load approved exception entries from disk."""
if not APPROVED_EXCEPTIONS_FILE.exists(): if not APPROVED_EXCEPTIONS_FILE.exists():
@ -232,17 +211,17 @@ def _append_audit_log(app_id: int, reason: str, event: str) -> None:
def add_pending_exception(app_id: int, reason: str) -> str: def add_pending_exception(app_id: int, reason: str) -> str:
"""Request a new whitelist exception for *app_id*. """Add a whitelist exception for *app_id* immediately.
The entry becomes active only after ``WHITELIST_COOLDOWN_SECONDS`` have The entry becomes active right away (no cooldown). Returns a
elapsed (24 h by default). Returns a human-readable status message. human-readable status message.
Args: Args:
app_id: Steam application ID to add. app_id: Steam application ID to add.
reason: Validated justification text (must pass :func:`validate_reason`). reason: Validated justification text (must pass :func:`validate_reason`).
Returns: Returns:
Human-readable confirmation or remaining-cooldown message. Human-readable confirmation message.
Raises: Raises:
ValueError: If the reason fails validation or the ID is already approved. ValueError: If the reason fails validation or the ID is already approved.
@ -256,92 +235,25 @@ def add_pending_exception(app_id: int, reason: str) -> str:
msg = f"AppID {app_id} is already in the approved exceptions list." msg = f"AppID {app_id} is already in the approved exceptions list."
raise ValueError(msg) raise ValueError(msg)
pending = _load_pending()
for entry in pending:
if int(entry["app_id"]) == app_id:
elapsed = time.time() - float(entry["requested_at"])
remaining = WHITELIST_COOLDOWN_SECONDS - elapsed
if remaining > 0:
hours = int(remaining // 3600)
mins = int((remaining % 3600) // 60)
return (
f"AppID {app_id} is already pending; approves in {hours}h {mins}m."
)
# Cooldown already elapsed for this pending entry — promote now.
break
entry_new: dict[str, object] = {
"app_id": app_id,
"reason": reason,
"requested_at": time.time(),
}
pending.append(entry_new)
_save_pending(pending)
_append_audit_log(app_id, reason, "REQUESTED")
hours = WHITELIST_COOLDOWN_SECONDS // 3600
return (
f"Exception requested for AppID {app_id}. "
f"Will become active in {hours}h. Reason logged."
)
def promote_pending_exceptions() -> list[int]:
"""Move cooldown-elapsed pending entries to the approved list.
Called by the enforce daemon on each loop iteration. Returns the list
of app IDs that were promoted this call.
Returns:
List of newly approved app IDs (may be empty).
"""
pending = _load_pending()
now = time.time() now = time.time()
still_pending: list[dict[str, object]] = [] approved.append(
newly_approved: list[int] = [] {
"app_id": app_id,
"reason": reason,
"approved_at": now,
}
)
_save_approved(approved)
_append_audit_log(app_id, reason, "APPROVED")
for entry in pending: return f"Exception approved for AppID {app_id}. Active immediately. Reason logged."
elapsed = now - float(entry["requested_at"])
if elapsed >= WHITELIST_COOLDOWN_SECONDS:
app_id = int(entry["app_id"])
approved = _load_approved()
if not any(int(e["app_id"]) == app_id for e in approved):
approved.append(
{
"app_id": app_id,
"reason": entry["reason"],
"approved_at": now,
}
)
_save_approved(approved)
_append_audit_log(app_id, str(entry["reason"]), "APPROVED")
newly_approved.append(app_id)
else:
still_pending.append(entry)
if len(still_pending) != len(pending):
_save_pending(still_pending)
return newly_approved
def get_approved_exception_ids() -> frozenset[int]: def get_approved_exception_ids() -> frozenset[int]:
"""Return the frozenset of currently approved exception app IDs. """Return the frozenset of currently approved exception app IDs.
Does NOT trigger promotion — call :func:`promote_pending_exceptions`
explicitly when timely promotion is required (e.g. the enforce loop).
Returns: Returns:
Frozenset of approved app IDs. Frozenset of approved app IDs.
""" """
approved = _load_approved() approved = _load_approved()
return frozenset(int(e["app_id"]) for e in approved) return frozenset(int(e["app_id"]) for e in approved)
def list_pending_exceptions() -> list[dict[str, object]]:
"""Return a copy of the current pending exception list.
Returns:
List of pending exception dicts with keys app_id, reason, requested_at.
"""
return list(_load_pending())

View File

@ -5,7 +5,6 @@ from __future__ import annotations
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
import logging import logging
import sys import sys
import time
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from steam_backlog_enforcer._actions import ( from steam_backlog_enforcer._actions import (
@ -32,9 +31,7 @@ from steam_backlog_enforcer._total_block import (
) )
from steam_backlog_enforcer._web_server import serve from steam_backlog_enforcer._web_server import serve
from steam_backlog_enforcer._whitelist import ( from steam_backlog_enforcer._whitelist import (
WHITELIST_COOLDOWN_SECONDS,
add_pending_exception, add_pending_exception,
list_pending_exceptions,
validate_reason, validate_reason,
) )
from steam_backlog_enforcer.config import ( from steam_backlog_enforcer.config import (
@ -364,17 +361,16 @@ _ADD_EXCEPTION_USAGE = (
"Example:\n" "Example:\n"
" add-exception 440 --reason " " add-exception 440 --reason "
'"TF2 is needed for a community event this weekend"\n\n' '"TF2 is needed for a community event this weekend"\n\n'
f"Exceptions become active after a {WHITELIST_COOLDOWN_SECONDS // 3600}h " "Exceptions become active immediately."
"cooldown."
) )
def cmd_add_exception(args: list[str]) -> None: def cmd_add_exception(args: list[str]) -> None:
"""Request a time-locked whitelist exception. """Add a whitelist exception, active immediately.
Usage: add-exception <app_id> --reason "<text>" Usage: add-exception <app_id> --reason "<text>"
The exception becomes active after a 24-hour cooldown. The reason must be The exception becomes active right away (no cooldown). The reason must be
a genuine justification of at least 5 words with sufficient entropy. a genuine justification of at least 5 words with sufficient entropy.
Args: Args:
@ -411,20 +407,6 @@ def cmd_add_exception(args: list[str]) -> None:
_echo(msg) _echo(msg)
# Show current pending list.
pending = list_pending_exceptions()
if pending:
_echo(f"\nPending exceptions ({len(pending)}):")
now = time.time()
for entry in pending:
aid = int(entry["app_id"])
elapsed = now - float(entry["requested_at"])
remaining = max(0.0, WHITELIST_COOLDOWN_SECONDS - elapsed)
hrs = int(remaining // 3600)
mins = int((remaining % 3600) // 60)
status = "ready" if remaining == 0.0 else f"approves in {hrs}h {mins}m"
_echo(f" AppID={aid} [{status}]")
def cmd_install(config: Config, state: State) -> None: def cmd_install(config: Config, state: State) -> None:
"""Manually trigger install of the assigned game.""" """Manually trigger install of the assigned game."""

View File

@ -113,10 +113,6 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]:
), ),
), ),
# Whitelist exception files (_whitelist module-level constants) # Whitelist exception files (_whitelist module-level constants)
patch(
"steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE",
fake_config / "pending_exceptions.json",
),
patch( patch(
"steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE", "steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE",
fake_config / "approved_exceptions.json", fake_config / "approved_exceptions.json",

View File

@ -100,22 +100,6 @@ class TestEnforceLoopIteration:
mock_guard.assert_not_called() mock_guard.assert_not_called()
mock_installed.assert_not_called() mock_installed.assert_not_called()
def test_promotes_newly_approved_exceptions(self) -> None:
"""Loop body at line 286 executes when promote returns non-empty list."""
config = Config(
kill_unauthorized_games=False,
uninstall_other_games=False,
)
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.is_game_installed", return_value=True),
patch(
f"{PKG}.promote_pending_exceptions",
return_value=[440],
),
):
_enforce_loop_iteration(config, state)
class TestDoEnforce: class TestDoEnforce:
"""Tests for do_enforce.""" """Tests for do_enforce."""

View File

@ -3,13 +3,11 @@
from __future__ import annotations from __future__ import annotations
import sys import sys
import time
from typing import Any from typing import Any
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from steam_backlog_enforcer._whitelist import WHITELIST_COOLDOWN_SECONDS
from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.main import ( from steam_backlog_enforcer.main import (
cmd_add_exception, cmd_add_exception,
@ -448,37 +446,17 @@ class TestCmdAddException:
): ):
cmd_add_exception(["440", "--reason", _VALID_REASON]) cmd_add_exception(["440", "--reason", _VALID_REASON])
def test_happy_path_no_pending(self) -> None: def test_happy_path(self) -> None:
with ( with (
patch(f"{PKG}._echo") as mock_echo, patch(f"{PKG}._echo") as mock_echo,
patch( patch(
f"{PKG}.add_pending_exception", f"{PKG}.add_pending_exception",
return_value="Exception requested for AppID 440.", return_value="Exception approved for AppID 440. Active immediately.",
), ),
patch(f"{PKG}.list_pending_exceptions", return_value=[]),
): ):
cmd_add_exception(["440", "--reason", _VALID_REASON]) cmd_add_exception(["440", "--reason", _VALID_REASON])
mock_echo.assert_called() mock_echo.assert_called()
def test_happy_path_with_pending_list(self) -> None:
now = time.time()
pending = [
{"app_id": 440, "requested_at": now - WHITELIST_COOLDOWN_SECONDS - 1},
{"app_id": 730, "requested_at": now},
]
with (
patch(f"{PKG}._echo") as mock_echo,
patch(
f"{PKG}.add_pending_exception",
return_value="Exception requested for AppID 440.",
),
patch(f"{PKG}.list_pending_exceptions", return_value=pending),
):
cmd_add_exception(["440", "--reason", _VALID_REASON])
# At least the "Pending exceptions" line should be echoed
calls = [str(c) for c in mock_echo.call_args_list]
assert any("Pending" in s for s in calls)
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
# main() dispatch to add-exception # main() dispatch to add-exception

View File

@ -2,26 +2,20 @@
from __future__ import annotations from __future__ import annotations
import time
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from steam_backlog_enforcer._whitelist import ( from steam_backlog_enforcer._whitelist import (
WHITELIST_COOLDOWN_SECONDS,
_append_audit_log, _append_audit_log,
_load_approved, _load_approved,
_load_pending,
_save_approved, _save_approved,
_save_pending,
_shannon_entropy, _shannon_entropy,
_try_set_immutable, _try_set_immutable,
add_pending_exception, add_pending_exception,
get_approved_exception_ids, get_approved_exception_ids,
list_pending_exceptions,
lock_enforcement_files, lock_enforcement_files,
promote_pending_exceptions,
unlock_for_write, unlock_for_write,
validate_reason, validate_reason,
) )
@ -194,39 +188,11 @@ class TestLockAndUnlock:
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
# Persistence helpers (_load_pending, _save_pending, etc.) # Persistence helpers (_load_approved, _save_approved)
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
class TestPersistence: class TestPersistence:
def test_load_pending_missing_file(self) -> None:
assert _load_pending() == []
def test_load_pending_corrupt_file(self, tmp_path: Path) -> None:
bad = tmp_path / "pending.json"
bad.write_text("not json{{", encoding="utf-8")
with patch(
"steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE",
bad,
):
assert _load_pending() == []
def test_load_pending_non_list(self, tmp_path: Path) -> None:
bad = tmp_path / "pending.json"
bad.write_text('{"key": "value"}', encoding="utf-8")
with patch(
"steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE",
bad,
):
assert _load_pending() == []
def test_save_and_load_pending_roundtrip(self) -> None:
entries: list[dict[str, object]] = [
{"app_id": 440, "reason": "test", "requested_at": 12345.0}
]
_save_pending(entries)
assert _load_pending() == entries
def test_load_approved_missing_file(self) -> None: def test_load_approved_missing_file(self) -> None:
assert _load_approved() == [] assert _load_approved() == []
@ -300,10 +266,12 @@ class TestAddPendingException:
with patch("shutil.which", return_value=None): with patch("shutil.which", return_value=None):
msg = add_pending_exception(440, _VALID_REASON) msg = add_pending_exception(440, _VALID_REASON)
assert "440" in msg assert "440" in msg
assert "24h" in msg or "hours" in msg or "active" in msg.lower() assert "immediately" in msg.lower()
pending = _load_pending() approved = _load_approved()
assert len(pending) == 1 assert len(approved) == 1
assert int(pending[0]["app_id"]) == 440 assert int(approved[0]["app_id"]) == 440
# Now active right away, no cooldown.
assert 440 in get_approved_exception_ids()
def test_invalid_reason_raises(self) -> None: def test_invalid_reason_raises(self) -> None:
with pytest.raises( with pytest.raises(
@ -322,100 +290,6 @@ class TestAddPendingException:
): ):
add_pending_exception(440, _VALID_REASON) add_pending_exception(440, _VALID_REASON)
def test_already_pending_cooldown_remaining(self) -> None:
existing: list[dict[str, object]] = [
{
"app_id": 440,
"reason": _VALID_REASON,
"requested_at": time.time(), # just now → full cooldown remaining
}
]
_save_pending(existing)
with patch("shutil.which", return_value=None):
msg = add_pending_exception(440, _VALID_REASON)
assert "already pending" in msg
def test_already_pending_cooldown_elapsed_promotes(self) -> None:
past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1
existing: list[dict[str, object]] = [
{"app_id": 440, "reason": _VALID_REASON, "requested_at": past}
]
_save_pending(existing)
with patch("shutil.which", return_value=None):
msg = add_pending_exception(440, _VALID_REASON)
# The elapsed entry is broken out of the pending-check loop via break,
# then a new entry is appended → still gets the "Will become active" msg.
assert "440" in msg
# ──────────────────────────────────────────────────────────────
# promote_pending_exceptions
# ──────────────────────────────────────────────────────────────
class TestPromotePendingExceptions:
def test_no_entries(self) -> None:
result = promote_pending_exceptions()
assert result == []
def test_cooldown_not_elapsed(self) -> None:
entries: list[dict[str, object]] = [
{"app_id": 440, "reason": _VALID_REASON, "requested_at": time.time()}
]
_save_pending(entries)
with patch("shutil.which", return_value=None):
result = promote_pending_exceptions()
assert result == []
# Still pending
assert len(_load_pending()) == 1
def test_cooldown_elapsed_promotes(self) -> None:
past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1
entries: list[dict[str, object]] = [
{"app_id": 440, "reason": _VALID_REASON, "requested_at": past}
]
_save_pending(entries)
with patch("shutil.which", return_value=None):
result = promote_pending_exceptions()
assert 440 in result
assert _load_pending() == []
approved_ids = get_approved_exception_ids()
assert 440 in approved_ids
def test_already_in_approved_not_duplicated(self) -> None:
"""If somehow already approved, skip duplicating it."""
past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1
entries: list[dict[str, object]] = [
{"app_id": 440, "reason": _VALID_REASON, "requested_at": past}
]
_save_pending(entries)
existing_approved: list[dict[str, object]] = [
{"app_id": 440, "reason": _VALID_REASON, "approved_at": 0.0}
]
_save_approved(existing_approved)
with patch("shutil.which", return_value=None):
result = promote_pending_exceptions()
# Not added again
assert 440 not in result
approved = _load_approved()
assert sum(1 for e in approved if int(e["app_id"]) == 440) == 1
def test_pending_list_saved_when_entries_removed(self) -> None:
"""_save_pending is called when the pending list shrinks."""
past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1
future = time.time()
entries: list[dict[str, object]] = [
{"app_id": 440, "reason": _VALID_REASON, "requested_at": past},
{"app_id": 730, "reason": _VALID_REASON, "requested_at": future},
]
_save_pending(entries)
with patch("shutil.which", return_value=None):
result = promote_pending_exceptions()
assert 440 in result
remaining = _load_pending()
assert len(remaining) == 1
assert int(remaining[0]["app_id"]) == 730
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
# get_approved_exception_ids # get_approved_exception_ids
@ -439,27 +313,6 @@ class TestGetApprovedExceptionIds:
assert result == frozenset({440, 730}) assert result == frozenset({440, 730})
# ──────────────────────────────────────────────────────────────
# list_pending_exceptions
# ──────────────────────────────────────────────────────────────
class TestListPendingExceptions:
def test_returns_copy(self) -> None:
"""Mutating the result must not affect stored state."""
entries: list[dict[str, object]] = [
{"app_id": 440, "reason": "r", "requested_at": 1.0}
]
_save_pending(entries)
result = list_pending_exceptions()
result.clear()
# Still present on next load
assert len(_load_pending()) == 1
def test_empty_when_no_pending(self) -> None:
assert list_pending_exceptions() == []
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
# Extra coverage for validate_reason branches 94 & 106 # Extra coverage for validate_reason branches 94 & 106
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
@ -481,16 +334,3 @@ class TestValidateReasonExtraBranches:
err = validate_reason(reason) err = validate_reason(reason)
assert err is not None assert err is not None
assert "repetitive" in err assert "repetitive" in err
def test_different_app_id_in_pending_does_not_block(self) -> None:
"""Pending entry with a different app_id must not block a new add (covers
the 264→263 loop-continue branch)."""
other: list[dict[str, object]] = [
{"app_id": 440, "reason": _VALID_REASON, "requested_at": 1.0}
]
_save_pending(other)
with patch("shutil.which", return_value=None):
msg = add_pending_exception(730, _VALID_REASON)
assert "730" in msg
pending = _load_pending()
assert len(pending) == 2 # original + new