Migrate API authentication to client credentials

README.rst

@@ -16,8 +16,8 @@ Licence
 
 ::
 
+  Copyright © 2021, 2022 Jonathan Weth <dev@jonathanweth.de>
   Copyright © 2021 Margarete Grassl <grasslma@katharineum.de>
-  Copyright © 2021 Jonathan Weth <dev@jonathanweth.de>
 
   Licenced under the EUPL, version 1.2 or later
 

aleksis/apps/kort/api.py

@@ -4,36 +4,71 @@ from django.utils import timezone
 
 from celery.result import allow_join_result
 from celery.states import SUCCESS
-from oauth2_provider.contrib.rest_framework import TokenHasScope
-from rest_framework import generics, permissions, serializers
+from oauth2_provider.oauth2_backends import get_oauthlib_core
+from oauthlib.common import Request as OauthlibRequest
+from rest_framework import generics, serializers
+from rest_framework.authentication import BaseAuthentication
 from rest_framework.exceptions import APIException, ValidationError
 from rest_framework.permissions import BasePermission
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
 from aleksis.apps.kort.models import Card, CardPrinter, CardPrintJob
+from aleksis.core.util.auth_helpers import AppScopes
 
 admin.autodiscover()
 
 
-class CorrectPrinterPermission(BasePermission):
+class OAuth2ClientAuthentication(BaseAuthentication):
+    """OAuth 2 authentication backend using client credentials authentication."""
+
+    www_authenticate_realm = "api"
+
+    def authenticate(self, request):
+        """Authenticate the request with client credentials."""
+        oauthlib_core = get_oauthlib_core()
+        uri, http_method, body, headers = oauthlib_core._extract_params(request)
+        oauth_request = OauthlibRequest(uri, http_method, body, headers)
+
+        # Verify general authentication of the client
+        if not oauthlib_core.server.request_validator.authenticate_client(oauth_request):
+            # Client credentials were invalid
+            return None
+
+        request.auth = oauth_request
+
+        return (oauth_request.client.client_id, oauth_request)
+
+
+class ClientProtectedResourcePermission(BasePermission):
+    def has_object_permission(self, request, view, obj):
+        # Verify scopes of configured application
+        # The OAuth request was enriched with a reference to the Application when using the
+        #  validator above.
+        if not request.auth.client.allowed_scopes:
+            # If there are no allowed scopes, the client is not allowed to access this resource
+            return None
+
+        required_scopes = set(self.get_scopes(request, view, obj) or [])
+        allowed_scopes = set(AppScopes().get_available_scopes(request.auth.client) or [])
+        return required_scopes.issubset(allowed_scopes)
+
+    def get_scopes(self, request, view, obj):
+        return view.get_scopes()
+
+
+class CorrectPrinterPermission(ClientProtectedResourcePermission):
     """Check whether the OAuth2 application belongs to the printer."""
 
-    def has_object_permission(self, request, view, obj) -> bool:
-        token = request.auth
-        if token.application == obj.oauth2_application:
-            return True
-        return False
+    def get_scopes(self, request, view, obj):
+        return [obj.scope]
 
 
 class CorrectJobPrinterPermission(BasePermission):
     """Check whether the OAuth2 application belongs to the job's printer."""
 
-    def has_object_permission(self, request, view, obj) -> bool:
-        token = request.auth
-        if token.application == obj.printer.oauth2_application:
-            return True
-        return False
+    def get_scopes(self, request, view, obj):
+        return [obj.printer.scope]
 
 
 class CardPrinterSerializer(serializers.ModelSerializer):
@@ -91,21 +126,22 @@ class CardChipNumberSerializer(serializers.ModelSerializer):
 class CardPrinterDetails(generics.RetrieveAPIView):
     """Show details about the card printer."""
 
-    permission_classes = [permissions.IsAuthenticated, TokenHasScope, CorrectPrinterPermission]
-    required_scopes = ["card_printer"]
+    authentication_classes = [OAuth2ClientAuthentication]
+    permission_classes = [CorrectPrinterPermission]
     serializer_class = CardPrinterSerializer
     queryset = CardPrinter.objects.all()
 
     def get_object(self):
         token = self.request.auth
-        return token.application.card_printers.all().first()
+        return token.client.card_printers.all().first()
 
 
 class CardPrinterUpdateStatus(generics.UpdateAPIView):
     """Update the status of the card printer."""
 
-    permission_classes = [permissions.IsAuthenticated, TokenHasScope, CorrectPrinterPermission]
-    required_scopes = ["card_printer"]
+    authentication_classes = [OAuth2ClientAuthentication]
+
+    permission_classes = [CorrectPrinterPermission]
     serializer_class = CardPrinterStatusSerializer
     queryset = CardPrinter.objects.all()
 
