Test use of PQC client certificates / CAs with Pulp + increase aiohttp default header size limits slightly

CHANGES/+header-too-large.bugfix

@@ -0,0 +1 @@
+Increased the content app's maximum HTTP header field size from 8190 to 16384 bytes to support PQC (post-quantum) X.509 certificates forwarded via the `X-CLIENT-CERT` header.

functest_requirements.txt

@@ -4,6 +4,7 @@ pytest-xdist
 python-gnupg
 proxy.py~=2.4.10
 trustme~=1.2.1
+cryptography>=49.0
 
 # pulp_file tests
 beautifulsoup4

pulp_certguard/tests/functional/api/test_x509_certguard.py

@@ -1,4 +1,5 @@
 import uuid
+from collections import namedtuple
 from urllib.parse import quote, urljoin
 
 import pytest
@@ -12,6 +13,11 @@
     X509_UNTRUSTED_CLIENT_CERT_FILE_PATH,
 )
 
+PQCGuardedDistribution = namedtuple(
+    "PQCGuardedDistribution",
+    ["distribution", "algorithm", "client_cert_pem", "untrusted_client_cert_pem"],
+)
+
 
 @pytest.fixture(scope="class")
 def x509_certguard_factory(x509_content_guards_api_client, gen_object_with_cleanup):
