Merge branch 'cloud-custodian:master' into master

.dockerignore

@@ -32,3 +32,4 @@ pip-selfcheck.json
 *~
 **/tests/
 **/tests_azure/
+tools/c7n_kube/.skaffold

.github/composites/docker-build-push/action.yaml

@@ -14,7 +14,7 @@ inputs:
     default: false
   poetry_version:
     description: "Poetry Version to use"
-    default: "1.2.1"
+    default: "1.2.2"
   trivy_version:
     description: "Trivy version to use"
     default: "0.5.4"

.github/workflows/ci-main.yml

@@ -1,6 +1,6 @@
 name: "CI"
 env:
-  POETRY_VERSION: "1.2.1"
+  POETRY_VERSION: "1.2.2"
 on:
   push:
     branches:
@@ -10,7 +10,9 @@ on:
     branches:
       - master
       - main
-
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
 jobs:
   Lint:
     runs-on: ubuntu-latest
@@ -219,7 +221,7 @@ jobs:
           fi
 
       - name: Upload Code Coverage
-        uses: codecov/codecov-action@v2
+        uses: codecov/codecov-action@v3
         if: contains(matrix.python-version, '3.10') && contains(matrix.os, 'ubuntu')
         with:
           files: ./coverage.xml

ADOPTERS.md

@@ -4,10 +4,16 @@ There are some unreferenceable users that are leveraging Cloud Custodian that ar
 
 There are many additional adopters of Cloud Custodian in the evaluating phase that will be added to this list as they transition to production deployments.
 
+- Avalara
 - Capital One
 - Code 42
+- CyberArk
+- Databricks
 - Grupo
 - HBO Max
+- Intuit
 - JP Morgan Chase & Co
 - Premise Data
+- Sage
+- Siemens
 - Zapier

Makefile

@@ -20,7 +20,7 @@ install:
 
 install-poetry:
 	poetry install
-	for pkg in $(PKG_SET); do echo "Install $$pkg" && cd $$pkg && poetry install && cd ../..; done
+	for pkg in $(PKG_SET); do echo "Install $$pkg" && cd $$pkg && poetry install --all-extras && cd ../..; done
 
 pkg-rebase:
 	rm -f poetry.lock

c7n/actions/autotag.py

@@ -109,6 +109,9 @@ def process(self, resources, event):
             return
 
         user_info = self.get_tag_value(event)
+        if user_info is None:
+            self.log.warning("user info not found in event")
+            return
 
         # will skip writing their UserName tag and not overwrite pre-existing values
         if not self.data.get('update', False):

c7n/actions/invoke.py

@@ -10,6 +10,7 @@
 from c7n import utils
 from c7n.manager import resources
 from c7n.version import version as VERSION
+from c7n.credentials import assumed_session
 
 
 class LambdaInvoke(EventAction):
@@ -32,6 +33,7 @@ class LambdaInvoke(EventAction):
 
      - type: invoke-lambda
        function: my-function
+       assume-role: iam-role-arn
 
     Note if your synchronously invoking the lambda, you may also need
     to configure the timeout, to avoid multiple invokes. The default
@@ -48,6 +50,7 @@ class LambdaInvoke(EventAction):
         'properties': {
             'type': {'enum': ['invoke-lambda']},
             'function': {'type': 'string'},
+            'assume-role': {'type': 'string'},
             'region': {'type': 'string'},
             'async': {'type': 'boolean'},
             'qualifier': {'type': 'string'},
@@ -61,17 +64,27 @@ class LambdaInvoke(EventAction):
                'iam:ListAccountAliases',)
 
     def process(self, resources, event=None):
+
+        config = Config(read_timeout=self.data.get(
+            'timeout', 90), region_name=self.data.get('region', None))
+        session = utils.local_session(self.manager.session_factory)
+        assumed_role = self.data.get('assume-role', '')
+
+        if assumed_role:
+            self.log.debug('Assuming role: {}'.format(assumed_role))
+            target_session = assumed_session(assumed_role, 'LambdaAssumedRoleSession', session)
+            client = target_session.client('lambda', config=config)
+        else:
+            client = utils.local_session(
+                self.manager.session_factory).client('lambda', config=config)
+
         params = dict(FunctionName=self.data['function'])
         if self.data.get('qualifier'):
             params['Qualifier'] = self.data['Qualifier']
 
         if self.data.get('async', True):
             params['InvocationType'] = 'Event'
 
-        config = Config(read_timeout=self.data.get(
-            'timeout', 90), region_name=self.data.get('region', None))
-        client = utils.local_session(
-            self.manager.session_factory).client('lambda', config=config)
         alias = utils.get_account_alias_from_sts(
             utils.local_session(self.manager.session_factory))
 

