<s> """<STR_LIT>""" <EOL> import base64 <EOL> import hashlib <EOL> import hmac <EOL> import sys <EOL> try : <EOL> from urlparse import urlsplit <EOL> except ImportError : <EOL> from urllib . parse import urlsplit <EOL> from contextlib import contextmanager <EOL> from datetime import datetime , timedelta <EOL> from flask import url_for , flash , current_app , request , session , render_template <EOL> from flask_login import login_user as _login_user , logout_user as _logout_user <EOL> from flask_mail import Message <EOL> from flask_principal import Identity , AnonymousIdentity , identity_changed <EOL> from itsdangerous import BadSignature , SignatureExpired <EOL> from werkzeug . local import LocalProxy <EOL> from . signals import user_registered , login_instructions_sent , reset_password_instructions_sent <EOL> _security = LocalProxy ( lambda : current_app . extensions [ '<STR_LIT>' ] ) <EOL> _datastore = LocalProxy ( lambda : _security . datastore ) <EOL> _pwd_context = LocalProxy ( lambda : _security . pwd_context ) <EOL> PY3 = sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> <EOL> if PY3 : <EOL> string_types = str , <EOL> text_type = str <EOL> else : <EOL> string_types = basestring , <EOL> text_type = unicode <EOL> def login_user ( user , remember = None ) : <EOL> """<STR_LIT>""" <EOL> if remember is None : <EOL> remember = config_value ( '<STR_LIT>' ) <EOL> if not _login_user ( user , remember ) : <EOL> return False <EOL> if _security . trackable : <EOL> if '<STR_LIT>' in request . headers : <EOL> remote_addr = request . headers . getlist ( "<STR_LIT>" ) [ <NUM_LIT:0> ] . rpartition ( '<STR_LIT:U+0020>' ) [ - <NUM_LIT:1> ] <EOL> else : <EOL> remote_addr = request . remote_addr or '<STR_LIT>' <EOL> old_current_login , new_current_login = user . current_login_at , datetime . utcnow ( ) <EOL> old_current_ip , new_current_ip = user . current_login_ip , remote_addr <EOL> user . last_login_at = old_current_login or new_current_login <EOL> user . current_login_at = new_current_login <EOL> user . last_login_ip = old_current_ip or new_current_ip <EOL> user . current_login_ip = new_current_ip <EOL> user . login_count = user . login_count + <NUM_LIT:1> if user . login_count else <NUM_LIT:1> <EOL> _datastore . put ( user ) <EOL> identity_changed . send ( current_app . _get_current_object ( ) , <EOL> identity = Identity ( user . id ) ) <EOL> return True <EOL> def logout_user ( ) : <EOL> """<STR_LIT>""" <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> session . pop ( key , None ) <EOL> identity_changed . send ( current_app . _get_current_object ( ) , <EOL> identity = AnonymousIdentity ( ) ) <EOL> _logout_user ( ) <EOL> def get_hmac ( password ) : <EOL> """<STR_LIT>""" <EOL> salt = _security . password_salt <EOL> if salt is None : <EOL> raise RuntimeError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % _security . password_hash ) <EOL> h = hmac . new ( encode_string ( salt ) , encode_string ( password ) , hashlib . sha512 ) <EOL> return base64 . b64encode ( h . digest ( ) ) <EOL> def verify_password ( password , password_hash ) : <EOL> """<STR_LIT>""" <EOL> if _security . password_hash != '<STR_LIT>' : <EOL> password = get_hmac ( password ) <EOL> return _pwd_context . verify ( password , password_hash ) <EOL> def verify_and_update_password ( password , user ) : <EOL> """<STR_LIT>""" <EOL> if _pwd_context . identify ( user . password ) != '<STR_LIT>' : <EOL> password = get_hmac ( password ) <EOL> verified , new_password = _pwd_context . verify_and_update ( password , user . password ) <EOL> if verified and new_password : <EOL> user . password = encrypt_password ( password ) <EOL> _datastore . put ( user ) <EOL> return verified <EOL> def encrypt_password ( password ) : <EOL> """<STR_LIT>""" <EOL> if _security . password_hash == '<STR_LIT>' : <EOL> return password <EOL> signed = get_hmac ( password ) . decode ( '<STR_LIT:ascii>' ) <EOL> return _pwd_context . encrypt ( signed ) <EOL> def encode_string ( string ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( string , text_type ) : <EOL> string = string . encode ( '<STR_LIT:utf-8>' ) <EOL> return string <EOL> def md5 ( data ) : <EOL> return hashlib . md5 ( encode_string ( data ) ) . hexdigest ( ) <EOL> def do_flash ( message , category = None ) : <EOL> """<STR_LIT>""" <EOL> if config_value ( '<STR_LIT>' ) : <EOL> flash ( message , category ) <EOL> def get_url ( endpoint_or_url ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return url_for ( endpoint_or_url ) <EOL> except : <EOL> return endpoint_or_url <EOL> def slash_url_suffix ( url , suffix ) : <EOL> """<STR_LIT>""" <EOL> return url . endswith ( '<STR_LIT:/>' ) and ( '<STR_LIT>' % suffix ) or ( '<STR_LIT>' % suffix ) <EOL> def get_security_endpoint_name ( endpoint ) : <EOL> return '<STR_LIT>' % ( _security . blueprint_name , endpoint ) <EOL> def url_for_security ( endpoint , ** values ) : <EOL> """<STR_LIT>""" <EOL> endpoint = get_security_endpoint_name ( endpoint ) <EOL> return url_for ( endpoint , ** values ) <EOL> def validate_redirect_url ( url ) : <EOL> if url is None or url . strip ( ) == '<STR_LIT>' : <EOL> return False <EOL> url_next = urlsplit ( url ) <EOL> url_base = urlsplit ( request . host_url ) <EOL> if ( url_next . netloc or url_next . scheme ) and url_next . netloc != url_base . netloc : <EOL> return False <EOL> return True <EOL> def get_post_action_redirect ( config_key , declared = None ) : <EOL> urls = [ <EOL> get_url ( request . args . get ( '<STR_LIT>' ) ) , <EOL> get_url ( request . form . get ( '<STR_LIT>' ) ) , <EOL> find_redirect ( config_key ) <EOL> ] <EOL> if declared : <EOL> urls . insert ( <NUM_LIT:0> , declared ) <EOL> for url in urls : <EOL> if validate_redirect_url ( url ) : <EOL> return url <EOL> def get_post_login_redirect ( declared = None ) : <EOL> return get_post_action_redirect ( '<STR_LIT>' , declared ) <EOL> def get_post_register_redirect ( declared = None ) : <EOL> return get_post_action_redirect ( '<STR_LIT>' , declared ) <EOL> def find_redirect ( key ) : <EOL> """<STR_LIT>""" <EOL> rv = ( get_url ( session . pop ( key . lower ( ) , None ) ) or <EOL> get_url ( current_app . config [ key . upper ( ) ] or None ) or '<STR_LIT:/>' ) <EOL> return rv <EOL> def get_config ( app ) : <EOL> """<STR_LIT>""" <EOL> items = app . config . items ( ) <EOL> prefix = '<STR_LIT>' <EOL> def strip_prefix ( tup ) : <EOL> return ( tup [ <NUM_LIT:0> ] . replace ( '<STR_LIT>' , '<STR_LIT>' ) , tup [ <NUM_LIT:1> ] ) <EOL> return dict ( [ strip_prefix ( i ) for i in items if i [ <NUM_LIT:0> ] . startswith ( prefix ) ] ) <EOL> def get_message ( key , ** kwargs ) : <EOL> rv = config_value ( '<STR_LIT>' + key ) <EOL> return rv [ <NUM_LIT:0> ] % kwargs , rv [ <NUM_LIT:1> ] <EOL> def config_value ( key , app = None , default = None ) : <EOL> """<STR_LIT>""" <EOL> app = app or current_app <EOL> return get_config ( app ) . get ( key . upper ( ) , default ) <EOL> def get_max_age ( key , app = None ) : <EOL> td = get_within_delta ( key + '<STR_LIT>' , app ) <EOL> return td . seconds + td . days * <NUM_LIT> * <NUM_LIT> <EOL> def get_within_delta ( key , app = None ) : <EOL> """<STR_LIT>""" <EOL> txt = config_value ( key , app = app ) <EOL> values = txt . split ( ) <EOL> return timedelta ( ** { values [ <NUM_LIT:1> ] : int ( values [ <NUM_LIT:0> ] ) } ) <EOL> def send_mail ( subject , recipient , template , ** context ) : <EOL> """<STR_LIT>""" <EOL> context . setdefault ( '<STR_LIT>' , _security ) <EOL> context . update ( _security . _run_ctx_processor ( '<STR_LIT>' ) ) <EOL> msg = Message ( subject , <EOL> sender = _security . email_sender , <EOL> recipients = [ recipient ] ) <EOL> ctx = ( '<STR_LIT>' , template ) <EOL> msg . body = render_template ( '<STR_LIT>' % ctx , ** context ) <EOL> msg . html = render_template ( '<STR_LIT>' % ctx , ** context ) <EOL> if _security . _send_mail_task : <EOL> _security . _send_mail_task ( msg ) <EOL> return <EOL> mail = current_app . extensions . get ( '<STR_LIT>' ) <EOL> mail . send ( msg ) <EOL> def get_token_status ( token , serializer , max_age = None , return_data = False ) : <EOL> """<STR_LIT>""" <EOL> serializer = getattr ( _security , serializer + '<STR_LIT>' ) <EOL> max_age = get_max_age ( max_age ) <EOL> user , data = None , None <EOL> expired , invalid = False , False <EOL> try : <EOL> data = serializer . loads ( token , max_age = max_age ) <EOL> except SignatureExpired : <EOL> d , data = serializer . loads_unsafe ( token ) <EOL> expired = True <EOL> except ( BadSignature , TypeError , ValueError ) : <EOL> invalid = True <EOL> if data : <EOL> user = _datastore . find_user ( id = data [ <NUM_LIT:0> ] ) <EOL> expired = expired and ( user is not None ) <EOL> if return_data : <EOL> return expired , invalid , user , data <EOL> else : <EOL> return expired , invalid , user <EOL> def get_identity_attributes ( app = None ) : <EOL> app = app or current_app <EOL> attrs = app . config [ '<STR_LIT>' ] <EOL> try : <EOL> attrs = [ f . strip ( ) for f in attrs . split ( '<STR_LIT:U+002C>' ) ] <EOL> except AttributeError : <EOL> pass <EOL> return attrs <EOL> @ contextmanager <EOL> def capture_passwordless_login_requests ( ) : <EOL> login_requests = [ ] <EOL> def _on ( app , ** data ) : <EOL> login_requests . append ( data ) <EOL> login_instructions_sent . connect ( _on ) <EOL> try : <EOL> yield login_requests <EOL> finally : <EOL> login_instructions_sent . disconnect ( _on ) <EOL> @ contextmanager <EOL> def capture_registrations ( ) : <EOL> """<STR_LIT>""" <EOL> registrations = [ ] <EOL> def _on ( app , ** data ) : <EOL> registrations . append ( data ) <EOL> user_registered . connect ( _on ) <EOL> try : <EOL> yield registrations <EOL> finally : <EOL> user_registered . disconnect ( _on ) <EOL> @ contextmanager <EOL> def capture_reset_password_requests ( reset_password_sent_at = None ) : <EOL> """<STR_LIT>""" <EOL> reset_requests = [ ] <EOL> def _on ( app , ** data ) : <EOL> reset_requests . append ( data ) <EOL> reset_password_instructions_sent . connect ( _on ) <EOL> try : <EOL> yield reset_requests <EOL> finally : <EOL> reset_password_instructions_sent . disconnect ( _on ) </s>
