diff --git a/CHANGES/+remote-policy.feature b/CHANGES/+remote-policy.feature new file mode 100644 index 00000000000..712da8f4e85 --- /dev/null +++ b/CHANGES/+remote-policy.feature @@ -0,0 +1 @@ +Added `UpstreamPulp.remote_policy` so remotes created during replication can use `on_demand` or `streamed` instead of defaulting to `immediate`. diff --git a/CHANGES/+replicate-ssl-tempfiles.bugfix b/CHANGES/+replicate-ssl-tempfiles.bugfix new file mode 100644 index 00000000000..3c9642aef77 --- /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/docs/user/guides/replication.md b/docs/user/guides/replication.md index a8b627e3e42..7bf07c3a4c6 100644 --- a/docs/user/guides/replication.md +++ b/docs/user/guides/replication.md @@ -48,6 +48,7 @@ pulp upstream-pulp create \ | `tls_validation` | Whether to verify the upstream server's TLS certificate. Defaults to `True`. | | `q_select` | A filter expression to select which upstream distributions to replicate. See [Filtering Distributions](#filtering-distributions-with-q_select). | | `policy` | Controls how replication manages local objects. One of `all`, `labeled`, or `nodelete`. See [Replication Policies](#replication-policies). Defaults to `all`. | +| `remote_policy` | Download policy for remotes created during replication. One of `immediate`, `on_demand`, or `streamed`. Distinct from `policy`. When unset, remotes use Pulp's default (`immediate`). | ## Running Replication @@ -151,7 +152,8 @@ pulp upstream-pulp replicate --upstream-pulp "my-upstream" ## Replication Policies The `policy` field controls how replication handles local objects, particularly when upstream -distributions are removed or no longer match a `q_select` filter. +distributions are removed or no longer match a `q_select` filter. It is not the same as a remote's +download policy (`immediate`, `on_demand`, or `streamed`); set that with `remote_policy`. ### `all` (default) diff --git a/pulpcore/app/migrations/0157_upstreampulp_remote_policy.py b/pulpcore/app/migrations/0157_upstreampulp_remote_policy.py new file mode 100644 index 00000000000..2c6ddc91c1f --- /dev/null +++ b/pulpcore/app/migrations/0157_upstreampulp_remote_policy.py @@ -0,0 +1,34 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("core", "0156_alter_contentartifact_relative_path_and_more"), + ] + + operations = [ + migrations.AddField( + model_name="upstreampulp", + name="remote_policy", + field=models.TextField( + choices=[ + ("immediate", "When syncing, download all metadata and content now."), + ( + "on_demand", + "When syncing, download metadata, but do not download content now. " + "Instead, download content as clients request it, and save it in Pulp " + "to be served for future client requests.", + ), + ( + "streamed", + "When syncing, download metadata, but do not download content now. " + "Instead,download content as clients request it, but never save it in " + "Pulp. This causes future requests for that same content to have to be " + "downloaded again.", + ), + ], + null=True, + ), + ), + ] diff --git a/pulpcore/app/models/replica.py b/pulpcore/app/models/replica.py index 29b32c4d5c0..4bbd7fa73a4 100644 --- a/pulpcore/app/models/replica.py +++ b/pulpcore/app/models/replica.py @@ -11,6 +11,8 @@ from pulpcore.app.util import get_domain_pk from pulpcore.plugin.models import AutoAddObjPermsMixin, BaseModel, EncryptedTextField +from .repository import Remote + class UpstreamPulp(BaseModel, AutoAddObjPermsMixin): ALL = "all" @@ -59,6 +61,7 @@ class UpstreamPulp(BaseModel, AutoAddObjPermsMixin): sock_read_timeout = models.FloatField( null=True, validators=[MinValueValidator(0.0, "Timeout must be >= 0")] ) + remote_policy = models.TextField(choices=Remote.POLICY_CHOICES, null=True) q_select = models.TextField(null=True) policy = models.TextField(choices=POLICY_CHOICES, default=ALL) diff --git a/pulpcore/app/serializers/replica.py b/pulpcore/app/serializers/replica.py index 7d13425865f..e130ac8afee 100644 --- a/pulpcore/app/serializers/replica.py +++ b/pulpcore/app/serializers/replica.py @@ -3,7 +3,7 @@ from rest_framework import serializers from rest_framework.validators import UniqueValidator -from pulpcore.app.models import UpstreamPulp +from pulpcore.app.models import Remote, UpstreamPulp from pulpcore.app.serializers import ( HiddenFieldsMixin, IdentityField, @@ -122,6 +122,16 @@ class UpstreamPulpSerializer(ModelSerializer, HiddenFieldsMixin): ), min_value=0.0, ) + remote_policy = serializers.ChoiceField( + choices=Remote.POLICY_CHOICES, + help_text=_( + "Download policy for remotes created during replication. One of 'immediate', " + "'on_demand', or 'streamed'. Distinct from 'policy', which controls how replicate " + "manages local objects. Defaults to the Remote default ('immediate') when unset." + ), + required=False, + allow_null=True, + ) pulp_last_updated = serializers.DateTimeField( help_text="Timestamp of the most recent update of the remote.", read_only=True @@ -178,6 +188,7 @@ class Meta: "connect_timeout", "sock_connect_timeout", "sock_read_timeout", + "remote_policy", "pulp_last_updated", "hidden_fields", "q_select", diff --git a/pulpcore/app/tasks/replica.py b/pulpcore/app/tasks/replica.py index 60510bc9d72..d8aad2f0471 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,36 +29,31 @@ def user_agent(): return f"pulpcore/{pulp_version} ({python}, {system}) (pulp-glue {pulp_glue_version})" -def replicate_distributions(server_pk, q_select=None, **kwargs): - server = UpstreamPulp.objects.get(pk=server_pk) - - # Write out temporary files related to SSL +@contextmanager +def _ssl_temp_files(server): + """Write UpstreamPulp TLS material to temp files that live for this context.""" 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. - } - ) - + 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 _build_remote_settings(server): + """Build fields copied onto remotes created during replication.""" remote_settings = { "ca_cert": server.ca_cert, "tls_validation": server.tls_validation, @@ -70,73 +66,102 @@ def replicate_distributions(server_pk, q_select=None, **kwargs): "sock_connect_timeout": server.sock_connect_timeout, "sock_read_timeout": server.sock_read_timeout, } + # Omit policy when unset so new remotes keep Remote.policy's default (immediate). + if (remote_policy := getattr(server, "remote_policy", None)) is not None: + remote_settings["policy"] = remote_policy + return remote_settings - 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 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. + } + ) + + remote_settings = _build_remote_settings(server) + 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/functional/api/test_replication.py b/pulpcore/tests/functional/api/test_replication.py index f208a4be082..a06d49fe3af 100644 --- a/pulpcore/tests/functional/api/test_replication.py +++ b/pulpcore/tests/functional/api/test_replication.py @@ -281,6 +281,80 @@ def test_replication_remote_settings_propagation( assert remote.max_retries == 2 +@pytest.mark.parallel +def test_replication_remote_policy( + domain_factory, + bindings_cfg, + pulpcore_bindings, + file_bindings, + monitor_task, + monitor_task_group, + pulp_settings, + gen_object_with_cleanup, + file_distribution_factory, + file_publication_factory, + file_repository_factory, + tmp_path, + add_domain_objects_to_cleanup, +): + """Remotes created by replicate() inherit UpstreamPulp.remote_policy when set.""" + source_domain = domain_factory() + add_domain_objects_to_cleanup(source_domain) + + repository = file_repository_factory(pulp_domain=source_domain.name) + file_path = tmp_path / "file.txt" + file_path.write_text("DEADBEEF") + monitor_task( + file_bindings.ContentFilesApi.create( + file=str(file_path), + relative_path="file.txt", + repository=repository.pulp_href, + pulp_domain=source_domain.name, + ).task + ) + publication = file_publication_factory( + pulp_domain=source_domain.name, repository=repository.pulp_href + ) + file_distribution_factory(pulp_domain=source_domain.name, publication=publication.pulp_href) + + replica_domain = domain_factory() + add_domain_objects_to_cleanup(replica_domain) + + upstream_pulp_body = { + "name": str(uuid.uuid4()), + "base_url": bindings_cfg.host, + "api_root": pulp_settings.API_ROOT, + "domain": source_domain.name, + "username": bindings_cfg.username, + "password": bindings_cfg.password, + "remote_policy": "on_demand", + } + upstream_pulp = gen_object_with_cleanup( + pulpcore_bindings.UpstreamPulpsApi, upstream_pulp_body, pulp_domain=replica_domain.name + ) + + response = pulpcore_bindings.UpstreamPulpsApi.replicate( + upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate() + ) + monitor_task_group(response.task_group) + + result = file_bindings.RemotesFileApi.list(pulp_domain=replica_domain.name) + assert result.count == 1 + remote = result.results[0] + assert remote.policy == "on_demand" + + pulpcore_bindings.UpstreamPulpsApi.partial_update( + upstream_pulp.pulp_href, {"remote_policy": "streamed"} + ) + response = pulpcore_bindings.UpstreamPulpsApi.replicate( + upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate() + ) + monitor_task_group(response.task_group) + + remote = file_bindings.RemotesFileApi.list(pulp_domain=replica_domain.name).results[0] + assert remote.policy == "streamed" + + @pytest.mark.parallel def test_replication_with_repo_based_distribution( domain_factory, diff --git a/pulpcore/tests/unit/test_replica.py b/pulpcore/tests/unit/test_replica.py new file mode 100644 index 00000000000..e5f56ed2e73 --- /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 diff --git a/pulpcore/tests/unit/test_replica_remote_policy.py b/pulpcore/tests/unit/test_replica_remote_policy.py new file mode 100644 index 00000000000..f5eea3d335a --- /dev/null +++ b/pulpcore/tests/unit/test_replica_remote_policy.py @@ -0,0 +1,35 @@ +from types import SimpleNamespace + +from pulpcore.app.tasks.replica import _build_remote_settings + + +def _server(**overrides): + base = { + "ca_cert": "api-ca", + "tls_validation": True, + "client_cert": "api-cert", + "client_key": "api-key", + "download_concurrency": 10, + "max_retries": 3, + "total_timeout": 30, + "connect_timeout": 5, + "sock_connect_timeout": 5, + "sock_read_timeout": 5, + "remote_policy": None, + } + base.update(overrides) + return SimpleNamespace(**base) + + +def test_build_remote_settings_omits_policy_when_unset(): + settings = _build_remote_settings(_server()) + + assert "policy" not in settings + assert settings["ca_cert"] == "api-ca" + assert settings["download_concurrency"] == 10 + + +def test_build_remote_settings_includes_policy_when_set(): + settings = _build_remote_settings(_server(remote_policy="on_demand")) + + assert settings["policy"] == "on_demand"