Gateway pairing codes stored and hashed securely

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #30383 fix(security): hash gateway pairing codes instead of storing plaintext (#8056)

  merged to main

  files: gateway/pairing.py, infographic/pr-8056-hash-pairing-codes/infographic.png, tests/gateway/test_pairing.py

  Show the diff

  • gateway/pairing.py

    @@ -18,6 +18,7 @@
     Storage: ~/.hermes/pairing/
     """
     
    +import hashlib
     import json
     import os
     import secrets
    @@ -148,6 +149,11 @@ def revoke(self, platform: str, user_id: str) -> bool:
     
         # ----- Pending codes -----
     
    +    @staticmethod
    +    def _hash_code(code: str, salt: bytes) -> str:
    +        """Hash a pairing code with the given salt using SHA-256."""
    +        return hashlib.sha256(salt + code.encode("utf-8")).hexdigest()
    +
         def generate_code(
             self, platform: str, user_id: str, user_name: str = ""
         ) -> Optional[str]:
    @@ -158,6 +164,9 @@ def generate_code(
               - User is rate-limited (too recent request)
               - Max pending codes reached for this platform
               - User/platform is in lockout due to failed attempts
    +
    +        The code is NOT stored in plaintext.  Only a salted SHA-256 hash is
    +        persisted so that reading the pending file does not reveal codes.
             """
             with self._lock:
                 self._cleanup_expired(platform)
    @@ -178,8 +187,17 @@ def generate_code(
                 # Generate cryptographically random code
                 code = "".join(secrets.choice(ALPHABET) for _ in range(CODE_LENGTH))
     
    -            # Store pending request
    -            pending[code] = {
    +            # Hash the code with a random salt before storing
    +            salt = os.urandom(16)
    +            code_hash = self._hash_code(code, salt)
    +
    +            # Use a unique entry id as the key (not the code itself)
    +            entry_id = secrets.token_hex(8)
    +
    +            # Store pending request with hashed code
    +            pending[entry_id] = {
    +                "hash": code_hash,
    +                "salt": salt.hex(),
                     "user_id": user_id,
                     "user_name": user_name,
                     "created_at": time.time(),
    @@ -195,10 +213,16 @@ def approve_code(self, platform: str, code: str) -> Optional[dict]:
             """
             Approve a pairing code. Adds the user to the approved list.
     
    -        Returns {user_id, user_name} on success, None if code is
    +        Returns ``{user_id, user_name}`` on success, ``None`` if the code is
             invalid/expired OR the platform is currently locked out after
             ``MAX_FAILED_ATTEMPTS`` failed approvals (#10195). Callers can
             disambiguate with ``_is_locked_out(platform)``.
    +
    +        Verification: the user-provided code is hashed with each stored
    +        entry's salt and compared to the stored hash using constant-time
    +        comparison. Pre-hash entries (legacy plaintext-key format from
    +        pre-upgrade pending.json files) are silently ignored — they get
    +        pruned at TTL by ``_cleanup_expired``.
             """
             with self._lock:
                 self._cleanup_expired(platform)
    @@ -213,34 +237,73 @@ def approve_code(self, platform: str, code: str) -> Optional[dict]:
                     return None
     
                 pending = self._load_json(self._pending_path(platform))
    -            if code not in pending:
    +
    +            # Find the entry whose hash matches the provided code.
    +            # Tolerate legacy plaintext-key entries (no salt/hash) and
    +            # malformed entries — skip them rather than KeyError, so an
    +            # in-place upgrade across an existing pending.json doesn't
    +            # crash on the first approve call. Legacy entries get pruned
    +            # at their TTL by _cleanup_expired.
    +            matched_key = None
    +            matched_entry = None
    +            for entry_id, entry in pending.items():
    +                if not isinstance(entry, dict):
    +                    continue
    +                if "salt" not in entry or "hash" not in entry:
    +                    continue
    +                try:
    +                    salt = bytes.fromhex(entry["salt"])
    +                except ValueError:
    +                    continue
    +                candidate_hash = self._hash_code(code, salt)
    +                if secrets.compare_digest(candidate_hash, entry["hash"]):
    +                    matched_key = entry_id
    +                    matched_entry = entry
    +                    break
    +
    +            if matched_key is None:
                     self._record_failed_attempt(platform)
                     return None
     
    -            entry = pending.pop(code)
    +            del pending[matched_key]
                 self._save_json(self._pending_path(platform), pending)
     
                 # Add to approved list
    -            self._approve_user(platform, entry["user_id"], entry.get("user_name", ""))
    +            self._approve_user(platform, matched_entry["user_id"],
    +                               matched_entry.get("user_name", ""))
     
                 return {
    -                "user_id": entry["user_id"],
    -                "user_name": entry.get("user_name", ""),
    +                "user_id": matched_entry["user_id"],
    +                "user_name": matched_entry.get("user_name", ""),
                 }
     
         def list_pending(self, platform: str = None) -> list:
    -        """List pending pairing requests, optionally filtered by platform."""
    +        """List pending pairing requests, optionally filtered by platform.
    +
    +        Codes are stored hashed — the ``code`` field is replaced with the
    +        first 8 hex characters of the hash so admins can distinguish entries
    +        without revealing the original code. Legacy plaintext-key entries
    +        (pre-hash format) are shown with a "legacy" placeholder so admins
    +        can see them age out without crashing on a missing ``hash`` field.
    +        """
             results = []
             platforms = [platform] if platform else self._all_platforms("pending")
             for p in platforms:
                 self._cleanup_expired(p)
                 pending = self._load_json(self._pending_path(p))
    -            for code, info in pending.items():
    -                age_min = int((time.time() - info["created_at"]) / 60)
    +            for entry_id, info in pending.items():
    +                if not isinstance(info, dict):
    +                    continue
    +                created_at = info.get("created_at")
    +                if not isinstance(created_at, (int, float)):
    +                    continue
    +                age_min = int((time.time() - created_at) / 60)
    +                hash_val = info.get("hash")
    +                code_display = hash_val[:8] if isinstance(hash_val, str) else "legacy"
                     results.append({
                         "platform": p,
    -                    "code": code,
    -                    "user_id": info["user_id"],
    +                    "code": code_display,
    +                    "user_id": info.get("user_id", ""),
                         "user_name": info.get("user_name", ""),
                         "age_minutes": age_min,
                     })
    @@ -297,17 +360,29 @@ def _record_failed_attempt(self, platform: str) -> None:
         # ----- Cleanup -----
     
         def _cleanup_expired(self, platform: str) -> None:
    -        """Remove expired pending codes."""
    +        """Remove expired pending codes.
    +
    +        Tolerant of malformed / legacy entries — anything without a numeric
    +        ``created_at`` is treated as expired (it's effectively unusable
    +        with the new hash-keyed schema anyway).
    +        """
             path = self._pending_path(platform)
             pending = self._load_json(path)
             now = time.time()
    -        expired = [
    -            code for code, info in pending.items()
    -            if (now - info["created_at"]) > CODE_TTL_SECONDS
    -        ]
    +        expired = []
    +        for entry_id, info in pending.items():
    +            if not isinstance(info, dict):
    +                expired.append(entry_id)
    +                continue
    +            created_at = info.get("created_at")
    +            if not isinstance(created_at, (int, float)):
    +                expired.append(entry_id)
    +                continue
    +            if (now - created_at) > CODE_TTL_SECONDS:
    +                expired.append(entry_id)
             if expired:
    -            for code in expired:
    -                del pending[code]
    +            for entry_id in expired:
    +                del pending[entry_id]
                 self._save_json(path, pending)
     
         def _all_platforms(self, suffix: str) -> list:

  • tests/gateway/test_pairing.py

    @@ -75,9 +75,197 @@ def test_stores_pending_entry(self, tmp_path):
                 code = store.generate_code("telegram", "user1", "Alice")
                 pending = store.list_pending("telegram")
             assert len(pending) == 1
    -        assert pending[0]["code"] == code
    +        # list_pending no longer returns the original code — it returns a
    +        # truncated hash prefix.  Verify the metadata is correct instead.
             assert pending[0]["user_id"] == "user1"
             assert pending[0]["user_name"] == "Alice"
    +        # The code field is now a hash prefix, not the original plaintext code
    +        assert pending[0]["code"] != code
    +
    +
    +# ---------------------------------------------------------------------------
    +# Hashed storage
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestHashedStorage:
    +    def test_pending_file_contains_hash_and_salt(self, tmp_path):
    +        """Stored entries must have 'hash' and 'salt', never the plaintext code."""
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            store = PairingStore()
    +            code = store.generate_code("telegram", "user1", "Alice")
    +            raw = json.loads(
    +                (tmp_path / "telegram-pending.json").read_text(encoding="utf-8")
    +            )
    +
    +        assert len(raw) == 1
    +        entry = next(iter(raw.values()))
    +        # Must have hash and salt fields
    +        assert "hash" in entry
    +        assert "salt" in entry
    +        # Hash must be a valid hex SHA-256 digest (64 hex chars)
    +        assert len(entry["hash"]) == 64
    +        assert all(c in "0123456789abcdef" for c in entry["hash"])
    +        # Salt must be a valid hex string (32 hex chars for 16 bytes)
    +        assert len(entry["salt"]) == 32
    +        assert all(c in "0123456789abcdef" for c in entry["salt"])
    +        # The plaintext code must NOT appear as a key or value anywhere
    +        assert code not in raw  # not a key
    +        for key, val in raw.items():
    +            assert code != key
    +            for field_val in val.values():
    +                if isinstance(field_val, str):
    +                    assert field_val != code
    +
    +    def test_plaintext_code_not_stored(self, tmp_path):
    +        """The raw JSON file must not contain the plaintext code anywhere."""
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            store = PairingStore()
    +            code = store.generate_code("telegram", "user1")
    +            raw_text = (tmp_path / "telegram-pending.json").read_text(encoding="utf-8")
    +        assert code not in raw_text
    +
    +    def test_valid_code_verifies_against_hash(self, tmp_path):
    +        """approve_code with the correct code should succeed."""
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            store = PairingStore()
    +            code = store.generate_code("telegram", "user1", "Bob")
    +            result = store.approve_code("telegram", code)
    +        assert result is not None
    +        assert result["user_id"] == "user1"
    +        assert result["user_name"] == "Bob"
    +
    +    def test_invalid_code_rejected(self, tmp_path):
    +        """approve_code with a wrong code should fail."""
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            store = PairingStore()
    +            store.generate_code("telegram", "user1")
    +            result = store.approve_code("telegram", "ZZZZZZZZ")
    +        assert result is None
    +
    +    def test_different_salts_per_entry(self, tmp_path):
    +        """Each pending entry should have a unique salt."""
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            store = PairingStore()
    +            store.generate_code("telegram", "user0")
    +            store.generate_code("telegram", "user1")
    +            store.generate_code("telegram", "user2")
    +            raw = json.loads(
    +                (tmp_path / "telegram-pending.json").read_text(encoding="utf-8")
    +            )
    +        salts = [entry["salt"] for entry in raw.values()]
    +        assert len(set(salts)) == 3  # all unique
    +
    +    def test_hash_code_static_method(self, tmp_path):
    +        """_hash_code should be deterministic for the same code+salt."""
    +        salt = os.urandom(16)
    +        h1 = PairingStore._hash_code("ABCD1234", salt)
    +        h2 = PairingStore._hash_code("ABCD1234", salt)
    +        assert h1 == h2
    +        # Different salt should produce a different hash
    +        salt2 = os.urandom(16)
    +        h3 = PairingStore._hash_code("ABCD1234", salt2)
    +        assert h3 != h1
    +
    +
    +class TestLegacyPendingFileCompat:
    +    """Defensive coverage for pre-hash pending.json on upgraded installs.
    +
    +    Existing user installs may have a pending.json written by the old
    +    code (plaintext code as key, no hash/salt fields). The new
    +    approve_code / list_pending / _cleanup_expired must not crash on
    +    those entries — they should be ignored and aged out at TTL.
    +    """
    +
    +    @staticmethod
    +    def _write_legacy(tmp_path, code="ABCD1234", created_at=None):
    +        """Write a pre-hash pending.json with plaintext code as the key."""
    +        import time as _time
    +        if created_at is None:
    +            created_at = _time.time()
    +        legacy = {
    +            code: {
    +                "user_id": "legacy-user",
    +                "user_name": "Legacy",
    +                "created_at": created_at,
    +            }
    +        }
    +        (tmp_path / "telegram-pending.json").write_text(
    +            json.dumps(legacy), encoding="utf-8"
    +        )
    +
    +    def test_approve_code_ignores_legacy_entries(self, tmp_path):
    +        """A valid old-format code must NOT silently approve under the new schema."""
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            self._write_legacy(tmp_path, code="LEGACY01")
    +            store = PairingStore()
    +            # The plaintext "code" used to be the key — under the new schema
    +            # it's not even looked at, and there's no hash/salt to verify.
    +            # Result: approve_code returns None, the legacy entry is left
    +            # alone (gets pruned by _cleanup_expired at TTL).
    +            result = store.approve_code("telegram", "LEGACY01")
    +            assert result is None
    +            # Approved list must be empty
    +            assert store.is_approved("telegram", "legacy-user") is False
    +
    +    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):
    +            self._write_legacy(tmp_path)
    +            store = PairingStore()
    +            pending = store.list_pending("telegram")
    +        assert len(pending) == 1
    +        assert pending[0]["user_id"] == "legacy-user"
    +        assert pending[0]["code"] == "legacy"  # placeholder
    +
    +    def test_cleanup_expired_removes_legacy_at_ttl(self, tmp_path):
    +        """Legacy entries past CODE_TTL must still get pruned."""
    +        import time as _time
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            self._write_legacy(
    +                tmp_path,
    +                code="LEGACY99",
    +                created_at=_time.time() - CODE_TTL_SECONDS - 1,
    +            )
    +            store = PairingStore()
    +            store._cleanup_expired("telegram")
    +            raw = json.loads(
    +                (tmp_path / "telegram-pending.json").read_text(encoding="utf-8")
    +            )
    +        assert raw == {}
    +
    +    def test_cleanup_expired_handles_malformed_entries(self, tmp_path):
    +        """Non-dict / missing-created_at entries get evicted, not crashed on."""
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            (tmp_path / "telegram-pending.json").write_text(
    +                json.dumps({
    +                    "broken1": "not a dict",
    +                    "broken2": {"user_id": "x"},  # no created_at
    +                    "broken3": {"created_at": "not a number"},
    +                }),
    +                encoding="utf-8",
    +            )
    +            store = PairingStore()
    +            store._cleanup_expired("telegram")
    +            raw = json.loads(
    +                (tmp_path / "telegram-pending.json").read_text(encoding="utf-8")
    +            )
    +        assert raw == {}
    +
    +    def test_approve_code_skips_malformed_entries(self, tmp_path):
    +        """Malformed entries must not crash approve_code's hash loop."""
    +        import time as _time
    +        with patch("gateway.pairing.PAIRING_DIR", tmp_path):
    +            (tmp_path / "telegram-pending.json").write_text(
    +                json.dumps({
    +                    "broken": {"user_id": "x", "created_at": _time.time(),
    +                               "salt": "not-hex", "hash": "doesntmatter"},
    +                }),
    +                encoding="utf-8",
    +            )
    +            store = PairingStore()
    +            # Approving with any code must just return None, not crash.
    +            assert store.approve_code("telegram", "ABCD1234") is None
     
     
     # ---------------------------------------------------------------------------
    @@ -300,9 +488,10 @@ def test_expired_codes_cleaned_up(self, tmp_path):
                 store = PairingStore()
                 code = store.generate_code("telegram", "user1")
     
    -            # Manually expire the code
    +            # Manually expire all pending entries
                 pending = store._load_json(store._pending_path("telegram"))
    -            pending[code]["created_at"] = time.time() - CODE_TTL_SECONDS - 1
    +            for entry_id in pending:
    +                pending[entry_id]["created_at"] = time.time() - CODE_TTL_SECONDS - 1
                 store._save_json(store._pending_path("telegram"), pending)
     
                 # Cleanup happens on next operation
    @@ -314,9 +503,10 @@ def test_expired_code_cannot_be_approved(self, tmp_path):
                 store = PairingStore()
                 code = store.generate_code("telegram", "user1")
     
    -            # Expire it
    +            # Expire all entries
                 pending = store._load_json(store._pending_path("telegram"))
    -            pending[code]["created_at"] = time.time() - CODE_TTL_SECONDS - 1
    +            for entry_id in pending:
    +                pending[entry_id]["created_at"] = time.time() - CODE_TTL_SECONDS - 1
                 store._save_json(store._pending_path("telegram"), pending)
     
                 result = store.approve_code("telegram", code)