DockerHook: obtain credentials and login to Amazon ECR

airflow/providers/docker/credentials/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

airflow/providers/docker/credentials/base.py

@@ -0,0 +1,78 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.log.secrets_masker import mask_secret
+
+if TYPE_CHECKING:
+    from airflow.models import Connection
+
+
+@dataclass
+class DockerLoginCredentials:
+    """Class for keeping authentication information for Docker Registry."""
+
+    username: str
+    password: str
+    registry: str
+    email: Optional[str] = None
+    reauth: bool = False
+
+    def __post_init__(self):
+        if self.password:
+            # Mask password if it specified.
+            mask_secret(self.password)
+
+
+class BaseDockerCredentialHelper(LoggingMixin):
+    """Base class for authentication in Docker Registry.
+
+    :param conn: Reference to Docker hook connection object.
+    """
+
+    def __init__(self, *, conn: "Connection", **kwargs):
+        super().__init__()
+        self.conn = conn
+        self.conn_extra = conn.extra_dejson
+
+    @property
+    def reauth(self) -> bool:
+        """The reauth property from connection."""
+        val = self.conn_extra.get("reauth", True)
+        if isinstance(val, bool):
+            return val
+        val = str(val).lower()
+        if val in ('y', 'yes', 't', 'true', 'on', '1'):
+            return True
+        if val in ('n', 'no', 'f', 'false', 'off', '0'):
+            return False
+        raise ValueError(f"{val!r} is not a boolean-like string value.")
+
+    @property
+    def email(self) -> Optional[str]:
+        """The email for the registry account from connection."""
+        return self.conn_extra.get("email")
+
+    def get_credentials(self) -> Optional[Sequence[DockerLoginCredentials]]:
+        """
+        Method uses for return credentials to Docker Registry.
+        It might or might not use credentials from Connection.
+        """
+        raise NotImplementedError()

airflow/providers/docker/credentials/connection.py

@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import Optional, Sequence
+
+from airflow.exceptions import AirflowException
+from airflow.providers.docker.credentials.base import BaseDockerCredentialHelper, DockerLoginCredentials
+
+
+class AirflowConnectionDockerCredentialHelper(BaseDockerCredentialHelper):
+    """Default helper for authentication in Docker Registry.
+    Use Airflow Connection information.
+    """
+
+    def get_credentials(self) -> Optional[Sequence[DockerLoginCredentials]]:
+        if not self.conn.host:
+            raise AirflowException('No Docker URL provided.')
+        if not self.conn.login:
+            raise AirflowException('No username provided.')
+
+        if self.conn.port:
+            registry = f"{self.conn.host}:{self.conn.port}"
+        else:
+            registry = self.conn.host
+
+        return [
+            DockerLoginCredentials(
+                username=self.conn.login,
+                password=self.conn.password,
+                registry=registry,
+                email=self.email,
+                reauth=self.reauth,
+            )
+        ]

airflow/providers/docker/credentials/ecr.py

@@ -0,0 +1,89 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import base64
+from typing import List, Optional, Sequence, Union
+
+from airflow.exceptions import AirflowException
+from airflow.providers.docker.credentials.base import BaseDockerCredentialHelper, DockerLoginCredentials
+
+
+class EcrDockerCredentialHelper(BaseDockerCredentialHelper):
+    """
+    Authenticate into Amazon ECR (Elastic Container Registry).
+
+    This helper ignore all connection information such as password and registry host otherwise use
+    credentials retrieved from AWS API.
+
+    .. seealso::
+        - `ECR Registry Auth <https://docs.aws.amazon.com/AmazonECR/latest/userguide/registry_auth.html>`_
+        - :ref:`howto/connection:aws:configuring-the-connection`
+
+    :param aws_conn_id: The Airflow connection used for AWS credentials.
+        If this is ``None`` or empty then the default boto3 behaviour is used.
+    :param region_name: AWS Region Name.
+    :param registry_ids: A list of Amazon Web Services account IDs that are associated with the registries
+        for which to get AuthorizationData objects.
+        If you do not specify a registry, the default registry is assumed.
+    """
+
+    def __init__(
+        self,
+        *,
+        aws_conn_id: Optional[str] = "aws_default",
+        region_name: Optional[str] = None,
+        registry_ids: Optional[Union[str, List[str]]] = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.aws_conn_id = aws_conn_id
+        self.region_name = region_name
+        if isinstance(registry_ids, str):
+            registry_ids = [registry_ids]
+        self.registry_ids = registry_ids
+
+    def get_credentials(self) -> Optional[Sequence[DockerLoginCredentials]]:
+        try:
+            from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+        except ImportError:
+            raise AirflowException(
+                "apache-airflow-providers-amazon not installed, run: "
+                "pip install 'apache-airflow-providers-docker[amazon]'."
+            )
+
+        aws_hook = AwsBaseHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name, client_type="ecr")
+        if self.registry_ids:
+            response = aws_hook.conn.get_authorization_token(registryIds=self.registry_ids)
+        else:
+            response = aws_hook.conn.get_authorization_token()
+
+        creds = []
+        for auth_data in response["authorizationData"]:
+            username, password = base64.b64decode(auth_data["authorizationToken"]).decode("utf-8").split(":")
+            registry: str = auth_data['proxyEndpoint']
+            creds.append(
+                DockerLoginCredentials(
+                    username=username,
+                    password=password,
+                    # https://github.com/docker/docker-py/issues/2256#issuecomment-824940506
+                    registry=registry.replace("https://", ""),
+                    reauth=True,
+                )
+            )
+            self.log.info("Credentials to Amazon ECR %r expires at %s.", registry, auth_data['expiresAt'])
+
+        return creds

