Commit 6067533e authored by Mukund Sivaraman's avatar Mukund Sivaraman
Browse files

Merge branch 'trac2641_3'

parents 8d178815 85a61f11
......@@ -33,6 +33,7 @@ import inspect
import pprint
import ssl, socket
import os, time, random, re
import os.path
import getpass
from hashlib import sha1
import csv
......@@ -214,22 +215,16 @@ WARNING: Python readline module isn't available, so the command line editor
return True
def __print_check_ssl_msg(self):
self._print("Please check the logs of b10-cmdctl, there may "
"be a problem accepting SSL connections, such "
"as a permission problem on the server "
"certificate file.")
def _try_login(self, username, password):
'''
Attempts to log in to cmdctl by sending a POST with
the given username and password.
On success of the POST (mind, not the login, only the network
operation), returns a tuple (response, data).
On failure, raises a FailToLogin exception, and prints some
information on the failure.
This call is essentially 'private', but made 'protected' for
easier testing.
Attempts to log into cmdctl by sending a POST with the given
username and password. On success of the POST (not the login,
but the network operation), it returns a tuple (response, data).
We check for some failures such as SSL errors and socket errors
which could happen due to the environment in which BIND 10 runs.
On failure, it raises a FailToLogin exception and prints some
information on the failure. This call is essentially 'private',
but made 'protected' for easier testing.
'''
param = {'username': username, 'password' : password}
try:
......@@ -237,16 +232,8 @@ WARNING: Python readline module isn't available, so the command line editor
data = response.read().decode()
# return here (will raise error after try block)
return (response, data)
except ssl.SSLError as err:
self._print("SSL error while sending login information: ", err)
if err.errno == ssl.SSL_ERROR_EOF:
self.__print_check_ssl_msg()
except socket.error as err:
self._print("Socket error while sending login information: ", err)
# An SSL setup error can also bubble up as a plain CONNRESET...
# (on some systems it usually does)
if err.errno == errno.ECONNRESET:
self.__print_check_ssl_msg()
except (ssl.SSLError, socket.error) as err:
self._print('Error while sending login information:', err)
pass
raise FailToLogin()
......@@ -270,9 +257,21 @@ WARNING: Python readline module isn't available, so the command line editor
# No valid logins were found, prompt the user for a username/password
count = 0
self._print('No stored password file found, please see sections '
'"Configuration specification for b10-cmdctl" and "bindctl '
'command-line options" of the BIND 10 guide.')
if not os.path.exists(self.csv_file_dir + CSV_FILE_NAME):
self._print('\nNo stored password file found.\n\n'
'When the system is first set up you need to create '
'at least one user account.\n'
'For information on how to set up a BIND 10 system, '
'please check see the\n'
'BIND 10 Guide: \n\n'
'http://bind10.isc.org/docs/bind10-guide.html#quick-start-auth-dns\n\n'
'If a user account has been set up, please check the '
'b10-cmdctl log for other\n'
'information.\n')
else:
self._print('Login failed: either the user name or password is '
'invalid.\n')
while True:
count = count + 1
if count > 3:
......@@ -317,14 +316,14 @@ WARNING: Python readline module isn't available, so the command line editor
return {}
def send_POST(self, url, post_param = None):
def send_POST(self, url, post_param=None):
'''Send POST request to cmdctl, session id is send with the name
'cookie' in header.
Format: /module_name/command_name
parameters of command is encoded as a map
'''
param = None
if (len(post_param) != 0):
if post_param is not None and len(post_param) != 0:
param = json.dumps(post_param)
headers = {"cookie" : self.session_id}
......@@ -938,5 +937,3 @@ WARNING: Python readline module isn't available, so the command line editor
if data != "" and data != "{}":
self._print(json.dumps(json.loads(data), sort_keys=True,
indent=4))
......@@ -26,6 +26,7 @@ import http.client
import pwd
import getpass
import re
import json
from optparse import OptionParser
from isc.config.config_data import ConfigData, MultiConfigData
from isc.config.module_spec import ModuleSpec
......@@ -386,7 +387,7 @@ class TestConfigCommands(unittest.TestCase):
self.tool.send_POST = send_POST_raiseImmediately
self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar")
expected_printed_messages.append(
'Socket error while sending login information: test error')
'Error while sending login information: test error')
self.__check_printed_messages(expected_printed_messages)
def create_send_POST_raiseOnRead(exception):
......@@ -405,7 +406,7 @@ class TestConfigCommands(unittest.TestCase):
create_send_POST_raiseOnRead(socket.error("read error"))
self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar")
expected_printed_messages.append(
'Socket error while sending login information: read error')
'Error while sending login information: read error')
self.__check_printed_messages(expected_printed_messages)
# connection reset
......@@ -415,13 +416,7 @@ class TestConfigCommands(unittest.TestCase):
create_send_POST_raiseOnRead(exc)
self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar")
expected_printed_messages.append(
'Socket error while sending login information: '
'connection reset')
expected_printed_messages.append(
'Please check the logs of b10-cmdctl, there may be a '
'problem accepting SSL connections, such as a permission '
'problem on the server certificate file.'
)
'Error while sending login information: connection reset')
self.__check_printed_messages(expected_printed_messages)
# 'normal' SSL error
......@@ -430,7 +425,7 @@ class TestConfigCommands(unittest.TestCase):
create_send_POST_raiseOnRead(exc)
self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar")
expected_printed_messages.append(
'SSL error while sending login information: .*')
'Error while sending login information: .*')
self.__check_printed_messages(expected_printed_messages)
# 'EOF' SSL error
......@@ -440,12 +435,7 @@ class TestConfigCommands(unittest.TestCase):
create_send_POST_raiseOnRead(exc)
self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar")
expected_printed_messages.append(
'SSL error while sending login information: .*')
expected_printed_messages.append(
'Please check the logs of b10-cmdctl, there may be a '
'problem accepting SSL connections, such as a permission '
'problem on the server certificate file.'
)
'Error while sending login information: .*')
self.__check_printed_messages(expected_printed_messages)
# any other exception should be passed through
......@@ -457,6 +447,44 @@ class TestConfigCommands(unittest.TestCase):
finally:
self.tool.send_POST = orig_send_POST
def test_try_login_calls_cmdctl(self):
# Make sure _try_login() makes the right API call to cmdctl.
orig_conn = self.tool.conn
try:
class MyConn:
def __init__(self):
self.method = None
self.url = None
self.param = None
self.headers = None
def request(self, method, url, param, headers):
self.method = method
self.url = url
self.param = param
self.headers = headers
def getresponse(self):
class MyResponse:
def __init__(self):
self.status = http.client.OK
def read(self):
class MyData:
def decode(self):
return json.dumps(True)
return MyData()
return MyResponse()
self.tool.conn = MyConn()
self.assertTrue(self.tool._try_login('user32', 'pass64'))
self.assertEqual(self.tool.conn.method, 'POST')
self.assertEqual(self.tool.conn.url, '/login')
self.assertEqual(json.loads(self.tool.conn.param),
{"password": "pass64", "username": "user32"})
self.assertIn('cookie', self.tool.conn.headers)
finally:
self.tool.conn = orig_conn
def test_run(self):
def login_to_cmdctl():
return True
......
......@@ -11,20 +11,18 @@ pylogmessagedir = $(pyexecdir)/isc/log_messages/
b10_cmdctldir = $(pkgdatadir)
USERSFILES = cmdctl-accounts.csv
CERTFILES = cmdctl-keyfile.pem cmdctl-certfile.pem
b10_cmdctl_DATA = cmdctl.spec
EXTRA_DIST = $(USERSFILES)
CLEANFILES= b10-cmdctl cmdctl.pyc cmdctl.spec
CLEANFILES += $(PYTHON_LOGMSGPKG_DIR)/work/cmdctl_messages.py
CLEANFILES += $(PYTHON_LOGMSGPKG_DIR)/work/cmdctl_messages.pyc
man_MANS = b10-cmdctl.8 b10-certgen.1
DISTCLEANFILES = $(man_MANS) cmdctl-certfile.pem cmdctl-keyfile.pem
EXTRA_DIST += $(man_MANS) b10-certgen.xml b10-cmdctl.xml cmdctl_messages.mes
EXTRA_DIST = $(man_MANS) b10-certgen.xml b10-cmdctl.xml cmdctl_messages.mes
EXTRA_DIST += cmdctl-accounts.csv
if GENERATE_DOCS
......
......@@ -97,6 +97,11 @@ def check_file(file_name):
class SecureHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
'''https connection request handler.
Currently only GET and POST are supported. '''
def __init__(self, request, client_address, server):
http.server.BaseHTTPRequestHandler.__init__(self, request,
client_address, server)
self.session_id = None
def do_GET(self):
'''The client should send its session id in header with
the name 'cookie'
......@@ -121,7 +126,7 @@ class SecureHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
return self.server.get_reply_data_for_GET(id, module)
def _is_session_valid(self):
return self.session_id
return self.session_id is not None
def _is_user_logged_in(self):
login_time = self.server.user_sessions.get(self.session_id)
......@@ -171,7 +176,7 @@ class SecureHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
is_user_valid, error_info = self._check_user_name_and_pwd()
if is_user_valid:
self.server.save_user_session_id(self.session_id)
return http.client.OK, ["login success "]
return http.client.OK, ["login success"]
else:
return http.client.UNAUTHORIZED, error_info
......
......@@ -25,8 +25,8 @@ endif
echo Running test: $$pytest ; \
$(LIBRARY_PATH_PLACEHOLDER) \
PYTHONPATH=$(COMMON_PYTHON_PATH):$(abs_top_builddir)/src/bin/cmdctl \
CMDCTL_BUILD_PATH=$(abs_top_builddir)/src/bin/cmdctl \
CMDCTL_SRC_PATH=$(abs_top_srcdir)/src/bin/cmdctl \
CMDCTL_BUILD_PATH=$(abs_top_builddir)/src/bin/cmdctl \
B10_LOCKFILE_DIR_FROM_BUILD=$(abs_top_builddir) \
$(PYCOVERAGE_RUN) $(abs_srcdir)/$$pytest || exit ; \
done
......@@ -172,10 +172,7 @@ class TestCertGenTool(unittest.TestCase):
"""
Tests a few pre-created certificates with the -c option
"""
if ('CMDCTL_SRC_PATH' in os.environ):
path = os.environ['CMDCTL_SRC_PATH'] + "/tests/testdata/"
else:
path = "testdata/"
path = os.environ['CMDCTL_SRC_PATH'] + '/tests/testdata/'
self.validate_certificate(10, path + 'expired-certfile.pem')
self.validate_certificate(100, path + 'mangled-certfile.pem')
self.validate_certificate(17, path + 'noca-certfile.pem')
......
......@@ -33,7 +33,7 @@ BUILD_FILE_PATH = os.environ['CMDCTL_BUILD_PATH'] + os.sep
# Rewrite the class for unittest.
class MySecureHTTPRequestHandler(SecureHTTPRequestHandler):
def __init__(self):
pass
self.session_id = None
def send_response(self, rcode):
self.rcode = rcode
......@@ -41,19 +41,6 @@ class MySecureHTTPRequestHandler(SecureHTTPRequestHandler):
def end_headers(self):
pass
def do_GET(self):
self.wfile = open('tmp.file', 'wb')
super().do_GET()
self.wfile.close()
os.remove('tmp.file')
def do_POST(self):
self.wfile = open("tmp.file", 'wb')
super().do_POST()
self.wfile.close()
os.remove('tmp.file')
class FakeSecureHTTPServer(SecureHTTPServer):
def __init__(self):
self.user_sessions = {}
......@@ -93,13 +80,22 @@ class TestSecureHTTPRequestHandler(unittest.TestCase):
self.handler.server.user_sessions = {}
self.handler.server._user_infos = {}
self.handler.headers = {}
self.handler.rfile = open("check.tmp", 'w+b')
self.handler.rfile = open('input.tmp', 'w+b')
self.handler.wfile = open('output.tmp', 'w+b')
def tearDown(self):
sys.stdout.close()
sys.stdout = self.old_stdout
self.handler.wfile.close()
os.remove('output.tmp')
self.handler.rfile.close()
os.remove('check.tmp')
os.remove('input.tmp')
def test_is_session_valid(self):
self.assertIsNone(self.handler.session_id)
self.assertFalse(self.handler._is_session_valid())
self.handler.session_id = 4234
self.assertTrue(self.handler._is_session_valid())
def test_parse_request_path(self):
self.handler.path = ''
......@@ -160,7 +156,7 @@ class TestSecureHTTPRequestHandler(unittest.TestCase):
self.handler.do_GET()
self.assertEqual(self.handler.rcode, http.client.OK)
def test_user_logged_in(self):
def test_is_user_logged_in(self):
self.handler.server.user_sessions = {}
self.handler.session_id = 12345
self.assertTrue(self.handler._is_user_logged_in() == False)
......@@ -294,6 +290,68 @@ class TestSecureHTTPRequestHandler(unittest.TestCase):
rcode, reply = self.handler._handle_post_request()
self.assertEqual(http.client.BAD_REQUEST, rcode)
def test_handle_login(self):
orig_is_user_logged_in = self.handler._is_user_logged_in
orig_check_user_name_and_pwd = self.handler._check_user_name_and_pwd
try:
def create_is_user_logged_in(status):
'''Create a replacement _is_user_logged_in() method.'''
def my_is_user_logged_in():
return status
return my_is_user_logged_in
# Check case where _is_user_logged_in() returns True
self.handler._is_user_logged_in = create_is_user_logged_in(True)
self.handler.headers['cookie'] = 12345
self.handler.path = '/login'
self.handler.do_POST()
self.assertEqual(self.handler.rcode, http.client.OK)
self.handler.wfile.seek(0, 0)
d = self.handler.wfile.read()
self.assertEqual(json.loads(d.decode()),
['user has already login'])
# Clear the output
self.handler.wfile.seek(0, 0)
self.handler.wfile.truncate()
# Check case where _is_user_logged_in() returns False
self.handler._is_user_logged_in = create_is_user_logged_in(False)
def create_check_user_name_and_pwd(status, error_info=None):
'''Create a replacement _check_user_name_and_pwd() method.'''
def my_check_user_name_and_pwd():
return status, error_info
return my_check_user_name_and_pwd
# (a) Check case where _check_user_name_and_pwd() returns
# valid user status
self.handler._check_user_name_and_pwd = \
create_check_user_name_and_pwd(True)
self.handler.do_POST()
self.assertEqual(self.handler.rcode, http.client.OK)
self.handler.wfile.seek(0, 0)
d = self.handler.wfile.read()
self.assertEqual(json.loads(d.decode()), ['login success'])
# Clear the output
self.handler.wfile.seek(0, 0)
self.handler.wfile.truncate()
# (b) Check case where _check_user_name_and_pwd() returns
# invalid user status
self.handler._check_user_name_and_pwd = \
create_check_user_name_and_pwd(False, ['login failed'])
self.handler.do_POST()
self.assertEqual(self.handler.rcode, http.client.UNAUTHORIZED)
self.handler.wfile.seek(0, 0)
d = self.handler.wfile.read()
self.assertEqual(json.loads(d.decode()), ['login failed'])
finally:
self.handler._is_user_logged_in = orig_is_user_logged_in
self.handler._check_user_name_and_pwd = orig_check_user_name_and_pwd
class MyCommandControl(CommandControl):
def _get_modules_specification(self):
return {}
......@@ -470,6 +528,12 @@ class TestSecureHTTPServer(unittest.TestCase):
self.assertEqual(1, len(self.server._user_infos))
self.assertTrue('root' in self.server._user_infos)
def test_get_user_info(self):
self.assertIsNone(self.server.get_user_info('root'))
self.server._create_user_info(SRC_FILE_PATH + 'cmdctl-accounts.csv')
self.assertIn('6f0c73bd33101a5ec0294b3ca39fec90ef4717fe',
self.server.get_user_info('root'))
def test_check_file(self):
# Just some file that we know exists
file_name = BUILD_FILE_PATH + 'cmdctl-keyfile.pem'
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment