diff --git a/cmd_chat/client/_srp_pure.py b/cmd_chat/client/_srp_pure.py index bc728a8..93e0978 100644 --- a/cmd_chat/client/_srp_pure.py +++ b/cmd_chat/client/_srp_pure.py @@ -5,7 +5,7 @@ often fails to build under Termux/Android. This module reproduces the *exact* client-side derivation of that package's pure-Python reference (``srp._pysrp``) with ``rfc5054_enable()`` active, using only the standard library. -Only the surface ``cmd_chat/client/client.py`` consumes is provided: +The client surface ``cmd_chat/client/client.py`` consumes: rfc5054_enable() # module-level toggle SHA256 # hash-alg constant @@ -15,10 +15,19 @@ Only the surface ``cmd_chat/client/client.py`` consumes is provided: usr.verify_session(H_AMK) # sets authenticated flag usr.authenticated() # -> bool -The server uses the real ``srp`` library (see cmd_chat/server/srp_auth.py), so -every constant, padding rule, and hash input here mirrors ``srp._pysrp`` byte -for byte — otherwise the M1 / H_AMK proofs would not match and auth would fail. -Correctness is gated by tests/test_srp_pure_interop.py. +The server surface ``cmd_chat/server/srp_auth.py`` consumes (so a phone/Termux +box with no ``srp`` C-ext can *host* a room, not only join one): + + salt, vkey = create_salted_verification_key(b"chat", pw, hash_alg=SHA256) + svr = Verifier(b"chat", salt, vkey, A, hash_alg=SHA256) + s, B = svr.get_challenge() # -> (salt, B_bytes) + H_AMK = svr.verify_session(M) # -> H_AMK or None + K = svr.get_session_key() + +Both sides mirror ``srp._pysrp`` byte for byte — every constant, padding rule, +and hash input — otherwise the M1 / H_AMK proofs would not match the real +library and auth would fail. Correctness is gated by +tests/test_srp_pure_interop.py (pure↔real in both directions). """ import hashlib @@ -334,3 +343,109 @@ class User: def verify_session(self, host_HAMK) -> None: if self.H_AMK == host_HAMK: self._authenticated = True + + +# ── server side ────────────────────────────────────────────────────────────── +def create_salted_verification_key( + username, + password, + hash_alg=SHA1, + ng_type=NG_2048, + n_hex=None, + g_hex=None, + salt_len=4, +): + """Return ``(salt, vkey)`` for a password — mirrors srp._pysrp exactly. + + ``v = g**x mod N`` where ``x = H(salt, H(username:password))``; the salt is + ``salt_len`` random bytes. This is what the room stores in place of the + password (SRP never sees the password on the wire).""" + if ng_type == NG_CUSTOM and (n_hex is None or g_hex is None): + raise ValueError("Both n_hex and g_hex are required when ng_type = NG_CUSTOM") + hash_class = _hash_map[hash_alg] + N, g = _get_ng(ng_type, n_hex, g_hex) + _s = _long_to_bytes(_get_random(salt_len)) + _v = _long_to_bytes(pow(g, _gen_x(hash_class, _s, username, password), N)) + return _s, _v + + +class Verifier: + """Server side of the SRP-6a exchange, matching srp._pysrp.Verifier's surface.""" + + def __init__( + self, + username, + bytes_s, + bytes_v, + bytes_A, + hash_alg=SHA1, + ng_type=NG_2048, + n_hex=None, + g_hex=None, + bytes_b=None, + ): + if ng_type == NG_CUSTOM and (n_hex is None or g_hex is None): + raise ValueError("Both n_hex and g_hex are required when ng_type = NG_CUSTOM") + if bytes_b and len(bytes_b) != 32: + raise ValueError("32 bytes required for bytes_b") + + self.A = _bytes_to_long(bytes_A) + self.M = None + self.K = None + self.H_AMK = None + self._authenticated = False + self.I = username + self.s = bytes_s + self.v = _bytes_to_long(bytes_v) + + N, g = _get_ng(ng_type, n_hex, g_hex) + hash_class = _hash_map[hash_alg] + k = _bytes_to_long(_H(hash_class, N, g, width=len(_long_to_bytes(N)))) + + self.hash_class = hash_class + self.N = N + self.g = g + self.k = k + + # SRP-6a safety check: abort (no challenge) if A is a multiple of N. + self.safety_failed = False + if (self.A % N) == 0: + self.safety_failed = True + self.B = None + return + + if bytes_b: + self.b = _bytes_to_long(bytes_b) + else: + self.b = _get_random_of_length(32) + self.B = (k * self.v + pow(g, self.b, N)) % N + self.u = _bytes_to_long(_H(hash_class, self.A, self.B, width=len(_long_to_bytes(N)))) + self.S = pow(self.A * pow(self.v, self.u, N), self.b, N) + self.K = hash_class(_long_to_bytes(self.S)).digest() + self.M = _calculate_M(hash_class, N, g, self.I, self.s, self.A, self.B, self.K) + self.H_AMK = _calculate_H_AMK(hash_class, self.A, self.M, self.K) + + def authenticated(self) -> bool: + return self._authenticated + + def get_username(self): + return self.I + + def get_ephemeral_secret(self) -> bytes: + return _long_to_bytes(self.b) + + def get_session_key(self): + return self.K if self._authenticated else None + + def get_challenge(self): + """Return ``(salt, B_bytes)`` or ``(None, None)`` if the safety check failed.""" + if self.safety_failed: + return None, None + return (self.s, _long_to_bytes(self.B)) + + def verify_session(self, user_M): + """Check the client proof M; return H_AMK (bytes) on success, else None.""" + if not self.safety_failed and user_M == self.M: + self._authenticated = True + return self.H_AMK + return None diff --git a/cmd_chat/server/srp_auth.py b/cmd_chat/server/srp_auth.py index d47915d..0ce1921 100644 --- a/cmd_chat/server/srp_auth.py +++ b/cmd_chat/server/srp_auth.py @@ -3,7 +3,10 @@ from dataclasses import dataclass, field from typing import Optional from uuid import uuid4 -import srp +try: + import srp +except ImportError: # no aarch64 wheel / C-ext build (Termux) — use the shim + from ..client import _srp_pure as srp srp.rfc5054_enable() diff --git a/tests/test_srp_pure_interop.py b/tests/test_srp_pure_interop.py index 946111c..d7f3975 100644 --- a/tests/test_srp_pure_interop.py +++ b/tests/test_srp_pure_interop.py @@ -107,3 +107,82 @@ def test_pure_client_full_http_handshake(test_client): assert usr.authenticated(), "server accepted us but H_AMK did not match" assert verify["ws_token"], "no ws_token issued — cannot join the room" + + +# --- Inverse direction: the pure *Verifier* (server side) -------------------- +# When the server itself runs on Termux (no srp C-extension — e.g. a room HOSTED +# ON THE PHONE) srp_auth.py falls back to pure.Verifier. These tests exercise +# that server surface, so a broken constant in the shim's Verifier can't ship. + + +def test_real_user_authenticates_against_pure_verifier(): + """A real srp.User (C-ext client, e.g. a laptop) must authenticate against + the pure Verifier — the exact cross-implementation handshake proven live + when the laptop joined a phone-hosted room.""" + salt, vkey = srp.create_salted_verification_key( + b"chat", PASSWORD, hash_alg=srp.SHA256 + ) + + usr = srp.User(b"chat", PASSWORD, hash_alg=srp.SHA256) + _, A = usr.start_authentication() + + svr = pure.Verifier(b"chat", salt, vkey, A, hash_alg=pure.SHA256) + s, B = svr.get_challenge() + assert B is not None, "pure Verifier aborted the SRP-6a safety check" + + M = usr.process_challenge(s, B) + assert M is not None, "real client failed the SRP-6a safety checks" + + H_AMK = svr.verify_session(M) + assert H_AMK is not None, "pure Verifier rejected the real client's proof M1" + + usr.verify_session(H_AMK) + assert usr.authenticated(), "real client rejected the pure Verifier's H_AMK" + assert svr.authenticated() + # Both sides must derive the identical session key or encryption breaks. + assert usr.get_session_key() == svr.get_session_key() + + +def test_pure_user_authenticates_against_pure_verifier(): + """Both ends on Termux (pure client + pure Verifier) — a room hosted and + joined entirely without the C-extension must still complete the handshake.""" + salt, vkey = pure.create_salted_verification_key( + b"chat", PASSWORD, hash_alg=pure.SHA256 + ) + + usr = pure.User(b"chat", PASSWORD, hash_alg=pure.SHA256) + _, A = usr.start_authentication() + + svr = pure.Verifier(b"chat", salt, vkey, A, hash_alg=pure.SHA256) + s, B = svr.get_challenge() + assert B is not None + + M = usr.process_challenge(s, B) + assert M is not None + + H_AMK = svr.verify_session(M) + assert H_AMK is not None + + usr.verify_session(H_AMK) + assert usr.authenticated() + assert svr.authenticated() + assert usr.get_session_key() == svr.get_session_key() + + +def test_pure_verifier_rejects_wrong_password(): + """Negative control for the pure Verifier: a bad password must never + authenticate, proving the positive tests aren't passing trivially.""" + salt, vkey = srp.create_salted_verification_key( + b"chat", PASSWORD, hash_alg=srp.SHA256 + ) + + usr = srp.User(b"chat", b"wrongpassword", hash_alg=srp.SHA256) + _, A = usr.start_authentication() + + svr = pure.Verifier(b"chat", salt, vkey, A, hash_alg=pure.SHA256) + s, B = svr.get_challenge() + + M = usr.process_challenge(s, B) + assert svr.verify_session(M) is None + assert not svr.authenticated() + assert svr.get_session_key() is None