💩(user-tokens) add back & front for Token auth

This provides:
 - a frontend to allow user to create/delete User Token
 - the authentication process to allow any API to be called when
   authenticating with a User Token.

Author: qbey

src/backend/core/api/viewsets.py

@@ -24,6 +24,7 @@
 import requests
 import rest_framework as drf
 from botocore.exceptions import ClientError
+from knox.auth import TokenAuthentication
 from lasuite.malware_detection import malware_detection
 from lasuite.oidc_resource_server.authentication import ResourceServerAuthentication
 from rest_framework import filters, status, viewsets
@@ -670,6 +671,7 @@ def trashbin(self, request, *args, **kwargs):
         authentication_classes=[
             authentication.ServerToServerAuthentication,
             ResourceServerAuthentication,
+            TokenAuthentication,
         ],
         detail=False,
         methods=["post"],

src/backend/core/tests/test_user_token_api.py

@@ -0,0 +1,131 @@
+"""
+Test user_token API endpoints in the impress core app.
+"""
+
+import pytest
+from knox.models import get_token_model
+from rest_framework.test import APIClient
+
+from core import factories, models
+
+pytestmark = pytest.mark.django_db
+AuthToken = get_token_model()
+
+def test_api_user_token_list_anonymous(client):
+    """Anonymous users should not be allowed to list user tokens."""
+    response = client.get("/api/v1.0/user-tokens/")
+    assert response.status_code == 403
+    assert response.json() == {
+        "detail": "Authentication credentials were not provided."
+    }
+
+
+def test_api_user_token_list_authenticated(client):
+    """
+    Authenticated users should be able to list their own tokens.
+    Tokens are identified by digest, and include created/expiry.
+    """
+    user = factories.UserFactory()
+    # Knox creates a token instance and a character string token key.
+    # The create method returns a tuple: (instance, token_key_string)
+    token_instance_1, _ = AuthToken.objects.create(user=user)
+    AuthToken.objects.create(user=user)  # Another token for the same user
+    AuthToken.objects.create(user=factories.UserFactory())  # Token for a different user
+
+    client.force_login(user)
+
+    response = client.get("/api/v1.0/user-tokens/")
+    assert response.status_code == 200
+    content = response.json()
+    assert len(content) == 2
+    
+    # Check that the response contains the digests of the tokens created for the user
+    response_token_digests = {item["digest"] for item in content}
+    assert token_instance_1.digest in response_token_digests
+    
+    # Ensure the token_key is not listed
+    for item in content:
+        assert "token_key" not in item
+        assert "digest" in item
+        assert "created" in item
+        assert "expiry" in item
+
+
+def test_api_user_token_create_anonymous(client):
+    """Anonymous users should not be allowed to create user tokens."""
+    # The create endpoint does not take any parameters as per TokenCreateSerializer
+    # (user is implicit, other fields are read_only)
+    response = client.post("/api/v1.0/user-tokens/", data={}) 
+    assert response.status_code == 403
+    assert response.json() == {
+        "detail": "Authentication credentials were not provided."
+    }
+
+
+def test_api_user_token_create_authenticated(client):
+    """
+    Authenticated users should be able to create a new token.
+    The token key should be returned in the response upon creation.
+    """
+    user = factories.UserFactory()
+
+    client.force_login(user)
+
+    # The create endpoint does not take any parameters as per TokenCreateSerializer
+    response = client.post("/api/v1.0/user-tokens/", data={})
+    assert response.status_code == 201
+    content = response.json()
+    
+    # Based on TokenCreateSerializer, these fields should be in the response
+    assert "token_key" in content 
+    assert "digest" in content
+    assert "created" in content
+    assert "expiry" in content
+    assert len(content["token_key"]) > 0 # Knox token key should be non-empty
+
+    # Verify the token was actually created in the database for the user
+    assert AuthToken.objects.filter(user=user, digest=content["digest"]).exists() 
+
+def test_api_user_token_destroy_anonymous(client):
+    """Anonymous users should not be allowed to delete user tokens."""
+    user = factories.UserFactory()
+    token_instance, _ = AuthToken.objects.create(user=user)
+    response = client.delete(f"/api/v1.0/user-tokens/{token_instance.digest}/")
+    assert response.status_code == 403
+    assert AuthToken.objects.filter(digest=token_instance.digest).exists()
+
+
+def test_api_user_token_destroy_authenticated_own_token(client):
+    """Authenticated users should be able to delete their own tokens."""
+    user = factories.UserFactory()
+    token_instance, _ = AuthToken.objects.create(user=user)
+
+    client.force_login(user)
+
+    response = client.delete(f"/api/v1.0/user-tokens/{token_instance.digest}/")
+    assert response.status_code == 204
+    assert not AuthToken.objects.filter(digest=token_instance.digest).exists()
+
+
+def test_api_user_token_destroy_authenticated_other_user_token(client):
+    """Authenticated users should not be able to delete other users' tokens."""
+    user = factories.UserFactory()
+    other_user = factories.UserFactory()
+    other_user_token_instance, _ = AuthToken.objects.create(user=other_user)
+
+    client.force_login(user) # Log in as 'user'
+
+    response = client.delete(f"/api/v1.0/user-tokens/{other_user_token_instance.digest}/")
+    # The default behavior for a non-found or non-permissioned item in DestroyModelMixin
+    # when the queryset is filtered (as in get_queryset) is often a 404.
+    assert response.status_code == 404 
+    assert AuthToken.objects.filter(digest=other_user_token_instance.digest).exists()
+
+
+def test_api_user_token_destroy_non_existent_token(client):
+    """Attempting to delete a non-existent token should result in a 404."""
+    user = factories.UserFactory()
+    client.force_login(user)
+
+    response = client.delete("/api/v1.0/user-tokens/nonexistentdigest/")
+    assert response.status_code == 404 

