c7n_kube - mode/k8s-admission - add admission controller mode (#7697)

c7n/filters/core.py

@@ -346,11 +346,20 @@ def __call__(self, r):
 
     def process_set(self, resources, event):
         rtype_id = self.get_resource_type_id()
-        resource_map = {r[rtype_id]: r for r in resources}
+        compiled = None
+        if '.' in rtype_id:
+            compiled = jmespath.compile(rtype_id)
+            resource_map = {compiled.search(r): r for r in resources}
+        else:
+            resource_map = {r[rtype_id]: r for r in resources}
         results = set()
         for f in self.filters:
-            results = results.union([
-                r[rtype_id] for r in f.process(resources, event)])
+            if compiled:
+                results = results.union([
+                    compiled.search(r) for r in f.process(resources, event)])
+            else:
+                results = results.union([
+                    r[rtype_id] for r in f.process(resources, event)])
         return [resource_map[r_id] for r_id in results]
 
 
@@ -390,7 +399,12 @@ def __call__(self, r):
 
     def process_set(self, resources, event):
         rtype_id = self.get_resource_type_id()
-        resource_map = {r[rtype_id]: r for r in resources}
+        compiled = None
+        if '.' in rtype_id:
+            compiled = jmespath.compile(rtype_id)
+            resource_map = {compiled.search(r): r for r in resources}
+        else:
+            resource_map = {r[rtype_id]: r for r in resources}
         sweeper = AnnotationSweeper(rtype_id, resources)
 
         for f in self.filters:
@@ -399,7 +413,10 @@ def process_set(self, resources, event):
                 break
 
         before = set(resource_map.keys())
-        after = {r[rtype_id] for r in resources}
+        if compiled:
+            after = {compiled.search(r) for r in resources}
+        else:
+            after = {r[rtype_id] for r in resources}
         results = before - after
         sweeper.sweep([])
 
@@ -415,16 +432,28 @@ def __init__(self, id_key, resources):
         self.id_key = id_key
         ra_map = {}
         resource_map = {}
+        compiled = None
+        if '.' in id_key:
+            compiled = jmespath.compile(self.id_key)
         for r in resources:
-            ra_map[r[id_key]] = {k: v for k, v in r.items() if k.startswith('c7n')}
-            resource_map[r[id_key]] = r
+            if compiled:
+                id_ = compiled.search(r)
+            else:
+                id_ = r[self.id_key]
+            ra_map[id_] = {k: v for k, v in r.items() if k.startswith('c7n')}
+            resource_map[id_] = r
         # We keep a full copy of the annotation keys to allow restore.
         self.ra_map = copy.deepcopy(ra_map)
         self.resource_map = resource_map
 
     def sweep(self, resources):
-        for rid in set(self.ra_map).difference([
-                r[self.id_key] for r in resources]):
+        compiled = None
+        if '.' in self.id_key:
+            compiled = jmespath.compile(self.id_key)
+            diff = set(self.ra_map).difference([compiled.search(r) for r in resources])
+        else:
+            diff = set(self.ra_map).difference([r[self.id_key] for r in resources])
+        for rid in diff:
             # Clear annotations if the block filter didn't match
             akeys = [k for k in self.resource_map[rid] if k.startswith('c7n')]
             for k in akeys:

c7n/policy.py

@@ -92,10 +92,11 @@ def from_data(cls, data: dict, options, session_factory=None):
     def __add__(self, other):
         return self.__class__(self.policies + other.policies, self.options)
 
-    def filter(self, policy_patterns=[], resource_types=[]):
+    def filter(self, policy_patterns=(), resource_types=(), modes=()):
         results = self.policies
         results = self._filter_by_patterns(results, policy_patterns)
         results = self._filter_by_resource_types(results, resource_types)
+        results = self._filter_by_modes(results, modes)
         # next line brings the result set in the original order of self.policies
         results = [x for x in self.policies if x in results]
         return PolicyCollection(results, self.options)
@@ -162,6 +163,32 @@ def _filter_by_resource_type(self, policies, resource_type):
 
         return results
 
+    def _filter_by_modes(self, policies, modes):
+        """
+        Takes a list of policies and returns only those matching a given mode
+        """
+        if not modes:
+            return policies
+        results = []
+        for mode in modes:
+            result = self._filter_by_mode(policies, mode)
+            results.extend(x for x in result if x not in results)
+        return results
+
+    def _filter_by_mode(self, policies, mode):
+        """
+        Takes a list of policies and returns only those matching a given mode
+        """
+        results = []
+        for policy in policies:
+            if policy.get_execution_mode().type == mode:
+                results.append(policy)
+        if not results:
+            self.log.warning((
+                'Filter by modes type "{}" '
+                'did not match any policies.').format(mode))
+        return results
+
     def __iter__(self):
         return iter(self.policies)
 

c7n/resources/aws.py

@@ -743,7 +743,6 @@ def get_service_region_map(regions, resource_types, provider='aws'):
             normalized_types.append(r[len(provider) + 1:])
         else:
             normalized_types.append(r)
-
     resource_service_map = {
         r: clouds[provider].resources.get(r).resource_type.service
         for r in normalized_types if r != 'account'}

tests/test_filters.py

@@ -1,5 +1,6 @@
 # Copyright The Cloud Custodian Authors.
 # SPDX-License-Identifier: Apache-2.0
+import copy
 import calendar
 from collections import namedtuple
 from datetime import datetime, timedelta
@@ -17,7 +18,7 @@
 from c7n.testing import mock_datetime_now
 from c7n.utils import annotation
 from .common import instance, event_data, Bag, BaseTest
-from c7n.filters.core import ValueRegex, parse_date as core_parse_date
+from c7n.filters.core import AnnotationSweeper, ValueRegex, parse_date as core_parse_date
 
 
 class BaseFilterTest(unittest.TestCase):
@@ -1503,5 +1504,42 @@ def test_group_regex_date_desc_null_first(self):
         )
 
 
