Repository: cloud-custodian/cloud-custodian

Issue:
Using `config-rule` mode with `aws.kinesis` resource results in `KeyError: 'StreamName'`
### Describe the bug

When upgrading from [`0.9.12`](https://github.com/cloud-custodian/cloud-custodian/releases/tag/0.9.12.0) to [`0.9.16`](https://github.com/cloud-custodian/cloud-custodian/releases/tag/0.9.16.0), it is suggested to use `mode: config-rule` for resources fully supported by Config. After making this change for a policy enforcing tag compliance on Kinesis resources, the corresponding Lambda function errors with `KeyError: 'StreamName'`.

Based on a number of similar issues reported, it seems the problem is related to the difference in the Lambda invocation event shape that Config produces. References to what I believe to be similar issues:

- https://github.com/cloud-custodian/cloud-custodian/issues/6197
- https://github.com/cloud-custodian/cloud-custodian/issues/6268
- https://github.com/cloud-custodian/cloud-custodian/issues/6556
- https://github.com/cloud-custodian/cloud-custodian/issues/7042
- https://github.com/cloud-custodian/cloud-custodian/issues/6940
- https://github.com/cloud-custodian/cloud-custodian/issues/7169

### What did you expect to happen?

I expected my Kinesis tag compliance policy to yield the same results as it did prior to switching over to `config-rule` mode. 

### Cloud Provider

Amazon Web Services (AWS)

### Cloud Custodian version and dependency information

```shell
Custodian:   0.9.16
Python:      3.9.13
Platform:    Ubuntu
Using venv:  True
Docker: False
Installed: 

argcomplete==2.0.0
attrs==21.4.0
boto3==1.21.42
botocore==1.24.42
docutils==0.17.1
importlib-metadata==4.11.3
importlib-resources==5.7.1
jmespath==1.0.0
jsonschema==4.4.0
pyrsistent==0.18.1
python-dateutil==2.8.2
pyyaml==6.0
s3transfer==0.5.2
six==1.16.0
tabulate==0.8.9
typing-extensions==4.2.0
urllib3==1.26.9
zipp==3.8.0
```


### Policy

```yaml
policies:
- name: kinesis-tag-compliance
  resource: aws.kinesis
  mode:
    role: custodian
    type: config-rule
    runtime: python3.9
    tags:
      Name: kinesis-tag-compliance
  filters:
  - or:
    - tag:Name: absent
    - tag:Source: absent
    - type: value
      key: tag:fake-tag-x
      op: not-in
      value:
      - foo
      - bar
    - type: value
      key: tag:fake-tag-y
      op: not-in
      value:
      - "1.2.3"
      - "4.5.6"
    - type: value
      key: "tag:foo/bar"
      op: not-in
      value:
      - foo-value
      - bar-value
    - tag:fake-tag-z: absent
    - type: value
      key: "tag:xxx"
      op: not-in
      value:
      - "1.0"
```


### Relevant log/traceback output

```shell
[ERROR] KeyError: 'StreamName'
Traceback (most recent call last):
  File "/var/task/custodian_policy.py", line 4, in run
    return handler.dispatch_event(event, context)
  File "/var/task/c7n/handler.py", line 165, in dispatch_event
    p.push(event, context)
  File "/var/task/c7n/policy.py", line 1201, in push
    return mode.run(event, lambda_ctx)
  File "/var/task/c7n/policy.py", line 898, in run
    resources = super(ConfigRuleMode, self).run(event, lambda_context)
  File "/var/task/c7n/policy.py", line 447, in run
    resources = self.policy.resource_manager.filter_resources(
  File "/var/task/c7n/manager.py", line 111, in filter_resources
    resources = f.process(resources, event)
  File "/var/task/c7n/filters/core.py", line 337, in process
    return self.process_set(resources, event)
  File "/var/task/c7n/filters/core.py", line 349, in process_set
    resource_map = {r[rtype_id]: r for r in resources}
  File "/var/task/c7n/filters/core.py", line 349, in <dictcomp>
    resource_map = {r[rtype_id]: r for r in resources}
```


### Extra information or context

It'd be great if the documentation was accurate. The example on https://cloudcustodian.io/docs/aws/topics/config.html#config-poll-rule doesn't pass a validate with v0.9.16:

```sh
$ custodian validate kinesis-one-stream.yml
2022-06-08 11:46:20,026: custodian.commands:ERROR Configuration invalid: kinesis-one-stream.yml
2022-06-08 11:46:20,026: custodian.commands:ERROR Policy: kinesis-one-stream is invalid: resource:aws.kinesis fully supported by config and should use mode: config-rule
```

Here's a sample Config resource history output:

```sh
$ aws configservice get-resource-config-history --resource-type AWS::Kinesis::Stream --resource-id test
{
    "configurationItems": [
        {
            "version": "1.3",
            "accountId": "xxxx",
            "configurationItemCaptureTime": "xxx",
            "configurationItemStatus": "OK",
            "configurationStateId": "xxxx",
            "configurationItemMD5Hash": "",
            "arn": "arn:aws:kinesis:us-west-1:xxxx:stream/test",
            "resourceType": "AWS::Kinesis::Stream",
            "resourceId": "test",
            "resourceName": "test",
            "awsRegion": "us-west-1",
            "availabilityZone": "Regional",
            "tags": {
                "test": "test"
            },
            "relatedEvents": [],
            "relationships": [],
            "configuration": "{\"Arn\":\"arn:aws:kinesis:us-west-1:xxxx:stream/test\",\"Name\":\"test\",\"RetentionPeriodHours\":24,\"ShardCount\":1,\"Tags\":[{\"Key\":\"test\",\"Value\":\"test\"}]}",
            "supplementaryConfiguration": {}
        },
        {
            "version": "1.3",
            "accountId": "xxxx",
            "configurationItemCaptureTime": "2022-06-08T08:19:31.170000-05:00",
            "configurationItemStatus": "OK",
            "configurationStateId": "xxxx",
            "configurationItemMD5Hash": "",
            "arn": "arn:aws:kinesis:us-west-1:xxxx:stream/test",
            "resourceType": "AWS::Kinesis::Stream",
            "resourceId": "test",
            "resourceName": "test",
            "awsRegion": "us-west-1",
            "availabilityZone": "Regional",
            "tags": {},
            "relatedEvents": [],
            "relationships": [],
            "configuration": "{\"Arn\":\"arn:aws:kinesis:us-west-1:xxxxx:stream/test\",\"Name\":\"test\",\"RetentionPeriodHours\":24,\"ShardCount\":1,\"Tags\":[]}",
            "supplementaryConfiguration": {}
        }
    ]
}
```


Discussion:

Patch:
diff --git a/c7n/ctx.py b/c7n/ctx.py
index c3fbd1631..94c8e6f47 100644
--- a/c7n/ctx.py
+++ b/c7n/ctx.py
@@ -11,7 +11,8 @@ from c7n.output import (
     log_outputs,
     metrics_outputs,
     sys_stats_outputs,
-    tracer_outputs)
+    tracer_outputs,
+)
 
 from c7n.utils import reset_session_cache, dumps, local_session
 from c7n.version import version
