1.0.0 🚀 (#155)

* Implement pluggy (#150)

* update requirements

* add hook specs

* add hook implementations

* clean

* add __main__.py

* reorg pluggy code

* reorg pluggy code

* add pluggy code

* upgrade requirements

* Add tests for pluggy

* add docs

* add typing

* upgrade version

* update changelog and docs

* update changelog

* Update tests/config/test_pluggy.py

* Update docs/plugin.md

Co-authored-by: Oliver Crawford <16978487+ocrawford555@users.noreply.github.com>

* update docs

Co-authored-by: Oliver Crawford <16978487+ocrawford555@users.noreply.github.com>

* Remove filters from ruleconfig to be able to load list of filters externally

* Fix some tests

* Fix tests

* Update docs
Add cli parameter

* Change cli help

* Update makefile & requirements.txt (#156)

* fix makefile & update requirements

* Removes clean from makefile

* Remove old whitelisting code in favour of filters (#154)

* PSAS-3128: remove old code

* update changelog

* fix typo

* Feature: don't run disabled rules (#157)

* dont run disabled rules

* add tests

* Refactor result (#158)

* refactor result and failure class

* update callers

* update test utils

* update tests

* update tests

* add more tests

* fix test

* add mising import

* Filters debug

* Rename whitelisted to allowed (#159)

* rename whitelisted to allowed

* update changelog

* fix tests and update changelog
gp

* Fix feature

* Add tests
Simplify implementation

* Fix test

* wip

* remove unncessary [./eu-west-1]

* Update cfripper/config/filter.py

Co-authored-by: Oscar Blanco <oscarbc1996@gmail.com>

* Update cfripper/config/filter.py

Co-authored-by: Oscar Blanco <oscarbc1996@gmail.com>

* improve test

* Complete docs

* add filter context

* fix format

* Rename whitelist Part2 (#161)

* Rename whitelist

* Rename whitelist 2

* Change to collection get_failures (#162)

* Add get exports function to get export values from AWS

* Move lambda main code to docs (#164)

* moved to examples

* add makefile

* update lambda example and remove tests

* revert makefile

* update docs

* update docs

* Update CHANGELOG.md

Co-authored-by: Oliver Crawford <16978487+ocrawford555@users.noreply.github.com>
Co-authored-by: ignaciobolonio <nachobc12@hotmail.com>
Co-authored-by: Oliver Crawford <oliver.crawford@skyscanner.net>

CHANGELOG.md

@@ -1,6 +1,17 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [1.0.0] - 2021-03-16
+### Breaking changes
+- `Filter` include the set of rules in which it is applied.
+- `RuleConfig` only contains `rule_mode` and `risk_value` now.
+- Removes old whitelisting methods in favour of Filters
+- Rename `RuleMode.WHITELISTED` to `RuleMode.ALLOWED`, and all `whitelist` word in strings.
+- Add debug flag to `Filter` class.
+### Improvements
+- Implements `pluggy` https://github.com/pytest-dev/pluggy to enable dynamic rule loading.
+- Add support to load filters from external files
+
 ## [0.23.3] - 2021-02-11
 ### Additions
 - All rules now support filter contexts!

Makefile

@@ -1,13 +1,9 @@
 SOURCES = $(shell find . -name "*.py")
 
-clean:
-	rm -f lambda.zip
-	rm -rf package
-
 install:
 	pip install -r requirements.txt
 
-install-dev: install
+install-dev:
 	pip install -e ".[dev]"
 
 install-docs:
@@ -40,17 +36,9 @@ coverage:
 test: lint unit
 
 freeze:
-	CUSTOM_COMPILE_COMMAND="make freeze" pip-compile --no-emit-index-url --output-file requirements.txt setup.py
+	CUSTOM_COMPILE_COMMAND="make freeze" pip-compile --no-emit-index-url --no-annotate --output-file requirements.txt setup.py
 
 freeze-upgrade:
-	CUSTOM_COMPILE_COMMAND="make freeze-upgrade" pip-compile --no-emit-index-url --upgrade --output-file requirements.txt setup.py
-
-lambda.zip: $(SOURCES) Makefile requirements.txt
-	if [ -f lambda.zip ]; then rm lambda.zip; fi
-	if [ -d "./package" ]; then rm -rf package/; fi
-	pip install -t package -r requirements.txt
-	cp -r cfripper package/cfripper
-	cd ./package && zip -rq ../lambda.zip .
-	rm -rf ./package
+	CUSTOM_COMPILE_COMMAND="make freeze" pip-compile --no-emit-index-url --upgrade --no-annotate --output-file requirements.txt setup.py
 
-.PHONY: clean install install-dev format lint isort-lint black-lint flake8-lint unit coverage test freeze freeze-upgrade
+.PHONY: install install-dev install-docs format lint isort-lint black-lint flake8-lint unit coverage test freeze freeze-upgrade

README.md

@@ -68,6 +68,15 @@ Analysing tests/test_templates/config/security_group_firehose_ips.json...
 Valid: True
 ```
 
+### Using rules filters files
+
+```bash
+$ cfripper tests/test_templates/config/security_group_firehose_ips.json --rules-filters-folder cfripper/config/rule_configs/
+example_rules_config_for_cli.py loaded
+Analysing tests/test_templates/config/security_group_firehose_ips.json...
+Valid: True
+```
+
 ### Exit Codes
 
 ```python

cfripper/__main__.py

@@ -0,0 +1,4 @@
+from cfripper.cli import cli
+
+if __name__ == "__main__":
+    cli()

cfripper/__version__.py

@@ -1,3 +1,3 @@
-VERSION = (0, 23, 3)
+VERSION = (1, 0, 0)
 
 __version__ = ".".join(map(str, VERSION))

cfripper/boto3_client.py

@@ -76,3 +76,11 @@ def download_template_to_dictionary(self, s3_url):
         file_contents = response["Body"].read().decode("utf-8")
 
         return convert_json_or_yaml_to_dict(file_contents)
+
+    def get_exports(self) -> Dict[str, str]:
+        client = self.session.client("cloudformation", region_name=self.region)
+        try:
+            return {export["Name"]: export["Value"] for export in client.list_exports()["Exports"]}
+        except Exception:
+            logger.exception(f"Could not get AWS Export values! ({self.account_id} - {self.region})")
+            return {}

cfripper/cli.py

@@ -10,11 +10,12 @@
 
 from cfripper.__version__ import __version__
 from cfripper.config.config import Config
+from cfripper.config.pluggy.utils import get_all_rules
 from cfripper.exceptions import FileEmptyException
+from cfripper.model.enums import RuleMode
 from cfripper.model.result import Result
 from cfripper.model.utils import convert_json_or_yaml_to_dict
 from cfripper.rule_processor import RuleProcessor
-from cfripper.rules import DEFAULT_RULES
 
 LOGGING_LEVELS = {
     "ERROR": logging.ERROR,
@@ -28,11 +29,16 @@ def setup_logging(level: str) -> None:
     logging.basicConfig(level=LOGGING_LEVELS[level], format="%(message)s")
 
 
-def init_cfripper(rules_config_file: Optional[TextIOWrapper]) -> Tuple[Config, RuleProcessor]:
-    config = Config(rules=DEFAULT_RULES.keys())
+def init_cfripper(
+    rules_config_file: Optional[TextIOWrapper], rules_filters_folder: Optional[str]
+) -> Tuple[Config, RuleProcessor]:
+    rules = get_all_rules()
+    config = Config(rules=rules.keys())
     if rules_config_file:
         config.load_rules_config_file(rules_config_file)
-    rule_processor = RuleProcessor(*[DEFAULT_RULES.get(rule)(config) for rule in config.rules])
+    if rules_filters_folder:
+        config.add_filters_from_dir(rules_filters_folder)
+    rule_processor = RuleProcessor(*[rules.get(rule)(config) for rule in config.rules])
     return config, rule_processor
 
 
@@ -54,12 +60,17 @@ def format_result_json(result: Result) -> str:
 
 def format_result_txt(result: Result) -> str:
     result_lines = [f"Valid: {result.valid}"]
-    if result.failed_rules:
+
+    blocking_rules = result.get_failures(include_rule_modes={RuleMode.BLOCKING})
+    if blocking_rules:
         result_lines.append("Issues found:")
-        [result_lines.append(f"\t- {r.rule}: {r.reason}") for r in result.failed_rules]
-    if result.failed_monitored_rules:
+        [result_lines.append(f"\t- {r.rule}: {r.reason}") for r in blocking_rules]
+
+    monitoring_rules = result.get_failures(include_rule_modes={RuleMode.MONITOR})
+    if monitoring_rules:
         result_lines.append("Monitored issues found:")
-        [result_lines.append(f"\t- {r.rule}: {r.reason}") for r in result.failed_monitored_rules]
+        [result_lines.append(f"\t- {r.rule}: {r.reason}") for r in monitoring_rules]
+
     return "\n".join(result_lines)
 
 
@@ -93,14 +104,15 @@ def process_template(
     output_folder: Optional[str],
     output_format: str,
     rules_config_file: Optional[TextIOWrapper],
+    rules_filters_folder: Optional[str],
 ) -> bool:
     logging.info(f"Analysing {template.name}...")
 
     cfmodel = get_cfmodel(template)
     if resolve:
         cfmodel = cfmodel.resolve(resolve_parameters)
 
-    config, rule_processor = init_cfripper(rules_config_file)
+    config, rule_processor = init_cfripper(rules_config_file, rules_filters_folder)
 
     result = analyse_template(cfmodel, rule_processor, config)
 
@@ -153,6 +165,11 @@ def process_template(
 @click.option(
     "--rules-config-file", type=click.File("r"), help="Loads rules configuration file (type: [.py, .pyc])",
 )
+@click.option(
+    "--rules-filters-folder",
+    type=click.Path(exists=True, resolve_path=True, readable=True, file_okay=False),
+    help="All files in the folder must be of type: [.py, .pyc]",
+)
 def cli(templates, logging_level, resolve_parameters, **kwargs):
     """
     Analyse AWS Cloudformation templates passed by parameter.

cfripper/config/config.py

@@ -1,18 +1,18 @@
 import importlib
+import itertools
 import logging
-import os
-import re
 import sys
+from collections import defaultdict
 from io import TextIOWrapper
-from typing import Dict, List
+from pathlib import Path
+from typing import DefaultDict, Dict, List
 
 from pydantic import BaseModel
 
+from cfripper.config.constants import AWS_ELASTICACHE_BACKUP_CANONICAL_IDS, AWS_ELB_LOGS_ACCOUNT_IDS
+
+from .filter import Filter
 from .rule_config import RuleConfig
-from .whitelist import AWS_ELASTICACHE_BACKUP_CANONICAL_IDS, AWS_ELB_LOGS_ACCOUNT_IDS
-from .whitelist import rule_to_action_whitelist as default_rule_to_action_whitelist
-from .whitelist import rule_to_resource_whitelist as default_rule_to_resource_whitelist
-from .whitelist import stack_whitelist as default_stack_whitelist
 
 logger = logging.getLogger(__file__)
 
@@ -92,10 +92,8 @@ def __init__(
         aws_user_agent=None,
         aws_principals=None,
         aws_service_accounts=None,
-        stack_whitelist=None,
-        rule_to_action_whitelist=None,
-        rule_to_resource_whitelist=None,
         rules_config=None,
+        rules_filters=None,
     ):
         self.project_name = project_name
         self.service_name = service_name
@@ -107,13 +105,6 @@ def __init__(
         self.aws_account_name = aws_account_name
         self.aws_account_id = aws_account_id
         self.aws_user_agent = aws_user_agent
-        self.rule_to_action_whitelist = (
-            rule_to_action_whitelist if rule_to_action_whitelist is not None else default_rule_to_action_whitelist
-        )
-        self.rule_to_resource_whitelist = (
-            rule_to_resource_whitelist if rule_to_resource_whitelist is not None else default_rule_to_resource_whitelist
-        )
-        self.stack_whitelist = stack_whitelist if stack_whitelist is not None else default_stack_whitelist
         if aws_service_accounts is None:
             self.aws_service_accounts = {
                 "elb_logs_account_ids": AWS_ELB_LOGS_ACCOUNT_IDS,
@@ -122,11 +113,6 @@ def __init__(
         else:
             self.aws_service_accounts = aws_service_accounts
 
-        if self.stack_name:
-            whitelisted_rules = self.get_whitelisted_rules()
-            # set difference to get a list of allowed rules to be ran for this stack
-            self.rules = list(set(self.rules) - set(whitelisted_rules))
-
         self.allowed_world_open_ports = list(self.DEFAULT_ALLOWED_WORLD_OPEN_PORTS)
 
         self.forbidden_managed_policy_arns = list(self.DEFAULT_FORBIDDEN_MANAGED_POLICY_ARNS)
@@ -135,7 +121,11 @@ def __init__(
 
         # Set up a string list of allowed principals. If kept empty it will allow any AWS principal
         self.aws_principals = aws_principals if aws_principals is not None else []
+
         self.rules_config = rules_config if rules_config is not None else {}
+        self.rules_filters: DefaultDict[str, List[Filter]] = defaultdict(list)
+        if rules_filters:
+            self.add_filters(rules_filters)
 
     def get_rule_config(self, rule_name: str) -> RuleConfig:
         rule_config = self.rules_config.get(rule_name)
@@ -145,38 +135,17 @@ def get_rule_config(self, rule_name: str) -> RuleConfig:
             return rule_config
         return RuleConfig(**rule_config)
 
-    def get_whitelisted_actions(self, rule_name: str) -> List[str]:
-        allowed_actions = []
-        for k, v in self.rule_to_action_whitelist.get(rule_name, {}).items():
-            if re.match(k, self.stack_name):
-                allowed_actions += v
-
-        return allowed_actions
-
-    def get_whitelisted_resources(self, rule_name: str) -> List[str]:
-        allowed_resources = []
-        for k, v in self.rule_to_resource_whitelist.get(rule_name, {}).items():
-            if re.match(k, self.stack_name):
-                allowed_resources += v
-
-        return allowed_resources
-
-    def get_whitelisted_rules(self) -> List[str]:
-        whitelisted_rules = []
-        for k, v in self.stack_whitelist.items():
-            if re.match(k, self.stack_name):
-                whitelisted_rules += v
-
-        return whitelisted_rules
+    def get_rule_filters(self, rule_name: str) -> List[Filter]:
+        return self.rules_filters.get(rule_name, [])
 
     def load_rules_config_file(self, rules_config_file: TextIOWrapper):
         filename = rules_config_file.name
 
-        if not os.path.exists(filename):
+        if not Path(filename).is_file():
             raise RuntimeError(f"{filename} doesn't exist")
 
         try:
-            ext = os.path.splitext(filename)[1]
+            ext = Path(filename).suffix
             module_name = "__rules_config__"
             if ext not in [".py", ".pyc"]:
                 raise RuntimeError("Configuration file should have a valid Python extension.")
@@ -192,6 +161,38 @@ def load_rules_config_file(self, rules_config_file: TextIOWrapper):
             logger.exception(f"Failed to read config file: {filename}")
             raise
 
+    def add_filters_from_dir(self, path: str):
+        if not Path(path).is_dir():
+            raise RuntimeError(f"{path} doesn't exist")
+
+        try:
+            module_name = "__rules_config__"
+            filenames = sorted(itertools.chain(Path(path).glob("*.py"), Path(path).glob("*.pyc")))
+            for filename in filenames:
+                spec = importlib.util.spec_from_file_location(module_name, filename.absolute())
+                module = importlib.util.module_from_spec(spec)
+                sys.modules[module_name] = module
+                spec.loader.exec_module(module)
+                filters = vars(module).get("FILTERS")
+                if not filters:
+                    continue
+                # Validate filters format
+                RulesFiltersMapping(__root__=filters)
+                self.add_filters(filters=filters)
+                logger.info(f"{filename} loaded")
+        except Exception:
+            logger.exception(f"Failed to read files in path: {path}")
+            raise
+
+    def add_filters(self, filters: List[Filter]):
+        for filter in filters:
+            for rule in filter.rules:
+                self.rules_filters[rule].append(filter)
+
 
 class RulesConfigMapping(BaseModel):
     __root__: Dict[str, RuleConfig]
+
+
+class RulesFiltersMapping(BaseModel):
+    __root__: List[Filter]