Merge branch 'main' into main

c7n/commands.py

@@ -290,8 +290,11 @@ def run(options, policies: List[Policy]) -> None:
     # AWS - Sanity check that we have an assumable role before executing policies
     # Todo - move this behind provider interface
     if options.assume_role and [p for p in policies if p.provider_name == 'aws']:
+        # the cli options we're being handed haven't been initialized by the
+        # provider, instead use one of the provider's policy options.
+        sample_aws = [p for p in policies if p.provider_name == 'aws'].pop()
         try:
-            local_session(clouds['aws']().get_session_factory(options))
+            local_session(clouds['aws']().get_session_factory(sample_aws.options))
         except ClientError:
             log.exception("Unable to assume role %s", options.assume_role)
             sys.exit(1)

c7n/manager.py

@@ -39,6 +39,7 @@ class ResourceManager:
     executor_factory = ThreadPoolExecutor
     retry = None
     permissions = ()
+    get_client = None
 
     def __init__(self, ctx, data):
         self.ctx = ctx

c7n/policy.py

@@ -1045,8 +1045,6 @@ class PolicyConditions:
 
     def __init__(self, policy, data):
         self.policy = policy
-        self.data = data
-        self.filters = self.data.get('conditions', [])
         # for value_from usage / we use the conditions class
         # to mimic a resource manager interface. we can't use
         # the actual resource manager as we're overriding block
@@ -1058,6 +1056,11 @@ def __init__(self, policy, data):
         self.session_factory = rm.session_factory
         # used by c7n-org to extend evaluation conditions
         self.env_vars = {}
+        self.update(data)
+
+    def update(self, data):
+        self.data = data
+        self.filters = self.data.get('conditions', [])
         self.initialized = False
 
     def validate(self):
@@ -1270,8 +1273,8 @@ def expand_variables(self, variables):
         # Update ourselves in place
         self.data = updated
 
-        # NOTE rebuild the policy conditions base on the new self.data
-        self.conditions = PolicyConditions(self, self.data)
+        # NOTE update the policy conditions base on the new self.data
+        self.conditions.update(self.data)
 
         # Reload filters/actions using updated data, we keep a reference
         # for some compatiblity preservation work.

c7n/query.py

@@ -264,8 +264,13 @@ def augment(self, resources):
             _augment = _batch_augment
         else:
             return resources
