refactor/sarif amqp delivery impl

Author: cx-nathan-leach

agent/resolver/results_agent.py

@@ -1,4 +1,4 @@
-from workflows.base_service import BaseWorkflowService
+from workflows.base_service import CxOneFlowAbstractWorkflowService
 from workflows.messaging import DelegatedScanResultMessage
 from services import CxOneFlowServices
 from api_utils.auth_factories import EventContext
@@ -8,7 +8,7 @@
 import aio_pika, gzip, importlib
 from typing import List
 
-class ResolverResultsAgent(BaseWorkflowService):
+class ResolverResultsAgent(CxOneFlowAbstractWorkflowService):
 
     def __init__(self, services : CxOneFlowServices):
         self.__services = services

agent/resolver/runner_agent.py

@@ -1,5 +1,5 @@
 from workflows.resolver_workflow import ResolverScanningWorkflow
-from workflows.base_service import BaseWorkflowService
+from workflows.base_service import CxOneFlowAbstractWorkflowService
 from workflows.resolver_scan_service import ResolverScanService
 from workflows.messaging import (
     DelegatedScanMessage,
@@ -17,7 +17,7 @@
 from _version import __version__
 
 
-class ResolverRunnerAgent(BaseWorkflowService):
+class ResolverRunnerAgent(CxOneFlowAbstractWorkflowService):
 
     def __init__(
         self,

agent/resolver/timeout_agent.py

@@ -1,9 +1,9 @@
-from workflows.base_service import BaseWorkflowService
+from workflows.base_service import CxOneFlowAbstractWorkflowService
 from services import CxOneFlowServices
 from workflows.messaging import DelegatedScanMessage
 import aio_pika
 
-class ResolverTimeoutAgent(BaseWorkflowService):
+class ResolverTimeoutAgent(CxOneFlowAbstractWorkflowService):
 
     def __init__(self, services : CxOneFlowServices):
         self.__services = services

config/server/__init__.py

@@ -2,7 +2,8 @@
 from _agent import __agent__
 from pathlib import Path
 from config import ConfigurationException, RouteNotFoundException, CommonConfig
-import re, uuid, sys
+import re, uuid, sys, dataclasses_json
+from dataclasses import make_dataclass, asdict
 from importlib import import_module
 from scm_services import SCMService, ADOEService, BBDCService, GHService, GLService
 from scm_services.cloner import Cloner
@@ -18,7 +19,7 @@
 from workflows.resolver_scan_service import ResolverScanService
 from workflows.pull_request import PullRequestWorkflow
 from workflows.push import PushWorkflow
-from workflows.base_service import BaseWorkflowService
+from workflows.base_service import CxOneFlowAbstractWorkflowService
 from workflows.resolver_workflow import (
     DummyResolverScanningWorkflow,
     ResolverScanningWorkflow,
@@ -29,6 +30,8 @@
 from cxone_api import CxOneClient
 from kickoff_services import DummyKickoffService, KickoffService
 from naming_services import ProjectNamingService
+from cxone_sarif import get_sarif_v210_log_for_scan
+from cxone_sarif.opts import DEFAULT as SARIF_DEFAULT_OPTS, ReportOpts
 
 
 class CxOneFlowConfig(CommonConfig):
@@ -229,7 +232,7 @@ def __resolver_service_factory(
 
     @staticmethod
     def __polling_service_factory(
-        config_path, services : List[BaseWorkflowService], **kwargs
+        config_path, services : List[CxOneFlowAbstractWorkflowService], **kwargs
     ) -> ScanPollingService:
 
 
@@ -330,16 +333,26 @@ def __pr_feedback_service_factory(
             )
 
 
+    @staticmethod
+    def __sarif_ReportOpts_factory(config_path : str, opts : dict) -> ReportOpts:
+        if opts is None:
+            return SARIF_DEFAULT_OPTS
+        @dataclasses_json.dataclass_json
+        class JsonSarifReportOpts(ReportOpts):
+            @classmethod
+            def from_dict(clazz, json : dict):
+                return make_dataclass(clazz.__name__, json)
+        
+        config = asdict(SARIF_DEFAULT_OPTS)
+        config.update(opts)
+        return JsonSarifReportOpts.from_dict(config)
+
+    
 
     @staticmethod
     def __push_feedback_service_factory(
         config_path, moniker, **kwargs
     ) -> PushFeedbackService:
-
-        # TODO: Need to get the defaults for timeouts.
-        # scan-timeout-hours
-        # poll-interval-seconds
-
         if kwargs is None or len(kwargs.keys()) == 0:
             return PushFeedbackService(
                 moniker, CxOneFlowConfig.__server_base_url, PushWorkflow(),
@@ -349,8 +362,42 @@ def __push_feedback_service_factory(
                 True,
             )
         else:
+            push_config_dict = CxOneFlowConfig._get_value_for_key_or_default("push", kwargs, {})
+            scan_monitor_dict = CxOneFlowConfig._get_value_for_key_or_default(
+                "scan-monitor", kwargs, {}
+            )
+
+            sarif_opts_dict = CxOneFlowConfig._get_value_for_key_or_default("sarif-opts", push_config_dict, None)
+
+            amqp_delivery_dict = CxOneFlowConfig._get_value_for_key_or_default("via-amqp", push_config_dict, None)
+
+            delivery_agents = []
+
+            if amqp_delivery_dict is not None and not 'amqp' in amqp_delivery_dict.keys():
+                raise ConfigurationException.missing_key_path(f"{config_path}/via-amqp/amqp")
+            elif amqp_delivery_dict is not None:
+                delivery_agents.append(PushFeedbackService.amqp_agent_factory
+                                       (moniker, 
+                                        CxOneFlowConfig._get_secret_from_value_of_key_or_fail
+                                            (f"{config_path}/via-amqp", "shared-secret", amqp_delivery_dict),
+                                        *CxOneFlowConfig._load_amqp_settings(config_path, **amqp_delivery_dict)))
+                
             return PushFeedbackService(
-                moniker, PushWorkflow(),
+                moniker, delivery_agents, CxOneFlowConfig.__sarif_ReportOpts_factory(f"{config_path}/push", sarif_opts_dict), 
+                PushWorkflow(
+                CxOneFlowConfig._get_value_for_key_or_default(
+                    "enabled", push_config_dict, False
+                ),
+                int(
+                    CxOneFlowConfig._get_value_for_key_or_default(
+                        "poll-interval-seconds", scan_monitor_dict, CxOneFlowConfig.DEFAULT_POLL_INTERVAL_SECS
+                    )
+                ),
+                int(
+                    CxOneFlowConfig._get_value_for_key_or_default(
+                        "scan-timeout-hours", scan_monitor_dict, CxOneFlowConfig.DEFAULT_SCAN_TIMEOUT_HOURS
+                    )
+                )),
                 *CxOneFlowConfig._load_amqp_settings(config_path, **kwargs)
             )
 

cxone_service/__init__.py

@@ -50,12 +50,16 @@ def __init__(self, moniker : str, cxone_client : CxOneClient, default_engines :
         self.__group_service = grouping_service
 
     @property
-    def moniker(self):
+    def moniker(self) -> str:
         return self.__moniker
 
     @property
-    def display_link(self):
+    def display_link(self) -> str:
         return self.__client.display_endpoint
+
+    @property
+    def client(self) -> CxOneClient:
+        return self.__client
 
     @staticmethod
     def __get_json_or_fail(response):

rabbit_config.py

@@ -8,7 +8,7 @@
 from workflows.push_feedback_service import PushFeedbackService
 from workflows.resolver_scan_service import ResolverScanService
 from workflows.scan_polling_service import ScanPollingService
-from workflows.base_service import BaseWorkflowService
+from workflows.base_service import CxOneFlowAbstractWorkflowService
 
 
 
@@ -41,7 +41,7 @@ async def setup() -> None:
                 durable=True,
             )
             scan_in_exchange = await channel.declare_exchange(
-                BaseWorkflowService.EXCHANGE_SCAN_INPUT,
+                CxOneFlowAbstractWorkflowService.EXCHANGE_SCAN_INPUT,
                 aio_pika.ExchangeType.FANOUT,
                 durable=True,
             )
@@ -53,7 +53,7 @@ async def setup() -> None:
                 internal=True,
             )
             scan_await_exchange = await channel.declare_exchange(
-                BaseWorkflowService.EXCHANGE_SCAN_WAIT,
+                CxOneFlowAbstractWorkflowService.EXCHANGE_SCAN_WAIT,
                 aio_pika.ExchangeType.TOPIC,
                 durable=True,
                 internal=True,
@@ -101,7 +101,7 @@ async def setup() -> None:
                 internal=True,
             )
             polling_delivery_exchange = await channel.declare_exchange(
-                BaseWorkflowService.EXCHANGE_SCAN_POLLING,
+                CxOneFlowAbstractWorkflowService.EXCHANGE_SCAN_POLLING,
                 aio_pika.ExchangeType.TOPIC,
                 durable=True,
                 internal=True,
@@ -125,7 +125,7 @@ async def setup() -> None:
                     "x-queue-type": "quorum",
                     "x-dead-letter-strategy": "at-least-once",
                     "x-overflow": "reject-publish",
-                    "x-dead-letter-exchange": BaseWorkflowService.EXCHANGE_SCAN_POLLING,
+                    "x-dead-letter-exchange": CxOneFlowAbstractWorkflowService.EXCHANGE_SCAN_POLLING,
                 },
             )
 

workflow_agent.py

@@ -4,6 +4,7 @@
 from config.server import CxOneFlowConfig
 from workflows.scan_polling_service import ScanPollingService
 from workflows.pr_feedback_service import PRFeedbackService
+from workflows.push_feedback_service import PushFeedbackService
 from workflows.resolver_scan_service import ResolverScanService
 from workflows.messaging import (
     ScanAwaitMessage,
@@ -18,6 +19,20 @@
 __log = logging.getLogger("WorkflowAgent")
 
 
+async def process_gen_sarif(msg: aio_pika.abc.AbstractIncomingMessage) -> None:
+    try:
+        __log.debug(
+            f"Received Sarif generation message on channel {msg.channel.number}: {msg.info()}"
+        )
+
+        sm = ScanFeedbackMessage.from_binary(msg.body)
+        services = CxOneFlowConfig.retrieve_services_by_moniker(sm.moniker)
+        await services.push.execute_sarif_generation(msg, services.cxone.client)
+    except BaseException as ex:
+        __log.exception(ex)
+        await msg.nack(requeue=False)
+
+
 async def process_poll(msg: aio_pika.abc.AbstractIncomingMessage) -> None:
     try:
         __log.debug(
@@ -66,6 +81,16 @@ async def spawn_agents():
     async with asyncio.TaskGroup() as g:
         for moniker in CxOneFlowConfig.get_service_monikers():
             services = CxOneFlowConfig.retrieve_services_by_moniker(moniker)
+
+            g.create_task(
+                mq_agent(
+                    process_gen_sarif,
+                    await services.pr.mq_client(),
+                    moniker,
+                    PushFeedbackService.QUEUE_SARIF_GEN,
+                )
+            )
+
             g.create_task(
                 mq_agent(
                     process_poll,
@@ -82,7 +107,6 @@ async def spawn_agents():
                     ScanPollingService.QUEUE_SCAN_POLLING,
                 )
             )
-
             g.create_task(
                 mq_agent(
                     process_pr_annotate,

workflows/base_service.py

@@ -1,20 +1,12 @@
 import urllib.parse, aio_pika, logging, asyncio, os
-from workflows import ScanStates
 from ssl import create_default_context, CERT_NONE
 from cxoneflow_logging import SecretRegistry
 from workflows.messaging.base_message import BaseMessage
 from workflows.messaging import ScanAwaitMessage
 from typing import Any
 
-class BaseWorkflowService:
-
-    ELEMENT_PREFIX = "cx:"
-    TOPIC_PREFIX = "cx."
-
-    EXCHANGE_SCAN_INPUT = f"{ELEMENT_PREFIX}Scan In"
-    EXCHANGE_SCAN_WAIT = f"{ELEMENT_PREFIX}Scan Await"
-    EXCHANGE_SCAN_POLLING = f"{ELEMENT_PREFIX}Scan Polling Delivery"
 
+class AMQPClient:
 
     def __init__(self, amqp_url : str, amqp_user : str, amqp_password : str, ssl_verify : bool):
         self.__lock = asyncio.Lock()
@@ -42,7 +34,7 @@ async def mq_client(self) -> aio_pika.abc.AbstractRobustConnection:
         async with self.__lock:
 
             if self.__client is None:
-                BaseWorkflowService.log().debug(f"Creating AMQP connection to: {self.__amqp_url}")
+                AMQPClient.log().debug(f"Creating AMQP connection to: {self.__amqp_url}")
                 ctx = None
 
                 if isinstance(self.__ssl_verify, bool):
@@ -63,17 +55,26 @@ async def mq_client(self) -> aio_pika.abc.AbstractRobustConnection:
                                                     ssl_context=ctx)
         return self.__client
 
+
+class CxOneFlowAbstractWorkflowService(AMQPClient):
+    ELEMENT_PREFIX = "cx:"
+    TOPIC_PREFIX = "cx."
+
+    EXCHANGE_SCAN_INPUT = f"{ELEMENT_PREFIX}Scan In"
+    EXCHANGE_SCAN_WAIT = f"{ELEMENT_PREFIX}Scan Await"
+    EXCHANGE_SCAN_POLLING = f"{ELEMENT_PREFIX}Scan Polling Delivery"
+
     async def _safe_deserialize_body(self, msg : aio_pika.abc.AbstractIncomingMessage, msg_class : BaseMessage) -> Any:
         try:
             ret_val = msg_class.from_binary(msg.body)
             return ret_val
         except BaseException as ex:
-            BaseWorkflowService.log().exception(ex)
+            CxOneFlowAbstractWorkflowService.log().exception(ex)
             await msg.nack(requeue=False)
             raise
 
     async def handle_completed_scan(self, msg : ScanAwaitMessage) -> None:
         raise NotImplementedError("handle_awaited_scan")
 
     async def handle_awaited_scan_error(self, msg : ScanAwaitMessage, error_msg : str) -> None:
-        raise NotImplementedError("handle_awaited_scan_error")    
+        raise NotImplementedError("handle_awaited_scan_error")    

workflows/pr_feedback_service.py

@@ -5,38 +5,38 @@
 from workflows.feedback_workflow_base import AbstractPRFeedbackWorkflow
 from workflows import ScanStates, ScanWorkflow, FeedbackWorkflow
 from workflows.pr import PullRequestAnnotation, PullRequestFeedback
-from workflows.base_service import BaseWorkflowService
+from workflows.base_service import CxOneFlowAbstractWorkflowService
 from cxone_service import CxOneException
 
-class PRFeedbackService(BaseWorkflowService):
+class PRFeedbackService(CxOneFlowAbstractWorkflowService):
     PR_ELEMENT_PREFIX = "pr:"
     PR_TOPIC_PREFIX = "pr."
 
-    EXCHANGE_SCAN_INPUT_LEGACY = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan In"
-    EXCHANGE_SCAN_WAIT_LEGACY = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Await"
-    EXCHANGE_SCAN_POLLING_LEGACY = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Polling Delivery"
+    EXCHANGE_SCAN_INPUT_LEGACY = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan In"
+    EXCHANGE_SCAN_WAIT_LEGACY = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Await"
+    EXCHANGE_SCAN_POLLING_LEGACY = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Polling Delivery"
 
 
-    EXCHANGE_SCAN_ANNOTATE = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Annotate"
-    EXCHANGE_SCAN_FEEDBACK = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Feedback"
+    EXCHANGE_SCAN_ANNOTATE = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Annotate"
+    EXCHANGE_SCAN_FEEDBACK = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Scan Feedback"
 
-    QUEUE_SCAN_POLLING_LEGACY = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Polling Scans"
-    QUEUE_SCAN_WAIT_LEGACY = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Awaited Scans"
+    QUEUE_SCAN_POLLING_LEGACY = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Polling Scans"
+    QUEUE_SCAN_WAIT_LEGACY = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}Awaited Scans"
 
 
-    QUEUE_ANNOTATE_PR = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}PR Annotating"
-    QUEUE_FEEDBACK_PR = f"{BaseWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}PR Feedback"
+    QUEUE_ANNOTATE_PR = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}PR Annotating"
+    QUEUE_FEEDBACK_PR = f"{CxOneFlowAbstractWorkflowService.ELEMENT_PREFIX}{PR_ELEMENT_PREFIX}PR Feedback"
 
-    ROUTEKEY_POLL_BINDING_LEGACY = f"{BaseWorkflowService.TOPIC_PREFIX}{PR_TOPIC_PREFIX}{ScanStates.AWAIT}.*.*"
+    ROUTEKEY_POLL_BINDING_LEGACY = f"{CxOneFlowAbstractWorkflowService.TOPIC_PREFIX}{PR_TOPIC_PREFIX}{ScanStates.AWAIT}.*.*"
 
 
-    ROUTEKEY_FEEDBACK_PR = f"{BaseWorkflowService.TOPIC_PREFIX}{PR_TOPIC_PREFIX}{ScanStates.FEEDBACK}.{FeedbackWorkflow.PR}.*"
-    ROUTEKEY_ANNOTATE_PR = f"{BaseWorkflowService.TOPIC_PREFIX}{PR_TOPIC_PREFIX}{ScanStates.ANNOTATE}.{FeedbackWorkflow.PR}.*"
+    ROUTEKEY_FEEDBACK_PR = f"{CxOneFlowAbstractWorkflowService.TOPIC_PREFIX}{PR_TOPIC_PREFIX}{ScanStates.FEEDBACK}.{FeedbackWorkflow.PR}.*"
+    ROUTEKEY_ANNOTATE_PR = f"{CxOneFlowAbstractWorkflowService.TOPIC_PREFIX}{PR_TOPIC_PREFIX}{ScanStates.ANNOTATE}.{FeedbackWorkflow.PR}.*"
 
 
     @staticmethod
     def make_topic(state : ScanStates, workflow : FeedbackWorkflow, moniker : str):
-        return f"{BaseWorkflowService.TOPIC_PREFIX}{PRFeedbackService.PR_TOPIC_PREFIX}{state}.{workflow}.{moniker}"
+        return f"{CxOneFlowAbstractWorkflowService.TOPIC_PREFIX}{PRFeedbackService.PR_TOPIC_PREFIX}{state}.{workflow}.{moniker}"
 
     @staticmethod
     def log():