From 614192691af75d31763156ac3848a8d9e9a884bc Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Fri, 3 Jul 2026 15:50:01 +0200
Subject: [PATCH 1/8] OIDC Support

---
 pulpcore/app/oidc/__init__.py         |   5 +
 pulpcore/app/oidc/authentication.py   | 105 +++++++++++++++++++
 pulpcore/app/oidc/authz.py            | 109 ++++++++++++++++++++
 pulpcore/app/oidc/config.py           |  35 +++++++
 pulpcore/app/oidc/principal.py        |  71 +++++++++++++
 pulpcore/app/oidc/rules.py            |  32 ++++++
 pulpcore/app/role_util.py             |  19 ++++
 pulpcore/app/settings.py              |   6 ++
 pulpcore/tests/unit/oidc/__init__.py  |   0
 pulpcore/tests/unit/oidc/test_oidc.py | 141 ++++++++++++++++++++++++++
 pyproject.toml                        |   1 +
 11 files changed, 524 insertions(+)
 create mode 100644 pulpcore/app/oidc/__init__.py
 create mode 100644 pulpcore/app/oidc/authentication.py
 create mode 100644 pulpcore/app/oidc/authz.py
 create mode 100644 pulpcore/app/oidc/config.py
 create mode 100644 pulpcore/app/oidc/principal.py
 create mode 100644 pulpcore/app/oidc/rules.py
 create mode 100644 pulpcore/tests/unit/oidc/__init__.py
 create mode 100644 pulpcore/tests/unit/oidc/test_oidc.py

diff --git a/pulpcore/app/oidc/__init__.py b/pulpcore/app/oidc/__init__.py
new file mode 100644
index 00000000000..f5690aef4bc
--- /dev/null
+++ b/pulpcore/app/oidc/__init__.py
@@ -0,0 +1,5 @@
+"""OIDC authentication for CI clients.
+
+A validated OIDC token becomes a stateless principal that carries grants computed per request
+from the ``OIDC_AUTH`` setting. Nothing is written to the role tables.
+"""
diff --git a/pulpcore/app/oidc/authentication.py b/pulpcore/app/oidc/authentication.py
new file mode 100644
index 00000000000..65dd49ce11f
--- /dev/null
+++ b/pulpcore/app/oidc/authentication.py
@@ -0,0 +1,105 @@
+"""DRF authentication that validates third-party OIDC tokens.
+
+This authenticator accepts an OIDC token issued by a configured third-party
+provider (for example a GitHub Actions workflow token), verifies it against the
+provider's JWKS, maps its claims to grants and returns a stateless
+``OIDCPrincipal``. No database user is involved.
+
+The token may arrive either as a ``Bearer`` token or inside a ``Basic`` header
+(the way ``docker login`` passes a token, where the password field carries it).
+"""
+
+import base64
+import binascii
+import logging
+
+import jwt
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.exceptions import AuthenticationFailed
+
+from pulpcore.app.oidc import config, rules
+from pulpcore.app.oidc.principal import OIDCPrincipal
+
+_logger = logging.getLogger("pulpcore.oidc")
+
+
+class OIDCAuthentication(BaseAuthentication):
+    """Authenticate requests bearing a third-party OIDC token.
+
+    On success this returns a stateless ``OIDCPrincipal`` whose permissions are
+    derived entirely from the grants earned by the token's claims. When the
+    request carries no token, or a token that is not meant for us, the
+    authenticator returns ``None`` so that other authenticators may run.
+    """
+
+    def _get_token(self, request):
+        """Return the token from the Authorization header (Bearer, or Basic password), or None."""
+        header = request.META.get("HTTP_AUTHORIZATION", "")
+        parts = header.split()
+        if len(parts) != 2:
+            return None
+        scheme, value = parts
+        scheme = scheme.lower()
+        if scheme == "bearer":
+            return value
+        if scheme == "basic":
+            try:
+                decoded = base64.b64decode(value).decode("utf-8")
+            except (binascii.Error, ValueError, UnicodeDecodeError):
+                return None
+            if ":" not in decoded:
+                return None
+            # docker login passes the token as the password; the username is ignored.
+            _, _, password = decoded.partition(":")
+            return password
+        return None
+
+    def authenticate(self, request):
+        """Validate an OIDC token and return ``(OIDCPrincipal, claims)``, or ``None`` if not ours."""
+        token = self._get_token(request)
+        if not token:
+            return None
+
+        # Peek at the issuer without verifying the signature. If this is not a
+        # JWT at all, it is not meant for us.
+        try:
+            unverified = jwt.decode(token, options={"verify_signature": False})
+        except jwt.PyJWTError:
+            return None
+        issuer = unverified.get("iss")
+
+        provider = config.provider_for_issuer(issuer)
+        if provider is None:
+            return None
+
+        # Verify the token for real against the provider's JWKS.
+        try:
+            signing_key = config.jwks_client(provider).get_signing_key_from_jwt(token)
+            claims = jwt.decode(
+                token,
+                signing_key.key,
+                algorithms=provider.get("algorithms", ["RS256"]),
+                issuer=provider["issuer"],
+                audience=provider["audience"],
+                options={"require": ["exp", "iss", "aud"]},
+            )
+        except jwt.PyJWTError as exc:
+            _logger.info("Rejecting OIDC token from %s: %s", issuer, exc)
+            raise AuthenticationFailed("Invalid OIDC token.")
+
+        grants = rules.grants_for(provider, claims)
+        if not grants:
+            _logger.info(
+                "No matching OIDC rule for sub=%r repository=%r",
+                claims.get("sub"),
+                claims.get("repository"),
+            )
+            raise AuthenticationFailed("No matching OIDC rule.")
+
+        # The username is intentionally empty: the container registry token
+        # subject must be empty for a principal with no database user.
+        return (OIDCPrincipal(grants, username=""), claims)
+
+    def authenticate_header(self, request):
+        """Return the ``WWW-Authenticate`` value so failures are 401, not 403."""
+        return "Bearer"
diff --git a/pulpcore/app/oidc/authz.py b/pulpcore/app/oidc/authz.py
new file mode 100644
index 00000000000..9a10eef9618
--- /dev/null
+++ b/pulpcore/app/oidc/authz.py
@@ -0,0 +1,109 @@
+"""Shared authorization logic over a list of grants.
+
+A grant is ``{"role": <role name>, "scope": {...}}``. Scope is one of:
+
+* ``{"type": "global"}``
+* ``{"type": "domain", "domain": "<name>"}``
+* ``{"type": "object", "name": "<name>"}`` (or ``"prn"``)
+
+This module is used both by the principal (single-object ``has_perm`` and the model-level
+permission set) and by ``get_objects_for_user`` (list filtering). Roles are read from the
+database to resolve their permissions; the grant assignment itself is never stored.
+"""
+
+from django.db.models import Q
+
+
+def _split(permission):
+    app_label, _, codename = permission.partition(".")
+    return app_label, codename
+
+
+def _role_has_perm(role_name, app_label, codename):
+    from pulpcore.app.models.role import Role
+
+    if not role_name:
+        return False
+    try:
+        role = Role.objects.get(name=role_name)
+    except Role.DoesNotExist:
+        return False
+    return role.permissions.filter(
+        content_type__app_label=app_label, codename=codename
+    ).exists()
+
+
+def _scope_matches(scope, obj):
+    """Whether a scope applies to a single object (``obj`` may be ``None`` for model-level)."""
+    from pulpcore.app.models import Domain
+
+    stype = scope.get("type")
+    if stype == "global":
+        return True
+    if obj is None:
+        return False
+    if stype == "domain":
+        if isinstance(obj, Domain):
+            return obj.name == scope.get("domain")
+        domain = getattr(obj, "pulp_domain", None)
+        return domain is not None and domain.name == scope.get("domain")
+    if stype == "object":
+        checks = (("prn", "prn"), ("name", "name"))
+        provided = [key for key, _ in checks if key in scope]
+        if not provided:
+            return False
+        return all(str(getattr(obj, attr, None)) == str(scope[key]) for key, attr in checks if key in scope)
+    return False
+
+
+def has_grant_perm(grants, permission, obj=None):
+    """True if any grant confers ``permission`` and its scope matches ``obj``."""
+    app_label, codename = _split(permission)
+    for grant in grants:
+        if _role_has_perm(grant.get("role"), app_label, codename) and _scope_matches(
+            grant.get("scope", {}), obj
+        ):
+            return True
+    return False
+
+
+def permissions_for(grants):
+    """The set of ``app_label.codename`` the grants confer, for model-level ``get_all_permissions``."""
+    from pulpcore.app.models.role import Role
+
+    names = {g.get("role") for g in grants if g.get("role")}
+    perms = set()
+    for role in Role.objects.filter(name__in=names).prefetch_related("permissions__content_type"):
+        for perm in role.permissions.all():
+            perms.add(f"{perm.content_type.app_label}.{perm.codename}")
+    return perms
+
+
+def grants_queryset(grants, permission, queryset):
+    """Return ``queryset`` filtered to the objects the grants allow for ``permission``."""
+    app_label, codename = _split(permission)
+    relevant = [g for g in grants if _role_has_perm(g.get("role"), app_label, codename)]
+    if not relevant:
+        return queryset.none()
+
+    has_domain = hasattr(queryset.model, "pulp_domain")
+    predicate = None
+    for grant in relevant:
+        scope = grant.get("scope", {})
+        stype = scope.get("type")
+        if stype == "global":
+            return queryset
+        clause = None
+        if stype == "domain" and has_domain:
+            clause = Q(pulp_domain__name=scope.get("domain"))
+        elif stype == "object":
+            if "prn" in scope:
+                clause = Q(prn=scope["prn"])
+            elif "name" in scope:
+                clause = Q(name=scope["name"])
+        if clause is not None:
+            predicate = clause if predicate is None else (predicate | clause)
+
+    if predicate is None:
+        return queryset.none()
+    return queryset.filter(predicate)
diff --git a/pulpcore/app/oidc/config.py b/pulpcore/app/oidc/config.py
new file mode 100644
index 00000000000..473d224088b
--- /dev/null
+++ b/pulpcore/app/oidc/config.py
@@ -0,0 +1,35 @@
+"""Access to the ``OIDC_AUTH`` setting and per-provider JWKS clients."""
+
+import functools
+
+from django.conf import settings
+from jwt import PyJWKClient
+
+
+def config():
+    return getattr(settings, "OIDC_AUTH", {}) or {}
+
+
+def strategy():
+    return config().get("strategy", "union")
+
+
+def providers():
+    return config().get("providers", {}) or {}
+
+
+def provider_for_issuer(issuer):
+    """Return the provider entry whose ``issuer`` matches, or ``None``."""
+    for entry in providers().values():
+        if entry.get("issuer") == issuer:
+            return entry
+    return None
+
+
+@functools.lru_cache(maxsize=None)
+def _jwks_client(jwks_url):
+    return PyJWKClient(jwks_url, cache_keys=True)
+
+
+def jwks_client(provider):
+    return _jwks_client(provider["jwks_url"])
diff --git a/pulpcore/app/oidc/principal.py b/pulpcore/app/oidc/principal.py
new file mode 100644
index 00000000000..34863556b34
--- /dev/null
+++ b/pulpcore/app/oidc/principal.py
@@ -0,0 +1,71 @@
+"""A stateless principal used as ``request.user`` for OIDC-authenticated clients.
+
+``OIDCPrincipal`` deliberately does not subclass Django's user model or
+``PermissionsMixin``. It implements its own permission-checking methods so that
+Django never dispatches permission checks for it to ``AUTHENTICATION_BACKENDS``.
+Authorization is derived entirely from the list of grants carried on the object.
+"""
+
+from pulpcore.app.oidc.authz import has_grant_perm, permissions_for
+
+
+class OIDCPrincipal:
+    """A user-like object backed by OIDC grants rather than a database row.
+
+    This object is safe to assign to ``request.user``. It exposes the attributes
+    the request path reads and answers permission checks from its ``grants``.
+
+    Attributes:
+        grants (list): The list of grant dicts backing this principal.
+        username (str): The principal's username (may be empty).
+    """
+
+    is_authenticated = True
+    is_active = True
+    is_anonymous = False
+    is_superuser = False
+    is_staff = False
+    pk = None
+    id = None
+
+    def __init__(self, grants, username=""):
+        """Initialize the principal.
+
+        Args:
+            grants (list): A list of grant dicts (see ``authz`` for the shape).
+            username (str): The principal's username. Defaults to an empty string.
+        """
+        self.grants = grants
+        self.username = username
+
+    @property
+    def groups(self):
+        """An empty ``Group`` queryset so ``user.groups.all()`` never crashes."""
+        from pulpcore.app.models import Group
+
+        return Group.objects.none()
+
+    def has_perm(self, perm, obj=None):
+        """Whether the grants confer ``perm`` (optionally scoped to ``obj``)."""
+        return has_grant_perm(self.grants, perm, obj)
+
+    def has_perms(self, perm_list, obj=None):
+        """Whether the grants confer every permission in ``perm_list``."""
+        return all(self.has_perm(perm, obj) for perm in perm_list)
+
+    def has_module_perms(self, app_label):
+        """Whether the grants confer any permission in ``app_label``."""
+        prefix = f"{app_label}."
+        return any(perm.startswith(prefix) for perm in self.get_all_permissions())
+
+    def get_all_permissions(self, obj=None):
+        """The set of ``app_label.codename`` permissions the grants confer."""
+        return permissions_for(self.grants)
+
+    def get_group_permissions(self, obj=None):
+        """The permissions from groups; always empty for a principal."""
+        return set()
+
+    def __str__(self):
+        """The username, or ``"oidc-principal"`` when it is empty."""
+        return self.username or "oidc-principal"
diff --git a/pulpcore/app/oidc/rules.py b/pulpcore/app/oidc/rules.py
new file mode 100644
index 00000000000..0058ffcb5e6
--- /dev/null
+++ b/pulpcore/app/oidc/rules.py
@@ -0,0 +1,32 @@
+"""Match token claims against the provider rules and collect the grants they earn.
+
+Nothing here touches the database. A grant is ``{"role": <name>, "scope": {...}}``.
+"""
+
+import fnmatch
+
+from pulpcore.app.oidc.config import strategy
+
+
+def _matches(match_spec, claims):
+    """All claim conditions must match (AND). Values support ``*`` globbing."""
+    for claim, pattern in match_spec.items():
+        value = claims.get(claim)
+        if value is None or not fnmatch.fnmatchcase(str(value), str(pattern)):
+            return False
+    return True
+
+
+def grants_for(provider, claims):
+    """Return the list of grants the claims earn for this provider.
+
+    ``strategy`` "union" accumulates every matching rule; "first-match" stops at the first.
+    """
+    grants = []
+    first_match = strategy() == "first-match"
+    for rule in provider.get("rules", []):
+        if _matches(rule.get("match", {}), claims):
+            grants.extend(rule.get("grants", []))
+            if first_match:
+                break
+    return grants
diff --git a/pulpcore/app/role_util.py b/pulpcore/app/role_util.py
index 103c68541e3..52c9fbfb074 100644
--- a/pulpcore/app/role_util.py
+++ b/pulpcore/app/role_util.py
@@ -173,6 +173,25 @@ def get_objects_for_user(
     accept_domain_perms=True,
     accept_global_perms=True,
 ):
