Add git support and remove driver interdependency.

Fix tests with new changes and add full coverage.

Author: William Chrisp

publisher/drivers/checkov.py

@@ -1,55 +1,31 @@
 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 (
+from publisher.entities.events import (
+    Event,
     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]]:
+class Checkov(EventSource):    
+    def get_events(self, file_data: dict, repo_name: 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"]:
+        if "results" in file_data:
+            for result in file_data["results"]["passed_checks"]:
+                events.append(GuardrailPassed(
+                    aggregate_id = repo_name + "." + result["resource"],
+                    guardrail_id = result["check_id"],
+                    time = current_time,
+                ))
+            for result in file_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,
-                    )
+                    aggregate_id = repo_name + "." + result["resource"],
+                    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}")
+        return Exception(f"Unable to read Checkov results from file")
 

publisher/drivers/event_bridge.py

@@ -1,16 +1,12 @@
 import json
-from typing import Dict, Any, Optional, List, Union
+from typing import Dict, Any, 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")
@@ -22,22 +18,22 @@ def _split_events(self, events: List[Dict[str, Any]]) -> Any:
 
     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,
+            "Source": "contino.flight_controller",
+            "DetailType": event.event_type,
+            "Detail": json.dumps(event.__dict__),
+            "EventBusName": "main_lambda_bus_cdktf",
         }
 
-    def send_events(self, events: List[Event]) -> Optional[Union[Exception, str]]:
+    def send_events(self, events: List[Event]) -> Union[Exception, str]:
         try:
             events = [self._format_events(event) for event in events]
             if len(events) > 10:
                 event_groups = self._split_events(events)
+                response = []
                 for event_group in event_groups:
-                    response = self.event_bridge_client.put_events(Entries=event_group)
+                    response.append(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/event_sink.py

@@ -6,5 +6,5 @@
 
 class EventSink(ABC):
     @abstractmethod
-    def send_events(self, events: List[Event]) -> Optional[Union[Exception, str]]:
+    def send_events(self, events: List[Event]) -> Union[Exception, str]:
         raise NotImplementedError()  # pragma: no cover

publisher/drivers/event_source.py

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

publisher/drivers/file_source.py

@@ -1,20 +1,13 @@
 import json
-from typing import Dict
-
-import structlog
-
-LOGGER = structlog.get_logger(__name__)
+from typing import Dict, Union
 
 
 class FileSource():
-    def read_file(self, file: str) -> Dict:
+    def read_file(self, file: str) -> Union[Exception, 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/git.py

@@ -0,0 +1,17 @@
+import subprocess
+from typing import Union
+
+
+class Git():
+    def _check_git_available(self) -> bool:
+        return subprocess.run(['git', '--version'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0
+
+    def get_repo_name(self) -> Union[Exception, str]:
+        if self._check_git_available() == True:
+            try:
+                repo_name = subprocess.check_output(['git', 'rev-parse', '--show-toplevel'], stderr=subprocess.DEVNULL).strip().decode('utf-8').split('/')[-1]
+                return repo_name
+            except Exception as err:
+                return err
+        else:
+            return Exception("Git not available")

publisher/drivers/open_policy_agent.py

@@ -1,54 +1,31 @@
 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 (
+from publisher.entities.events import (
+    Event,
     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]]:
+    def get_events(self, file_data: dict, repo_name: 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 "results" in file_data:
+            for result in file_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,
-                        )
+                        aggregate_id = repo_name + "." + result["input"]["metadata"]["namespace"] + "."  + result["input"]["metadata"]["name"],
+                        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,
-                        )
+                        aggregate_id = repo_name + "." + result["input"]["metadata"]["namespace"] + "." + result["input"]["metadata"]["name"],
+                        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}")
-
+        return Exception(f"Unable to read Open Policy Agent results from file")

publisher/entities/base.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
-from typing import Literal
+
 
 # Base event used across all other events
 @dataclass
 class BaseEvent:
-    detail_type: str
+    aggregate_id: str

publisher/entities/events.py

@@ -2,9 +2,7 @@
 
 from publisher.entities.guardrail import (
     GuardrailActivated,
-    GuardrailActivatedDetail,
     GuardrailPassed,
-    GuardrailPassedDetail,
 )
 
 
@@ -20,12 +18,6 @@
 ]
 
 
-Detail = Union[
-    GuardrailActivatedDetail,
-    GuardrailPassedDetail
-]
-
-
 EVENT_CLASSES = {
     "guardrail_activated": GuardrailActivated,
     "guardrail_passed": GuardrailPassed

publisher/entities/guardrail.py

@@ -1,31 +1,18 @@
 from dataclasses import dataclass
-from typing import Union, Literal
+from typing import Literal
 
 from publisher.entities.base import BaseEvent
 
 
 @dataclass
-class GuardrailActivatedDetail():
-    aggregate_id: str
+class GuardrailActivated(BaseEvent):
     guardrail_id: str
     time: int
     event_type: Literal["guardrail_activated"] = "guardrail_activated"
 
-@dataclass
-class GuardrailActivated(BaseEvent):
-    source: str
-    detail: GuardrailActivatedDetail
-    event_bus_name: Literal["main_lambda_bus_cdktf"] = "main_lambda_bus_cdktf"
 
 @dataclass
-class GuardrailPassedDetail():
-    aggregate_id: str
+class GuardrailPassed(BaseEvent):
     guardrail_id: str
     time: int
-    event_type: Literal["guardrail_passed"] = "guardrail_passed"
-
-@dataclass
-class GuardrailPassed(BaseEvent):
-    source: str
-    detail: GuardrailPassedDetail
-    event_bus_name: Literal["main_lambda_bus_cdktf"] = "main_lambda_bus_cdktf"
+    event_type: Literal["guardrail_passed"] = "guardrail_passed"