mirror of https://github.com/huashengdun/webssh
Added test for bad authentication type
parent
105c1f62ee
commit
e1fbc417fd
|
@ -72,11 +72,13 @@ class Server(paramiko.ServerInterface):
|
||||||
|
|
||||||
def check_auth_publickey(self, username, key):
|
def check_auth_publickey(self, username, key):
|
||||||
print('Auth attempt with username: {!r} & key: {!r}'.format(username, u(hexlify(key.get_fingerprint())))) # noqa
|
print('Auth attempt with username: {!r} & key: {!r}'.format(username, u(hexlify(key.get_fingerprint())))) # noqa
|
||||||
if (username == 'robey') and (key == self.good_pub_key):
|
if (username in ['robey', 'keyonly']) and (key == self.good_pub_key):
|
||||||
return paramiko.AUTH_SUCCESSFUL
|
return paramiko.AUTH_SUCCESSFUL
|
||||||
return paramiko.AUTH_FAILED
|
return paramiko.AUTH_FAILED
|
||||||
|
|
||||||
def get_allowed_auths(self, username):
|
def get_allowed_auths(self, username):
|
||||||
|
if username == 'keyonly':
|
||||||
|
return 'publickey'
|
||||||
return 'password,publickey'
|
return 'password,publickey'
|
||||||
|
|
||||||
def check_channel_exec_request(self, channel, command):
|
def check_channel_exec_request(self, channel, command):
|
||||||
|
|
|
@ -28,14 +28,14 @@ class TestApp(AsyncHTTPTestCase):
|
||||||
running = [True]
|
running = [True]
|
||||||
sshserver_port = 2200
|
sshserver_port = 2200
|
||||||
body = 'hostname=127.0.0.1&port={}&username=robey&password=foo'.format(sshserver_port) # noqa
|
body = 'hostname=127.0.0.1&port={}&username=robey&password=foo'.format(sshserver_port) # noqa
|
||||||
body_dict = {
|
|
||||||
|
def get_app(self):
|
||||||
|
self.body_dict = {
|
||||||
'hostname': '127.0.0.1',
|
'hostname': '127.0.0.1',
|
||||||
'port': str(sshserver_port),
|
'port': str(self.sshserver_port),
|
||||||
'username': 'robey',
|
'username': 'robey',
|
||||||
'password': ''
|
'password': ''
|
||||||
}
|
}
|
||||||
|
|
||||||
def get_app(self):
|
|
||||||
loop = self.io_loop
|
loop = self.io_loop
|
||||||
options.debug = False
|
options.debug = False
|
||||||
options.policy = random.choice(['warning', 'autoadd'])
|
options.policy = random.choice(['warning', 'autoadd'])
|
||||||
|
@ -449,3 +449,19 @@ class TestApp(AsyncHTTPTestCase):
|
||||||
yield client.fetch(url, method='POST', body=body)
|
yield client.fetch(url, method='POST', body=body)
|
||||||
self.assertEqual(ctx.exception.code, 400)
|
self.assertEqual(ctx.exception.code, 400)
|
||||||
self.assertIn('Bad Request', ctx.exception.message)
|
self.assertIn('Bad Request', ctx.exception.message)
|
||||||
|
|
||||||
|
@tornado.testing.gen_test
|
||||||
|
def test_app_with_user_keyonly_for_bad_authentication_type(self):
|
||||||
|
url = self.get_url('/')
|
||||||
|
client = self.get_http_client()
|
||||||
|
response = yield client.fetch(url)
|
||||||
|
self.assertEqual(response.code, 200)
|
||||||
|
|
||||||
|
self.body_dict.update(username='keyonly', password='foo')
|
||||||
|
body = urlencode(self.body_dict)
|
||||||
|
response = yield client.fetch(url, method='POST', body=body)
|
||||||
|
self.assertEqual(response.code, 200)
|
||||||
|
data = json.loads(to_str(response.body))
|
||||||
|
self.assertIsNone(data['id'])
|
||||||
|
self.assertIsNone(data['encoding'])
|
||||||
|
self.assertIn('Bad authentication type', data['status'])
|
||||||
|
|
Loading…
Reference in New Issue