+    from pulpcore.app.oidc.principal import OIDCPrincipal
+
+    if isinstance(user, OIDCPrincipal):
+        # Stateless OIDC principal: scope from its per-request grants, not the role tables.
+        from pulpcore.app.oidc.authz import grants_queryset
+
+        grants = user.grants
+        if isinstance(perms, str):
+            return grants_queryset(grants, perms, qs)
+        if any_perm:
+            result = qs.none()
+            for permission_name in perms:
+                result |= grants_queryset(grants, permission_name, qs)
+            return result
+        result = qs.all()
+        for permission_name in perms:
+            result &= grants_queryset(grants, permission_name, qs)
+        return result
+
     new_qs = qs.none()
     replace = False
     if "pulpcore.backends.ObjectRolePermissionBackend" in settings.AUTHENTICATION_BACKENDS:
diff --git a/pulpcore/app/settings.py b/pulpcore/app/settings.py
index 6961b88f302..2f63cf95290 100644
--- a/pulpcore/app/settings.py
+++ b/pulpcore/app/settings.py
@@ -316,6 +316,12 @@
 AUTHENTICATION_JSON_HEADER_JQ_FILTER = ""
 AUTHENTICATION_JSON_HEADER_OPENAPI_SECURITY_SCHEME = {}
 
+# OIDC authentication for CI clients. Empty by default (feature off). See
+# pulpcore.app.oidc.config for the schema. Enable by adding
+# "pulpcore.app.oidc.authentication.OIDCAuthentication" to
+# REST_FRAMEWORK["DEFAULT_AUTHENTICATION_CLASSES"] and populating this setting.
+OIDC_AUTH = {}
+
 ALLOWED_IMPORT_PATHS = []
 
 ALLOWED_EXPORT_PATHS = []
diff --git a/pulpcore/tests/unit/oidc/__init__.py b/pulpcore/tests/unit/oidc/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/pulpcore/tests/unit/oidc/test_oidc.py b/pulpcore/tests/unit/oidc/test_oidc.py
new file mode 100644
index 00000000000..5017bde4efa
--- /dev/null
+++ b/pulpcore/tests/unit/oidc/test_oidc.py
@@ -0,0 +1,141 @@
+import pytest
+from django.contrib.auth.models import Permission
+from django.test import override_settings
+
+from pulpcore.app.models import Repository
+from pulpcore.app.models.role import Role
+from pulpcore.app.role_util import get_objects_for_user
+
+from pulpcore.app.oidc.authz import grants_queryset, has_grant_perm, permissions_for
+from pulpcore.app.oidc.principal import OIDCPrincipal
+from pulpcore.app.oidc.rules import grants_for
+
+
+PROVIDER = {
+    "issuer": "https://issuer",
+    "rules": [
+        {
+            "match": {"repository": "org/infra", "ref": "refs/heads/main"},
+            "grants": [{"role": "role1", "scope": {"type": "global"}}],
+        },
+        {
+            "match": {"repository": "org/*"},
+            "grants": [{"role": "role2", "scope": {"type": "domain", "domain": "default"}}],
+        },
+    ],
+}
+
+
+# --- rules.grants_for (no database) ---
+
+
+def test_rules_no_match():
+    assert grants_for(PROVIDER, {"repository": "other/x"}) == []
+
+
+@override_settings(OIDC_AUTH={"strategy": "union"})
+def test_rules_union_accumulates():
+    grants = grants_for(PROVIDER, {"repository": "org/infra", "ref": "refs/heads/main"})
+    assert {g["role"] for g in grants} == {"role1", "role2"}
+
+
+@override_settings(OIDC_AUTH={"strategy": "first-match"})
+def test_rules_first_match_stops():
+    grants = grants_for(PROVIDER, {"repository": "org/infra", "ref": "refs/heads/main"})
+    assert [g["role"] for g in grants] == ["role1"]
+
+
+def test_rules_glob_only():
+    grants = grants_for(PROVIDER, {"repository": "org/app"})
+    assert [g["role"] for g in grants] == ["role2"]
+
+
+# --- database fixtures ---
+
+
+@pytest.fixture
+def repo_viewer_role(db):
+    role = Role.objects.create(name="repo_viewer")
+    role.permissions.add(
+        Permission.objects.get(content_type__app_label="core", codename="view_repository")
+    )
+    return role
+
+
+@pytest.fixture
+def repo_a(db):
+    return Repository.objects.create(name="repo-a")
+
+
+@pytest.fixture
+def repo_b(db):
+    return Repository.objects.create(name="repo-b")
+
+
+# --- authz.has_grant_perm ---
+
+
+def test_has_grant_perm_global(repo_viewer_role, repo_a):
+    grants = [{"role": "repo_viewer", "scope": {"type": "global"}}]
+    assert has_grant_perm(grants, "core.view_repository", repo_a) is True
+
+
+def test_has_grant_perm_object_scope(repo_viewer_role, repo_a, repo_b):
+    grants = [{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}]
+    assert has_grant_perm(grants, "core.view_repository", repo_a) is True
+    assert has_grant_perm(grants, "core.view_repository", repo_b) is False
+
+
+def test_has_grant_perm_wrong_permission(repo_viewer_role, repo_a):
+    grants = [{"role": "repo_viewer", "scope": {"type": "global"}}]
+    assert has_grant_perm(grants, "core.change_repository", repo_a) is False
+
+
+def test_has_grant_perm_missing_role(db, repo_a):
+    grants = [{"role": "does_not_exist", "scope": {"type": "global"}}]
+    assert has_grant_perm(grants, "core.view_repository", repo_a) is False
+
+
+# --- authz.permissions_for ---
+
+
+def test_permissions_for(repo_viewer_role):
+    grants = [{"role": "repo_viewer", "scope": {"type": "global"}}]
+    assert "core.view_repository" in permissions_for(grants)
+
+
+# --- authz.grants_queryset ---
+
+
+def test_grants_queryset_global(repo_viewer_role, repo_a, repo_b):
+    grants = [{"role": "repo_viewer", "scope": {"type": "global"}}]
+    qs = grants_queryset(grants, "core.view_repository", Repository.objects.all())
+    assert {"repo-a", "repo-b"} <= set(qs.values_list("name", flat=True))
+
+
+def test_grants_queryset_object(repo_viewer_role, repo_a, repo_b):
+    grants = [{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}]
+    qs = grants_queryset(grants, "core.view_repository", Repository.objects.all())
+    assert set(qs.values_list("name", flat=True)) == {"repo-a"}
+
+
+def test_grants_queryset_no_relevant_grant(db, repo_a):
+    grants = [{"role": "does_not_exist", "scope": {"type": "global"}}]
+    qs = grants_queryset(grants, "core.view_repository", Repository.objects.all())
+    assert qs.count() == 0
+
+
+# --- principal + get_objects_for_user branch ---
+
+
+def test_principal_is_authenticated_and_has_perm(repo_viewer_role, repo_a):
+    principal = OIDCPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
+    assert principal.is_authenticated is True
+    assert principal.pk is None
+    assert principal.has_perm("core.view_repository", repo_a) is True
+
+
+def test_get_objects_for_user_scopes_by_grants(repo_viewer_role, repo_a, repo_b):
+    principal = OIDCPrincipal([{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}])
+    qs = get_objects_for_user(principal, "core.view_repository", Repository.objects.all())
+    assert set(qs.values_list("name", flat=True)) == {"repo-a"}
diff --git a/pyproject.toml b/pyproject.toml
index 4502deab44b..700f1ba489a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -31,6 +31,7 @@ dependencies = [
   "backoff>=2.1.2,<2.3",  # Looks like only bugfixes in z-Stream.
   "click>=8.1.0,<8.5",  # Uses milestone.feature.fix https://palletsprojects.com/versions .
   "cryptography>=44.0.3,<50.0",  # SemVer compatible https://cryptography.io/en/latest/api-stability/#versioning .
+  "pyjwt>=2.4,<3",  # SemVer. Used to validate OIDC tokens against a provider JWKS.
   "Django>=5.2.0,<5.3",  # LTS version, we aim at supporting one or two at a time.
   "django-filter>=24.3,<=25.2",  # Uses CalVer, not released often https://github.com/carltongibson/django-filter
   "django-guid>=3.4.0,<3.7",  # Looks like only bugfixes in z-Stream.

From 88ecdfdedab3a968049236990fcc0a01d84a052e Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Wed, 8 Jul 2026 11:28:54 +0200
Subject: [PATCH 2/8] Rename OIDC auth to workload identity, add docs and fixes

---
 docs/admin/guides/_SUMMARY.md                 |   1 +
 docs/admin/guides/auth/workload_identity.md   | 102 ++++++++++++++++++
 pulpcore/app/access_policy.py                 |   7 ++
 pulpcore/app/oidc/__init__.py                 |   5 -
 pulpcore/app/role_util.py                     |   7 +-
 pulpcore/app/settings.py                      |   7 +-
 pulpcore/app/workload_identity/__init__.py    |   5 +
 .../authentication.py                         |  31 ++----
 .../app/{oidc => workload_identity}/authz.py  |  48 ++++++---
 .../app/{oidc => workload_identity}/config.py |   4 +-
 .../{oidc => workload_identity}/principal.py  |  36 +++----
 .../app/{oidc => workload_identity}/rules.py  |   2 +-
 .../{oidc => workload_identity}/__init__.py   |   0
 .../test_workload_identity.py}                |  14 +--
 14 files changed, 187 insertions(+), 82 deletions(-)
 create mode 100644 docs/admin/guides/auth/workload_identity.md
 delete mode 100644 pulpcore/app/oidc/__init__.py
 create mode 100644 pulpcore/app/workload_identity/__init__.py
 rename pulpcore/app/{oidc => workload_identity}/authentication.py (67%)
 rename pulpcore/app/{oidc => workload_identity}/authz.py (69%)
 rename pulpcore/app/{oidc => workload_identity}/config.py (82%)
 rename pulpcore/app/{oidc => workload_identity}/principal.py (53%)
 rename pulpcore/app/{oidc => workload_identity}/rules.py (94%)
 rename pulpcore/tests/unit/{oidc => workload_identity}/__init__.py (100%)
 rename pulpcore/tests/unit/{oidc/test_oidc.py => workload_identity/test_workload_identity.py} (88%)

