r47036 - improve type clarity
hawkowl-TA+aISz0psMTMxyoc4vAAJOcrHinNvQL0E9HWUfgJXw@public.gmane.org Sun, 20 Mar 2016 19:19:25 -0600 (MDT)
| Newsgroups | gmane.comp.python.twisted.commits |
|---|---|
| Message-ID | <[email protected]> |
Author: hawkowl
Date: Sun Mar 20 19:19:21 2016
New Revision: 47036
Modified:
branches/clarity-sshuserauth-8240/twisted/conch/ssh/userauth.py
branches/clarity-sshuserauth-8240/twisted/conch/test/test_userauth.py
Log:
improve type clarity
Modified: branches/clarity-sshuserauth-8240/twisted/conch/ssh/userauth.py
==============================================================================
--- branches/clarity-sshuserauth-8240/twisted/conch/ssh/userauth.py (original)
+++ branches/clarity-sshuserauth-8240/twisted/conch/ssh/userauth.py Sun Mar 20 19:19:21 2016
@@ -9,7 +9,10 @@
Maintainer: Paul Swartz
"""
+from __future__ import absolute_import, division
+
import struct
+
from twisted.conch import error, interfaces
from twisted.conch.ssh import keys, transport, service
from twisted.conch.ssh.common import NS, getNS
@@ -72,8 +75,8 @@
passwordDelay = 1 # number of seconds to delay on a failed password
clock = reactor
interfaceToMethod = {
- credentials.ISSHPrivateKey : 'publickey',
- credentials.IUsernamePassword : 'password',
+ credentials.ISSHPrivateKey : b'publickey',
+ credentials.IUsernamePassword : b'password',
}
@@ -97,8 +100,8 @@
if not self.transport.isEncrypted('in'):
# don't let us transport password in plaintext
- if 'password' in self.supportedAuthentications:
- self.supportedAuthentications.remove('password')
+ if b'password' in self.supportedAuthentications:
+ self.supportedAuthentications.remove(b'password')
self._cancelLoginTimeout = self.clock.callLater(
self.loginTimeout,
self.timeoutAuthentication)
@@ -122,7 +125,7 @@
self._cancelLoginTimeout = None
self.transport.sendDisconnect(
transport.DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE,
- 'you took too long')
+ b'you took too long')
def tryAuth(self, kind, user, data):
@@ -165,7 +168,7 @@
string method
<authentication specific data>
- @type packet: C{str}
+ @type packet: C{bytes}
"""
user, nextService, method, rest = getNS(packet, 3)
if user != self.user or nextService != self.nextService:
@@ -198,7 +201,7 @@
raise error.ConchError('could not get next service: %s'
% self.nextService)
log.msg('%s authenticated with %s' % (self.user, self.method))
- self.transport.sendPacket(MSG_USERAUTH_SUCCESS, '')
+ self.transport.sendPacket(MSG_USERAUTH_SUCCESS, b'')
self.transport.setService(service())
@@ -212,7 +215,7 @@
"""
reason.trap(error.NotEnoughAuthentication)
self.transport.sendPacket(MSG_USERAUTH_FAILURE,
- NS(','.join(self.supportedAuthentications)) + '\xff')
+ NS(b','.join(self.supportedAuthentications)) + b'\xff')
def _ebBadAuth(self, reason):
@@ -239,11 +242,11 @@
if self.loginAttempts > self.attemptsBeforeDisconnect:
self.transport.sendDisconnect(
transport.DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE,
- 'too many bad auths')
+ b'too many bad auths')
return
self.transport.sendPacket(
MSG_USERAUTH_FAILURE,
- NS(','.join(self.supportedAuthentications)) + '\x00')
+ NS(b','.join(self.supportedAuthentications)) + b'\x00')
def auth_publickey(self, packet):
@@ -323,11 +326,11 @@
first, in order of preference, if supported by the server
@type preferredOrder: C{list}
@ivar user: the name of the user to authenticate as
- @type user: C{str}
+ @type user: C{bytes}
@ivar instance: the service to start after authentication has finished
@type instance: L{service.SSHService}
@ivar authenticatedWith: a list of strings of authentication methods we've tried
- @type authenticatedWith: C{list} of C{str}
+ @type authenticatedWith: C{list} of C{bytes}
@ivar triedPublicKeys: a list of public key objects that we've tried to
authenticate with
@type triedPublicKeys: C{list} of L{Key}
@@ -338,7 +341,7 @@
name = 'ssh-userauth'
- preferredOrder = ['publickey', 'password', 'keyboard-interactive']
+ preferredOrder = [b'publickey', b'password', b'keyboard-interactive']
def __init__(self, user, instance):
@@ -350,7 +353,7 @@
self.authenticatedWith = []
self.triedPublicKeys = []
self.lastPublicKey = None
- self.askForAuth('none', '')
+ self.askForAuth(b'none', b'')
def askForAuth(self, kind, extraData):
@@ -358,9 +361,9 @@
Send a MSG_USERAUTH_REQUEST.
@param kind: the authentication method to try.
- @type kind: C{str}
+ @type kind: C{bytes}
@param extraData: method-specific data to go in the packet
- @type extraData: C{str}
+ @type extraData: C{bytes}
"""
self.lastAuth = kind
self.transport.sendPacket(MSG_USERAUTH_REQUEST, NS(self.user) +
@@ -374,7 +377,7 @@
@param kind: the authentication method
@type kind: C{str}
"""
- kind = kind.replace('-', '_')
+ kind = kind.replace(b'-', b'_')
log.msg('trying to auth with %s' % (kind,))
f = getattr(self,'auth_%s' % (kind,), None)
if f:
@@ -386,7 +389,7 @@
Generic callback for a failed authentication attempt. Respond by
asking for the list of accepted methods (the 'none' method)
"""
- self.askForAuth('none', '')
+ self.askForAuth(b'none', b'')
def ssh_USERAUTH_SUCCESS(self, packet):
@@ -441,7 +444,7 @@
# put the element at the end of the list.
return len(self.preferredOrder)
- canContinue = sorted([meth for meth in canContinue.split(',')
+ canContinue = sorted([meth for meth in canContinue.split(b',')
if meth not in self.authenticatedWith],
key=orderByPreference)
@@ -453,11 +456,11 @@
if result:
return
try:
- method = iterator.next()
+ method = next(iterator)
except StopIteration:
self.transport.sendDisconnect(
transport.DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE,
- 'no more authentication methods available')
+ b'no more authentication methods available')
else:
d = defer.maybeDeferred(self.tryAuth, method)
d.addCallback(self._cbUserauthFailure, iterator)
@@ -475,7 +478,7 @@
if func is not None:
return func(packet)
else:
- self.askForAuth('none', '')
+ self.askForAuth(b'none', b'')
def ssh_USERAUTH_PK_OK_publickey(self, packet):
@@ -489,7 +492,7 @@
'\x01' + NS(publicKey.sshType()) + NS(publicKey.blob()))
d = self.signData(publicKey, b)
if not d:
- self.askForAuth('none', '')
+ self.askForAuth(b'none', b'')
# this will fail, we'll move on
return
d.addCallback(self._cbSignedData)
@@ -504,7 +507,7 @@
"""
prompt, language, rest = getNS(packet, 2)
self._oldPass = self._newPass = None
- d = self.getPassword('Old Password: ')
+ d = self.getPassword(b'Old Password: ')
d = d.addCallbacks(self._setOldPass, self._ebAuth)
d.addCallback(lambda ignored: self.getPassword(prompt))
d.addCallbacks(self._setNewPass, self._ebAuth)
@@ -539,7 +542,7 @@
@type signedData: C{str}
"""
publicKey = self.lastPublicKey
- self.askForAuth('publickey', '\x01' + NS(publicKey.sshType()) +
+ self.askForAuth(b'publickey', b'\x01' + NS(publicKey.sshType()) +
NS(publicKey.blob()) + NS(signedData))
@@ -564,7 +567,7 @@
"""
op = self._oldPass
self._oldPass = None
- self.askForAuth('password', '\xff' + NS(op) + NS(np))
+ self.askForAuth(b'password', b'\xff' + NS(op) + NS(np))
def _cbGenericAnswers(self, responses):
@@ -602,7 +605,7 @@
self.lastPublicKey = publicKey
self.triedPublicKeys.append(publicKey)
log.msg('using key of type %s' % publicKey.type())
- self.askForAuth('publickey', '\x00' + NS(publicKey.sshType()) +
+ self.askForAuth(b'publickey', b'\x00' + NS(publicKey.sshType()) +
NS(publicKey.blob()))
return True
else:
@@ -633,7 +636,7 @@
@rtype: C{bool}
"""
log.msg('authing with keyboard-interactive')
- self.askForAuth('keyboard-interactive', NS('') + NS(''))
+ self.askForAuth(b'keyboard-interactive', NS(b'') + NS(b''))
return True
@@ -645,7 +648,7 @@
@param password: the password the user entered
@type password: C{str}
"""
- self.askForAuth('password', '\x00' + NS(password))
+ self.askForAuth(b'password', b'\x00' + NS(password))
def signData(self, publicKey, signData):
@@ -744,8 +747,8 @@
MSG_USERAUTH_PK_OK = 60
messages = {}
-for k, v in locals().items():
- if k[:4]=='MSG_':
+for k, v in items(locals()):
+ if k[:4] == 'MSG_':
messages[v] = k
SSHUserAuthServer.protocolMessages = messages
Modified: branches/clarity-sshuserauth-8240/twisted/conch/test/test_userauth.py
==============================================================================
--- branches/clarity-sshuserauth-8240/twisted/conch/test/test_userauth.py (original)
+++ branches/clarity-sshuserauth-8240/twisted/conch/test/test_userauth.py Sun Mar 20 19:19:21 2016
@@ -1,4 +1,3 @@
-# -*- test-case-name: twisted.conch.test.test_userauth -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
@@ -8,6 +7,8 @@
Maintainer: Paul Swartz
"""
+from __future__ import absolute_import, division
+
from zope.interface import implementer
from twisted.cred.checkers import ICredentialsChecker
@@ -73,14 +74,14 @@
"""
Return 'foo' as the password.
"""
- return defer.succeed('foo')
+ return defer.succeed(b'foo')
def getGenericAnswers(self, name, information, answers):
"""
Return 'foo' as the answer to two questions.
"""
- return defer.succeed(('foo', 'foo'))
+ return defer.succeed((b'foo', b'foo'))
@@ -99,6 +100,8 @@
def getPublicKey(self):
return keys.Key.fromString(keydata.publicRSA_openssh).blob()
+
+
class ClientAuthWithoutPrivateKey(userauth.SSHUserAuthClient):
"""
This client doesn't have a private key, but it does have a public key.
@@ -281,7 +284,7 @@
"""
self.assertEqual(self.authServer.transport.packets[-1],
(userauth.MSG_USERAUTH_FAILURE,
- NS('password,publickey') + '\x00'))
+ NS(b'password,publickey') + b'\x00'))
def test_noneAuthentication(self):
@@ -291,8 +294,8 @@
See RFC 4252 Section 5.2.
"""
- d = self.authServer.ssh_USERAUTH_REQUEST(NS('foo') + NS('service') +
- NS('none'))
+ d = self.authServer.ssh_USERAUTH_REQUEST(NS(b'foo') + NS(b'service') +
+ NS(b'none'))
return d.addCallback(self._checkFailed)
@@ -304,12 +307,12 @@
See RFC 4252, Section 5.1.
"""
- packet = NS('foo') + NS('none') + NS('password') + chr(0) + NS('foo')
+ packet = NS(b'foo') + NS(b'none') + NS(b'password') + chr(0) + NS(b'foo')
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
def check(ignored):
self.assertEqual(
self.authServer.transport.packets,
- [(userauth.MSG_USERAUTH_SUCCESS, '')])
+ [(userauth.MSG_USERAUTH_SUCCESS, b'')])
return d.addCallback(check)
@@ -323,7 +326,7 @@
See RFC 4252, Section 5.1.
"""
# packet = username, next_service, authentication type, FALSE, password
- packet = NS('foo') + NS('none') + NS('password') + chr(0) + NS('bar')
+ packet = NS(b'foo') + NS(b'none') + NS(b'password') + chr(0) + NS(b'bar')
self.authServer.clock = task.Clock()
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
self.assertEqual(self.authServer.transport.packets, [])
@@ -337,16 +340,16 @@
"""
blob = keys.Key.fromString(keydata.publicRSA_openssh).blob()
obj = keys.Key.fromString(keydata.privateRSA_openssh)
- packet = (NS('foo') + NS('none') + NS('publickey') + '\xff'
+ packet = (NS(b'foo') + NS(b'none') + NS(b'publickey') + b'\xff'
+ NS(obj.sshType()) + NS(blob))
- self.authServer.transport.sessionID = 'test'
- signature = obj.sign(NS('test') + chr(userauth.MSG_USERAUTH_REQUEST)
+ self.authServer.transport.sessionID = b'test'
+ signature = obj.sign(NS(b'test') + chr(userauth.MSG_USERAUTH_REQUEST)
+ packet)
packet += NS(signature)
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
def check(ignored):
self.assertEqual(self.authServer.transport.packets,
- [(userauth.MSG_USERAUTH_SUCCESS, '')])
+ [(userauth.MSG_USERAUTH_SUCCESS, b'')])
return d.addCallback(check)
@@ -370,7 +373,7 @@
self.patch(self.authServer, '_cbFinishedAuth', mockCbFinishedAuth)
self.patch(self.authServer, '_ebBadAuth', mockEbBadAuth)
- packet = NS('user') + NS('none') + NS('public-key') + NS('data')
+ packet = NS(b'user') + NS(b'none') + NS(b'public-key') + NS(b'data')
# If an error other than ConchError is raised, this will trigger an
# exception.
self.authServer.ssh_USERAUTH_REQUEST(packet)
@@ -382,12 +385,12 @@
Test that verifying a valid private key works.
"""
blob = keys.Key.fromString(keydata.publicRSA_openssh).blob()
- packet = (NS('foo') + NS('none') + NS('publickey') + '\x00'
- + NS('ssh-rsa') + NS(blob))
+ packet = (NS(b'foo') + NS(b'none') + NS(b'publickey') + b'\x00'
+ + NS(b'ssh-rsa') + NS(blob))
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
def check(ignored):
self.assertEqual(self.authServer.transport.packets,
- [(userauth.MSG_USERAUTH_PK_OK, NS('ssh-rsa') + NS(blob))])
+ [(userauth.MSG_USERAUTH_PK_OK, NS(b'ssh-rsa') + NS(blob))])
return d.addCallback(check)
@@ -397,8 +400,8 @@
is invalid.
"""
blob = keys.Key.fromString(keydata.publicDSA_openssh).blob()
- packet = (NS('foo') + NS('none') + NS('publickey') + '\x00'
- + NS('ssh-dsa') + NS(blob))
+ packet = (NS(b'foo') + NS(b'none') + NS(b'publickey') + b'\x00'
+ + NS(b'ssh-dsa') + NS(blob))
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
return d.addCallback(self._checkFailed)
@@ -410,9 +413,9 @@
"""
blob = keys.Key.fromString(keydata.publicRSA_openssh).blob()
obj = keys.Key.fromString(keydata.privateRSA_openssh)
- packet = (NS('foo') + NS('none') + NS('publickey') + '\xff'
- + NS('ssh-rsa') + NS(blob) + NS(obj.sign(blob)))
- self.authServer.transport.sessionID = 'test'
+ packet = (NS(b'foo') + NS(b'none') + NS(b'publickey') + b'\xff'
+ + NS(b'ssh-rsa') + NS(blob) + NS(obj.sign(blob)))
+ self.authServer.transport.sessionID = b'test'
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
return d.addCallback(self._checkFailed)
@@ -433,7 +436,7 @@
server.serviceStopped()
server.supportedAuthentications.sort() # give a consistent order
self.assertEqual(server.supportedAuthentications,
- ['password', 'publickey'])
+ [b'password', b'publickey'])
def test_removePasswordIfUnencrypted(self):
@@ -441,21 +444,21 @@
Test that the userauth service does not advertise password
authentication if the password would be send in cleartext.
"""
- self.assertIn('password', self.authServer.supportedAuthentications)
+ self.assertIn(b'password', self.authServer.supportedAuthentications)
# no encryption
clearAuthServer = userauth.SSHUserAuthServer()
clearAuthServer.transport = FakeTransport(self.portal)
clearAuthServer.transport.isEncrypted = lambda x: False
clearAuthServer.serviceStarted()
clearAuthServer.serviceStopped()
- self.assertNotIn('password', clearAuthServer.supportedAuthentications)
+ self.assertNotIn(b'password', clearAuthServer.supportedAuthentications)
# only encrypt incoming (the direction the password is sent)
halfAuthServer = userauth.SSHUserAuthServer()
halfAuthServer.transport = FakeTransport(self.portal)
halfAuthServer.transport.isEncrypted = lambda x: x == 'in'
halfAuthServer.serviceStarted()
halfAuthServer.serviceStopped()
- self.assertIn('password', halfAuthServer.supportedAuthentications)
+ self.assertIn(b'password', halfAuthServer.supportedAuthentications)
def test_unencryptedConnectionWithoutPasswords(self):
@@ -475,7 +478,7 @@
clearAuthServer.serviceStarted()
clearAuthServer.serviceStopped()
self.assertEqual(clearAuthServer.supportedAuthentications,
- ['publickey'])
+ [b'publickey'])
# only encrypt incoming (the direction the password is sent)
halfAuthServer = userauth.SSHUserAuthServer()
@@ -484,7 +487,7 @@
halfAuthServer.serviceStarted()
halfAuthServer.serviceStopped()
self.assertEqual(clearAuthServer.supportedAuthentications,
- ['publickey'])
+ [b'publickey'])
def test_loginTimeout(self):
@@ -499,9 +502,9 @@
timeoutAuthServer.serviceStopped()
self.assertEqual(timeoutAuthServer.transport.packets,
[(transport.MSG_DISCONNECT,
- '\x00' * 3 +
+ b'\x00' * 3 +
chr(transport.DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE) +
- NS("you took too long") + NS(''))])
+ NS(b"you took too long") + NS(b''))])
self.assertTrue(timeoutAuthServer.transport.lostConnection)
@@ -524,7 +527,7 @@
Test that the server disconnects if the client fails authentication
too many times.
"""
- packet = NS('foo') + NS('none') + NS('password') + chr(0) + NS('bar')
+ packet = NS(b'foo') + NS(b'none') + NS(b'password') + chr(0) + NS(b'bar')
self.authServer.clock = task.Clock()
for i in range(21):
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
@@ -532,9 +535,9 @@
def check(ignored):
self.assertEqual(self.authServer.transport.packets[-1],
(transport.MSG_DISCONNECT,
- '\x00' * 3 +
+ b'\x00' * 3 +
chr(transport.DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE) +
- NS("too many bad auths") + NS('')))
+ NS(b"too many bad auths") + NS(b'')))
return d.addCallback(check)
@@ -543,7 +546,7 @@
If the user requests a service that we don't support, the
authentication should fail.
"""
- packet = NS('foo') + NS('') + NS('password') + chr(0) + NS('foo')
+ packet = NS(b'foo') + NS(b'') + NS(b'password') + chr(0) + NS(b'foo')
self.authServer.clock = task.Clock()
d = self.authServer.ssh_USERAUTH_REQUEST(packet)
return d.addCallback(self._checkFailed)
@@ -567,10 +570,10 @@
self.patch(self.authServer, 'auth_password', None) # second case
def secondTest(ignored):
- d2 = self.authServer.tryAuth('password', None, None)
+ d2 = self.authServer.tryAuth(b'password', None, None)
return self.assertFailure(d2, ConchError)
- d1 = self.authServer.tryAuth('publickey', None, None)
+ d1 = self.authServer.tryAuth(b'publickey', None, None)
return self.assertFailure(d1, ConchError).addCallback(secondTest)
@@ -581,15 +584,14 @@
Tests for SSHUserAuthClient.
"""
-
if keys is None:
skip = "cannot run without cryptography"
def setUp(self):
- self.authClient = ClientUserAuth('foo', FakeTransport.Service())
+ self.authClient = ClientUserAuth(b'foo', FakeTransport.Service())
self.authClient.transport = FakeTransport(None)
- self.authClient.transport.sessionID = 'test'
+ self.authClient.transport.sessionID = b'test'
self.authClient.serviceStarted()
@@ -602,11 +604,11 @@
"""
Test that client is initialized properly.
"""
- self.assertEqual(self.authClient.user, 'foo')
+ self.assertEqual(self.authClient.user, b'foo')
self.assertEqual(self.authClient.instance.name, 'nancy')
self.assertEqual(self.authClient.transport.packets,
- [(userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy')
- + NS('none'))])
+ [(userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy')
+ + NS(b'none'))])
def test_USERAUTH_SUCCESS(self):
@@ -617,7 +619,7 @@
def stubSetService(service):
instance[0] = service
self.authClient.transport.setService = stubSetService
- self.authClient.ssh_USERAUTH_SUCCESS('')
+ self.authClient.ssh_USERAUTH_SUCCESS(b'')
self.assertEqual(instance[0], self.authClient.instance)
@@ -625,28 +627,28 @@
"""
Test that the client can authenticate with a public key.
"""
- self.authClient.ssh_USERAUTH_FAILURE(NS('publickey') + '\x00')
+ self.authClient.ssh_USERAUTH_FAILURE(NS(b'publickey') + b'\x00')
self.assertEqual(self.authClient.transport.packets[-1],
- (userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy')
- + NS('publickey') + '\x00' + NS('ssh-dss')
+ (userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy')
+ + NS(b'publickey') + b'\x00' + NS(b'ssh-dss')
+ NS(keys.Key.fromString(
keydata.publicDSA_openssh).blob())))
# that key isn't good
- self.authClient.ssh_USERAUTH_FAILURE(NS('publickey') + '\x00')
+ self.authClient.ssh_USERAUTH_FAILURE(NS(b'publickey') + b'\x00')
blob = NS(keys.Key.fromString(keydata.publicRSA_openssh).blob())
self.assertEqual(self.authClient.transport.packets[-1],
- (userauth.MSG_USERAUTH_REQUEST, (NS('foo') + NS('nancy')
- + NS('publickey') + '\x00'+ NS('ssh-rsa') + blob)))
- self.authClient.ssh_USERAUTH_PK_OK(NS('ssh-rsa')
+ (userauth.MSG_USERAUTH_REQUEST, (NS(b'foo') + NS(b'nancy')
+ + NS(b'publickey') + b'\x00' + NS(b'ssh-rsa') + blob)))
+ self.authClient.ssh_USERAUTH_PK_OK(NS(b'ssh-rsa')
+ NS(keys.Key.fromString(keydata.publicRSA_openssh).blob()))
sigData = (NS(self.authClient.transport.sessionID)
- + chr(userauth.MSG_USERAUTH_REQUEST) + NS('foo')
- + NS('nancy') + NS('publickey') + '\x01' + NS('ssh-rsa')
+ + chr(userauth.MSG_USERAUTH_REQUEST) + NS(b'foo')
+ + NS(b'nancy') + NS(b'publickey') + b'\x01' + NS(b'ssh-rsa')
+ blob)
obj = keys.Key.fromString(keydata.privateRSA_openssh)
self.assertEqual(self.authClient.transport.packets[-1],
- (userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy')
- + NS('publickey') + '\x01' + NS('ssh-rsa') + blob
+ (userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy')
+ + NS(b'publickey') + b'\x01' + NS(b'ssh-rsa') + blob
+ NS(obj.sign(sigData))))
@@ -656,18 +658,18 @@
the client should start the authentication over again by requesting
'none' authentication.
"""
- authClient = ClientAuthWithoutPrivateKey('foo',
+ authClient = ClientAuthWithoutPrivateKey(b'foo',
FakeTransport.Service())
authClient.transport = FakeTransport(None)
- authClient.transport.sessionID = 'test'
+ authClient.transport.sessionID = b'test'
authClient.serviceStarted()
- authClient.tryAuth('publickey')
+ authClient.tryAuth(b'publickey')
authClient.transport.packets = []
- self.assertIs(authClient.ssh_USERAUTH_PK_OK(''), None)
+ self.assertIs(authClient.ssh_USERAUTH_PK_OK(b''), None)
self.assertEqual(authClient.transport.packets, [
- (userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy') +
- NS('none'))])
+ (userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy') +
+ NS(b'none'))])
def test_no_publickey(self):
@@ -676,24 +678,25 @@
called back with a False value.
"""
self.authClient.getPublicKey = lambda x: None
- d = self.authClient.tryAuth('publickey')
+ d = self.authClient.tryAuth(b'publickey')
def check(result):
self.assertFalse(result)
return d.addCallback(check)
+
def test_password(self):
"""
Test that the client can authentication with a password. This
includes changing the password.
"""
- self.authClient.ssh_USERAUTH_FAILURE(NS('password') + '\x00')
+ self.authClient.ssh_USERAUTH_FAILURE(NS(b'password') + b'\x00')
self.assertEqual(self.authClient.transport.packets[-1],
- (userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy')
- + NS('password') + '\x00' + NS('foo')))
- self.authClient.ssh_USERAUTH_PK_OK(NS('') + NS(''))
+ (userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy')
+ + NS(b'password') + b'\x00' + NS(b'foo')))
+ self.authClient.ssh_USERAUTH_PK_OK(NS(b'') + NS(b''))
self.assertEqual(self.authClient.transport.packets[-1],
- (userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy')
- + NS('password') + '\xff' + NS('foo') * 2))
+ (userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy')
+ + NS(b'password') + b'\xff' + NS(b'foo') * 2))
def test_no_password(self):
@@ -701,7 +704,7 @@
If getPassword returns None, tryAuth should return False.
"""
self.authClient.getPassword = lambda: None
- self.assertFalse(self.authClient.tryAuth('password'))
+ self.assertFalse(self.authClient.tryAuth(b'password'))
def test_USERAUTH_PK_OK_unknown_method(self):
@@ -710,12 +713,12 @@
expecting it, it should fail the current authentication and move on to
the next type.
"""
- self.authClient.lastAuth = 'unknown'
+ self.authClient.lastAuth = b'unknown'
self.authClient.transport.packets = []
- self.authClient.ssh_USERAUTH_PK_OK('')
+ self.authClient.ssh_USERAUTH_PK_OK(b'')
self.assertEqual(self.authClient.transport.packets,
- [(userauth.MSG_USERAUTH_REQUEST, NS('foo') +
- NS('nancy') + NS('none'))])
+ [(userauth.MSG_USERAUTH_REQUEST, NS(b'foo') +
+ NS(b'nancy') + NS(b'none'))])
def test_USERAUTH_FAILURE_sorting(self):
@@ -725,25 +728,25 @@
preferredOrder should be sorted at the end of that list.
"""
def auth_firstmethod():
- self.authClient.transport.sendPacket(255, 'here is data')
+ self.authClient.transport.sendPacket(255, b'here is data')
def auth_anothermethod():
- self.authClient.transport.sendPacket(254, 'other data')
+ self.authClient.transport.sendPacket(254, b'other data')
return True
self.authClient.auth_firstmethod = auth_firstmethod
self.authClient.auth_anothermethod = auth_anothermethod
# although they shouldn't get called, method callbacks auth_* MUST
# exist in order for the test to work properly.
- self.authClient.ssh_USERAUTH_FAILURE(NS('anothermethod,password') +
- '\x00')
+ self.authClient.ssh_USERAUTH_FAILURE(NS(b'anothermethod,password') +
+ b'\x00')
# should send password packet
self.assertEqual(self.authClient.transport.packets[-1],
- (userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy')
- + NS('password') + '\x00' + NS('foo')))
+ (userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy')
+ + NS(b'password') + b'\x00' + NS(b'foo')))
self.authClient.ssh_USERAUTH_FAILURE(
- NS('firstmethod,anothermethod,password') + '\xff')
+ NS(b'firstmethod,anothermethod,password') + b'\xff')
self.assertEqual(self.authClient.transport.packets[-2:],
- [(255, 'here is data'), (254, 'other data')])
+ [(255, b'here is data'), (254, b'other data')])
def test_disconnectIfNoMoreAuthentication(self):
@@ -752,12 +755,12 @@
the SSHUserAuthClient should disconnect with code
DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE.
"""
- self.authClient.ssh_USERAUTH_FAILURE(NS('password') + '\x00')
- self.authClient.ssh_USERAUTH_FAILURE(NS('password') + '\xff')
+ self.authClient.ssh_USERAUTH_FAILURE(NS(b'password') + b'\x00')
+ self.authClient.ssh_USERAUTH_FAILURE(NS(b'password') + b'\xff')
self.assertEqual(self.authClient.transport.packets[-1],
- (transport.MSG_DISCONNECT, '\x00\x00\x00\x0e' +
- NS('no more authentication methods available') +
- '\x00\x00\x00\x00'))
+ (transport.MSG_DISCONNECT, b'\x00\x00\x00\x0e' +
+ NS(b'no more authentication methods available') +
+ b'\x00\x00\x00\x00'))
def test_ebAuth(self):
@@ -768,8 +771,8 @@
self.authClient.transport.packets = []
self.authClient._ebAuth(None)
self.assertEqual(self.authClient.transport.packets,
- [(userauth.MSG_USERAUTH_REQUEST, NS('foo') + NS('nancy')
- + NS('none'))])
+ [(userauth.MSG_USERAUTH_REQUEST, NS(b'foo') + NS(b'nancy')
+ + NS(b'none'))])
def test_defaults(self):
@@ -778,7 +781,7 @@
failed Deferred. getPassword() should return a failed Deferred.
getGenericAnswers() should return a failed Deferred.
"""
- authClient = userauth.SSHUserAuthClient('foo', FakeTransport.Service())
+ authClient = userauth.SSHUserAuthClient(b'foo', FakeTransport.Service())
self.assertIs(authClient.getPublicKey(), None)
def check(result):
result.trap(NotImplementedError)
@@ -824,7 +827,7 @@
Test that the userauth server and client play nicely with each other.
"""
server = userauth.SSHUserAuthServer()
- client = ClientUserAuth('foo', self.Factory.Service())
+ client = ClientUserAuth(b'foo', self.Factory.Service())
# set up transports
server.transport = transport.SSHTransportBase()
@@ -832,7 +835,7 @@
server.transport.isEncrypted = lambda x: True
client.transport = transport.SSHTransportBase()
client.transport.service = client
- server.transport.sessionID = client.transport.sessionID = ''
+ server.transport.sessionID = client.transport.sessionID = b''
# don't send key exchange packet
server.transport.sendKexInit = client.transport.sendKexInit = \
lambda: None