PADV-3559 - add OAuth 2.0 Client Credentials support for Credly

credentials/apps/badges/admin.py

@@ -160,13 +160,37 @@ class CredlyOrganizationAdmin(admin.ModelAdmin):
     list_display = (
         "name",
         "uuid",
+        "oauth_client_id",
         "api_key_hidden",
     )
-    fields = [
-        "name",
-        "uuid",
-        "api_key_hidden",
-    ]
+    fieldsets = (
+        (
+            _("General Information"),
+            {
+                "fields": (
+                    "name",
+                    "uuid",
+                ),
+            },
+        ),
+        (
+            _("OAuth 2.0 Credentials"),
+            {
+                "description": _("Authentication credentials for OAuth 2.0 Client Credentials flow."),
+                "fields": (
+                    "oauth_client_id",
+                    "oauth_client_secret",
+                ),
+            },
+        ),
+        (
+            _("Legacy API Key Authentication"),
+            {
+                "description": _("Legacy Basic Auth API key. Use this only if OAuth 2.0 credentials are not available."),
+                "fields": ("api_key",),
+            },
+        ),
+    )
     readonly_fields = [
         "name",
     ]

credentials/apps/badges/admin_forms.py

@@ -5,10 +5,11 @@
 from django import forms
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
+from credentials.apps.badges.exceptions import BadgeProviderError
 from model_utils import Choices
 
 from credentials.apps.badges.credly.api_client import CredlyAPIClient
