| 24 | print_exception(f'Failed to create rule {rule.filename}: {e}') |
| 25 | |
| 26 | def run(self, cloud_provider, skip_dashboard=False): |
| 27 | # Clean up existing findings |
| 28 | for service in cloud_provider.services: |
| 29 | cloud_provider.services[service][self.ruleset.rule_type] = {} |
| 30 | |
| 31 | # Process each rule |
| 32 | for finding_path in self._filter_rules(self.rules, cloud_provider.service_list): |
| 33 | for rule in self.rules[finding_path]: |
| 34 | |
| 35 | if not rule.enabled: # or rule.service not in []: # TODO: handle this... |
| 36 | continue |
| 37 | |
| 38 | print_debug(f'Processing {rule.service} rule "{rule.description}" ({rule.filename})') |
| 39 | finding_path = rule.path |
| 40 | path = finding_path.split('.') |
| 41 | service = path[0] |
| 42 | manage_dictionary(cloud_provider.services[service], self.ruleset.rule_type, {}) |
| 43 | cloud_provider.services[service][self.ruleset.rule_type][rule.key] = {} |
| 44 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['description'] = rule.description |
| 45 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['path'] = rule.path |
| 46 | for attr in ['level', 'id_suffix', 'class_suffix', 'display_path']: |
| 47 | if hasattr(rule, attr): |
| 48 | cloud_provider.services[service][self.ruleset.rule_type][rule.key][attr] = getattr(rule, attr) |
| 49 | try: |
| 50 | setattr(rule, 'checked_items', 0) |
| 51 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['items'] = recurse( |
| 52 | cloud_provider.services, cloud_provider.services, path, [], rule, True) |
| 53 | if skip_dashboard: |
| 54 | continue |
| 55 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['dashboard_name'] = \ |
| 56 | rule.dashboard_name |
| 57 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['checked_items'] = \ |
| 58 | rule.checked_items |
| 59 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['flagged_items'] = \ |
| 60 | len(cloud_provider.services[service][self.ruleset.rule_type][rule.key]['items']) |
| 61 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['service'] = rule.service |
| 62 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['rationale'] = \ |
| 63 | rule.rationale if hasattr(rule, 'rationale') else None |
| 64 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['remediation'] = \ |
| 65 | rule.remediation if hasattr(rule, 'remediation') else None |
| 66 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['compliance'] = \ |
| 67 | rule.compliance if hasattr(rule, 'compliance') else None |
| 68 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['references'] = \ |
| 69 | rule.references if hasattr(rule, 'references') else None |
| 70 | except Exception as e: |
| 71 | print_exception(f'Failed to process rule defined in {rule.filename}: {e}') |
| 72 | # Fallback if process rule failed to ensure report creation and data dump still happen |
| 73 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['checked_items'] = 0 |
| 74 | cloud_provider.services[service][self.ruleset.rule_type][rule.key]['flagged_items'] = 0 |
| 75 | |
| 76 | @staticmethod |
| 77 | def _filter_rules(rules, services): |