fix(pairing): reset the failed-approval counter on a successful approval

approve_code()'s success path never cleared _failures:{platform}. The
counter is incremented on every non-matching code, persisted in
_rate_limits.json, and only ever reset to 0 when it reaches
MAX_FAILED_ATTEMPTS (firing the lockout). So it counts failures over the
gateway's entire lifetime, not consecutive ones.

An owner who mistypes a pairing code on a handful of separate occasions
— each time immediately retyping it correctly and successfully pairing —
accumulates those isolated typos. A later single fresh typo then hits
MAX_FAILED_ATTEMPTS and locks the whole platform out for an hour, at
which point _is_locked_out gates approve_code and even the *correct*
code is rejected.

Reset the counter on a successful approval, matching standard
brute-force-guard semantics (the counter tracks consecutive failures).
This does not weaken protection: an attacker cannot produce a success
without a valid code, and 5 consecutive wrong attempts still lock out.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
solyanviktor-star 2026-07-10 18:18:34 +03:00 committed by Brooklyn Nicholson
parent 57ad222620
commit c352a322b0
2 changed files with 45 additions and 0 deletions

View file

@ -419,6 +419,14 @@ class PairingStore:
del pending[matched_key]
self._save_json(self._pending_path(platform), pending)
# A successful approval proves the requester is legitimate, so the
# brute-force failure streak must not carry over. Without this,
# isolated mistyped codes accumulate across the gateway's lifetime
# (the counter is persisted in _rate_limits.json and only ever
# reset when a lockout fires) and eventually trip a spurious
# lockout on a single fresh typo — rejecting even a valid code.
self._reset_failed_attempts(platform)
self._approve_user(
platform, matched_entry["user_id"], matched_entry.get("user_name", "")
)
@ -661,6 +669,19 @@ class PairingStore:
f"after {MAX_FAILED_ATTEMPTS} failed attempts", flush=True)
self._save_json(self._rate_limit_path(), limits)
def _reset_failed_attempts(self, platform: str) -> None:
"""Clear the accumulated failed-approval counter after a success.
Called from the ``approve_code`` success path so that a legitimate
approval resets the brute-force streak (standard lockout semantics:
the counter tracks *consecutive* failures, not lifetime ones).
"""
limits = self._load_json(self._rate_limit_path())
fail_key = f"_failures:{platform}"
if limits.get(fail_key):
limits[fail_key] = 0
self._save_json(self._rate_limit_path(), limits)
# ----- Cleanup -----
def _cleanup_expired(self, platform: str) -> None:

View file

@ -302,6 +302,30 @@ class TestApprovalFlow:
class TestLockout:
def test_successful_approval_resets_failure_counter(self, tmp_path):
"""A successful approval clears the brute-force streak, so isolated
typos across the gateway's lifetime don't accumulate into a spurious
lockout that rejects a valid code.
"""
with patch("gateway.pairing.PAIRING_DIR", tmp_path):
store = PairingStore()
# One short of the lockout threshold — not locked out yet.
for _ in range(MAX_FAILED_ATTEMPTS - 1):
assert store.approve_code("telegram", "WRONGCODE") is None
assert store._is_locked_out("telegram") is False
# A legitimate approval must reset the accumulated failures.
code = store.generate_code("telegram", "user1", "Alice")
assert store.approve_code("telegram", code) is not None
limits = store._load_json(store._rate_limit_path())
assert limits.get("_failures:telegram", 0) == 0
# Because the streak was cleared, a single fresh typo afterwards
# must NOT trip the lockout (it would have with the stale count).
assert store.approve_code("telegram", "WRONGCODE") is None
assert store._is_locked_out("telegram") is False
def test_lockout_blocks_code_approval(self, tmp_path):
"""Regression guard for #10195: lockout must also gate approve_code.