Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n/resources/securityhub.py: 30%
301 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:51 +0000
1# All Rights Reserved.
2# Copyright The Cloud Custodian Authors.
3# SPDX-License-Identifier: Apache-2.0
4from collections import Counter
5from datetime import datetime
6from dateutil.tz import tzutc
7import json
8import hashlib
9import logging
10import sys
12from c7n import deprecated, query
13from c7n.actions import Action
14from c7n.filters import Filter
15from c7n.manager import resources
16from c7n.query import DescribeSource
17from c7n.exceptions import PolicyValidationError, PolicyExecutionError
18from c7n.policy import LambdaMode, execution
19from c7n.utils import (
20 local_session, type_schema, get_retry,
21 chunks, dumps, filter_empty, get_partition, jmespath_search,
22 merge_dict_list
23)
24from c7n.version import version
26from .aws import AWS
28log = logging.getLogger('c7n.securityhub')
31class SecurityHubFindingFilter(Filter):
32 """Check if there are Security Hub Findings related to the resources
34 :example:
36 By default, this filter checks to see if *any* findings exist for a given
37 resource.
39 .. code-block:: yaml
41 policies:
42 - name: iam-roles-with-findings
43 resource: aws.iam-role
44 filters:
45 - finding
47 :example:
49 The ``query`` parameter can look for specific findings. Consult this
50 `reference <https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_AwsSecurityFindingFilters.html>`_
51 for more information about available filters and their structure. Note that when matching
52 by finding Id, it can be helpful to combine ``PREFIX`` comparisons with parameterized
53 account and region information.
55 .. code-block:: yaml
57 policies:
58 - name: iam-roles-with-global-kms-decrypt
59 resource: aws.iam-role
60 filters:
61 - type: finding
62 query:
63 Id:
64 - Comparison: PREFIX
65 Value: 'arn:aws:securityhub:{region}:{account_id}:subscription/aws-foundational-security-best-practices/v/1.0.0/KMS.2'
66 Title:
67 - Comparison: EQUALS
68 Value: >-
69 KMS.2 IAM principals should not have IAM inline policies
70 that allow decryption actions on all KMS keys
71 ComplianceStatus:
72 - Comparison: EQUALS
73 Value: 'FAILED'
74 RecordState:
75 - Comparison: EQUALS
76 Value: 'ACTIVE'
77 """ # noqa: E501
78 schema = type_schema(
79 'finding',
80 # Many folks do an aggregator region, allow them to use that
81 # for filtering.
82 region={'type': 'string'},
83 query={'type': 'object'})
84 schema_alias = True
85 permissions = ('securityhub:GetFindings',)
86 annotation_key = 'c7n:finding-filter'
87 query_shape = 'AwsSecurityFindingFilters'
89 def validate(self):
90 query = self.data.get('query')
91 if query:
92 from c7n.resources import aws
93 aws.shape_validate(query, self.query_shape, 'securityhub')
95 def process(self, resources, event=None):
96 client = local_session(
97 self.manager.session_factory).client(
98 'securityhub', region_name=self.data.get('region'))
99 found = []
100 params = dict(self.data.get('query', {}))
101 for r_arn, resource in zip(self.manager.get_arns(resources), resources):
102 params['ResourceId'] = [{"Value": r_arn, "Comparison": "EQUALS"}]
103 if resource.get("InstanceId"):
104 params['ResourceId'].append(
105 {"Value": resource["InstanceId"], "Comparison": "EQUALS"})
106 retry = get_retry(('TooManyRequestsException'))
107 findings = retry(client.get_findings, Filters=params).get("Findings")
108 if len(findings) > 0:
109 resource[self.annotation_key] = findings
110 found.append(resource)
111 return found
113 @classmethod
114 def register_resources(klass, registry, resource_class):
115 """ meta model subscriber on resource registration.
117 SecurityHub Findings Filter
118 """
119 if 'post-finding' not in resource_class.action_registry:
120 return
121 if not resource_class.has_arn():
122 return
123 resource_class.filter_registry.register('finding', klass)
126@execution.register('hub-finding')
127class SecurityHub(LambdaMode):
128 """Deploy a policy lambda that executes on security hub finding ingestion events.
130 .. example:
132 This policy will provision a lambda that will process findings from
133 guard duty (note custodian also has support for guard duty events directly)
134 on iam users by removing access keys.
136 .. code-block:: yaml
138 policy:
139 - name: remediate
140 resource: aws.iam-user
141 mode:
142 type: hub-finding
143 role: MyRole
144 filters:
145 - type: event
146 key: detail.findings[].ProductFields.aws/securityhub/ProductName
147 value: GuardDuty
148 - type: event
149 key: detail.findings[].ProductFields.aws/securityhub/ProductName
150 value: GuardDuty
151 actions:
152 - remove-keys
154 Note, for custodian we support additional resources in the finding via the Other resource,
155 so these modes work for resources that security hub doesn't natively support.
157 https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-cloudwatch-events.html
159 """
161 schema = type_schema(
162 'hub-finding', aliases=('hub-action',),
163 rinherit=LambdaMode.schema)
165 ActionFinding = 'Security Hub Findings - Custom Action'
166 ActionInsight = 'Security Hub Insight Results'
167 ImportFinding = 'Security Hub Findings - Imported'
169 handlers = {
170 ActionFinding: 'resolve_action_finding',
171 ActionInsight: 'resolve_action_insight',
172 ImportFinding: 'resolve_import_finding'
173 }
175 def resolve_findings(self, findings):
176 rids = set()
177 for f in findings:
178 for r in f['Resources']:
179 # Security hub invented some new arn format for a few resources...
180 # detect that and normalize to something sane.
181 if r['Id'].startswith('AWS') and r['Type'] == 'AwsIamAccessKey':
182 if 'PrincipalName' in r['Details']['AwsIamAccessKey']:
183 label = r['Details']['AwsIamAccessKey']['PrincipalName']
184 else:
185 label = r['Details']['AwsIamAccessKey']['UserName']
186 rids.add('arn:{}:iam::{}:user/{}'.format(get_partition(r['Region']),
187 f['AwsAccountId'], label))
188 elif not r['Id'].startswith('arn'):
189 if r['Type'] == 'AwsEc2Instance':
190 rids.add('arn:{}:ec2:{}:{}:instance/{}'.format(get_partition(r['Region']),
191 r['Region'], f['AwsAccountId'], r['Id']))
192 else:
193 log.warning("security hub unknown id:%s rtype:%s",
194 r['Id'], r['Type'])
195 else:
196 rids.add(r['Id'])
197 return rids
199 def resolve_action_insight(self, event):
200 rtype = event['detail']['resultType']
201 rids = [list(i.keys())[0] for i in event['detail']['insightResults']]
202 client = local_session(
203 self.policy.session_factory).client('securityhub')
204 insights = client.get_insights(
205 InsightArns=[event['detail']['insightArn']]).get(
206 'Insights', ())
207 if not insights or len(insights) > 1:
208 return []
209 insight = insights.pop()
210 params = {}
211 params['Filters'] = insight['Filters']
212 params['Filters'][rtype] = [
213 {'Comparison': 'EQUALS', 'Value': r} for r in rids]
214 findings = client.get_findings(**params).get('Findings', ())
215 return self.resolve_findings(findings)
217 def resolve_action_finding(self, event):
218 return self.resolve_findings(event['detail']['findings'])
220 def resolve_import_finding(self, event):
221 return self.resolve_findings(event['detail']['findings'])
223 def run(self, event, lambda_context):
224 self.setup_exec_environment(event)
225 resource_sets = self.get_resource_sets(event)
226 result_sets = {}
227 for (account_id, region), rarns in resource_sets.items():
228 self.assume_member({'account': account_id, 'region': region})
229 resources = self.resolve_resources(event)
230 rset = result_sets.setdefault((account_id, region), [])
231 if resources:
232 rset.extend(self.run_resource_set(event, resources))
233 return result_sets
235 def get_resource_sets(self, event):
236 # return a mapping of (account_id, region): [resource_arns]
237 # per the finding in the event.
238 resource_arns = self.get_resource_arns(event)
239 # Group resources by account_id, region for role assumes
240 resource_sets = {}
241 for rarn in resource_arns:
242 resource_sets.setdefault((rarn.account_id, rarn.region), []).append(rarn)
243 # Warn if not configured for member-role and have multiple accounts resources.
244 if (not self.policy.data['mode'].get('member-role') and
245 {self.policy.options.account_id} != {
246 rarn.account_id for rarn in resource_arns}):
247 msg = ('hub-mode not configured for multi-account member-role '
248 'but multiple resource accounts found')
249 self.policy.log.warning(msg)
250 raise PolicyExecutionError(msg)
251 return resource_sets
253 def get_resource_arns(self, event):
254 event_type = event['detail-type']
255 arn_resolver = getattr(self, self.handlers[event_type])
256 arns = arn_resolver(event)
257 # Lazy import to avoid aws sdk runtime dep in core
258 from c7n.resources.aws import Arn
259 return {Arn.parse(r) for r in arns}
261 def resolve_resources(self, event):
262 # For centralized setups in a hub aggregator account
263 resource_map = self.get_resource_arns(event)
265 # sanity check on finding resources matching policy resource
266 # type's service.
267 if self.policy.resource_manager.type != 'account':
268 log.info(
269 "mode:security-hub resolve resources %s", list(resource_map))
270 if not resource_map:
271 return []
272 resource_arns = [
273 r for r in resource_map
274 if r.service == self.policy.resource_manager.resource_type.service]
275 if not resource_arns:
276 log.info("mode:security-hub no matching resources arns")
277 return []
278 resources = self.policy.resource_manager.get_resources(
279 [r.resource for r in resource_arns])
280 else:
281 resources = self.policy.resource_manager.get_resources([])
282 resources[0]['resource-arns'] = resource_arns
283 return resources
286@execution.register('hub-action')
287class SecurityHubAction(SecurityHub):
288 """Deploys a policy lambda as a Security Hub Console Action.
290 .. example:
292 This policy will provision a lambda and security hub custom
293 action. The action can be invoked on a finding or insight result
294 (collection of findings) from within the console. The action name
295 will have the resource type prefixed as custodian actions are
296 resource specific.
298 .. code-block:: yaml
300 policy:
301 - name: remediate
302 resource: aws.ec2
303 mode:
304 type: hub-action
305 role: MyRole
306 actions:
307 - snapshot
308 - type: set-instance-profile
309 name: null
310 - stop
311 """
314FindingTypes = {
315 "Software and Configuration Checks",
316 "TTPs",
317 "Effects",
318 "Unusual Behaviors",
319 "Sensitive Data Identifications"
320}
322# Mostly undocumented value size limit
323SECHUB_VALUE_SIZE_LIMIT = 1024
326class PostFinding(Action):
327 """Report a finding to AWS Security Hub.
329 Custodian acts as a finding provider, allowing users to craft
330 policies that report to the AWS SecurityHub in the AWS Security Finding Format documented at
331 https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format.html
333 For resources that are taggable, we will tag the resource with an identifier
334 such that further findings generate updates. The name of the tag comes from the ``title``
335 parameter of the ``post-finding`` action, or the policy name if ``title`` is empty. This
336 allows different policies to update the same finding if they specify the same ``title``.
338 Example generate a finding for accounts that don't have shield enabled.
340 Note with Cloud Custodian (0.9+) you need to enable the Custodian integration
341 to post-findings, see Getting Started with :ref:`Security Hub <aws-securityhub>`.
343 :example:
345 .. code-block:: yaml
347 policies:
349 - name: account-shield-enabled
350 resource: account
351 filters:
352 - shield-enabled
353 actions:
354 - type: post-finding
355 description: |
356 Shield should be enabled on account to allow for DDOS protection (1 time 3k USD Charge).
357 severity_label: LOW
358 types:
359 - "Software and Configuration Checks/Industry and Regulatory Standards/NIST CSF Controls (USA)"
360 recommendation: "Enable shield"
361 recommendation_url: "https://www.example.com/policies/AntiDDoS.html"
362 title: "shield-enabled"
363 confidence: 100
364 compliance_status: FAILED
366 """ # NOQA
368 deprecations = (
369 deprecated.field('severity_normalized', 'severity_label'),
370 )
372 FindingVersion = "2018-10-08"
374 permissions = ('securityhub:BatchImportFindings',)
376 resource_type = ""
378 schema_alias = True
379 schema = type_schema(
380 "post-finding",
381 required=["types"],
382 title={"type": "string", 'default': 'policy.name'},
383 description={'type': 'string', 'default':
384 'policy.description, or if not defined in policy then policy.name'},
385 severity={"type": "number", 'default': 0},
386 severity_normalized={"type": "number", "min": 0, "max": 100, 'default': 0},
387 severity_label={
388 "type": "string", 'default': 'INFORMATIONAL',
389 "enum": ["INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL"],
390 },
391 confidence={"type": "number", "min": 0, "max": 100},
392 criticality={"type": "number", "min": 0, "max": 100},
393 # Cross region aggregation
394 region={'type': 'string', 'description': 'cross-region aggregation target'},
395 recommendation={"type": "string"},
396 recommendation_url={"type": "string"},
397 fields={"type": "object"},
398 batch_size={'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 1},
399 types={
400 "type": "array",
401 "minItems": 1,
402 "items": {"type": "string"},
403 },
404 compliance_status={
405 "type": "string",
406 "enum": ["PASSED", "WARNING", "FAILED", "NOT_AVAILABLE"],
407 },
408 record_state={
409 "type": "string", 'default': 'ACTIVE',
410 "enum": ["ACTIVE", "ARCHIVED"],
411 },
412 )
414 NEW_FINDING = 'New'
416 def validate(self):
417 for finding_type in self.data["types"]:
418 if finding_type.count('/') > 2 or finding_type.split('/')[0] not in FindingTypes:
419 raise PolicyValidationError(
420 "Finding types must be in the format 'namespace/category/classifier'."
421 " Found {}. Valid namespace values are: {}.".format(
422 finding_type, " | ".join([ns for ns in FindingTypes])))
424 def get_finding_tag(self, resource):
425 finding_tag = None
426 tags = resource.get('Tags', [])
428 finding_key = '{}:{}'.format('c7n:FindingId',
429 self.data.get('title', self.manager.ctx.policy.name))
431 # Support Tags as dictionary
432 if isinstance(tags, dict):
433 return tags.get(finding_key)
435 # Support Tags as list of {'Key': 'Value'}
436 for t in tags:
437 key = t['Key']
438 value = t['Value']
439 if key == finding_key:
440 finding_tag = value
441 return finding_tag
443 def group_resources(self, resources):
444 grouped_resources = {}
445 for r in resources:
446 finding_tag = self.get_finding_tag(r) or self.NEW_FINDING
447 grouped_resources.setdefault(finding_tag, []).append(r)
448 return grouped_resources
450 def process(self, resources, event=None):
451 region_name = self.data.get('region', self.manager.config.region)
452 client = local_session(
453 self.manager.session_factory).client(
454 "securityhub", region_name=region_name)
456 now = datetime.now(tzutc()).isoformat()
457 # default batch size to one to work around security hub console issue
458 # which only shows a single resource in a finding.
459 batch_size = self.data.get('batch_size', 1)
460 stats = Counter()
461 for resource_set in chunks(resources, batch_size):
462 findings = []
463 for key, grouped_resources in self.group_resources(resource_set).items():
464 for resource in grouped_resources:
465 stats['Finding'] += 1
466 if key == self.NEW_FINDING:
467 finding_id = None
468 created_at = now
469 updated_at = now
470 else:
471 finding_id, created_at = self.get_finding_tag(
472 resource).split(':', 1)
473 updated_at = now
475 finding = self.get_finding(
476 [resource], finding_id, created_at, updated_at)
477 findings.append(finding)
478 if key == self.NEW_FINDING:
479 stats['New'] += 1
480 # Tag resources with new finding ids
481 tag_action = self.manager.action_registry.get('tag')
482 if tag_action is None:
483 continue
484 tag_action({
485 'key': '{}:{}'.format(
486 'c7n:FindingId',
487 self.data.get(
488 'title', self.manager.ctx.policy.name)),
489 'value': '{}:{}'.format(
490 finding['Id'], created_at)},
491 self.manager).process([resource])
492 else:
493 stats['Update'] += 1
494 import_response = self.manager.retry(
495 client.batch_import_findings, Findings=findings
496 )
497 if import_response['FailedCount'] > 0:
498 stats['Failed'] += import_response['FailedCount']
499 self.log.error(
500 "import_response=%s" % (import_response))
501 self.log.debug(
502 "policy:%s securityhub %d findings resources %d new %d updated %d failed",
503 self.manager.ctx.policy.name,
504 stats['Finding'],
505 stats['New'],
506 stats['Update'],
507 stats['Failed'])
509 def get_finding(self, resources, existing_finding_id, created_at, updated_at):
510 policy = self.manager.ctx.policy
511 model = self.manager.resource_type
512 region = self.data.get('region', self.manager.config.region)
514 if existing_finding_id:
515 finding_id = existing_finding_id
516 else:
517 # for fips compliance we need to explicit pass the usage param but it doesn't
518 # exist on python 3.8, directly pass when we drop 3.8 support.
519 params = (sys.version_info.major > 3 and sys.version_info.minor > 8) and {
520 'usedforsecurity': False} or {}
521 finding_id = '{}/{}/{}/{}'.format(
522 self.manager.config.region,
523 self.manager.config.account_id,
524 # we use md5 for id, equiv to using crc32
525 hashlib.md5( # nosec nosemgrep
526 json.dumps(policy.data).encode('utf8'),
527 **params).hexdigest(),
528 hashlib.md5( # nosec nosemgrep
529 json.dumps(list(sorted([r[model.id] for r in resources]))).encode('utf8'),
530 **params).hexdigest()
531 )
532 finding = {
533 "SchemaVersion": self.FindingVersion,
534 "ProductArn": "arn:{}:securityhub:{}::product/cloud-custodian/cloud-custodian".format(
535 get_partition(self.manager.config.region),
536 region
537 ),
538 "AwsAccountId": self.manager.config.account_id,
539 # Long search chain for description values, as this was
540 # made required long after users had policies deployed, so
541 # use explicit description, or policy description, or
542 # explicit title, or policy name, in that order.
543 "Description": self.data.get(
544 "description", policy.data.get(
545 "description",
546 self.data.get('title', policy.name))).strip(),
547 "Title": self.data.get("title", policy.name),
548 'Id': finding_id,
549 "GeneratorId": policy.name,
550 'CreatedAt': created_at,
551 'UpdatedAt': updated_at,
552 "RecordState": "ACTIVE",
553 }
555 severity = {'Product': 0, 'Normalized': 0, 'Label': 'INFORMATIONAL'}
556 if self.data.get("severity") is not None:
557 severity["Product"] = self.data["severity"]
558 if self.data.get("severity_label") is not None:
559 severity["Label"] = self.data["severity_label"]
560 # severity_normalized To be deprecated per https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format.html#asff-severity # NOQA
561 if self.data.get("severity_normalized") is not None:
562 severity["Normalized"] = self.data["severity_normalized"]
563 if severity:
564 finding["Severity"] = severity
566 recommendation = {}
567 if self.data.get("recommendation"):
568 recommendation["Text"] = self.data["recommendation"]
569 if self.data.get("recommendation_url"):
570 recommendation["Url"] = self.data["recommendation_url"]
571 if recommendation:
572 finding["Remediation"] = {"Recommendation": recommendation}
574 if "confidence" in self.data:
575 finding["Confidence"] = self.data["confidence"]
576 if "criticality" in self.data:
577 finding["Criticality"] = self.data["criticality"]
578 if "compliance_status" in self.data:
579 finding["Compliance"] = {"Status": self.data["compliance_status"]}
580 if "record_state" in self.data:
581 finding["RecordState"] = self.data["record_state"]
583 fields = {
584 'resource': policy.resource_type,
585 'ProviderName': 'CloudCustodian',
586 'ProviderVersion': version
587 }
589 if "fields" in self.data:
590 fields.update(self.data["fields"])
591 else:
592 tags = {}
593 for t in policy.tags:
594 if ":" in t:
595 k, v = t.split(":", 1)
596 else:
597 k, v = t, ""
598 tags[k] = v
599 fields.update(tags)
600 if fields:
601 finding["ProductFields"] = fields
603 finding_resources = []
604 for r in resources:
605 finding_resources.append(self.format_resource(r))
606 finding["Resources"] = finding_resources
607 finding["Types"] = list(self.data["types"])
609 return filter_empty(finding)
611 def format_envelope(self, r):
612 details = {}
613 envelope = filter_empty({
614 'Id': self.manager.get_arns([r])[0],
615 'Region': self.manager.config.region,
616 'Tags': {t['Key']: t['Value'] for t in r.get('Tags', [])},
617 'Partition': get_partition(self.manager.config.region),
618 'Details': {self.resource_type: details},
619 'Type': self.resource_type
620 })
621 return envelope, details
623 filter_empty = staticmethod(filter_empty)
625 def format_resource(self, r):
626 raise NotImplementedError("subclass responsibility")
629class OtherResourcePostFinding(PostFinding):
631 fields = ()
632 resource_type = 'Other'
634 def format_resource(self, r):
635 details = {}
636 for k in r:
637 if isinstance(k, (list, dict)):
638 continue
639 details[k] = r[k]
641 for f in self.fields:
642 value = jmespath_search(f['expr'], r)
643 if not value:
644 continue
645 details[f['key']] = value
647 for k, v in details.items():
648 if isinstance(v, datetime):
649 v = v.isoformat()
650 elif isinstance(v, (list, dict)):
651 v = dumps(v)
652 elif isinstance(v, (int, float, bool)):
653 v = str(v)
654 else:
655 continue
656 details[k] = v[:SECHUB_VALUE_SIZE_LIMIT]
658 details['c7n:resource-type'] = self.manager.type
659 other = {
660 'Type': self.resource_type,
661 'Id': self.manager.get_arns([r])[0],
662 'Region': self.manager.config.region,
663 'Partition': get_partition(self.manager.config.region),
664 'Details': {self.resource_type: filter_empty(details)}
665 }
666 tags = {t['Key']: t['Value'] for t in r.get('Tags', [])}
667 if tags:
668 other['Tags'] = tags
669 return other
671 @classmethod
672 def register_resource(klass, registry, resource_class):
673 if 'post-finding' not in resource_class.action_registry:
674 resource_class.action_registry.register('post-finding', klass)
677AWS.resources.subscribe(OtherResourcePostFinding.register_resource)
678AWS.resources.subscribe(SecurityHubFindingFilter.register_resources)
681class DescribeSecurityhubFinding(DescribeSource):
682 def resources(self, query):
683 """Only show active compliance failures by default
685 Unless overridden by policy, use these default filters:
687 - Workflow status: anything but RESOLVED
688 - Record state: ACTIVE
689 - Compliance status: FAILED
690 """
692 query = merge_dict_list(
693 [
694 {
695 "Filters": {
696 "WorkflowStatus": [
697 {
698 "Comparison": "NOT_EQUALS",
699 "Value": "RESOLVED",
700 },
701 ],
702 "RecordState": [
703 {
704 "Comparison": "EQUALS",
705 "Value": "ACTIVE",
706 },
707 ],
708 "ComplianceStatus": [
709 {
710 "Comparison": "EQUALS",
711 "Value": "FAILED",
712 }
713 ],
714 }
715 },
716 *self.manager.data.get("query", []),
717 query,
718 ]
719 )
721 return super().resources(query=query)
724@resources.register("securityhub-finding")
725class SecurityhubFinding(query.QueryResourceManager):
726 """AWS SecurityHub Findings
728 :example:
730 Use the default filter set, which includes active unresolved findings
731 that have failed compliance checks
733 .. code-block:: yaml
735 policies:
736 - name: aws-security-hub-findings
737 resource: aws.securityhub-finding
739 :example:
741 Show findings for a specific control ID, overriding default filters
743 .. code-block:: yaml
745 policies:
746 - name: aws-security-hub-findings
747 resource: aws.securityhub-finding
748 query:
749 - Filters:
750 ComplianceSecurityControlId:
751 - Comparison: EQUALS
752 Value: RDS.23
754 Reference for available filters:
756 https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_GetFindings.html#API_GetFindings_RequestSyntax
757 """ # noqa: E501
759 class resource_type(query.TypeInfo):
760 service = "securityhub"
761 enum_spec = ('get_findings', 'Findings', None)
762 arn = False
763 id = "Id"
764 name = "ProductName"
766 source_mapping = {
767 "describe": DescribeSecurityhubFinding,
768 }