<s> import re <EOL> from trac . admin . api import AdminCommandError , IAdminCommandProvider , console_date_format , get_console_locale <EOL> from trac . core import Component , ExtensionPoint , TracError , implements <EOL> from trac . util import as_bool , as_int , hex_entropy , lazy <EOL> from trac . util . datefmt import get_datetime_format_hint , format_date , parse_date , time_now , to_datetime , to_timestamp <EOL> from trac . util . text import print_table <EOL> from trac . util . translation import _ <EOL> from trac . web . api import IRequestHandler , is_valid_default_handler <EOL> UPDATE_INTERVAL = <NUM_LIT> * <NUM_LIT> <EOL> PURGE_AGE = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> <EOL> COOKIE_KEY = '<STR_LIT>' <EOL> class DetachedSession ( dict ) : <EOL> def __init__ ( self , env , sid ) : <EOL> dict . __init__ ( self ) <EOL> self . env = env <EOL> self . sid = None <EOL> if sid : <EOL> self . get_session ( sid , authenticated = True ) <EOL> else : <EOL> self . authenticated = False <EOL> self . last_visit = <NUM_LIT:0> <EOL> self . _new = True <EOL> self . _old = { } <EOL> def __setitem__ ( self , key , value ) : <EOL> dict . __setitem__ ( self , key , unicode ( value ) ) <EOL> def as_bool ( self , key , default = None ) : <EOL> """<STR_LIT>""" <EOL> if key not in self : <EOL> return default <EOL> return as_bool ( self [ key ] , default ) <EOL> def as_int ( self , key , default = None , min = None , max = None ) : <EOL> """<STR_LIT>""" <EOL> if key not in self : <EOL> return default <EOL> return as_int ( self [ key ] , default , min , max ) <EOL> def set ( self , key , value , default = None ) : <EOL> """<STR_LIT>""" <EOL> value = unicode ( value ) <EOL> if default is not None : <EOL> default = unicode ( default ) <EOL> if value == default : <EOL> self . pop ( key , None ) <EOL> return <EOL> dict . __setitem__ ( self , key , value ) <EOL> def get_session ( self , sid , authenticated = False ) : <EOL> self . env . log . debug ( "<STR_LIT>" , sid ) <EOL> with self . env . db_query as db : <EOL> self . sid = sid <EOL> self . authenticated = authenticated <EOL> self . clear ( ) <EOL> for last_visit , in db ( """<STR_LIT>""" , ( sid , int ( authenticated ) ) ) : <EOL> self . _new = False <EOL> self . last_visit = int ( last_visit or <NUM_LIT:0> ) <EOL> self . update ( db ( """<STR_LIT>""" , ( sid , int ( authenticated ) ) ) ) <EOL> self . _old = self . copy ( ) <EOL> break <EOL> else : <EOL> self . last_visit = <NUM_LIT:0> <EOL> self . _new = True <EOL> self . _old = { } <EOL> def save ( self ) : <EOL> items = self . items ( ) <EOL> if not self . _old and not items : <EOL> return <EOL> authenticated = int ( self . authenticated ) <EOL> now = int ( time_now ( ) ) <EOL> session_saved = False <EOL> with self . env . db_transaction as db : <EOL> if self . _new : <EOL> self . last_visit = now <EOL> self . _new = False <EOL> try : <EOL> db ( """<STR_LIT>""" , ( self . sid , self . last_visit , authenticated ) ) <EOL> except self . env . db_exc . IntegrityError : <EOL> self . env . log . warning ( '<STR_LIT>' , self . sid ) <EOL> db . rollback ( ) <EOL> return <EOL> if self . _old != self : <EOL> if self . _old . get ( '<STR_LIT:name>' ) != self . get ( '<STR_LIT:name>' ) or self . _old . get ( '<STR_LIT:email>' ) != self . get ( '<STR_LIT:email>' ) : <EOL> self . env . invalidate_known_users_cache ( ) <EOL> if not items and not authenticated : <EOL> db ( "<STR_LIT>" , <EOL> ( self . sid , ) ) <EOL> db ( """<STR_LIT>""" , ( self . sid , authenticated ) ) <EOL> self . _old = dict ( self . items ( ) ) <EOL> try : <EOL> db . executemany ( """<STR_LIT>""" , [ ( self . sid , authenticated , k , v ) <EOL> for k , v in items ] ) <EOL> except self . env . db_exc . IntegrityError : <EOL> self . env . log . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' , self . sid ) <EOL> db . rollback ( ) <EOL> return <EOL> session_saved = True <EOL> if session_saved and now - self . last_visit > UPDATE_INTERVAL : <EOL> self . last_visit = now <EOL> mintime = now - PURGE_AGE <EOL> with self . env . db_transaction as db : <EOL> self . env . log . info ( "<STR_LIT>" , self . sid ) <EOL> db ( """<STR_LIT>""" , ( self . last_visit , self . sid , authenticated ) ) <EOL> self . env . log . debug ( '<STR_LIT>' ) <EOL> db ( """<STR_LIT>""" , ( mintime , ) ) <EOL> with self . env . db_transaction as db : <EOL> db ( """<STR_LIT>""" , ( mintime , ) ) <EOL> class Session ( DetachedSession ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , env , req ) : <EOL> super ( Session , self ) . __init__ ( env , None ) <EOL> self . req = req <EOL> if req . authname == '<STR_LIT>' : <EOL> if COOKIE_KEY not in req . incookie : <EOL> self . sid = hex_entropy ( <NUM_LIT> ) <EOL> self . bake_cookie ( ) <EOL> else : <EOL> sid = req . incookie [ COOKIE_KEY ] . value <EOL> self . get_session ( sid ) <EOL> else : <EOL> if COOKIE_KEY in req . incookie : <EOL> sid = req . incookie [ COOKIE_KEY ] . value <EOL> self . promote_session ( sid ) <EOL> self . get_session ( req . authname , authenticated = True ) <EOL> def bake_cookie ( self , expires = PURGE_AGE ) : <EOL> assert self . sid , '<STR_LIT>' <EOL> self . req . outcookie [ COOKIE_KEY ] = self . sid <EOL> self . req . outcookie [ COOKIE_KEY ] [ '<STR_LIT:path>' ] = self . req . base_path or '<STR_LIT:/>' <EOL> self . req . outcookie [ COOKIE_KEY ] [ '<STR_LIT>' ] = expires <EOL> if self . env . secure_cookies : <EOL> self . req . outcookie [ COOKIE_KEY ] [ '<STR_LIT>' ] = True <EOL> self . req . outcookie [ COOKIE_KEY ] [ '<STR_LIT>' ] = True <EOL> _valid_sid_re = re . compile ( r'<STR_LIT>' ) <EOL> def get_session ( self , sid , authenticated = False ) : <EOL> refresh_cookie = False <EOL> if not authenticated and not self . _valid_sid_re . match ( sid ) : <EOL> raise TracError ( _ ( "<STR_LIT>" ) ) <EOL> if self . sid and sid != self . sid : <EOL> refresh_cookie = True <EOL> super ( Session , self ) . get_session ( sid , authenticated ) <EOL> if self . last_visit and time_now ( ) - self . last_visit > UPDATE_INTERVAL : <EOL> refresh_cookie = True <EOL> if not authenticated and refresh_cookie : <EOL> self . bake_cookie ( ) <EOL> def change_sid ( self , new_sid ) : <EOL> assert self . req . authname == '<STR_LIT>' , '<STR_LIT>' <EOL> assert new_sid , '<STR_LIT>' <EOL> if new_sid == self . sid : <EOL> return <EOL> if not self . _valid_sid_re . match ( new_sid ) : <EOL> raise TracError ( _ ( "<STR_LIT>" ) , <EOL> _ ( "<STR_LIT>" ) ) <EOL> with self . env . db_transaction as db : <EOL> if db ( "<STR_LIT>" , ( new_sid , ) ) : <EOL> raise TracError ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> id = new_sid ) , <EOL> _ ( "<STR_LIT>" ) ) <EOL> self . env . log . debug ( "<STR_LIT>" , self . sid , <EOL> new_sid ) <EOL> db ( "<STR_LIT>" , <EOL> ( new_sid , self . sid ) ) <EOL> db ( """<STR_LIT>""" , ( new_sid , self . sid ) ) <EOL> self . sid = new_sid <EOL> self . bake_cookie ( ) <EOL> def promote_session ( self , sid ) : <EOL> """<STR_LIT>""" <EOL> assert self . req . authname != '<STR_LIT>' , "<STR_LIT>" <EOL> with self . env . db_transaction as db : <EOL> authenticated_flags = [ authenticated for authenticated , in db ( <EOL> "<STR_LIT>" , <EOL> ( sid , self . req . authname ) ) ] <EOL> if len ( authenticated_flags ) == <NUM_LIT:2> : <EOL> db ( "<STR_LIT>" , <EOL> ( sid , ) ) <EOL> db ( """<STR_LIT>""" , ( sid , ) ) <EOL> elif len ( authenticated_flags ) == <NUM_LIT:1> : <EOL> if not authenticated_flags [ <NUM_LIT:0> ] : <EOL> self . env . log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sid , self . req . authname ) <EOL> db ( """<STR_LIT>""" , ( self . req . authname , sid ) ) <EOL> db ( """<STR_LIT>""" , ( self . req . authname , sid ) ) <EOL> else : <EOL> try : <EOL> db ( """<STR_LIT>""" , ( self . req . authname , int ( time_now ( ) ) ) ) <EOL> except self . env . db_exc . IntegrityError : <EOL> self . env . log . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' , self . req . authname ) <EOL> db . rollback ( ) <EOL> self . _new = False <EOL> self . sid = sid <EOL> self . bake_cookie ( <NUM_LIT:0> ) <EOL> class SessionAdmin ( Component ) : <EOL> """<STR_LIT>""" <EOL> implements ( IAdminCommandProvider ) <EOL> request_handlers = ExtensionPoint ( IRequestHandler ) <EOL> def get_admin_commands ( self ) : <EOL> hints = { <EOL> '<STR_LIT>' : get_datetime_format_hint ( get_console_locale ( self . env ) ) , <EOL> '<STR_LIT>' : get_datetime_format_hint ( '<STR_LIT>' ) , <EOL> } <EOL> yield ( '<STR_LIT>' , '<STR_LIT>' , <EOL> """<STR_LIT>""" , <EOL> self . _complete_list , self . _do_list ) <EOL> yield ( '<STR_LIT>' , '<STR_LIT>' , <EOL> """<STR_LIT>""" , <EOL> None , self . _do_add ) <EOL> yield ( '<STR_LIT>' , '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> """<STR_LIT>""" , <EOL> self . _complete_set , self . _do_set ) <EOL> yield ( '<STR_LIT>' , '<STR_LIT>' , <EOL> """<STR_LIT>""" , <EOL> self . _complete_delete , self . _do_delete ) <EOL> yield ( '<STR_LIT>' , '<STR_LIT>' , <EOL> """<STR_LIT>""" % hints , <EOL> None , self . _do_purge ) <EOL> @ lazy <EOL> def _valid_default_handlers ( self ) : <EOL> return sorted ( handler . __class__ . __name__ <EOL> for handler in self . request_handlers <EOL> if is_valid_default_handler ( handler ) ) <EOL> def _split_sid ( self , sid ) : <EOL> if sid . endswith ( '<STR_LIT>' ) : <EOL> return sid [ : - <NUM_LIT:2> ] , <NUM_LIT:0> <EOL> elif sid . endswith ( '<STR_LIT>' ) : <EOL> return sid [ : - <NUM_LIT:2> ] , <NUM_LIT:1> <EOL> else : <EOL> return sid , <NUM_LIT:1> <EOL> def _get_sids ( self ) : <EOL> rows = self . env . db_query ( "<STR_LIT>" ) <EOL> return [ '<STR_LIT>' % ( sid , auth ) for sid , auth in rows ] <EOL> def _get_list ( self , sids ) : <EOL> all_anon = '<STR_LIT>' in sids or '<STR_LIT:*>' in sids <EOL> all_auth = '<STR_LIT>' in sids or '<STR_LIT:*>' in sids <EOL> sids = set ( self . _split_sid ( sid ) for sid in sids <EOL> if sid not in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:*>' ) ) <EOL> rows = self . env . db_query ( """<STR_LIT>""" ) <EOL> for sid , authenticated , last_visit , name , email , handler in rows : <EOL> if all_anon and not authenticated or all_auth and authenticated or ( sid , authenticated ) in sids : <EOL> yield ( sid , authenticated , <EOL> format_date ( to_datetime ( last_visit ) , <EOL> console_date_format ) , <EOL> name , email , handler ) <EOL> def _complete_list ( self , args ) : <EOL> all_sids = self . _get_sids ( ) + [ '<STR_LIT:*>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> return set ( all_sids ) - set ( args ) <EOL> def _complete_set ( self , args ) : <EOL> if len ( args ) == <NUM_LIT:1> : <EOL> return [ '<STR_LIT:name>' , '<STR_LIT:email>' ] <EOL> elif len ( args ) == <NUM_LIT:2> : <EOL> return self . _get_sids ( ) <EOL> def _complete_delete ( self , args ) : <EOL> all_sids = self . _get_sids ( ) + [ '<STR_LIT>' ] <EOL> return set ( all_sids ) - set ( args ) <EOL> def _do_list ( self , * sids ) : <EOL> if not sids : <EOL> sids = [ '<STR_LIT:*>' ] <EOL> headers = ( _ ( "<STR_LIT>" ) , _ ( "<STR_LIT>" ) , _ ( "<STR_LIT>" ) , _ ( "<STR_LIT:Name>" ) , <EOL> _ ( "<STR_LIT>" ) , _ ( "<STR_LIT>" ) ) <EOL> print_table ( self . _get_list ( sids ) , headers ) <EOL> def _do_add ( self , sid , name = None , email = None ) : <EOL> sid , authenticated = self . _split_sid ( sid ) <EOL> with self . env . db_transaction as db : <EOL> try : <EOL> db ( "<STR_LIT>" , <EOL> ( sid , authenticated , int ( time_now ( ) ) ) ) <EOL> except Exception : <EOL> raise AdminCommandError ( _ ( "<STR_LIT>" , <EOL> sid = sid ) ) <EOL> if name is not None : <EOL> db ( "<STR_LIT>" , <EOL> ( sid , authenticated , name ) ) <EOL> if email is not None : <EOL> db ( "<STR_LIT>" , <EOL> ( sid , authenticated , email ) ) <EOL> self . env . invalidate_known_users_cache ( ) <EOL> def _do_set ( self , attr , sid , val ) : <EOL> if attr not in ( '<STR_LIT:name>' , '<STR_LIT:email>' , '<STR_LIT>' ) : <EOL> raise AdminCommandError ( _ ( "<STR_LIT>" , <EOL> attr = attr ) ) <EOL> if attr == '<STR_LIT>' : <EOL> if val and val not in self . _valid_default_handlers : <EOL> raise AdminCommandError ( _ ( "<STR_LIT>" , <EOL> val = val ) ) <EOL> sid , authenticated = self . _split_sid ( sid ) <EOL> with self . env . db_transaction as db : <EOL> if not db ( """<STR_LIT>""" , <EOL> ( sid , authenticated ) ) : <EOL> raise AdminCommandError ( _ ( "<STR_LIT>" , <EOL> sid = sid ) ) <EOL> db ( """<STR_LIT>""" , ( sid , authenticated , attr ) ) <EOL> db ( "<STR_LIT>" , <EOL> ( sid , authenticated , attr , val ) ) <EOL> self . env . invalidate_known_users_cache ( ) <EOL> def _do_delete ( self , * sids ) : <EOL> with self . env . db_transaction as db : <EOL> for sid in sids : <EOL> sid , authenticated = self . _split_sid ( sid ) <EOL> if sid == '<STR_LIT>' : <EOL> db ( "<STR_LIT>" ) <EOL> db ( "<STR_LIT>" ) <EOL> else : <EOL> db ( """<STR_LIT>""" , ( sid , authenticated ) ) <EOL> db ( """<STR_LIT>""" , ( sid , authenticated ) ) <EOL> self . env . invalidate_known_users_cache ( ) <EOL> def _do_purge ( self , age ) : <EOL> when = parse_date ( age , hint = '<STR_LIT>' , <EOL> locale = get_console_locale ( self . env ) ) <EOL> with self . env . db_transaction as db : <EOL> ts = to_timestamp ( when ) <EOL> db ( """<STR_LIT>""" , ( ts , ) ) <EOL> db ( """<STR_LIT>""" ) </s>