airflow/providers/docker/hooks/docker.py

@@ -15,18 +15,23 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
+import json
 from typing import Any, Dict, Optional
 
-from docker import APIClient  # type: ignore[attr-defined]
-from docker.constants import DEFAULT_TIMEOUT_SECONDS  # type: ignore[attr-defined]
-from docker.errors import APIError  # type: ignore[attr-defined]
+from docker import APIClient
+from docker.constants import DEFAULT_TIMEOUT_SECONDS
+from docker.errors import APIError
 
+from airflow.compat.functools import cached_property
 from airflow.exceptions import AirflowException
 from airflow.hooks.base import BaseHook
-from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.providers.docker.credentials.base import BaseDockerCredentialHelper
+from airflow.providers.docker.credentials.connection import AirflowConnectionDockerCredentialHelper
+from airflow.utils.module_loading import import_string
 
 
-class DockerHook(BaseHook, LoggingMixin):
+class DockerHook(BaseHook):
     """
     Interact with a Docker Daemon or Registry.
 
@@ -48,6 +53,16 @@ def get_ui_field_behaviour() -> Dict[str, Any]:
                 'host': 'Registry URL',
                 'login': 'Username',
             },
+            "placeholders": {
+                'extra': json.dumps(
+                    {
+                        'reauth': False,
+                        'email': 'Jane.Doe@example.org',
+                        'credential_helper': 'dotted.path.to.credential.Helper',
+                        'credential_helper_kwargs': {"foo": "bar"},
+                    }
+                )
+            },
         }
 
     def __init__(
@@ -66,46 +81,51 @@ def __init__(
 
         if not docker_conn_id:
             raise AirflowException('No Docker connection id provided')
-
-        conn = self.get_connection(docker_conn_id)
-
-        if not conn.host:
-            raise AirflowException('No Docker URL provided')
-        if not conn.login:
-            raise AirflowException('No username provided')
-        extra_options = conn.extra_dejson
-
+        self.docker_conn_id = docker_conn_id
         self.__base_url = base_url
         self.__version = version
         self.__tls = tls
         self.__timeout = timeout
-        if conn.port:
-            self.__registry = f"{conn.host}:{conn.port}"
-        else:
-            self.__registry = conn.host
-        self.__username = conn.login
-        self.__password = conn.password
-        self.__email = extra_options.get('email')
-        self.__reauth = extra_options.get('reauth') != 'no'
 
-    def get_conn(self) -> APIClient:
+    @cached_property
+    def api_client(self) -> APIClient:
+        """Create connection to docker host and login to the docker registries. (cached)"""
+        conn = self.get_connection(self.docker_conn_id)
         client = APIClient(
             base_url=self.__base_url, version=self.__version, tls=self.__tls, timeout=self.__timeout
         )
-        self.__login(client)
+
+        credential_helper = conn.extra_dejson.get("credential_helper")
+        if not credential_helper:
+            # If not specified credential helper than retrieve information from Connection.
+            credential_helper = AirflowConnectionDockerCredentialHelper
+            credential_helper_kwargs = {}
+        else:
+            credential_helper = import_string(credential_helper)
+            credential_helper_kwargs = conn.extra_dejson.get("credential_helper_kwargs", {})
+
+            if not issubclass(credential_helper, BaseDockerCredentialHelper):
+                raise TypeError(
+                    f"Your credential_helper `{credential_helper.__name__}` is not a subclass "
+                    f"of `{BaseDockerCredentialHelper.__name__}`."
+                )
+
+        for creds in credential_helper(conn=conn, **credential_helper_kwargs).get_credentials() or []:
+            try:
+                self.log.info('Login into Docker Registry: %s', creds.registry)
+                client.login(
+                    username=creds.username,
+                    password=creds.password,
+                    registry=creds.registry,
+                    email=creds.email,
+                    reauth=creds.reauth,
+                )
+                self.log.debug('Login successful')
+            except APIError as docker_error:
+                self.log.error('Docker login failed: %s', str(docker_error))
+                raise AirflowException(f'Docker login failed: {docker_error}')
         return client
 
-    def __login(self, client) -> None:
-        self.log.debug('Logging into Docker')
-        try:
-            client.login(
-                username=self.__username,
-                password=self.__password,
-                registry=self.__registry,
-                email=self.__email,
-                reauth=self.__reauth,
-            )
-            self.log.debug('Login successful')
-        except APIError as docker_error:
-            self.log.error('Docker login failed: %s', str(docker_error))
-            raise AirflowException(f'Docker login failed: {docker_error}')
+    def get_conn(self) -> APIClient:
+        """Create connection to docker host and login to the docker registries. (cached)"""
+        return self.api_client

docs/spelling_wordlist.txt

@@ -1147,6 +1147,7 @@ ReadOnlyCredentials
 readthedocs
 Realtime
 realtime
+reauth
 rebase
 Rebasing
 Rebrand

generated/provider_dependencies.json

@@ -253,7 +253,9 @@
       "apache-airflow>=2.2.0",
       "docker>=5.0.3"
     ],
-    "cross-providers-deps": []
+    "cross-providers-deps": [
+      "amazon"
+    ]
   },
   "elasticsearch": {
     "deps": [