Set X-Registry-Auth header on manifest push and bump to new API

podman/api/__init__.py

@@ -3,7 +3,7 @@
 from podman.api.cached_property import cached_property
 from podman.api.client import APIClient
 from podman.api.api_versions import VERSION, COMPATIBLE_VERSION
-from podman.api.http_utils import prepare_body, prepare_filters
+from podman.api.http_utils import encode_auth_header, prepare_body, prepare_filters
 from podman.api.parse_utils import (
     decode_header,
     frames,
@@ -27,6 +27,7 @@
     'cached_property',
     'create_tar',
     'decode_header',
+    'encode_auth_header',
     'frames',
     'parse_repository',
     'prepare_body',

podman/api/http_utils.py

@@ -100,5 +100,5 @@ def _filter_values(mapping: Mapping[str, Any], recursion=False) -> dict[str, Any
     return canonical
 
 
-def encode_auth_header(auth_config: dict[str, str]) -> str:
+def encode_auth_header(auth_config: dict[str, str]) -> bytes:
     return base64.urlsafe_b64encode(json.dumps(auth_config).encode('utf-8'))

podman/domain/images_manager.py

@@ -13,7 +13,6 @@
 
 from podman import api
 from podman.api.parse_utils import parse_repository
-from podman.api.http_utils import encode_auth_header
 from podman.domain.images import Image
 from podman.domain.images_build import BuildMixin
 from podman.domain.json_stream import json_stream
@@ -264,7 +263,7 @@ def push(
 
         headers = {
             # A base64url-encoded auth configuration
-            "X-Registry-Auth": encode_auth_header(auth_config) if auth_config else ""
+            "X-Registry-Auth": api.encode_auth_header(auth_config) if auth_config else ""
         }
 
         params = {
@@ -359,7 +358,7 @@ def pull(
 
         headers = {
             # A base64url-encoded auth configuration
-            "X-Registry-Auth": encode_auth_header(auth_config) if auth_config else ""
+            "X-Registry-Auth": api.encode_auth_header(auth_config) if auth_config else ""
         }
 
         params = {

podman/domain/manifests.py

@@ -97,22 +97,40 @@ def push(
         self,
         destination: str,
         all: Optional[bool] = None,  # pylint: disable=redefined-builtin
+        **kwargs,
     ) -> None:
         """Push a manifest list or image index to a registry.
 
         Args:
             destination: Target for push.
             all: Push all images.
 
+        Keyword Args:
+            auth_config (Mapping[str, str]: Override configured credentials. Must include
+                username and password keys.
+
         Raises:
             NotFound: when the Manifest could not be found
             APIError: when service reports an error
         """
+        auth_config: Optional[dict[str, str]] = kwargs.get("auth_config")
+
+        headers = {
+            # A base64url-encoded auth configuration
+            "X-Registry-Auth": api.encode_auth_header(auth_config) if auth_config else ""
+        }
+
         params = {
             "all": all,
             "destination": destination,
         }
-        response = self.client.post(f"/manifests/{self.quoted_name}/push", params=params)
+
+        destination_quoted = urllib.parse.quote_plus(destination)
+        response = self.client.post(
+            f"/manifests/{self.quoted_name}/registry/{destination_quoted}",
+            params=params,
+            headers=headers,
+        )
         response.raise_for_status()
 
     def remove(self, digest: str) -> None:

podman/tests/unit/test_api_utils.py

@@ -164,6 +164,15 @@ def test_prepare_body_dict_empty_string(self):
 
         self.assertDictEqual(payload, actual_dict)
 
+    def test_encode_auth_header(self):
+        auth_config = {
+            "username": "user",
+            "password": "pass",
+        }
+        expected = b"eyJ1c2VybmFtZSI6ICJ1c2VyIiwgInBhc3N3b3JkIjogInBhc3MifQ=="
+        actual = api.encode_auth_header(auth_config)
+        self.assertEqual(expected, actual)
+
 
 if __name__ == '__main__':
     unittest.main()

podman/tests/unit/test_manifests.py

@@ -1,8 +1,15 @@
 import unittest
 
+import requests_mock
+
 from podman import PodmanClient, tests
 from podman.domain.manifests import Manifest, ManifestsManager
 
+FIRST_MANIFEST = {
+    "Id": "326dd9d7add24646a389e8eaa82125294027db2332e49c5828d96312c5d773ab",
+    "names": "quay.io/fedora:latest",
+}
+
 
 class ManifestTestCase(unittest.TestCase):
     def setUp(self) -> None:
@@ -23,6 +30,34 @@ def test_name(self):
         manifest = Manifest()
         self.assertIsNone(manifest.name)
 
+    @requests_mock.Mocker()
+    def test_push(self, mock):
+        adapter = mock.post(
+            tests.LIBPOD_URL + "/manifests/quay.io%2Ffedora%3Alatest/registry/quay.io%2Ffedora%3Av1"
+        )
+
+        manifest = Manifest(attrs=FIRST_MANIFEST, client=self.client.api)
+        manifest.push(destination="quay.io/fedora:v1")
+
+        self.assertTrue(adapter.called_once)
+
+    @requests_mock.Mocker()
+    def test_push_with_auth(self, mock):
+        adapter = mock.post(
+            tests.LIBPOD_URL
+            + "/manifests/quay.io%2Ffedora%3Alatest/registry/quay.io%2Ffedora%3Av1",
+            request_headers={
+                "X-Registry-Auth": b"eyJ1c2VybmFtZSI6ICJ1c2VyIiwgInBhc3N3b3JkIjogInBhc3MifQ=="
+            },
+        )
+
+        manifest = Manifest(attrs=FIRST_MANIFEST, client=self.client.api)
+        manifest.push(
+            destination="quay.io/fedora:v1", auth_config={"username": "user", "password": "pass"}
+        )
+
+        self.assertTrue(adapter.called_once)
+
 
 if __name__ == '__main__':
     unittest.main()