Add a way to configure basic auth without storing passwords in plaintext in settings

fastapi_security/api.py

@@ -5,7 +5,12 @@
 from fastapi.security.http import HTTPAuthorizationCredentials
 from starlette.datastructures import Headers
 
-from .basic import BasicAuthValidator, IterableOfHTTPBasicCredentials
+from .basic import (
+    BasicAuthValidator,
+    BasicAuthWithDigestValidator,
+    IterableOfHTTPBasicCredentials,
+    IterableOfHTTPBasicCredentialsDigest,
+)
 from .entities import AuthMethod, User, UserAuth, UserInfo
 from .exceptions import AuthNotConfigured
 from .oauth2 import Oauth2JwtAccessTokenValidator
@@ -26,6 +31,7 @@ class FastAPISecurity:
 
     def __init__(self, *, user_permission_class: Type[UserPermission] = UserPermission):
         self.basic_auth = BasicAuthValidator()
+        self.basic_auth_with_digest = BasicAuthWithDigestValidator()
         self.oauth2_jwt = Oauth2JwtAccessTokenValidator()
         self.oidc_discovery = OpenIdConnectDiscovery()
         self._permission_overrides: Dict[str, List[str]] = {}
@@ -37,6 +43,9 @@ def __init__(self, *, user_permission_class: Type[UserPermission] = UserPermissi
     def init_basic_auth(self, basic_auth_credentials: IterableOfHTTPBasicCredentials):
         self.basic_auth.init(basic_auth_credentials)
 
+    def init_basic_auth_with_digest(self, salt: str, basic_auth_with_digest_credentials: IterableOfHTTPBasicCredentialsDigest):
+        self.basic_auth_with_digest.init(salt, basic_auth_with_digest_credentials)
+
     def init_oauth2_through_oidc(
         self, oidc_discovery_url: str, *, audiences: Iterable[str] = None
     ):
@@ -169,7 +178,7 @@ async def dependency(
         ) -> Optional[UserAuth]:
             oidc_configured = self.oidc_discovery.is_configured()
             oauth2_configured = self.oauth2_jwt.is_configured()
-            basic_auth_configured = self.basic_auth.is_configured()
+            basic_auth_configured = self.basic_auth.is_configured() or self.basic_auth_with_digest.is_configured()
 
             if not any([oidc_configured, oauth2_configured, basic_auth_configured]):
                 raise AuthNotConfigured()
@@ -185,8 +194,12 @@ async def dependency(
                     return self._maybe_override_permissions(
                         UserAuth.from_jwt_access_token(access_token)
                     )
-            elif http_credentials is not None and self.basic_auth.is_configured():
-                if self.basic_auth.validate(http_credentials):
+            elif http_credentials is not None:
+                is_valid = (
+                    self.basic_auth.validate(http_credentials) or
+                    self.basic_auth_with_digest.validate(http_credentials)
+                )
+                if is_valid:
                     return self._maybe_override_permissions(
                         UserAuth(
                             subject=http_credentials.username,

fastapi_security/basic.py

@@ -1,12 +1,29 @@
 import secrets
+from base64 import urlsafe_b64encode
 from typing import Dict, Iterable, List, Union
 
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.hashes import SHA512, Hash
+
 from fastapi.security.http import HTTPBasicCredentials
 
-__all__ = ("HTTPBasicCredentials",)
+__all__ = ("HTTPBasicCredentials", "generate_digest")
+
+
+from pydantic import BaseModel
+
+
+class HTTPBasicCredentialsDigest(BaseModel):
+    username: str
+    digest: str
+
 
 IterableOfHTTPBasicCredentials = Iterable[Union[HTTPBasicCredentials, Dict]]
 
+IterableOfHTTPBasicCredentialsDigest = Iterable[
+    Union[HTTPBasicCredentialsDigest, Dict]
+]
+
 
 class BasicAuthValidator:
     def __init__(self):
@@ -36,3 +53,47 @@ def _make_credentials(
             c if isinstance(c, HTTPBasicCredentials) else HTTPBasicCredentials(**c)
             for c in credentials
         ]
+
+
+class BasicAuthWithDigestValidator:
+    def __init__(self):
+        self._salt = None
+        self._credentials = []
+
+    def init(self, salt: str, credentials: IterableOfHTTPBasicCredentialsDigest):
+        self._salt = salt
+        self._credentials = self._make_credentials(credentials)
+
+    def is_configured(self) -> bool:
+        return self._salt and len(self._credentials) > 0
+
+    def validate(self, credentials: HTTPBasicCredentials) -> bool:
+        if not self.is_configured():
+            return False
+        return any(
+            (
+                secrets.compare_digest(c.username, credentials.username)
+                and c.digest == self.generate_digest(self._salt, credentials.password)
+            )
+            for c in self._credentials
+        )
+
+    def generate_digest(self, secret: str):
+        if not self._salt:
+            raise ValueError('BasicAuthWithDigestValidator: cannot generate digest, salt is empty')
+        return generate_digest(self._salt, secret)
+
+    def _make_credentials(
+        self, credentials: IterableOfHTTPBasicCredentialsDigest
+    ) -> List[HTTPBasicCredentialsDigest]:
+        return [
+            c if isinstance(c, HTTPBasicCredentialsDigest) else HTTPBasicCredentialsDigest(**c)
+            for c in credentials
+        ]
+
+
+def generate_digest(salt: str, secret: str):
+    hash_obj = Hash(algorithm=SHA512(), backend=default_backend())
+    hash_obj.update((salt + secret).encode('latin1'))
+    result = hash_obj.finalize()
+    return urlsafe_b64encode(result).decode('latin1')

fastapi_security/gendigest.py

@@ -0,0 +1,133 @@
+"""Generate digest for basic_auth_with_digest credentials.
+
+Takes an instance of FastAPISecurity that has basic_auth_with_digest configured
+(even if with empty credential list), prompts for password and generates a
+digest that can be appended to that instance's list of credentials.
+
+Example:
+
+$ python -m fastapi_security.gendigest fastapi_security.gendigest:obj
+Password:
+Confirm password:
+0jFS-cNapwQf_lpyULF7_hEelbl_zreNVHbxqKwKIFmPRQ09bYTEDQLrr_UEWZc9fdYFiU5F3il3rovJQ_UEpg==
+
+"""
+
+import argparse
+import importlib
+import sys
+import textwrap
+from getpass import getpass
+from types import ModuleType
+from typing import Union
+
+from fastapi_security import FastAPISecurity
+
+
+def _wrap_paragraphs(s):
+    paragraphs = s.strip().split('\n\n')
+    wrapped_paragraphs = [
+        '\n'.join(textwrap.wrap(paragraph)) for paragraph in paragraphs
+    ]
+    return '\n\n'.join(wrapped_paragraphs)
+
+
+def import_from_string(import_str: Union[ModuleType, str]) -> ModuleType:
+    """import_from_string: part of uvicorn codebase
+
+    Copyright © 2017-present, Encode OSS Ltd. All rights reserved.
+
+    Redistribution and use in source and binary forms, with or without
+    modification, are permitted provided that the following conditions are met:
+
+    Redistributions of source code must retain the above copyright notice, this
+    list of conditions and the following disclaimer.
+
+    Redistributions in binary form must reproduce the above copyright notice,
+    this list of conditions and the following disclaimer in the documentation
+    and/or other materials provided with the distribution.
+
+    Neither the name of the copyright holder nor the names of its contributors
+    may be used to endorse or promote products derived from this software
+    without specific prior written permission.
+
+    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+    AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+    IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+    ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+    LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+    CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+    SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+    INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+    CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+    ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+    POSSIBILITY OF SUCH DAMAGE.
+    """
+    if not isinstance(import_str, str):
+        return import_str
+
+    module_str, _, attrs_str = import_str.partition(":")
+    if not module_str or not attrs_str:
+        message = (
+            'Import string "{import_str}" must be in format "<module>:<attribute>".'
+        )
+        raise ValueError(message.format(import_str=import_str))
+
+    try:
+        module = importlib.import_module(module_str)
+    except ImportError as exc:
+        if exc.name != module_str:
+            raise exc from None
+        message = 'Could not import module "{module_str}".'
+        raise ValueError(message.format(module_str=module_str))
+
+    instance = module
+    try:
+        for attr_str in attrs_str.split("."):
+            instance = getattr(instance, attr_str)
+    except AttributeError:
+        message = 'Attribute "{attrs_str}" not found in module "{module_str}".'
+        raise ValueError(
+            message.format(attrs_str=attrs_str, module_str=module_str)
+        )
+
+    return instance
+
+
+parser = argparse.ArgumentParser(
+    description=_wrap_paragraphs(__doc__),
+    formatter_class=argparse.RawDescriptionHelpFormatter,
+)
+parser.add_argument('fastapi_security_obj')
+
+
+obj = FastAPISecurity()
+obj.init_basic_auth_with_digest('salt123', [])
+
+
+def main():
+    args = parser.parse_args()
+
+    fastapi_security_obj = import_from_string(args.fastapi_security_obj)
+    if callable(fastapi_security_obj):
+        instance = fastapi_security_obj()
+    elif isinstance(fastapi_security_obj, FastAPISecurity):
+        instance = fastapi_security_obj
+    else:
+        print("Cannot generate digest: ", args.fastapi_security_obj,
+              "must point to a FastAPISecurity object or a function returning one",
+              file=sys.error)
+        sys.exit(1)
+
+    password = getpass(prompt='Password: ')
+    password_confirmation = getpass(prompt='Confirm password: ')
+
+    if password != password_confirmation:
+        print("Cannot generate digest: passwords don't match", file=sys.stderr)
+        sys.exit(1)
+
+    print(instance.basic_auth_with_digest.generate_digest(password))
+
+
+if __name__ == '__main__':
+    main()

pyproject.toml

@@ -38,6 +38,7 @@ aiohttp = "^3"
 fastapi = "^0"
 pydantic = "^1"
 PyJWT = {version = "^2", extras = ["crypto"]}
+cryptography = "^3.4.7"
 
 [tool.poetry.dev-dependencies]
 aioresponses = "^0.7.2"

tests/integration/test_basic_auth.py

@@ -2,6 +2,7 @@
 
 from fastapi_security import FastAPISecurity, HTTPBasicCredentials, User
 from fastapi_security.basic import BasicAuthValidator
+from fastapi_security.basic import generate_digest
 
 from ..helpers.jwks import dummy_audience, dummy_jwks_uri
 
@@ -65,3 +66,42 @@ def get_products(user: User = Depends(security.authenticated_user_or_401)):
 
     resp = client.get("/", auth=("user", "pass"))
     assert resp.status_code == 200
+
+
+def test_that_basic_auth_with_digest_rejects_incorrect_credentials(app, client):
+    security = FastAPISecurity()
+
+    @app.get("/")
+    def get_products(user: User = Depends(security.authenticated_user_or_401)):
+        return []
+
+    pass_digest = generate_digest('salt123', 'pass')
+    credentials = [{"username": "user", "digest": pass_digest}]
+    security.init_basic_auth_with_digest('salt123', credentials)
+
+    resp = client.get("/")
+    assert resp.status_code == 401
+
+    resp = client.get("/", auth=("user", ""))
+    assert resp.status_code == 401
+
+    resp = client.get("/", auth=("", "pass"))
+    assert resp.status_code == 401
+
+    resp = client.get("/", auth=("abc", "123"))
+    assert resp.status_code == 401
+
+
+def test_that_basic_auth_with_digest_accepts_correct_credentials(app, client):
+    security = FastAPISecurity()
+
+    @app.get("/")
+    def get_products(user: User = Depends(security.authenticated_user_or_401)):
+        return []
+
+    pass_digest = generate_digest('salt123', 'pass')
+    credentials = [{"username": "user", "digest": pass_digest}]
+    security.init_basic_auth_with_digest('salt123', credentials)
+
+    resp = client.get("/", auth=("user", "pass"))
+    assert resp.status_code == 200