<s> __doc__ = """<STR_LIT>""" <EOL> __author__ = '<STR_LIT>' <EOL> __date__ = '<STR_LIT>' <EOL> import time <EOL> from cherrypy . _cpcompat import parse_http_list , parse_keqv_list <EOL> import cherrypy <EOL> from cherrypy . _cpcompat import md5 , ntob <EOL> md5_hex = lambda s : md5 ( ntob ( s ) ) . hexdigest ( ) <EOL> qop_auth = '<STR_LIT>' <EOL> qop_auth_int = '<STR_LIT>' <EOL> valid_qops = ( qop_auth , qop_auth_int ) <EOL> valid_algorithms = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def TRACE ( msg ) : <EOL> cherrypy . log ( msg , context = '<STR_LIT>' ) <EOL> def get_ha1_dict_plain ( user_password_dict ) : <EOL> """<STR_LIT>""" <EOL> def get_ha1 ( realm , username ) : <EOL> password = user_password_dict . get ( username ) <EOL> if password : <EOL> return md5_hex ( '<STR_LIT>' % ( username , realm , password ) ) <EOL> return None <EOL> return get_ha1 <EOL> def get_ha1_dict ( user_ha1_dict ) : <EOL> """<STR_LIT>""" <EOL> def get_ha1 ( realm , username ) : <EOL> return user_ha1_dict . get ( user ) <EOL> return get_ha1 <EOL> def get_ha1_file_htdigest ( filename ) : <EOL> """<STR_LIT>""" <EOL> def get_ha1 ( realm , username ) : <EOL> result = None <EOL> f = open ( filename , '<STR_LIT:r>' ) <EOL> for line in f : <EOL> u , r , ha1 = line . rstrip ( ) . split ( '<STR_LIT::>' ) <EOL> if u == username and r == realm : <EOL> result = ha1 <EOL> break <EOL> f . close ( ) <EOL> return result <EOL> return get_ha1 <EOL> def synthesize_nonce ( s , key , timestamp = None ) : <EOL> """<STR_LIT>""" <EOL> if timestamp is None : <EOL> timestamp = int ( time . time ( ) ) <EOL> h = md5_hex ( '<STR_LIT>' % ( timestamp , s , key ) ) <EOL> nonce = '<STR_LIT>' % ( timestamp , h ) <EOL> return nonce <EOL> def H ( s ) : <EOL> """<STR_LIT>""" <EOL> return md5_hex ( s ) <EOL> class HttpDigestAuthorization ( object ) : <EOL> """<STR_LIT>""" <EOL> def errmsg ( self , s ) : <EOL> return '<STR_LIT>' % s <EOL> def __init__ ( self , auth_header , http_method , debug = False ) : <EOL> self . http_method = http_method <EOL> self . debug = debug <EOL> scheme , params = auth_header . split ( "<STR_LIT:U+0020>" , <NUM_LIT:1> ) <EOL> self . scheme = scheme . lower ( ) <EOL> if self . scheme != '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . auth_header = auth_header <EOL> items = parse_http_list ( params ) <EOL> paramsd = parse_keqv_list ( items ) <EOL> self . realm = paramsd . get ( '<STR_LIT>' ) <EOL> self . username = paramsd . get ( '<STR_LIT:username>' ) <EOL> self . nonce = paramsd . get ( '<STR_LIT>' ) <EOL> self . uri = paramsd . get ( '<STR_LIT>' ) <EOL> self . method = paramsd . get ( '<STR_LIT>' ) <EOL> self . response = paramsd . get ( '<STR_LIT>' ) <EOL> self . algorithm = paramsd . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . cnonce = paramsd . get ( '<STR_LIT>' ) <EOL> self . opaque = paramsd . get ( '<STR_LIT>' ) <EOL> self . qop = paramsd . get ( '<STR_LIT>' ) <EOL> self . nc = paramsd . get ( '<STR_LIT>' ) <EOL> if self . algorithm not in valid_algorithms : <EOL> raise ValueError ( self . errmsg ( "<STR_LIT>" % self . algorithm ) ) <EOL> has_reqd = self . username and self . realm and self . nonce and self . uri and self . response <EOL> if not has_reqd : <EOL> raise ValueError ( self . errmsg ( "<STR_LIT>" ) ) <EOL> if self . qop : <EOL> if self . qop not in valid_qops : <EOL> raise ValueError ( self . errmsg ( "<STR_LIT>" % self . qop ) ) <EOL> if not ( self . cnonce and self . nc ) : <EOL> raise ValueError ( self . errmsg ( "<STR_LIT>" ) ) <EOL> else : <EOL> if self . cnonce or self . nc : <EOL> raise ValueError ( self . errmsg ( "<STR_LIT>" ) ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % self . auth_header <EOL> def validate_nonce ( self , s , key ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> timestamp , hashpart = self . nonce . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> s_timestamp , s_hashpart = synthesize_nonce ( s , key , timestamp ) . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> is_valid = s_hashpart == hashpart <EOL> if self . debug : <EOL> TRACE ( '<STR_LIT>' % is_valid ) <EOL> return is_valid <EOL> except ValueError : <EOL> pass <EOL> return False <EOL> def is_nonce_stale ( self , max_age_seconds = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> timestamp , hashpart = self . nonce . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> if int ( timestamp ) + max_age_seconds > int ( time . time ( ) ) : <EOL> return False <EOL> except ValueError : <EOL> pass <EOL> if self . debug : <EOL> TRACE ( "<STR_LIT>" ) <EOL> return True <EOL> def HA2 ( self , entity_body = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if self . qop is None or self . qop == "<STR_LIT>" : <EOL> a2 = '<STR_LIT>' % ( self . http_method , self . uri ) <EOL> elif self . qop == "<STR_LIT>" : <EOL> a2 = "<STR_LIT>" % ( self . http_method , self . uri , H ( entity_body ) ) <EOL> else : <EOL> raise ValueError ( self . errmsg ( "<STR_LIT>" ) ) <EOL> return H ( a2 ) <EOL> def request_digest ( self , ha1 , entity_body = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> ha2 = self . HA2 ( entity_body ) <EOL> if self . qop : <EOL> req = "<STR_LIT>" % ( self . nonce , self . nc , self . cnonce , self . qop , ha2 ) <EOL> else : <EOL> req = "<STR_LIT>" % ( self . nonce , ha2 ) <EOL> if self . algorithm == '<STR_LIT>' : <EOL> ha1 = H ( '<STR_LIT>' % ( ha1 , self . nonce , self . cnonce ) ) <EOL> digest = H ( '<STR_LIT>' % ( ha1 , req ) ) <EOL> return digest <EOL> def www_authenticate ( realm , key , algorithm = '<STR_LIT>' , nonce = None , qop = qop_auth , stale = False ) : <EOL> """<STR_LIT>""" <EOL> if qop not in valid_qops : <EOL> raise ValueError ( "<STR_LIT>" % qop ) <EOL> if algorithm not in valid_algorithms : <EOL> raise ValueError ( "<STR_LIT>" % algorithm ) <EOL> if nonce is None : <EOL> nonce = synthesize_nonce ( realm , key ) <EOL> s = '<STR_LIT>' % ( <EOL> realm , nonce , algorithm , qop ) <EOL> if stale : <EOL> s += '<STR_LIT>' <EOL> return s <EOL> def digest_auth ( realm , get_ha1 , key , debug = False ) : <EOL> """<STR_LIT>""" <EOL> request = cherrypy . serving . request <EOL> auth_header = request . headers . get ( '<STR_LIT>' ) <EOL> nonce_is_stale = False <EOL> if auth_header is not None : <EOL> try : <EOL> auth = HttpDigestAuthorization ( auth_header , request . method , debug = debug ) <EOL> except ValueError : <EOL> raise cherrypy . HTTPError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> if debug : <EOL> TRACE ( str ( auth ) ) <EOL> if auth . validate_nonce ( realm , key ) : <EOL> ha1 = get_ha1 ( realm , auth . username ) <EOL> if ha1 is not None : <EOL> digest = auth . request_digest ( ha1 , entity_body = request . body ) <EOL> if digest == auth . response : <EOL> if debug : <EOL> TRACE ( "<STR_LIT>" ) <EOL> nonce_is_stale = auth . is_nonce_stale ( max_age_seconds = <NUM_LIT> ) <EOL> if not nonce_is_stale : <EOL> request . login = auth . username <EOL> if debug : <EOL> TRACE ( "<STR_LIT>" % auth . username ) <EOL> return <EOL> header = www_authenticate ( realm , key , stale = nonce_is_stale ) <EOL> if debug : <EOL> TRACE ( header ) <EOL> cherrypy . serving . response . headers [ '<STR_LIT>' ] = header <EOL> raise cherrypy . HTTPError ( <NUM_LIT> , "<STR_LIT>" ) </s>
