Allow authentication by JWT Bearer token

specifyweb/backend/accounts/access_token_utils.py

@@ -0,0 +1,211 @@
+import uuid
+
+from datetime import datetime, timezone, timedelta
+from typing import Literal, Generator, Any, cast
+from enum import StrEnum
+
+import jwt
+from django import http
+from django.contrib.auth import authenticate
+from django.conf import settings
+
+from specifyweb.specify.models import Collection, Specifyuser
+from specifyweb.backend.cache.redis.connect import RedisConnection, RedisString
+from specifyweb.backend.context.views import has_collection_access
+
+DEFAULT_AUTH_LIFESPAN_SECONDS = 1800
+
+# See https://pyjwt.readthedocs.io/en/latest/api.html#jwt.decode
+AUTH_JWT_DECODE_OPTIONS: jwt.types.Options = {
+    "require": ["iat", "exp", "jti"],
+    "verify_signature": True,
+    "verify_iat": True,
+    "verify_exp": True
+}
+
+AUTH_TOKEN_ALGORITHMS = ["HS256"]
+
+
+def generate_access_token(user, collection_id: int, expires_in: int = DEFAULT_AUTH_LIFESPAN_SECONDS):
+    jti = str(uuid.uuid4())
+
+    jwt_payload = {
+        "sub": str(user.id),
+        "username": user.name,
+        "collection": collection_id,
+        "jti": jti,
+        "iat": datetime.now(timezone.utc),
+        "exp": datetime.now(timezone.utc) + timedelta(seconds=expires_in)
+    }
+    token = jwt.encode(jwt_payload, settings.SECRET_KEY,
+                       algorithm=AUTH_TOKEN_ALGORITHMS[0])
+    return token
+
+
+def revoke_access_token(token: dict[str, Any]):
+    """
+    Accepts and revokes a decoded JWT Auth Token.
+    Specifically, stores the token in a "blacklist" in Redis for the remaining
+    time of the token.
+    The JWT Auth Middleware checks to see if the token is blacklisted during
+    authorization
+    """
+    required_claims = ("jti", "exp")
+    if not all(k in token for k in required_claims):
+        raise ValueError(f"Token missing required claims: {required_claims}")
+    jti = token["jti"]
+    expires_at = token["exp"]
+    current_time = int(datetime.now(timezone.utc).timestamp())
+    blacklist_ttl = expires_at - current_time
+    if blacklist_ttl < 1:
+        return
+    redis_string = RedisString(RedisConnection())
+    token_key = f"revoked:{jti}"
+    redis_string.set(token_key, "true", time_to_live=max(blacklist_ttl, 1))
+
+
+def get_token_from_request(request: http.HttpRequest) -> Literal[False] | None | dict:
+    """
+    Attempts to get an Access Token from a HTTP Request.
+    Currently, we only support using access tokens with an Authorization
+    Header.
+
+    Particularly, clients should provide an access token like the following as
+    a Header:
+        Authorization: Bearer <token>
+
+    For more information, see Section 2.1 of RFC 6750:
+        https://www.rfc-editor.org/info/rfc6750/#section-2.1
+    """
+    auth_header = request.headers.get("Authorization")
+    if auth_header is None or not auth_header.lower().startswith("bearer "):
+        return None
+
+    auth_fields = auth_header.strip().split(" ")
+    # If there is more than one space in the Authorization header, consider it
+    # invalid
+    if len(auth_fields) != 2:
+        return False
+
+    encoded_token = auth_fields[1]
+
+    token = decode_access_token(encoded_token)
+
+    if token is None:
+        return False
+
+    return token
+
+def decode_access_token(encoded_token: str) -> dict[str, Any] | None:
+    """
+    Attempts to decode an Access Token. Returns the token if it's valid and
+    None otherwise
+    """
+    try:
+        token = jwt.decode(
+            encoded_token,
+            settings.SECRET_KEY,
+            options=AUTH_JWT_DECODE_OPTIONS,
+            algorithms=AUTH_TOKEN_ALGORITHMS
+        )
+    except jwt.exceptions.InvalidTokenError:
+        return None
+    return token
+
+def token_is_revoked(token: dict):
+    token_identifier = token["jti"]
+    redis_string = RedisString(RedisConnection())
+    token_key = f"revoked:{token_identifier}"
+    return redis_string.exists(token_key)
+
+class ACCESS_TOKEN_GRANT_TYPE(StrEnum):
+    # The client provides the server with a username, password, and collection
+    # id
+    PASSWORD = "password"
+
+    @classmethod
+    def types(cls) -> Generator[str, None, None]:
+        return (item.value for item in cls)
+
+
+def handle_password_token_grant(request: http.HttpRequest) -> http.HttpResponse:
+    """
+    The client indiciated they want to authenticate using username and password.
+    They should have also provided a collection id to scope the token to.
+
+    See Section 4.3 of RFC 6749 for more information about this specific grant
+    type according to the OAuth 2.0 Framework:
+        https://www.rfc-editor.org/info/rfc6749/#section-4.3
+    """
+    username = request.POST.get("username")
+    password = request.POST.get("password")
+    raw_collection_id = request.POST.get("collectionid")
+    raw_expires_in = request.POST.get("expires_in", DEFAULT_AUTH_LIFESPAN_SECONDS)
+
+    if None in (username, password, raw_collection_id):
+        error_response = {
+            "error": "invalid_request",
+            "error_description": "username, password, and collectionid are required"
+        }
+        return http.JsonResponse(error_response, status=400)
+
+    try:
+        expires_in = int(raw_expires_in)
+        if expires_in <= 0:
+            error_response = {
+                "error": "invalid_request",
+                "error_description": "Token expiry time must be greater than 0"
+            }
+            return http.JsonResponse(error_response, status=400)
+    except ValueError:
+        error_response = {
+            "error": "invalid_request",
+            "error_description": "Expiry time could not be parsed as integer"
+        }
+        return http.JsonResponse(error_response, status=400)
+
+    try:
+        # We know that the raw_collection_id can not be NONE here as that was
+        # caught by the first check for username, password, and collectionid
+        # In the worst case scenario where the above check is changed and this
+        # can be None, this will raise TypeError
+        collection_id = int(cast(str, raw_collection_id))
+    except ValueError:
+        error_response = {
+            "error": "invalid_request",
+            "error_description": "Collection ID could not be parsed as integer"
+        }
+        return http.JsonResponse(error_response, status=400)
+
+    try:
+        collection = Collection.objects.get(id=collection_id)
+    except Collection.DoesNotExist:
+        error_response = {
+            "error": "invalid_request",
+            "error_description": f"Collection {collection_id} does not exist"
+        }
+        return http.JsonResponse(error_response, status=400)
+
+    untyped_user = authenticate(username=username, password=password)
+    # authenticate should always return a Specifyuser, as that's our only Auth
+    # User Model
+    user = cast(Specifyuser | None, untyped_user)
+
+    if user is None or not has_collection_access(collection.id, user.id):
+        error_response = {
+            "error": "invalid_client",
+            "error_description": "Credentials were incorrect or user does not have access to the Collection"
+        }
+        return http.JsonResponse(error_response, status=401)
+
+    token = generate_access_token(user, collection.id, expires_in=expires_in)
+
+    # TODO: issue refresh tokens that can be used to re-issue auth tokens
+    # Issue refresh tokens as HTTP-only cookies? Need to keep CSRF protection
+    # and replay attacks in mind. Maybe just store them in Redis...
+    response = {
+        "access_token": token,
+        "expires_in": expires_in,
+        "token_type": "Bearer"
+    }
+    return http.JsonResponse(response)

