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