Initial custom publisher files and test.

Author: William Chrisp

Makefile

@@ -22,7 +22,7 @@ integration-test:
 	pipenv run pytest -m 'integration' tests/
 
 test:
-	pipenv run pytest --cov=src --cov-fail-under=81 --cov-report term-missing tests/
+	pipenv run pytest --cov=src --cov=publisher --cov-fail-under=81 --cov-report term-missing tests/
 
 watch:
 	ptw -- -m 'not integration' tests/

publisher/README.md

@@ -0,0 +1,69 @@
+<h1 align="center">Flight Controller Custom Publisher</h1>
+<p align="center">Allows you to publish custom events for flight controller to produce metrics.</p>
+
+# Usage
+
+## Prerequsites
+- Python version 3.10
+- Pipenv (version 2022.10.12 proven working)
+- Ensure you have working AWS account creds/tokens.
+
+`make local` - Install python prerequisites and enter the pip environment.
+
+## Base command:
+`python -m publisher.entrypoints.main`
+
+## Arguments: 
+
+| Option (Short) | Options (Long) |                                   Description                                  |
+|:--------------:|:--------------:|:------------------------------------------------------------------------------:|
+| -h             | --help         | Shows command help and all available options                                   |
+| -so            | --source       | {open_policy_agent,checkov} - The source in which you want to get events from. |
+| -f             | --file         | The json file in which to parse result into events.                            |
+| -si            | --sink         | {event_bridge} - The sink in which you want to send events to.                 |
+
+## Example Workflow:
+
+You can test with the examples in `tests/examples` and just running the the publisher tool or by using in your workflow like below.
+
+**Custom Checkov Guardrail Events** 
+1. `terraform plan --out tfplan.binary; terraform show -json tfplan.binary > tfplan.json`
+2. `checkov -f tfplan.json -o json checkov.json`
+3. `python -m publisher.entrypoints.main -so checkov -f tests/examples/checkov.json -si event_bridge`
+
+**Custom Open Policy Agent Events**  
+`python -m publisher.entrypoints.main -so open_policy_agent -f tests/examples/opa.json -si event_bridge`
+
+**Example Workflow**  
+1. Run Checkov on your Infrastructure as Code and output this to a .json file
+2. Run Flight Controller handler.py on this file to parse results and generate events.  
+   
+**This can be done with in your CICD**
+
+# Developing
+
+## Code Structure
+The code is structured in the [Clean Architecture](https://blog.cleancoder.com/uncle-bob/2012/08/13/the-clean-architecture.html) pattern.
+
+![Clean Architecture](../images/CleanArchitecture.jpeg)
+
+The current layers are:
+
+1. `Entities`, which contains domain objects
+2. `Drivers`, which interact with data storage
+3. `Entrypoints`, which handle the event from AWS, retrieve and store data through drivers and call adapters to perform the needed business logic
+
+The core rule of Clean Architecture, is that a layer can only depend on the layers that have come before it. E.g. code in the `usecases` layer, may depend on `entities`, but cannot depend on `adapters` or `drivers`.
+
+When developing, it is simplest to start at the first layer and work down ending up with the entrypoint. This forces you to focus on the domain objects first before considering external services.
+
+## Adding more support
+
+Adding support for a new source:  
+`publisher/drivers/` 
+
+Adding support for a new event type:  
+`publisher/entities/`
+
+Adding support for a new sink:  
+`publisher/drivers/`

tests/adapters/__init__.py renamed to publisher/__init__.py

@@ -0,0 +1,55 @@
+from time import time
+from typing import Tuple, Union
+from uuid import uuid4
+
+import structlog
+
+from publisher.drivers.event_source import EventSource
+from publisher.drivers.file_source import FileSource
+from publisher.entities.events import Event
+from publisher.entities.guardrail import (
+    GuardrailActivated,
+    GuardrailActivatedDetail,
+    GuardrailPassed,
+    GuardrailPassedDetail,
+)
+
+LOGGER = structlog.get_logger(__name__)
+
+
+class Checkov(EventSource):
+    def __init__(self) -> None:
+         self.file_source = FileSource()
+    
+    def get_events(self, file: str) -> Union[Exception, Tuple[Event]]:
+        current_time = int(time())
+        data = self.file_source.read_file(file)
+        if isinstance(data, Exception):
+            LOGGER.error(f"Unable to read Checkov results file: {file}", exception=str(data))
+            return data
+        events = []
+        if "results" in data:
+            for result in data["results"]["passed_checks"]:
+                    events.append(GuardrailPassed(
+                        source = "contino.custom",
+                        detail_type = "Checkov Guardrail Passed",
+                        detail = GuardrailPassedDetail(
+                            aggregate_id = result["resource"], # Will need a better way of getting resource id, current method is not live id
+                            guardrail_id = result["check_id"],
+                            time = current_time,
+                        )
+                    ))
+            for result in data["results"]["failed_checks"]:
+                events.append(GuardrailActivated(
+                    source = "contino.custom",
+                    detail_type = "Checkov Guardrail Activated",
+                    detail = GuardrailActivatedDetail(
+                        aggregate_id = result["resource"], # Will need a better way of getting resource id, current method is not live id
+                        guardrail_id = result["check_id"],
+                        time = current_time,
+                    )
+                ))
+            return tuple(events)
+        LOGGER.error(f"Unable to read Checkov results from file: {file}")
+        return Exception(f"Unable to read Checkov results from file: {file}")
+        

tests/drivers/__init__.py renamed to publisher/drivers/__init__.py

@@ -0,0 +1,43 @@
+import json
+from typing import Dict, Any, Optional, List, Union
+
+import boto3
+import structlog
+
+from publisher.drivers.event_sink import EventSink
+from publisher.entities.events import Event
+
+
+LOGGER = structlog.get_logger(__name__)
+
+
+class EventBridge(EventSink):
+    def __init__(self) -> None:
+        self.event_bridge_client = boto3.client("events")
+
+    def _split_events(self, events: List[Dict[str, Any]]) -> Any:
+        # Will split events based on eventbridge constraint for 10 at a time
+        for i in range(0, len(events), 10):
+            yield events[i:i + 10]
+
+    def _format_events(self, event: Dict) -> Dict:
+        return {
+            "Source": event.source,
+            "DetailType": event.detail_type,
+            "Detail": json.dumps(event.detail.__dict__),
+            "EventBusName": event.event_bus_name,
+        }
+
+    def send_events(self, events: List[Event]) -> Optional[Union[Exception, str]]:
+        try:
+            events = [self._format_events(event) for event in events]
+            if len(events) > 10:
+                event_groups = self._split_events(events)
+                for event_group in event_groups:
+                    response = self.event_bridge_client.put_events(Entries=event_group)
+            else:
+                response = self.event_bridge_client.put_events(Entries=events)
+            return str(response)
+        except Exception as err:
+            return err
+        

publisher/drivers/checkov.py

@@ -0,0 +1,10 @@
+from abc import ABC, abstractmethod
+from typing import List, Optional, Union
+
+from publisher.entities.events import Event
+
+
+class EventSink(ABC):
+    @abstractmethod
+    def send_events(self, events: List[Event]) -> Optional[Union[Exception, str]]:
+        raise NotImplementedError()  # pragma: no cover

publisher/drivers/event_bridge.py

@@ -0,0 +1,11 @@
+from abc import ABC, abstractmethod
+from typing import Tuple, Union
+
+from publisher.entities.events import Event
+
+
+class EventSource(ABC):
+    @abstractmethod
+    def get_events(self, file: str) -> Union[Exception, Tuple[Event]]:
+        raise NotImplementedError()  # pragma: no cover
+    

publisher/drivers/event_sink.py

@@ -0,0 +1,20 @@
+import json
+from typing import Dict
+
+import structlog
+
+LOGGER = structlog.get_logger(__name__)
+
+
+class FileSource():
+    def read_file(self, file: str) -> Dict:
+        try:
+            with open(file, "r") as file:
+                return json.loads(file.read())
+        except FileNotFoundError as err:
+            LOGGER.error(f"File {file} not found!", exception=str(err))
+            return err
+        except Exception as err:
+            LOGGER.error(f"file_source.py raised an exception", exception=str(err))
+            return err
+

publisher/drivers/event_source.py

@@ -0,0 +1,54 @@
+from time import time
+from typing import Tuple, Union
+
+import structlog
+
+from publisher.drivers.event_source import EventSource
+from publisher.drivers.file_source import FileSource
+from publisher.entities.events import Event
+from publisher.entities.guardrail import (
+    GuardrailActivated,
+    GuardrailActivatedDetail,
+    GuardrailPassed,
+    GuardrailPassedDetail,
+)
+
+LOGGER = structlog.get_logger(__name__)
+
+class OpenPolicyAgent(EventSource):
+    def __init__(self) -> None:
+        self.file_source = FileSource()
+
+    def get_events(self, file: str) -> Union[Exception, Tuple[Event]]:
+        current_time = int(time())
+        data = self.file_source.read_file(file)
+        if isinstance(data, Exception):
+            LOGGER.error(f"Unable to read Open Policy Agent results file: {file}", exception=str(data))
+            return data
+        events = []
+        if "results" in data:
+            for result in data["results"]:
+                if result["allow"] == True:
+                    events.append(GuardrailPassed(
+                        source = "contino.custom",
+                        detail_type = "Open Policy Agent Guardrail Passed",
+                        detail = GuardrailPassedDetail(
+                            aggregate_id = result["input"]["metadata"]["name"] + "-" + result["input"]["metadata"]["namespace"],
+                            guardrail_id = result["query"],
+                            time = current_time,
+                        )
+                    ))
+                else:
+                    events.append(GuardrailActivated(
+                        source = "contino.custom",
+                        detail_type = "Open Policy Agent Guardrail Activated",
+                        detail = GuardrailActivatedDetail(
+                            aggregate_id = result["input"]["metadata"]["name"] + "-" + result["input"]["metadata"]["namespace"],
+                            guardrail_id = result["query"],
+                            time = current_time,
+                        )
+                    ))
+            return tuple(events)
+        LOGGER.error(f"Unable to read Open Policy Agent results from file: {file}")
+        return Exception(f"Unable to read Open Policy Agent results from file: {file}")
+