Fix jq expression injection in user-supplied filters (CVE-2026-123084)

CHANGES/+cve-2026-123084.bugfix

@@ -0,0 +1 @@
+Fixed an expression language injection in user-supplied jq filters (content guards and JSON header authentication) that could expose the environment of the content process, including the Django secret key and the database password. This addresses CVE-2026-123084. Filters that reference env or $ENV now evaluate to an empty object and deny access; use EnvVarHeaderContentGuard for allow-listed environment variables.

pulpcore/app/authentication.py

@@ -1,10 +1,8 @@
 import json
 import logging
 from base64 import b64decode
-from binascii import Error as Base64DecodeError
 from gettext import gettext as _
 
-import jq
 from django.conf import settings
 from django.contrib.auth import authenticate
 from django.contrib.auth.backends import RemoteUserBackend
@@ -17,6 +15,8 @@
 )
 from rest_framework.exceptions import AuthenticationFailed
 
+from pulpcore.app.jq_utils import compile_safe_jq
+
 _logger = logging.getLogger(__name__)
 
 
@@ -61,18 +61,16 @@ def authenticate(self, request):
         try:
             header_content = request.META.get(self.header)
             header_decoded_content = b64decode(header_content)
-        except Base64DecodeError:
+        except ValueError:
             _logger.debug(_("Access not allowed - Header content is not Base64 encoded."))
             raise AuthenticationFailed(_("Access denied."))
 
         try:
             header_value = json.loads(header_decoded_content)
-            json_path = jq.compile(self.jq_filter)
-
-            remote_user = json_path.input_value(header_value).first()
-        except json.JSONDecodeError:
-            _logger.debug(_("Access not allowed - Invalid JSON."))
-            raise AuthenticationFailed(_("Access denied. Invalid JSON."))
+            remote_user = compile_safe_jq(self.jq_filter).input_value(header_value).first()
+        except (StopIteration, ValueError):
+            _logger.debug(_("Access not allowed - Invalid JSON or jq filter."))
+            raise AuthenticationFailed(_("Access denied."))
 
         if not remote_user:
             _logger.debug(_("Path or value not found."))

pulpcore/app/jq_utils.py

@@ -0,0 +1,19 @@
+"""Helpers for evaluating user-supplied jq filters safely."""
+
+import jq
+
+
+def compile_safe_jq(program):
+    """
+    Compile a user-supplied jq filter with the process environment hidden.
+
+    jq exposes the environment of the running process through the `env`
+    builtin and the `$ENV` variable. Filters may come from untrusted sources,
+    such as a content guard created by a domain administrator, so evaluating
+    them as-is can leak secrets held in the environment of the process, e.g. the
+    Django `SECRET_KEY` or the database password.
+
+    Shadowing both interfaces keeps the rest of the jq language available while
+    making the environment unreachable from the filter.
+    """
+    return jq.compile(f"def env: {{}}; {{}} as $ENV | ({program}\n)")

pulpcore/app/models/publication.py

@@ -10,7 +10,6 @@
 from gettext import gettext as _
 from urllib.parse import urljoin, urlparse
 
-import jq
 from aiohttp.web_exceptions import HTTPNotFound
 from django.conf import settings
 from django.contrib.postgres.fields import HStoreField
@@ -22,6 +21,7 @@
 from url_normalize import url_normalize
 
 from pulpcore.app.files import PulpTemporaryUploadedFile
+from pulpcore.app.jq_utils import compile_safe_jq
 from pulpcore.app.models import AutoAddObjPermsMixin
 from pulpcore.app.models.fields import RelativePathField
 from pulpcore.app.util import cache_key, get_domain_pk, get_url, retain_distributed_pub_enabled
@@ -538,19 +538,25 @@ def permit(self, request):
 
         try:
             header_decoded_content = b64decode(header_content)
-        except Base64DecodeError:
+        except ValueError:
             _logger.debug("Access not allowed - Header content is not Base64 encoded.")
             raise PermissionError(_("Access denied."))
 
         if self.jq_filter:
             try:
                 header_value = json.loads(header_decoded_content)
