launchpad-reviewers team mailing list archive

["Thiago F. Pappacena" <pappacena@xxxxxxxxx>]

Diff comments:

> diff --git a/lib/lp/oci/model/ociregistryclient.py b/lib/lp/oci/model/ociregistryclient.py
> index 56225da..0307dd1 100644
> --- a/lib/lp/oci/model/ociregistryclient.py
> +++ b/lib/lp/oci/model/ociregistryclient.py
> @@ -41,6 +43,8 @@ from tenacity import (
>      )
>  from zope.interface import implementer
>  
> +from lp.services.config import config as lp_config

Ok! Changing.

> +from lp.services.features import getFeatureFlag
>  from lp.oci.interfaces.ociregistryclient import (

It got confused by the try/except block a bit above this. Fixing manually.

>      BlobUploadFailed,
>      IOCIRegistryClient,
> @@ -577,29 +597,76 @@ class BearerTokenRegistryClient(RegistryHTTPClient):
>              raise
>  
>  
> -class AWSRegistryHTTPClient(RegistryHTTPClient):
> +class AWSAuthenticatorMixin:
> +    """Basic class to override the way we get credentials, exchanging
> +    registered aws_access_key_id and aws_secret_access_key with the
> +    temporary token got from AWS API.
> +    """
> +
> +    def _getClientParameters(self):
> +        if lp_config.launchpad.http_proxy:
> +            boto_config = Config(proxies={
> +                'http': lp_config.launchpad.http_proxy,
> +                'https': lp_config.launchpad.http_proxy})
> +        else:
> +            boto_config = Config()
> +        auth = self.push_rule.registry_credentials.getCredentials()
> +        username, password = auth['username'], auth.get('password')
> +        region = self._getRegion()
> +        log.info("Trying to authenticate with AWS in region %s" % region)
> +        return dict(
> +            aws_access_key_id=username,
> +            aws_secret_access_key=password, region_name=region,
> +            config=boto_config)
> +
> +    def _getBotoClient(self):
> +        params = self._getClientParameters()
> +        if not self.should_use_aws_extra_model:
> +            return boto3.client('ecr', **params)
> +        model_path = getFeatureFlag(OCI_AWS_BOT_EXTRA_MODEL_PATH)
> +        model_name = getFeatureFlag(OCI_AWS_BOT_EXTRA_MODEL_NAME)
> +        if not model_path or not model_name:
> +            log.warning(
> +                "%s or %s feature rules are not set. Using default model." %
> +                (OCI_AWS_BOT_EXTRA_MODEL_PATH, OCI_AWS_BOT_EXTRA_MODEL_NAME))
> +            return boto3.client('ecr', **params)
> +        session = boto3.Session()
> +        session._loader.search_paths.extend([model_path])
> +        return session.client(model_name, **params)
> +
> +    @property
> +    def should_use_aws_extra_model(self):
> +        """Returns True if the given registry domain requires extra boto API
> +        model.
> +        """
> +        domain = urlparse(self.push_rule.registry_url).netloc
> +        return is_aws_bearer_token_domain(domain)
>  
>      def _getRegion(self):
>          """Returns the region from the push URL domain."""
> -        domain = urlparse(self.push_rule.registry_url).netloc
> -        # The domain format should be something like
> +        push_rule = self.push_rule
> +        region = push_rule.registry_credentials.getCredentialsValue("region")
> +        if region is not None:
> +            return region
> +        # Try to guess from the domain. The format should be something like
>          # 'xxx.dkr.ecr.sa-east-1.amazonaws.com'. 'sa-east-1' is the region.
> -        return domain.split(".")[-3]
> +        domain = urlparse(self.push_rule.registry_url).netloc
> +        if re.match(r'.+\.dkr\.ecr\..+\.amazonaws\.com', domain):
> +            return domain.split(".")[-3]
> +        raise OCIRegistryAuthenticationError("Unknown AWS region.")
>  
>      @cachedproperty
>      def credentials(self):
>          """Exchange aws_access_key_id and aws_secret_access_key with the
>          authentication token that should be used when talking to ECR."""
>          try:
> -            auth = self.push_rule.registry_credentials.getCredentials()
> -            username, password = auth['username'], auth.get('password')
> -            region = self._getRegion()
> -            log.info("Trying to authenticate with AWS in region %s" % region)
> -            client = boto3.client('ecr', aws_access_key_id=username,
> -                                  aws_secret_access_key=password,
> -                                  region_name=region)
> +            client = self._getBotoClient()
>              token = client.get_authorization_token()
> -            auth_data = token["authorizationData"][0]
> +            auth_data = token["authorizationData"]
> +            # ecr-public returns a dict directly, but ecr returns list with

Ok!

> +            # one element inside. Go figure...
> +            if isinstance(auth_data, list):
> +                auth_data = auth_data[0]
>              authorization_token = auth_data['authorizationToken']
>              username, password = base64.b64decode(
>                  authorization_token).decode().split(':')
> diff --git a/lib/lp/oci/tests/test_ociregistryclient.py b/lib/lp/oci/tests/test_ociregistryclient.py
> index 65e416b..7d48ec1 100644
> --- a/lib/lp/oci/tests/test_ociregistryclient.py
> +++ b/lib/lp/oci/tests/test_ociregistryclient.py
> @@ -873,10 +881,33 @@ class TestRegistryHTTPClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
>  
>          instance = RegistryHTTPClient.getInstance(push_rule)
>          self.assertEqual(AWSRegistryHTTPClient, type(instance))
> +        self.assertFalse(instance.should_use_aws_extra_model)
> +        self.assertIsInstance(instance, RegistryHTTPClient)
> +
> +    @responses.activate
> +    def test_get_aws_bearer_token_auth_client_instance(self):
> +        self.useFixture(FeatureFixture({
> +            OCI_RECIPE_ALLOW_CREATE: 'on',
> +            OCI_AWS_BEARER_TOKEN_DOMAINS_FLAG: (
> +                'foo.domain.com fake.domain.com'),

Ok!

> +        }))
> +        credentials = self.factory.makeOCIRegistryCredentials(
> +            url="https://fake.domain.com";,
> +            credentials={
> +                'username': 'aws_access_key_id',
> +                'password': "aws_secret_access_key"})
> +        push_rule = removeSecurityProxy(self.factory.makeOCIPushRule(
> +            registry_credentials=credentials,
> +            image_name="ecr-test"))
> +
> +        instance = RegistryHTTPClient.getInstance(push_rule)
> +        self.assertEqual(AWSRegistryBearerTokenClient, type(instance))
> +        self.assertTrue(instance.should_use_aws_extra_model)
>          self.assertIsInstance(instance, RegistryHTTPClient)
>  
>      @responses.activate
>      def test_aws_credentials(self):
> +        self.pushConfig('launchpad', http_proxy='http://proxy.local.com:123')
>          boto_patch = self.useFixture(
>              MockPatch('lp.oci.model.ociregistryclient.boto3'))
>          boto = boto_patch.mock


-- 
https://code.launchpad.net/~pappacena/launchpad/+git/launchpad/+merge/394598
Your team Launchpad code reviewers is subscribed to branch ~pappacena/launchpad:public-ecr-aws.