chore: merge upstream into stacklet/integration and stamp 0.9.53a1

c7n/filters/iamaccess.py

@@ -257,7 +257,7 @@ def normalize_conditions(self, s):
             'ArnLike',
             'IpAddress',
             'NotIpAddress')
-        set_conditions = ('ForAllValues', 'ForAnyValues')
+        set_conditions = ('ForAllValues', 'ForAnyValue')
 
         for s_cond_op in list(s['Condition'].keys()):
             if s_cond_op not in conditions:

c7n/filters/waf.py

@@ -223,8 +223,14 @@ def _legacy_match(self, r_acl):
     # based on the scope
     def _get_web_acls(self, scope):
         if self._cached_web_acls is None:
-            self._cached_web_acls = self.manager.get_resource_manager('wafv2').resources(
-                query=dict(Scope=scope),
+            # Scope must be supplied as resource manager data rather than a query
+            # kwarg. The wafv2 manager derives both the API Scope and the region
+            # used for the call from its own data (CLOUDFRONT web acls are global
+            # and must be addressed via us-east-1), and a Scope passed via
+            # resources(query=...) is overwritten by the manager's own value.
+            wafv2 = self.manager.get_resource_manager(
+                'wafv2', {'query': [{'Scope': scope}]})
+            self._cached_web_acls = wafv2.resources(
                 # required to get the additional detail needed for this filter (e.g. Rules), but
                 # the legacy mode does not require additional detail
                 augment=(not self._is_legacy)

c7n/lookup.py

@@ -6,28 +6,52 @@
 import copy
 
 
+class Skip:
+    """Sentinel returned by Lookup.resolve_value to omit a value entirely."""
+
+    __slots__ = ()
+
+    def __repr__(self):
+        return 'Lookup.SKIP'
+
+
 class Lookup:
     RESOURCE_SOURCE = 'resource'
 
+    SKIP = Skip()
+
+    # A lookup needs a key, a fallback, or both. With a key it resolves against
+    # the resource and falls back to 'default-value' when the path yields
+    # nothing. With no key it is just the fallback -- which callers that pass a
+    # name to resolve_value treat as a conditional default, writing it only
+    # when that name is not already present on the resource.
     schema = {
         'type': 'object',
-        'oneOf': [
-            {
-                'properties': {
-                    'type': {'type': 'string', 'enum': [RESOURCE_SOURCE]},
-                    'key': {'type': 'string'}
-                },
-                'additionalProperties': False,
-                'required': ['type', 'key']
-            }
+        'properties': {
+            'type': {'type': 'string', 'enum': [RESOURCE_SOURCE]},
+            'key': {'type': 'string'},
+            'default-value': {},
+        },
+        'additionalProperties': False,
+        'required': ['type'],
+        'anyOf': [
+            {'required': ['key']},
+            {'required': ['default-value']},
         ]
     }
 
     @staticmethod
-    def lookup_type(schema):
+    def lookup_type(schema, default_schema=None):
+        """Schema for a value that may be given directly or looked up.
+
+        ``default_schema`` types the lookup's fallback, and defaults to
+        ``schema``. Pass it when the fallback is narrower than the value --
+        tag values, for instance, accept unquoted YAML scalars for backwards
+        compatibility, but a fallback is always written as a string.
+        """
         lookup_schema = copy.deepcopy(Lookup.schema)
-        for lookup in lookup_schema['oneOf']:
-            lookup['properties']['default-value'] = schema
+        lookup_schema['properties']['default-value'] = (
+            schema if default_schema is None else default_schema)
 
         return {
             'oneOf': [
@@ -36,6 +60,30 @@ def lookup_type(schema):
             ]
         }
 
+    @staticmethod
+    def resolve_value(spec, resource, name=None, current_names=()):
+        """Resolve a value spec against a resource.
+
+        Handles both lookup forms and plain scalars:
+
+        - a scalar is returned unchanged
+        - ``{type: resource, key: <jmespath>}`` resolves against the resource,
+          falling back to ``default-value`` when the path yields nothing
+        - ``{type: resource, default-value: <x>}`` (no key) is a conditional
+          default: it resolves to ``x`` only when ``name`` is absent from
+          ``current_names``, and to ``Lookup.SKIP`` otherwise.
+        """
+        if not Lookup.is_lookup(spec):
+            return spec
+        if 'key' not in spec and name in current_names:
+            return Lookup.SKIP
+        return Lookup.extract(spec, resource)
+
+    @staticmethod
+    def has_lookups(spec_map):
+        """True if any value in the mapping is a lookup rather than a scalar."""
+        return any(Lookup.is_lookup(v) for v in spec_map.values())
+
     @staticmethod
     def extract(source, data=None):
         if Lookup.is_lookup(source):
@@ -54,10 +102,11 @@ def get_value(source, data=None):
 
     @staticmethod
     def get_value_from_resource(source, resource):
-        value = jmespath_search(source['key'], resource)
+        if 'key' in source:
+            value = jmespath_search(source['key'], resource)
+            if value is not None:
+                return value
 
-        if value is not None:
-            return value
         if 'default-value' not in source:
             raise Exception('Lookup for key, {}, returned None'.format(source['key']))
         else:

c7n/query.py

@@ -461,13 +461,14 @@ class QueryResourceManager(ResourceManager, metaclass=QueryMeta):
 
     retry = staticmethod(
         get_retry((
-            'TooManyRequestsException',
-            'ThrottlingException',
+            'Client.RequestLimitExceeded',
+            'InternalServerException',
             'RequestLimitExceeded',
             'Throttled',
             'ThrottledException',
             'Throttling',
-            'Client.RequestLimitExceeded')))
+            'ThrottlingException',
+            'TooManyRequestsException')))
 
     source_mapping = sources
 