+        if self.manager.get_client:
+            client = self.manager.get_client()
+        else:
+            client = local_session(self.manager.session_factory).client(
+                model.service, region_name=self.manager.config.region)
         _augment = functools.partial(
-            _augment, self.manager, model, detail_spec)
+            _augment, self.manager, model, detail_spec, client)
         with self.manager.executor_factory(
                 max_workers=self.manager.max_workers) as w:
             results = list(w.map(
@@ -451,12 +456,8 @@ class QueryResourceManager(ResourceManager, metaclass=QueryMeta):
     max_workers = 3
     chunk_size = 20
 
-    permissions = ()
-
     _generate_arn = None
 
-    get_client = None
-
     retry = staticmethod(
         get_retry((
             'TooManyRequestsException',
@@ -718,10 +719,8 @@ def get_parent_manager(self):
         return self.get_resource_manager(self.resource_type.parent_spec[0])
 
 
-def _batch_augment(manager, model, detail_spec, resource_set):
+def _batch_augment(manager, model, detail_spec, client, resource_set):
     detail_op, param_name, param_key, detail_path, detail_args = detail_spec
-    client = local_session(manager.session_factory).client(
-        model.service, region_name=manager.config.region)
     op = getattr(client, detail_op)
     if manager.retry:
         args = (op,)
@@ -735,10 +734,8 @@ def _batch_augment(manager, model, detail_spec, resource_set):
     return response[detail_path]
 
 
-def _scalar_augment(manager, model, detail_spec, resource_set):
+def _scalar_augment(manager, model, detail_spec, client, resource_set):
     detail_op, param_name, param_key, detail_path = detail_spec
-    client = local_session(manager.session_factory).client(
-        model.service, region_name=manager.config.region)
     op = getattr(client, detail_op)
     if manager.retry:
         args = (op,)

tests/test_policy.py

@@ -1416,6 +1416,30 @@ def test_env_var_extension(self):
         p.conditions.env_vars['account'] = {'name': 'mickey'}
         self.assertFalse(p.is_runnable())
 
+    def test_env_var_extension_with_expand_variables(self):
+        p_json = {
+            'name': 'profx',
+            'resource': 'aws.ec2',
+            'description': 'Test var extension {var1}',
+            'conditions': [{
+                'type': 'value',
+                'key': 'account.name',
+                'value': 'deputy'}]}
+
+        p = self.load_policy(p_json)
+        p.conditions.env_vars['account'] = {'name': 'deputy'}
+        p.expand_variables({"var1":"value1"})
+        p.validate()
+        self.assertEqual("Test var extension value1", p.data["description"])
+        self.assertTrue(p.is_runnable())
+
+        p = self.load_policy(p_json)
+        p.conditions.env_vars['account'] = {'name': 'mickey'}
+        p.expand_variables({"var1":"value2"})
+        p.validate()
+        self.assertEqual("Test var extension value2", p.data["description"])
+        self.assertFalse(p.is_runnable())
+
     def test_event_filter(self):
         p = self.load_policy({
             'name': 'profx',

tools/c7n_left/c7n_left/cli.py

@@ -68,22 +68,25 @@ def run(
 
 @cli.command()
 @click.option("-p", "--policy-dir", type=click.Path(), required=True)
-def test(policy_dir):
+@click.option(
+    "--filters", help="filter policies or resources as k=v pairs with globbing"
+)
+def test(policy_dir, filters):
     """Run policy tests."""
     policy_dir = Path(policy_dir)
     source_dir = policy_dir / "tests"
+
     config = Config.empty(
         source_dir=source_dir,
         policy_dir=policy_dir,
         output_file=sys.stdout,
-        # output=output,
-        # output_file=output_file,
-        # output_query=output_query,
-        # summary=summary,
-        # filters=filters,
+        filters=filters,
     )
+
     reporter = TestReporter(None, config)
-    policies = load_policies(policy_dir, config)
+    exec_filter = ExecutionFilter.parse(config)
+    config["exec_filter"] = exec_filter
+    policies = exec_filter.filter_policies(load_policies(policy_dir, config))
     runner = TestRunner(policies, config, reporter)
     sys.exit(int(runner.run()))
 

tools/c7n_left/c7n_left/test.py

@@ -1,6 +1,7 @@
 # Copyright The Cloud Custodian Authors.
 # SPDX-License-Identifier: Apache-2.0
 #
+import operator
 import time
 
 from c7n.config import Config
@@ -23,7 +24,7 @@ def __init__(self, policies, options, reporter):
     def run(self) -> bool:
         policy_tests = self.get_policy_tests()
         self.reporter.on_tests_discovered(self, policy_tests)
-        for test in policy_tests:
+        for test in sorted(policy_tests, key=operator.attrgetter("name")):
             self.run_test(test)
             self.reporter.on_test_result(test)
         self.reporter.on_tests_complete()
@@ -33,7 +34,9 @@ def run_test(self, test) -> bool:
         checker = TestChecker(test, self.options)
         runner = CollectionRunner(
             [test.policy],
-            self.options.copy(exec_filter=None, source_dir=test.test_dir),
+            self.options.copy(
+                exec_filter=self.options.get("exec_filter"), source_dir=test.test_dir
+            ),
             checker,
         )
         runner.run()
@@ -154,9 +157,14 @@ def on_tests_discovered(self, runner, tests):
         if runner.unmatched_policies:
             header += f" - {len(runner.unmatched_policies)}/{len(runner.policies)}"
             header += " Policies Untested"
-        if runner.unmatched_tests:
-            header += " - [red]{len(runner.unmatched_tests)} Unused Tests"
+        if runner.unmatched_tests and not self.config.get("filters"):
+            header += f" - [red]{len(runner.unmatched_tests)} Unused Tests"
         self.console.print(header)
+        if self.config.get("verbose", True) and not self.config.get("filters"):
+            for p in runner.unmatched_policies:
+                self.console.print(f"no test for {p}")
+            for t in runner.unmatched_tests:
+                self.console.print(f"no policy for {t}")
 
     def on_tests_complete(self):
         status = f"{self.total} "
@@ -169,7 +177,7 @@ def on_tests_complete(self):
         self.console.print(status)
 
     def on_test_load_error(self, test_path, error):
-        self.console.print("[yellow]test load error[yellow] {test_path} - {error}")
+        self.console.print(f"[yellow]test load error[yellow] {test_path} - {error}")
 
     def on_test_result(self, test: Test):
         self.total += 1

tools/c7n_left/tests/test_left_test.py

@@ -26,16 +26,13 @@
 data_dir = Path(os.curdir).absolute() / "data"
 
 
-def test_test_reporter_discovery():
+def test_test_reporter_discovery(capsys):
     reporter = left_test.TestReporter(None, Bag(output_file=sys.stdout))
-
     runner = Bag(unmatched_policies=[1], policies=[2, 3], unmatched_tests=[1])
 
-    console = MagicMock()
-    reporter.console = console
     reporter.on_tests_discovered(runner, [1])
-
-    console.print.assert_called_once()
+    captured = capsys.readouterr()
+    assert "Discovered 1 Tests" in captured.out
 
 
 def test_test_reporter_result():
@@ -114,7 +111,7 @@ def test_cli_test_assertion_not_used(tmp_path):
     (test_case_dir / "left.plan.yaml").write_text(
         """
         - "resource.__tfmeta.path": "google_pubsub_topic.example"
-        - "resource.__tfmeta.path": "google_pubsub_topic.example2"        
+        - "resource.__tfmeta.path": "google_pubsub_topic.example2"
         """
     )
 

tools/c7n_tencentcloud/c7n_tencentcloud/resources/security_group.py

@@ -1,5 +1,6 @@
 # Copyright The Cloud Custodian Authors.
 # SPDX-License-Identifier: Apache-2.0
+from c7n.utils import type_schema, chunks
 from c7n_tencentcloud.provider import resources
 from c7n_tencentcloud.query import ResourceTypeInfo, QueryResourceManager
 from c7n_tencentcloud.utils import PageMethod
@@ -200,3 +201,55 @@ class IPPermissionEgress(SGPermission):
             'CidrV6': {}
         },
         'required': ['type']}
+
+
+@SecurityGroup.filter_registry.register('used')
+class StatisticsFilter(ValueFilter):
+    """statistics
+
+    :example:
+
+    .. code-block:: yaml
+
+        policies:
+        - name: used
+          resource: tencentcloud.security-group
+          description: security group used statistical
+          filters:
+            - type: used
+              key: CVM
+              op: greater-than
+              value: 0
+    """
+
+    schema = type_schema('used', rinherit=ValueFilter.schema)
+    annotation_key = "c7n:usage_stats"
+
+    def match(self, i):
+        return super().match(i[self.annotation_key])
+
+    def process(self, resources, event=None):
+        self.augment([r for r in resources if self.annotation_key not in r])
+        return super().process(resources)
+
+    def augment(self, resources):
+        client = self.manager.get_client()
+
+        # DescribeSecurityGroupAssociationStatistics Maximum support 100
+        for batch in chunks(resources, 50):
+            id_resource_map = {r['SecurityGroupId']: r for r in batch}
+            resp = client.execute_query(
+                "DescribeSecurityGroupAssociationStatistics",
+                {"SecurityGroupIds": list(id_resource_map)}
+            )
+            statistics = resp["Response"]["SecurityGroupAssociationStatisticsSet"]
+            for stat in statistics:
+                group = id_resource_map[stat['SecurityGroupId']]
+                group[self.annotation_key] = {
+                    istat['InstanceType']: istat['InstanceCount'] for istat
+                    in stat['InstanceStatistics']
+                }
+                group[self.annotation_key].update(
+                    {'TotalCount': stat['TotalCount'], 'SG': stat['SG']}
+                )
+        return resources