<s> """<STR_LIT>""" <EOL> import os , base64 , binascii , errno <EOL> try : <EOL> import pwd <EOL> except ImportError : <EOL> pwd = None <EOL> else : <EOL> import crypt <EOL> try : <EOL> import spwd <EOL> except ImportError : <EOL> spwd = None <EOL> try : <EOL> import shadow <EOL> except ImportError : <EOL> shadow = None <EOL> else : <EOL> shadow = None <EOL> try : <EOL> from twisted . cred import pamauth <EOL> except ImportError : <EOL> pamauth = None <EOL> from zope . interface import implements , providedBy <EOL> from twisted . conch import error <EOL> from twisted . conch . ssh import keys <EOL> from twisted . cred . checkers import ICredentialsChecker <EOL> from twisted . cred . credentials import IUsernamePassword , ISSHPrivateKey <EOL> from twisted . cred . error import UnauthorizedLogin , UnhandledCredentials <EOL> from twisted . internet import defer <EOL> from twisted . python import failure , reflect , log <EOL> from twisted . python . util import runAsEffectiveUser <EOL> from twisted . python . filepath import FilePath <EOL> def verifyCryptedPassword ( crypted , pw ) : <EOL> return crypt . crypt ( pw , crypted ) == crypted <EOL> def _pwdGetByName ( username ) : <EOL> """<STR_LIT>""" <EOL> if pwd is None : <EOL> return None <EOL> return pwd . getpwnam ( username ) <EOL> def _shadowGetByName ( username ) : <EOL> """<STR_LIT>""" <EOL> if spwd is not None : <EOL> f = spwd . getspnam <EOL> elif shadow is not None : <EOL> f = shadow . getspnam <EOL> else : <EOL> return None <EOL> return runAsEffectiveUser ( <NUM_LIT:0> , <NUM_LIT:0> , f , username ) <EOL> class UNIXPasswordDatabase : <EOL> """<STR_LIT>""" <EOL> credentialInterfaces = IUsernamePassword , <EOL> implements ( ICredentialsChecker ) <EOL> def __init__ ( self , getByNameFunctions = None ) : <EOL> if getByNameFunctions is None : <EOL> getByNameFunctions = [ _pwdGetByName , _shadowGetByName ] <EOL> self . _getByNameFunctions = getByNameFunctions <EOL> def requestAvatarId ( self , credentials ) : <EOL> for func in self . _getByNameFunctions : <EOL> try : <EOL> pwnam = func ( credentials . username ) <EOL> except KeyError : <EOL> return defer . fail ( UnauthorizedLogin ( "<STR_LIT>" ) ) <EOL> else : <EOL> if pwnam is not None : <EOL> crypted = pwnam [ <NUM_LIT:1> ] <EOL> if crypted == '<STR_LIT>' : <EOL> continue <EOL> if verifyCryptedPassword ( crypted , credentials . password ) : <EOL> return defer . succeed ( credentials . username ) <EOL> return defer . fail ( UnauthorizedLogin ( "<STR_LIT>" ) ) <EOL> class SSHPublicKeyDatabase : <EOL> """<STR_LIT>""" <EOL> implements ( ICredentialsChecker ) <EOL> credentialInterfaces = ( ISSHPrivateKey , ) <EOL> _userdb = pwd <EOL> def requestAvatarId ( self , credentials ) : <EOL> d = defer . maybeDeferred ( self . checkKey , credentials ) <EOL> d . addCallback ( self . _cbRequestAvatarId , credentials ) <EOL> d . addErrback ( self . _ebRequestAvatarId ) <EOL> return d <EOL> def _cbRequestAvatarId ( self , validKey , credentials ) : <EOL> """<STR_LIT>""" <EOL> if not validKey : <EOL> return failure . Failure ( UnauthorizedLogin ( "<STR_LIT>" ) ) <EOL> if not credentials . signature : <EOL> return failure . Failure ( error . ValidPublicKey ( ) ) <EOL> else : <EOL> try : <EOL> pubKey = keys . Key . fromString ( credentials . blob ) <EOL> if pubKey . verify ( credentials . signature , credentials . sigData ) : <EOL> return credentials . username <EOL> except : <EOL> log . err ( ) <EOL> return failure . Failure ( UnauthorizedLogin ( '<STR_LIT>' ) ) <EOL> return failure . Failure ( UnauthorizedLogin ( "<STR_LIT>" ) ) <EOL> def getAuthorizedKeysFiles ( self , credentials ) : <EOL> """<STR_LIT>""" <EOL> pwent = self . _userdb . getpwnam ( credentials . username ) <EOL> root = FilePath ( pwent . pw_dir ) . child ( '<STR_LIT>' ) <EOL> files = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> return [ root . child ( f ) for f in files ] <EOL> def checkKey ( self , credentials ) : <EOL> """<STR_LIT>""" <EOL> uid , gid = os . geteuid ( ) , os . getegid ( ) <EOL> ouid , ogid = self . _userdb . getpwnam ( credentials . username ) [ <NUM_LIT:2> : <NUM_LIT:4> ] <EOL> for filepath in self . getAuthorizedKeysFiles ( credentials ) : <EOL> if not filepath . exists ( ) : <EOL> continue <EOL> try : <EOL> lines = filepath . open ( ) <EOL> except IOError , e : <EOL> if e . errno == errno . EACCES : <EOL> lines = runAsEffectiveUser ( ouid , ogid , filepath . open ) <EOL> else : <EOL> raise <EOL> for l in lines : <EOL> l2 = l . split ( ) <EOL> if len ( l2 ) < <NUM_LIT:2> : <EOL> continue <EOL> try : <EOL> if base64 . decodestring ( l2 [ <NUM_LIT:1> ] ) == credentials . blob : <EOL> return True <EOL> except binascii . Error : <EOL> continue <EOL> return False <EOL> def _ebRequestAvatarId ( self , f ) : <EOL> if not f . check ( UnauthorizedLogin ) : <EOL> log . msg ( f ) <EOL> return failure . Failure ( UnauthorizedLogin ( "<STR_LIT>" ) ) <EOL> return f <EOL> class SSHProtocolChecker : <EOL> """<STR_LIT>""" <EOL> implements ( ICredentialsChecker ) <EOL> def __init__ ( self ) : <EOL> self . checkers = { } <EOL> self . successfulCredentials = { } <EOL> def get_credentialInterfaces ( self ) : <EOL> return self . checkers . keys ( ) <EOL> credentialInterfaces = property ( get_credentialInterfaces ) <EOL> def registerChecker ( self , checker , * credentialInterfaces ) : <EOL> if not credentialInterfaces : <EOL> credentialInterfaces = checker . credentialInterfaces <EOL> for credentialInterface in credentialInterfaces : <EOL> self . checkers [ credentialInterface ] = checker <EOL> def requestAvatarId ( self , credentials ) : <EOL> """<STR_LIT>""" <EOL> ifac = providedBy ( credentials ) <EOL> for i in ifac : <EOL> c = self . checkers . get ( i ) <EOL> if c is not None : <EOL> d = defer . maybeDeferred ( c . requestAvatarId , credentials ) <EOL> return d . addCallback ( self . _cbGoodAuthentication , <EOL> credentials ) <EOL> return defer . fail ( UnhandledCredentials ( "<STR_LIT>" % '<STR_LIT:U+002CU+0020>' . join ( map ( reflect . qual , ifac ) ) ) ) <EOL> def _cbGoodAuthentication ( self , avatarId , credentials ) : <EOL> """<STR_LIT>""" <EOL> if avatarId not in self . successfulCredentials : <EOL> self . successfulCredentials [ avatarId ] = [ ] <EOL> self . successfulCredentials [ avatarId ] . append ( credentials ) <EOL> if self . areDone ( avatarId ) : <EOL> del self . successfulCredentials [ avatarId ] <EOL> return avatarId <EOL> else : <EOL> raise error . NotEnoughAuthentication ( ) <EOL> def areDone ( self , avatarId ) : <EOL> """<STR_LIT>""" <EOL> return True </s>