@@ -91,8 +92,7 @@ class ExecutionContext:
     def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):
         if exc_type is not None and self.metrics:
             self.metrics.put_metric('PolicyException', 1, "Count")
-        self.policy._write_file(
-            'metadata.json', dumps(self.get_metadata(), indent=2))
+        self.output.write_file('metadata.json', dumps(self.get_metadata(), indent=2))
         self.api_stats.__exit__(exc_type, exc_value, exc_traceback)
 
         with self.tracer.subsegment('output'):
@@ -121,8 +121,9 @@ class ExecutionContext:
                 'id': self.execution_id,
                 'start': self.start_time,
                 'end_time': t,
-                'duration': t - self.start_time},
-            'config': dict(self.options)
+                'duration': t - self.start_time,
+            },
+            'config': dict(self.options),
         }
 
         if 'sys-stats' in include and self.sys_stats:
diff --git a/c7n/output.py b/c7n/output.py
index b301af3ce..95cef6f77 100644
--- a/c7n/output.py
+++ b/c7n/output.py
@@ -17,6 +17,7 @@ import tempfile
 import time
 import uuid
 
+from abc import ABC, abstractmethod
 
 from c7n.exceptions import InvalidOutputConfig
 from c7n.registry import PluginRegistry
@@ -24,6 +25,7 @@ from c7n.utils import parse_url_config
 
 try:
     import psutil
+
     HAVE_PSUTIL = True
 except ImportError:
     HAVE_PSUTIL = False
@@ -157,7 +159,7 @@ class NullStats:
         """Take a snapshot of the system stats and append to the stack."""
 
     def pop_snapshot(self):
