<s> import hashlib <EOL> import re <EOL> import time <EOL> import gntp . shim <EOL> import gntp . errors as errors <EOL> __all__ = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> GNTP_INFO_LINE = re . compile ( <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' , <EOL> re . IGNORECASE <EOL> ) <EOL> GNTP_INFO_LINE_SHORT = re . compile ( <EOL> '<STR_LIT>' , <EOL> re . IGNORECASE <EOL> ) <EOL> GNTP_HEADER = re . compile ( '<STR_LIT>' ) <EOL> GNTP_EOL = gntp . shim . b ( '<STR_LIT:\r\n>' ) <EOL> GNTP_SEP = gntp . shim . b ( '<STR_LIT>' ) <EOL> class _GNTPBuffer ( gntp . shim . StringIO ) : <EOL> """<STR_LIT>""" <EOL> def writeln ( self , value = None ) : <EOL> if value : <EOL> self . write ( gntp . shim . b ( value ) ) <EOL> self . write ( GNTP_EOL ) <EOL> def writeheader ( self , key , value ) : <EOL> if not isinstance ( value , str ) : <EOL> value = str ( value ) <EOL> self . write ( gntp . shim . b ( key ) ) <EOL> self . write ( GNTP_SEP ) <EOL> self . write ( gntp . shim . b ( value ) ) <EOL> self . write ( GNTP_EOL ) <EOL> class _GNTPBase ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , messagetype = None , version = '<STR_LIT:1.0>' , encryption = None ) : <EOL> self . info = { <EOL> '<STR_LIT:version>' : version , <EOL> '<STR_LIT>' : messagetype , <EOL> '<STR_LIT>' : encryption <EOL> } <EOL> self . hash_algo = { <EOL> '<STR_LIT>' : hashlib . md5 , <EOL> '<STR_LIT>' : hashlib . sha1 , <EOL> '<STR_LIT>' : hashlib . sha256 , <EOL> '<STR_LIT>' : hashlib . sha512 , <EOL> } <EOL> self . headers = { } <EOL> self . resources = { } <EOL> def __str__ ( self ) : <EOL> return self . encode ( ) <EOL> def _parse_info ( self , data ) : <EOL> """<STR_LIT>""" <EOL> match = GNTP_INFO_LINE . match ( data ) <EOL> if not match : <EOL> raise errors . ParseError ( '<STR_LIT>' ) <EOL> info = match . groupdict ( ) <EOL> if info [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> info [ '<STR_LIT>' ] = None <EOL> return info <EOL> def set_password ( self , password , encryptAlgo = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if not password : <EOL> self . info [ '<STR_LIT>' ] = None <EOL> self . info [ '<STR_LIT>' ] = None <EOL> return <EOL> self . password = gntp . shim . b ( password ) <EOL> self . encryptAlgo = encryptAlgo . upper ( ) <EOL> if not self . encryptAlgo in self . hash_algo : <EOL> raise errors . UnsupportedError ( '<STR_LIT>' % self . encryptAlgo ) <EOL> hashfunction = self . hash_algo . get ( self . encryptAlgo ) <EOL> password = password . encode ( '<STR_LIT:utf8>' ) <EOL> seed = time . ctime ( ) . encode ( '<STR_LIT:utf8>' ) <EOL> salt = hashfunction ( seed ) . hexdigest ( ) <EOL> saltHash = hashfunction ( seed ) . digest ( ) <EOL> keyBasis = password + saltHash <EOL> key = hashfunction ( keyBasis ) . digest ( ) <EOL> keyHash = hashfunction ( key ) . hexdigest ( ) <EOL> self . info [ '<STR_LIT>' ] = self . encryptAlgo <EOL> self . info [ '<STR_LIT>' ] = keyHash . upper ( ) <EOL> self . info [ '<STR_LIT>' ] = salt . upper ( ) <EOL> def _decode_hex ( self , value ) : <EOL> """<STR_LIT>""" <EOL> result = '<STR_LIT>' <EOL> for i in range ( <NUM_LIT:0> , len ( value ) , <NUM_LIT:2> ) : <EOL> tmp = int ( value [ i : i + <NUM_LIT:2> ] , <NUM_LIT:16> ) <EOL> result += chr ( tmp ) <EOL> return result <EOL> def _decode_binary ( self , rawIdentifier , identifier ) : <EOL> rawIdentifier += '<STR_LIT>' <EOL> dataLength = int ( identifier [ '<STR_LIT>' ] ) <EOL> pointerStart = self . raw . find ( rawIdentifier ) + len ( rawIdentifier ) <EOL> pointerEnd = pointerStart + dataLength <EOL> data = self . raw [ pointerStart : pointerEnd ] <EOL> if not len ( data ) == dataLength : <EOL> raise errors . ParseError ( '<STR_LIT>' % ( dataLength , len ( data ) ) ) <EOL> return data <EOL> def _validate_password ( self , password ) : <EOL> """<STR_LIT>""" <EOL> self . password = password <EOL> if password is None : <EOL> raise errors . AuthError ( '<STR_LIT>' ) <EOL> keyHash = self . info . get ( '<STR_LIT>' , None ) <EOL> if keyHash is None and self . password is None : <EOL> return True <EOL> if keyHash is None : <EOL> raise errors . AuthError ( '<STR_LIT>' ) <EOL> if self . password is None : <EOL> raise errors . AuthError ( '<STR_LIT>' ) <EOL> keyHashAlgorithmID = self . info . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> password = self . password . encode ( '<STR_LIT:utf8>' ) <EOL> saltHash = self . _decode_hex ( self . info [ '<STR_LIT>' ] ) <EOL> keyBasis = password + saltHash <EOL> self . key = self . hash_algo [ keyHashAlgorithmID ] ( keyBasis ) . digest ( ) <EOL> keyHash = self . hash_algo [ keyHashAlgorithmID ] ( self . key ) . hexdigest ( ) <EOL> if not keyHash . upper ( ) == self . info [ '<STR_LIT>' ] . upper ( ) : <EOL> raise errors . AuthError ( '<STR_LIT>' ) <EOL> return True <EOL> def validate ( self ) : <EOL> """<STR_LIT>""" <EOL> for header in self . _requiredHeaders : <EOL> if not self . headers . get ( header , False ) : <EOL> raise errors . ParseError ( '<STR_LIT>' + header ) <EOL> def _format_info ( self ) : <EOL> """<STR_LIT>""" <EOL> info = '<STR_LIT>' % ( <EOL> self . info . get ( '<STR_LIT:version>' ) , <EOL> self . info . get ( '<STR_LIT>' ) , <EOL> ) <EOL> if self . info . get ( '<STR_LIT>' , None ) : <EOL> info += '<STR_LIT>' % ( <EOL> self . info . get ( '<STR_LIT>' ) , <EOL> self . info . get ( '<STR_LIT>' ) , <EOL> ) <EOL> else : <EOL> info += '<STR_LIT>' <EOL> if self . info . get ( '<STR_LIT>' , None ) : <EOL> info += '<STR_LIT>' % ( <EOL> self . info . get ( '<STR_LIT>' ) , <EOL> self . info . get ( '<STR_LIT>' ) , <EOL> self . info . get ( '<STR_LIT>' ) <EOL> ) <EOL> return info <EOL> def _parse_dict ( self , data ) : <EOL> """<STR_LIT>""" <EOL> d = { } <EOL> for line in data . split ( '<STR_LIT:\r\n>' ) : <EOL> match = GNTP_HEADER . match ( line ) <EOL> if not match : <EOL> continue <EOL> key = match . group ( <NUM_LIT:1> ) . strip ( ) <EOL> val = match . group ( <NUM_LIT:2> ) . strip ( ) <EOL> d [ key ] = val <EOL> return d <EOL> def add_header ( self , key , value ) : <EOL> self . headers [ key ] = value <EOL> def add_resource ( self , data ) : <EOL> """<STR_LIT>""" <EOL> data = gntp . shim . b ( data ) <EOL> identifier = hashlib . md5 ( data ) . hexdigest ( ) <EOL> self . resources [ identifier ] = data <EOL> return '<STR_LIT>' % identifier <EOL> def decode ( self , data , password = None ) : <EOL> """<STR_LIT>""" <EOL> self . password = password <EOL> self . raw = gntp . shim . u ( data ) <EOL> parts = self . raw . split ( '<STR_LIT>' ) <EOL> self . info = self . _parse_info ( self . raw ) <EOL> self . headers = self . _parse_dict ( parts [ <NUM_LIT:0> ] ) <EOL> def encode ( self ) : <EOL> """<STR_LIT>""" <EOL> buff = _GNTPBuffer ( ) <EOL> buff . writeln ( self . _format_info ( ) ) <EOL> for k , v in self . headers . items ( ) : <EOL> buff . writeheader ( k , v ) <EOL> buff . writeln ( ) <EOL> for resource , data in self . resources . items ( ) : <EOL> buff . writeheader ( '<STR_LIT>' , resource ) <EOL> buff . writeheader ( '<STR_LIT>' , len ( data ) ) <EOL> buff . writeln ( ) <EOL> buff . write ( data ) <EOL> buff . writeln ( ) <EOL> buff . writeln ( ) <EOL> return buff . getvalue ( ) <EOL> class GNTPRegister ( _GNTPBase ) : <EOL> """<STR_LIT>""" <EOL> _requiredHeaders = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> _requiredNotificationHeaders = [ '<STR_LIT>' ] <EOL> def __init__ ( self , data = None , password = None ) : <EOL> _GNTPBase . __init__ ( self , '<STR_LIT>' ) <EOL> self . notifications = [ ] <EOL> if data : <EOL> self . decode ( data , password ) <EOL> else : <EOL> self . set_password ( password ) <EOL> self . add_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . add_header ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> def validate ( self ) : <EOL> '''<STR_LIT>''' <EOL> for header in self . _requiredHeaders : <EOL> if not self . headers . get ( header , False ) : <EOL> raise errors . ParseError ( '<STR_LIT>' + header ) <EOL> for notice in self . notifications : <EOL> for header in self . _requiredNotificationHeaders : <EOL> if not notice . get ( header , False ) : <EOL> raise errors . ParseError ( '<STR_LIT>' + header ) <EOL> def decode ( self , data , password ) : <EOL> """<STR_LIT>""" <EOL> self . raw = gntp . shim . u ( data ) <EOL> parts = self . raw . split ( '<STR_LIT>' ) <EOL> self . info = self . _parse_info ( self . raw ) <EOL> self . _validate_password ( password ) <EOL> self . headers = self . _parse_dict ( parts [ <NUM_LIT:0> ] ) <EOL> for i , part in enumerate ( parts ) : <EOL> if i == <NUM_LIT:0> : <EOL> continue <EOL> if part . strip ( ) == '<STR_LIT>' : <EOL> continue <EOL> notice = self . _parse_dict ( part ) <EOL> if notice . get ( '<STR_LIT>' , False ) : <EOL> self . notifications . append ( notice ) <EOL> elif notice . get ( '<STR_LIT>' , False ) : <EOL> notice [ '<STR_LIT>' ] = self . _decode_binary ( part , notice ) <EOL> self . resources [ notice . get ( '<STR_LIT>' ) ] = notice <EOL> def add_notification ( self , name , enabled = True ) : <EOL> """<STR_LIT>""" <EOL> notice = { } <EOL> notice [ '<STR_LIT>' ] = name <EOL> notice [ '<STR_LIT>' ] = enabled <EOL> self . notifications . append ( notice ) <EOL> self . add_header ( '<STR_LIT>' , len ( self . notifications ) ) <EOL> def encode ( self ) : <EOL> """<STR_LIT>""" <EOL> buff = _GNTPBuffer ( ) <EOL> buff . writeln ( self . _format_info ( ) ) <EOL> for k , v in self . headers . items ( ) : <EOL> buff . writeheader ( k , v ) <EOL> buff . writeln ( ) <EOL> if len ( self . notifications ) > <NUM_LIT:0> : <EOL> for notice in self . notifications : <EOL> for k , v in notice . items ( ) : <EOL> buff . writeheader ( k , v ) <EOL> buff . writeln ( ) <EOL> for resource , data in self . resources . items ( ) : <EOL> buff . writeheader ( '<STR_LIT>' , resource ) <EOL> buff . writeheader ( '<STR_LIT>' , len ( data ) ) <EOL> buff . writeln ( ) <EOL> buff . write ( data ) <EOL> buff . writeln ( ) <EOL> buff . writeln ( ) <EOL> return buff . getvalue ( ) <EOL> class GNTPNotice ( _GNTPBase ) : <EOL> """<STR_LIT>""" <EOL> _requiredHeaders = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> def __init__ ( self , data = None , app = None , name = None , title = None , password = None ) : <EOL> _GNTPBase . __init__ ( self , '<STR_LIT>' ) <EOL> if data : <EOL> self . decode ( data , password ) <EOL> else : <EOL> self . set_password ( password ) <EOL> if app : <EOL> self . add_header ( '<STR_LIT>' , app ) <EOL> if name : <EOL> self . add_header ( '<STR_LIT>' , name ) <EOL> if title : <EOL> self . add_header ( '<STR_LIT>' , title ) <EOL> def decode ( self , data , password ) : <EOL> """<STR_LIT>""" <EOL> self . raw = gntp . shim . u ( data ) <EOL> parts = self . raw . split ( '<STR_LIT>' ) <EOL> self . info = self . _parse_info ( self . raw ) <EOL> self . _validate_password ( password ) <EOL> self . headers = self . _parse_dict ( parts [ <NUM_LIT:0> ] ) <EOL> for i , part in enumerate ( parts ) : <EOL> if i == <NUM_LIT:0> : <EOL> continue <EOL> if part . strip ( ) == '<STR_LIT>' : <EOL> continue <EOL> notice = self . _parse_dict ( part ) <EOL> if notice . get ( '<STR_LIT>' , False ) : <EOL> notice [ '<STR_LIT>' ] = self . _decode_binary ( part , notice ) <EOL> self . resources [ notice . get ( '<STR_LIT>' ) ] = notice <EOL> class GNTPSubscribe ( _GNTPBase ) : <EOL> """<STR_LIT>""" <EOL> _requiredHeaders = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> def __init__ ( self , data = None , password = None ) : <EOL> _GNTPBase . __init__ ( self , '<STR_LIT>' ) <EOL> if data : <EOL> self . decode ( data , password ) <EOL> else : <EOL> self . set_password ( password ) <EOL> class GNTPOK ( _GNTPBase ) : <EOL> """<STR_LIT>""" <EOL> _requiredHeaders = [ '<STR_LIT>' ] <EOL> def __init__ ( self , data = None , action = None ) : <EOL> _GNTPBase . __init__ ( self , '<STR_LIT>' ) <EOL> if data : <EOL> self . decode ( data ) <EOL> if action : <EOL> self . add_header ( '<STR_LIT>' , action ) <EOL> class GNTPError ( _GNTPBase ) : <EOL> """<STR_LIT>""" <EOL> _requiredHeaders = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> def __init__ ( self , data = None , errorcode = None , errordesc = None ) : <EOL> _GNTPBase . __init__ ( self , '<STR_LIT>' ) <EOL> if data : <EOL> self . decode ( data ) <EOL> if errorcode : <EOL> self . add_header ( '<STR_LIT>' , errorcode ) <EOL> self . add_header ( '<STR_LIT>' , errordesc ) <EOL> def error ( self ) : <EOL> return ( self . headers . get ( '<STR_LIT>' , None ) , <EOL> self . headers . get ( '<STR_LIT>' , None ) ) <EOL> def parse_gntp ( data , password = None ) : <EOL> """<STR_LIT>""" <EOL> data = gntp . shim . u ( data ) <EOL> match = GNTP_INFO_LINE_SHORT . match ( data ) <EOL> if not match : <EOL> raise errors . ParseError ( '<STR_LIT>' ) <EOL> info = match . groupdict ( ) <EOL> if info [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> return GNTPRegister ( data , password = password ) <EOL> elif info [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> return GNTPNotice ( data , password = password ) <EOL> elif info [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> return GNTPSubscribe ( data , password = password ) <EOL> elif info [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> return GNTPOK ( data ) <EOL> elif info [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> return GNTPError ( data ) <EOL> raise errors . ParseError ( '<STR_LIT>' ) </s>
