feat: adds support for OAuth 2 client credentials auth in xapi

integrated_channels/xapi/migrations/0008_adds_auth_method_scope_and_url.py

@@ -0,0 +1,28 @@
+# Generated by Django 3.2.12 on 2023-09-01 07:26
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('xapi', '0007_auto_20220405_2311'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='xapilrsconfiguration',
+            name='auth_method',
+            field=models.CharField(choices=[('BASIC', 'HTTP Basic'), ('OAUTH2_CC', 'OAuth 2.0 Client Credentials')], default='BASIC', help_text='The Authentication Method to use when sending the xAPI data to the endpoint.', max_length=16, verbose_name='xAPI POST Authentication Method'),
+        ),
+        migrations.AddField(
+            model_name='xapilrsconfiguration',
+            name='auth_url',
+            field=models.URLField(blank=True, help_text='URL to use for authentication. Eg., Token URL for OAuth', null=True),
+        ),
+        migrations.AddField(
+            model_name='xapilrsconfiguration',
+            name='oauth_scope',
+            field=models.CharField(blank=True, help_text='The "scope" to pass for OAuth authentication.', max_length=255, null=True, verbose_name='OAuth scope'),
+        ),
+    ]

integrated_channels/xapi/models.py

@@ -3,8 +3,12 @@
 """
 
 import base64
+from functools import cached_property
+
+import requests
 
 from django.contrib import auth
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 
@@ -16,6 +20,11 @@
 User = auth.get_user_model()
 
 
+class XAPIAuthMethods(models.TextChoices):
+    HTTP_BASIC = 'BASIC', _('HTTP Basic')
+    OAUTH2_CLIENT_CREDS = 'OAUTH2_CC', _('OAuth 2.0 Client Credentials')
+
+
 class XAPILRSConfiguration(TimeStampedModel):
     """
     xAPI LRS configurations.
@@ -39,6 +48,25 @@ class XAPILRSConfiguration(TimeStampedModel):
         null=False,
         help_text=_('Is this configuration active?'),
     )
+    auth_method = models.CharField(
+        max_length=16,
+        verbose_name="xAPI POST Authentication Method",
+        choices=XAPIAuthMethods.choices,
+        default=XAPIAuthMethods.HTTP_BASIC,
+        help_text=_('The Authentication Method to use when sending the xAPI data to the endpoint.')
+    )
+    auth_url = models.URLField(
+        blank=True,
+        null=True,
+        help_text=_("URL to use for authentication. Eg., Token URL for OAuth")
+    )
+    oauth_scope = models.CharField(
+        max_length=255,
+        verbose_name=_('OAuth scope'),
+        blank=True,
+        null=True,
+        help_text=_('The "scope" to pass for OAuth authentication.')
+    )
 
     class Meta:
         app_label = 'xapi'
@@ -62,10 +90,43 @@ def authorization_header(self):
         """
         Authorization header for authenticating requests to LRS.
         """
+        if self.auth_method == XAPIAuthMethods.OAUTH2_CLIENT_CREDS:
+            return f'Bearer {self.access_token}'
+
         return 'Basic {}'.format(
             base64.b64encode('{key}:{secret}'.format(key=self.key, secret=self.secret).encode()).decode()
         )
 
+    def clean(self):
+        errors = {}
+        # Don't allow OAuth2 Client Credentials method to be set without auth_url, or oauth_scope
+        if self.auth_method == XAPIAuthMethods.OAUTH2_CLIENT_CREDS:
+            if not self.auth_url:
+                errors['auth_url'] = _("Authentication URL is required for the OAuth2 authentication method.")
+            if not self.oauth_scope:
+                errors['oauth_scope'] = _("OAuth scope is required for the OAuth2 authentication method.")
+
+        if errors:
+            raise ValidationError(errors)
+
+    @cached_property
+    def access_token(self):
+        """
+        Gets the access token from the OAuth2 Authentication endpoint return it.
+        """
+        if self.auth_method != XAPIAuthMethods.OAUTH2_CLIENT_CREDS:
+            raise RuntimeError("Access Token can be fetched only for OAuth2 Client Credentials authenication method.")
+
+        data = {
+            "grant_type": "client_credentials",
+            "scope": self.oauth_scope,
+            "client_id": self.key,
+            "client_secret": self.secret
+        }
+        response = requests.post(self.auth_url, data=data)
+        response.raise_for_status()
+        return response.json()["access_token"]
+
 
 class XAPILearnerDataTransmissionAudit(LearnerDataTransmissionAudit):
     """

integrated_channels/xapi/utils.py

@@ -147,6 +147,6 @@ def is_success_response(response_fields):
     Returns: Boolean
     """
     success_response = False
