r47044 - Merge clarity-sshuserauth-8240: Increase type clarity in twisted.conch.ssh.userauth
hawkowl-TA+aISz0psMTMxyoc4vAAJOcrHinNvQL0E9HWUfgJXw@public.gmane.org Mon, 21 Mar 2016 20:42:59 -0600 (MDT)
| Newsgroups | gmane.comp.python.twisted.commits |
|---|---|
| Message-ID | <[email protected]> |
Author: hawkowl
Date: Mon Mar 21 20:42:54 2016
New Revision: 47044
Added:
trunk/twisted/conch/topfiles/8240.misc
Modified:
trunk/twisted/conch/ssh/userauth.py
trunk/twisted/conch/test/test_userauth.py
Log:
Merge clarity-sshuserauth-8240: Increase type clarity in twisted.conch.ssh.userauth
Author: hawkowl
Reviewer: adiroiban
Fixes: #8240
Modified: trunk/twisted/conch/ssh/userauth.py
==============================================================================
--- trunk/twisted/conch/ssh/userauth.py (original)
+++ trunk/twisted/conch/ssh/userauth.py Mon Mar 21 20:42:54 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
@@ -49,11 +52,11 @@
@type interfaceToMethod: C{dict}
@ivar supportedAuthentications: A list of the supported authentication
methods.
- @type supportedAuthentications: C{list} of C{str}
+ @type supportedAuthentications: C{list} of C{bytes}
@ivar user: the last username the client tried to authenticate with
- @type user: C{str}
+ @type user: C{bytes}
@ivar method: the current authentication method
- @type method: C{str}
+ @type method: C{bytes}
@ivar nextService: the service the user wants started after authentication
has been completed.
@type nextService: C{str}
@@ -63,7 +66,6 @@
@ivar clock: an object with a callLater method. Stubbed out for testing.
"""
-
name = 'ssh-userauth'
loginTimeout = 10 * 60 * 60
# 10 minutes before we disconnect them
@@ -72,8 +74,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 +99,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 +124,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):
@@ -131,11 +133,11 @@
auth_* method.
@param kind: the authentication method to try.
- @type kind: C{str}
+ @type kind: C{bytes}
@param user: the username the client is authenticating with.
- @type user: C{str}
+ @type user: C{bytes}
@param data: authentication specific data sent by the client.
- @type data: C{str}
+ @type data: C{bytes}
@return: A Deferred called back if the method succeeded, or erred back
if it failed.
@rtype: C{defer.Deferred}
@@ -144,17 +146,18 @@
if kind not in self.supportedAuthentications:
return defer.fail(
error.ConchError('unsupported authentication, failing'))
- kind = kind.replace('-', '_')
- f = getattr(self,'auth_%s'%kind, None)
+ kind = kind.replace(b'-', b'_')
+ f = getattr(self, 'auth_%s' % (kind,), None)
if f:
ret = f(data)
if not ret:
return defer.fail(
- error.ConchError('%s return None instead of a Deferred'
- % kind))
+ error.ConchError(
+ '%s return None instead of a Deferred'
+ % (kind, )))
else:
return ret
- return defer.fail(error.ConchError('bad auth type: %s' % kind))
+ return defer.fail(error.ConchError('bad auth type: %s' % (kind,)))
def ssh_USERAUTH_REQUEST(self, packet):
@@ -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):
@@ -262,7 +265,7 @@
signature = hasSig and getNS(rest)[0] or None
if hasSig:
b = (NS(self.transport.sessionID) + chr(MSG_USERAUTH_REQUEST) +
- NS(self.user) + NS(self.nextService) + NS('publickey') +
+ NS(self.user) + NS(self.nextService) + NS(b'publickey') +
chr(hasSig) + NS(pubKey.sshType()) + NS(blob))
c = credentials.SSHPrivateKey(self.user, algName, blob, b,
signature)
@@ -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}
@@ -336,9 +339,8 @@
@type lastPublicKey: L{Key}
"""
-
name = 'ssh-userauth'
- preferredOrder = ['publickey', 'password', 'keyboard-interactive']
+ preferredOrder = [b'publickey', b'password', b'keyboard-interactive']
def __init__(self, user, instance):
@@ -350,7 +352,7 @@
self.authenticatedWith = []
self.triedPublicKeys = []
self.lastPublicKey = None
- self.askForAuth('none', '')
+ self.askForAuth(b'none', b'')
def askForAuth(self, kind, extraData):
@@ -358,9 +360,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) +
@@ -372,9 +374,9 @@
Dispatch to an authentication method.
@param kind: the authentication method
- @type kind: C{str}
+ @type kind: C{bytes}
"""
- 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 +388,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):
@@ -412,7 +414,7 @@
C{self.tryAuth} with the most preferred method.
@param packet: the L{MSG_USERAUTH_FAILURE} payload.
- @type packet: C{str}
+ @type packet: C{bytes}
@return: a L{defer.Deferred} that will be callbacked with C{None} as
soon as all authentication methods have been tried, or C{None} if no
@@ -430,7 +432,7 @@
comparison key which is then used for sorting.
@param meth: the authentication method.
- @type meth: C{str}
+ @type meth: C{bytes}
@return: the comparison key for C{meth}.
@rtype: C{int}
@@ -441,7 +443,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 +455,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)
@@ -471,11 +473,11 @@
in order to handle this request.
"""
func = getattr(self, 'ssh_USERAUTH_PK_OK_%s' %
- self.lastAuth.replace('-', '_'), None)
+ self.lastAuth.replace(b'-', b'_'), None)
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):
@@ -485,11 +487,11 @@
"""
publicKey = self.lastPublicKey
b = (NS(self.transport.sessionID) + chr(MSG_USERAUTH_REQUEST) +
- NS(self.user) + NS(self.instance.name) + NS('publickey') +
- '\x01' + NS(publicKey.sshType()) + NS(publicKey.blob()))
+ NS(self.user) + NS(self.instance.name) + NS(b'publickey') +
+ b'\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 +506,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)
@@ -536,10 +538,10 @@
authentication request with the signature.
@param signedData: the data signed by the user's private key.
- @type signedData: C{str}
+ @type signedData: C{bytes}
"""
publicKey = self.lastPublicKey
- self.askForAuth('publickey', '\x01' + NS(publicKey.sshType()) +
+ self.askForAuth(b'publickey', b'\x01' + NS(publicKey.sshType()) +
NS(publicKey.blob()) + NS(signedData))
@@ -549,7 +551,7 @@
password for now.
@param op: the old password as entered by the user
- @type op: C{str}
+ @type op: C{bytes}
"""
self._oldPass = op
@@ -560,11 +562,11 @@
and send the authentication message with both.
@param np: the new password as entered by the user
- @type np: C{str}
+ @type np: C{bytes}
"""
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):
@@ -573,7 +575,7 @@
questions. Send the info back to the server in a
MSG_USERAUTH_INFO_RESPONSE.
- @param responses: a list of C{str} responses
+ @param responses: a list of C{bytes} responses
@type responses: C{list}
"""
data = struct.pack('!L', len(responses))
@@ -602,7 +604,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 +635,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
@@ -643,9 +645,9 @@
server.
@param password: the password the user entered
- @type password: C{str}
+ @type password: C{bytes}
"""
- self.askForAuth('password', '\x00' + NS(password))
+ self.askForAuth(b'password', b'\x00' + NS(password))
def signData(self, publicKey, signData):
@@ -662,7 +664,7 @@
@type publicKey: L{keys.Key}
@param signData: the data to be signed by the private key.
- @type signData: C{str}
+ @type signData: C{bytes}
@return: a Deferred that's called back with the signature
@rtype: L{defer.Deferred}
"""
@@ -680,9 +682,9 @@
@param privateKey: the private key object
@type publicKey: L{keys.Key}
@param signData: the data to be signed by the private key.
- @type signData: C{str}
+ @type signData: C{bytes}
@return: the signature
- @rtype: C{str}
+ @rtype: C{bytes}
"""
return privateKey.sign(signData)
@@ -717,7 +719,7 @@
prompt is a string to display for the password, or None for a generic
'user@hostname's password: '.
- @type prompt: C{str}/C{None}
+ @type prompt: C{bytes}/C{None}
@rtype: L{defer.Deferred}
"""
return defer.fail(NotImplementedError())
@@ -745,7 +747,7 @@
messages = {}
for k, v in locals().items():
- if k[:4]=='MSG_':
+ if k[:4] == 'MSG_':
messages[v] = k
SSHUserAuthServer.protocolMessages = messages
Modified: trunk/twisted/conch/test/test_userauth.py
==============================================================================
--- trunk/twisted/conch/test/test_userauth.py (original)
+++ trunk/twisted/conch/test/test_userauth.py Mon Mar 21 20:42:54 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
@@ -49,7 +50,6 @@
A mock user auth client.
"""
-
def getPublicKey(self):
"""
If this is the first time we've been called, return a blob for
@@ -59,7 +59,8 @@
if self.lastPublicKey:
return keys.Key.fromString(keydata.publicRSA_openssh)
else:
- return defer.succeed(keys.Key.fromString(keydata.publicDSA_openssh))
+ return defer.succeed(
+ keys.Key.fromString(keydata.publicDSA_openssh))
def getPrivateKey(self):
@@ -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'))
@@ -90,7 +91,6 @@
getPrivateKey() and a string from getPublicKey
"""
-
def getPrivateKey(self):
return defer.succeed(keys.Key.fromString(
keydata.privateRSA_openssh).keyObject)
@@ -99,12 +99,13 @@
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.
"""
-
def getPrivateKey(self):
return
@@ -131,7 +132,6 @@
@type lostConnection: C{bool}
"""
-
class Service(object):
"""
A mock service, representing the other service offered by the server.
@@ -143,14 +143,12 @@
pass
-
class Factory(object):
"""
A mock factory, representing the factory that spawned this user auth
service.
"""
-
def getService(self, transport, service):
"""
Return our fake service.
@@ -159,7 +157,6 @@
return FakeTransport.Service
-
def __init__(self, portal):
self.factory = self.Factory()
self.factory.portal = portal
@@ -168,7 +165,6 @@
self.packets = []
-
def sendPacket(self, messageType, message):
"""
Record the packet sent by the service.
@@ -253,7 +249,6 @@
Tests for SSHUserAuthServer.
"""
-
if keys is None:
skip = "cannot run without cryptography"
@@ -281,7 +276,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 +286,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 +299,13 @@
See RFC 4252, Section 5.1.
"""
- packet = NS('foo') + NS('none') + NS('password') + chr(0) + NS('foo')
+ packet = b''.join([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 +319,8 @@
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 = b''.join([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 +334,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 +367,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 +379,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 +394,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 +407,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 +430,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 +438,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 +472,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 +481,7 @@
halfAuthServer.serviceStarted()
halfAuthServer.serviceStopped()
self.assertEqual(clearAuthServer.supportedAuthentications,
- ['publickey'])
+ [b'publickey'])
def test_loginTimeout(self):
@@ -499,9 +496,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 +521,8 @@
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 = b''.join([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 +530,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 +541,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,29 +565,27 @@
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)
-
class SSHUserAuthClientTests(unittest.TestCase):
"""
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 +598,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 +613,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 +621,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 +652,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 +672,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 +698,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 +707,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 +722,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 +749,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 +765,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 +775,8 @@
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)
@@ -797,7 +795,6 @@
class LoopbackTests(unittest.TestCase):
-
if keys is None:
skip = "cannot run without cryptography or PyASN1"
@@ -824,7 +821,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 +829,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