codekingpro/portable-devtools
114k
1"""2 flask_security.utils3 ~~~~~~~~~~~~~~~~~~~~4 5 Flask-Security utils module6 7 :copyright: (c) 2012-2019 by Matt Wright.8 :copyright: (c) 2019-2024 by J. Christopher Wagner (jwag).9 :license: MIT, see LICENSE for more details.10"""11 12from __future__ import annotations13 14import abc15import base6416from datetime import datetime, timedelta, timezone17from functools import partial18import hashlib19import hmac20import time21import typing as t22from urllib.parse import parse_qsl, quote, urlsplit, urlunsplit, urlencode23import urllib.request24import urllib.error25import warnings26 27from flask import (28 after_this_request,29 current_app,30 flash,31 g,32 request,33 render_template,34 session,35 url_for,36)37from flask_login import login_user as _login_user38from flask_login import logout_user as _logout_user39from flask_login import current_user40from flask_login import COOKIE_NAME as REMEMBER_COOKIE_NAME41from flask_principal import AnonymousIdentity, Identity, identity_changed, Need42from flask_wtf import csrf, FlaskForm43from wtforms import ValidationError44from itsdangerous import BadSignature, SignatureExpired45from werkzeug.local import LocalProxy46from werkzeug.datastructures import MultiDict47 48from .quart_compat import best, get_quart_status49from .proxies import _security, _datastore, _pwd_context, _hashing_context50from .signals import user_authenticated51 52if t.TYPE_CHECKING: # pragma: no cover53 from flask import Flask, Response54 from flask.typing import ResponseValue55 from .datastore import User56 57localize_callback = LocalProxy(lambda: _security.i18n_domain.gettext)58 59FsPermNeed = partial(Need, "fsperm")60FsPermNeed.__doc__ = """A need with the method preset to `"fsperm"`."""61 62 63def _(translate):64 """Identity function to mark strings for translation."""65 return translate66 67 68def get_request_attr(name: str) -> t.Any:69 """Retrieve a request local attribute.70 71 Current public attributes are:72 73 **fs_authn_via**74 will be set to the authentication mechanism (session, token, basic)75 that the current request was authenticated with.76 77 Returns None if attribute doesn't exist.78 79 .. versionadded:: 4.0.080 .. versionchanged:: 4.1.581 Use 'g' rather than request_ctx stack which is going away post Flask 2.282 """83 return getattr(g, name, None)84 85 86def set_request_attr(name, value):87 return setattr(g, name, value)88 89 90"""91Most view functions that modify the DB will call ``after_this_request(view_commit)``92Quart compatibility needs an async version93"""94if get_quart_status(): # pragma: no cover95 96 async def view_commit(response=None):97 _datastore.commit()98 return response99 100else:101 102 def view_commit(response=None):103 _datastore.commit()104 return response105 106 107# From a miguel grinberg blog around dealing with 3.12.108# Our default SQLAlchemy Datetime is naive.109# Note that most code should call _security.datetime_factory()110def aware_utcnow():111 return datetime.now(timezone.utc)112 113 114def aware_utcfromtimestamp(timestamp):115 return datetime.fromtimestamp(timestamp, timezone.utc)116 117 118def naive_utcnow():119 return aware_utcnow().replace(tzinfo=None)120 121 122def naive_utcfromtimestamp(timestamp):123 return aware_utcfromtimestamp(timestamp).replace(tzinfo=None)124 125 126def find_csrf_field_name():127 """128 We need to clear it on logout (since that isn't being done by Flask-WTF).129 The field name is configurable withing Flask-WTF as well as being130 overridable.131 We take the field name from the login_form as set by the configuration.132 """133 from .forms import DummyForm134 135 form = DummyForm(formdata=None)136 if hasattr(form.meta, "csrf_field_name"):137 return form.meta.csrf_field_name138 return None139 140 141def is_user_authenticated(user: User | None) -> bool:142 """143 return True if user is authenticated.144 145 With Flask-Login <=0.6.x and Flask-Security <5.4 current_user was always146 set - for non-authenticated users it pointed to an AnonymousUser147 Flask-Login is experimenting (11/5/23) with a LOGIN_NO_ANONYMOUS which will set148 current_user to None and deprecate is_authenticated (current_user non None implies149 authenticated).150 We have a configuration variable ANONYMOUS_USER_DISABLED which if true will force151 current_user to None on unauthenticated as well152 """153 if config_value("ANONYMOUS_USER_DISABLED"):154 # Note that user often is current_user which is a proxy and isn't ever actually155 # 'None'156 return bool(user)157 return bool(user and user.is_authenticated)158 159 160def login_user(161 user: User,162 remember: bool | None = None,163 authn_via: list[str] | None = None,164) -> bool:165 """Perform the login routine.166 167 If :py:data:`SECURITY_TRACKABLE` is used, make sure you commit changes after this168 request (i.e. ``app.security.datastore.commit()``).169 170 :param user: The user to login171 :param remember: Flag specifying if the remember cookie should be set.172 If ``None`` use value of :py:data:`SECURITY_DEFAULT_REMEMBER_ME`173 :param authn_via: A list of strings denoting which mechanism(s) the user174 authenticated with.175 These should be one or more of ["password", "sms", "authenticator", "email"] or176 other 'auto-login' mechanisms.177 :return: True if user successfully logged in.178 """179 180 if remember is None:181 remember = config_value("DEFAULT_REMEMBER_ME")182 183 if not _login_user(user, remember): # pragma: no cover184 return False185 186 if _security.trackable:187 remote_addr = request.remote_addr or None # make sure it is None188 189 old_current_login, new_current_login = (190 user.current_login_at,191 _security.datetime_factory(),192 )193 old_current_ip, new_current_ip = user.current_login_ip, remote_addr194 195 user.last_login_at = old_current_login or new_current_login196 user.current_login_at = new_current_login197 user.last_login_ip = old_current_ip198 user.current_login_ip = new_current_ip199 user.login_count = user.login_count + 1 if user.login_count else 1200 201 _datastore.put(user)202 203 session["fs_cc"] = "set" # CSRF cookie204 session["fs_paa"] = time.time() # Primary authentication at - timestamp205 206 identity_changed.send(207 current_app._get_current_object(), # type: ignore[attr-defined]208 _async_wrapper=current_app.ensure_sync,209 identity=Identity(user.fs_uniquifier),210 )211 212 user_authenticated.send(213 current_app._get_current_object(), # type: ignore[attr-defined]214 _async_wrapper=current_app.ensure_sync,215 user=user,216 authn_via=authn_via,217 )218 return True219 220 221def logout_user() -> None:222 """Logs out the current user.223 224 This will also clean up the remember me cookie if it exists.225 226 This sends an ``identity_changed`` signal to note that the current227 identity is now the `AnonymousIdentity`228 """229 230 for key in (231 "identity.name",232 "identity.auth_type",233 "fs_paa",234 "fs_gexp",235 "fs_oauth_next",236 ):237 session.pop(key, None)238 239 # Clear csrf token between sessions.240 # Ideally this would be handled by Flask-WTF but...241 # We don't clear entire session since Flask-Login seems to like having it.242 csrf_field_name = find_csrf_field_name()243 if csrf_field_name:244 session.pop(csrf_field_name, None)245 # Flask-WTF 'caches' csrf_token - and only set the session if not already246 # in 'g'. Be sure to clear both. This affects at least /confirm247 g.pop(csrf_field_name, None)248 session["fs_cc"] = "clear"249 identity_changed.send(250 current_app._get_current_object(), # type: ignore251 _async_wrapper=current_app.ensure_sync,252 identity=AnonymousIdentity(),253 )254 _logout_user()255 256 257def check_and_update_authn_fresh(258 within: timedelta,259 grace: timedelta,260 method: str | None = None,261) -> bool:262 """Check if user authenticated within specified time and update grace period.263 264 :param within: A timedelta specifying the maximum time in the past that the caller265 authenticated that is still considered 'fresh'.266 :param grace: A timedelta that, if the current session is considered 'fresh'267 will set a grace period for which freshness won't be checked.268 The intent here is that the caller shouldn't get part-way though269 a set of operations and suddenly be required to authenticate again.270 :param method: Optional - if set and == "basic" then will always return True.271 (since basic-auth sends username/password on every request)272 273 If within.total_seconds() is negative, will always return True (always 'fresh').274 This effectively just disables this entire mechanism.275 276 If "fs_gexp" is in the session and the current timestamp is less than that,277 return True and extend grace time (i.e. set fs_gexp to current time + grace).278 279 If not within the grace period, and within.total_seconds() is 0,280 return False (not fresh).281 282 Be aware that for this to work, sessions and therefore session cookies283 must be functioning and being sent as part of the request. If the required284 state isn't in the session cookie then return False (not 'fresh').285 286 .. warning::287 Be sure the caller is already authenticated PRIOR to calling this method.288 289 .. versionadded:: 3.4.0290 291 .. versionchanged:: 4.0.0292 Added `method` parameter.293 """294 295 if method == "basic":296 return True297 298 if within.total_seconds() < 0:299 # this means 'always fresh'300 return True301 302 if "fs_paa" not in session:303 # No session, you can't play.304 return False305 306 now = naive_utcnow()307 new_exp = now + grace308 grace_ts = int(new_exp.timestamp())309 310 if fs_gexp := session.get("fs_gexp", None):311 if now.timestamp() < fs_gexp:312 # Within grace period - extend it, and we're good.313 session["fs_gexp"] = grace_ts314 return True315 316 # Special case 0 - return False always, but set grace period.317 if within.total_seconds() == 0:318 session["fs_gexp"] = grace_ts319 return False320 321 authn_time = naive_utcfromtimestamp(session["fs_paa"])322 # allow for some time drift where it's possible authn_time is in the future323 # but let's be cautious and not allow arbitrary future times324 delta = now - authn_time325 if within > delta > -within:326 session["fs_gexp"] = grace_ts327 return True328 return False329 330 331def get_hmac(password: str | bytes) -> bytes:332 """Returns a Base64 encoded HMAC+SHA512 of the password signed with333 the salt specified by :py:data:`SECURITY_PASSWORD_SALT`.334 335 :param password: The password to sign336 """337 if not (salt := config_value("PASSWORD_SALT")):338 raise RuntimeError(339 "The configuration value `SECURITY_PASSWORD_SALT` must "340 "not be None when the value of `SECURITY_PASSWORD_HASH` is "341 'set to "%s"' % config_value("PASSWORD_HASH")342 )343 344 h = hmac.new(encode_string(salt), encode_string(password), hashlib.sha512)345 return base64.b64encode(h.digest())346 347 348def verify_password(password: str | bytes, password_hash: str | bytes) -> bool:349 """Returns ``True`` if the password matches the supplied hash.350 351 :param password: A plaintext password to verify352 :param password_hash: The expected hash value of the password353 (usually from your database)354 355 .. note::356 Make sure that the password passed in has already been normalized.357 """358 if use_double_hash(password_hash):359 password = get_hmac(password)360 361 return _pwd_context.verify(password, password_hash)362 363 364def verify_and_update_password(password: str | bytes, user: User) -> bool:365 """Returns ``True`` if the password is valid for the specified user.366 367 Additionally, the hashed password in the database is updated if the368 hashing algorithm happens to have changed.369 370 N.B. you MUST call DB commit if you are using a session-based datastore371 (such as SqlAlchemy) since the user instance might have been altered372 (i.e. ``app.security.datastore.commit()``).373 This is usually handled in the view.374 375 :param password: A plaintext password to verify376 :param user: The user to verify against377 378 .. tip::379 This should not be called directly - rather use380 :meth:`.UserMixin.verify_and_update_password`381 382 """383 if use_double_hash(user.password):384 verified = _pwd_context.verify(get_hmac(password), user.password)385 else:386 # Try with original password.387 verified = _pwd_context.verify(password, user.password)388 389 if verified and _pwd_context.needs_update(user.password):390 user.password = hash_password(password)391 _datastore.put(user)392 return verified393 394 395def hash_password(password: str | bytes) -> str:396 """Hash the specified plaintext password.397 398 Unless the hash algorithm (as specified by399 :py:data:`SECURITY_PASSWORD_HASH`) is listed in400 the configuration variable :py:data:`SECURITY_PASSWORD_SINGLE_HASH`,401 perform a double hash - first create an HMAC from the plaintext password402 and the value of :py:data:`SECURITY_PASSWORD_SALT`,403 then use the configured hashing algorithm.404 This satisfies OWASP/ASVS section 2.4.5: 'provide additional405 iteration of a key derivation'.406 407 .. versionadded:: 2.0.2408 409 :param password: The plaintext password to hash410 """411 if use_double_hash():412 password = get_hmac(password).decode("ascii")413 414 # Passing in options as part of hash is deprecated in passlib 1.7415 # and new algorithms like argon2 don't even support it.416 return _pwd_context.hash(417 password,418 **config_value("PASSWORD_HASH_OPTIONS", default={}).get(419 config_value("PASSWORD_HASH"), {}420 ),421 )422 423 424def encode_string(string):425 """Encodes a string to bytes, if it isn't already.426 427 :param string: The string to encode"""428 429 if isinstance(string, str):430 string = string.encode("utf-8")431 return string432 433 434def hash_data(data):435 return _hashing_context.hash(encode_string(data))436 437 438def verify_hash(hashed_data, compare_data):439 return _hashing_context.verify(encode_string(compare_data), hashed_data)440 441 442def suppress_form_csrf():443 """444 Return meta contents if we should suppress form from attempting to validate CSRF.445 446 If app doesn't want CSRF for unauth endpoints then check if caller is authenticated447 or not (many endpoints can be called either way).448 """449 if config_value("CSRF_IGNORE_UNAUTH_ENDPOINTS") and not is_user_authenticated(450 current_user451 ):452 return {"csrf": False}453 return {}454 455 456def do_flash(message: str, category: str) -> None:457 """Flash a message depending on if the `FLASH_MESSAGES` configuration458 value is set.459 460 :param message: The flash message461 :param category: The flash message category462 """463 if config_value("FLASH_MESSAGES"):464 flash(message, category)465 466 467def parse_auth_token(auth_token: str) -> dict[str, t.Any]:468 """Parse an authentication token.469 This will raise an exception if not properly signed or expired470 """471 tdata = dict()472 473 # This can raise BadSignature or SignatureExpired exceptions from itsdangerous474 raw_data = _security.remember_token_serializer.loads(475 auth_token, max_age=config_value("TOKEN_MAX_AGE")476 )477 478 # Version 3.x generated tokens that map to data with 3 elements,479 # and fs_uniquifier was on last element.480 # Version 4.0.0 generates tokens that map to data with only 1 element,481 # which maps to fs_uniquifier.482 # Version 5 and up are already a dict (with a version #)483 if isinstance(raw_data, dict):484 # new format - starting at ver=5485 if not all(k in raw_data for k in ["ver", "uid", "exp", "sid"]):486 raise ValueError("Token missing keys")487 tdata = raw_data488 if ts := tdata.get("exp"):489 if ts < int(time.time()):490 raise SignatureExpired("token[exp] value expired")491 else:492 # old tokens that were lists493 if len(raw_data) == 1:494 # version 4495 tdata["ver"] = "4"496 tdata["uid"] = raw_data[0]497 else:498 # version 3499 tdata["ver"] = "3"500 tdata["uid"] = raw_data[2]501 502 return tdata503 504 505def get_url(endpoint_or_url: str, qparams: dict[str, str] | None = None) -> str:506 """Returns a URL if a valid endpoint is found. Otherwise, returns the507 provided value.508 509 .. warning::510 If an endpoint ISN'T provided, then it is assumed that the URL511 is external to Flask and if the spa configuration REDIRECT_HOST512 is set will redirect to that host. This could be an issue in513 development.514 515 :param endpoint_or_url: The endpoint name or URL to default to516 :param qparams: additional query params to add to end of url517 :return: URL518 """519 try:520 return transform_url(url_for(endpoint_or_url), qparams)521 except Exception:522 # This is an external URL (no endpoint defined in app)523 # For (mostly) testing - allow changing/adding the url - for example524 # add a different host:port for cases where the UI is running525 # separately.526 if config_value("REDIRECT_HOST"):527 url = transform_url(528 endpoint_or_url, qparams, netloc=config_value("REDIRECT_HOST")529 )530 else:531 url = transform_url(endpoint_or_url, qparams)532 533 return url534 535 536def slash_url_suffix(url, suffix):537 """Adds a slash either to the beginning or the end of a suffix538 (which is to be appended to a URL), depending on whether or not539 the URL ends with a slash."""540 return url.endswith("/") and f"{suffix}/" or f"/{suffix}"541 542 543def transform_url(544 url: str, qparams: dict[str, str] | None = None, **kwargs: str545) -> str:546 """Modify url547 548 :param url: url to transform (can be relative)549 :param qparams: additional query params to add to end of url550 :param kwargs: pieces of URL to modify - e.g. netloc=localhost:8000551 :return: Modified URL552 553 .. versionadded:: 3.2.0554 """555 link_parse = urlsplit(url)556 if qparams:557 current_query = dict(parse_qsl(link_parse.query))558 current_query.update(qparams)559 link_parse = link_parse._replace(query=urlencode(current_query))560 return urlunsplit(link_parse._replace(**kwargs))561 562 563def get_security_endpoint_name(endpoint):564 return f"{config_value('BLUEPRINT_NAME')}.{endpoint}"565 566 567def url_for_security(endpoint: str, **values: t.Any) -> str:568 """Return a URL for the security blueprint569 570 :param endpoint: the endpoint of the URL (name of the function)571 :param values: the variable arguments of the URL rule572 :param _external: if set to `True`, an absolute URL is generated. Server573 address can be changed via `SERVER_NAME` configuration variable which574 defaults to `localhost`.575 :param _anchor: if provided this is added as anchor to the URL.576 :param _method: if provided this explicitly specifies an HTTP method.577 """578 endpoint = get_security_endpoint_name(endpoint)579 # mypy is complaining about this - but I think it's wrong?580 return url_for(endpoint, **values) # type: ignore581 582 583def validate_redirect_url(url: str) -> bool:584 """Validate that the URL for redirect is relative.585 Allowing an absolute redirect is a security issue - a so-called open-redirect.586 Note that by default Werkzeug will always take this URL and make it relative587 when setting the Location header - but that behavior can be overridden.588 589 The complexity here is that urlsplit() does pretty well, but browsers even today590 May 2021 are very lenient in what they accept as URLs - for example:591 next=\\\\github.com592 next=%5C%5C%5Cgithub.com593 next=/////github.com594 next=%20\\\\github.com595 next=%20///github.com596 next=%20//github.com597 next=%19////github.com - i.e. browser will strip control chars598 next=%E2%80%8A///github.com - doesn't redirect! That is a unicode thin space.599 600 All will result in a null netloc and scheme from urlsplit - however many browsers601 will gladly strip off uninteresting characters and convert backslashes to forward602 slashes - and the cases above will actually cause a redirect to github.com603 Sigh.604 605 Some articles claim that a relative url has to start with a '/' - but that isn't606 strictly true. From: https://datatracker.ietf.org/doc/html/rfc3986#section-5607 a relative path can start with a "//", "/", a non-colon, or be empty. So it seems608 that all the above URLs are valid.609 By the time we get the URL, it has been unencoded - so we can't really determine610 if it is 'valid' since it appears that '/'s can appear in the URL if escaped.611 """612 if url is None or url.strip() == "":613 return False614 url_next = urlsplit(url)615 url_base = urlsplit(request.host_url)616 if (url_next.netloc or url_next.scheme) and url_next.netloc != url_base.netloc:617 base_domain = current_app.config.get("SERVER_NAME")618 if (619 config_value("REDIRECT_ALLOW_SUBDOMAINS")620 and base_domain621 and (622 url_next.netloc == base_domain623 or url_next.netloc.endswith(f".{base_domain}")624 )625 ):626 return True627 else:628 return False629 return True630 631 632def get_post_action_redirect(633 config_key: str, next_loc: FlaskForm | MultiDict | dict | None634) -> str:635 """636 There is a security angle here - the result of this method is637 sent to Flask::redirect() - and we need to be sure that it can't be638 interpreted as a user-input external URL - that would mean we would639 have an 'open-redirect' vulnerability.640 """641 rurl = propagate_next(find_redirect(config_key), next_loc)642 (scheme, netloc, path, query, fragment) = urlsplit(rurl)643 safe_url = urlunsplit((scheme, netloc, quote(path), query, fragment))644 return safe_url645 646 647def get_post_login_redirect() -> str:648 return get_post_action_redirect("SECURITY_POST_LOGIN_VIEW", request.form)649 650 651def get_post_register_redirect() -> str:652 return get_post_action_redirect("SECURITY_POST_REGISTER_VIEW", request.form)653 654 655def get_post_logout_redirect() -> str:656 return get_post_action_redirect("SECURITY_POST_LOGOUT_VIEW", request.form)657 658 659def get_post_verify_redirect() -> str:660 return get_post_action_redirect("SECURITY_POST_VERIFY_VIEW", request.form)661 662 663def find_redirect(key: str) -> str:664 """Returns the URL to redirect to.665 666 :param key: The application configuration key to search for667 """668 app_url = None669 if app_value := current_app.config[key.upper()]:670 app_url = get_url(app_value)671 rv = app_url or str(current_app.config.get("APPLICATION_ROOT", "/"))672 return rv673 674 675def propagate_next(fallback_url: str, form: FlaskForm | MultiDict | dict | None) -> str:676 """Compute appropriate redirect URL677 The application can add a 'next' query parameter or have 'next' as a form field.678 If either exist, make sure they are valid (not pointing to external location)679 If neither, return the fallback_url680 681 Can be passed either request.form682 (which is really a MultiDict OR a real form OR a dict with a 'next' key).683 """684 form_next = None685 if form and isinstance(form, FlaskForm):686 if hasattr(form, "next") and form.next.data:687 form_next = form.next.data688 elif form and form.get("next", None):689 form_next = str(form.get("next"))690 arg_next = request.args.get("next")691 692 urls = [693 get_url(form_next) if form_next else None,694 get_url(arg_next) if arg_next else None,695 fallback_url,696 ]697 for url in urls:698 if url and validate_redirect_url(url):699 return url700 raise ValueError("No valid redirect URL found - configuration error")701 702 703def simplify_url(base_url: str, redirect_url: str) -> str:704 """705 Reduces the scheme and host from the redirect_url so it can be passed706 as a relative URL in a query (e.g. next) param.707 For this method we aren't worrying about a valid url (e.g. if it points708 externally) - that will be handled by later requests.709 710 :param base_url: The URL to simplify 'against'.711 :param redirect_url: The URL to reduce.712 """713 b_url = urlsplit(base_url)714 r_url = urlsplit(redirect_url)715 716 if (not r_url.scheme or r_url.scheme == b_url.scheme) and (717 not r_url.netloc or r_url.netloc == b_url.netloc718 ):719 return urlunsplit(("", "", r_url.path, r_url.query, r_url.fragment))720 return redirect_url721 722 723def get_message(key: str, **kwargs: t.Any) -> tuple[str, str]:724 rv = config_value("MSG_" + key)725 return localize_callback(rv[0], **kwargs), rv[1]726 727 728def config_value(key, app=None, default=None, strict=True):729 """Get a Flask-Security configuration value.730 731 :param key: The configuration key without the prefix `SECURITY_`732 :param app: An optional specific application to inspect. Defaults to733 Flask's `current_app`734 :param default: An optional default value if the value is not set735 :param strict: if True, will raise ValueError if key doesn't exist736 """737 app = app or current_app738 key = f"SECURITY_{key.upper()}"739 # protect against spelling mistakes740 if strict and key not in app.config:741 raise ValueError(f"Key {key} doesn't exist")742 return app.config.get(key, default)743 744 745def get_max_age(key, app=None):746 td = get_within_delta(key + "_WITHIN", app)747 return td.seconds + td.days * 24 * 3600748 749 750def get_within_delta(key, app=None):751 """Get a timedelta object from the application configuration following752 the internal convention of::753 754 <Amount of Units> <Type of Units>755 756 Examples of valid config values::757 758 5 days759 10 minutes760 761 :param key: The config value key without the `SECURITY_` prefix762 :param app: Optional application to inspect. Defaults to Flask's763 `current_app`764 """765 txt = config_value(key, app=app)766 values = txt.split()767 return timedelta(**{values[1]: int(values[0])})768 769 770def send_mail(subject, recipient, template, **context):771 """Send an email.772 773 :param subject: Email subject774 :param recipient: Email recipient775 :param template: The name of the email template776 :param context: The context to render the template with777 778 This formats the email and passes it off to :class:`.MailUtil` to actually send the779 message.780 """781 782 context.setdefault("security", _security)783 context.update(_security._run_ctx_processor("mail"))784 785 body = None786 html = None787 template_path = f"security/email/{template}"788 if config_value("EMAIL_PLAINTEXT"):789 body = _security.render_template(f"{template_path}.txt", **context)790 if config_value("EMAIL_HTML"):791 html = _security.render_template(f"{template_path}.html", **context)792 793 subject = localize_callback(subject)794 795 sender = config_value("EMAIL_SENDER")796 if isinstance(sender, LocalProxy):797 sender = sender._get_current_object()798 799 _security._mail_util.send_mail(800 template,801 subject,802 recipient,803 sender,804 body,805 html,806 **context,807 )808 809 810def get_token_status(token, serializer, max_age=None, return_data=False):811 """Get the status of a token.812 813 :param token: The token to check814 :param serializer: The name of the serializer. Can be one of the815 following: ``confirm``, ``login``, ``reset``816 :param max_age: The name of the max age config option. Can be one of817 the following: ``CONFIRM_EMAIL``, ``LOGIN``,818 ``RESET_PASSWORD``819 820 .. deprecated:: 5.0.0821 """822 warnings.warn(823 "'get_token_status' is deprecated - use check_and_get_token_status instead",824 DeprecationWarning,825 stacklevel=2,826 )827 serializer = getattr(_security, serializer + "_serializer")828 max_age = get_max_age(max_age)829 user, data = None, None830 expired, invalid = False, False831 832 try:833 data = serializer.loads(token, max_age=max_age)834 except SignatureExpired:835 d, data = serializer.loads_unsafe(token)836 expired = True837 except (BadSignature, TypeError, ValueError):838 invalid = True839 840 if data:841 user = _datastore.find_user(fs_uniquifier=data[0])842 843 expired = expired and (user is not None)844 845 if return_data:846 return expired, invalid, user, data847 else:848 return expired, invalid, user849 850 851def check_and_get_token_status(852 token: str, serializer_name: str, within: timedelta853) -> tuple[bool, bool, t.Any]:854 """Get the status of a token and return data.855 856 :param token: The token to check857 :param serializer_name: The name of the serializer. Can be one of the858 following: ``confirm``, ``login``, ``reset``, ``us_setup``859 ``remember``, ``two_factor_validity``, ``wan``860 :param within: max age - passed as a timedelta861 862 :return: a tuple of (expired, invalid, data)863 864 .. versionadded:: 3.4.0865 """866 serializer = getattr(_security, serializer_name + "_serializer")867 max_age = within.total_seconds()868 data = None869 expired, invalid = False, False870 871 try:872 data = serializer.loads(token, max_age=max_age)873 except SignatureExpired:874 d, data = serializer.loads_unsafe(token)875 expired = True876 except (BadSignature, TypeError, ValueError):877 invalid = True878 879 return expired, invalid, data880 881 882def get_identity_attributes(app: Flask | None = None) -> list[str]:883 # Return list of keys of identity attributes884 # Is it possible to not have any?885 app = app or current_app886 iattrs = app.config["SECURITY_USER_IDENTITY_ATTRIBUTES"]887 if iattrs:888 return [[*f][0] for f in iattrs]889 return []890 891 892def get_identity_attribute(attr: str, app: Flask | None = None) -> dict[str, t.Any]:893 """Given an user_identity_attribute, return the defining dict.894 A bit annoying since USER_IDENTITY_ATTRIBUTES is a list of dict895 where each dict has just one key.896 """897 app = app or current_app898 iattrs = app.config["SECURITY_USER_IDENTITY_ATTRIBUTES"]899 if iattrs:900 details = [901 mapping[attr] for mapping in iattrs if list(mapping.keys())[0] == attr902 ]903 if details:904 return details[0]905 return {}906 907 908def lookup_identity(identity):909 """910 Lookup identity in DB.911 This loops through, in order, :py:data:`SECURITY_USER_IDENTITY_ATTRIBUTES`,912 and first calls the mapper function to validate/normalize.913 Then the db.find_user is called on the specified user model attribute.914 """915 for mapping in config_value("USER_IDENTITY_ATTRIBUTES"):916 attr = list(mapping.keys())[0]917 details = mapping[attr]918 idata = details["mapper"](identity)919 if idata:920 user = _datastore.find_user(921 case_insensitive=details.get("case_insensitive", False), **{attr: idata}922 )923 return user924 return None925 926 927def uia_phone_mapper(identity: str) -> str | None:928 """Used to match identity as a phone number. This is a simple proxy929 to :py:class:`PhoneUtil`930 931 See :py:data:`SECURITY_USER_IDENTITY_ATTRIBUTES`.932 933 .. versionadded:: 3.4.0934 """935 ph = _security._phone_util.get_canonical_form(identity)936 return ph937 938 939def uia_email_mapper(identity: str) -> str | None:940 """Used to match identity as an email.941 942 :return: Normalized email or None if not valid email.943 944 See :py:data:`SECURITY_USER_IDENTITY_ATTRIBUTES`.945 946 .. versionadded:: 3.4.0947 """948 949 try:950 return _security._mail_util.normalize(identity)951 except ValueError:952 return None953 954 955def uia_username_mapper(identity: str) -> str | None:956 """Used to match identity as a username. This is a simple proxy957 to :py:class:`UsernameUtil`958 959 See :py:data:`SECURITY_USER_IDENTITY_ATTRIBUTES`.960 961 .. versionadded:: 4.1.0962 """963 return _security._username_util.normalize(identity)964 965 966def use_double_hash(password_hash=None):967 """Return a bool indicating whether a password should be hashed twice."""968 # Default to plaintext for backward compatibility with969 # :py:data:`SECURITY_PASSWORD_SINGLE_HASH` = False970 single_hash = config_value("PASSWORD_SINGLE_HASH") or {"plaintext"}971 972 if password_hash is None:973 scheme = config_value("PASSWORD_HASH")974 else:975 scheme = _pwd_context.identify(password_hash)976 977 return not (single_hash is True or scheme in single_hash)978 979 980def csrf_cookie_handler(response: Response) -> Response:981 """Called at end of every request.982 Uses session to track state (set/clear)983 984 Ideally we just need to set this once - however by default985 Flask-WTF has a time-out on these tokens governed by *WTF_CSRF_TIME_LIMIT*.986 While we could set that to None - and OWASP implies this is fine - that might987 not be agreeable to everyone.988 So as a basic usability hack - we check if it is expired and re-generate so at least989 the user doesn't have to log out and back in (just refresh).990 We also support a *CSRF_COOKIE_REFRESH_EACH_REQUEST* analogous to Flask's991 *SESSION_REFRESH_EACH_REQUEST*992 993 It is of course removed on logout/session end.994 Other info on web suggests replacing on every POST and accepting up to 'age' ago.995 """996 csrf_cookie = config_value("CSRF_COOKIE")997 csrf_cookie_name = config_value("CSRF_COOKIE_NAME")998 if not csrf_cookie_name:999 return response1000 1001 op = session.get("fs_cc", None)1002 if not op:1003 remember_cookie_name = current_app.config.get(1004 "REMEMBER_COOKIE_NAME", REMEMBER_COOKIE_NAME1005 )1006 has_remember_cookie = (1007 remember_cookie_name in request.cookies1008 and session.get("remember") != "clear"1009 )1010 # Set cookie if successfully logged in with flask_login's remember cookie1011 if has_remember_cookie and is_user_authenticated(current_user):1012 op = "set"1013 else:1014 return response1015 1016 if op == "clear":1017 # Alas delete_cookie only accepts some of the keywords set_cookie does1018 allowed = ["path", "domain", "secure", "httponly", "samesite"]1019 args = {k: csrf_cookie.get(k) for k in allowed if k in csrf_cookie}1020 response.delete_cookie(csrf_cookie_name, **args)1021 session.pop("fs_cc")1022 return response1023 1024 # Send a cookie if any of:1025 # 1) CSRF_COOKIE_REFRESH_EACH_REQUEST is true1026 # 2) fs_cc == "set" - this is on first login1027 # 3) existing cookie has expired1028 send = False1029 if op == "set":1030 send = True1031 session["fs_cc"] = "sent"1032 elif config_value("CSRF_COOKIE_REFRESH_EACH_REQUEST"):1033 send = True1034 elif current_app.config["WTF_CSRF_TIME_LIMIT"]:1035 current_cookie = request.cookies.get(csrf_cookie_name, None)1036 if current_cookie:1037 # Lets make sure it isn't expired if app doesn't set TIME_LIMIT to None.1038 try:1039 csrf.validate_csrf(current_cookie)1040 except ValidationError:1041 send = True1042 1043 if send:1044 response.set_cookie(csrf_cookie_name, value=csrf.generate_csrf(), **csrf_cookie)1045 return response1046 1047 1048def base_render_json(1049 form: FlaskForm,1050 include_user: bool = True,1051 include_auth_token: bool = False,1052 additional: dict[str, t.Any] | None = None,1053 error_status_code: int = 400,1054) -> ResponseValue:1055 """1056 This method is called by all views that return JSON responses.1057 This fills in the response and then calls :meth:`.Security.render_json`1058 which can be overridden by the app.1059 """1060 user = getattr(form, "user", None)1061 if form.errors:1062 code = error_status_code1063 # wtforms 3.0 introduces form-level errors - these show up as part of the1064 # errors dict with a key of 'None'1065 payload = json_error_response(field_errors=form.errors)1066 else:1067 code = 2001068 payload = dict()1069 if user:1070 # This allows anonymous GETs via JSON1071 if include_user:1072 payload["user"] = user.get_security_payload()1073 1074 if include_auth_token:1075 # view willing to return auth_token - check behavior config1076 if (1077 config_value("BACKWARDS_COMPAT_AUTH_TOKEN")1078 or "include_auth_token" in request.args1079 ):1080 try:1081 token = user.get_auth_token()1082 except ValueError:1083 # application has fs_token_uniquifier attribute but it1084 # hasn't been initialized. Since we are in a request context1085 # we can do that here.1086 _datastore.set_token_uniquifier(user)1087 after_this_request(view_commit)1088 token = user.get_auth_token()1089 payload["user"]["authentication_token"] = token1090 1091 # Return csrf_token on each JSON response - just as every form1092 # has it rendered.1093 payload["csrf_token"] = csrf.generate_csrf()1094 if additional:1095 payload.update(additional)1096 1097 return _security._render_json(payload, code, None, user)1098 1099 1100def simple_render_json(1101 additional: dict[str, t.Any] | None = None,1102) -> ResponseValue:1103 payload = dict(csrf_token=csrf.generate_csrf())1104 if additional:1105 payload.update(additional)1106 return _security._render_json(payload, 200, None, None)1107 1108 1109def default_want_json(req):1110 """Return True if response should be in json1111 N.B. do not call this directly - use security._want_json()1112 1113 :param req: Flask/Werkzeug Request1114 """1115 if req.is_json:1116 return True1117 # TODO should this handle json sub-types?1118 accept_mimetypes = req.accept_mimetypes1119 if not hasattr(req.accept_mimetypes, "best"): # pragma: no cover1120 # Alright. we don't have the best property, lets add it ourselves.1121 # This is for quart compatibility1122 accept_mimetypes.best = best1123 if accept_mimetypes.best == "application/json":1124 return True1125 return False1126 1127 1128def json_error_response(1129 errors: str | list | None = None,1130 field_errors: dict[str | None, list] | None = None,1131) -> dict[str, t.Any]:1132 """Helper to create an error response.1133 1134 The "errors" key holds a simple list of errors - which is made up of any passed1135 errors (either a string or list) as well as the (localized) error msgs from the1136 passed in field_errors.1137 1138 The "field_errors" key which is exactly what is returned from WTForms - namely1139 a dict of field-name: msg. For form-level errors (WTForms 3.0) the 'field-name' is1140 None - which alas means it isn't sortable and Flask's default JSONProvider1141 sorts keys - so we change that to '__all__' which is what django uses1142 apparently and was suggested as part of WTForms 3.0.1143 """1144 response_json: dict[str, list | dict[str, list]] = dict()1145 plain_errors = []1146 if errors:1147 if isinstance(errors, str):1148 plain_errors = [errors]1149 elif isinstance(errors, list):1150 plain_errors = errors1151 else:1152 raise TypeError("The errors argument should be either a str or list.")1153 if field_errors:1154 # This is default from WTForms - a dictionary of field name and list of errors1155 # we return that, as well as create a simple list of errors.1156 for e in field_errors.values():1157 plain_errors.extend(e)1158 if None in field_errors.keys():1159 # Ugh - wtforms decided to use None as a key - which json1160 # a) can't sort1161 # b) converts to "null"1162 # Issue filed - maybe they will change it1163 field_errors[""] = field_errors[None]1164 del field_errors[None]1165 response_json["field_errors"] = field_errors # type: ignore1166 response_json["errors"] = plain_errors1167 1168 return response_json1169 1170 1171def default_render_template(*args, **kwargs):1172 return render_template(*args, **kwargs)1173 1174 1175class SmsSenderBaseClass(metaclass=abc.ABCMeta):1176 @abc.abstractmethod1177 def send_sms(1178 self, from_number: str, to_number: str, msg: str1179 ) -> None: # pragma: no cover1180 """Abstract method for sending sms messages1181 1182 .. versionadded:: 3.2.01183 """1184 return1185 1186 1187class DummySmsSender(SmsSenderBaseClass):1188 def send_sms(self, from_number, to_number, msg): # pragma: no cover1189 """Do nothing."""1190 return1191 1192 1193class SmsSenderFactory:1194 senders: dict[str, t.Type[SmsSenderBaseClass]] = {"Dummy": DummySmsSender}1195 1196 @classmethod1197 def createSender(cls, name, *args, **kwargs):1198 """Initialize an SMS sender.1199 1200 :param name: Name as registered in SmsSenderFactory:senders (e.g. 'Twilio')