c7n/actions/notify.py

@@ -220,6 +220,14 @@ def prepare_ec2(self, resources):
                 r.pop('c7n:user-data')
         return resources
 
+    def prepare_iam_saml_provider(self, resources):
+        for r in resources:
+            if 'SAMLMetadataDocument' in r:
+                r.pop('SAMLMetadataDocument')
+            if 'IDPSSODescriptor' in r:
+                r.pop('IDPSSODescriptor')
+        return resources
+
     def send_data_message(self, message):
         if self.data['transport']['type'] == 'sqs':
             return self.send_sqs(message)

c7n/filters/backup.py

@@ -0,0 +1,73 @@
+# Copyright The Cloud Custodian Authors.
+# SPDX-License-Identifier: Apache-2.0
+from .core import Filter
+from datetime import datetime, timedelta
+from c7n.utils import type_schema, local_session, chunks
+from c7n.query import RetryPageIterator
+
+
+class ConsecutiveAwsBackupsFilter(Filter):
+    """Returns resources where number of consective backups (based on the
+    periodicity defined in the filter) is equal to/or greater than n units.
+    This filter supports the resources that use AWS Backup service for backups.
+    :example:
+    .. code-block:: yaml
+            policies:
+              - name: dynamodb-consecutive-aws-backup-count
+                resource: dynamodb-table
+                filters:
+                  - type: consecutive-aws-backups
+                    count: 7
+                    period: days
+                    status: 'COMPLETED'
+    """
+    schema = type_schema('consecutive-aws-backups', count={'type': 'number', 'minimum': 1},
+        period={'enum': ['hours', 'days', 'weeks']},
+        status={'enum': ['COMPLETED', 'PARTIAL', 'DELETING', 'EXPIRED']},
+        required=['count', 'period', 'status'])
+    permissions = ('backup:ListRecoveryPointsByResource', )
+    annotation = 'c7n:AwsBackups'
+
+    def process_resource_set(self, resources):
+        arns = self.manager.get_arns(resources)
+
+        client = local_session(self.manager.session_factory).client('backup')
+        paginator = client.get_paginator('list_recovery_points_by_resource')
+        paginator.PAGE_ITERATOR_CLS = RetryPageIterator
+        for r, arn in zip(resources, arns):
+            r[self.annotation] = paginator.paginate(
+                ResourceArn=arn).build_full_result().get('RecoveryPoints', [])
+
+    def get_date(self, time):
+        period = self.data.get('period')
+        if period == 'weeks':
+            date = (datetime.utcnow() - timedelta(weeks=time)).strftime('%Y-%m-%d')
+        elif period == 'hours':
+            date = (datetime.utcnow() - timedelta(hours=time)).strftime('%Y-%m-%d-%H')
+        else:
+            date = (datetime.utcnow() - timedelta(days=time)).strftime('%Y-%m-%d')
+        return date
+
+    def process(self, resources, event=None):
+        results = []
+        retention = self.data.get('count')
+        expected_dates = set()
+        for time in range(1, retention + 1):
+            expected_dates.add(self.get_date(time))
+
+        for resource_set in chunks(
+                [r for r in resources if self.annotation not in r], 50):
+            self.process_resource_set(resource_set)
+
+        for r in resources:
+            backup_dates = set()
+            for backup in r[self.annotation]:
+                if backup['Status'] == self.data.get('status'):
+                    if self.data.get('period') == 'hours':
+                        backup_dates.add(backup['CreationDate'].strftime('%Y-%m-%d-%H'))
+                    else:
+                        backup_dates.add(backup['CreationDate'].strftime('%Y-%m-%d'))
+
+            if expected_dates.issubset(backup_dates):
+                results.append(r)
+        return results

c7n/filters/iamaccess.py

@@ -72,7 +72,7 @@ def check_actions(self):
 
     @property
     def whitelist_conditions(self):
-        return self.checker_config.get('whitelist_conditions', ())
+        return set(v.lower() for v in self.checker_config.get('whitelist_conditions', ()))
 
     @property
     def allowed_vpce(self):

c7n/loader.py

@@ -174,7 +174,6 @@ def load_file(self):
 
 class DirectoryLoader(PolicyLoader):
     def load_directory(self, directory):
-
         structure = StructureParser()
 
         def _validate(data):
@@ -191,26 +190,28 @@ def _validate(data):
             errors += schema.validate(data, schm)
             return errors
 
