fix: Match by digest only for combined tag@digest image lookups

When using combined tag@digest references (e.g., nginx:1.21@sha256:abc...),
Docker does NOT store the tag in RepoTags. It only stores the digest in
RepoDigests. The previous implementation required BOTH to match, which
always failed because RepoTags was empty.

This caused docker_container to pull the image on every run even when
the image with the correct digest already existed locally, breaking
idempotency.

The fix: When a digest is specified, match by digest only since it's the
authoritative identifier. The tag is informational for human readability.

Real-world example from docker image inspect:
  "RepoTags": [],  # Empty when pulled by digest\!
  "RepoDigests": ["portainer/portainer-ee@sha256:7ecf2008..."]

Updated tests to reflect the correct behavior:
- test_empty_repo_tags_matches_by_digest (the critical fix case)
- test_combined_tag_digest_matches_even_if_tag_differs
- test_multiple_tags_irrelevant_for_combined

plugins/module_utils/_common.py

@@ -27,6 +27,7 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
     DOCKER_COMMON_ARGS,
     DOCKER_MUTUALLY_EXCLUSIVE,
     DOCKER_REQUIRED_TOGETHER,
+    build_pull_arguments,
     filter_images_by_tag,
     sanitize_result,
     update_tls_hostname,
@@ -572,17 +573,22 @@ class AnsibleDockerClientBase(Client):
         """
         Pull an image
         """
