<s> """<STR_LIT>""" <EOL> import struct <EOL> from twisted . python import log <EOL> from twisted . internet import defer <EOL> from twisted . conch . interfaces import IConchUser <EOL> from twisted . conch . ssh import userauth <EOL> from twisted . conch . ssh . common import NS , getNS <EOL> from twisted . conch import error <EOL> from cowrie . core import credentials <EOL> class HoneyPotSSHUserAuthServer ( userauth . SSHUserAuthServer ) : <EOL> """<STR_LIT>""" <EOL> def serviceStarted ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . interfaceToMethod [ credentials . IUsername ] = '<STR_LIT:none>' <EOL> self . interfaceToMethod [ credentials . IUsernamePasswordIP ] = '<STR_LIT:password>' <EOL> self . interfaceToMethod [ credentials . IPluggableAuthenticationModulesIP ] = '<STR_LIT>' <EOL> self . bannerSent = False <EOL> self . _pamDeferred = None <EOL> userauth . SSHUserAuthServer . serviceStarted ( self ) <EOL> def sendBanner ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . bannerSent : <EOL> return <EOL> self . bannerSent = True <EOL> try : <EOL> honeyfs = self . portal . realm . cfg . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> issuefile = honeyfs + "<STR_LIT>" <EOL> data = open ( issuefile ) . read ( ) <EOL> except IOError : <EOL> return <EOL> if not data or not len ( data . strip ( ) ) : <EOL> return <EOL> self . transport . sendPacket ( <EOL> userauth . MSG_USERAUTH_BANNER , NS ( data ) + NS ( '<STR_LIT>' ) ) <EOL> def ssh_USERAUTH_REQUEST ( self , packet ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . sendBanner ( ) <EOL> return userauth . SSHUserAuthServer . ssh_USERAUTH_REQUEST ( self , packet ) <EOL> def auth_publickey ( self , packet ) : <EOL> """<STR_LIT>""" <EOL> algName , blob , rest = getNS ( packet [ <NUM_LIT:1> : ] , <NUM_LIT:2> ) <EOL> if not algName in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> log . msg ( "<STR_LIT>" % ( algName , ) ) <EOL> return defer . fail ( error . ConchError ( "<STR_LIT>" ) ) <EOL> return userauth . SSHUserAuthServer . auth_publickey ( self , packet ) <EOL> def auth_none ( self , packet ) : <EOL> """<STR_LIT>""" <EOL> c = credentials . Username ( self . user ) <EOL> srcIp = self . transport . transport . getPeer ( ) . host <EOL> return self . portal . login ( c , srcIp , IConchUser ) <EOL> def auth_password ( self , packet ) : <EOL> """<STR_LIT>""" <EOL> password = getNS ( packet [ <NUM_LIT:1> : ] ) [ <NUM_LIT:0> ] <EOL> srcIp = self . transport . transport . getPeer ( ) . host <EOL> c = credentials . UsernamePasswordIP ( self . user , password , srcIp ) <EOL> return self . portal . login ( c , srcIp , <EOL> IConchUser ) . addErrback ( self . _ebPassword ) <EOL> def auth_keyboard_interactive ( self , packet ) : <EOL> """<STR_LIT>""" <EOL> if self . _pamDeferred is not None : <EOL> self . transport . sendDisconnect ( <EOL> transport . DISCONNECT_PROTOCOL_ERROR , <EOL> "<STR_LIT>" ) <EOL> return defer . fail ( error . IgnoreAuthentication ( ) ) <EOL> src_ip = self . transport . transport . getPeer ( ) . host <EOL> c = credentials . PluggableAuthenticationModulesIP ( self . user , <EOL> self . _pamConv , src_ip ) <EOL> return self . portal . login ( c , src_ip , <EOL> IConchUser ) . addErrback ( self . _ebPassword ) <EOL> def _pamConv ( self , items ) : <EOL> """<STR_LIT>""" <EOL> resp = [ ] <EOL> for message , kind in items : <EOL> if kind == <NUM_LIT:1> : <EOL> resp . append ( ( message , <NUM_LIT:0> ) ) <EOL> elif kind == <NUM_LIT:2> : <EOL> resp . append ( ( message , <NUM_LIT:1> ) ) <EOL> elif kind in ( <NUM_LIT:3> , <NUM_LIT:4> ) : <EOL> return defer . fail ( error . ConchError ( <EOL> '<STR_LIT>' ) ) <EOL> else : <EOL> return defer . fail ( error . ConchError ( <EOL> '<STR_LIT>' % ( kind , ) ) ) <EOL> packet = NS ( '<STR_LIT>' ) + NS ( '<STR_LIT>' ) + NS ( '<STR_LIT>' ) <EOL> packet += struct . pack ( '<STR_LIT>' , len ( resp ) ) <EOL> for prompt , echo in resp : <EOL> packet += NS ( prompt ) <EOL> packet += chr ( echo ) <EOL> self . transport . sendPacket ( userauth . MSG_USERAUTH_INFO_REQUEST , packet ) <EOL> self . _pamDeferred = defer . Deferred ( ) <EOL> return self . _pamDeferred <EOL> def ssh_USERAUTH_INFO_RESPONSE ( self , packet ) : <EOL> """<STR_LIT>""" <EOL> d , self . _pamDeferred = self . _pamDeferred , None <EOL> try : <EOL> resp = [ ] <EOL> numResps = struct . unpack ( '<STR_LIT>' , packet [ : <NUM_LIT:4> ] ) [ <NUM_LIT:0> ] <EOL> packet = packet [ <NUM_LIT:4> : ] <EOL> while len ( resp ) < numResps : <EOL> response , packet = getNS ( packet ) <EOL> resp . append ( ( response , <NUM_LIT:0> ) ) <EOL> if packet : <EOL> raise error . ConchError ( <EOL> "<STR_LIT>" . format ( len ( packet ) ) ) <EOL> except : <EOL> d . errback ( failure . Failure ( ) ) <EOL> else : <EOL> d . callback ( resp ) </s>
