diff --git a/CHANGES/1657.bugfix b/CHANGES/1657.bugfix
new file mode 100644
index 000000000..8c3e751cb
--- /dev/null
+++ b/CHANGES/1657.bugfix
@@ -0,0 +1,2 @@
+Disallowed anonymous users to pull new content via a pull-through caching distribution. Content that
+is already cached/downloaded can be still pulled.
diff --git a/pulp_container/app/registry_api.py b/pulp_container/app/registry_api.py
index 8ea7d04ba..9bcabbe47 100644
--- a/pulp_container/app/registry_api.py
+++ b/pulp_container/app/registry_api.py
@@ -306,7 +306,7 @@ def get_pull_through_drv(self, path):
             .order_by("-base_path")
             .first()
         )
-        if not pull_through_cache_distribution:
+        if not pull_through_cache_distribution or not self.request.user.is_authenticated:
             raise RepositoryNotFound(name=path)
 
         try:
@@ -1028,7 +1028,11 @@ def handle_safe_method(self, request, path, pk):
                 tag = models.Tag.objects.get(name=pk, pk__in=repository_version.content)
             except models.Tag.DoesNotExist:
                 distribution = distribution.cast()
-                if distribution.remote and distribution.pull_through_distribution_id:
+                if (
+                    distribution.remote
+                    and distribution.pull_through_distribution_id
+                    and request.user.is_authenticated
+                ):
                     remote = distribution.remote.cast()
                     # issue a head request first to ensure that the content exists on the remote
                     # source; we want to prevent immediate "not found" error responses from
@@ -1077,7 +1081,11 @@ def handle_safe_method(self, request, path, pk):
                     pass
 
                 distribution = distribution.cast()
-                if distribution.remote and distribution.pull_through_distribution_id:
+                if (
+                    distribution.remote
+                    and distribution.pull_through_distribution_id
+                    and request.user.is_authenticated
+                ):
                     remote = distribution.remote.cast()
                     self.fetch_manifest(remote, pk)
                     return redirects.redirect_to_content_app("manifests", pk)