@@ -42,6 +48,31 @@ def x509_guarded_distribution(
     return distribution
 
 
+@pytest.fixture(scope="class")
+def pqc_guarded_distribution(
+    pqc_certificate_authority,
+    x509_content_guards_api_client,
+    gen_object_with_cleanup,
+    file_distribution_factory,
+    repository_test_file,
+):
+    pca = pqc_certificate_authority
+    content_guard = gen_object_with_cleanup(
+        x509_content_guards_api_client,
+        {
+            "name": str(uuid.uuid4()),
+            "ca_certificate": pca.ca_cert_pem,
+        },
+    )
+    distribution = file_distribution_factory(
+        repository=repository_test_file.pulp_href,
+        content_guard=content_guard.pulp_href,
+    )
+    return PQCGuardedDistribution(
+        distribution, pca.algorithm, pca.client_cert_pem, pca.untrusted_client_cert_pem
+    )
+
+
 @pytest.fixture(
     scope="module",
     params=[
@@ -92,3 +123,50 @@ def test_download(
             headers=cert_data and {"X-CLIENT-CERT": cert_data},
         )
         assert response.status_code == status_code
+
+
+class TestPQCX509CertGuard:
+    """Test X.509 content guard with PQC (ML-DSA) certificates.
+
+    Parameterized over ML-DSA-65 (~7.5KB, under the default 8190-byte header
+    limit) and ML-DSA-87 (~10KB, over the limit).
+    """
+
+    def test_download_with_valid_cert(
+        self,
+        pqc_guarded_distribution,
+        distribution_base_url,
+    ):
+        distribution = pqc_guarded_distribution.distribution
+        url = distribution_base_url(distribution.base_url)
+        cert_pem = quote(pqc_guarded_distribution.client_cert_pem)
+        response = requests.get(
+            urljoin(url, "test_file"),
+            headers={"X-CLIENT-CERT": cert_pem},
+            allow_redirects=False,
+        )
+        assert response.status_code in (200, 302)
+
+    def test_download_with_untrusted_cert(
+        self,
+        pqc_guarded_distribution,
+        distribution_base_url,
+    ):
+        distribution = pqc_guarded_distribution.distribution
+        url = distribution_base_url(distribution.base_url)
+        cert_pem = quote(pqc_guarded_distribution.untrusted_client_cert_pem)
+        response = requests.get(
+            urljoin(url, "test_file"),
+            headers={"X-CLIENT-CERT": cert_pem},
+        )
+        assert response.status_code == 403
+
+    def test_download_with_no_cert(
+        self,
+        pqc_guarded_distribution,
+        distribution_base_url,
+    ):
+        distribution = pqc_guarded_distribution.distribution
+        url = distribution_base_url(distribution.base_url)
+        response = requests.get(urljoin(url, "test_file"))
+        assert response.status_code == 403

pulp_certguard/tests/functional/constants.py

@@ -14,7 +14,6 @@
     X509_CERTS_BASE_PATH, "un_urlencoded_cert.txt"
 )
 
-
 RHSM_CA_CERT_FILE_PATH = os.path.join(_CURRENT_DIR, "artifacts", "rhsm", "katello-default-ca.crt")
 
 RHSM_CLIENT_CERT_FROM_UNTRUSTED_CA = os.path.join(

pulp_file/pytest_plugin.py

@@ -255,6 +255,64 @@ def _file_remote_client_cert_req_factory(*, manifest_path, policy, **kwargs):
     return _file_remote_client_cert_req_factory
 
 
+@pytest.fixture(scope="class")
+def file_fixture_server_pqc_ssl(pqc_ssl_ctx, file_fixtures_root, gen_fixture_server):
+    return gen_fixture_server(file_fixtures_root, pqc_ssl_ctx)
+
+
+@pytest.fixture(scope="class")
+def file_fixture_server_pqc_ssl_client_cert_req(
+    pqc_ssl_ctx_req_client_auth, file_fixtures_root, gen_fixture_server
+):
+    return gen_fixture_server(file_fixtures_root, pqc_ssl_ctx_req_client_auth)
+
+
+@pytest.fixture(scope="class")
+def file_remote_pqc_ssl_factory(
+    file_fixture_server_pqc_ssl,
+    file_bindings,
+    pqc_certificate_authority,
+    gen_object_with_cleanup,
+):
+    def _file_remote_pqc_ssl_factory(*, manifest_path, policy, **kwargs):
+        url = file_fixture_server_pqc_ssl.make_url(manifest_path)
+        kwargs.update(
+            {
+                "url": str(url),
+                "policy": policy,
+                "name": str(uuid.uuid4()),
+                "ca_cert": pqc_certificate_authority.ca_cert_pem,
+            }
+        )
+        return gen_object_with_cleanup(file_bindings.RemotesFileApi, kwargs)
+
+    return _file_remote_pqc_ssl_factory
+
+
+@pytest.fixture(scope="class")
+def file_remote_pqc_client_cert_req_factory(
+    file_fixture_server_pqc_ssl_client_cert_req,
+    file_bindings,
+    pqc_certificate_authority,
+    gen_object_with_cleanup,
+):
+    def _file_remote_pqc_client_cert_req_factory(*, manifest_path, policy, **kwargs):
+        url = file_fixture_server_pqc_ssl_client_cert_req.make_url(manifest_path)
+        kwargs.update(
+            {
+                "url": str(url),
+                "policy": policy,
+                "name": str(uuid.uuid4()),
+                "ca_cert": pqc_certificate_authority.ca_cert_pem,
+                "client_cert": pqc_certificate_authority.client_cert_pem,
+                "client_key": pqc_certificate_authority.client_key_pem,
+            }
+        )
+        return gen_object_with_cleanup(file_bindings.RemotesFileApi, kwargs)
+
+    return _file_remote_pqc_client_cert_req_factory
+
+
 @pytest.fixture(scope="class")
 def file_repository_factory(file_bindings, gen_object_with_cleanup):
     """A factory to generate a File Repository with auto-deletion after the test run."""

pulp_file/tests/functional/api/test_remote_settings.py

@@ -217,6 +217,52 @@ def test_http_sync_ssl_with_client_cert_req(
     )
 
 
+@pytest.mark.parallel
+def test_http_sync_pqc_ssl_tls_validation_on(
+    file_bindings,
+    file_remote_pqc_ssl_factory,
+    file_repo,
+    basic_manifest_path,
+    monitor_task,
+):
+    """
+    Test file on_demand sync with https:// using PQC (ML-DSA-65) server certificate.
+    """
+    remote_on_demand = file_remote_pqc_ssl_factory(
+        manifest_path=basic_manifest_path, policy="on_demand", tls_validation=True
+    )
+
+    _run_basic_sync_and_assert(
+        file_bindings,
+        remote_on_demand,
+        file_repo,
+        monitor_task,
+    )
+
+
+@pytest.mark.parallel
+def test_http_sync_pqc_ssl_with_client_cert_req(
+    file_bindings,
+    file_remote_pqc_client_cert_req_factory,
+    file_repo,
+    basic_manifest_path,
+    monitor_task,
+):
+    """
+    Test file on_demand sync with https:// using PQC (ML-DSA-65) mutual TLS authentication.
+    """
+    remote_on_demand = file_remote_pqc_client_cert_req_factory(
+        manifest_path=basic_manifest_path, policy="on_demand"
+    )
+
+    _run_basic_sync_and_assert(
+        file_bindings,
+        remote_on_demand,
+        file_repo,
+        monitor_task,
+    )
+
+
 @pytest.mark.parallel
 def test_ondemand_to_immediate_sync(
     file_bindings,

pulpcore/content/__init__.py

@@ -29,12 +29,18 @@
 
 log = logging.getLogger(__name__)
 
+# PQC (post-quantum) X.509 certificates can exceed aiohttp's default 8190-byte header
+# limit when forwarded via X-CLIENT-CERT by a reverse proxy.
+_HANDLER_ARGS = {"max_field_size": 16 * 1024}
+
 if settings.OTEL_ENABLED:
     from .instrumentation import instrumentation  # noqa: E402
 
-    app = web.Application(middlewares=[guid, authenticate, instrumentation()])
+    app = web.Application(
+        middlewares=[guid, authenticate, instrumentation()], handler_args=_HANDLER_ARGS
+    )
 else:
-    app = web.Application(middlewares=[guid, authenticate])
+    app = web.Application(middlewares=[guid, authenticate], handler_args=_HANDLER_ARGS)
 
 
 if settings.UVLOOP_ENABLED: