r47005 - porting
hawkowl-TA+aISz0psMTMxyoc4vAAJOcrHinNvQL0E9HWUfgJXw@public.gmane.org Wed, 16 Mar 2016 01:48:23 -0600 (MDT)
| Newsgroups | gmane.comp.python.twisted.commits |
|---|---|
| Message-ID | <[email protected]> |
Author: hawkowl
Date: Wed Mar 16 01:48:18 2016
New Revision: 47005
Modified:
branches/conch-transport-py3-8232/twisted/conch/ssh/connection.py
branches/conch-transport-py3-8232/twisted/conch/ssh/userauth.py
branches/conch-transport-py3-8232/twisted/conch/test/test_userauth.py
branches/conch-transport-py3-8232/twisted/python/dist3.py
Log:
porting
Modified: branches/conch-transport-py3-8232/twisted/conch/ssh/connection.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/ssh/connection.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/ssh/connection.py Wed Mar 16 01:48:18 2016
@@ -143,7 +143,7 @@
channel.localWindowSize,
channel.localMaxPacket)+channel.specificData)
log.callWithLogger(channel, channel.channelOpen, packet)
- except Exception, e:
+ except Exception as e:
log.err(e, 'channel open failed')
if isinstance(e, error.ConchError):
textualInfo, reason = e.args
@@ -630,7 +630,7 @@
messages[value] = name # doesn't handle doubles
import string
-alphanums = string.letters + string.digits
+alphanums = string.ascii_letters + string.digits
TRANSLATE_TABLE = ''.join([chr(i) in alphanums and chr(i) or '_'
for i in range(256)])
SSHConnection.protocolMessages = messages
Modified: branches/conch-transport-py3-8232/twisted/conch/ssh/userauth.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/ssh/userauth.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/ssh/userauth.py Wed Mar 16 01:48:18 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
@@ -17,6 +20,8 @@
from twisted.cred.error import UnauthorizedLogin
from twisted.internet import defer, reactor
from twisted.python import failure, log
+from twisted.python.compat import (items, nativeString, networkString,
+ _bytesChr as chr)
@@ -72,8 +77,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 +102,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 +127,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):
@@ -144,8 +149,8 @@
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' % (nativeString(kind),), None)
if f:
ret = f(data)
if not ret:
@@ -165,13 +170,13 @@
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:
self.authenticatedWith = [] # clear auth state
self.user = user
- self.nextService = nextService
+ self.nextService = nativeString(nextService)
self.method = method
d = self.tryAuth(method, user, rest)
if not d:
@@ -184,12 +189,13 @@
return d
- def _cbFinishedAuth(self, (interface, avatar, logout)):
+ def _cbFinishedAuth(self, args):
"""
The callback when user has successfully been authenticated. For a
description of the arguments, see L{twisted.cred.portal.Portal.login}.
We start the service requested by the user.
"""
+ (interface, avatar, logout) = args
self.transport.avatar = avatar
self.transport.logoutFunction = logout
service = self.transport.factory.getService(self.transport,
@@ -198,7 +204,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 +218,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 +245,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):
@@ -256,14 +262,15 @@
Create a SSHPublicKey credential and verify it using our portal.
"""
- hasSig = ord(packet[0])
+ hasSig = ord(packet[0:1])
algName, blob, rest = getNS(packet[1:], 2)
pubKey = keys.Key.fromString(blob)
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') +
- chr(hasSig) + NS(pubKey.sshType()) + NS(blob))
+ NS(self.user) + NS(networkString(self.nextService)) +
+ NS(b'publickey') + chr(hasSig) + NS(pubKey.sshType()) +
+ NS(blob))
c = credentials.SSHPrivateKey(self.user, algName, blob, b,
signature)
return self.portal.login(c, None, interfaces.IConchUser)
@@ -323,11 +330,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 +345,7 @@
name = 'ssh-userauth'
- preferredOrder = ['publickey', 'password', 'keyboard-interactive']
+ preferredOrder = [b'publickey', b'password', b'keyboard-interactive']
def __init__(self, user, instance):
@@ -350,7 +357,7 @@
self.authenticatedWith = []
self.triedPublicKeys = []
self.lastPublicKey = None
- self.askForAuth('none', '')
+ self.askForAuth(b'none', b'')
def askForAuth(self, kind, extraData):
@@ -358,13 +365,13 @@
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) +
- NS(self.instance.name) + NS(kind) + extraData)
+ NS(networkString(self.instance.name)) + NS(kind) + extraData)
def tryAuth(self, kind):
@@ -374,9 +381,9 @@
@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)
+ f = getattr(self, 'auth_%s' % (nativeString(kind),), None)
if f:
return f()
@@ -386,7 +393,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 +448,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 +460,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 +478,11 @@
in order to handle this request.
"""
func = getattr(self, 'ssh_USERAUTH_PK_OK_%s' %
- self.lastAuth.replace('-', '_'), None)
+ nativeString(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 +492,12 @@
"""
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(networkString(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 +512,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)
@@ -517,12 +525,12 @@
responses.
"""
name, instruction, lang, data = getNS(packet, 3)
- numPrompts = struct.unpack('!L', data[:4])[0]
+ numPrompts = struct.unpack('!L', data[:4])[0:1]
data = data[4:]
prompts = []
for i in range(numPrompts):
prompt, data = getNS(data)
- echo = bool(ord(data[0]))
+ echo = bool(ord(data[0:1]))
data = data[1:]
prompts.append((prompt, echo))
d = self.getGenericAnswers(name, instruction, prompts)
@@ -539,7 +547,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 +572,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 +610,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 +641,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 +653,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 +752,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/conch-transport-py3-8232/twisted/conch/test/test_userauth.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/test/test_userauth.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/test/test_userauth.py Wed Mar 16 01:48:18 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
@@ -19,6 +20,7 @@
from twisted.internet import defer, task
from twisted.protocols import loopback
from twisted.python.reflect import requireModule
+from twisted.python.compat import _bytesChr as chr
from twisted.trial import unittest
if requireModule('cryptography') and requireModule('pyasn1'):
@@ -73,14 +75,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 +101,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 +285,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 +295,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 +308,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 +327,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 +341,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 +374,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 +386,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 +401,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 +414,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 +437,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 +445,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 +479,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 +488,7 @@
halfAuthServer.serviceStarted()
halfAuthServer.serviceStopped()
self.assertEqual(clearAuthServer.supportedAuthentications,
- ['publickey'])
+ [b'publickey'])
def test_loginTimeout(self):
@@ -499,9 +503,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 +528,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 +536,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 +547,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 +571,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 +585,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 +605,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 +620,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 +628,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 +659,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 +679,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 +705,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 +714,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 +729,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 +756,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 +772,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 +782,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 +828,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 +836,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
Modified: branches/conch-transport-py3-8232/twisted/python/dist3.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/python/dist3.py (original)
+++ branches/conch-transport-py3-8232/twisted/python/dist3.py Wed Mar 16 01:48:18 2016
@@ -49,9 +49,13 @@
"twisted.conch.error",
"twisted.conch.ssh.__init__",
"twisted.conch.ssh._cryptography_backports",
+ "twisted.conch.ssh._kex",
+ "twisted.conch.ssh.address",
"twisted.conch.ssh.common",
+ "twisted.conch.ssh.factory",
"twisted.conch.ssh.keys",
"twisted.conch.ssh.sexpy",
+ "twisted.conch.ssh.transport",
"twisted.conch.telnet",
"twisted.conch.test.__init__",
"twisted.copyright",
@@ -260,7 +264,9 @@
"twisted.application.test.test_service",
"twisted.conch.test.test_checkers",
"twisted.conch.test.test_keys",
+ "twisted.conch.test.test_transport",
"twisted.conch.test.test_telnet",
+ "twisted.conch.test.test_userauth",
"twisted.cred.test.test_cramauth",
"twisted.cred.test.test_cred",
"twisted.cred.test.test_digestauth",