From d777af9c71fc12c6870520a2f2315754167d5d03 Mon Sep 17 00:00:00 2001 From: Kali Kaneko Date: Wed, 9 Sep 2015 01:08:36 -0400 Subject: move srp_auth to _srp --- src/leap/bonafide/_srp.py | 145 ++++++++++++++++++++++++++++++++++++++++++ src/leap/bonafide/srp_auth.py | 115 --------------------------------- 2 files changed, 145 insertions(+), 115 deletions(-) create mode 100644 src/leap/bonafide/_srp.py delete mode 100644 src/leap/bonafide/srp_auth.py (limited to 'src/leap/bonafide') diff --git a/src/leap/bonafide/_srp.py b/src/leap/bonafide/_srp.py new file mode 100644 index 0000000..dc856ab --- /dev/null +++ b/src/leap/bonafide/_srp.py @@ -0,0 +1,145 @@ +# -*- coding: utf-8 -*- +# srp_auth.py +# Copyright (C) 2015 LEAP +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +SRP Authentication. +""" + +import binascii +import json + +import srp + + +class SRPAuthMechanism(object): + + """ + Implement a protocol-agnostic SRP Authentication mechanism. + """ + + def initialize(self, username, password): + srp_user = srp.User(username.encode('utf-8'), + password.encode('utf-8'), + srp.SHA256, srp.NG_1024) + _, A = srp_user.start_authentication() + return srp_user, A + + def get_handshake_params(self, username, A): + return {'login': bytes(username), 'A': binascii.hexlify(A)} + + def process_handshake(self, srp_user, handshake_response): + challenge = json.loads(handshake_response) + self._check_for_errors(challenge) + salt = challenge.get('salt', None) + B = challenge.get('B', None) + unhex_salt, unhex_B = self._unhex_salt_B(salt, B) + M = srp_user.process_challenge(unhex_salt, unhex_B) + return M + + def get_authentication_params(self, M, A): + # It looks A is not used server side + return {'client_auth': binascii.hexlify(M), 'A': binascii.hexlify(A)} + + def process_authentication(self, authentication_response): + auth = json.loads(authentication_response) + self._check_for_errors(auth) + uuid = auth.get('id', None) + token = auth.get('token', None) + M2 = auth.get('M2', None) + self._check_auth_params(uuid, token, M2) + return uuid, token, M2 + + def verify_authentication(self, srp_user, M2): + unhex_M2 = _safe_unhexlify(M2) + srp_user.verify_session(unhex_M2) + assert srp_user.authenticated() + + def _check_for_errors(self, response): + if 'errors' in response: + msg = response['errors']['base'] + raise SRPAuthError(msg) + + def _unhex_salt_B(self, salt, B): + if salt is None: + raise SRPAuthNoSalt() + if B is None: + raise SRPAuthNoB() + try: + unhex_salt = _safe_unhexlify(salt) + unhex_B = _safe_unhexlify(B) + except (TypeError, ValueError) as e: + raise SRPAuthBadDataFromServer(str(e)) + return unhex_salt, unhex_B + + def _check_auth_params(self, uuid, token, M2): + if not all((uuid, token, M2)): + msg = '%s' % str((M2, uuid, token)) + raise SRPAuthBadDataFromServer(msg) + + +class SRPSignupMechanism(object): + + """ + Implement a protocol-agnostic SRP Registration mechanism. + """ + + def get_signup_params(self, username, password): + salt, verifier = srp.create_salted_verification_key( + bytes(username), bytes(password), + srp.SHA256, srp.NG_1024) + user_data = { + 'user[login]': username, + 'user[password_salt]': binascii.hexlify(salt), + 'user[password_verifier]': binascii.hexlify(verifier)} + return user_data + + def process_signup(self, signup_response): + signup = json.loads(signup_response) + errors = signup.get('errors') + if errors: + msg = 'username ' + errors.get('login')[0] + raise SRPRegistrationError(msg) + else: + username = signup.get('login') + return username + + +def _safe_unhexlify(val): + return binascii.unhexlify(val) \ + if (len(val) % 2 == 0) else binascii.unhexlify('0' + val) + + +class SRPAuthError(Exception): + """ + Base exception for srp authentication errors + """ + + +class SRPAuthNoSalt(SRPAuthError): + message = 'The server didn\'t send the salt parameter' + + +class SRPAuthNoB(SRPAuthError): + message = 'The server didn\'t send the B parameter' + + +class SRPAuthBadDataFromServer(SRPAuthError): + pass + +class SRPRegistrationError(Exception): + pass + diff --git a/src/leap/bonafide/srp_auth.py b/src/leap/bonafide/srp_auth.py deleted file mode 100644 index d48214f..0000000 --- a/src/leap/bonafide/srp_auth.py +++ /dev/null @@ -1,115 +0,0 @@ -# -*- coding: utf-8 -*- -# srp_auth.py -# Copyright (C) 2015 LEAP -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -""" -SRP Authentication. -""" - -import binascii -import json - -import srp - - - -class SRPAuthMechanism(object): - - """ - Implement a protocol-agnostic SRP Authentication mechanism. - """ - - def initialize(self, username, password): - srp_user = srp.User(username.encode('utf-8'), - password.encode('utf-8'), - srp.SHA256, srp.NG_1024) - _, A = srp_user.start_authentication() - return srp_user, A - - def get_handshake_params(self, username, A): - return {'login': bytes(username), 'A': binascii.hexlify(A)} - - def process_handshake(self, srp_user, handshake_response): - challenge = json.loads(handshake_response) - self._check_for_errors(challenge) - salt = challenge.get('salt', None) - B = challenge.get('B', None) - unhex_salt, unhex_B = self._unhex_salt_B(salt, B) - M = srp_user.process_challenge(unhex_salt, unhex_B) - return M - - def get_authentication_params(self, M, A): - # It looks A is not used server side - return {'client_auth': binascii.hexlify(M), 'A': binascii.hexlify(A)} - - def process_authentication(self, authentication_response): - auth = json.loads(authentication_response) - self._check_for_errors(auth) - uuid = auth.get('id', None) - token = auth.get('token', None) - M2 = auth.get('M2', None) - self._check_auth_params(uuid, token, M2) - return uuid, token, M2 - - def verify_authentication(self, srp_user, M2): - unhex_M2 = _safe_unhexlify(M2) - srp_user.verify_session(unhex_M2) - assert srp_user.authenticated() - - def _check_for_errors(self, response): - if 'errors' in response: - msg = response['errors']['base'] - raise SRPAuthError(msg) - - def _unhex_salt_B(self, salt, B): - if salt is None: - raise SRPAuthNoSalt() - if B is None: - raise SRPAuthNoB() - try: - unhex_salt = _safe_unhexlify(salt) - unhex_B = _safe_unhexlify(B) - except (TypeError, ValueError) as e: - raise SRPAuthBadDataFromServer(str(e)) - return unhex_salt, unhex_B - - def _check_auth_params(self, uuid, token, M2): - if not all((uuid, token, M2)): - msg = '%s' % str((M2, uuid, token)) - raise SRPAuthBadDataFromServer(msg) - - -def _safe_unhexlify(val): - return binascii.unhexlify(val) \ - if (len(val) % 2 == 0) else binascii.unhexlify('0' + val) - - -class SRPAuthError(Exception): - """ - Base exception for srp authentication errors - """ - - -class SRPAuthNoSalt(SRPAuthError): - message = 'The server didn\'t send the salt parameter' - - -class SRPAuthNoB(SRPAuthError): - message = 'The server didn\'t send the B parameter' - - -class SRPAuthBadDataFromServer(SRPAuthError): - pass -- cgit v1.2.3