c7n/resources/asg.py

@@ -20,9 +20,11 @@
 from c7n.manager import resources
 from c7n import query
 from c7n.resources.securityhub import PostFinding
-from c7n.tags import TagActionFilter, DEFAULT_TAG, TagCountFilter, TagTrim, TagDelayedAction
+from c7n.tags import (
+    TagActionFilter, DEFAULT_TAG, TagCountFilter, TagTrim, TagDelayedAction,
+    TagValueResolver, has_dynamic_tag_values, tag_value_schema)
 from c7n.utils import (
-    FormatDate, local_session, type_schema, chunks, get_retry, select_keys)
+    local_session, type_schema, chunks, get_retry, select_keys)
 
 from .ec2 import deserialize_user_data
 
@@ -1104,7 +1106,7 @@ def process_resource_set(self, client, asgs, tags):
 
 @ASG.action_registry.register('tag')
 @ASG.action_registry.register('mark')
-class Tag(Action):
+class Tag(TagValueResolver, Action):
     """Action to add a tag to an ASG
 
     The *propagate* parameter can be used to specify that the tag being added
@@ -1125,13 +1127,32 @@ class Tag(Action):
                     key: OwnerName
                     value: OwnerName
                     propagate: true
+
+    Tag values may also be looked up per-asg from the asg's attributes:
+
+    .. code-block:: yaml
+
+            policies:
+              - name: asg-tag-owner
+                resource: asg
+                actions:
+                  - type: tag
+                    propagate: true
+                    tags:
+                      Owner:
+                        type: resource
+                        key: "Tags[?Key=='Team'].Value | [0]"  # from the Team tag
+                        default-value: unknown
+                      CostCenter:
+                        type: resource
+                        default-value: unassigned  # set only if CostCenter absent
     """
 
     schema = type_schema(
         'tag',
         key={'type': 'string'},
-        value={'type': 'string'},
-        tags={'type': 'object'},
+        value=tag_value_schema(),
+        tags={'type': 'object', 'additionalProperties': tag_value_schema()},
         # Backwards compatibility
         tag={'type': 'string'},
         msg={'type': 'string'},
@@ -1141,34 +1162,57 @@ class Tag(Action):
     permissions = ('autoscaling:CreateOrUpdateTags',)
     batch_size = 1
 
-    def get_tag_set(self):
-        tags = []
+    def get_tag_spec_map(self):
+        spec_map = {}
+        tags = self.data.get('tags', {})
         key = self.data.get('key', self.data.get('tag', DEFAULT_TAG))
-        value = self.data.get(
-            'value', self.data.get(
-                'msg', 'AutoScaleGroup does not meet policy guidelines'))
+        value = self.data.get('value', self.data.get('msg'))
+        # The mark default only applies to the single-tag form. A policy that
+        # spells out a tags mapping gets that mapping and nothing else --
+        # otherwise every asg picks up a marker tag it never asked for, and a
+        # conditional default can never resolve to an empty payload.
+        if value is None and not tags:
+            value = 'AutoScaleGroup does not meet policy guidelines'
         if key and value:
-            tags.append({'Key': key, 'Value': value})
+            spec_map[key] = value
+
+        spec_map.update(tags)
 
-        for k, v in self.data.get('tags', {}).items():
-            tags.append({'Key': k, 'Value': v})
+        return spec_map
 
-        return tags
+    def resolve_tag_sets(self, asgs, spec_map):
+        """Yield (asg_set, tags) pairs to submit.
+
+        Static values are shared by every asg, so they go out in batches.
+        Dynamic values are resolved per asg and then regrouped, so asgs that
+        land on the same payload still batch together, and an asg whose tags
+        all resolve to skip drops out entirely.
+        """
+        if not has_dynamic_tag_values(spec_map):
+            tags = [{'Key': k, 'Value': v} for k, v in spec_map.items()]
+            self.interpolate_values(tags)
+            for asg_set in chunks(asgs, self.batch_size):
+                yield asg_set, tags
+            return
+
+        for asg_set, resolved in self.resolve_and_group(asgs, spec_map):
+            tags = [{'Key': k, 'Value': v} for k, v in resolved.items()]
+            for chunk in chunks(asg_set, self.batch_size):
+                yield chunk, tags
 
     def process(self, asgs):
-        tags = self.get_tag_set()
+        spec_map = self.get_tag_spec_map()
         error = None
 
-        self.interpolate_values(tags)
-
         client = self.get_client()
         with self.executor_factory(max_workers=2) as w:
             futures = {}
-            for asg_set in chunks(asgs, self.batch_size):
+            for asg_set, tags in self.resolve_tag_sets(asgs, spec_map):
                 futures[w.submit(
-                    self.process_resource_set, client, asg_set, tags)] = asg_set
+                    self.process_resource_set, client, asg_set, tags)] = (
+                        asg_set, tags)
             for f in as_completed(futures):
