5 changes: 0 additions & 5 deletions twisted/conch/checkers.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,6 @@
else:
shadow = None

try:
from twisted.cred import pamauth
except ImportError:
pamauth = None

from zope.interface import providedBy, implementer, Interface


Expand Down
86 changes: 3 additions & 83 deletions twisted/conch/ssh/userauth.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,22 +74,20 @@ class SSHUserAuthServer(service.SSHService):
interfaceToMethod = {
credentials.ISSHPrivateKey : 'publickey',
credentials.IUsernamePassword : 'password',
credentials.IPluggableAuthenticationModules : 'keyboard-interactive',
}


def serviceStarted(self):
"""
Called when the userauth service is started. Set up instance
variables, check if we should allow password/keyboard-interactive
authentication (only allow if the outgoing connection is encrypted) and
set up a login timeout.
variables, check if we should allow password authentication (only
allow if the outgoing connection is encrypted) and set up a login
timeout.
"""
self.authenticatedWith = []
self.loginAttempts = 0
self.user = None
self.nextService = None
self._pamDeferred = None
self.portal = self.transport.factory.portal

self.supportedAuthentications = []
Expand All @@ -101,8 +99,6 @@ def serviceStarted(self):
# don't let us transport password in plaintext
if 'password' in self.supportedAuthentications:
self.supportedAuthentications.remove('password')
if 'keyboard-interactive' in self.supportedAuthentications:
self.supportedAuthentications.remove('keyboard-interactive')
self._cancelLoginTimeout = self.clock.callLater(
self.loginTimeout,
self.timeoutAuthentication)
Expand Down Expand Up @@ -313,82 +309,6 @@ def _ebPassword(self, f):
return d


def auth_keyboard_interactive(self, packet):
"""
Keyboard interactive authentication. No payload. We create a
PluggableAuthenticationModules credential and authenticate with our
portal.
"""
if self._pamDeferred is not None:
self.transport.sendDisconnect(
transport.DISCONNECT_PROTOCOL_ERROR,
"only one keyboard interactive attempt at a time")
return defer.fail(error.IgnoreAuthentication())
c = credentials.PluggableAuthenticationModules(self.user,
self._pamConv)
return self.portal.login(c, None, interfaces.IConchUser)


def _pamConv(self, items):
"""
Convert a list of PAM authentication questions into a
MSG_USERAUTH_INFO_REQUEST. Returns a Deferred that will be called
back when the user has responses to the questions.
@param items: a list of 2-tuples (message, kind). We only care about
kinds 1 (password) and 2 (text).
@type items: C{list}
@rtype: L{defer.Deferred}
"""
resp = []
for message, kind in items:
if kind == 1: # password
resp.append((message, 0))
elif kind == 2: # text
resp.append((message, 1))
elif kind in (3, 4):
return defer.fail(error.ConchError(
'cannot handle PAM 3 or 4 messages'))
else:
return defer.fail(error.ConchError(
'bad PAM auth kind %i' % kind))
packet = NS('') + NS('') + NS('')
packet += struct.pack('>L', len(resp))
for prompt, echo in resp:
packet += NS(prompt)
packet += chr(echo)
self.transport.sendPacket(MSG_USERAUTH_INFO_REQUEST, packet)
self._pamDeferred = defer.Deferred()
return self._pamDeferred


def ssh_USERAUTH_INFO_RESPONSE(self, packet):
"""
The user has responded with answers to PAMs authentication questions.
Parse the packet into a PAM response and callback self._pamDeferred.
Payload::
uint32 numer of responses
string response 1
...
string response n
"""
d, self._pamDeferred = self._pamDeferred, None

try:
resp = []
numResps = struct.unpack('>L', packet[:4])[0]
packet = packet[4:]
while len(resp) < numResps:
response, packet = getNS(packet)
resp.append((response, 0))
if packet:
raise error.ConchError("%i bytes of extra data" % len(packet))
except:
d.errback(failure.Failure())
else:
d.callback(resp)