+class AnnotationSweeperTest(unittest.TestCase):
+    def test_annotation_sweep_jmespath(self):
+        resources = [
+            {
+                "metadata": {"uid": "foo"}, "c7n:annotation": "bar"
+            },
+            {
+                "metadata": {"uid": "bar"}, "c7n:annotation": "bar"
+            },
+            {
+                "metadata": {"uid": "baz"},
+            }
+        ]
+        sweeper = AnnotationSweeper(
+            id_key="metadata.uid", resources=resources)
+        sweeper.sweep(resources=resources)
+        self.assertEqual(len(resources), 3)
+        for r in resources:
+            self.assertTrue("c7n:annotation" not in resources)
+
+    def test_annotation_sweep_jmespath_no_annotations(self):
+        resources = [
+            {
+                "metadata": {"uid": "foo"},
+            },
+            {
+                "metadata": {"uid": "bar"},
+            }
+        ]
+        swept = copy.deepcopy(resources)
+        sweeper = AnnotationSweeper(
+            id_key="metadata.uid", resources=swept)
+        sweeper.sweep(resources=swept)
+        self.assertEqual(len(resources), 2)
+        self.assertEqual(resources, swept)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_policy.py

@@ -776,6 +776,34 @@ def test_policy_region_expand_global(self):
         self.assertEqual(iam[0].options.output_dir, "/test/output/eu-west-1")
         self.assertEqual(len(collection), 3)
 
+    def test_policy_filter_mode(self):
+        cfg = Config.empty(regions=['us-east-1'])
+        original = policy.PolicyCollection.from_data(
+            {"policies": [
+                {
+                    "name": "bar",
+                    "resource": "lambda",
+                    "mode": {
+                        "type": "cloudtrail",
+                        "events": ["CreateFunction"],
+                        "role": "custodian"
+                    }
+                },
+                {
+                    "name": "two",
+                    "resource": "ec2",
+                    "mode": {
+                        "type": "periodic",
+                        "role": "cutodian",
+                        "schedule": "rate(1 day)"
+                    }
+                }
+            ]}, cfg)
+        collection = AWS().initialize_policies(original, cfg)
+        result = collection.filter(modes=['cloudtrail'])
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result.policies[0].name, 'bar')
+
 
 class TestPolicy(BaseTest):
 

tools/c7n_kube/c7n_kube/cli.py