-        kwargs = {
+        kwargs: dict[str, t.Any] = {
-            "tag": tag,
             "stream": True,
             "decode": True,
         }
         if image_platform is not None:
             kwargs["platform"] = image_platform
+
+        # Build pull arguments - handles combined tag@digest format
+        pull_name, pull_tag = build_pull_arguments(name, tag)
+        if pull_tag is not None:
+            kwargs["tag"] = pull_tag
+
         self.log(f"Pulling image {name}:{tag}")
         old_tag = self.find_image(name, tag)
         try:
-            for line in self.pull(name, **kwargs):
+            for line in self.pull(pull_name, **kwargs):
                 self.log(line, pretty_print=True)
                 if line.get("error"):
                     if line.get("errorDetail"):

plugins/module_utils/_common_api.py

@@ -56,6 +56,7 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
     DOCKER_COMMON_ARGS,
     DOCKER_MUTUALLY_EXCLUSIVE,
     DOCKER_REQUIRED_TOGETHER,
+    build_pull_arguments,
     filter_images_by_tag,
     sanitize_result,
     update_tls_hostname,
@@ -531,10 +532,15 @@ class AnsibleDockerClientBase(Client):
         try:
             repository, image_tag = parse_repository_tag(name)
             registry, dummy_repo_name = auth.resolve_repository_name(repository)
-            params = {
+
-                "tag": tag or image_tag or "latest",
+            # Build pull arguments - handles combined tag@digest format
-                "fromImage": repository,
+            effective_tag = tag or image_tag or "latest"
-            }
+            pull_name, pull_tag = build_pull_arguments(repository, effective_tag)
+
+            params: dict[str, t.Any] = {"fromImage": pull_name}
+            if pull_tag is not None:
+                params["tag"] = pull_tag
+
             if image_platform is not None:
                 params["platform"] = image_platform
 

plugins/module_utils/_util.py

@@ -125,13 +125,13 @@ def filter_images_by_tag(
     # The @ is for digest, but if it doesn't start with sha256:, it's combined format
     if "@" in tag and not tag.startswith("sha256:"):
         tag_part, digest_part = tag.split("@", 1)
-        lookup_tag = f"{name}:{tag_part}"
         lookup_digest = f"{name}@{digest_part}"
         for image in images:
-            repo_tags = image.get("RepoTags") or []
             repo_digests = image.get("RepoDigests") or []
-            # For combined format, match BOTH tag AND digest
+            # When digest is specified, match by digest only.
-            if lookup_tag in repo_tags and lookup_digest in repo_digests:
+            # Docker doesn't preserve the tag in RepoTags when pulling by digest,
+            # so we can't require both to match.
+            if lookup_digest in repo_digests:
                 return [image]
         return []
 
@@ -146,6 +146,36 @@ def filter_images_by_tag(
     return []
 
 
+def build_pull_arguments(
+    name: str, tag: str
+) -> tuple[str, str | None]:
+    """
+    Build the correct arguments for Docker pull operations.
+
+    Docker SDK and API don't accept combined "tag@digest" in the tag parameter.
+    This function handles the three formats:
+    - Tag-only: "v1.0" -> (name, "v1.0")
+    - Digest-only: "sha256:abc..." -> (name, "sha256:abc...")
+    - Combined tag@digest: "v1.0@sha256:abc..." -> ("name:v1.0@sha256:abc...", None)
+
+    Args:
+        name: Repository name (e.g., "nginx", "ghcr.io/user/repo")
+        tag: Tag, digest, or combined tag@digest
+
+    Returns:
+        Tuple of (pull_name, pull_tag) where:
+        - For combined format: pull_name is full reference, pull_tag is None
+        - For other formats: pull_name is just the name, pull_tag is the tag/digest
+    """
+    # Handle combined tag@digest format (e.g., "v1.0@sha256:abc123")
+    # The @ indicates a digest, but if it doesn't START with sha256:, it's combined
+    if "@" in tag and not tag.startswith("sha256:"):
+        tag_part, digest = tag.split("@", 1)
+        return f"{name}:{tag_part}@{digest}", None
+    else:
+        return name, tag
+
+
 def is_valid_tag(tag: str, allow_empty: bool = False) -> bool:
     """Check whether the given string is a valid docker tag name."""
     if not tag:

plugins/modules/docker_image.py

@@ -405,6 +405,7 @@ from ansible_collections.community.docker.plugins.module_utils._image_archive im
 )
 from ansible_collections.community.docker.plugins.module_utils._util import (
     DockerBaseClass,
+    build_pull_arguments,
     clean_dict_booleans_for_docker_api,
     is_image_name_id,
     is_valid_tag,
@@ -784,6 +785,10 @@ class ImageManager(DockerBaseClass):
                         push_repository, push_tag = parse_repository_tag(
                             push_repository
                         )
+
+                    # Handle combined tag@digest format - Docker API doesn't accept it in tag param
+                    push_name, push_tag_param = build_pull_arguments(push_repository, push_tag)
+
                     push_registry, dummy = resolve_repository_name(push_repository)
                     headers = {}
                     header = get_config_header(self.client, push_registry)
@@ -792,12 +797,17 @@ class ImageManager(DockerBaseClass):
                         # See https://github.com/moby/moby/issues/50614.
                         header = base64.urlsafe_b64encode(b"{}")
                     headers["X-Registry-Auth"] = header
+
+                    params: dict[str, t.Any] = {}
+                    if push_tag_param is not None:
+                        params["tag"] = push_tag_param
+
                     response = self.client._post_json(
-                        self.client._url("/images/{0}/push", push_repository),
+                        self.client._url("/images/{0}/push", push_name),
                         data=None,
                         headers=headers,
                         stream=True,
-                        params={"tag": push_tag},
+                        params=params,
                     )
                     self.client._raise_for_status(response)
                     for line in self.client._stream_helper(response, decode=True):

tests/unit/plugins/module_utils/test__util.py

@@ -544,9 +544,11 @@ class TestFilterImagesByTag:
         assert len(result) == 0
 
     # ========== Combined tag@digest tests ==========
+    # When a digest is specified, we match by digest ONLY because Docker doesn't
+    # preserve tags in RepoTags when pulling by digest. The tag is informational.
 
-    def test_combined_tag_digest_matches_both(self) -> None:
+    def test_combined_tag_digest_matches_by_digest(self) -> None:
-        """Combined tag@digest should match when BOTH tag AND digest match."""
+        """Combined tag@digest should match when digest matches (tag is informational)."""
         image = _create_mock_image(
             repo_tags=["nginx:1.21"],
             repo_digests=["nginx@sha256:" + SHA],
@@ -566,14 +568,18 @@ class TestFilterImagesByTag:
         )
         assert len(result) == 1
 
-    def test_combined_tag_digest_fails_if_tag_mismatch(self) -> None:
+    def test_combined_tag_digest_matches_even_if_tag_differs(self) -> None:
-        """Combined tag@digest should NOT match if tag doesn't match."""
+        """Combined tag@digest SHOULD match even if tag differs - digest is authoritative.
+
+        Docker doesn't store tags when pulling by digest, so requiring tag match
+        would cause lookups to always fail. The digest is the authoritative identifier.
+        """
         image = _create_mock_image(
-            repo_tags=["nginx:1.20"],  # Different tag
+            repo_tags=["nginx:1.20"],  # Different tag - doesn't matter
             repo_digests=["nginx@sha256:" + SHA],
         )
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
-        assert len(result) == 0
+        assert len(result) == 1  # Should match by digest
 
     def test_combined_tag_digest_fails_if_digest_mismatch(self) -> None:
         """Combined tag@digest should NOT match if digest doesn't match."""
@@ -585,27 +591,40 @@ class TestFilterImagesByTag:
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 0
 
-    def test_combined_tag_digest_fails_if_both_mismatch(self) -> None:
+    def test_combined_tag_digest_fails_if_digest_mismatch_tag_irrelevant(self) -> None:
-        """Combined tag@digest should NOT match if both tag and digest don't match."""
+        """Combined tag@digest should NOT match if digest doesn't match, regardless of tag."""
         other_sha = "a" * 64
         image = _create_mock_image(
-            repo_tags=["nginx:1.20"],  # Different tag
+            repo_tags=["nginx:1.20"],  # Different tag - irrelevant
-            repo_digests=["nginx@sha256:" + other_sha],  # Different digest
+            repo_digests=["nginx@sha256:" + other_sha],  # Different digest - this matters
         )
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 0
 
     # ========== Edge cases ==========
 
-    def test_empty_repo_tags(self) -> None:
+    def test_empty_repo_tags_matches_by_digest(self) -> None:
-        """Handle images where RepoTags is empty or None."""
+        """Handle images where RepoTags is empty or None - should still match by digest.
+
+        This is the critical real-world case: Docker doesn't store tags when pulling
+        by digest, so RepoTags is often empty. We must match by digest only.
+        """
         image = _create_mock_image(
-            repo_tags=None,
+            repo_tags=None,  # Empty - typical when pulled by digest
             repo_digests=["nginx@sha256:" + SHA],
         )
-        # Combined format should not match if RepoTags is empty
+        # Combined format SHOULD match because digest matches
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
-        assert len(result) == 0
+        assert len(result) == 1
+
+    def test_empty_repo_tags_list_matches_by_digest(self) -> None:
+        """Empty list RepoTags should still match by digest."""
+        image = _create_mock_image(
+            repo_tags=[],  # Empty list - typical when pulled by digest
+            repo_digests=["nginx@sha256:" + SHA],
+        )
+        result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
+        assert len(result) == 1
 
     def test_empty_repo_digests(self) -> None:
         """Handle images where RepoDigests is empty or None."""
@@ -613,22 +632,24 @@ class TestFilterImagesByTag:
             repo_tags=["nginx:1.21"],
             repo_digests=None,
         )
-        # Combined format should not match if RepoDigests is empty
+        # Combined format should not match if RepoDigests is empty (digest is authoritative)
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 0
 
-    def test_multiple_tags(self) -> None:
+    def test_multiple_tags_irrelevant_for_combined(self) -> None:
-        """Image with multiple tags should match on any of them."""
+        """For combined tag@digest, tags are irrelevant - digest is authoritative."""
         image = _create_mock_image(
             repo_tags=["nginx:1.21", "nginx:latest", "nginx:stable"],
             repo_digests=["nginx@sha256:" + SHA],
         )
-        # Should match on 1.21
+        # All of these should match because the digest matches
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 1
-        # Should also match on latest
         result = filter_images_by_tag("nginx", f"latest@sha256:{SHA}", [image])
         assert len(result) == 1
+        # Even a non-existent tag should match because digest matches
+        result = filter_images_by_tag("nginx", f"nonexistent@sha256:{SHA}", [image])
+        assert len(result) == 1
 
     def test_multiple_digests(self) -> None:
         """Image with multiple digests should match on any of them."""
@@ -637,12 +658,15 @@ class TestFilterImagesByTag:
             repo_tags=["nginx:1.21"],
             repo_digests=["nginx@sha256:" + SHA, "nginx@sha256:" + other_sha],
         )
-        # Should match on first digest
+        # Should match on first digest (tag is irrelevant)
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 1
         # Should also match on second digest
         result = filter_images_by_tag("nginx", f"1.21@sha256:{other_sha}", [image])
         assert len(result) == 1
+        # Should match even with different tag
+        result = filter_images_by_tag("nginx", f"different@sha256:{SHA}", [image])
+        assert len(result) == 1
 
     def test_port_in_registry_name(self) -> None:
         """Registry with port number should not be confused with tag."""

tests/unit/plugins/module_utils/test_pull_image.py

@@ -0,0 +1,218 @@
+# Copyright (c) Ansible Project
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""
+Tests for build_pull_arguments handling of tag@digest format.
+
+When users specify an image with both tag and digest (e.g., nginx:1.21@sha256:abc...),
+the parse_repository_tag function returns ("nginx", "1.21@sha256:abc...").
+
+The Docker SDK and API don't accept "tag@digest" in the tag parameter.
+build_pull_arguments handles this by:
+- Returning full reference as name when tag contains @
+- Returning None for tag in that case
+
+This function is used by:
+- pull_image() in _common.py and _common_api.py (for pulling images)
+- push_image() in docker_image.py (for pushing images)
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from ansible_collections.community.docker.plugins.module_utils._util import (
+    build_pull_arguments,
+)
+
+
+SHA = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+
+
+class TestBuildPullArguments(unittest.TestCase):
+    """Test cases for build_pull_arguments function."""
+
+    # ========== Tag-only tests ==========
+
+    def test_tag_only_simple(self) -> None:
+        """Tag-only should return name and tag separately."""
+        assert build_pull_arguments("nginx", "1.21") == ("nginx", "1.21")
+
+    def test_tag_only_latest(self) -> None:
+        """Latest tag should return name and tag separately."""
+        assert build_pull_arguments("nginx", "latest") == ("nginx", "latest")
+
+    def test_tag_only_with_registry(self) -> None:
+        """Tag-only with registry should return name and tag separately."""
+        assert build_pull_arguments("ghcr.io/user/repo", "v1.0") == (
+            "ghcr.io/user/repo",
+            "v1.0",
+        )
+
+    def test_tag_only_with_registry_port(self) -> None:
+        """Tag-only with registry port should return name and tag separately."""
+        assert build_pull_arguments("localhost:5000/myapp", "v2.0") == (
+            "localhost:5000/myapp",
+            "v2.0",
+        )
+
+    # ========== Digest-only tests ==========
+
+    def test_digest_only(self) -> None:
+        """Digest-only (sha256:...) should return name and digest separately."""
+        assert build_pull_arguments("nginx", f"sha256:{SHA}") == (
+            "nginx",
+            f"sha256:{SHA}",
+        )
+
+    def test_digest_only_with_registry(self) -> None:
+        """Digest-only with registry should return name and digest separately."""
+        assert build_pull_arguments("ghcr.io/user/repo", f"sha256:{SHA}") == (
+            "ghcr.io/user/repo",
+            f"sha256:{SHA}",
+        )
+
+    # ========== Combined tag@digest tests ==========
+
+    def test_combined_tag_digest_simple(self) -> None:
+        """Combined tag@digest should return full reference with None tag."""
+        assert build_pull_arguments("nginx", f"1.21@sha256:{SHA}") == (
+            f"nginx:1.21@sha256:{SHA}",
+            None,
+        )
+
+    def test_combined_tag_digest_with_user_repo(self) -> None:
+        """Combined tag@digest with user/repo should return full reference."""
+        assert build_pull_arguments("portainer/portainer-ee", f"2.35.0-alpine@sha256:{SHA}") == (
+            f"portainer/portainer-ee:2.35.0-alpine@sha256:{SHA}",
+            None,
+        )
+
+    def test_combined_tag_digest_with_ghcr(self) -> None:
+        """Combined tag@digest with GHCR should return full reference."""
+        assert build_pull_arguments("ghcr.io/gethomepage/homepage", f"v1.7@sha256:{SHA}") == (
+            f"ghcr.io/gethomepage/homepage:v1.7@sha256:{SHA}",
+            None,
+        )
+
+    def test_combined_tag_digest_with_registry_port(self) -> None:
+        """Combined tag@digest with registry port should return full reference."""
+        assert build_pull_arguments("localhost:5000/myapp", f"v2.0@sha256:{SHA}") == (
+            f"localhost:5000/myapp:v2.0@sha256:{SHA}",
+            None,
+        )
+
+    # ========== Edge cases ==========
+
+    def test_empty_tag_part_before_digest(self) -> None:
+        """Handle edge case where tag part is empty like '@sha256:...'."""
+        # This is an unusual case but should still work
+        assert build_pull_arguments("nginx", f"@sha256:{SHA}") == (
+            f"nginx:@sha256:{SHA}",
+            None,
+        )
+
+    def test_tag_with_hyphen_and_digest(self) -> None:
+        """Tag with hyphens and digest should work."""
+        assert build_pull_arguments("nginx", f"1.21-alpine@sha256:{SHA}") == (
+            f"nginx:1.21-alpine@sha256:{SHA}",
+            None,
+        )
+
+    def test_tag_with_dots_and_digest(self) -> None:
+        """Tag with dots and digest should work."""
+        assert build_pull_arguments("nginx", f"1.21.0@sha256:{SHA}") == (
+            f"nginx:1.21.0@sha256:{SHA}",
+            None,
+        )
+
+
+class TestBuildPullArgumentsIntegration(unittest.TestCase):
+    """Integration tests with parse_repository_tag."""
+
+    def test_full_flow_docker_hub(self) -> None:
+        """Test parse_repository_tag -> build_pull_arguments for Docker Hub."""
+        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
+            parse_repository_tag,
+        )
+
+        # User specifies: portainer/portainer-ee:2.35.0-alpine@sha256:abc...
+        image_ref = f"portainer/portainer-ee:2.35.0-alpine@sha256:{SHA}"
+        repo, tag = parse_repository_tag(image_ref)
+
+        # parse_repository_tag returns repo and combined tag@digest
+        assert repo == "portainer/portainer-ee"
+        assert tag == f"2.35.0-alpine@sha256:{SHA}"
+
+        # build_pull_arguments converts to full reference
+        pull_name, pull_tag = build_pull_arguments(repo, tag)
+        assert pull_name == f"portainer/portainer-ee:2.35.0-alpine@sha256:{SHA}"
+        assert pull_tag is None
+
+    def test_full_flow_ghcr(self) -> None:
+        """Test parse_repository_tag -> build_pull_arguments for GHCR."""
+        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
+            parse_repository_tag,
+        )
+
+        # User specifies: ghcr.io/gethomepage/homepage:v1.7@sha256:abc...
+        image_ref = f"ghcr.io/gethomepage/homepage:v1.7@sha256:{SHA}"
+        repo, tag = parse_repository_tag(image_ref)
+
+        assert repo == "ghcr.io/gethomepage/homepage"
+        assert tag == f"v1.7@sha256:{SHA}"
+
+        pull_name, pull_tag = build_pull_arguments(repo, tag)
+        assert pull_name == f"ghcr.io/gethomepage/homepage:v1.7@sha256:{SHA}"
+        assert pull_tag is None
+
+    def test_full_flow_private_registry_with_port(self) -> None:
+        """Test full flow for private registry with port number."""
+        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
+            parse_repository_tag,
+        )
+
+        # User specifies: localhost:5000/myapp:v2.0@sha256:abc...
+        image_ref = f"localhost:5000/myapp:v2.0@sha256:{SHA}"
+        repo, tag = parse_repository_tag(image_ref)
+
+        # Port should be part of repo, not confused with tag
+        assert repo == "localhost:5000/myapp"
+        assert tag == f"v2.0@sha256:{SHA}"
+
+        pull_name, pull_tag = build_pull_arguments(repo, tag)
+        assert pull_name == f"localhost:5000/myapp:v2.0@sha256:{SHA}"
+        assert pull_tag is None
+
+    def test_full_flow_tag_only(self) -> None:
+        """Test full flow for tag-only (no digest)."""
+        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
+            parse_repository_tag,
+        )
+
+        image_ref = "nginx:1.21"
+        repo, tag = parse_repository_tag(image_ref)
+
+        assert repo == "nginx"
+        assert tag == "1.21"
+
+        pull_name, pull_tag = build_pull_arguments(repo, tag)
+        assert pull_name == "nginx"
+        assert pull_tag == "1.21"
+
+    def test_full_flow_digest_only(self) -> None:
+        """Test full flow for digest-only (no tag)."""
+        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
+            parse_repository_tag,
+        )
+
+        image_ref = f"nginx@sha256:{SHA}"
+        repo, tag = parse_repository_tag(image_ref)
+
+        assert repo == "nginx"
+        assert tag == f"sha256:{SHA}"
+
+        pull_name, pull_tag = build_pull_arguments(repo, tag)
+        assert pull_name == "nginx"
+        assert pull_tag == f"sha256:{SHA}"