From 95fc7da078f1c764d215968991b9ac8f72cfc73d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20M=C3=A9ndez=20Hern=C3=A1ndez?= Date: Wed, 26 Aug 2026 12:35:14 +0200 Subject: [PATCH] fix: keep replicate TLS temp files alive Replication over HTTPS wrote the CA, client cert, and client key to NamedTemporaryFiles but only kept their paths, so CPython deleted earlier files before pulp-glue could load the upstream API with TLS verification. Keep those files alive for the duration of replicate(), pass the CA bundle path directly as verify_ssl, add regression coverage for the TLS handoff, and clean up the temp files on exit. Assisted-By: Cursor Co-authored-by: Cursor --- CHANGES/+replicate-ssl-tempfiles.bugfix | 1 + pulpcore/app/tasks/replica.py | 226 +++++++++++++----------- pulpcore/tests/unit/test_replica.py | 155 ++++++++++++++++ 3 files changed, 277 insertions(+), 105 deletions(-) create mode 100644 CHANGES/+replicate-ssl-tempfiles.bugfix create mode 100644 pulpcore/tests/unit/test_replica.py diff --git a/CHANGES/+replicate-ssl-tempfiles.bugfix b/CHANGES/+replicate-ssl-tempfiles.bugfix new file mode 100644 index 0000000000..3c9642aef7 --- /dev/null +++ b/CHANGES/+replicate-ssl-tempfiles.bugfix @@ -0,0 +1 @@ +Fixed replicate() deleting temporary TLS files before pulp-glue could use them, and stopped leaking PULP_CA_BUNDLE into later worker tasks. diff --git a/pulpcore/app/tasks/replica.py b/pulpcore/app/tasks/replica.py index 60510bc9d7..19de2bd60e 100644 --- a/pulpcore/app/tasks/replica.py +++ b/pulpcore/app/tasks/replica.py @@ -1,6 +1,7 @@ import os import platform import sys +from contextlib import contextmanager from tempfile import NamedTemporaryFile from django.db import transaction @@ -28,115 +29,130 @@ def user_agent(): return f"pulpcore/{pulp_version} ({python}, {system}) (pulp-glue {pulp_glue_version})" +@contextmanager +def _ssl_temp_files(server): + """Write UpstreamPulp TLS material to temp files that live for this context.""" + ssl_files = {} + try: + for key in ["ca_cert", "client_cert", "client_key"]: + if value := getattr(server, key): + suffix = ".key" if key == "client_key" else ".pem" + with NamedTemporaryFile( + dir=".", mode="w", encoding="utf-8", delete=False, suffix=suffix + ) as f: + f.write(value) + f.flush() + ssl_files[key] = f.name + yield ssl_files + finally: + for path in ssl_files.values(): + try: + os.unlink(path) + except FileNotFoundError: + pass + + def replicate_distributions(server_pk, q_select=None, **kwargs): server = UpstreamPulp.objects.get(pk=server_pk) + with _ssl_temp_files(server) as ssl_files: + verify_ssl = ( + ssl_files["ca_cert"] + if server.tls_validation and "ca_cert" in ssl_files + else server.tls_validation + ) + ctx = ReplicaContext.from_config( + { + "base_url": server.base_url, + "api_root": server.api_root, + "domain": server.domain, + "username": server.username, + "password": server.password, + "cert": ssl_files.get("client_cert"), + "key": ssl_files.get("client_key"), + "user_agent": user_agent(), + "verify_ssl": verify_ssl, + "dry_run": True, # We only want to read from upstream anyway. + } + ) - # Write out temporary files related to SSL - ssl_files = {} - for key in ["ca_cert", "client_cert", "client_key"]: - if value := getattr(server, key): - f = NamedTemporaryFile(dir=".") - f.write(bytes(value, "utf-8")) - f.flush() - ssl_files[key] = f.name - - if "ca_cert" in ssl_files: - os.environ["PULP_CA_BUNDLE"] = ssl_files["ca_cert"] - - ctx = ReplicaContext.from_config( - { - "base_url": server.base_url, - "api_root": server.api_root, - "domain": server.domain, - "username": server.username, - "password": server.password, - "cert": ssl_files.get("client_cert"), - "key": ssl_files.get("client_key"), - "user_agent": user_agent(), - "verify_ssl": server.tls_validation, - "dry_run": True, # We only want to read from upstream anyway. + remote_settings = { + "ca_cert": server.ca_cert, + "tls_validation": server.tls_validation, + "client_cert": server.client_cert, + "client_key": server.client_key, + "download_concurrency": server.download_concurrency, + "max_retries": server.max_retries, + "total_timeout": server.total_timeout, + "connect_timeout": server.connect_timeout, + "sock_connect_timeout": server.sock_connect_timeout, + "sock_read_timeout": server.sock_read_timeout, } - ) - - remote_settings = { - "ca_cert": server.ca_cert, - "tls_validation": server.tls_validation, - "client_cert": server.client_cert, - "client_key": server.client_key, - "download_concurrency": server.download_concurrency, - "max_retries": server.max_retries, - "total_timeout": server.total_timeout, - "connect_timeout": server.connect_timeout, - "sock_connect_timeout": server.sock_connect_timeout, - "sock_read_timeout": server.sock_read_timeout, - } - - try: - task_group = TaskGroup.current() - supported_replicators = [] - # Load all the available replicators - for config in pulp_plugin_configs(): - if config.replicator_classes: - for replicator_class in config.replicator_classes: - req = PluginRequirement( - config.label, specifier=replicator_class.required_version - ) - if ctx.has_plugin(req): - replicator = replicator_class(ctx, task_group, remote_settings, server) - supported_replicators.append(replicator) - - effective_q_select = q_select if q_select is not None else server.q_select - distro_repo_pairs = [] - for replicator in supported_replicators: - distro_names = [] - pending_distributions = [] - distros = replicator.upstream_distributions(q=effective_q_select) - for distro in distros: - # Create remote - remote = replicator.create_or_update_remote(upstream_distribution=distro) - if not remote: - # The upstream distribution is not serving any content, - # let it fall through the cracks and be cleaned up below. - continue - # Check if there is already a repository - repository = replicator.create_or_update_repository(remote=remote) - if not repository: - # No update occurred because server.policy==LABELED and there was - # an already existing local repository with the same name - continue - - # Dispatch a sync task if needed - if replicator.requires_syncing(distro): - replicator.sync(repository, remote) - - # Add name to the list of known distribution names - distro_names.append(distro["name"]) - distro_repo_pairs.append((distro["name"], str(repository.pk))) - pending_distributions.append((repository, distro)) - - # Get or create distributions BEFORE remove_missing so that - # create_or_update_distribution can synchronously rename any existing - # distribution matched by base_path. remove_missing then sees the - # updated name in the DB and won't schedule it for deletion. - for repository, distro in pending_distributions: - replicator.create_or_update_distribution(repository, distro) - - # When a per-request q_select override is used, this is a selective sync - # of a subset of distributions. Skipping remove_missing avoids deleting - # distributions that simply weren't included in the filter — but it also - # means that distributions removed from upstream won't be cleaned up until - # a full (non-overridden) replication runs. - if q_select is None: - replicator.remove_missing(distro_names) - except GluePulpException as e: - raise ExternalServiceError(service_name=server.base_url, details=str(e)) - - dispatch( - finalize_replication, - task_group=task_group, - exclusive_resources=[server, distros_lock_uri(server.pulp_domain_id)], - args=[server.pk, distro_repo_pairs], - ) + try: + task_group = TaskGroup.current() + supported_replicators = [] + # Load all the available replicators + for config in pulp_plugin_configs(): + if config.replicator_classes: + for replicator_class in config.replicator_classes: + req = PluginRequirement( + config.label, specifier=replicator_class.required_version + ) + if ctx.has_plugin(req): + replicator = replicator_class(ctx, task_group, remote_settings, server) + supported_replicators.append(replicator) + + effective_q_select = q_select if q_select is not None else server.q_select + distro_repo_pairs = [] + for replicator in supported_replicators: + distro_names = [] + pending_distributions = [] + distros = replicator.upstream_distributions(q=effective_q_select) + for distro in distros: + # Create remote + remote = replicator.create_or_update_remote(upstream_distribution=distro) + if not remote: + # The upstream distribution is not serving any content, + # let it fall through the cracks and be cleaned up below. + continue + # Check if there is already a repository + repository = replicator.create_or_update_repository(remote=remote) + if not repository: + # No update occurred because server.policy==LABELED and there was + # an already existing local repository with the same name + continue + + # Dispatch a sync task if needed + if replicator.requires_syncing(distro): + replicator.sync(repository, remote) + + # Add name to the list of known distribution names + distro_names.append(distro["name"]) + distro_repo_pairs.append((distro["name"], str(repository.pk))) + pending_distributions.append((repository, distro)) + + # Get or create distributions BEFORE remove_missing so that + # create_or_update_distribution can synchronously rename any existing + # distribution matched by base_path. remove_missing then sees the + # updated name in the DB and won't schedule it for deletion. + for repository, distro in pending_distributions: + replicator.create_or_update_distribution(repository, distro) + + # When a per-request q_select override is used, this is a selective sync + # of a subset of distributions. Skipping remove_missing avoids deleting + # distributions that simply weren't included in the filter — but it also + # means that distributions removed from upstream won't be cleaned up until + # a full (non-overridden) replication runs. + if q_select is None: + replicator.remove_missing(distro_names) + except GluePulpException as e: + raise ExternalServiceError(service_name=server.base_url, details=str(e)) + + dispatch( + finalize_replication, + task_group=task_group, + exclusive_resources=[server, distros_lock_uri(server.pulp_domain_id)], + args=[server.pk, distro_repo_pairs], + ) def finalize_replication(server_pk, distro_repo_pairs, **kwargs): diff --git a/pulpcore/tests/unit/test_replica.py b/pulpcore/tests/unit/test_replica.py new file mode 100644 index 0000000000..e5f56ed2e7 --- /dev/null +++ b/pulpcore/tests/unit/test_replica.py @@ -0,0 +1,155 @@ +import os +from types import SimpleNamespace + +from pulpcore.app.tasks import replica +from pulpcore.app.tasks.replica import _ssl_temp_files + + +def test_ssl_temp_files_keep_all_certs_until_context_exits(tmp_path, monkeypatch): + """Katello replicate() sends ca_cert + client_cert + client_key together. + + The old loop stored only filenames, so the CA NamedTemporaryFile was + garbage-collected (and unlinked) before pulp-glue opened it. + """ + monkeypatch.chdir(tmp_path) + server = SimpleNamespace( + ca_cert="-----BEGIN CA-----\nca\n-----END CA-----", + client_cert="-----BEGIN CERT-----\ncert\n-----END CERT-----", + client_key="-----BEGIN KEY-----\nkey\n-----END KEY-----", + ) + + with _ssl_temp_files(server) as ssl_files: + for key, expected in ( + ("ca_cert", server.ca_cert), + ("client_cert", server.client_cert), + ("client_key", server.client_key), + ): + path = ssl_files[key] + assert os.path.exists(path) + with open(path, encoding="utf-8") as f: + assert f.read() == expected + ca_path = ssl_files["ca_cert"] + cert_path = ssl_files["client_cert"] + key_path = ssl_files["client_key"] + + assert not os.path.exists(ca_path) + assert not os.path.exists(cert_path) + assert not os.path.exists(key_path) + + +def test_ssl_temp_files_skips_missing_material(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + server = SimpleNamespace(ca_cert="ca", client_cert=None, client_key=None) + + with _ssl_temp_files(server) as ssl_files: + assert set(ssl_files) == {"ca_cert"} + assert os.path.exists(ssl_files["ca_cert"]) + + +def test_replicate_distributions_passes_ca_path_as_verify_ssl(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + captured = {} + server = SimpleNamespace( + base_url="https://example.com", + api_root="/pulp/", + domain="default", + username="user", + password="pass", + ca_cert="ca", + client_cert="cert", + client_key="key", + tls_validation=True, + download_concurrency=10, + max_retries=3, + total_timeout=30, + connect_timeout=5, + sock_connect_timeout=5, + sock_read_timeout=5, + q_select=None, + pulp_domain_id="domain-id", + pk="server-pk", + ) + + class DummyContext: + def has_plugin(self, req): + assert os.path.exists(captured["config"]["verify_ssl"]) + assert os.path.exists(captured["config"]["cert"]) + assert os.path.exists(captured["config"]["key"]) + return False + + class DummyReplicator: + required_version = ">=0" + + def fake_from_config(config): + captured["config"] = config + assert os.path.exists(config["verify_ssl"]) + assert os.path.exists(config["cert"]) + assert os.path.exists(config["key"]) + return DummyContext() + + monkeypatch.setattr(replica.UpstreamPulp.objects, "get", lambda pk: server) + monkeypatch.setattr(replica.ReplicaContext, "from_config", fake_from_config) + monkeypatch.setattr( + replica, + "pulp_plugin_configs", + lambda: [SimpleNamespace(label="core", replicator_classes=[DummyReplicator])], + ) + monkeypatch.setattr(replica.TaskGroup, "current", lambda: "task-group") + monkeypatch.setattr(replica, "dispatch", lambda *args, **kwargs: None) + + replica.replicate_distributions(server.pk) + + assert isinstance(captured["config"]["verify_ssl"], str) + + +def test_replicate_distributions_uses_false_verify_ssl_when_disabled(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + captured = {} + server = SimpleNamespace( + base_url="https://example.com", + api_root="/pulp/", + domain="default", + username="user", + password="pass", + ca_cert="ca", + client_cert="cert", + client_key="key", + tls_validation=False, + download_concurrency=10, + max_retries=3, + total_timeout=30, + connect_timeout=5, + sock_connect_timeout=5, + sock_read_timeout=5, + q_select=None, + pulp_domain_id="domain-id", + pk="server-pk", + ) + + class DummyContext: + def has_plugin(self, req): + return False + + class DummyReplicator: + required_version = ">=0" + + def fake_from_config(config): + captured["config"] = config + assert config["verify_ssl"] is False + assert os.path.exists(config["cert"]) + assert os.path.exists(config["key"]) + return DummyContext() + + monkeypatch.setattr(replica.UpstreamPulp.objects, "get", lambda pk: server) + monkeypatch.setattr(replica.ReplicaContext, "from_config", fake_from_config) + monkeypatch.setattr( + replica, + "pulp_plugin_configs", + lambda: [SimpleNamespace(label="core", replicator_classes=[DummyReplicator])], + ) + monkeypatch.setattr(replica.TaskGroup, "current", lambda: "task-group") + monkeypatch.setattr(replica, "dispatch", lambda *args, **kwargs: None) + + replica.replicate_distributions(server.pk) + + assert captured["config"]["verify_ssl"] is False