-from credentials.apps.badges.credly.exceptions import CredlyAPIError
+from credentials.apps.badges.credly.exceptions import CredlyError
 from credentials.apps.badges.models import (
     AbstractDataRule,
     BadgePenalty,
@@ -42,21 +43,45 @@ def clean(self):
 
         uuid = cleaned_data.get("uuid")
         api_key = cleaned_data.get("api_key")
+        oauth_client_id = cleaned_data.get("oauth_client_id")
+        oauth_client_secret = cleaned_data.get("oauth_client_secret")
 
-        if str(uuid) in CredlyOrganization.get_preconfigured_organizations().keys():
-            if api_key:
-                raise forms.ValidationError(_("You can't provide an API key for a configured organization."))
+        is_preconfigured = str(uuid) in CredlyOrganization.get_preconfigured_organizations().keys()
 
+        if is_preconfigured:
+            if api_key or oauth_client_id or oauth_client_secret:
+                raise forms.ValidationError(
+                    _("You can't provide API keys or OAuth credentials for a pre-configured organization.")
+                )
             api_key = settings.BADGES_CONFIG["credly"]["ORGANIZATIONS"][str(uuid)]
 
-        credly_api_client = CredlyAPIClient(uuid, api_key)
+        else:
+            has_oauth = bool(oauth_client_id and oauth_client_secret)
+            has_api_key = bool(api_key)
+
+            if not (has_oauth or has_api_key):
+                raise forms.ValidationError(
+                    _("You must provide either OAuth credentials (Client ID & Client Secret) or a legacy API Key.")
+                )
+
+            if bool(oauth_client_id) != bool(oauth_client_secret):
+                raise forms.ValidationError(
+                    _("Both Client ID and Client Secret are required for OAuth authentication.")
+                )
+
+        credly_api_client = CredlyAPIClient(
+            organization_id=uuid,
+            api_key=api_key,
+            oauth_client_id=oauth_client_id,
+            oauth_client_secret=oauth_client_secret,
+        )
         self.ensure_organization_exists(credly_api_client)
 
         return cleaned_data
 
     def save(self, commit=True):
         """
-        Auto-fill addition properties.
+        Auto-fill additional properties.
         """
         instance = super().save(commit=False)
         instance.name = self.api_data.get("name")
@@ -69,11 +94,26 @@ def ensure_organization_exists(self, api_client):
         Try to fetch organization data by the configured Credly Organization ID.
         """
         try:
-            response_json = api_client.fetch_organization()
-            if org_data := response_json.get("data"):
-                self.api_data = org_data
-        except CredlyAPIError as err:
-            raise forms.ValidationError(message=str(err))
+            if getattr(api_client, "oauth_client_id", None) and getattr(api_client, "oauth_client_secret", None):
+                api_client.fetch_badge_templates()
+
+                self.api_data = {
+                    "name": f"Credly OAuth Org ({api_client.organization_id})"
+                }
+            else:
+                response_json = api_client.fetch_organization()
+                if org_data := response_json.get("data"):
+                    self.api_data = org_data
+
+        except (CredlyError, BadgeProviderError) as exc:
+            if "401" in str(exc) or "Unauthorized" in str(exc):
+                raise forms.ValidationError(
+                    _("Invalid OAuth credentials or API key. Credly rejected the authentication request.")
+                ) from exc
+
+            raise forms.ValidationError(
+                _("Error communicating with Credly API: %(error)s") % {"error": str(exc)}
+            ) from exc
 
 
 class BadgePenaltyForm(forms.ModelForm):

credentials/apps/badges/credly/api_client.py

@@ -5,6 +5,7 @@
 
 import requests  # pylint: disable=unused-import
 from attrs import asdict
+from django.core.cache import cache
 from django.conf import settings
 from django.contrib.sites.models import Site
 
@@ -28,20 +29,34 @@ class CredlyAPIClient(BaseBadgeProviderClient):
 
     PROVIDER_NAME = "Credly"
 
-    def __init__(self, organization_id, api_key=None):  # pylint: disable=super-init-not-called
+    def __init__(
+        self,
+        organization_id,
+        api_key=None,
+        oauth_client_id=None,
+        oauth_client_secret=None,
+    ):  # pylint: disable=super-init-not-called
         """
         Initializes a CredlyRestAPI object.
 
         Args:
             organization_id (str, uuid): ID of the organization.
-            api_key (str): optional ID of the organization.
+            api_key (str): Optional legacy API key of the organization.
+            oauth_client_id (str): Optional OAuth Client ID.
+            oauth_client_secret (str): Optional OAuth Client Secret.
         """
-        if api_key is None:
+        self.organization_id = organization_id
+        self.organization = None
+
+        if not (api_key or (oauth_client_id and oauth_client_secret)):
             self.organization = self._get_organization(organization_id)
             api_key = self.organization.api_key
+            oauth_client_id = getattr(self.organization, "oauth_client_id", None)
+            oauth_client_secret = getattr(self.organization, "oauth_client_secret", None)
 
         self.api_key = api_key
-        self.organization_id = organization_id
+        self.oauth_client_id = oauth_client_id
+        self.oauth_client_secret = oauth_client_secret
 
     def _get_base_api_url(self):
         return urljoin(get_credly_api_base_url(settings), f"organizations/{self.organization_id}/")
@@ -56,24 +71,80 @@ def _get_organization(self, organization_id):
         except CredlyOrganization.DoesNotExist:
             raise CredlyError(f"CredlyOrganization with the uuid {organization_id} does not exist!")
 
+    def _get_oauth_token(self):
+        """
+        Obtains a Bearer Access Token from Credly OAuth endpoint using Client Credentials grant.
+        """
+        if not (self.oauth_client_id and self.oauth_client_secret):
+            return None
+
+        cache_key = f"credly_oauth_access_token_{self.oauth_client_id}"
+        token = cache.get(cache_key)
+
+        if not token:
+            token_url = urljoin(get_credly_api_base_url(settings), "/oauth/token")
+
+            try:
+                payload = {
+                    "grant_type": "client_credentials",
+                    "scope": "badge_templates issued_badges",
+                }
+
+                response = requests.post(
+                    token_url,
+                    data=payload,
+                    auth=(self.oauth_client_id, self.oauth_client_secret),
+                    headers={"Accept": "application/json"},
+                    timeout=10,
+                )
+                response.raise_for_status()
+                data = response.json()
+
+                token = data.get("access_token")
+                expires_in = data.get("expires_in", 7200)
+
+                if not token:
+                    raise CredlyError(f"Credly response did not contain access_token: {data}")
+
+                cache.set(cache_key, token, timeout=max(expires_in - 60, 60))
+
+            except requests.RequestException as exc:
+                raise CredlyError(f"Failed to fetch OAuth token from Credly: {str(exc)}") from exc
+
+        return token
+
     def _get_headers(self):
         """
         Returns the headers for making API requests to Credly.
+        Supports both OAuth Bearer Tokens and legacy Basic Auth API Keys.
         """
-        return {
+        headers = {
             "Accept": "application/json",
             "Content-Type": "application/json",
-            "Authorization": f"Basic {self._build_authorization_token()}",
         }
 
+        if self.oauth_client_id and self.oauth_client_secret:
+            bearer_token = self._get_oauth_token()
+            headers["Authorization"] = f"Bearer {bearer_token}"
+
+        elif self.api_key:
+            headers["Authorization"] = f"Basic {self._build_authorization_token()}"
+
+        else:
+            raise CredlyError("No valid authentication credentials (OAuth or API Key) available for Credly.")
+
+        return headers
+
     @lru_cache
     def _build_authorization_token(self):
         """
-        Build the authorization token for the Credly API.
+        Build the authorization token for the Credly API (Legacy).
 
         Returns:
             str: Authorization token.
         """
+        if not self.api_key:
+            return ""
         return base64.b64encode(self.api_key.encode("ascii")).decode("ascii")
 
     def fetch_organization(self):
@@ -154,6 +225,9 @@ def sync_organization_badge_templates(self, site_id):
             logger.error(f"Site with the id {site_id} does not exist!")
             raise
 
+        if not self.organization:
+            self.organization = self._get_organization(self.organization_id)
+
         badge_templates_data = self.fetch_badge_templates()
         raw_badge_templates = badge_templates_data.get("data", [])
 

credentials/apps/badges/migrations/0003_credlyorganization_oauth_client_id_and_more.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.2.7 on 2026-08-03 13:47
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('badges', '0002_accredibleapiconfig_accrediblebadge_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='credlyorganization',
+            name='oauth_client_id',
+            field=models.CharField(blank=True, help_text='OAuth 2.0 Client ID for Credly Organization.', max_length=255, null=True),
+        ),
+        migrations.AddField(
+            model_name='credlyorganization',
+            name='oauth_client_secret',
+            field=models.CharField(blank=True, help_text='OAuth 2.0 Client Secret for Credly Organization.', max_length=255, null=True),
+        ),
+    ]

credentials/apps/badges/models.py

@@ -48,6 +48,18 @@ class CredlyOrganization(TimeStampedModel):
         blank=True,
         help_text=_("Verbose name for Credly Organization."),
     )
+    oauth_client_id = models.CharField(
+        max_length=255,
+        blank=True,
+        null=True,
+        help_text=_("OAuth 2.0 Client ID for Credly Organization.")
+    )
+    oauth_client_secret = models.CharField(
+        max_length=255,
+        blank=True,
+        null=True,
+        help_text=_("OAuth 2.0 Client Secret for Credly Organization.")
+    )
 
     def __str__(self):
         return f"{self.name or self.uuid}"