diff --git a/docs/admin/guides/_SUMMARY.md b/docs/admin/guides/_SUMMARY.md
index fe1d8f15e76..1965cc35575 100644
--- a/docs/admin/guides/_SUMMARY.md
+++ b/docs/admin/guides/_SUMMARY.md
@@ -3,6 +3,7 @@
     * [Using external service](auth/external.md)
     * [Using Keycloak](auth/keycloak.md)
     * [Using JSON Header](auth/json_header.md)
+    * [Using Workload Identity](auth/workload_identity.md)
     * auth/*.md
 * Configuration
     * [Introduction](configure-pulp/index.md)
diff --git a/docs/admin/guides/auth/workload_identity.md b/docs/admin/guides/auth/workload_identity.md
new file mode 100644
index 00000000000..01cb87c95fe
--- /dev/null
+++ b/docs/admin/guides/auth/workload_identity.md
@@ -0,0 +1,102 @@
+# Workload Identity Authentication
+
+A CI job can authenticate to Pulp with a short-lived OIDC token from a third-party provider (for
+example GitHub Actions) instead of a stored username and password. The token is verified against the
+provider's public keys, its claims are matched against a set of rules, and the request is granted
+roles for that request only. No user is created and nothing is written to the role tables.
+
+This suits supply-chain workflows where a pipeline pushes content and you want its permissions scoped
+to specific repositories without long-lived secrets.
+
+!!! note
+    The token is an OIDC token, but this is unrelated to the user-facing SSO login covered in
+    [Using external service](external.md). It identifies a workload, not a person.
+
+## How it works
+
+On each request the token is read from the `Authorization` header, either as a `Bearer` token or as
+the password of a `Basic` header (the way `docker login` sends a token). The `iss` claim selects a
+configured provider, the signature is verified against the provider's JWKS, and `iss`, `aud` and
+`exp` are checked. The remaining claims are matched against the provider's rules to compute the roles
+and scopes for the request. A token that matches no rule is rejected with a 401.
+
+## Enabling
+
+Add the authentication class to `DEFAULT_AUTHENTICATION_CLASSES`, before `BasicAuthentication` so the
+`docker login` path reaches it, then populate `WORKLOAD_IDENTITY`:
+
+```python title="settings.py"
+REST_FRAMEWORK["DEFAULT_AUTHENTICATION_CLASSES"] = [
+    "pulpcore.app.workload_identity.authentication.WorkloadIdentityAuthentication",
+    "pulpcore.app.authentication.BasicAuthentication",
+    "rest_framework.authentication.SessionAuthentication",
+]
+```
+
+No change to `AUTHENTICATION_BACKENDS` is needed. The feature stays off while `WORKLOAD_IDENTITY` is
+empty, so adding the class alone changes nothing.
+
+With the example below, a push from the `main` branch of `my-org/app` is granted the
+`file.filerepository_owner` role on the repository named `prod`, and nothing else. See the
+configuration reference at the end for every option.
+
+## Roles for asynchronous tasks
+
+Operations that dispatch a task, such as a sync, return a task the client polls. A workload identity
+request is not a database user, so it is not automatically granted a role on the tasks it creates.
+Grant a role carrying `core.view_task` when the CI needs to read its own tasks.
+
+## Configuration reference
+
+Every option of the `WORKLOAD_IDENTITY` setting, annotated:
+
+```python title="settings.py"
+WORKLOAD_IDENTITY = {
+    # How matching rules combine.
+    # "union" (default) collects the grants of every matching rule.
+    # "first-match" stops at the first matching rule.
+    "strategy": "union",
+
+    # One entry per trusted provider. The key is a name for your own reference.
+    "providers": {
+        "github": {
+            # Required. Expected "iss" claim. Selects the provider and is verified while decoding.
+            "issuer": "https://token.actions.githubusercontent.com",
+
+            # Required. URL of the provider's JWKS. Keys are fetched and cached.
+            "jwks_url": "https://token.actions.githubusercontent.com/.well-known/jwks",
+
+            # Required. Expected "aud" claim.
+            "audience": "https://pulp.example.com",
+
+            # Optional. Allowed signing algorithms. Default: ["RS256"].
+            "algorithms": ["RS256"],
+
+            # Rules are evaluated in order. Each maps claims to grants.
+            "rules": [
+                {
+                    # Claim name to expected value. Values support "*" globbing.
+                    # Every entry must match (AND). A missing claim never matches.
+                    "match": {"repository": "my-org/app", "ref": "refs/heads/main"},
+
+                    # Grants awarded when the rule matches.
+                    "grants": [
+                        {
+                            # Required. Name of a role that already exists in Pulp.
+                            # A role that does not exist confers no permission.
+                            "role": "file.filerepository_owner",
+
+                            # Required. Where the role applies. One of:
+                            #   {"type": "global"}                       everywhere
+                            #   {"type": "domain", "domain": "<name>"}   objects in a domain
+                            #   {"type": "object", "name": "<name>"}     one object by name
+                            #   {"type": "object", "prn": "<prn>"}       one object by PRN
+                            "scope": {"type": "object", "name": "prod"},
+                        },
+                    ],
+                },
+            ],
+        },
+    },
+}
+```
diff --git a/pulpcore/app/access_policy.py b/pulpcore/app/access_policy.py
index 6cb41c02b7b..a6175f993f1 100644
--- a/pulpcore/app/access_policy.py
+++ b/pulpcore/app/access_policy.py
@@ -15,6 +15,13 @@ class DefaultAccessPolicy(AccessPolicy):
     An AccessPolicy that takes default statements from the view(set).
     """
 
+    def get_user_group_values(self, user):
+        """Let a stateless principal supply its groups via ``group_names`` instead of the ORM."""
+        group_names = getattr(user, "group_names", None)
+        if group_names is not None:
+            return list(group_names)
+        return super().get_user_group_values(user)
+
     @classmethod
     def get_access_policy(cls, view):
         """
diff --git a/pulpcore/app/oidc/__init__.py b/pulpcore/app/oidc/__init__.py
deleted file mode 100644
index f5690aef4bc..00000000000
--- a/pulpcore/app/oidc/__init__.py
+++ /dev/null
@@ -1,5 +0,0 @@
-"""OIDC authentication for CI clients.
-
-A validated OIDC token becomes a stateless principal that carries grants computed per request
-from the ``OIDC_AUTH`` setting. Nothing is written to the role tables.
-"""
diff --git a/pulpcore/app/role_util.py b/pulpcore/app/role_util.py
index 52c9fbfb074..2bd319793a6 100644
--- a/pulpcore/app/role_util.py
+++ b/pulpcore/app/role_util.py
@@ -173,11 +173,10 @@ def get_objects_for_user(
     accept_domain_perms=True,
     accept_global_perms=True,
 ):
-    from pulpcore.app.oidc.principal import OIDCPrincipal
+    from pulpcore.app.workload_identity.principal import WorkloadIdentityPrincipal
 
-    if isinstance(user, OIDCPrincipal):
-        # Stateless OIDC principal: scope from its per-request grants, not the role tables.
-        from pulpcore.app.oidc.authz import grants_queryset
+    if isinstance(user, WorkloadIdentityPrincipal):
+        from pulpcore.app.workload_identity.authz import grants_queryset
 
         grants = user.grants
         if isinstance(perms, str):
diff --git a/pulpcore/app/settings.py b/pulpcore/app/settings.py
index 2f63cf95290..f901015d2e6 100644
--- a/pulpcore/app/settings.py
+++ b/pulpcore/app/settings.py
@@ -316,11 +316,8 @@
 AUTHENTICATION_JSON_HEADER_JQ_FILTER = ""
 AUTHENTICATION_JSON_HEADER_OPENAPI_SECURITY_SCHEME = {}
 
-# OIDC authentication for CI clients. Empty by default (feature off). See
-# pulpcore.app.oidc.config for the schema. Enable by adding
-# "pulpcore.app.oidc.authentication.OIDCAuthentication" to
-# REST_FRAMEWORK["DEFAULT_AUTHENTICATION_CLASSES"] and populating this setting.
-OIDC_AUTH = {}
+# Workload identity authentication for CI clients. Off while empty.
+WORKLOAD_IDENTITY = {}
 
 ALLOWED_IMPORT_PATHS = []
 
diff --git a/pulpcore/app/workload_identity/__init__.py b/pulpcore/app/workload_identity/__init__.py
new file mode 100644
index 00000000000..63efe9d54b2
--- /dev/null
+++ b/pulpcore/app/workload_identity/__init__.py
@@ -0,0 +1,5 @@
+"""Workload identity authentication for CI clients.
+
+A short-lived OIDC token from a third-party provider (for example GitHub Actions) becomes a
+stateless principal whose grants are computed per request from the ``WORKLOAD_IDENTITY`` setting.
+"""
diff --git a/pulpcore/app/oidc/authentication.py b/pulpcore/app/workload_identity/authentication.py
similarity index 67%
rename from pulpcore/app/oidc/authentication.py
rename to pulpcore/app/workload_identity/authentication.py
index 65dd49ce11f..64a9b3c5d3a 100644
--- a/pulpcore/app/oidc/authentication.py
+++ b/pulpcore/app/workload_identity/authentication.py
@@ -1,12 +1,7 @@
-"""DRF authentication that validates third-party OIDC tokens.
+"""DRF authentication that validates a third-party OIDC token against its provider's JWKS.
 
-This authenticator accepts an OIDC token issued by a configured third-party
-provider (for example a GitHub Actions workflow token), verifies it against the
-provider's JWKS, maps its claims to grants and returns a stateless
-``OIDCPrincipal``. No database user is involved.
-
-The token may arrive either as a ``Bearer`` token or inside a ``Basic`` header
-(the way ``docker login`` passes a token, where the password field carries it).
+The token arrives as a ``Bearer`` token or as the password in a ``Basic`` header (``docker login``).
+On success its claims map to grants and a stateless ``WorkloadIdentityPrincipal`` is returned.
 """
 
 import base64
@@ -17,16 +12,16 @@
 from rest_framework.authentication import BaseAuthentication
 from rest_framework.exceptions import AuthenticationFailed
 
-from pulpcore.app.oidc import config, rules
-from pulpcore.app.oidc.principal import OIDCPrincipal
+from pulpcore.app.workload_identity import config, rules
+from pulpcore.app.workload_identity.principal import WorkloadIdentityPrincipal
 
-_logger = logging.getLogger("pulpcore.oidc")
+_logger = logging.getLogger("pulpcore.workload_identity")
 
 
-class OIDCAuthentication(BaseAuthentication):
+class WorkloadIdentityAuthentication(BaseAuthentication):
     """Authenticate requests bearing a third-party OIDC token.
 
-    On success this returns a stateless ``OIDCPrincipal`` whose permissions are
+    On success this returns a stateless ``WorkloadIdentityPrincipal`` whose permissions are
     derived entirely from the grants earned by the token's claims. When the
     request carries no token, or a token that is not meant for us, the
     authenticator returns ``None`` so that other authenticators may run.
@@ -49,19 +44,16 @@ def _get_token(self, request):
                 return None
             if ":" not in decoded:
                 return None
-            # docker login passes the token as the password; the username is ignored.
             _, _, password = decoded.partition(":")
             return password
         return None
 
     def authenticate(self, request):
-        """Validate an OIDC token and return ``(OIDCPrincipal, claims)``, or ``None`` if not ours."""
+        """Validate an OIDC token and return ``(WorkloadIdentityPrincipal, claims)``, or ``None`` if not ours."""
         token = self._get_token(request)
         if not token:
             return None
 
-        # Peek at the issuer without verifying the signature. If this is not a
-        # JWT at all, it is not meant for us.
         try:
             unverified = jwt.decode(token, options={"verify_signature": False})
         except jwt.PyJWTError:
@@ -72,7 +64,6 @@ def authenticate(self, request):
         if provider is None:
             return None
 
-        # Verify the token for real against the provider's JWKS.
         try:
             signing_key = config.jwks_client(provider).get_signing_key_from_jwt(token)
             claims = jwt.decode(
@@ -96,9 +87,7 @@ def authenticate(self, request):
             )
             raise AuthenticationFailed("No matching OIDC rule.")
 
-        # The username is intentionally empty: the container registry token
-        # subject must be empty for a principal with no database user.
-        return (OIDCPrincipal(grants, username=""), claims)
+        return (WorkloadIdentityPrincipal(grants, username=""), claims)
 
     def authenticate_header(self, request):
         """Return the ``WWW-Authenticate`` value so failures are 401, not 403."""
diff --git a/pulpcore/app/oidc/authz.py b/pulpcore/app/workload_identity/authz.py
similarity index 69%
rename from pulpcore/app/oidc/authz.py
rename to pulpcore/app/workload_identity/authz.py
index 9a10eef9618..c6a28a0c26b 100644
--- a/pulpcore/app/oidc/authz.py
+++ b/pulpcore/app/workload_identity/authz.py
@@ -6,9 +6,7 @@
 * ``{"type": "domain", "domain": "<name>"}``
 * ``{"type": "object", "name": "<name>"}`` (or ``"prn"``)
 
-This module is used both by the principal (single-object ``has_perm`` and the model-level
-permission set) and by ``get_objects_for_user`` (list filtering). Roles are read from the
-database to resolve their permissions; the grant assignment itself is never stored.
+Roles are read from the database to resolve their permissions; the grant assignment is never stored.
 """
 
 from django.db.models import Q
@@ -48,11 +46,19 @@ def _scope_matches(scope, obj):
         domain = getattr(obj, "pulp_domain", None)
         return domain is not None and domain.name == scope.get("domain")
     if stype == "object":
-        checks = (("prn", "prn"), ("name", "name"))
-        provided = [key for key, _ in checks if key in scope]
-        if not provided:
+        if "prn" not in scope and "name" not in scope:
             return False
-        return all(str(getattr(obj, attr, None)) == str(scope[key]) for key, attr in checks if key in scope)
+        if "prn" in scope:
+            from pulpcore.app.util import get_prn
+
+            try:
+                if get_prn(obj) != scope["prn"]:
+                    return False
+            except Exception:
+                return False
+        if "name" in scope and str(getattr(obj, "name", None)) != str(scope["name"]):
+            return False
+        return True
     return False
 
 
@@ -67,15 +73,26 @@ def has_grant_perm(grants, permission, obj=None):
     return False
 
 
-def permissions_for(grants):
-    """The set of ``app_label.codename`` the grants confer, for model-level ``get_all_permissions``."""
+def permissions_for(grants, obj=None):
+    """The set of ``app_label.codename`` the grants confer, scoped to ``obj`` when given."""
     from pulpcore.app.models.role import Role
 
     names = {g.get("role") for g in grants if g.get("role")}
-    perms = set()
+    if not names:
+        return set()
+    role_perms = {}
     for role in Role.objects.filter(name__in=names).prefetch_related("permissions__content_type"):
-        for perm in role.permissions.all():
-            perms.add(f"{perm.content_type.app_label}.{perm.codename}")
+        role_perms[role.name] = {
+            f"{perm.content_type.app_label}.{perm.codename}" for perm in role.permissions.all()
+        }
+    perms = set()
+    for grant in grants:
+        conferred = role_perms.get(grant.get("role"))
+        if not conferred:
+            continue
+        if obj is not None and not _scope_matches(grant.get("scope", {}), obj):
+            continue
+        perms |= conferred
     return perms
 
 
@@ -98,7 +115,12 @@ def grants_queryset(grants, permission, queryset):
             clause = Q(pulp_domain__name=scope.get("domain"))
         elif stype == "object":
             if "prn" in scope:
-                clause = Q(prn=scope["prn"])
+                from pulpcore.app.util import extract_pk
+
+                try:
+                    clause = Q(pk=extract_pk(scope["prn"], only_prn=True))
+                except Exception:
+                    clause = None
             elif "name" in scope:
                 clause = Q(name=scope["name"])
         if clause is not None:
diff --git a/pulpcore/app/oidc/config.py b/pulpcore/app/workload_identity/config.py
similarity index 82%
rename from pulpcore/app/oidc/config.py
rename to pulpcore/app/workload_identity/config.py
index 473d224088b..d72c5c4d818 100644
--- a/pulpcore/app/oidc/config.py
+++ b/pulpcore/app/workload_identity/config.py
@@ -1,4 +1,4 @@
-"""Access to the ``OIDC_AUTH`` setting and per-provider JWKS clients."""
+"""Access to the ``WORKLOAD_IDENTITY`` setting and per-provider JWKS clients."""
 
 import functools
 
@@ -7,7 +7,7 @@
 
 
 def config():
-    return getattr(settings, "OIDC_AUTH", {}) or {}
+    return getattr(settings, "WORKLOAD_IDENTITY", {}) or {}
 
 
 def strategy():
diff --git a/pulpcore/app/oidc/principal.py b/pulpcore/app/workload_identity/principal.py
similarity index 53%
rename from pulpcore/app/oidc/principal.py
rename to pulpcore/app/workload_identity/principal.py
index 34863556b34..be5fe2145f5 100644
--- a/pulpcore/app/oidc/principal.py
+++ b/pulpcore/app/workload_identity/principal.py
@@ -1,24 +1,14 @@
-"""A stateless principal used as ``request.user`` for OIDC-authenticated clients.
+"""A stateless ``request.user`` for workload-identity clients.
 
-``OIDCPrincipal`` deliberately does not subclass Django's user model or
-``PermissionsMixin``. It implements its own permission-checking methods so that
-Django never dispatches permission checks for it to ``AUTHENTICATION_BACKENDS``.
-Authorization is derived entirely from the list of grants carried on the object.
+It does not subclass the user model, and answers permission checks from its own grants so Django
+never dispatches them to ``AUTHENTICATION_BACKENDS``.
 """
 
-from pulpcore.app.oidc.authz import has_grant_perm, permissions_for
+from pulpcore.app.workload_identity.authz import has_grant_perm, permissions_for
 
 
-class OIDCPrincipal:
-    """A user-like object backed by OIDC grants rather than a database row.
-
-    This object is safe to assign to ``request.user``. It exposes the attributes
-    the request path reads and answers permission checks from its ``grants``.
-
-    Attributes:
-        grants (list): The list of grant dicts backing this principal.
-        username (str): The principal's username (may be empty).
-    """
+class WorkloadIdentityPrincipal:
+    """A user-like object backed by grants rather than a database row, safe as ``request.user``."""
 
     is_authenticated = True
     is_active = True
@@ -27,14 +17,9 @@ class OIDCPrincipal:
     is_staff = False
     pk = None
     id = None
+    group_names = ()
 
     def __init__(self, grants, username=""):
-        """Initialize the principal.
-
-        Args:
-            grants (list): A list of grant dicts (see ``authz`` for the shape).
-            username (str): The principal's username. Defaults to an empty string.
-        """
         self.grants = grants
         self.username = username
 
@@ -59,8 +44,11 @@ def has_module_perms(self, app_label):
         return any(perm.startswith(prefix) for perm in self.get_all_permissions())
 
     def get_all_permissions(self, obj=None):
-        """The set of ``app_label.codename`` permissions the grants confer."""
-        return permissions_for(self.grants)
+        """``app_label.codename`` at the model level, bare codenames for an object (Pulp's convention)."""
+        perms = permissions_for(self.grants, obj)
+        if obj is None:
+            return perms
+        return {perm.split(".", 1)[1] for perm in perms}
 
     def get_group_permissions(self, obj=None):
         """The permissions from groups; always empty for a principal."""
diff --git a/pulpcore/app/oidc/rules.py b/pulpcore/app/workload_identity/rules.py
similarity index 94%
rename from pulpcore/app/oidc/rules.py
rename to pulpcore/app/workload_identity/rules.py
index 0058ffcb5e6..fc0a498cebf 100644
--- a/pulpcore/app/oidc/rules.py
+++ b/pulpcore/app/workload_identity/rules.py
@@ -5,7 +5,7 @@
 
 import fnmatch
 
-from pulpcore.app.oidc.config import strategy
+from pulpcore.app.workload_identity.config import strategy
 
 
 def _matches(match_spec, claims):
diff --git a/pulpcore/tests/unit/oidc/__init__.py b/pulpcore/tests/unit/workload_identity/__init__.py
similarity index 100%
rename from pulpcore/tests/unit/oidc/__init__.py
rename to pulpcore/tests/unit/workload_identity/__init__.py
diff --git a/pulpcore/tests/unit/oidc/test_oidc.py b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
similarity index 88%
rename from pulpcore/tests/unit/oidc/test_oidc.py
rename to pulpcore/tests/unit/workload_identity/test_workload_identity.py
index 5017bde4efa..33ce57af165 100644
--- a/pulpcore/tests/unit/oidc/test_oidc.py
+++ b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
@@ -6,9 +6,9 @@
 from pulpcore.app.models.role import Role
 from pulpcore.app.role_util import get_objects_for_user
 
-from pulpcore.app.oidc.authz import grants_queryset, has_grant_perm, permissions_for
-from pulpcore.app.oidc.principal import OIDCPrincipal
-from pulpcore.app.oidc.rules import grants_for
+from pulpcore.app.workload_identity.authz import grants_queryset, has_grant_perm, permissions_for
+from pulpcore.app.workload_identity.principal import WorkloadIdentityPrincipal
+from pulpcore.app.workload_identity.rules import grants_for
 
 
 PROVIDER = {
@@ -33,13 +33,13 @@ def test_rules_no_match():
     assert grants_for(PROVIDER, {"repository": "other/x"}) == []
 
 
-@override_settings(OIDC_AUTH={"strategy": "union"})
+@override_settings(WORKLOAD_IDENTITY={"strategy": "union"})
 def test_rules_union_accumulates():
     grants = grants_for(PROVIDER, {"repository": "org/infra", "ref": "refs/heads/main"})
     assert {g["role"] for g in grants} == {"role1", "role2"}
 
 
-@override_settings(OIDC_AUTH={"strategy": "first-match"})
+@override_settings(WORKLOAD_IDENTITY={"strategy": "first-match"})
 def test_rules_first_match_stops():
     grants = grants_for(PROVIDER, {"repository": "org/infra", "ref": "refs/heads/main"})
     assert [g["role"] for g in grants] == ["role1"]
@@ -129,13 +129,13 @@ def test_grants_queryset_no_relevant_grant(db, repo_a):
 
 
 def test_principal_is_authenticated_and_has_perm(repo_viewer_role, repo_a):
-    principal = OIDCPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
+    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
     assert principal.is_authenticated is True
     assert principal.pk is None
     assert principal.has_perm("core.view_repository", repo_a) is True
 
 
 def test_get_objects_for_user_scopes_by_grants(repo_viewer_role, repo_a, repo_b):
-    principal = OIDCPrincipal([{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}])
+    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}])
     qs = get_objects_for_user(principal, "core.view_repository", Repository.objects.all())
     assert set(qs.values_list("name", flat=True)) == {"repo-a"}

From 4accbf49d157321ca1b88137acd6c35f447e2454 Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Mon, 13 Jul 2026 11:11:28 +0200
Subject: [PATCH 3/8] fix ruff format reports

---
 pulpcore/app/workload_identity/authz.py                       | 4 +---
 .../tests/unit/workload_identity/test_workload_identity.py    | 4 +++-
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/pulpcore/app/workload_identity/authz.py b/pulpcore/app/workload_identity/authz.py
index c6a28a0c26b..17548e4c19c 100644
--- a/pulpcore/app/workload_identity/authz.py
+++ b/pulpcore/app/workload_identity/authz.py
@@ -26,9 +26,7 @@ def _role_has_perm(role_name, app_label, codename):
         role = Role.objects.get(name=role_name)
     except Role.DoesNotExist:
         return False
-    return role.permissions.filter(
-        content_type__app_label=app_label, codename=codename
-    ).exists()
+    return role.permissions.filter(content_type__app_label=app_label, codename=codename).exists()
 
 
 def _scope_matches(scope, obj):
diff --git a/pulpcore/tests/unit/workload_identity/test_workload_identity.py b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
index 33ce57af165..f7391c278b6 100644
--- a/pulpcore/tests/unit/workload_identity/test_workload_identity.py
+++ b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
@@ -136,6 +136,8 @@ def test_principal_is_authenticated_and_has_perm(repo_viewer_role, repo_a):
 
 
 def test_get_objects_for_user_scopes_by_grants(repo_viewer_role, repo_a, repo_b):
-    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}])
+    principal = WorkloadIdentityPrincipal(
+        [{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}]
+    )
     qs = get_objects_for_user(principal, "core.view_repository", Repository.objects.all())
     assert set(qs.values_list("name", flat=True)) == {"repo-a"}

From 2c4aeb516b8336914b89461c23146cb97ad1f0a5 Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Mon, 13 Jul 2026 11:45:01 +0200
Subject: [PATCH 4/8] fix lint and update documentation format

---
 docs/admin/guides/auth/workload_identity.md   | 53 +++++++++++--------
 .../test_workload_identity.py                 |  2 -
 2 files changed, 31 insertions(+), 24 deletions(-)

diff --git a/docs/admin/guides/auth/workload_identity.md b/docs/admin/guides/auth/workload_identity.md
index 01cb87c95fe..9f80b639517 100644
--- a/docs/admin/guides/auth/workload_identity.md
+++ b/docs/admin/guides/auth/workload_identity.md
@@ -1,29 +1,35 @@
 # Workload Identity Authentication
 
-A CI job can authenticate to Pulp with a short-lived OIDC token from a third-party provider (for
-example GitHub Actions) instead of a stored username and password. The token is verified against the
-provider's public keys, its claims are matched against a set of rules, and the request is granted
-roles for that request only. No user is created and nothing is written to the role tables.
+A CI job can authenticate to Pulp with a short-lived OIDC token from a third-party provider (for example GitHub Actions),
+instead of a stored username and password.
+The token is verified against the provider's public keys,
+its claims are matched against a set of rules,
+and the request is granted roles for that request only.
+No user is created and nothing is written to the role tables.
 
-This suits supply-chain workflows where a pipeline pushes content and you want its permissions scoped
-to specific repositories without long-lived secrets.
+This suits supply-chain workflows where a pipeline pushes content
+and you want its permissions scoped to specific repositories without long-lived secrets.
 
 !!! note
-    The token is an OIDC token, but this is unrelated to the user-facing SSO login covered in
-    [Using external service](external.md). It identifies a workload, not a person.
+    The token is an OIDC token,
+    but this is unrelated to the user-facing SSO login covered in [Using external service](external.md).
+    It identifies a workload, not a person.
 
 ## How it works
 
-On each request the token is read from the `Authorization` header, either as a `Bearer` token or as
-the password of a `Basic` header (the way `docker login` sends a token). The `iss` claim selects a
-configured provider, the signature is verified against the provider's JWKS, and `iss`, `aud` and
-`exp` are checked. The remaining claims are matched against the provider's rules to compute the roles
-and scopes for the request. A token that matches no rule is rejected with a 401.
+On each request the token is read from the `Authorization` header,
+either as a `Bearer` token or as the password of a `Basic` header (the way `docker login` sends a token).
+The `iss` claim selects a configured provider,
+the signature is verified against the provider's JWKS,
+and `iss`, `aud` and `exp` are checked.
+The remaining claims are matched against the provider's rules to compute the roles and scopes for the request.
+A token that matches no rule is rejected with a 401.
 
 ## Enabling
 
-Add the authentication class to `DEFAULT_AUTHENTICATION_CLASSES`, before `BasicAuthentication` so the
-`docker login` path reaches it, then populate `WORKLOAD_IDENTITY`:
+Add the authentication class to `DEFAULT_AUTHENTICATION_CLASSES`,
+before `BasicAuthentication` so the `docker login` path reaches it,
+then populate `WORKLOAD_IDENTITY`:
 
 ```python title="settings.py"
 REST_FRAMEWORK["DEFAULT_AUTHENTICATION_CLASSES"] = [
@@ -33,17 +39,20 @@ REST_FRAMEWORK["DEFAULT_AUTHENTICATION_CLASSES"] = [
 ]
 ```
 
-No change to `AUTHENTICATION_BACKENDS` is needed. The feature stays off while `WORKLOAD_IDENTITY` is
-empty, so adding the class alone changes nothing.
+No change to `AUTHENTICATION_BACKENDS` is needed.
+The feature stays off while `WORKLOAD_IDENTITY` is empty,
+so adding the class alone changes nothing.
 
-With the example below, a push from the `main` branch of `my-org/app` is granted the
-`file.filerepository_owner` role on the repository named `prod`, and nothing else. See the
-configuration reference at the end for every option.
+With the example below,
+a push from the `main` branch of `my-org/app` is granted the `file.filerepository_owner` role on the repository named `prod`,
+and nothing else.
+See the configuration reference at the end for every option.
 
 ## Roles for asynchronous tasks
 
-Operations that dispatch a task, such as a sync, return a task the client polls. A workload identity
-request is not a database user, so it is not automatically granted a role on the tasks it creates.
+Operations that dispatch a task, such as a sync, return a task the client polls.
+A workload identity request is not a database user,
+so it is not automatically granted a role on the tasks it creates.
 Grant a role carrying `core.view_task` when the CI needs to read its own tasks.
 
 ## Configuration reference
diff --git a/pulpcore/tests/unit/workload_identity/test_workload_identity.py b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
index f7391c278b6..a7d97303948 100644
--- a/pulpcore/tests/unit/workload_identity/test_workload_identity.py
+++ b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
@@ -5,12 +5,10 @@
 from pulpcore.app.models import Repository
 from pulpcore.app.models.role import Role
 from pulpcore.app.role_util import get_objects_for_user
-
 from pulpcore.app.workload_identity.authz import grants_queryset, has_grant_perm, permissions_for
 from pulpcore.app.workload_identity.principal import WorkloadIdentityPrincipal
 from pulpcore.app.workload_identity.rules import grants_for
 
-
 PROVIDER = {
     "issuer": "https://issuer",
     "rules": [

From cd380ef900c17384a010ae0e3306e7424dfaac9c Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Mon, 13 Jul 2026 12:00:51 +0200
Subject: [PATCH 5/8] better basic auth support

---
 docs/admin/guides/auth/workload_identity.md   |  4 +--
 pulpcore/app/checks.py                        | 30 +++++++++++++++++++
 .../app/workload_identity/authentication.py   | 16 ++++++----
 pulpcore/app/workload_identity/config.py      |  5 ++++
 pulpcore/app/workload_identity/principal.py   |  4 +--
 5 files changed, 49 insertions(+), 10 deletions(-)

diff --git a/docs/admin/guides/auth/workload_identity.md b/docs/admin/guides/auth/workload_identity.md
index 9f80b639517..c2dd5996649 100644
--- a/docs/admin/guides/auth/workload_identity.md
+++ b/docs/admin/guides/auth/workload_identity.md
@@ -17,8 +17,7 @@ and you want its permissions scoped to specific repositories without long-lived
 
 ## How it works
 
-On each request the token is read from the `Authorization` header,
-either as a `Bearer` token or as the password of a `Basic` header (the way `docker login` sends a token).
+On each request the token is read from the `Authorization: Bearer` header.
 The `iss` claim selects a configured provider,
 the signature is verified against the provider's JWKS,
 and `iss`, `aud` and `exp` are checked.
@@ -28,7 +27,6 @@ A token that matches no rule is rejected with a 401.
 ## Enabling
 
 Add the authentication class to `DEFAULT_AUTHENTICATION_CLASSES`,
-before `BasicAuthentication` so the `docker login` path reaches it,
 then populate `WORKLOAD_IDENTITY`:
 
 ```python title="settings.py"
diff --git a/pulpcore/app/checks.py b/pulpcore/app/checks.py
index 3920cad44fc..dee19f85a4a 100644
--- a/pulpcore/app/checks.py
+++ b/pulpcore/app/checks.py
@@ -123,3 +123,33 @@ def check_artifact_checksums(app_configs, **kwargs):
         )
 
     return messages
+
+
+@register(deploy=True)
+def workload_identity_reserved_username(app_configs, **kwargs):
+    from pulpcore.app.workload_identity import config
+
+    messages = []
+    if not config.config():
+        return messages
+
+    username = config.basic_username()
+    try:
+        from django.contrib.auth import get_user_model
+
+        collides = get_user_model().objects.filter(username=username).exists()
+    except Exception:
+        return messages
+
+    if collides:
+        messages.append(
+            CheckWarning(
+                f"The WORKLOAD_IDENTITY basic_auth_username '{username}' is also a database user. "
+                "A token presented with this username over Basic auth is validated as a workload "
+                "identity token, not as that user's password. Set basic_auth_username to a name "
+                "that is not a real user to avoid ambiguity.",
+                id="pulpcore.W006",
+            )
+        )
+
+    return messages
diff --git a/pulpcore/app/workload_identity/authentication.py b/pulpcore/app/workload_identity/authentication.py
index 64a9b3c5d3a..7b3c9c9421d 100644
--- a/pulpcore/app/workload_identity/authentication.py
+++ b/pulpcore/app/workload_identity/authentication.py
@@ -1,7 +1,8 @@
 """DRF authentication that validates a third-party OIDC token against its provider's JWKS.
 
-The token arrives as a ``Bearer`` token or as the password in a ``Basic`` header (``docker login``).
-On success its claims map to grants and a stateless ``WorkloadIdentityPrincipal`` is returned.
+The token arrives as a ``Bearer`` token, or as the password of a ``Basic`` header whose username
+is the reserved workload-identity name. On success its claims map to grants and a stateless
+``WorkloadIdentityPrincipal`` is returned.
 """
 
 import base64
@@ -28,7 +29,12 @@ class WorkloadIdentityAuthentication(BaseAuthentication):
     """
 
     def _get_token(self, request):
-        """Return the token from the Authorization header (Bearer, or Basic password), or None."""
+        """Return the token from the Authorization header, or None.
+
+        Accepts a ``Bearer`` token, or a token carried as the password of a ``Basic`` header whose
+        username is the reserved workload-identity name. Any other ``Basic`` header is left for the
+        regular authenticators.
+        """
         header = request.META.get("HTTP_AUTHORIZATION", "")
         parts = header.split()
         if len(parts) != 2:
@@ -42,9 +48,9 @@ def _get_token(self, request):
                 decoded = base64.b64decode(value).decode("utf-8")
             except (binascii.Error, ValueError, UnicodeDecodeError):
                 return None
-            if ":" not in decoded:
+            username, sep, password = decoded.partition(":")
+            if not sep or username != config.basic_username():
                 return None
-            _, _, password = decoded.partition(":")
             return password
         return None
 
diff --git a/pulpcore/app/workload_identity/config.py b/pulpcore/app/workload_identity/config.py
index d72c5c4d818..c443ec91e56 100644
--- a/pulpcore/app/workload_identity/config.py
+++ b/pulpcore/app/workload_identity/config.py
@@ -18,6 +18,11 @@ def providers():
     return config().get("providers", {}) or {}
 
 
+def basic_username():
+    """The reserved username that flags a token carried in the ``Basic`` password field."""
+    return config().get("basic_auth_username", "workload-identity")
+
+
 def provider_for_issuer(issuer):
     """Return the provider entry whose ``issuer`` matches, or ``None``."""
     for entry in providers().values():
diff --git a/pulpcore/app/workload_identity/principal.py b/pulpcore/app/workload_identity/principal.py
index be5fe2145f5..8a8a6d99410 100644
--- a/pulpcore/app/workload_identity/principal.py
+++ b/pulpcore/app/workload_identity/principal.py
@@ -55,5 +55,5 @@ def get_group_permissions(self, obj=None):
         return set()
 
     def __str__(self):
-        """The username, or ``"oidc-principal"`` when it is empty."""
-        return self.username or "oidc-principal"
+        """The username, or ``"workload-identity"`` when it is empty."""
+        return self.username or "workload-identity"

From 35e29b2989593fdffe2c93416ea16db969e2a64b Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Mon, 13 Jul 2026 12:51:08 +0200
Subject: [PATCH 6/8] Derive groups from group_names

---
 pulpcore/app/access_policy.py               | 9 +++++----
 pulpcore/app/workload_identity/principal.py | 4 ++--
 2 files changed, 7 insertions(+), 6 deletions(-)

diff --git a/pulpcore/app/access_policy.py b/pulpcore/app/access_policy.py
index a6175f993f1..e3696ae34bb 100644
--- a/pulpcore/app/access_policy.py
+++ b/pulpcore/app/access_policy.py
@@ -16,10 +16,11 @@ class DefaultAccessPolicy(AccessPolicy):
     """
 
     def get_user_group_values(self, user):
-        """Let a stateless principal supply its groups via ``group_names`` instead of the ORM."""
-        group_names = getattr(user, "group_names", None)
-        if group_names is not None:
-            return list(group_names)
+        """Let a stateless principal supply its groups directly instead of via the ORM."""
+        from pulpcore.app.workload_identity.principal import WorkloadIdentityPrincipal
+
+        if isinstance(user, WorkloadIdentityPrincipal):
+            return list(user.group_names)
         return super().get_user_group_values(user)
 
     @classmethod
diff --git a/pulpcore/app/workload_identity/principal.py b/pulpcore/app/workload_identity/principal.py
index 8a8a6d99410..19b0b5209ea 100644
--- a/pulpcore/app/workload_identity/principal.py
+++ b/pulpcore/app/workload_identity/principal.py
@@ -25,10 +25,10 @@ def __init__(self, grants, username=""):
 
     @property
     def groups(self):
-        """An empty ``Group`` queryset so ``user.groups.all()`` never crashes."""
+        """The ``Group`` objects named by ``group_names`` (empty by default)."""
         from pulpcore.app.models import Group
 
-        return Group.objects.none()
+        return Group.objects.filter(name__in=self.group_names)
 
     def has_perm(self, perm, obj=None):
         """Whether the grants confer ``perm`` (optionally scoped to ``obj``)."""

From 970726077b39a552baa42b7a5f90d2c30c645d78 Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Mon, 13 Jul 2026 13:54:24 +0200
Subject: [PATCH 7/8] Multitenant support

---
 docs/admin/guides/auth/workload_identity.md | 18 +++++--
 pulpcore/app/checks.py                      | 60 +++++++++++++++++++++
 pulpcore/app/workload_identity/authz.py     |  6 +++
 3 files changed, 80 insertions(+), 4 deletions(-)

diff --git a/docs/admin/guides/auth/workload_identity.md b/docs/admin/guides/auth/workload_identity.md
index c2dd5996649..66e7e78ab64 100644
--- a/docs/admin/guides/auth/workload_identity.md
+++ b/docs/admin/guides/auth/workload_identity.md
@@ -53,6 +53,13 @@ A workload identity request is not a database user,
 so it is not automatically granted a role on the tasks it creates.
 Grant a role carrying `core.view_task` when the CI needs to read its own tasks.
 
+## Domains
+
+When `DOMAIN_ENABLED` is on, scope an object grant to a single tenant:
+use a `prn`, or add a `domain` to a `name` scope.
+A bare `name` matches that name in every domain, which breaks the isolation between domains.
+Pulp raises a startup check warning when a name scope is left unqualified while domains are enabled.
+
 ## Configuration reference
 
 Every option of the `WORKLOAD_IDENTITY` setting, annotated:
@@ -94,10 +101,13 @@ WORKLOAD_IDENTITY = {
                             "role": "file.filerepository_owner",
 
                             # Required. Where the role applies. One of:
-                            #   {"type": "global"}                       everywhere
-                            #   {"type": "domain", "domain": "<name>"}   objects in a domain
-                            #   {"type": "object", "name": "<name>"}     one object by name
-                            #   {"type": "object", "prn": "<prn>"}       one object by PRN
+                            #   {"type": "global"}                                    everywhere
+                            #   {"type": "domain", "domain": "<name>"}                every object in a domain
+                            #   {"type": "object", "name": "<name>"}                  one object by name
+                            #   {"type": "object", "name": "<name>", "domain": "<d>"} one object by name in a domain
+                            #   {"type": "object", "prn": "<prn>"}                    one object by PRN (domain-safe)
+                            # With DOMAIN_ENABLED, qualify a name scope with a domain (or use prn):
+                            # a bare name otherwise matches that name in every domain.
                             "scope": {"type": "object", "name": "prod"},
                         },
                     ],
diff --git a/pulpcore/app/checks.py b/pulpcore/app/checks.py
index dee19f85a4a..93725835f45 100644
--- a/pulpcore/app/checks.py
+++ b/pulpcore/app/checks.py
@@ -153,3 +153,63 @@ def workload_identity_reserved_username(app_configs, **kwargs):
         )
 
     return messages
+
+
+@register(deploy=True)
+def workload_identity_domain_scopes(app_configs, **kwargs):
+    from pulpcore.app.workload_identity import config
+
+    messages = []
+    if settings.DOMAIN_ENABLED or not config.config():
+        return messages
+
+    uses_domain = any(
+        grant.get("scope", {}).get("type") == "domain" or "domain" in grant.get("scope", {})
+        for provider in config.providers().values()
+        for rule in provider.get("rules", [])
+        for grant in rule.get("grants", [])
+    )
+    if uses_domain:
+        messages.append(
+            CheckWarning(
+                "WORKLOAD_IDENTITY has grant scopes that reference a domain, but DOMAIN_ENABLED is "
+                "False. Domain scoping has no effect while domains are disabled.",
+                id="pulpcore.W007",
+            )
+        )
+
+    return messages
+
+
+@register(deploy=True)
+def workload_identity_unqualified_name_scopes(app_configs, **kwargs):
+    from pulpcore.app.workload_identity import config
+
+    messages = []
+    if not settings.DOMAIN_ENABLED or not config.config():
+        return messages
+
+    risky = False
+    for provider in config.providers().values():
+        for rule in provider.get("rules", []):
+            for grant in rule.get("grants", []):
+                scope = grant.get("scope", {})
+                if (
+                    scope.get("type") == "object"
+                    and "name" in scope
+                    and "domain" not in scope
+                    and "prn" not in scope
+                ):
+                    risky = True
+
+    if risky:
+        messages.append(
+            CheckWarning(
+                "WORKLOAD_IDENTITY has object scopes matched by name only while DOMAIN_ENABLED is "
+                "True. A bare name matches that object in every domain, breaking domain isolation. "
+                "Add a 'domain' to the scope or use a 'prn'.",
+                id="pulpcore.W008",
+            )
+        )
+
+    return messages
diff --git a/pulpcore/app/workload_identity/authz.py b/pulpcore/app/workload_identity/authz.py
index 17548e4c19c..9098aec54ae 100644
--- a/pulpcore/app/workload_identity/authz.py
+++ b/pulpcore/app/workload_identity/authz.py
@@ -56,6 +56,10 @@ def _scope_matches(scope, obj):
                 return False
         if "name" in scope and str(getattr(obj, "name", None)) != str(scope["name"]):
             return False
+        if "domain" in scope:
+            domain = getattr(obj, "pulp_domain", None)
+            if domain is None or domain.name != scope["domain"]:
+                return False
         return True
     return False
 
@@ -121,6 +125,8 @@ def grants_queryset(grants, permission, queryset):
                     clause = None
             elif "name" in scope:
                 clause = Q(name=scope["name"])
+                if has_domain and "domain" in scope:
+                    clause &= Q(pulp_domain__name=scope["domain"])
         if clause is not None:
             predicate = clause if predicate is None else (predicate | clause)
 

From 60e5109e3513e18f99dd053fed14d1ec089da89f Mon Sep 17 00:00:00 2001
From: baptiste <BaptisteCentreon@users.noreply.github.com>
Date: Mon, 13 Jul 2026 14:11:07 +0200
Subject: [PATCH 8/8] Unit tests

---
 .../test_workload_identity.py                 | 489 +++++++++++++++++-
 1 file changed, 486 insertions(+), 3 deletions(-)

diff --git a/pulpcore/tests/unit/workload_identity/test_workload_identity.py b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
index a7d97303948..634bd097977 100644
--- a/pulpcore/tests/unit/workload_identity/test_workload_identity.py
+++ b/pulpcore/tests/unit/workload_identity/test_workload_identity.py
@@ -1,11 +1,34 @@
+import base64
+import time
+from types import SimpleNamespace
+
+import jwt
 import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Permission
 from django.test import override_settings
-
-from pulpcore.app.models import Repository
+from rest_framework.exceptions import AuthenticationFailed
+
+from pulpcore.app.access_policy import DefaultAccessPolicy
+from pulpcore.app.checks import (
+    workload_identity_domain_scopes,
+    workload_identity_reserved_username,
+    workload_identity_unqualified_name_scopes,
+)
+from pulpcore.app.models import Domain, Group, Repository
 from pulpcore.app.models.role import Role
 from pulpcore.app.role_util import get_objects_for_user
-from pulpcore.app.workload_identity.authz import grants_queryset, has_grant_perm, permissions_for
+from pulpcore.app.util import get_prn
+from pulpcore.app.workload_identity import config as wi_config
+from pulpcore.app.workload_identity.authentication import WorkloadIdentityAuthentication
+from pulpcore.app.workload_identity.authz import (
+    _scope_matches,
+    grants_queryset,
+    has_grant_perm,
+    permissions_for,
+)
 from pulpcore.app.workload_identity.principal import WorkloadIdentityPrincipal
 from pulpcore.app.workload_identity.rules import grants_for
 
@@ -139,3 +162,463 @@ def test_get_objects_for_user_scopes_by_grants(repo_viewer_role, repo_a, repo_b)
     )
     qs = get_objects_for_user(principal, "core.view_repository", Repository.objects.all())
     assert set(qs.values_list("name", flat=True)) == {"repo-a"}
+
+
+# --- domain fixtures ---
+
+
+@pytest.fixture
+def domain_a(db):
+    return Domain.objects.create(
+        name="tenant-a",
+        storage_class="pulpcore.app.models.storage.FileSystem",
+        storage_settings={"base_path": "/foo"},
+    )
+
+
+@pytest.fixture
+def domain_b(db):
+    return Domain.objects.create(
+        name="tenant-b",
+        storage_class="pulpcore.app.models.storage.FileSystem",
+        storage_settings={"base_path": "/foo"},
+    )
+
+
+@pytest.fixture
+def prod_in_a(domain_a):
+    return Repository.objects.create(name="prod", pulp_domain=domain_a)
+
+
+@pytest.fixture
+def prod_in_b(domain_b):
+    return Repository.objects.create(name="prod", pulp_domain=domain_b)
+
+
+# --- domain scope ---
+
+
+def test_has_grant_perm_domain_scope(repo_viewer_role, prod_in_a, prod_in_b):
+    grants = [{"role": "repo_viewer", "scope": {"type": "domain", "domain": "tenant-a"}}]
+    assert has_grant_perm(grants, "core.view_repository", prod_in_a) is True
+    assert has_grant_perm(grants, "core.view_repository", prod_in_b) is False
+
+
+def test_grants_queryset_domain_scope(repo_viewer_role, prod_in_a, prod_in_b):
+    grants = [{"role": "repo_viewer", "scope": {"type": "domain", "domain": "tenant-a"}}]
+    qs = grants_queryset(grants, "core.view_repository", Repository.objects.filter(name="prod"))
+    assert set(qs.values_list("pulp_domain__name", flat=True)) == {"tenant-a"}
+
+
+# --- object scope: domain-qualified name (tenant isolation) ---
+
+
+def test_has_grant_perm_object_name_domain_qualified(repo_viewer_role, prod_in_a, prod_in_b):
+    grants = [
+        {"role": "repo_viewer", "scope": {"type": "object", "name": "prod", "domain": "tenant-a"}}
+    ]
+    assert has_grant_perm(grants, "core.view_repository", prod_in_a) is True
+    assert has_grant_perm(grants, "core.view_repository", prod_in_b) is False
+
+
+def test_grants_queryset_object_name_domain_qualified(repo_viewer_role, prod_in_a, prod_in_b):
+    grants = [
+        {"role": "repo_viewer", "scope": {"type": "object", "name": "prod", "domain": "tenant-a"}}
+    ]
+    qs = grants_queryset(grants, "core.view_repository", Repository.objects.filter(name="prod"))
+    assert set(qs.values_list("pulp_domain__name", flat=True)) == {"tenant-a"}
+
+
+def test_object_name_bare_matches_across_domains(repo_viewer_role, prod_in_a, prod_in_b):
+    # Documents the unqualified behaviour the pulpcore.W008 check warns about.
+    grants = [{"role": "repo_viewer", "scope": {"type": "object", "name": "prod"}}]
+    assert has_grant_perm(grants, "core.view_repository", prod_in_a) is True
+    assert has_grant_perm(grants, "core.view_repository", prod_in_b) is True
+
+
+# --- object scope: prn ---
+
+
+def test_has_grant_perm_prn_scope(repo_viewer_role, repo_a, repo_b):
+    grants = [{"role": "repo_viewer", "scope": {"type": "object", "prn": get_prn(repo_a)}}]
+    assert has_grant_perm(grants, "core.view_repository", repo_a) is True
+    assert has_grant_perm(grants, "core.view_repository", repo_b) is False
+
+
+def test_grants_queryset_prn_scope(repo_viewer_role, repo_a, repo_b):
+    grants = [{"role": "repo_viewer", "scope": {"type": "object", "prn": get_prn(repo_a)}}]
+    qs = grants_queryset(grants, "core.view_repository", Repository.objects.all())
+    assert set(qs.values_list("name", flat=True)) == {"repo-a"}
+
+
+# --- principal surface ---
+
+
+def test_principal_get_all_permissions_model_level(repo_viewer_role):
+    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
+    assert principal.get_all_permissions() == {"core.view_repository"}
+
+
+def test_principal_get_all_permissions_object_level(repo_viewer_role, repo_a):
+    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
+    assert principal.get_all_permissions(repo_a) == {"view_repository"}
+
+
+def test_principal_has_module_perms(repo_viewer_role):
+    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
+    assert principal.has_module_perms("core") is True
+    assert principal.has_module_perms("auth") is False
+
+
+def test_principal_groups_default_empty(db):
+    principal = WorkloadIdentityPrincipal([])
+    assert principal.group_names == ()
+    assert list(principal.groups) == []
+
+
+def test_principal_groups_from_group_names(db):
+    group = Group.objects.create(name="grp-a")
+    principal = WorkloadIdentityPrincipal([])
+    principal.group_names = ("grp-a",)
+    assert list(principal.groups) == [group]
+
+
+def test_principal_str():
+    assert str(WorkloadIdentityPrincipal([])) == "workload-identity"
+    assert str(WorkloadIdentityPrincipal([], username="ci")) == "ci"
+
+
+# --- access_policy.get_user_group_values override ---
+
+
+def test_get_user_group_values_principal_empty(db):
+    assert DefaultAccessPolicy().get_user_group_values(WorkloadIdentityPrincipal([])) == []
+
+
+def test_get_user_group_values_principal_reports_group_names(db):
+    principal = WorkloadIdentityPrincipal([])
+    principal.group_names = ("grp-a",)
+    assert DefaultAccessPolicy().get_user_group_values(principal) == ["grp-a"]
+
+
+def test_get_user_group_values_real_user_fallthrough(db):
+    user = get_user_model().objects.create(username="alice")
+    user.groups.add(Group.objects.create(name="grp-a"))
+    assert DefaultAccessPolicy().get_user_group_values(user) == ["grp-a"]
+
+
+# --- authentication._get_token gating ---
+
+
+class _FakeRequest:
+    def __init__(self, authorization):
+        self.META = {"HTTP_AUTHORIZATION": authorization} if authorization else {}
+
+
+def _basic(username, password):
+    return "Basic " + base64.b64encode(f"{username}:{password}".encode()).decode()
+
+
+def test_get_token_bearer():
+    auth = WorkloadIdentityAuthentication()
+    assert auth._get_token(_FakeRequest("Bearer abc.def.ghi")) == "abc.def.ghi"
+
+
+def test_get_token_basic_reserved_username():
+    auth = WorkloadIdentityAuthentication()
+    assert auth._get_token(_FakeRequest(_basic("workload-identity", "the-token"))) == "the-token"
+
+
+def test_get_token_basic_wrong_username_falls_through():
+    auth = WorkloadIdentityAuthentication()
+    assert auth._get_token(_FakeRequest(_basic("someuser", "the-token"))) is None
+
+
+def test_get_token_basic_no_colon():
+    auth = WorkloadIdentityAuthentication()
+    value = base64.b64encode(b"nocolon").decode()
+    assert auth._get_token(_FakeRequest("Basic " + value)) is None
+
+
+def test_get_token_malformed_header():
+    auth = WorkloadIdentityAuthentication()
+    assert auth._get_token(_FakeRequest("Bearer")) is None
+    assert auth._get_token(_FakeRequest("")) is None
+
+
+# --- config helpers ---
+
+
+@override_settings(WORKLOAD_IDENTITY={"providers": {"p": {"issuer": "https://issuer"}}})
+def test_config_provider_for_issuer():
+    assert wi_config.provider_for_issuer("https://issuer") == {"issuer": "https://issuer"}
+    assert wi_config.provider_for_issuer("https://other") is None
+
+
+@override_settings(WORKLOAD_IDENTITY={})
+def test_config_basic_username_default():
+    assert wi_config.basic_username() == "workload-identity"
+
+
+@override_settings(WORKLOAD_IDENTITY={"basic_auth_username": "ci-bot"})
+def test_config_basic_username_override():
+    assert wi_config.basic_username() == "ci-bot"
+
+
+@override_settings(WORKLOAD_IDENTITY={})
+def test_config_strategy_default():
+    assert wi_config.strategy() == "union"
+
+
+# --- deploy checks W006 / W007 / W008 ---
+
+
+@override_settings(WORKLOAD_IDENTITY={"basic_auth_username": "ci-bot"})
+def test_check_reserved_username_collision(db):
+    get_user_model().objects.create(username="ci-bot")
+    assert "pulpcore.W006" in [m.id for m in workload_identity_reserved_username(None)]
+
+
+@override_settings(WORKLOAD_IDENTITY={"basic_auth_username": "ci-bot"})
+def test_check_reserved_username_no_collision(db):
+    assert workload_identity_reserved_username(None) == []
+
+
+@override_settings(
+    DOMAIN_ENABLED=False,
+    WORKLOAD_IDENTITY={
+        "providers": {
+            "p": {
+                "rules": [{"grants": [{"role": "r", "scope": {"type": "domain", "domain": "d"}}]}]
+            }
+        }
+    },
+)
+def test_check_domain_scope_while_domains_off():
+    assert "pulpcore.W007" in [m.id for m in workload_identity_domain_scopes(None)]
+
+
+@override_settings(
+    DOMAIN_ENABLED=True,
+    WORKLOAD_IDENTITY={
+        "providers": {
+            "p": {
+                "rules": [{"grants": [{"role": "r", "scope": {"type": "object", "name": "prod"}}]}]
+            }
+        }
+    },
+)
+def test_check_unqualified_name_scope_while_domains_on():
+    assert "pulpcore.W008" in [m.id for m in workload_identity_unqualified_name_scopes(None)]
+
+
+@override_settings(
+    DOMAIN_ENABLED=True,
+    WORKLOAD_IDENTITY={
+        "providers": {
+            "p": {
+                "rules": [
+                    {
+                        "grants": [
+                            {
+                                "role": "r",
+                                "scope": {"type": "object", "name": "prod", "domain": "d"},
+                            }
+                        ]
+                    }
+                ]
+            }
+        }
+    },
+)
+def test_check_qualified_name_scope_no_warning():
+    assert workload_identity_unqualified_name_scopes(None) == []
+
+
+# --- authz: remaining branches ---
+
+
+def test_scope_matches_global_obj_none():
+    assert _scope_matches({"type": "global"}, None) is True
+
+
+def test_scope_matches_object_obj_none():
+    assert _scope_matches({"type": "object", "name": "x"}, None) is False
+
+
+def test_scope_matches_domain_on_domain_object(domain_a):
+    assert _scope_matches({"type": "domain", "domain": "tenant-a"}, domain_a) is True
+    assert _scope_matches({"type": "domain", "domain": "other"}, domain_a) is False
+
+
+def test_has_grant_perm_empty_role(db, repo_a):
+    grants = [{"role": "", "scope": {"type": "global"}}]
+    assert has_grant_perm(grants, "core.view_repository", repo_a) is False
+
+
+# --- get_objects_for_user: list-of-perms paths ---
+
+
+def test_get_objects_for_user_any_perm(repo_viewer_role, repo_a, repo_b):
+    principal = WorkloadIdentityPrincipal(
+        [{"role": "repo_viewer", "scope": {"type": "object", "name": "repo-a"}}]
+    )
+    qs = get_objects_for_user(
+        principal,
+        ["core.view_repository", "core.change_repository"],
+        Repository.objects.all(),
+        any_perm=True,
+    )
+    assert set(qs.values_list("name", flat=True)) == {"repo-a"}
+
+
+def test_get_objects_for_user_all_perms_intersection(repo_viewer_role, repo_a):
+    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
+    qs = get_objects_for_user(
+        principal,
+        ["core.view_repository", "core.change_repository"],
+        Repository.objects.all(),
+    )
+    # AND of a granted (view) and a non-granted (change) permission -> empty.
+    assert qs.count() == 0
+
+
+# --- principal: remaining surface ---
+
+
+def test_principal_has_perms(repo_viewer_role, repo_a):
+    principal = WorkloadIdentityPrincipal([{"role": "repo_viewer", "scope": {"type": "global"}}])
+    assert principal.has_perms(["core.view_repository"], repo_a) is True
+    assert principal.has_perms(["core.view_repository", "core.change_repository"], repo_a) is False
+
+
+def test_principal_get_group_permissions(db):
+    assert WorkloadIdentityPrincipal([]).get_group_permissions() == set()
+
+
+# --- authentication: _get_token edge + header ---
+
+
+def test_get_token_basic_non_utf8():
+    value = base64.b64encode(b"\xff\xfe").decode()
+    assert WorkloadIdentityAuthentication()._get_token(_FakeRequest("Basic " + value)) is None
+
+
+def test_authenticate_header():
+    assert WorkloadIdentityAuthentication().authenticate_header(_FakeRequest("")) == "Bearer"
+
+
+# --- authentication.authenticate: full JWT validation (RSA key + mocked JWKS) ---
+
+
+_ISSUER = "https://issuer"
+WI_AUTH = {
+    "providers": {
+        "p": {
+            "issuer": _ISSUER,
+            "jwks_url": "https://issuer/jwks",
+            "audience": "pulp",
+            "algorithms": ["RS256"],
+            "rules": [
+                {
+                    "match": {"repository": "org/app"},
+                    "grants": [{"role": "r", "scope": {"type": "global"}}],
+                }
+            ],
+        }
+    }
+}
+
+
+def _private_pem(key):
+    return key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    )
+
+
+@pytest.fixture(scope="module")
+def rsa_keys():
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    public_pem = key.public_key().public_bytes(
+        serialization.Encoding.PEM,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    return _private_pem(key), public_pem
+
+
+@pytest.fixture
+def mock_jwks(rsa_keys, monkeypatch):
+    _, public_pem = rsa_keys
+    fake = SimpleNamespace(get_signing_key_from_jwt=lambda token: SimpleNamespace(key=public_pem))
+    monkeypatch.setattr(wi_config, "jwks_client", lambda provider: fake)
+
+
+def _token(private_pem, **overrides):
+    now = int(time.time())
+    claims = {"iss": _ISSUER, "aud": "pulp", "iat": now, "exp": now + 300, "repository": "org/app"}
+    claims.update(overrides)
+    return jwt.encode(claims, private_pem, algorithm="RS256")
+
+
+def _bearer(token):
+    return _FakeRequest(f"Bearer {token}")
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_valid_token(rsa_keys, mock_jwks):
+    private_pem, _ = rsa_keys
+    principal, claims = WorkloadIdentityAuthentication().authenticate(_bearer(_token(private_pem)))
+    assert isinstance(principal, WorkloadIdentityPrincipal)
+    assert claims["repository"] == "org/app"
+    assert principal.grants == [{"role": "r", "scope": {"type": "global"}}]
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_bad_signature(rsa_keys, mock_jwks):
+    other_pem = _private_pem(rsa.generate_private_key(public_exponent=65537, key_size=2048))
+    with pytest.raises(AuthenticationFailed):
+        WorkloadIdentityAuthentication().authenticate(_bearer(_token(other_pem)))
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_wrong_audience(rsa_keys, mock_jwks):
+    private_pem, _ = rsa_keys
+    with pytest.raises(AuthenticationFailed):
+        WorkloadIdentityAuthentication().authenticate(_bearer(_token(private_pem, aud="wrong")))
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_expired(rsa_keys, mock_jwks):
+    private_pem, _ = rsa_keys
+    now = int(time.time())
+    with pytest.raises(AuthenticationFailed):
+        token = _token(private_pem, exp=now - 10, iat=now - 20)
+        WorkloadIdentityAuthentication().authenticate(_bearer(token))
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_unknown_issuer_returns_none(rsa_keys):
+    private_pem, _ = rsa_keys
+    request = _bearer(_token(private_pem, iss="https://evil"))
+    assert WorkloadIdentityAuthentication().authenticate(request) is None
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_no_matching_rule(rsa_keys, mock_jwks):
+    private_pem, _ = rsa_keys
+    with pytest.raises(AuthenticationFailed):
+        WorkloadIdentityAuthentication().authenticate(
+            _bearer(_token(private_pem, repository="other/x"))
+        )
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_not_a_jwt_returns_none():
+    assert WorkloadIdentityAuthentication().authenticate(_bearer("not-a-jwt")) is None
+
+
+@override_settings(WORKLOAD_IDENTITY=WI_AUTH)
+def test_authenticate_no_token_returns_none():
+    assert WorkloadIdentityAuthentication().authenticate(_FakeRequest("")) is None