我们从Python开源项目中，提取了以下15个代码示例，用于说明如何使用itsdangerous.BadData()。
def try_refresh_token(self, session_id): morsel = context.cookies.get(self.refresh_token_key) if not morsel or morsel.value is None or not morsel.value.strip(): self.bad() return refresh_token_encoded = morsel.value # Decoding the refresh token try: refresh_principal = JwtRefreshToken.load(refresh_token_encoded) self.ok( self.create_principal(member_id=refresh_principal.id, session_id=session_id), setup_header=True ) except itsdangerous.SignatureExpired: self.bad() except itsdangerous.BadData: self.bad() raise HttpBadRequest()
def authenticate_request(self): if self.token_key not in context.environ: self.bad() return encoded_token = context.environ[self.token_key] if encoded_token is None or not encoded_token.strip(): self.bad() return try: self.ok(self.verify_token(encoded_token)) except itsdangerous.SignatureExpired as ex: # The token has expired. So we're trying to restore it using refresh-token. session_id = ex.payload.get('sessionId') if session_id: self.try_refresh_token(session_id) else: self.bad() raise HttpUnauthorized() except itsdangerous.BadData: # The token is Malformed self.bad() raise HttpBadRequest()
def confirm_account(): """Confirm account endpoint.""" key = request.args.get('key') if key is None: abort(403) try: timeout = current_app.config.get('ACCOUNT_LINK_EXPIRATION', 3600) userdict = signer.loads(key, max_age=timeout, salt='account-validation') except BadData: abort(403) # First check if the user exists user = user_repo.get_by_name(userdict['name']) if user is not None: return _update_user_with_valid_email(user, userdict['email_addr']) return _create_account(userdict)
def reset_password(): """ Reset password method. Returns a Jinja2 template. """ key = request.args.get('key') if key is None: abort(403) userdict = {} try: timeout = current_app.config.get('ACCOUNT_LINK_EXPIRATION', 3600) userdict = signer.loads(key, max_age=timeout, salt='password-reset') except BadData: abort(403) username = userdict.get('user') if not username or not userdict.get('password'): abort(403) user = user_repo.get_by_name(username) if user.passwd_hash != userdict.get('password'): abort(403) form = ChangePasswordForm(request.body) if form.validate_on_submit(): user.set_password(form.new_password.data) user_repo.update(user) flash(gettext('You reset your password successfully!'), 'success') return _sign_in_user(user) if request.method == 'POST' and not form.validate(): flash(gettext('Please correct the errors'), 'error') response = dict(template='/account/password_reset.html', form=form) return handle_content_type(response)
def _get_auth_status(self, authuser, authpassword): try: val = self.serializer.loads(authpassword, max_age=self.LOGIN_EXPIRATION) except itsdangerous.SignatureExpired: return dict(status="expired") except itsdangerous.BadData: # check if we got user/password direct authentication return self._validate(authuser, authpassword) else: if not isinstance(val, list) or len(val) != 2 or val[0] != authuser: threadlog.debug("mismatch credential for user %r", authuser) return dict(status="nouser") return dict(status="ok", groups=val[1])
def _decode_reset_otk(self, otk): reset_signer = itsdangerous.URLSafeTimedSerializer( self.config.reset_secret, 'password-recovery') try: # we allow 6 hours name, pwfrag = reset_signer.loads(otk, max_age=6*60*60) except itsdangerous.BadData: return None user = self.store.get_user(name) if pwfrag == user['password'][-4:]: return user return None
def get_payload_from_token(token): s = Serializer(current_app.config['SECRET_KEY']) cipher = AESCipher(current_app.config['SECRET_KEY'][:16]) try: data = json.loads(cipher.decrypt(s.loads(token))) return data except SignatureExpired, se: time_offset = (datetime.now()- se.date_signed).total_seconds() current_app.logger.error('token expired: %s. signature date %s. offset with current date = %s'%(se.message,str(se.date_signed),str(time_offset))) current_app.logger.error('current date %s, token date %s'%(str(datetime.now()), str(se.date_signed))) if -1<= time_offset < 0:#allow for 1 seconds out of sync machines current_app.logger.info('token time offset within grace period. allowing auth') return json.loads(cipher.decrypt(se.payload)) else: LogApiTokenExpired() # raise SignatureExpired(se) raise TokenExpired() # abort(419, message = 'Authentication expired.') except BadSignature, e: current_app.logger.error('bad signature in token') encoded_payload = e.payload if encoded_payload is not None: try: decoded_payload = s.load_payload(encoded_payload) payload= json.loads(cipher.decrypt(decoded_payload)) LogApiTokenInvalid(payload) except BadData: LogApiTokenInvalid(dict(error='bad data in token', token=token)) abort(401, message = 'bad signature in token')
def validate_csrf(data, secret_key=None, time_limit=None, token_key=None): """Check if the given data is a valid CSRF token. This compares the given signed token to the one stored in the session. :param data: The signed CSRF token to be checked. :param secret_key: Used to securely sign the token. Default is ``WTF_CSRF_SECRET_KEY`` or ``SECRET_KEY``. :param time_limit: Number of seconds that the token is valid. Default is ``WTF_CSRF_TIME_LIMIT`` or 3600 seconds (60 minutes). :param token_key: Key where token is stored in session for comparision. Default is ``WTF_CSRF_FIELD_NAME`` or ``'csrf_token'``. :raises ValidationError: Contains the reason that validation failed. .. versionchanged:: 0.14 Raises ``ValidationError`` with a specific error message rather than returning ``True`` or ``False``. """ secret_key = _get_config( secret_key, 'WTF_CSRF_SECRET_KEY', current_app.secret_key, message='A secret key is required to use CSRF.' ) field_name = _get_config( token_key, 'WTF_CSRF_FIELD_NAME', 'csrf_token', message='A field name is required to use CSRF.' ) time_limit = _get_config( time_limit, 'WTF_CSRF_TIME_LIMIT', 3600, required=False ) if not data: raise ValidationError('The CSRF token is missing.') if field_name not in session: raise ValidationError('The CSRF session token is missing.') s = URLSafeTimedSerializer(secret_key, salt='wtf-csrf-token') try: token = s.loads(data, max_age=time_limit) except SignatureExpired: raise ValidationError('The CSRF token has expired.') except BadData: raise ValidationError('The CSRF token is invalid.') if not safe_str_cmp(session[field_name], token): raise ValidationError('The CSRF tokens do not match.')