src/backend/core/urls.py

@@ -8,12 +8,18 @@
 from rest_framework.routers import DefaultRouter
 
 from core.api import viewsets
+from core.user_token import viewsets as user_token_viewsets
 
 # - Main endpoints
 router = DefaultRouter()
 router.register("templates", viewsets.TemplateViewSet, basename="templates")
 router.register("documents", viewsets.DocumentViewSet, basename="documents")
 router.register("users", viewsets.UserViewSet, basename="users")
+router.register(
+    "user-tokens",
+    user_token_viewsets.UserTokenViewset,
+    basename="user_tokens",
+)
 
 # - Routes nested under a document
 document_related_router = DefaultRouter()

src/backend/core/user_token/__init__.py

@@ -0,0 +1,27 @@
+from knox.models import get_token_model
+from rest_framework import serializers
+
+
+class TokenReadSerializer(serializers.ModelSerializer):
+    """Serialize token for list purpose."""
+
+    class Meta:
+        model = get_token_model()
+        fields = ["digest", "created", "expiry"]
+        read_only_fields = ["digest", "created", "expiry"]
+
+
+class TokenCreateSerializer(serializers.ModelSerializer):
+    """Serialize token for creation purpose."""
+
+    class Meta:
+        model = get_token_model()
+        fields = ["user", "digest", "token_key", "created", "expiry"]
+        read_only_fields = ["digest", "token_key", "created", "expiry"]
+        extra_kwargs = {"user": {"write_only": True}}
+
+    def create(self, validated_data):
+        """The default knox token create manager returns a tuple."""
+        instance, token = super().create(validated_data)
+        instance.token_key = token  # warning do not save this
+        return instance

src/backend/core/user_token/serializers.py

@@ -0,0 +1,50 @@
+"""API endpoints for user token management"""
+
+from knox.models import get_token_model
+from rest_framework import permissions, viewsets, mixins
+from rest_framework.authentication import SessionAuthentication
+
+from . import serializers
+
+
+class UserTokenViewset(
+    mixins.CreateModelMixin,
+    mixins.ListModelMixin,
+    mixins.DestroyModelMixin,
+    viewsets.GenericViewSet,
+):
+    """API ViewSet for user invitations to document.
+
+    This view access is restricted to the session ie from frontend.
+
+    GET /api/v1.0/user-token/
+        Return list of existing tokens.
+
+    POST /api/v1.0/user-token/
+        Return newly created token.
+
+    DELETE  /api/v1.0/user-token/<token_id>/
+        Delete targeted token.
+    """
+
+    authentication_classes = [SessionAuthentication]
+    pagination_class = None
+    permission_classes = [permissions.IsAuthenticated]
+    queryset = get_token_model().objects.all()
+    serializer_class = serializers.TokenReadSerializer
+
+    def get_queryset(self):
+        """Return the queryset restricted to the logged-in user."""
+        queryset = super().get_queryset()
+        queryset = queryset.filter(user_id=self.request.user.pk)
+        return queryset
+
+    def get_serializer_class(self):
+        if self.action == "create":
+            return serializers.TokenCreateSerializer
+        return super().get_serializer_class()
+
+    def create(self, request, *args, **kwargs):
+        """Enforce request data to use current user."""
+        request.data["user"] = self.request.user.pk
+        return super().create(request, *args, **kwargs)

src/backend/core/user_token/viewsets.py