-    if response_fields['status'] == 200:
+    if 200 <= response_fields['status'] <= 299:
         success_response = True
     return success_response

test_utils/factories.py

@@ -60,7 +60,7 @@
     SAPSuccessFactorsGlobalConfiguration,
     SapSuccessFactorsLearnerDataTransmissionAudit,
 )
-from integrated_channels.xapi.models import XAPILearnerDataTransmissionAudit, XAPILRSConfiguration
+from integrated_channels.xapi.models import XAPIAuthMethods, XAPILearnerDataTransmissionAudit, XAPILRSConfiguration
 
 FAKER = FakerFactory.create()
 User = auth.get_user_model()
@@ -786,6 +786,9 @@ class Meta:
     key = factory.LazyAttribute(lambda x: FAKER.slug())
     secret = factory.LazyAttribute(lambda x: FAKER.uuid4())
     active = True
+    auth_method = XAPIAuthMethods.HTTP_BASIC
+    auth_url = factory.LazyAttribute(lambda x: FAKER.url())
+    oauth_scope = factory.LazyAttribute(lambda x: FAKER.slug())
 
 
 class XAPILearnerDataTransmissionAuditFactory(factory.django.DjangoModelFactory):

tests/test_integrated_channels/test_xapi/test_models.py

@@ -3,10 +3,15 @@
 """
 
 import base64
+import json
 import unittest
 
+import responses
 from pytest import mark
 
+from django.core.exceptions import ValidationError
+
+from integrated_channels.xapi.models import XAPIAuthMethods, XAPILRSConfiguration
 from test_utils import factories
 
 
@@ -19,6 +24,9 @@ class TestXAPILRSConfiguration(unittest.TestCase):
     def setUp(self):
         super().setUp()
         self.x_api_lrs_config = factories.XAPILRSConfigurationFactory()
+        self.x_api_oauth_lrs_config = factories.XAPILRSConfigurationFactory(
+            auth_method=XAPIAuthMethods.OAUTH2_CLIENT_CREDS
+        )
 
     def test_string_representation(self):
         """
@@ -41,6 +49,52 @@ def test_authorization_header(self):
         )
         assert expected_header == self.x_api_lrs_config.authorization_header
 
+        with responses.RequestsMock() as rsps:
+            rsps.add(
+                responses.POST,
+                self.x_api_oauth_lrs_config.auth_url,
+                body=json.dumps({
+                    'access_token': 'test_token',
+                    'token_type': 'bearer',
+                    'expires_in': 3600
+                }),
+                status=200,
+                content_type="application/json"
+            )
+            expected_header = 'Bearer test_token'
+
+            assert expected_header == self.x_api_oauth_lrs_config.authorization_header
+
+    def test_auth_url_and_scope_are_required_oauth2_auth_method(self):
+        """
+        Test that a validation error is raised when the auth_url or scope is not set for auth method OAUTH2_CC.
+        """
+        conf = XAPILRSConfiguration(
+            enterprise_customer=factories.EnterpriseCustomerFactory(),
+            endpoint="https://xapi.endpoint",
+            key="key",
+            secret="secret",
+            active=True,
+            auth_method="OAUTH2_CC",
+        )
+        with self.assertRaises(ValidationError) as context:
+            conf.full_clean()
+            self.assertIn('auth_url', context.exception.message)
+            self.assertIn('oauth_scope', context.exception.message)
+
+        conf.auth_url = "https://auth.url"
+
+        with self.assertRaises(ValidationError) as context:
+            conf.full_clean()
+            self.assertIn('oauth_scope', context.exception.message)
+
+        conf.auth_url = None
+        conf.oauth_scope = "xapi:write"
+
+        with self.assertRaises(ValidationError) as context:
+            conf.full_clean()
+            self.assertIn('auth_url', context.exception.message)
+
 
 @mark.django_db
 class TestXAPILearnerDataTransmissionAudit(unittest.TestCase):

tests/test_integrated_channels/test_xapi/test_utils.py

@@ -151,6 +151,12 @@ def test_is_success_response(self):
         response_fields = {'status': 200, 'error_message': None}
         self.assertTrue(is_success_response(response_fields))
 
+        response_fields = {'status': 201, 'error_message': None}
+        self.assertTrue(is_success_response(response_fields))
+
+        response_fields = {'status': 202, 'error_message': None}
+        self.assertTrue(is_success_response(response_fields))
+
         response_fields = {'status': 400, 'error_message': None}
         self.assertFalse(is_success_response(response_fields))