class SSHUserAuthClient(service.SSHService):
"""
Expand Down
14 changes: 3 additions & 11 deletions twisted/conch/tap.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,21 +9,16 @@
from twisted.conch import unix
from twisted.conch import checkers as conch_checkers
from twisted.conch.openssh_compat import factory
from twisted.cred import portal, checkers, strcred
from twisted.cred import portal, strcred
from twisted.python import usage
from twisted.application import strports
try:
from twisted.cred import pamauth
except ImportError:
pamauth = None



class Options(usage.Options, strcred.AuthOptionMixin):
synopsis = "[-i <interface>] [-p <port>] [-d <dir>] "
longdesc = ("Makes a Conch SSH server. If no authentication methods are "
"specified, the default authentication methods are UNIX passwords, "
"SSH public keys, and PAM if it is available. If --auth options are "
"specified, the default authentication methods are UNIX passwords "
"and SSH public keys. If --auth options are "
"passed, only the measures specified will be used.")
optParameters = [
["interface", "i", "", "local interface to which we listen"],
Expand All @@ -49,9 +44,6 @@ def __init__(self, *a, **kw):
super(Options, self).addChecker(conch_checkers.UNIXPasswordDatabase())
super(Options, self).addChecker(conch_checkers.SSHPublicKeyChecker(
conch_checkers.UNIXAuthorizedKeysFiles()))
if pamauth is not None:
super(Options, self).addChecker(
checkers.PluggableAuthenticationModulesChecker())
self._usingDefaultAuth = True


Expand Down
35 changes: 3 additions & 32 deletions twisted/conch/test/test_tap.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,8 @@

from twisted.application.internet import StreamServerEndpointService
from twisted.cred import error
from twisted.cred.credentials import IPluggableAuthenticationModules
from twisted.cred.credentials import ISSHPrivateKey
from twisted.cred.credentials import IUsernamePassword, UsernamePassword
from twisted.python.reflect import requireModule

from twisted.trial.unittest import TestCase

Expand Down Expand Up @@ -78,17 +76,10 @@ def test_basic(self):
def test_defaultAuths(self):
"""
Make sure that if the C{--auth} command-line option is not passed,
the default checkers are (for backwards compatibility): SSH, UNIX, and
PAM if available
the default checkers are (for backwards compatibility): SSH and UNIX
"""
numCheckers = 2

if requireModule('twisted.cred.pamauth'):
self.assertIn(IPluggableAuthenticationModules,
self.options['credInterfaces'],
"PAM should be one of the modules")
numCheckers += 1

self.assertIn(ISSHPrivateKey, self.options['credInterfaces'],
"SSH should be one of the default checkers")
self.assertIn(IUsernamePassword, self.options['credInterfaces'],
Expand Down Expand Up @@ -148,32 +139,12 @@ def checkSuccess(username):
return d.addCallback(checkSuccess)


def test_checkersPamAuth(self):
"""
The L{OpenSSHFactory} built by L{tap.makeService} has a portal with
L{IPluggableAuthenticationModules}, L{ISSHPrivateKey} and
L{IUsernamePassword} interfaces registered as checkers if C{pamauth} is
available.
"""
# Fake the presence of pamauth, even if PyPAM is not installed
self.patch(tap, "pamauth", object())
config = tap.Options()
service = tap.makeService(config)
portal = service.factory.portal
self.assertEqual(
set(portal.checkers.keys()),
set([IPluggableAuthenticationModules, ISSHPrivateKey,
IUsernamePassword]))


def test_checkersWithoutPamAuth(self):
def test_checkers(self):
"""
The L{OpenSSHFactory} built by L{tap.makeService} has a portal with
L{ISSHPrivateKey} and L{IUsernamePassword} interfaces registered as
checkers if C{pamauth} is not available.
checkers.
"""
# Fake the absence of pamauth, even if PyPAM is installed
self.patch(tap, "pamauth", None)
config = tap.Options()
service = tap.makeService(config)
portal = service.factory.portal
Expand Down
Loading