@@ -0,0 +1,129 @@
+# Copyright The Cloud Custodian Authors.
+# SPDX-License-Identifier: Apache-2.0
+import argparse
+import logging
+import os
+import pkg_resources
+
+import yaml
+
+from c7n_kube.server import init
+
+from c7n.config import Config
+from c7n.loader import DirectoryLoader
+
+log = logging.getLogger('custodian.k8s.cli')
+logging.basicConfig(
+    # TODO: make this configurable
+    level=logging.INFO,
+    format="%(asctime)s: %(name)s:%(levelname)s %(message)s")
+
+
+PORT = '8800'
+HOST = '0.0.0.0'
+
+TEMPLATE = {
+    "apiVersion": "admissionregistration.k8s.io/v1",
+    "kind": "MutatingWebhookConfiguration",
+    "metadata": {
+        "name": "c7n-admission",
+        "labels": {
+            "app.kubernetes.io/name": "c7n-kates",
+            "app.kubernetes.io/instance": "c7n-kates",
+            "app.kubernetes.io/version": pkg_resources.get_distribution("c7n_kube").version,
+            "app.kubernetes.io/component": "AdmissionController",
+            "app.kubernetes.io/part-of": "c7n_kube",
+            "app.kubernetes.io/managed-by": "c7n"
+        }
+    },
+    "webhooks": [
+        {
+            "name": "admission.cloudcustodian.io",
+            "rules": [
+                {
+                    "operations": [],
+                    "scope": "*",
+                    "apiGroups": [],
+                    "apiVersions": [],
+                    "resources": [],
+                }
+            ],
+            "admissionReviewVersions": [
+                "v1",
+                "v1beta1"
+            ],
+            "clientConfig": {
+                "url": "${ENDPOINT}"
+            },
+            "sideEffects": "None",
+            "failurePolicy": "Fail"
+        }
+    ]
+}
+
+
+def _parser():
+    parser = argparse.ArgumentParser(description='Cloud Custodian Admission Controller')
+    parser.add_argument('--port', type=int, help='Server port', nargs='?', default=PORT)
+    parser.add_argument('--policy-dir', type=str, required=True, help='policy directory')
+    parser.add_argument(
+        '--on-exception', type=str.lower, required=False, default='warn',
+        choices=['warn', 'deny'],
+        help='warn or deny on policy exceptions')
+    parser.add_argument(
+        '--endpoint',
+        help='Endpoint for webhook, used for generating manfiest',
+        required=True,
+    )
+    parser.add_argument(
+        '--generate', default=False, action="store_true",
+        help='Generate a k8s manifest for ValidatingWebhookConfiguration')
+    return parser
+
+
+def cli():
+    """
+    Cloud Custodian Admission Controller
+    """
+    parser = _parser()
+    args = parser.parse_args()
+    if args.generate:
+        directory_loader = DirectoryLoader(Config.empty())
+        policy_collection = directory_loader.load_directory(
+            os.path.abspath(args.policy_dir))
+        operations = []
+        groups = []
+        api_versions = []
+        resources = []
+        for p in policy_collection:
+            execution_mode = p.get_execution_mode()
+            # We only support `k8s-admission` policies for the admission
+            # controller.
+            if execution_mode.type != 'k8s-admission':
+                policy = execution_mode.policy
+                type_ = execution_mode.type
+                log.warning(
+                    f"skipping policy {policy.name} with type {type_}, should be k8s-admission"
+                )
+                continue
+            mvals = p.get_execution_mode().get_match_values()
+            operations.extend(mvals['operations'])
+            groups.append(mvals['group'])
+            api_versions.append(mvals['apiVersions'])
+            resources.extend(mvals['resources'])
+
+        TEMPLATE['webhooks'][0]['rules'][0]['operations'] = sorted(list(set(operations)))
+        TEMPLATE['webhooks'][0]['rules'][0]['apiGroups'] = sorted(list(set(groups)))
+        TEMPLATE['webhooks'][0]['rules'][0]['apiVersions'] = sorted(list(set(api_versions)))
+        TEMPLATE['webhooks'][0]['rules'][0]['resources'] = sorted(list(set(resources)))
+
+        if args.endpoint:
+            TEMPLATE['webhooks'][0]['clientConfig']['url'] = args.endpoint
+
+        print(yaml.dump(TEMPLATE))
+    else:
+        init(args.port, args.policy_dir, args.on_exception)
+
+
+if __name__ == '__main__':
+    cli()

tools/c7n_kube/c7n_kube/entry.py

@@ -24,6 +24,8 @@
 
 from c7n_kube.resources import crd
 
+import c7n_kube.policy  # noqa
+
 log = logging.getLogger('custodian.k8s')
 
 ALL = [

tools/c7n_kube/c7n_kube/exceptions.py

@@ -0,0 +1,13 @@
+from c7n.exceptions import CustodianError
+
+
+class EventNotMatchedException(CustodianError):
+    """
+    Event not matched
+    """
+
+
+class PolicyNotRunnableException(CustodianError):
+    """
+    Policy is not runnable
+    """