@@ -120,8 +156,9 @@ class CardPrinterUpdateStatus(generics.UpdateAPIView):
 class GetNextPrintJob(APIView):
     """Get the next print job."""
 
-    permission_classes = [permissions.IsAuthenticated, TokenHasScope, CorrectPrinterPermission]
-    required_scopes = ["card_printer"]
+    authentication_classes = [OAuth2ClientAuthentication]
+
+    permission_classes = [CorrectPrinterPermission]
     serializer_class = CardPrinterSerializer
     queryset = CardPrinter.objects.all()
 
@@ -140,8 +177,9 @@ class GetNextPrintJob(APIView):
 class CardPrintJobUpdateStatusView(generics.UpdateAPIView):
     """Update the status of the card printer."""
 
-    permission_classes = [permissions.IsAuthenticated, TokenHasScope, CorrectJobPrinterPermission]
-    required_scopes = ["card_printer"]
+    authentication_classes = [OAuth2ClientAuthentication]
+
+    permission_classes = [CorrectJobPrinterPermission]
     serializer_class = CardPrintJobStatusSerializer
     queryset = CardPrintJob.objects.all()
 
@@ -149,8 +187,9 @@ class CardPrintJobUpdateStatusView(generics.UpdateAPIView):
 class CardPrintJobSetChipNumberView(generics.UpdateAPIView):
     """Update the status of the card printer."""
 
-    permission_classes = [permissions.IsAuthenticated, TokenHasScope, CorrectJobPrinterPermission]
-    required_scopes = ["card_printer"]
+    authentication_classes = [OAuth2ClientAuthentication]
+
+    permission_classes = [CorrectJobPrinterPermission]
     serializer_class = CardChipNumberSerializer
     queryset = CardPrintJob.objects.all()
 

aleksis/apps/kort/apps.py

+from django.apps import apps
+from django.db import models
+from django.db.models import functions
 from django.utils.translation import gettext as _
 
 from aleksis.core.util.apps import AppConfig
@@ -13,15 +16,29 @@ class DefaultConfig(AppConfig):
     }
     licence = "EUPL-1.2+"
     copyright_info = (
-        ([2021], "Margarete Grassl", "grasslma@katharineum.de"),
         (
-            [2021],
+            [2021, 2022],
             "Jonathan Weth",
             "dev@jonathanweth.de",
         ),
+        ([2021], "Margarete Grassl", "grasslma@katharineum.de"),
     )
 
     @classmethod
     def get_all_scopes(cls) -> dict[str, str]:
         """Return all OAuth scopes and their descriptions for this app."""
-        return {"card_printer": _("Access and manage printer status and print jobs")}
+        CardPrinter = apps.get_model("kort", "CardPrinter")
+        label_prefix = _("Access and manage printer status and print jobs")
+        scopes = dict(
+            CardPrinter.objects.annotate(
+                scope=functions.Concat(
+                    models.Value(f"{CardPrinter.SCOPE_PREFIX}_"),
+                    models.F("id"),
+                    output_field=models.CharField(),
+                ),
+                label=functions.Concat(models.Value(f"{label_prefix}: "), models.F("name")),
+            )
+            .values_list("scope", "label")
+            .distinct()
+        )
+        return scopes

aleksis/apps/kort/migrations/0015_migrate_scopes.py

+from django.db import migrations
+
+def _migrate_scopes(apps, schema_editor):
+    CardPrinter = apps.get_model("kort", "CardPrinter")
+    for printer in CardPrinter.objects.all():
+        application = printer.oauth2_application
+        application.allowed_scopes = [f"card_printer_{printer.id}"]
+        application.authorization_grant_type = "client-credentials"
+        application.save()
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('kort', '0014_auto_20220803_0025'),
+    ]
+
+    operations = [
+        migrations.RunPython(_migrate_scopes)
+    ]

aleksis/apps/kort/models.py

@@ -94,9 +94,10 @@ class CardPrinter(ExtensibleModel):
         if not self.oauth2_application:
             application = OAuthApplication(
                 client_type=OAuthApplication.CLIENT_CONFIDENTIAL,
-                authorization_grant_type=OAuthApplication.GRANT_AUTHORIZATION_CODE,
+                authorization_grant_type=OAuthApplication.GRANT_CLIENT_CREDENTIALS,
                 name=f"Card printer: {self.name}",
                 redirect_uris="urn:ietf:wg:oauth:2.0:oob",
+                allowed_scopes=[self.scope],
             )
             application.save()
             self.oauth2_application = application
@@ -171,6 +172,11 @@ class CardPrinter(ExtensibleModel):
             return jobs.first()
         return None
 
+    @property
+    def scope(self) -> str:
+        """Return OAuth2 scope name to access PDF file via API."""
+        return f"{self.SCOPE_PREFIX}_{self.id}"
+
     class Meta:
         verbose_name = _("Card printer")
         verbose_name_plural = _("Card printers")