feat: add validation to issue badge

credentials/apps/badges/admin.py

@@ -137,6 +137,7 @@ class FulfillmentInline(admin.TabularInline):
     extra = 0
     readonly_fields = [
         "requirement",
+        "course_key",
     ]
 
 

credentials/apps/badges/migrations/0004_fulfillment_course_key.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.7 on 2026-09-06 17:22
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('badges', '0003_credlyorganization_oauth_client_id_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='fulfillment',
+            name='course_key',
+            field=models.CharField(blank=True, help_text='Course key where this requirement was fulfilled', max_length=255, null=True),
+        ),
+    ]

credentials/apps/badges/models.py

@@ -222,7 +222,7 @@ class BadgeRequirement(models.Model):
     def __str__(self):
         return f"BadgeRequirement:{self.id}:{self.template.uuid}"
 
-    def fulfill(self, username: str):
+    def fulfill(self, username: str, course_key: str = None):
         """
         Marks itself as "done" for the user.
 
@@ -233,7 +233,12 @@ def fulfill(self, username: str):
         """
         template_id = self.template.id
         progress = BadgeProgress.for_user(username=username, template_id=template_id, create_if_absent=True)
-        fulfillment, created = Fulfillment.objects.get_or_create(progress=progress, requirement=self, blend=self.blend)
+        fulfillment, created = Fulfillment.objects.get_or_create(
+            progress=progress, 
+            requirement=self, 
+            blend=self.blend, 
+            course_key=course_key
+        )
 
         if created:
             notify_requirement_fulfilled(
@@ -635,6 +640,12 @@ class Fulfillment(models.Model):
         help_text=_("Group ID for the requirement."),
         verbose_name=_("group"),
     )
+    course_key = models.CharField(
+        max_length=255,
+        null=True,
+        blank=True,
+        help_text="Course key where this requirement was fulfilled"
+    )
 
 
 class CredlyBadge(UserCredential):

credentials/apps/badges/processing/progression.py

@@ -31,6 +31,12 @@ def process_requirements(event_type, username, payload):
 
     logger.debug("BADGES: found %s requirements to process.", len(requirements))
 
+    course_key = None
+
+    if hasattr(payload, 'course') and hasattr(payload.course, 'ccx_course_key'):
+        course_key = str(payload.course.ccx_course_key)
+        logger.debug(f"BADGES: extracted course_key={course_key}")
+
     for requirement in requirements:
 
         # remember: the badge template is already "done"
@@ -47,4 +53,4 @@ def process_requirements(event_type, username, payload):
 
         # process: payload rules
         if requirement.apply_rules(asdict(payload)):
-            requirement.fulfill(username)
+            requirement.fulfill(username, course_key=course_key)

credentials/apps/badges/processing/restrictions.py

@@ -0,0 +1,123 @@
+"""Badge issuance restriction checks."""
+
+import logging
+
+from django.conf import settings
+from django.core.exceptions import ObjectDoesNotExist
+
+from credentials.apps.badges.models import CredlyBadgeTemplate
+from credentials.apps.core.models import SiteConfiguration
+
+logger = logging.getLogger(__name__)
+
+
+def is_badge_issuance_allowed(username, badge_template_id, progress):
+    """
+    Check whether badge issuance is allowed for the completed badge progress.
+
+    The check is disabled by default. When enabled, each fulfillment with a
+    course key is validated against the configured restrictions API.
+
+    Badge issuance is blocked only when an exact course match explicitly
+    disables the configured badge flag. Missing data or API errors allow
+    issuance by default.
+    """
+    if not getattr(settings, "BADGES_ENABLE_ISSUANCE_RESTRICTIONS", False):
+        return True
+
+    if not progress:
+        return True
+
+    try:
+        badge_template = CredlyBadgeTemplate.objects.get(id=badge_template_id)
+        site_configuration = SiteConfiguration.objects.get(site=badge_template.site)
+        api_client = site_configuration.api_client
+    except ObjectDoesNotExist:
+        logger.exception(
+            "Unable to load SiteConfiguration for badge template %s. "
+            "Badge issuance will be allowed by default.",
+            badge_template_id,
+        )
+        return True
+
+    fulfillments = progress.fulfillment_set.all()
+    restrictions_api_url = getattr(
+        settings,
+        "BADGES_ISSUANCE_RESTRICTIONS_API_URL",
+        "",
+    )
+    restriction_flag_name = getattr(
+        settings,
+        "BADGES_ISSUANCE_RESTRICTIONS_FLAG_NAME",
+        "",
+    )
+
+    for fulfillment in fulfillments:
+        if not fulfillment.course_key:
+            continue
+
+        course_key = str(fulfillment.course_key)
+
+        try:
+            response = api_client.get(
+                restrictions_api_url,
+                params={"ccx_id": course_key},
+                timeout=5,
+            )
+        except Exception:
+            logger.exception(
+                "Error checking badge issuance restrictions for course %s.",
+                course_key,
+            )
+            continue
+
+        if response.status_code != 200:
+            logger.warning(
+                "Restrictions API returned status %s for course %s.",
+                response.status_code,
+                course_key,
+            )
+            continue
+
+        try:
+            response_data = response.json()
+        except ValueError:
+            logger.warning(
+                "Restrictions API returned invalid JSON for course %s.",
+                course_key,
+            )
+            continue
+
+        results = response_data.get("results", [])
+
+        matched_result = next(
+            (
+                result
+                for result in results
+                if result.get("ccx_id") == course_key
+            ),
+            None,
+        )
+
+        if not matched_result:
+            logger.warning(
+                "No exact restriction result found for course %s. "
+                "Badge issuance will be allowed by default.",
+                course_key,
+            )
+            continue
+
+        allow_badges = matched_result.get(
+            restriction_flag_name,
+            True,
+        )
+
+        if not allow_badges:
+            logger.info(
+                "Badge issuance blocked for user %s and course %s.",
+                username,
+                course_key,
+            )
+            return False
+
+    return True

credentials/apps/badges/signals/handlers.py

@@ -7,6 +7,7 @@
 import logging
 
 from django.dispatch import receiver
+from credentials.apps.badges.processing.restrictions import is_badge_issuance_allowed
 from openedx_events.tooling import OpenEdxPublicSignal, load_all_signals
 
 from credentials.apps.badges.issuers import AccredibleBadgeTemplateIssuer, CredlyBadgeTemplateIssuer
@@ -81,6 +82,15 @@ def handle_badge_completion(sender, username, badge_template_id, origin, **kwarg
     logger.debug("BADGES: progress is complete for %s on the %s", username, badge_template_id)
 
     if origin == CredlyBadgeTemplate.ORIGIN:
+        progress = BadgeProgress.for_user(username=username, template_id=badge_template_id)
+
+        if not is_badge_issuance_allowed(
+            username=username,
+            badge_template_id=badge_template_id,
+            progress=progress,
+        ):
+            return
+
         CredlyBadgeTemplateIssuer().award(username=username, credential_id=badge_template_id)
     elif origin == AccredibleGroup.ORIGIN:
         AccredibleBadgeTemplateIssuer().award(username=username, credential_id=badge_template_id)