diff --git a/pulp_container/app/token_verification.py b/pulp_container/app/token_verification.py
index 5e04fb2a6..344f5f16c 100644
--- a/pulp_container/app/token_verification.py
+++ b/pulp_container/app/token_verification.py
@@ -75,7 +75,7 @@ def authenticate(self, request):
         If basic authentication could not success, remote webserver authentication is considered.
         """
         if request.headers.get("Authorization") == "Basic Og==":
-            return (AnonymousUser, None)
+            return (AnonymousUser(), None)
 
         try:
             user = super().authenticate(request)
diff --git a/pulp_container/tests/functional/api/test_pull_through_cache.py b/pulp_container/tests/functional/api/test_pull_through_cache.py
index 6a623d72d..a101594eb 100644
--- a/pulp_container/tests/functional/api/test_pull_through_cache.py
+++ b/pulp_container/tests/functional/api/test_pull_through_cache.py
@@ -2,12 +2,14 @@
 import subprocess
 import pytest
 
+from subprocess import CalledProcessError
 from uuid import uuid4
 
 from pulp_container.tests.functional.constants import (
     REGISTRY_V2,
     REGISTRY_V2_FEED_URL,
     PULP_HELLO_WORLD_REPO,
+    PULP_HELLO_WORLD_LINUX_AMD64_DIGEST,
     PULP_FIXTURE_1,
 )
 
@@ -30,12 +32,30 @@ def pull_through_distribution(
 
 
 @pytest.fixture
-def pull_and_verify(
+def add_pull_through_entities_to_cleanup(
+    container_repository_api,
+    container_remote_api,
+    container_distribution_api,
     add_to_cleanup,
+):
+    def _add_pull_through_entities_to_cleanup(path):
+        repository = container_repository_api.list(name=path).results[0]
+        add_to_cleanup(container_repository_api, repository.pulp_href)
+        remote = container_remote_api.list(name=path).results[0]
+        add_to_cleanup(container_remote_api, remote.pulp_href)
+        distribution = container_distribution_api.list(name=path).results[0]
+        add_to_cleanup(container_distribution_api, distribution.pulp_href)
+
+    return _add_pull_through_entities_to_cleanup
+
+
+@pytest.fixture
+def pull_and_verify(
+    anonymous_user,
+    add_pull_through_entities_to_cleanup,
     container_pull_through_distribution_api,
     container_distribution_api,
     container_repository_api,
-    container_remote_api,
     container_tag_api,
     registry_client,
     local_registry,
@@ -46,24 +66,25 @@ def _pull_and_verify(images, pull_through_distribution):
             remote_image_path = f"{REGISTRY_V2}/{image_path}"
             local_image_path = f"{pull_through_distribution.base_path}/{image_path}"
 
+            # 0. test if an anonymous user cannot pull new content through the pull-through cache
+            with anonymous_user, pytest.raises(CalledProcessError):
+                local_registry.pull(local_image_path)
+
             # 1. pull remote content through the pull-through distribution
             local_registry.pull(local_image_path)
             local_image = local_registry.inspect(local_image_path)
 
-            # when the client pulls the image, a repository, distribution, and remote is created in
-            # the background; therefore, scheduling the cleanup for these entities is necessary
             path, tag = local_image_path.split(":")
             tags_to_verify.append(tag)
-            repository = container_repository_api.list(name=path).results[0]
-            add_to_cleanup(container_repository_api, repository.pulp_href)
-            remote = container_remote_api.list(name=path).results[0]
-            add_to_cleanup(container_remote_api, remote.pulp_href)
-            distribution = container_distribution_api.list(name=path).results[0]
-            add_to_cleanup(container_distribution_api, distribution.pulp_href)
+
+            # when the client pulls the image, a repository, distribution, and remote is created in
+            # the background; therefore, scheduling the cleanup for these entities is necessary
+            add_pull_through_entities_to_cleanup(path)
 
             pull_through_distribution = container_pull_through_distribution_api.list(
                 name=pull_through_distribution.name
             ).results[0]
+            distribution = container_distribution_api.list(name=path).results[0]
             assert [distribution.pulp_href] == pull_through_distribution.distributions
 
             # 2. verify if the pulled content is the same as on the remote
@@ -89,6 +110,10 @@ def _pull_and_verify(images, pull_through_distribution):
             tags = container_tag_api.list(repository_version=repository.latest_version_href).results
             assert sorted(tags_to_verify) == sorted([tag.name for tag in tags])
 
+            # 6. test if the anonymous user can pull existing content via the pull-through cache
+            with anonymous_user:
+                local_registry.pull(local_image_path)
+
     return _pull_and_verify
 
 
@@ -102,6 +127,25 @@ def test_manifest_pull(delete_orphans_pre, pull_through_distribution, pull_and_v
     pull_and_verify(images, pull_through_distribution)
 
 
+def test_anonymous_pull_by_digest(
+    delete_orphans_pre,
+    add_pull_through_entities_to_cleanup,
+    anonymous_user,
+    local_registry,
+    pull_through_distribution,
+):
+    image = f"{PULP_HELLO_WORLD_REPO}@{PULP_HELLO_WORLD_LINUX_AMD64_DIGEST}"
+    local_image_path = f"{pull_through_distribution.base_path}/{image}"
+
+    with anonymous_user, pytest.raises(CalledProcessError):
+        local_registry.pull(local_image_path)
+
+    local_registry.pull(local_image_path)
+
+    with anonymous_user:
+        local_registry.pull(local_image_path)
+
+
 def test_conflicting_names_and_paths(
     container_remote_api,
     container_remote_factory,
diff --git a/pulp_container/tests/functional/conftest.py b/pulp_container/tests/functional/conftest.py
index 457e443f7..c9789f062 100644
--- a/pulp_container/tests/functional/conftest.py
+++ b/pulp_container/tests/functional/conftest.py
@@ -5,7 +5,7 @@
 import requests
 import subprocess
 
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from urllib.parse import urljoin, urlparse
 from uuid import uuid4
 
@@ -196,12 +196,16 @@ def pull(image_path):
                 registry_client.login(
                     "-u", bindings_cfg.username, "-p", bindings_cfg.password, registry_name
                 )
+
+                try:
+                    registry_client.pull("/".join([registry_name, image_path]))
+                finally:
+                    registry_client.logout(registry_name)
             else:
-                registry_client.logout(registry_name)
-            try:
+                with suppress(subprocess.CalledProcessError):
+                    registry_client.logout(registry_name)
+
                 registry_client.pull("/".join([registry_name, image_path]))
-            finally:
-                registry_client.logout(registry_name)
 
         @staticmethod
         def tag_and_push(image_path, local_url, *args):
diff --git a/pulp_container/tests/functional/constants.py b/pulp_container/tests/functional/constants.py
index 584f80464..cb6c29004 100644
--- a/pulp_container/tests/functional/constants.py
+++ b/pulp_container/tests/functional/constants.py
@@ -31,6 +31,9 @@
 
 # a repository having the size of 1.84kB
 PULP_HELLO_WORLD_REPO = "pulp/hello-world"
+PULP_HELLO_WORLD_LINUX_AMD64_DIGEST = (
+    "sha256:239de6dd745ed1a211123322865b4c342c706e7c1e01644a1bbefe8f8846c5ff"
+)
 
 # a repository containing 4 manifest lists and 5 manifests
 PULP_FIXTURE_1 = "pulp/test-fixture-1"