-                json_path = jq.compile(self.jq_filter)
-
-                header_value = json_path.input_value(header_value).first()
-
-            except json.JSONDecodeError:
-                _logger.debug("Access not allowed - Invalid JSON or Path not found.")
+            except ValueError:
+                _logger.debug("Access not allowed - Header content is not valid JSON.")
+                raise PermissionError(_("Access denied."))
+            try:
+                header_value = compile_safe_jq(self.jq_filter).input_value(header_value).first()
+            except StopIteration:
+                _logger.debug("Access not allowed - jq filter produced no result.")
+                raise PermissionError(_("Access denied."))
+            except ValueError as exc:
+                _logger.warning(
+                    "Access not allowed - Invalid jq filter %r: %s", self.jq_filter, exc
+                )
                 raise PermissionError(_("Access denied."))
         else:
             header_value = header_decoded_content.decode("utf8")

pulpcore/app/serializers/publication.py

@@ -6,6 +6,7 @@
 from rest_framework import serializers
 
 from pulpcore.app import models
+from pulpcore.app.jq_utils import compile_safe_jq
 from pulpcore.app.role_util import get_groups_with_perms, get_users_with_perms
 from pulpcore.app.serializers import (
     BaseURLField,
@@ -169,6 +170,14 @@ class HeaderContentGuardSerializer(ContentGuardSerializer, GetOrCreateSerializer
         required=False,
     )
 
+    def validate_jq_filter(self, value):
+        if value:
+            try:
+                compile_safe_jq(value)
+            except ValueError:
+                raise serializers.ValidationError(_("Invalid jq filter."))
+        return value
+
     class Meta(ContentGuardSerializer.Meta):
         model = models.HeaderContentGuard
         fields = ContentGuardSerializer.Meta.fields + ("header_name", "header_value", "jq_filter")

pulpcore/tests/unit/test_authentication.py

@@ -0,0 +1,60 @@
+from base64 import b64encode
+from unittest.mock import Mock, patch
+
+import pytest
+from rest_framework.exceptions import AuthenticationFailed
+
+from pulpcore.app.authentication import JSONHeaderRemoteAuthentication
+
+HEADER = "x-rh-identity"
+
+
+def get_auth(jq_filter):
+    auth = JSONHeaderRemoteAuthentication()
+    auth.header = HEADER
+    auth.jq_filter = jq_filter
+    return auth
+
+
+def get_request(payload):
+    request = Mock()
+    request.META = {HEADER: b64encode(payload)}
+    return request
+
+
+def test_environment_is_not_reachable(monkeypatch):
+    """A configured filter must not be able to read the environment of the API process."""
+
+    monkeypatch.setenv("PULP_TEST_SECRET", "topsecret")
+
+    with patch("pulpcore.app.authentication.authenticate") as authenticate:
+        assert get_auth("env.PULP_TEST_SECRET").authenticate(get_request(b'{"sub": "u"}')) is None
+
+    authenticate.assert_not_called()
+
+
+def test_invalid_filter_denies():
+    with pytest.raises(AuthenticationFailed):
+        get_auth("def broken:").authenticate(get_request(b'{"sub": "u"}'))
+
+
+def test_filter_without_result_denies():
+    with pytest.raises(AuthenticationFailed):
+        get_auth(".missing[]").authenticate(get_request(b'{"sub": "u"}'))
+
+
+def test_non_base64_header_denies():
+    request = Mock()
+    request.META = {HEADER: "é"}
+
+    with pytest.raises(AuthenticationFailed):
+        get_auth(".sub").authenticate(request)
+
+
+def test_valid_filter_authenticates():
+    user = Mock()
+
+    with patch("pulpcore.app.authentication.authenticate", return_value=user):
+        result = get_auth(".sub").authenticate(get_request(b'{"sub": "u"}'))
+
+    assert result == (user, None)

pulpcore/tests/unit/test_content_guard.py

@@ -12,7 +12,10 @@
     EnvVarHeaderContentGuard,
     HeaderContentGuard,
 )
-from pulpcore.app.serializers import EnvVarHeaderContentGuardSerializer
+from pulpcore.app.serializers import (
+    EnvVarHeaderContentGuardSerializer,
+    HeaderContentGuardSerializer,
+)
 
 ENVVAR_HEADER = "X-Test-Content-Guard-Header"
 ENVVAR_NAME = "ENVVAR_HEADER_GUARD_TEST_SECRET"
@@ -248,3 +251,136 @@ def test_envvar_header_content_guard_serializer_rejects_disallowed_env_var(db):
     serializer = EnvVarHeaderContentGuardSerializer()
     with pytest.raises(ValidationError):
         serializer.validate_env_var("DJANGO_SECRET_KEY")
