feat: add ban_tags_enabled/ban_tags to docker remotes to block named tags (#43)

Adds two per-remote config keys for docker remotes:

  ban_tags_enabled: false   # opt-in, default off
  ban_tags:
    - latest
    - edge

When ban_tags_enabled is true and a manifest request arrives for a named
tag in ban_tags, the proxy returns 403. sha256-addressed pulls are never
blocked, so images already pulled can still be referenced by digest.
Blob requests are unaffected.

Reviewed-on: #43

README.md

@@ -11,6 +11,7 @@ FastAPI caching proxy that downloads and stores files from remote sources in S3-
 - Stale-on-upstream-error: refreshes TTL when backend is unreachable rather than evicting
 - URL rewriting for PyPI simple index, npm metadata, and Helm `index.yaml`
 - Access control via regex patterns — unmatched paths return 403
+- Docker tag banning — block named tags (e.g. `latest`) while allowing digest pulls
 
 ## Architecture
 
@@ -243,6 +244,26 @@ remotes:
 
 Tag manifests and `/tags/list` are built-in mutable patterns. Digest-addressed blobs are immutable.
 
+#### Banning tags
+
+Set `ban_tags_enabled: true` and list named tags in `ban_tags` to block specific tag references. Requests for a banned tag return `403`. Digest-addressed pulls (`sha256:…`) are never blocked, so images already in use can still be referenced by digest.
+
+```yaml
+remotes:
+  dockerhub:
+    base_url: "https://registry-1.docker.io"
+    package: "docker"
+    ban_tags_enabled: true
+    ban_tags:
+      - latest   # force pinned tags in CI/CD
+      - edge
+    cache:
+      immutable_ttl: 0
+      mutable_ttl: 300
+```
+
+`ban_tags_enabled` defaults to `false`. Setting it to `true` with an empty `ban_tags` list has no effect.
+
 For RKE2/containerd, configure `/etc/rancher/rke2/registries.yaml`:
 
 ```yaml

src/artifactapi/artifact/docker.py

@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import json
 import logging
@@ -33,6 +34,16 @@ async def proxy(request: Request, remote_name: str, path: str, storage, cache, c
             logger.info(f"PATTERN BLOCKED: {remote_name}/{path}")
             raise HTTPException(status_code=403, detail="Image not allowed by configuration patterns")
 
+    if remote_config.get("ban_tags_enabled", False):
+        ban_tags = remote_config.get("ban_tags", [])
+        if ban_tags:
+            tag_match = re.search(r"/manifests/([^/]+)$", path)
+            if tag_match:
+                tag = tag_match.group(1)
+                if not tag.startswith("sha256:") and tag in ban_tags:
+                    logger.info(f"TAG BANNED: {remote_name}/{path} (tag: {tag})")
+                    raise HTTPException(status_code=403, detail=f"Tag '{tag}' is not permitted on this remote")
+
     base_url = remote_config.get("base_url", "").rstrip("/")
     remote_url = f"{base_url}/v2/{path}"
 
@@ -47,8 +58,21 @@ async def proxy(request: Request, remote_name: str, path: str, storage, cache, c
             if not await _proxy.handle_expired_mutable(remote_name, path, remote_url, config, cache, storage):
                 cached_key = None
 
+    lock_acquired = False
+    if not cached_key:
+        lock_acquired = cache.acquire_fetch_lock(remote_name, path)
+        if not lock_acquired:
+            # Another pod is already fetching — poll storage briefly before issuing a duplicate upstream request
+            for _ in range(10):
+                await asyncio.sleep(0.5)
+                probe_key = storage.get_object_key(remote_name, path)
+                if storage.exists(probe_key):
+                    cached_key = probe_key
+                    break
+
     if not cached_key:
         logger.info(f"Cache MISS: {remote_name}/{path} - fetching from remote: {remote_url}")
+        try:
             result = await _proxy.cache_single_artifact(remote_url, remote_name, path, storage, remote_config)
             if result["status"] == "error":
                 raise HTTPException(status_code=502, detail=f"Failed to fetch: {result['error']}")
@@ -64,6 +88,9 @@ async def proxy(request: Request, remote_name: str, path: str, storage, cache, c
                 if published:
                     cache.store_artifact_published(remote_name, path, published)
                 _proxy._check_quarantine(remote_name, published, config)
+        finally:
+            if lock_acquired:
+                cache.release_fetch_lock(remote_name, path)
     elif not is_mutable:
         published = cache.get_artifact_published(remote_name, path)
         if not published:
@@ -90,6 +117,14 @@ async def proxy(request: Request, remote_name: str, path: str, storage, cache, c
             content_type = "application/vnd.oci.image.manifest.v1+json"
 
     digest = f"sha256:{hashlib.sha256(artifact_data).hexdigest()}"
+
+    # Cross-link tag manifests to their sha256 digest key so digest-addressed pulls hit cache
+    if is_mutable and "/manifests/" in path:
+        digest_path = re.sub(r"/manifests/[^/]+$", f"/manifests/{digest}", path)
+        digest_key = storage.get_object_key(remote_name, digest_path)
+        if not storage.exists(digest_key):
+            storage.upload(digest_key, artifact_data)
+
     headers = {
         "Docker-Distribution-Api-Version": "registry/2.0",
         "Docker-Content-Digest": digest,

src/artifactapi/cache/redis.py

@@ -99,6 +99,25 @@ class RedisCache:
         except Exception:
             return None
 
+    def acquire_fetch_lock(self, remote_name: str, path: str, ttl: int = 30) -> bool:
+        """Try to acquire a short-lived fetch lock. Returns True if acquired, False if held by another caller."""
+        if not self.available:
+            return True  # fail open: no Redis → behave as if we always hold the lock
+        key = f"fetchlock:{remote_name}:{hashlib.sha256(path.encode()).hexdigest()[:16]}"
+        try:
+            return bool(self.client.set(key, 1, nx=True, ex=ttl))
+        except Exception:
+            return True
+
+    def release_fetch_lock(self, remote_name: str, path: str) -> None:
+        if not self.available:
+            return
+        key = f"fetchlock:{remote_name}:{hashlib.sha256(path.encode()).hexdigest()[:16]}"
+        try:
+            self.client.delete(key)
+        except Exception:
+            pass
+
     def cleanup_expired_index(self, storage, remote_name: str, path: str) -> None:
         if not self.available:
             return

tests/conftest.py

@@ -41,6 +41,13 @@ TEST_REMOTES = {
             "immutable_patterns": ["^library/nginx"],
             "cache": {"immutable_ttl": 0, "mutable_ttl": 300},
         },
+        "docker-bantags-test": {
+            "base_url": "https://registry.example.com",
+            "package": "docker",
+            "ban_tags_enabled": True,
+            "ban_tags": ["latest", "edge"],
+            "cache": {"immutable_ttl": 0, "mutable_ttl": 300},
+        },
         "generic-test": {
             "base_url": "https://releases.example.com",
             "package": "generic",

tests/test_cache.py

@@ -327,3 +327,74 @@ class TestArtifactPublished:
 
     def test_get_returns_none_when_unavailable(self, unavailable_cache):
         assert unavailable_cache.get_artifact_published("remote", "path") is None
+
+
+# ---------------------------------------------------------------------------
+# fetch lock (thundering-herd deduplication)
+# ---------------------------------------------------------------------------
+
+
+class TestFetchLock:
+    def test_acquire_returns_true_when_lock_obtained(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        result = cache_with_redis.acquire_fetch_lock("myremote", "library/nginx/manifests/latest")
+        assert result is True
+
+    def test_acquire_calls_set_nx_with_ttl(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        cache_with_redis.acquire_fetch_lock("myremote", "library/nginx/manifests/latest", ttl=15)
+        _, kwargs = mock_redis_client.set.call_args
+        assert kwargs.get("nx") is True
+        assert kwargs.get("ex") == 15
+
+    def test_acquire_returns_false_when_lock_already_held(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = None  # Redis SET NX → None when key exists
+        result = cache_with_redis.acquire_fetch_lock("myremote", "library/nginx/manifests/latest")
+        assert result is False
+
+    def test_acquire_fails_open_when_unavailable(self, unavailable_cache):
+        # caller must be allowed to proceed when Redis is down
+        assert unavailable_cache.acquire_fetch_lock("myremote", "some/path") is True
+
+    def test_acquire_fails_open_on_redis_exception(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.side_effect = Exception("connection reset")
+        assert cache_with_redis.acquire_fetch_lock("myremote", "some/path") is True
+
+    def test_lock_key_embeds_path_hash(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        path = "library/nginx/manifests/latest"
+        cache_with_redis.acquire_fetch_lock("myremote", path)
+        args, _ = mock_redis_client.set.call_args
+        expected_hash = hashlib.sha256(path.encode()).hexdigest()[:16]
+        assert args[0] == f"fetchlock:myremote:{expected_hash}"
+
+    def test_lock_key_hash_is_16_chars(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        cache_with_redis.acquire_fetch_lock("myremote", "some/long/path/file.tar.gz")
+        args, _ = mock_redis_client.set.call_args
+        # key format: fetchlock:<remote>:<16-char hash>
+        parts = args[0].split(":")
+        assert len(parts) == 3
+        assert len(parts[2]) == 16
+
+    def test_different_paths_produce_different_lock_keys(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        cache_with_redis.acquire_fetch_lock("myremote", "path/a/manifests/latest")
+        key_a = mock_redis_client.set.call_args[0][0]
+        mock_redis_client.set.reset_mock()
+        cache_with_redis.acquire_fetch_lock("myremote", "path/b/manifests/latest")
+        key_b = mock_redis_client.set.call_args[0][0]
+        assert key_a != key_b
+
+    def test_release_deletes_correct_key(self, cache_with_redis, mock_redis_client):
+        path = "library/nginx/manifests/latest"
+        cache_with_redis.release_fetch_lock("myremote", path)
+        expected_hash = hashlib.sha256(path.encode()).hexdigest()[:16]
+        mock_redis_client.delete.assert_called_once_with(f"fetchlock:myremote:{expected_hash}")
+
+    def test_release_no_op_when_unavailable(self, unavailable_cache):
+        unavailable_cache.release_fetch_lock("myremote", "some/path")  # must not raise
+
+    def test_release_no_op_on_redis_exception(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.delete.side_effect = Exception("timeout")
+        cache_with_redis.release_fetch_lock("myremote", "some/path")  # must not raise

tests/test_routes.py

@@ -260,6 +260,211 @@ class TestDockerProxy:
         mock_fetch.assert_called_once()
         assert response.status_code == 200
 
+    # --- Issue 1: sha256 digest cross-linking ---
+
+    def test_tag_manifest_is_stored_under_digest_key_on_cache_hit(self, client, patched_deps):
+        # When serving a cached tag manifest the handler must also write the content
+        # under the sha256 digest key so subsequent sha256-addressed pulls hit cache.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # First exists call (tag manifest): hit. Second (digest key): miss → triggers upload.
+        deps["storage"].exists.side_effect = [True, False]
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+
+        response = client.get("/v2/docker-test/library/nginx/manifests/v1.25.3")
+
+        assert response.status_code == 200
+        deps["storage"].upload.assert_called_once_with(deps["storage"].get_object_key.return_value, manifest)
+
+    def test_tag_manifest_digest_key_not_written_when_already_exists(self, client, patched_deps):
+        # When the digest key already exists in storage upload must not be called.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # Both the tag key and the digest key already present.
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+
+        client.get("/v2/docker-test/library/nginx/manifests/v1.25.3")
+
+        deps["storage"].upload.assert_not_called()
+
+    def test_sha256_manifest_request_is_not_cross_linked(self, client, patched_deps):
+        # sha256-addressed manifests are immutable — the cross-link logic must not apply.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = False  # sha256 manifest is immutable
+
+        with patch("artifactapi.artifact.proxy._fetch_last_modified", new_callable=AsyncMock, return_value=None):
+            client.get("/v2/docker-test/library/nginx/manifests/sha256:" + "a" * 64)
+
+        deps["storage"].upload.assert_not_called()
+
+    # --- Issue 2: thundering herd distributed lock ---
+
+    def test_lock_acquired_and_released_on_upstream_fetch(self, client, patched_deps):
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.side_effect = [False, False]  # initial miss; digest key also absent
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = True
+
+        with patch(
+            "artifactapi.artifact.proxy.cache_single_artifact",
+            new_callable=AsyncMock,
+            return_value={"status": "cached"},
+        ):
+            response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        deps["cache"].acquire_fetch_lock.assert_called_once()
+        deps["cache"].release_fetch_lock.assert_called_once()
+        assert response.status_code == 200
+
+    def test_lock_released_even_when_fetch_returns_error(self, client, patched_deps):
+        deps = patched_deps
+        deps["storage"].exists.return_value = False
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = True
+
+        with patch(
+            "artifactapi.artifact.proxy.cache_single_artifact",
+            new_callable=AsyncMock,
+            return_value={"status": "error", "error": "upstream down"},
+        ):
+            response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        deps["cache"].release_fetch_lock.assert_called_once()
+        assert response.status_code == 502
+
+    def test_thundering_herd_polls_storage_when_lock_not_acquired(self, client, patched_deps):
+        # When the lock is held by another pod the handler must poll storage and serve
+        # from cache once the competing fetch completes, without issuing its own upstream request.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # Initial cache check: miss. First poll iteration: another pod has written it.
+        # Third call is for the digest cross-link check (is_mutable=True path); digest key exists.
+        deps["storage"].exists.side_effect = [False, True, True]
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = False  # lock held by peer
+
+        with patch("artifactapi.artifact.docker.asyncio.sleep", new_callable=AsyncMock):
+            with patch(
+                "artifactapi.artifact.proxy.cache_single_artifact",
+                new_callable=AsyncMock,
+            ) as mock_fetch:
+                response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        mock_fetch.assert_not_called()
+        assert response.status_code == 200
+
+    def test_thundering_herd_falls_through_to_fetch_if_poll_times_out(self, client, patched_deps):
+        # If the item never appears in storage during the poll window the handler must
+        # still issue its own upstream fetch as a fallback.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # All exists calls return False — item never appears during polling.
+        deps["storage"].exists.return_value = False
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = False  # lock held by peer
+
+        with patch("artifactapi.artifact.docker.asyncio.sleep", new_callable=AsyncMock):
+            with patch(
+                "artifactapi.artifact.proxy.cache_single_artifact",
+                new_callable=AsyncMock,
+                return_value={"status": "cached"},
+            ) as mock_fetch:
+                response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        mock_fetch.assert_called_once()
+        assert response.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# Docker ban_tags feature
+# ---------------------------------------------------------------------------
+
+
+class TestDockerBanTags:
+    def test_banned_tag_returns_403(self, client, patched_deps):
+        response = client.get("/v2/docker-bantags-test/library/nginx/manifests/latest")
+        assert response.status_code == 403
+        assert "latest" in response.json()["detail"]
+
+    def test_second_banned_tag_returns_403(self, client, patched_deps):
+        response = client.get("/v2/docker-bantags-test/library/nginx/manifests/edge")
+        assert response.status_code == 403
+        assert "edge" in response.json()["detail"]
+
+    def test_allowed_tag_proceeds(self, client, patched_deps):
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+
+        response = client.get("/v2/docker-bantags-test/library/nginx/manifests/1.25.3")
+        assert response.status_code == 200
+
+    def test_digest_pull_bypasses_ban(self, client, patched_deps):
+        # sha256-addressed pulls must never be blocked by the tag ban list
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = False
+
+        digest = "sha256:" + "a" * 64
+        with patch("artifactapi.artifact.proxy._fetch_last_modified", new_callable=AsyncMock, return_value=None):
+            response = client.get(f"/v2/docker-bantags-test/library/nginx/manifests/{digest}")
+        assert response.status_code == 200
+
+    def test_ban_tags_disabled_by_default(self, client, patched_deps):
+        # docker-test has no ban_tags_enabled — "latest" must pass through
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+
+        response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+        assert response.status_code == 200
+
+    def test_ban_tags_enabled_but_empty_list_allows_all(self, client, patched_deps):
+        # If ban_tags_enabled is true but ban_tags is empty nothing should be blocked.
+        # docker-test doesn't have ban_tags_enabled, but we can verify via the
+        # docker-bantags-test remote with an unlisted tag.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+
+        response = client.get("/v2/docker-bantags-test/library/nginx/manifests/stable")
+        assert response.status_code == 200
+
+    def test_ban_check_does_not_apply_to_blobs(self, client, patched_deps):
+        # Blob paths don't contain /manifests/ — the ban check must not interfere
+        deps = patched_deps
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = b"\x00" * 100
+        deps["cache"].is_mutable_file.return_value = False
+
+        with patch("artifactapi.artifact.proxy._fetch_last_modified", new_callable=AsyncMock, return_value=None):
+            response = client.get("/v2/docker-bantags-test/library/nginx/blobs/sha256:" + "b" * 64)
+        assert response.status_code == 200
+
 
 # ---------------------------------------------------------------------------
 # Generic artifact route  /api/v1/remote/{remote}/{path}