-                asg_set = futures[f]
+                asg_set, tags = futures[f]
                 if f.exception():
                     self.log.exception(
                         "Exception tagging tag:%s error:%s asg:%s" % (
@@ -1194,14 +1238,6 @@ def process_resource_set(self, client, asgs, tags):
                 a.setdefault('Tags', []).append(atags)
         self.manager.retry(client.create_or_update_tags, Tags=tag_params)
 
-    def interpolate_values(self, tags):
-        params = {
-            'account_id': self.manager.config.account_id,
-            'now': FormatDate.utcnow(),
-            'region': self.manager.config.region}
-        for t in tags:
-            t['Value'] = t['Value'].format(**params)
-
     def get_client(self):
         return local_session(self.manager.session_factory).client('autoscaling')
 

c7n/resources/aws.py

@@ -726,12 +726,22 @@ def transfer(self):
             self.ctx.session_factory(region=bucket_region, assume=False).client('s3'))
         return self._transfer
 
+    def get_sse_args(self):
+        # encryption configurable via output url query params, eg
+        # s3://bucket/prefix?sse=aws:kms&kms-key=alias/custodian
+        kms_key = self.config.get('kms-key')
+        sse = self.config.get('sse') or (kms_key and 'aws:kms') or 'AES256'
+        args = {'ServerSideEncryption': sse}
+        if sse == 'aws:kms' and kms_key:
+            args['SSEKMSKeyId'] = kms_key
+        return args
+
     def upload_file(self, path, key):
+        extra_args = {'ACL': 'bucket-owner-full-control'}
+        extra_args.update(self.get_sse_args())
         self.transfer.upload_file(
             path, self.bucket, key,
-            extra_args={
-                'ACL': 'bucket-owner-full-control',
-                'ServerSideEncryption': 'AES256'})
+            extra_args=extra_args)
 
 
 @clouds.register('aws')

c7n/resources/cleanrooms.py

@@ -1,10 +1,12 @@
 # Copyright The Cloud Custodian Authors.
 # SPDX-License-Identifier: Apache-2.0
 
+from c7n.filters import ListItemFilter
 from c7n.manager import resources
 from c7n.query import (
     ChildResourceManager, QueryResourceManager, TypeInfo,
     DescribeWithResourceTags)
+from c7n.utils import local_session, type_schema
 
 
 @resources.register('cleanrooms-collaboration')
@@ -27,6 +29,53 @@ class resource_type(TypeInfo):
     source_mapping = {'describe': DescribeWithResourceTags}
 
 
+@CleanRoomsCollaboration.filter_registry.register('members')
+class CleanRoomsCollaborationMemberFilter(ListItemFilter):
+    """Filter Clean Rooms collaborations on their members details.
+
+    :example:
+
+    .. code-block:: yaml
+
+        policies:
+          - name: cleanrooms-collaboration-with-members
+            resource: cleanrooms-collaboration
+            filters:
+              - type: members
+                attrs:
+                  - type: value
+                    key: status
+                    value: ACTIVE
+                  - type: value
+                    key: abilities
+                    op: intersect
+                    value: [CAN_QUERY, CAN_RECEIVE]
+    """
+
+    schema = type_schema('members',
+        attrs={"$ref": "#/definitions/filters_common/list_item_attrs"},
+        count={"type": "number"},
+        count_op={"$ref": "#/definitions/filters_common/comparison_operators"})
+
+    annotation_key = 'c7n:Members'
+    permissions = ('cleanrooms:ListMembers',)
+
+    def __init__(self, data, manager=None):
+        super().__init__(data, manager)
+        data['key'] = f'"{self.annotation_key}"'
+
+    def get_item_values(self, collaboration):
+        if self.annotation_key in collaboration:
+            return collaboration[self.annotation_key]
+        client = local_session(self.manager.session_factory).client('cleanrooms')
+        members = self.manager.retry(
+            client.get_paginator('list_members').paginate,
+            collaborationIdentifier=collaboration['id']
+        ).build_full_result().get('memberSummaries', [])
+        collaboration[self.annotation_key] = members
+        return members
+
+
 @resources.register('cleanrooms-membership')
 class CleanRoomsMembership(QueryResourceManager):
     """AWS Clean Rooms Membership"""