diff options
| -rw-r--r-- | src/leap/base/auth.py | 17 | ||||
| -rw-r--r-- | src/leap/base/tests/test_auth.py | 85 |
2 files changed, 94 insertions, 8 deletions
diff --git a/src/leap/base/auth.py b/src/leap/base/auth.py index 5a9ebe1d..d91e138b 100644 --- a/src/leap/base/auth.py +++ b/src/leap/base/auth.py @@ -119,12 +119,11 @@ safe_unhexlify = lambda x: binascii.unhexlify(x) \ class SRPAuth(requests.auth.AuthBase): - def __init__(self, username, password): + def __init__(self, username, password, server=SERVER, verify=True): self.username = username self.password = password - - # XXX init something similar to - # SERVER... + self.server = server + self.verify = verify self.init_data = None self.session = requests.session() @@ -153,8 +152,9 @@ class SRPAuth(requests.auth.AuthBase): def get_init_data(self): init_session = self.session.post( - SERVER + '/sessions', - data=self.get_auth_data()) + self.server + '/sessions', + data=self.get_auth_data(), + verify=self.verify) self.init_data = self.get_data(init_session) return self.init_data @@ -174,8 +174,9 @@ class SRPAuth(requests.auth.AuthBase): ) auth_result = self.session.put( - SERVER + '/sessions/' + self.username, - data={'client_auth': binascii.hexlify(self.M)}) + self.server + '/sessions/' + self.username, + data={'client_auth': binascii.hexlify(self.M)}, + verify=self.verify) # XXX check for errors auth_data = self.get_data(auth_result) diff --git a/src/leap/base/tests/test_auth.py b/src/leap/base/tests/test_auth.py new file mode 100644 index 00000000..7b5df99e --- /dev/null +++ b/src/leap/base/tests/test_auth.py @@ -0,0 +1,85 @@ +import cgi +import binascii +import json +import requests +import urlparse +try: + import unittest2 as unittest +except ImportError: + import unittest + +from mock import (patch, Mock) + +#XXX should be moved to a general location +from leap.eip.tests.test_checks import NoLogRequestHandler + +from leap.testing.basetest import BaseLeapTest +from BaseHTTPServer import BaseHTTPRequestHandler +from leap.testing.https_server import BaseHTTPSServerTestCase + +from leap.base.auth import SRPAuth + +USERNAME = "0ACOJK" +PASSWORD = "WG3HD06E7ZF3" +INIT_DATA = {u'B': u'd74a9f592193bba8a818dcf500f412f60ce1b999aa9b5166f59fbe02aee97be9ec71a5d62fd16dedd973041efd4c7de0568c0d0c38a3806c78fc96f9ffa59dde89e5a04969905a83b8e700ee9c03b5636ad99624ed1514319b3bdac10cde498c8e064adf2fe04bfc5ee5df0dd06693961190a16caa182c090e59ac52feec693e', + u'salt': u'd09ed33e'} +AUTH_RESULT = {u'M2': u'b040d0cd7ab1f93c4e87ffccdec07491782f2af303ad14f33dc4f0b4b2e40824'} + + +class SRP_SERVER_HTTPSTests(BaseHTTPSServerTestCase, BaseLeapTest): + class request_handler(NoLogRequestHandler, BaseHTTPRequestHandler): + responses = { + '/': [ 'OK', '' ], + '/1/sessions': [ json.dumps(INIT_DATA) ], + '/1/sessions/' + USERNAME: [ json.dumps(AUTH_RESULT) ] + } + + def do_GET(self): + path = urlparse.urlparse(self.path) + message = '\n'.join(self.responses.get( + path.path, None)) + self.send_response(200) + self.end_headers() + self.wfile.write(message) + + def do_PUT(self): + form = cgi.FieldStorage( + fp=self.rfile, + headers=self.headers, + environ={'REQUEST_METHOD': 'PUT', + 'CONTENT_TYPE': self.headers['Content-Type'], + }) + data = dict( + (key, form[key].value) for key in form.keys()) + path = urlparse.urlparse(self.path) + message = '\n'.join( + self.responses.get( + path.path, '')) + + self.send_response(200) + self.end_headers() + self.wfile.write(message) + + def do_POST(self): + form = cgi.FieldStorage( + fp=self.rfile, + headers=self.headers, + environ={'REQUEST_METHOD': 'POST', + 'CONTENT_TYPE': self.headers['Content-Type'], + }) + data = dict( + (key, form[key].value) for key in form.keys()) + path = urlparse.urlparse(self.path) + message = '\n'.join( + self.responses.get( + path.path, '')) + + self.send_response(200) + self.end_headers() + self.wfile.write(message) + + def test_srp_authenticate(self): + srp_auth = SRPAuth(USERNAME, PASSWORD, + "https://%s/1" % (self.get_server()), verify=False) + + srp_auth.authenticate() |