feat(api): setup login with tokens

See #318

config/api.py

@@ -2,6 +2,8 @@
 from ninja.security import django_auth
 
 from legadilo.reading.api import reading_api_router
+from legadilo.users.api import AuthBearer, users_api_router
 
-api = NinjaAPI(title="Legadilo API", auth=[django_auth], docs_url="/docs/")
+api = NinjaAPI(title="Legadilo API", auth=[django_auth, AuthBearer()], docs_url="/docs/")
 api.add_router("reading/", reading_api_router)
+api.add_router("users/", users_api_router)

config/settings.py

@@ -2,6 +2,7 @@
 
 import concurrent
 import warnings
+from datetime import timedelta
 from pathlib import Path
 
 import asgiref
@@ -620,3 +621,5 @@ def before_send_to_sentry(event, hint):
 RSS_FETCH_TIMEOUT = env.int("LEGADILO_RSS_FETCH_TIMEOUT", default=300)
 CONTACT_EMAIL = env.str("LEGADILO_CONTACT_EMAIL", default=None)
 TOKEN_LENGTH = 50
+JWT_ALGORITHM = "HS256"
+JWT_MAX_AGE = timedelta(hours=4)

legadilo/reading/api.py

@@ -27,7 +27,7 @@
     build_article_data_from_content,
     get_article_from_url,
 )
-from legadilo.users.user_types import AuthenticatedHttpRequest
+from legadilo.users.user_types import AuthenticateApiRequest
 from legadilo.utils.validators import ValidUrlValidator
 
 reading_api_router = Router(tags=["reading"])
@@ -59,12 +59,12 @@ def has_data(self) -> bool:
 
 @reading_api_router.get("/articles/", response=list[OutArticleSchema])
 @paginate
-def list_articles(request: AuthenticatedHttpRequest):
-    return Article.objects.get_queryset().for_user(request.user).default_order_by()
+def list_articles(request: AuthenticateApiRequest):
+    return Article.objects.get_queryset().for_user(request.auth).default_order_by()
 
 
 @reading_api_router.post("/articles/", response=OutArticleSchema)
-def create_article(request: AuthenticatedHttpRequest, article: InArticleSchema):
+def create_article(request: AuthenticateApiRequest, article: InArticleSchema):
     if article.has_data:
         article_data = build_article_data_from_content(
             url=article.link, title=article.title, content=article.content
@@ -74,10 +74,10 @@ def create_article(request: AuthenticatedHttpRequest, article: InArticleSchema):
 
     # Tags specified in article data are the raw tags used in feeds, they are not used to link an
     # article to tag objects.
-    tags = Tag.objects.get_or_create_from_list(request.user, article.tags)
+    tags = Tag.objects.get_or_create_from_list(request.auth, article.tags)
     article_data = article_data.model_copy(update={"tags": ()})
 
     articles = Article.objects.update_or_create_from_articles_list(
-        request.user, [article_data], tags, source_type=constants.ArticleSourceType.MANUAL
+        request.auth, [article_data], tags, source_type=constants.ArticleSourceType.MANUAL
     )
     return articles[0]

legadilo/users/api.py

@@ -0,0 +1,99 @@
+# Legadilo
+# Copyright (C) 2023-2024 by Legadilo contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from datetime import datetime
+
+import jwt
+from django.core.exceptions import BadRequest
+from django.http import HttpRequest
+from django.shortcuts import get_object_or_404
+from ninja import Router, Schema
+from ninja.security import HttpBearer
+from pydantic import BaseModel as BaseSchema
+from pydantic import ValidationError as PydanticValidationError
+
+from config import settings
+from legadilo.users.models import ApplicationToken
+from legadilo.utils.time_utils import utcnow
+
+from .models import User
+
+users_api_router = Router(tags=["auth"])
+
+
+class AuthBearer(HttpBearer):
+    def authenticate(self, request, token) -> User | None:
+        if not token:
+            return None
+
+        decoded_jwt = _decode_jwt(token)
+        return _get_user_from_jwt(decoded_jwt)
+
+
+class JWT(BaseSchema):
+    application_token: str
+    user_id: int
+    exp: datetime
+
+
+def _decode_jwt(token: str) -> JWT:
+    try:
+        decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
+        return JWT.model_validate(decoded_token)
+    except jwt.ExpiredSignatureError as e:
+        raise BadRequest("Expired JWT token") from e
+    except (jwt.PyJWTError, PydanticValidationError) as e:
+        raise BadRequest("Invalid JWT token") from e
+
+
+def _get_user_from_jwt(decoded_jwt: JWT) -> User | None:
+    try:
+        return User.objects.get(id=decoded_jwt.user_id)
+    except User.DoesNotExist:
+        return None
+
+
+class RefreshTokenPayload(Schema):
+    application_token: str
+
+
+class Token(Schema):
+    jwt: str
+
+
+@users_api_router.post("/refresh/", auth=None, response=Token)
+def refresh_token(request: HttpRequest, payload: RefreshTokenPayload) -> Token:
+    application_token = get_object_or_404(
+        ApplicationToken.objects.get_queryset().only_valid().defer(None),
+        token=payload.application_token,
+    )
+    application_token.last_used_at = utcnow()
+    application_token.save()
+    jwt = _create_jwt(application_token.user_id, application_token.token)
+
+    return Token(jwt=jwt)
+
+
+def _create_jwt(user_id: int, application_token: str) -> str:
+    return jwt.encode(
+        {
+            "application_token": application_token,
+            "user_id": user_id,
+            "exp": utcnow() + settings.JWT_MAX_AGE,
+        },
+        settings.SECRET_KEY,
+        algorithm=settings.JWT_ALGORITHM,
+    )

legadilo/users/models/application_token.py

@@ -24,6 +24,7 @@
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 
+from ...utils.time_utils import utcnow
 from .user import User
 
 if TYPE_CHECKING:
@@ -32,9 +33,18 @@
     TypedModelMeta = object
 
 
+class ApplicationTokenQuerySet(models.QuerySet["ApplicationToken"]):
+    def only_valid(self):
+        return self.filter(models.Q(validity_end=None) | models.Q(validity_end__lt=utcnow()))
+
+
 class ApplicationTokenManager(models.Manager["ApplicationToken"]):
+    _hints: dict
+
     def get_queryset(self):
-        return super().get_queryset().defer("token")
+        return ApplicationTokenQuerySet(model=self.model, using=self._db, hints=self._hints).defer(
+            "token"
+        )
 
     def create_new_token(
         self, user: User, title: str, validity_end: datetime | None = None

legadilo/users/user_types.py

@@ -28,3 +28,10 @@ class AuthenticatedHttpRequest(HttpRequest):
     @abstractmethod
     async def auser(self) -> User:
         pass
+
+
+class AuthenticateApiRequest(HttpRequest):
+    # In the API, we cannot use user because it's not defined when using auth tokens. We must rely
+    # on auth which will always contains the proper user object.
+    user: None  # type: ignore[assignment]
+    auth: User