diff --git a/gateway/pairing.py b/gateway/pairing.py index f5dd131721a..5a1388f4760 100644 --- a/gateway/pairing.py +++ b/gateway/pairing.py @@ -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: diff --git a/tests/gateway/test_pairing.py b/tests/gateway/test_pairing.py index 6ecbc35b6a9..8a46ca7aa43 100644 --- a/tests/gateway/test_pairing.py +++ b/tests/gateway/test_pairing.py @@ -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.