specifyweb/backend/accounts/middleware.py

@@ -0,0 +1,77 @@
+from django.utils.functional import SimpleLazyObject
+from django.core.exceptions import PermissionDenied
+from django.http import HttpResponse
+
+from specifyweb.specify.models import Collection, Specifyuser, Agent
+from specifyweb.specify.api.filter_by_col import filter_by_collection
+from specifyweb.backend.accounts.access_token_utils import get_token_from_request, token_is_revoked
+from specifyweb.backend.context.views import has_collection_access
+
+
+def get_agent(request):
+    try:
+        return filter_by_collection(Agent.objects, request.specify_collection) \
+            .select_related('specifyuser') \
+            .get(specifyuser=request.specify_user)
+    except Agent.DoesNotExist:
+        return None
+
+
+class JWTAuthMiddleware:
+    """
+    This middleware is in charge of handling requests that contain a JWT
+    Authorization Bearer Token following the Auth 2.0 framework.
+    See RFCs 6749 and 6750 for the Auth 2.0 spec:
+    https://www.rfc-editor.org/info/rfc6749/
+    https://www.rfc-editor.org/info/rfc6750/
+    """
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+        token = get_token_from_request(request)
+        # The request doesn't have an access token, so pass through
+        if token is None:
+            return self.get_response(request)
+
+        # There was an access token in the request, but it was invalid or
+        # revoked. Stop here and return a 401 Unauthorized
+        if token == False or token_is_revoked(token):
+            response = HttpResponse('Invalid access token', status=401)
+            # This header is formatted per OAuth 2.0, see section 5.2 of RFC
+            # 6749 and section 3 of RFC 6750
+            # See https://www.rfc-editor.org/info/rfc6749/#section-5.2 and
+            # https://www.rfc-editor.org/info/rfc6750/#section-3
+            response["WWW-Authenticate"] = 'Bearer error=\"invalid_token\", error_description=\"The access token is expired, revoked, or invalid\"'
+            return response
+
+        # REFACTOR: explicitly handle cases where these values can't be cast as
+        # an integer?
+        user_id = int(token["sub"])
+        collection_id = int(token["collection"])
+
+        # This shouldn't happen often in practice as this is also enforced when
+        # the tokens are generated, but just in case a token is forged or the
+        # user's collection access was revoked since the token was generated,
+        # this prevents users from accessing Collections they shouldn't
+        if not has_collection_access(collection_id, user_id):
+            raise PermissionDenied()
+
+        request.specify_collection = SimpleLazyObject(
+            lambda: Collection.objects.get(id=collection_id))
+        lazy_user = SimpleLazyObject(
+            lambda: Specifyuser.objects.get(id=user_id))
+        request.specify_user = lazy_user
+        request.user = lazy_user
+        request.specify_user_agent = SimpleLazyObject(
+            lambda: get_agent(request))
+
+        # We can disable CSRF checks with users authenticated via JWT.
+        # This is ONLY because the end user must explicitly pass the auth token
+        # as a header, and is not stored within the session, cookies, etc.
+        # Essentially, with CSRF protection disabled for users authenticated
+        # via token, we have to be careful not to store any auth information in
+        # a stateful way within the session
+        # e.g., avoid calling django.contrib.auth.login
+        request._dont_enforce_csrf_checks = True
+        return self.get_response(request)