<s> """<STR_LIT>""" <EOL> import os , time , datetime , random , base64 <EOL> import os . path <EOL> from copy import deepcopy <EOL> try : <EOL> import cPickle as pickle <EOL> except ImportError : <EOL> import pickle <EOL> try : <EOL> import hashlib <EOL> sha1 = hashlib . sha1 <EOL> except ImportError : <EOL> import sha <EOL> sha1 = sha . new <EOL> import utils <EOL> import webapi as web <EOL> __all__ = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> ] <EOL> web . config . session_parameters = utils . storage ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False <EOL> } ) <EOL> class SessionExpired ( web . HTTPError ) : <EOL> def __init__ ( self , message ) : <EOL> web . HTTPError . __init__ ( self , '<STR_LIT>' , { } , data = message ) <EOL> class Session ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = [ <EOL> "<STR_LIT:store>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" <EOL> ] <EOL> def __init__ ( self , app , store , initializer = None ) : <EOL> self . store = store <EOL> self . _initializer = initializer <EOL> self . _last_cleanup_time = <NUM_LIT:0> <EOL> self . _config = utils . storage ( web . config . session_parameters ) <EOL> self . _data = utils . threadeddict ( ) <EOL> self . __getitem__ = self . _data . __getitem__ <EOL> self . __setitem__ = self . _data . __setitem__ <EOL> self . __delitem__ = self . _data . __delitem__ <EOL> if app : <EOL> app . add_processor ( self . _processor ) <EOL> def __contains__ ( self , name ) : <EOL> return name in self . _data <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( self . _data , name ) <EOL> def __setattr__ ( self , name , value ) : <EOL> if name in self . __slots__ : <EOL> object . __setattr__ ( self , name , value ) <EOL> else : <EOL> setattr ( self . _data , name , value ) <EOL> def __delattr__ ( self , name ) : <EOL> delattr ( self . _data , name ) <EOL> def _processor ( self , handler ) : <EOL> """<STR_LIT>""" <EOL> self . _cleanup ( ) <EOL> self . _load ( ) <EOL> try : <EOL> return handler ( ) <EOL> finally : <EOL> self . _save ( ) <EOL> def _load ( self ) : <EOL> """<STR_LIT>""" <EOL> cookie_name = self . _config . cookie_name <EOL> cookie_domain = self . _config . cookie_domain <EOL> cookie_path = self . _config . cookie_path <EOL> httponly = self . _config . httponly <EOL> self . session_id = web . cookies ( ) . get ( cookie_name ) <EOL> if self . session_id and not self . _valid_session_id ( self . session_id ) : <EOL> self . session_id = None <EOL> self . _check_expiry ( ) <EOL> if self . session_id : <EOL> d = self . store [ self . session_id ] <EOL> self . update ( d ) <EOL> self . _validate_ip ( ) <EOL> if not self . session_id : <EOL> self . session_id = self . _generate_session_id ( ) <EOL> if self . _initializer : <EOL> if isinstance ( self . _initializer , dict ) : <EOL> self . update ( deepcopy ( self . _initializer ) ) <EOL> elif hasattr ( self . _initializer , '<STR_LIT>' ) : <EOL> self . _initializer ( ) <EOL> self . ip = web . ctx . ip <EOL> def _check_expiry ( self ) : <EOL> if self . session_id and self . session_id not in self . store : <EOL> if self . _config . ignore_expiry : <EOL> self . session_id = None <EOL> else : <EOL> return self . expired ( ) <EOL> def _validate_ip ( self ) : <EOL> if self . session_id and self . get ( '<STR_LIT>' , None ) != web . ctx . ip : <EOL> if not self . _config . ignore_change_ip : <EOL> return self . expired ( ) <EOL> def _save ( self ) : <EOL> if not self . get ( '<STR_LIT>' ) : <EOL> self . _setcookie ( self . session_id ) <EOL> self . store [ self . session_id ] = dict ( self . _data ) <EOL> else : <EOL> self . _setcookie ( self . session_id , expires = - <NUM_LIT:1> ) <EOL> def _setcookie ( self , session_id , expires = '<STR_LIT>' , ** kw ) : <EOL> cookie_name = self . _config . cookie_name <EOL> cookie_domain = self . _config . cookie_domain <EOL> cookie_path = self . _config . cookie_path <EOL> httponly = self . _config . httponly <EOL> secure = self . _config . secure <EOL> web . setcookie ( cookie_name , session_id , expires = expires , domain = cookie_domain , httponly = httponly , secure = secure , path = cookie_path ) <EOL> def _generate_session_id ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> rand = os . urandom ( <NUM_LIT:16> ) <EOL> now = time . time ( ) <EOL> secret_key = self . _config . secret_key <EOL> session_id = sha1 ( "<STR_LIT>" % ( rand , now , utils . safestr ( web . ctx . ip ) , secret_key ) ) <EOL> session_id = session_id . hexdigest ( ) <EOL> if session_id not in self . store : <EOL> break <EOL> return session_id <EOL> def _valid_session_id ( self , session_id ) : <EOL> rx = utils . re_compile ( '<STR_LIT>' ) <EOL> return rx . match ( session_id ) <EOL> def _cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> current_time = time . time ( ) <EOL> timeout = self . _config . timeout <EOL> if current_time - self . _last_cleanup_time > timeout : <EOL> self . store . cleanup ( timeout ) <EOL> self . _last_cleanup_time = current_time <EOL> def expired ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _killed = True <EOL> self . _save ( ) <EOL> raise SessionExpired ( self . _config . expired_message ) <EOL> def kill ( self ) : <EOL> """<STR_LIT>""" <EOL> del self . store [ self . session_id ] <EOL> self . _killed = True <EOL> class Store : <EOL> """<STR_LIT>""" <EOL> def __contains__ ( self , key ) : <EOL> raise NotImplementedError <EOL> def __getitem__ ( self , key ) : <EOL> raise NotImplementedError <EOL> def __setitem__ ( self , key , value ) : <EOL> raise NotImplementedError <EOL> def cleanup ( self , timeout ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def encode ( self , session_dict ) : <EOL> """<STR_LIT>""" <EOL> pickled = pickle . dumps ( session_dict ) <EOL> return base64 . encodestring ( pickled ) <EOL> def decode ( self , session_data ) : <EOL> """<STR_LIT>""" <EOL> pickled = base64 . decodestring ( session_data ) <EOL> return pickle . loads ( pickled ) <EOL> class DiskStore ( Store ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , root ) : <EOL> if not os . path . exists ( root ) : <EOL> os . makedirs ( <EOL> os . path . abspath ( root ) <EOL> ) <EOL> self . root = root <EOL> def _get_path ( self , key ) : <EOL> if os . path . sep in key : <EOL> raise ValueError , "<STR_LIT>" % repr ( key ) <EOL> return os . path . join ( self . root , key ) <EOL> def __contains__ ( self , key ) : <EOL> path = self . _get_path ( key ) <EOL> return os . path . exists ( path ) <EOL> def __getitem__ ( self , key ) : <EOL> path = self . _get_path ( key ) <EOL> if os . path . exists ( path ) : <EOL> pickled = open ( path ) . read ( ) <EOL> return self . decode ( pickled ) <EOL> else : <EOL> raise KeyError , key <EOL> def __setitem__ ( self , key , value ) : <EOL> path = self . _get_path ( key ) <EOL> pickled = self . encode ( value ) <EOL> try : <EOL> f = open ( path , '<STR_LIT:w>' ) <EOL> try : <EOL> f . write ( pickled ) <EOL> finally : <EOL> f . close ( ) <EOL> except IOError : <EOL> pass <EOL> def __delitem__ ( self , key ) : <EOL> path = self . _get_path ( key ) <EOL> if os . path . exists ( path ) : <EOL> os . remove ( path ) <EOL> def cleanup ( self , timeout ) : <EOL> now = time . time ( ) <EOL> for f in os . listdir ( self . root ) : <EOL> path = self . _get_path ( f ) <EOL> atime = os . stat ( path ) . st_atime <EOL> if now - atime > timeout : <EOL> os . remove ( path ) <EOL> class DBStore ( Store ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db , table_name ) : <EOL> self . db = db <EOL> self . table = table_name <EOL> def __contains__ ( self , key ) : <EOL> data = self . db . select ( self . table , where = "<STR_LIT>" , vars = locals ( ) ) <EOL> return bool ( list ( data ) ) <EOL> def __getitem__ ( self , key ) : <EOL> now = datetime . datetime . now ( ) <EOL> try : <EOL> s = self . db . select ( self . table , where = "<STR_LIT>" , vars = locals ( ) ) [ <NUM_LIT:0> ] <EOL> self . db . update ( self . table , where = "<STR_LIT>" , atime = now , vars = locals ( ) ) <EOL> except IndexError : <EOL> raise KeyError <EOL> else : <EOL> return self . decode ( s . data ) <EOL> def __setitem__ ( self , key , value ) : <EOL> pickled = self . encode ( value ) <EOL> now = datetime . datetime . now ( ) <EOL> if key in self : <EOL> self . db . update ( self . table , where = "<STR_LIT>" , data = pickled , vars = locals ( ) ) <EOL> else : <EOL> self . db . insert ( self . table , False , session_id = key , data = pickled ) <EOL> def __delitem__ ( self , key ) : <EOL> self . db . delete ( self . table , where = "<STR_LIT>" , vars = locals ( ) ) <EOL> def cleanup ( self , timeout ) : <EOL> timeout = datetime . timedelta ( timeout / ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> ) ) <EOL> last_allowed_time = datetime . datetime . now ( ) - timeout <EOL> self . db . delete ( self . table , where = "<STR_LIT>" , vars = locals ( ) ) <EOL> class ShelfStore : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , shelf ) : <EOL> self . shelf = shelf <EOL> def __contains__ ( self , key ) : <EOL> return key in self . shelf <EOL> def __getitem__ ( self , key ) : <EOL> atime , v = self . shelf [ key ] <EOL> self [ key ] = v <EOL> return v <EOL> def __setitem__ ( self , key , value ) : <EOL> self . shelf [ key ] = time . time ( ) , value <EOL> def __delitem__ ( self , key ) : <EOL> try : <EOL> del self . shelf [ key ] <EOL> except KeyError : <EOL> pass <EOL> def cleanup ( self , timeout ) : <EOL> now = time . time ( ) <EOL> for k in self . shelf . keys ( ) : <EOL> atime , v = self . shelf [ k ] <EOL> if now - atime > timeout : <EOL> del self [ k ] <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import doctest <EOL> doctest . testmod ( ) </s>