-        """Remove a snapshot from the snack and return a delta of the current stats to it.
+        """Remove a snapshot from the stack and return a delta of the current stats to it.
         """
         return {}
 
@@ -382,8 +384,33 @@ class NullLog(LogOutput):
         return None
 
 
+class OutputFileHandler(ABC):
+    """Base class for types registered with the blob_outputs registry.
+
+    Provides explicit interface definition for the types.
+
+    The file handlers are treated as context managers.
+    """
+
+    type: str  # Injected by the register method, matches the string type passed.
+    root_dir: str  # The base directory that will hold the output files.
+
+    @abstractmethod
+    def __enter__(self):
+        raise NotImplementedError()
+
+    @abstractmethod
+    def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):
+        raise NotImplementedError()
+
+    @abstractmethod
+    def write_file(self, rel_path, value):
+        "Write a file at the relative path specified with the value as the content."
+        raise NotImplementedError()
+
+
 @blob_outputs.register('null')
-class NullBlobOutput:
+class NullBlobOutput(OutputFileHandler):
     # default - for unit tests
 
     def __init__(self, ctx, config):
@@ -395,15 +422,18 @@ class NullBlobOutput:
         return "<null blob output>"
 
     def __enter__(self):
-        return
+        "A no-op for the null handler."
 
     def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):
-        return
+        "A no-op for the null handler."
+
+    def write_file(self, rel_path, value):
+        "A no-op for the null handler."
 
 
 @blob_outputs.register('file')
 @blob_outputs.register('default')
-class DirectoryOutput:
+class DirectoryOutput(OutputFileHandler):
 
     permissions = ()
 
@@ -428,6 +458,10 @@ class DirectoryOutput:
     def __repr__(self):
         return "<%s to dir:%s>" % (self.__class__.__name__, self.root_dir)
 
+    def write_file(self, rel_path, value):
+        with open(os.path.join(self.root_dir, rel_path), 'w') as fh:
+            fh.write(value)
+
     def compress(self):
         # Compress files individually so thats easy to walk them, without
         # downloading tar and extracting.
@@ -450,7 +484,8 @@ class DirectoryOutput:
             'region': self.ctx.options.region,
             'policy_name': self.ctx.policy.name,
             'now': datetime.datetime.utcnow(),
-            'uuid': str(uuid.uuid4())}
+            'uuid': str(uuid.uuid4()),
+        }
         return data
 
 
@@ -471,13 +506,15 @@ class BlobOutput(DirectoryOutput):
         return "<output:%s to bucket:%s prefix:%s>" % (
             self.type,
             self.bucket,
-            self.key_prefix)
+            self.key_prefix,
+        )
 
     def get_output_path(self, output_url):
         if '{' not in output_url:
             date_path = datetime.datetime.utcnow().strftime('%Y/%m/%d/%H')
-            return "/".join([s.strip('/') for s in [
-                output_url, self.ctx.policy.name, date_path]])
+            return "/".join(
+                [s.strip('/') for s in [output_url, self.ctx.policy.name, date_path]]
+            )
         return output_url.format(**self.get_output_vars()).rstrip('/')
 
     def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):
@@ -489,8 +526,10 @@ class BlobOutput(DirectoryOutput):
 
     def upload(self):
         for root, dirs, files in os.walk(self.root_dir):
+            len_root_dir = len(self.root_dir)
             for f in files:
-                key = "/".join(filter(None, [self.key_prefix, root[len(self.root_dir):], f]))
+                rel_path = root[len_root_dir:]
+                key = "/".join(filter(None, [self.key_prefix, rel_path, f]))
                 self.upload_file(os.path.join(root, f), key)
 
     def upload_file(self, path, key):
diff --git a/c7n/policy.py b/c7n/policy.py
index 6d94b092b..3336acbe2 100644
--- a/c7n/policy.py
+++ b/c7n/policy.py
@@ -16,7 +16,7 @@ from c7n.ctx import ExecutionContext
 from c7n.exceptions import PolicyValidationError, ClientError, ResourceLimitExceeded
 from c7n.filters import FilterRegistry, And, Or, Not
 from c7n.manager import iter_filters
-from c7n.output import DEFAULT_NAMESPACE, NullBlobOutput
+from c7n.output import DEFAULT_NAMESPACE
 from c7n.resources import load_resources
 from c7n.registry import PluginRegistry
 from c7n.provider import clouds, get_resource_class
@@ -279,35 +279,39 @@ class PullMode(PolicyExecutionMode):
         if not self.policy.is_runnable():
             return []
 
-        with self.policy.ctx:
+        with self.policy.ctx as ctx:
             self.policy.log.debug(
                 "Running policy:%s resource:%s region:%s c7n:%s",
-                self.policy.name, self.policy.resource_type,
+                self.policy.name,
+                self.policy.resource_type,
                 self.policy.options.region or 'default',
-                version)
+                version,
+            )
 
             s = time.time()
             try:
                 resources = self.policy.resource_manager.resources()
             except ResourceLimitExceeded as e:
                 self.policy.log.error(str(e))
-                self.policy.ctx.metrics.put_metric(
-                    'ResourceLimitExceeded', e.selection_count, "Count")
+                ctx.metrics.put_metric(
+                    'ResourceLimitExceeded', e.selection_count, "Count"
+                )
                 raise
 
             rt = time.time() - s
             self.policy.log.info(
-                "policy:%s resource:%s region:%s count:%d time:%0.2f" % (
-                    self.policy.name,
-                    self.policy.resource_type,
-                    self.policy.options.region,
-                    len(resources), rt))
-            self.policy.ctx.metrics.put_metric(
-                "ResourceCount", len(resources), "Count", Scope="Policy")
-            self.policy.ctx.metrics.put_metric(
-                "ResourceTime", rt, "Seconds", Scope="Policy")
-            self.policy._write_file(
-                'resources.json', utils.dumps(resources, indent=2))
+                "policy:%s resource:%s region:%s count:%d time:%0.2f",
+                self.policy.name,
+                self.policy.resource_type,
+                self.policy.options.region,
+                len(resources),
+                rt,
+            )
+            ctx.metrics.put_metric(
+                "ResourceCount", len(resources), "Count", Scope="Policy"
+            )
+            ctx.metrics.put_metric("ResourceTime", rt, "Seconds", Scope="Policy")
+            ctx.output.write_file('resources.json', utils.dumps(resources, indent=2))
 
             if not resources:
                 return []
@@ -319,19 +323,19 @@ class PullMode(PolicyExecutionMode):
             at = time.time()
             for a in self.policy.resource_manager.actions:
                 s = time.time()
-                with self.policy.ctx.tracer.subsegment('action:%s' % a.type):
+                with ctx.tracer.subsegment('action:%s' % a.type):
                     results = a.process(resources)
                 self.policy.log.info(
                     "policy:%s action:%s"
                     " resources:%d"
-                    " execution_time:%0.2f" % (
-                        self.policy.name, a.name,
-                        len(resources), time.time() - s))
+                    " execution_time:%0.2f"
+                    % (self.policy.name, a.name, len(resources), time.time() - s)
+                )
                 if results:
-                    self.policy._write_file(
-                        "action-%s" % a.name, utils.dumps(results))
-            self.policy.ctx.metrics.put_metric(
-                "ActionTime", time.time() - at, "Seconds", Scope="Policy")
+                    ctx.output.write_file("action-%s" % a.name, utils.dumps(results))
+            ctx.metrics.put_metric(
+                "ActionTime", time.time() - at, "Seconds", Scope="Policy"
+            )
             return resources
 
 
@@ -351,6 +355,7 @@ class LambdaMode(ServerlessExecutionMode):
             # Lambda passthrough config
             'layers': {'type': 'array', 'items': {'type': 'string'}},
             'concurrency': {'type': 'integer'},
+            # Do we really still support 2.7 and 3.6?
             'runtime': {'enum': ['python2.7', 'python3.6',
                                  'python3.7', 'python3.8', 'python3.9']},
             'role': {'type': 'string'},
@@ -468,28 +473,31 @@ class LambdaMode(ServerlessExecutionMode):
 
     def run_resource_set(self, event, resources):
         from c7n.actions import EventAction
-        with self.policy.ctx:
-            self.policy.ctx.metrics.put_metric(
-                'ResourceCount', len(resources), 'Count', Scope="Policy",
-                buffer=False)
+
+        with self.policy.ctx as ctx:
+            ctx.metrics.put_metric(
+                'ResourceCount', len(resources), 'Count', Scope="Policy", buffer=False
+            )
 
             if 'debug' in event:
                 self.policy.log.info(
-                    "Invoking actions %s", self.policy.resource_manager.actions)
+                    "Invoking actions %s", self.policy.resource_manager.actions
+                )
 
-            self.policy._write_file(
-                'resources.json', utils.dumps(resources, indent=2))
+            ctx.output.write_file('resources.json', utils.dumps(resources, indent=2))
 
             for action in self.policy.resource_manager.actions:
                 self.policy.log.info(
                     "policy:%s invoking action:%s resources:%d",
-                    self.policy.name, action.name, len(resources))
+                    self.policy.name,
+                    action.name,
+                    len(resources),
+                )
                 if isinstance(action, EventAction):
                     results = action.process(resources, event)
                 else:
                     results = action.process(resources)
-                self.policy._write_file(
-                    "action-%s" % action.name, utils.dumps(results))
+                ctx.output.write_file("action-%s" % action.name, utils.dumps(results))
         return resources
 
     @property
@@ -576,6 +584,7 @@ class PHDMode(LambdaMode):
         entities = []
         paginator = client.get_paginator('describe_affected_entities')
         for event_set in utils.chunks(event_arns, 10):
+            # Note: we aren't using event_set here, just event_arns.
             entities.extend(list(itertools.chain(
                             *[p['entities'] for p in paginator.paginate(
                                 filter={'eventArns': event_arns})])))
@@ -1282,10 +1291,11 @@ class Policy:
     run = __call__
 
     def _write_file(self, rel_path, value):
-        if isinstance(self.ctx.output, NullBlobOutput):
-            return
-        with open(os.path.join(self.ctx.log_dir, rel_path), 'w') as fh:
-            fh.write(value)
+        """This method is no longer called within c7n, and despite being a private
+        method, caution is taken here to not break any external callers.
+        """
+        log.warning("policy _write_file is deprecated, use ctx.output.write_file")
+        self.ctx.output.write_file(rel_path, value)
 
     def load_resource_manager(self):
         factory = get_resource_class(self.data.get('resource'))
diff --git a/c7n/resources/cloudfront.py b/c7n/resources/cloudfront.py
index 7834f0119..46e94e8c0 100644
--- a/c7n/resources/cloudfront.py
+++ b/c7n/resources/cloudfront.py
@@ -371,7 +371,7 @@ class DistributionPostFinding(PostFinding):
 
         payload.update(self.filter_empty({
             'DomainName': r['DomainName'],
-            'WebACLId': r.get('WebACLId'),
+            "WebAclId": r.get('WebACLId'),
             'LastModifiedTime': r['LastModifiedTime'].isoformat(),
             'Status': r['Status'],
             'Logging': self.filter_empty(r.get('Logging', {})),
diff --git a/c7n/resources/kinesis.py b/c7n/resources/kinesis.py
index c68a6ba65..f0a8ff36d 100644
--- a/c7n/resources/kinesis.py
+++ b/c7n/resources/kinesis.py
@@ -17,6 +17,21 @@ class DescribeStream(DescribeSource):
         return universal_augment(self.manager, super().augment(resources))
 
 
+class ConfigStream(ConfigSource):
+
+    def load_resource(self, item):
+        resource = super().load_resource(item)
+        for ck, dk in {
+                'Arn': 'StreamARN',
+                'Name': 'StreamName'}.items():
+            resource[dk] = resource.pop(ck, None)
+        if 'StreamEncryption' in resource:
+            encrypt = resource.pop('StreamEncryption')
+            resource['EncryptionType'] = encrypt['EncryptionType']
+            resource['KeyId'] = encrypt['KeyId']
+        return resource
+
+
 @resources.register('kinesis')
 class KinesisStream(QueryResourceManager):
     retry = staticmethod(
@@ -36,7 +51,7 @@ class KinesisStream(QueryResourceManager):
 
     source_mapping = {
         'describe': DescribeStream,
-        'config': ConfigSource
+        'config': ConfigStream
     }
 
 
diff --git a/tools/c7n_azure/c7n_azure/policy.py b/tools/c7n_azure/c7n_azure/policy.py
index c4f5bc981..88d0b9603 100644
--- a/tools/c7n_azure/c7n_azure/policy.py
+++ b/tools/c7n_azure/c7n_azure/policy.py
@@ -323,15 +323,15 @@ class AzureModeCommon:
             resources = policy.resource_manager.filter_resources(
                 resources, event)
 
-        with policy.ctx:
+        with policy.ctx as ctx:
             rt = time.time() - s
 
-            policy.ctx.metrics.put_metric(
+            ctx.metrics.put_metric(
                 'ResourceCount', len(resources), 'Count', Scope="Policy",
                 buffer=False)
-            policy.ctx.metrics.put_metric(
+            ctx.metrics.put_metric(
                 "ResourceTime", rt, "Seconds", Scope="Policy")
-            policy._write_file(
+            ctx.output.write_file(
                 'resources.json', utils.dumps(resources, indent=2))
 
             if not resources:
@@ -345,13 +345,13 @@ class AzureModeCommon:
                 policy.log.info(
                     "policy: %s invoking action: %s resources: %d",
                     policy.name, action.name, len(resources))
-                with policy.ctx.tracer.subsegment('action:%s' % action.type):
+                with ctx.tracer.subsegment('action:%s' % action.type):
                     if isinstance(action, EventAction):
                         results = action.process(resources, event)
                     else:
                         results = action.process(resources)
                 try:
-                    policy._write_file(
+                    ctx.output.write_file(
                         "action-%s" % action.name, utils.dumps(results))
                 except (TypeError, OverflowError):
                     pass


Test patch:
diff --git a/tests/data/config/kinesis-stream.json b/tests/data/config/kinesis-stream.json
new file mode 100644
index 000000000..13ece0f60
--- /dev/null
+++ b/tests/data/config/kinesis-stream.json
@@ -0,0 +1,21 @@
+{
+    "version": "1.3",
+    "accountId": "644160558196",
+    "configurationItemCaptureTime": "2022-07-21T08:54:37.664000-04:00",
+    "configurationItemStatus": "ResourceDiscovered",
+    "configurationStateId": "1658408077664",
+    "configurationItemMD5Hash": "",
+    "arn": "arn:aws:kinesis:us-east-1:644160558196:stream/stream-encrypted",
+    "resourceType": "AWS::Kinesis::Stream",
+    "resourceId": "stream-encrypted",
+    "resourceName": "stream-encrypted",
+    "awsRegion": "us-east-1",
+    "availabilityZone": "Regional",
+    "tags": {
+        "App": "Magic"
+    },
+    "relatedEvents": [],
+    "relationships": [],
+    "configuration": "{\"Arn\":\"arn:aws:kinesis:us-east-1:644160558196:stream/stream-encrypted\",\"Name\":\"stream-encrypted\",\"RetentionPeriodHours\":24,\"ShardCount\":4,\"Tags\":[{\"Key\":\"App\",\"Value\":\"Magic\"}],\"StreamEncryption\":{\"EncryptionType\":\"KMS\",\"KeyId\":\"alias/aws/kinesis\"}}",
+    "supplementaryConfiguration": {}
+}
diff --git a/tests/data/placebo/test_distribution_post_finding/cloudfront.GetDistributionConfig_1.json b/tests/data/placebo/test_distribution_post_finding/cloudfront.GetDistributionConfig_1.json
index 3b31425b8..61a468656 100644
--- a/tests/data/placebo/test_distribution_post_finding/cloudfront.GetDistributionConfig_1.json
+++ b/tests/data/placebo/test_distribution_post_finding/cloudfront.GetDistributionConfig_1.json
@@ -233,9 +233,9 @@
                     ]
                 }
             },
-            "WebACLId": "",
+            "WebACLId": "1ebe0b46-0fd2-4e07-a74c-27bf25adc0bf",
             "HttpVersion": "http2",
             "IsIPV6Enabled": false
         }
     }
-}
\ No newline at end of file
+}
diff --git a/tests/data/placebo/test_distribution_waf/cloudfront.ListDistributions_2.json b/tests/data/placebo/test_distribution_waf/cloudfront.ListDistributions_2.json
index 06f48eebc..b73a915a3 100644
--- a/tests/data/placebo/test_distribution_waf/cloudfront.ListDistributions_2.json
+++ b/tests/data/placebo/test_distribution_waf/cloudfront.ListDistributions_2.json
@@ -278,4 +278,4 @@
             }
         }
     }
-}
\ No newline at end of file
+}
diff --git a/tests/test_kinesis.py b/tests/test_kinesis.py
index e9235b9e6..f8f8603ae 100644
--- a/tests/test_kinesis.py
+++ b/tests/test_kinesis.py
@@ -1,6 +1,18 @@
 # Copyright The Cloud Custodian Authors.
 # SPDX-License-Identifier: Apache-2.0
-from .common import BaseTest
+from .common import BaseTest, event_data
+
+
+def test_stream_config_source(test):
+    p = test.load_policy({
+        'name': 'stream-config',
+        'resource': 'aws.kinesis',
+        'mode': {'type': 'config-rule'}})
+    item = event_data('kinesis-stream.json', 'config')
+    source = p.resource_manager.get_source('config')
+    resource = source.load_resource(item)
+    assert resource['StreamName'] == 'stream-encrypted'
+    assert resource['KeyId'] == 'alias/aws/kinesis'
 
 
 class Kinesis(BaseTest):