Repository: cloud-custodian/cloud-custodian

Issue:
aws.cloudtrail status IsLogging fails with KeyError: 'c7n:TrailStatus'
**Describe the bug**
cloudtrail status / IsLogging returns key error KeyError: 'c7n:TrailStatus'

https://cloudcustodian.io/docs/aws/resources/cloudtrail.html#status indicates this should work.

**To Reproduce**
Authenticate to an account that has both an Organizational trail and local trail.

Policy file:
```yaml
  - name: awslogs-cloudtrail
    resource: aws.cloudtrail
    filters:
      - type: status
        key: IsLogging
        value: True
```

Output is:
```
2020-05-05 09:03:16,332: custodian.commands:ERROR Error while executing policy awslogs-cloudtrail, continuing
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/c7n/commands.py", line 281, in run
    policy()
  File "/usr/local/lib/python3.7/site-packages/c7n/policy.py", line 1062, in __call__
    resources = mode.run()
  File "/usr/local/lib/python3.7/site-packages/c7n/policy.py", line 291, in run
    resources = self.policy.resource_manager.resources()
  File "/usr/local/lib/python3.7/site-packages/c7n/query.py", line 466, in resources
    resources = self.filter_resources(resources)
  File "/usr/local/lib/python3.7/site-packages/c7n/manager.py", line 109, in filter_resources
    resources = f.process(resources, event)
  File "/usr/local/lib/python3.7/site-packages/c7n/resources/cloudtrail.py", line 122, in process
    return super(Status, self).process(resources)
  File "/usr/local/lib/python3.7/site-packages/c7n/filters/core.py", line 499, in process
    return super(ValueFilter, self).process(resources, event)
  File "/usr/local/lib/python3.7/site-packages/c7n/filters/core.py", line 197, in process
    return list(filter(self, resources))
  File "/usr/local/lib/python3.7/site-packages/c7n/resources/cloudtrail.py", line 125, in __call__
    return self.match(r['c7n:TrailStatus'])
KeyError: 'c7n:TrailStatus'
```

**Expected behavior**
Only CloudTrails that are enabled are returned, no error.


**Background (please complete the following information):**
 - OS: OSX 10.14.6
 - Python Version: Python 3.7.4
 - Custodian Version: 0.9.1
 - Cloud Provider: aws
 - Policy:
```yaml
policies: 
  - name: awslogs-cloudtrail
    resource: cloudtrail
    filters:
      - type: status
        key: IsLogging
        value: True
```
 - Traceback: [if applicable, please exclude sensitive/account information]
 ```
Custodian:   0.9.1
Python:      3.7.4 (default, Sep  7 2019, 18:27:02)
             [Clang 10.0.1 (clang-1001.0.46.4)]
Platform:    posix.uname_result(sysname='Darwin', nodename='xxx', release='18.7.0', version='Darwin Kernel Version 18.7.0: Tue Aug 20 16:57:14 PDT 2019; root:xnu-4903.271.2~2/RELEASE_X86_64', machine='x86_64')
Using venv:  False
Docker: False
Installed:

argcomplete==1.11.1
attrs==19.3.0
boto3==1.12.47
botocore==1.15.47
docutils==0.15.2
importlib-metadata==1.6.0
jmespath==0.9.5
jsonschema==3.2.0
pyrsistent==0.16.0
python-dateutil==2.8.1
pyyaml==5.3.1
s3transfer==0.3.3
setuptools==41.0.1
six==1.14.0
tabulate==0.8.7
urllib3==1.25.9
zipp==3.1.0
```

Under --debug, self has:
```
(Pdb) p dir(self)
['__call__', '__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', '_validate_resource_count', '_validate_value_regex', 'annotate', 'annotation_key', 'data', 'executor_factory', 'expr', 'get_block_operator', 'get_permissions', 'get_resource_value', 'log', 'manager', 'match', 'merge_annotation', 'metrics', 'op', 'permissions', 'process', 'process_value_type', 'required_keys', 'schema', 'schema_alias', 'type', 'type_aliases', 'v', 'validate', 'vtype']
```

Adding this filter _before_ the status filter works around the issue:
```yaml
      - type: is-shadow
        state: False
```