+
+
+@pytest.mark.parametrize("jq_filter", ["env.PULP_TEST_SECRET", "$ENV.PULP_TEST_SECRET"])
+def test_header_content_guard_hides_environment(db, monkeypatch, jq_filter):
+    """A jq filter must not be able to read the environment of the content process."""
+
+    monkeypatch.setenv("PULP_TEST_SECRET", "topsecret")
+
+    content_guard = HeaderContentGuard(
+        name="header_guard",
+        header_name="x-header-name",
+        jq_filter=jq_filter,
+        header_value="topsecret",
+    )
+
+    request = Mock()
+    request.headers = {"x-header-name": b64encode(b"{}")}
+
+    with pytest.raises(PermissionError):
+        content_guard.permit(request)
+
+
+def test_header_content_guard_denies_per_character_environment_oracle(db, monkeypatch):
+    """The jq result must not work as a per-character allow/deny oracle for secrets."""
+
+    monkeypatch.setenv("PULP_TEST_SECRET", "supersecret")
+
+    def permits(jq_filter, guess, payload=b"{}"):
+        content_guard = HeaderContentGuard(
+            name="header_guard",
+            header_name="x-header-name",
+            header_value=guess,
+            jq_filter=jq_filter,
+        )
+        request = Mock()
+        request.headers = {"x-header-name": b64encode(payload)}
+        try:
+            content_guard.permit(request)
+        except PermissionError:
+            return False
+        return True
+
+    # Control: an honest filter still authorizes the matching value.
+    assert permits(".k", "v", b'{"k": "v"}')
+    assert not permits(".k", "x", b'{"k": "v"}')
+
+    for index, character in enumerate("supersecret"):
+        for jq_filter in (
+            f"env.PULP_TEST_SECRET | .[{index}:{index + 1}]",
+            f"$ENV.PULP_TEST_SECRET | .[{index}:{index + 1}]",
+        ):
+            assert not permits(jq_filter, character)
+
+
+def test_header_content_guard_supports_complex_jq_filters(db):
+    """Legitimate filters keep working, including pipes, select and interpolation."""
+
+    content_guard = HeaderContentGuard(
+        name="header_guard",
+        header_name="x-header-name",
+        jq_filter=r'.roles[] | select(.active) | "\(.name)"',
+        header_value="reader",
+    )
+
+    request = Mock()
+    payload = {"roles": [{"name": "reader", "active": True}]}
+    request.headers = {"x-header-name": b64encode(json.dumps(payload).encode())}
+
+    assert content_guard.permit(request) is None
+
+
+def test_header_content_guard_denies_when_filter_has_no_result(db):
+    """Filters that produce no result must deny access instead of raising an error."""
+
+    content_guard = HeaderContentGuard(
+        name="header_guard",
+        header_name="x-header-name",
+        jq_filter=".roles[] | select(.active == false) | .name",
+        header_value="reader",
+    )
+
+    request = Mock()
+    payload = {"roles": [{"name": "reader", "active": True}]}
+    request.headers = {"x-header-name": b64encode(json.dumps(payload).encode())}
+
+    with pytest.raises(PermissionError):
+        content_guard.permit(request)
+
+
+def test_header_content_guard_denies_invalid_filter(db):
+    """A broken stored filter denies instead of raising a server error."""
+
+    content_guard = HeaderContentGuard(
+        name="header_guard",
+        header_name="x-header-name",
+        jq_filter="def broken:",
+        header_value="v",
+    )
+
+    request = Mock()
+    request.headers = {"x-header-name": b64encode(b'{"k": "v"}')}
+
+    with pytest.raises(PermissionError):
+        content_guard.permit(request)
+
+
+def test_header_content_guard_denies_non_ascii_header(db):
+    """A non-ASCII header value must not raise a server error."""
+
+    content_guard = HeaderContentGuard(
+        name="header_guard",
+        header_name="x-header-name",
+        jq_filter=".k",
+        header_value="v",
+    )
+
+    request = Mock()
+    request.headers = {"x-header-name": "é"}
+
+    with pytest.raises(PermissionError):
+        content_guard.permit(request)
+
+
+def test_header_content_guard_serializer_rejects_invalid_filter(db):
+    """Invalid filters are rejected at write time."""
+
+    serializer = HeaderContentGuardSerializer()
+
+    with pytest.raises(ValidationError):
+        serializer.validate_jq_filter("def broken:")
+
+    assert serializer.validate_jq_filter(".k # comment") == ".k # comment"
+    assert serializer.validate_jq_filter(None) is None