<s> """<STR_LIT>""" <EOL> import bcrypt , re , urllib . parse , urllib . request , flask , json <EOL> import api <EOL> from api . common import check , validate , safe_fail <EOL> from api . common import WebException , InternalException <EOL> from api . annotations import log_action <EOL> from voluptuous import Required , Length , Schema <EOL> _check_email_format = lambda email : re . match ( r"<STR_LIT>" , email ) is not None <EOL> user_schema = Schema ( { <EOL> Required ( '<STR_LIT:email>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:5> , max = <NUM_LIT:50> ) ] ) , <EOL> ( "<STR_LIT>" , [ _check_email_format ] ) <EOL> ) , <EOL> Required ( '<STR_LIT>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:1> , max = <NUM_LIT:50> ) ] ) <EOL> ) , <EOL> Required ( '<STR_LIT>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:1> , max = <NUM_LIT:50> ) ] ) <EOL> ) , <EOL> Required ( '<STR_LIT>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:2> , max = <NUM_LIT:2> ) ] ) <EOL> ) , <EOL> Required ( '<STR_LIT:username>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:20> ) ] ) , <EOL> ( "<STR_LIT>" , [ <EOL> lambda name : safe_fail ( get_user , name = name ) is None ] ) <EOL> ) , <EOL> Required ( '<STR_LIT:password>' ) : <EOL> check ( ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:20> ) ] ) <EOL> ) , <EOL> Required ( '<STR_LIT>' ) : <EOL> check ( ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:20> ) ] ) <EOL> ) <EOL> } , extra = True ) <EOL> new_eligible_team_schema = Schema ( { <EOL> Required ( '<STR_LIT>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT> ) ] ) , <EOL> ( "<STR_LIT>" , [ <EOL> lambda name : safe_fail ( api . team . get_team , name = name ) is None ] ) <EOL> ) , <EOL> Required ( '<STR_LIT>' ) : <EOL> check ( ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:20> ) ] ) ) , <EOL> Required ( '<STR_LIT>' ) : <EOL> check ( ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:100> ) ] ) ) <EOL> } , extra = True ) <EOL> new_team_schema = Schema ( { <EOL> Required ( '<STR_LIT>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT> ) ] ) , <EOL> ( "<STR_LIT>" , [ <EOL> lambda name : safe_fail ( api . team . get_team , name = name ) is None ] ) <EOL> ) , <EOL> Required ( '<STR_LIT>' ) : <EOL> check ( ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:20> ) ] ) ) , <EOL> } , extra = True ) <EOL> existing_team_schema = Schema ( { <EOL> Required ( '<STR_LIT>' ) : check ( <EOL> ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:50> ) ] ) , <EOL> ( "<STR_LIT>" , [ <EOL> lambda name : api . team . get_team ( name = name ) != None ] ) , <EOL> ( "<STR_LIT>" , [ <EOL> lambda name : len ( api . team . get_team_uids ( name = name , show_disabled = False ) ) < api . team . max_team_users <EOL> ] ) <EOL> ) , <EOL> Required ( '<STR_LIT>' ) : <EOL> check ( ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:50> ) ] ) ) <EOL> } , extra = True ) <EOL> teacher_schema = Schema ( { <EOL> Required ( '<STR_LIT>' ) : <EOL> check ( ( "<STR_LIT>" , [ str , Length ( min = <NUM_LIT:3> , max = <NUM_LIT:100> ) ] ) ) <EOL> } , extra = True ) <EOL> def hash_password ( password ) : <EOL> """<STR_LIT>""" <EOL> return bcrypt . hashpw ( password , bcrypt . gensalt ( <NUM_LIT:8> ) ) <EOL> def get_team ( uid = None ) : <EOL> """<STR_LIT>""" <EOL> user = get_user ( uid = uid ) <EOL> return api . team . get_team ( tid = user [ "<STR_LIT>" ] ) <EOL> def get_user ( name = None , uid = None ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> match = { } <EOL> if uid is not None : <EOL> match . update ( { '<STR_LIT>' : uid } ) <EOL> elif name is not None : <EOL> match . update ( { '<STR_LIT:username>' : name } ) <EOL> elif api . auth . is_logged_in ( ) : <EOL> match . update ( { '<STR_LIT>' : api . auth . get_uid ( ) } ) <EOL> else : <EOL> raise InternalException ( "<STR_LIT>" ) <EOL> user = db . users . find_one ( match ) <EOL> if user is None : <EOL> raise InternalException ( "<STR_LIT>" ) <EOL> return user <EOL> def create_user ( username , firstname , lastname , email , password_hash , tid , teacher = False , <EOL> background = "<STR_LIT>" , country = "<STR_LIT>" , receive_ctf_emails = False ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> uid = api . common . token ( ) <EOL> if safe_fail ( get_user , name = username ) is not None : <EOL> raise InternalException ( "<STR_LIT>" ) <EOL> updated_team = db . teams . find_and_modify ( <EOL> query = { "<STR_LIT>" : tid , "<STR_LIT:size>" : { "<STR_LIT>" : api . team . max_team_users } } , <EOL> update = { "<STR_LIT>" : { "<STR_LIT:size>" : <NUM_LIT:1> } } , <EOL> new = True ) <EOL> if not updated_team : <EOL> raise InternalException ( "<STR_LIT>" ) <EOL> user = { <EOL> '<STR_LIT>' : uid , <EOL> '<STR_LIT>' : firstname , <EOL> '<STR_LIT>' : lastname , <EOL> '<STR_LIT:username>' : username , <EOL> '<STR_LIT:email>' : email , <EOL> '<STR_LIT>' : password_hash , <EOL> '<STR_LIT>' : tid , <EOL> '<STR_LIT>' : teacher , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : background , <EOL> '<STR_LIT>' : country , <EOL> '<STR_LIT>' : receive_ctf_emails <EOL> } <EOL> db . users . insert ( user ) <EOL> return uid <EOL> def get_all_users ( show_teachers = False ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> match = { } <EOL> projection = { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT:username>" : <NUM_LIT:1> , "<STR_LIT:email>" : <NUM_LIT:1> , "<STR_LIT>" : <NUM_LIT:1> } <EOL> if not show_teachers : <EOL> match . update ( { "<STR_LIT>" : False } ) <EOL> projection . update ( { "<STR_LIT>" : <NUM_LIT:1> } ) <EOL> return list ( db . users . find ( match , projection ) ) <EOL> def _validate_captcha ( data ) : <EOL> """<STR_LIT>""" <EOL> post_data = urllib . parse . urlencode ( { <EOL> "<STR_LIT>" : api . config . reCAPTCHA_private_key , <EOL> "<STR_LIT>" : data [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : flask . request . remote_addr <EOL> } ) . encode ( "<STR_LIT:utf-8>" ) <EOL> request = urllib . request . Request ( api . config . captcha_url , post_data , method = '<STR_LIT:POST>' ) <EOL> response = urllib . request . urlopen ( request ) . read ( ) . decode ( "<STR_LIT:utf-8>" ) <EOL> parsed_response = json . loads ( response ) <EOL> return parsed_response [ '<STR_LIT:success>' ] == True <EOL> @ log_action <EOL> def create_user_request ( params ) : <EOL> """<STR_LIT>""" <EOL> validate ( user_schema , params ) <EOL> if api . config . enable_captcha and not _validate_captcha ( params ) : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> if params . get ( "<STR_LIT>" , "<STR_LIT:false>" ) == "<STR_LIT:true>" : <EOL> if not api . config . enable_teachers : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> validate ( teacher_schema , params ) <EOL> tid = api . team . create_team ( { <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : params [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" + api . common . token ( ) <EOL> } ) <EOL> return create_user ( <EOL> params [ "<STR_LIT:username>" ] , <EOL> params [ "<STR_LIT>" ] , <EOL> params [ "<STR_LIT>" ] , <EOL> params [ "<STR_LIT:email>" ] , <EOL> hash_password ( params [ "<STR_LIT:password>" ] ) , <EOL> tid , <EOL> teacher = True , <EOL> background = params [ "<STR_LIT>" ] , <EOL> country = params [ "<STR_LIT>" ] , <EOL> receive_ctf_emails = params [ "<STR_LIT>" ] <EOL> ) <EOL> elif params . get ( "<STR_LIT>" , "<STR_LIT:false>" ) == "<STR_LIT:true>" : <EOL> eligible = True <EOL> if eligible : <EOL> validate ( new_eligible_team_schema , params ) <EOL> else : <EOL> validate ( new_team_schema , params ) <EOL> team_params = { <EOL> "<STR_LIT>" : params [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : params [ "<STR_LIT>" ] , <EOL> "<STR_LIT:password>" : params [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : eligible <EOL> } <EOL> tid = api . team . create_team ( team_params ) <EOL> if tid is None : <EOL> raise InternalException ( "<STR_LIT>" ) <EOL> team = api . team . get_team ( tid = tid ) <EOL> else : <EOL> validate ( existing_team_schema , params ) <EOL> team = api . team . get_team ( name = params [ "<STR_LIT>" ] ) <EOL> if team [ '<STR_LIT:password>' ] != params [ '<STR_LIT>' ] : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> uid = create_user ( <EOL> params [ "<STR_LIT:username>" ] , <EOL> params [ "<STR_LIT>" ] , <EOL> params [ "<STR_LIT>" ] , <EOL> params [ "<STR_LIT:email>" ] , <EOL> hash_password ( params [ "<STR_LIT:password>" ] ) , <EOL> team [ "<STR_LIT>" ] , <EOL> background = params [ "<STR_LIT>" ] , <EOL> country = params [ "<STR_LIT>" ] , <EOL> receive_ctf_emails = params [ "<STR_LIT>" ] <EOL> ) <EOL> if uid is None : <EOL> raise InternalException ( "<STR_LIT>" ) <EOL> api . team . determine_eligibility ( tid = team [ '<STR_LIT>' ] ) <EOL> return uid <EOL> def is_teacher ( uid = None ) : <EOL> """<STR_LIT>""" <EOL> user = get_user ( uid = uid ) <EOL> return user . get ( '<STR_LIT>' , False ) <EOL> def set_password_reset_token ( uid , token ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> db . users . update ( { '<STR_LIT>' : uid } , { '<STR_LIT>' : { '<STR_LIT>' : token } } ) <EOL> def delete_password_reset_token ( uid ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> db . users . update ( { '<STR_LIT>' : uid } , { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> def find_user_by_reset_token ( token ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> user = db . users . find_one ( { '<STR_LIT>' : token } ) <EOL> if user is None : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> return user <EOL> @ log_action <EOL> def update_password_request ( params , uid = None , check_current = False ) : <EOL> """<STR_LIT>""" <EOL> user = get_user ( uid = uid ) <EOL> if check_current and not api . auth . confirm_password ( params [ "<STR_LIT>" ] , user [ '<STR_LIT>' ] ) : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> if params [ "<STR_LIT>" ] != params [ "<STR_LIT>" ] : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> if len ( params [ "<STR_LIT>" ] ) == <NUM_LIT:0> : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> update_password ( user [ '<STR_LIT>' ] , params [ "<STR_LIT>" ] ) <EOL> def update_password ( uid , password ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> db . users . update ( { '<STR_LIT>' : uid } , { '<STR_LIT>' : { '<STR_LIT>' : hash_password ( password ) } } ) <EOL> def disable_account ( uid ) : <EOL> """<STR_LIT>""" <EOL> db = api . common . get_conn ( ) <EOL> result = db . users . update ( <EOL> { "<STR_LIT>" : uid , "<STR_LIT>" : False } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : True } } ) <EOL> tid = api . user . get_team ( uid = uid ) [ "<STR_LIT>" ] <EOL> if result [ "<STR_LIT:n>" ] == <NUM_LIT:1> : <EOL> db . teams . find_and_modify ( <EOL> query = { "<STR_LIT>" : tid , "<STR_LIT:size>" : { "<STR_LIT>" : <NUM_LIT:0> } } , <EOL> update = { "<STR_LIT>" : { "<STR_LIT:size>" : - <NUM_LIT:1> } } , <EOL> new = True ) <EOL> api . team . determine_eligibility ( tid ) <EOL> @ log_action <EOL> def disable_account_request ( params , uid = None , check_current = False ) : <EOL> """<STR_LIT>""" <EOL> user = get_user ( uid = uid ) <EOL> if check_current and not api . auth . confirm_password ( params [ "<STR_LIT>" ] , user [ '<STR_LIT>' ] ) : <EOL> raise WebException ( "<STR_LIT>" ) <EOL> disable_account ( user [ '<STR_LIT>' ] ) <EOL> api . auth . logout ( ) </s>