Discussion:

Patch:
diff --git a/c7n/resources/cloudtrail.py b/c7n/resources/cloudtrail.py
index 33511347b..dcff8de90 100644
--- a/c7n/resources/cloudtrail.py
+++ b/c7n/resources/cloudtrail.py
@@ -102,12 +102,16 @@ class Status(ValueFilter):
     annotation_key = 'c7n:TrailStatus'
 
     def process(self, resources, event=None):
+
+        non_account_trails = set()
+
         for r in resources:
             region = self.manager.config.region
             trail_arn = Arn.parse(r['TrailARN'])
 
             if (r.get('IsOrganizationTrail') and
                     self.manager.config.account_id != trail_arn.account_id):
+                non_account_trails.add(r['TrailARN'])
                 continue
             if r.get('HomeRegion') and r['HomeRegion'] != region:
                 region = trail_arn.region
@@ -119,7 +123,12 @@ class Status(ValueFilter):
             status.pop('ResponseMetadata')
             r[self.annotation_key] = status
 
-        return super(Status, self).process(resources)
+        if non_account_trails:
+            self.log.warning(
+                'found %d org cloud trail from different account that cant be processed',
+                len(non_account_trails))
+        return super(Status, self).process([
+            r for r in resources if r['TrailARN'] not in non_account_trails])
 
     def __call__(self, r):
         return self.match(r['c7n:TrailStatus'])


Test patch:
diff --git a/tests/data/placebo/test_cloudtrail_org_trail_status_skip/cloudtrail.DescribeTrails_1.json b/tests/data/placebo/test_cloudtrail_org_trail_status_skip/cloudtrail.DescribeTrails_1.json
new file mode 100644
index 000000000..0e13b2f64
--- /dev/null
+++ b/tests/data/placebo/test_cloudtrail_org_trail_status_skip/cloudtrail.DescribeTrails_1.json
@@ -0,0 +1,19 @@
+{
+    "status_code": 200,
+    "data": {
+        "trailList": [
+            {
+                "Name": "orgTrail",
+                "S3BucketName": "c7n-vpc-flow-logs",
+                "IncludeGlobalServiceEvents": true,
+                "IsMultiRegionTrail": true,
+                "HomeRegion": "us-east-1",
+                "TrailARN": "arn:aws:cloudtrail:us-east-1:1122333445566:trail/orgTrail",
+                "LogFileValidationEnabled": true,
+                "HasCustomEventSelectors": false,
+                "IsOrganizationTrail": true
+            }
+        ],
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/data/placebo/test_cloudtrail_org_trail_status_skip/tagging.GetResources_1.json b/tests/data/placebo/test_cloudtrail_org_trail_status_skip/tagging.GetResources_1.json
new file mode 100644
index 000000000..e0ea10a50
--- /dev/null
+++ b/tests/data/placebo/test_cloudtrail_org_trail_status_skip/tagging.GetResources_1.json
@@ -0,0 +1,1 @@
+{"status_code": 200, "data": {"ResourceTagMappingList": []}}
diff --git a/tests/test_cloudtrail.py b/tests/test_cloudtrail.py
index 896335d74..01f8045d0 100644
--- a/tests/test_cloudtrail.py
+++ b/tests/test_cloudtrail.py
@@ -29,6 +29,21 @@ class CloudTrail(BaseTest):
         self.assertEqual(len(resources), 1)
         self.assertTrue('c7n:TrailStatus' in resources[0])
 
+    def test_org_trail_status(self):
+        factory = self.replay_flight_data('test_cloudtrail_org_trail_status_skip')
+        output = self.capture_logging('custodian')
+        p = self.load_policy({
+            'name': 'resource',
+            'resource': 'cloudtrail',
+            'filters': [{'type': 'status', 'key': 'IsLogging', 'value': True}]},
+            session_factory=factory)
+        resources = p.run()
+        self.assertIn(
+            ("found 1 org cloud trail from different"
+             " account that cant be processed"),
+            output.getvalue())
+        self.assertEqual(len(resources), 0)
+
     def test_trail_update(self):
         factory = self.replay_flight_data('test_cloudtrail_update')
         p = self.load_policy({