-        def _load(directory, raw_policies, errors):
-            raw_policies = []
-            for root, dirs, files in os.walk(directory, topdown=False):
+        def _load(path, raw_policies, errors):
+            for root, dirs, files in os.walk(path):
+                files = [f for f in files if not is_hidden(f)]
+                dirs[:] = [d for d in dirs if not is_hidden(d)]
+
                 for name in files:
                     fmt = name.rsplit('.', 1)[-1]
                     if fmt in ('yaml', 'yml', 'json',):
-                        data = load_file(f"{root}/{name}")
-                        errors.extend(_validate(data))
+                        data = load_file(os.path.join(root, name))
+                        errors += _validate(data)
                         raw_policies.append(data)
                 for name in dirs:
                     _load(os.path.abspath(name), raw_policies, errors)
-            return raw_policies, errors
 
-        loaded, errors = _load(directory, [], [])
+        policy_collections, all_errors = [], []
+        _load(directory, policy_collections, all_errors)
 
-        if errors:
-            raise PolicyValidationError(errors)
+        if all_errors:
+            raise PolicyValidationError(all_errors)
 
         policies = []
-        for p in loaded:
+        for p in policy_collections:
             if not p.get('policies'):
                 continue
             policies.extend(p['policies'])
@@ -224,3 +225,11 @@ def _load(directory, raw_policies, errors):
                 names.append(p['name'])
 
         return self.load_data({'policies': policies}, directory)
+
+
+def is_hidden(path):
+    for part in os.path.split(path):
+        if part != '.' and part.startswith('.'):
+            return True
+
+    return False

c7n/mu.py

@@ -17,6 +17,7 @@
 import time
 import tempfile
 import zipfile
+import platform
 
 
 # We use this for freezing dependencies for serverless environments
@@ -487,6 +488,10 @@ def _create_or_update(self, func, role=None, s3_uri=None, qualifier=None):
             if self._update_tags(existing, new_config.pop('Tags', {})):
                 changed = True
 
+            if self._update_architecture(func, existing,
+                    new_config.pop('Architectures', ["x86_64"]), code_ref):
+                changed = True
+
             config_changed = self.delta_function(old_config, new_config)
             if config_changed:
                 log.debug("Updating function: %s config %s",
@@ -524,6 +529,20 @@ def _update_concurrency(self, existing, func):
             FunctionName=func.name,
             ReservedConcurrentExecutions=func.concurrency)
 
+    def _update_architecture(self, func, existing, new_architecture, code_ref):
+        existing_config = existing.get('Configuration', {})
+        existing_architecture = existing_config.get('Architectures', ["x86_64"])
+        diff = existing_architecture != new_architecture
+        changed = False
+        if diff:
+            log.debug("Updating function architecture: %s" % func.name)
+            params = dict(FunctionName=func.name, Publish=True,
+                          Architectures=new_architecture)
+            params.update(code_ref)
+            self.client.update_function_code(**params)
+            changed = True
+        return changed
+
     def _update_tags(self, existing, new_tags):
         # tag dance
         base_arn = existing['Configuration']['FunctionArn']
@@ -686,6 +705,10 @@ def get_events(self, session_factory):
     def get_archive(self):
         """Return the lambda distribution archive object."""
 
+    @abc.abstractproperty
+    def architectures(self):
+        """ """
+
     def get_config(self):
 
         conf = {
@@ -712,6 +735,8 @@ def get_config(self):
             conf['VpcConfig'] = {
                 'SubnetIds': self.subnets,
                 'SecurityGroupIds': self.security_groups}
+        if self.architectures:
+            conf['Architectures'] = self.architectures
         return conf
 
 
@@ -907,6 +932,16 @@ def layers(self):
     def packages(self):
         return self.policy.data['mode'].get('packages')
 
+    @property
+    def architectures(self):
+        architecture = []
+        arm64_arch = ('aarch64', 'arm64')
+        if platform.machine().lower() in arm64_arch:
+            architecture.append('arm64')
+        else:
+            architecture.append('x86_64')
+        return architecture
+
     def get_events(self, session_factory):
         events = []
         if self.policy.data['mode']['type'] in (

c7n/reports/csvout.py

@@ -193,8 +193,14 @@ def uniq_by_id(self, records):
         """Only the first record for each id"""
         uniq = []
         keys = set()
+        compiled = None
+        if '.' in self._id_field:
+            compiled = jmespath.compile(self._id_field)
         for rec in records:
-            rec_id = rec[self._id_field]
+            if compiled:
+                rec_id = compiled.search(rec)
+            else:
+                rec_id = rec[self._id_field]
             if rec_id not in keys:
                 uniq.append(rec)
                 keys.add(rec_id)