Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion gateway/pairing.py
Original file line number Diff line number Diff line change
Expand Up @@ -246,11 +246,13 @@ def approve_code(self, platform: str, code: str) -> Optional[dict]:
# at their TTL by _cleanup_expired.
matched_key = None
matched_entry = None
any_new_format_checked = False
for entry_id, entry in pending.items():
if not isinstance(entry, dict):
continue
if "salt" not in entry or "hash" not in entry:
continue
any_new_format_checked = True
try:
salt = bytes.fromhex(entry["salt"])
except ValueError:
Expand All @@ -262,7 +264,13 @@ def approve_code(self, platform: str, code: str) -> Optional[dict]:
break

if matched_key is None:
self._record_failed_attempt(platform)
# Skip failure recording only when pending is non-empty but
# every entry is legacy format (no salt/hash). This is the
# post-upgrade window where admins still have plaintext codes
# pending. Empty pending or a real hash-format mismatch both
# count as genuine failures.
if not pending or any_new_format_checked:
self._record_failed_attempt(platform)
return None

del pending[matched_key]
Expand Down
23 changes: 23 additions & 0 deletions tests/gateway/test_pairing.py
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,29 @@ def test_approve_code_ignores_legacy_entries(self, tmp_path):
# Approved list must be empty
assert store.is_approved("telegram", "legacy-user") is False

def test_legacy_only_pending_does_not_trigger_lockout(self, tmp_path):
"""Regression: approve_code with only legacy entries must not count as a
failed attempt (#30383). A platform upgraded mid-session has pending.json
entries written by the old code (no salt/hash fields). Calling approve_code
before those entries expire must not penalise the platform's failure counter
and must not cause lockout after MAX_FAILED_ATTEMPTS such calls."""
with patch("gateway.pairing.PAIRING_DIR", tmp_path):
self._write_legacy(tmp_path, code="LEGACY01")
store = PairingStore()
# Drive approve_code more than MAX_FAILED_ATTEMPTS times —
# if the bug is present, this would lock the platform out.
for _ in range(MAX_FAILED_ATTEMPTS + 1):
result = store.approve_code("telegram", "LEGACY01")
assert result is None
# Platform must NOT be locked out.
assert store._is_locked_out("telegram") is False
# A legitimately generated new-format code must still work.
new_code = store.generate_code("telegram", "realuser", "Real")
assert new_code is not None
approved = store.approve_code("telegram", new_code)
assert approved is not None
assert approved["user_id"] == "realuser"

def test_list_pending_handles_legacy_entries(self, tmp_path):
"""list_pending must not KeyError on a missing 'hash' field."""
with patch("gateway.pairing.PAIRING_DIR", tmp_path):
Expand Down
Loading