@@ -10,6 +10,9 @@
 https://docs.djangoproject.com/en/3.1/ref/settings/
 """
 
+# pylint: disable=too-many-lines
+
+import datetime
 import os
 import tomllib
 from socket import gethostbyname, gethostname
@@ -303,6 +306,7 @@ class Base(Configuration):
         "django_filters",
         "dockerflow.django",
         "rest_framework",
+        "knox",
         "parler",
         "treebeard",
         "easy_thumbnails",
@@ -328,6 +332,7 @@ class Base(Configuration):
     REST_FRAMEWORK = {
         "DEFAULT_AUTHENTICATION_CLASSES": (
             "rest_framework.authentication.SessionAuthentication",
+            "knox.auth.TokenAuthentication",
             "lasuite.oidc_resource_server.authentication.ResourceServerAuthentication",
         ),
         "DEFAULT_PARSER_CLASSES": [
@@ -640,6 +645,18 @@ class Base(Configuration):
         [], environ_name="OIDC_RS_SCOPES", environ_prefix=None
     )
 
+    # User token (knox)
+    REST_KNOX = {
+        "SECURE_HASH_ALGORITHM": "hashlib.sha512",
+        "AUTH_TOKEN_CHARACTER_LENGTH": 64,
+        "TOKEN_TTL": datetime.timedelta(hours=24 * 7),
+        "TOKEN_LIMIT_PER_USER": None,
+        "AUTO_REFRESH": False,
+        "AUTO_REFRESH_MAX_TTL": None,
+        "MIN_REFRESH_INTERVAL": 60,
+        "AUTH_HEADER_PREFIX": "Token",
+    }
+
     # AI service
     AI_FEATURE_ENABLED = values.BooleanValue(
         default=False, environ_name="AI_FEATURE_ENABLED", environ_prefix=None

src/backend/impress/settings.py

@@ -36,6 +36,7 @@ dependencies = [
     "django-lasuite[all]==0.0.8",
     "django-parler==2.3",
     "django-redis==5.4.0",
+    "django-rest-knox==5.0.2",
     "django-storages[s3]==1.14.6",
     "django-timezone-field>=5.1",
     "django==5.1.9",

src/backend/pyproject.toml

@@ -1,3 +1,4 @@
+import { Button } from '@openfun/cunningham-react';
 import { useTranslation } from 'react-i18next';
 import { css } from 'styled-components';
 
@@ -59,6 +60,13 @@ export const Header = () => {
       ) : (
         <Box $align="center" $gap={spacingsTokens['sm']} $direction="row">
           <ButtonLogin />
+          <Button
+            onClick={() => (window.location.href = '/user-tokens')}
+            aria-label={t('API Tokens', 'API Tokens')}
+            color="primary-text"
+          >
+            {t('API Tokens', 'API Tokens')}
+          </Button>
           <LanguagePicker />
           <LaGaufre />
         </Box>

src/frontend/apps/impress/src/features/header/components/Header.tsx

@@ -0,0 +1,3 @@
+export * from './useListUserTokens';
+export * from './useCreateUserToken';
+export * from './useDeleteUserToken';

src/frontend/apps/impress/src/features/user-tokens/api/index.ts

@@ -0,0 +1,28 @@
+import { useMutation } from '@tanstack/react-query';
+
+import { APIError, errorCauses, fetchAPI } from '@/api';
+
+import { NewUserToken } from '../types';
+
+export const createUserToken = async (): Promise<NewUserToken> => {
+  const response = await fetchAPI(`user-tokens/`, {
+    method: 'POST',
+    // The backend test indicates no data is sent for creation, so body is an empty object
+    body: JSON.stringify({}),
+  });
+
+  if (!response.ok) {
+    throw new APIError(
+      'Failed to create user token',
+      await errorCauses(response),
+    );
+  }
+
+  return response.json() as Promise<NewUserToken>;
+};
+
+export function useCreateUserToken() {
+  return useMutation<NewUserToken, APIError>({
+    mutationFn: createUserToken,
+  });
+}

src/frontend/apps/impress/src/features/user-tokens/api/useCreateUserToken.ts

@@ -0,0 +1,30 @@
+import { useMutation } from '@tanstack/react-query';
+
+import { APIError, errorCauses, fetchAPI } from '@/api';
+
+export const deleteUserToken = async (digest: string): Promise<void> => {
+  const response = await fetchAPI(`user-tokens/${digest}/`, {
+    method: 'DELETE',
+  });
+
+  if (!response.ok && response.status !== 204) {
+    // 204 is a valid response for delete
+    throw new APIError(
+      'Failed to delete user token',
+      await errorCauses(response),
+    );
+  }
+  // For 204, there's no content, and for other successful deletions, we don't expect content.
+  // So, we don't try to parse JSON.
+  return Promise.resolve();
+};
+
+export type DeleteUserTokenParams = {
+  digest: string;
+};
+
+export function useDeleteUserToken() {
+  return useMutation<void, APIError, DeleteUserTokenParams>({
+    mutationFn: ({ digest }) => deleteUserToken(digest),
+  });
+}