xsoar: initial commit

README.md

@@ -49,7 +49,7 @@ Github issues and feature requests welcomed.
 | Logging         | sentryio                                                    |
 | MacOS           | airport<br/>macchanger<br/>wdutil                           |
 | Python          | logging<br/>requests                                        |
-| SOAR            | swimlane<br/>splunk soar                                    |
+| SOAR            | swimlane<br/>splunk soar<br/>xsoar                          |
 | Recon           | nmap                                                        |
 | Test Automation | selenium                                                    |
 

automon/integrations/xsoar/__init__.py

@@ -0,0 +1,2 @@
+from .client import XSOARClient
+from .config import XSOARConfig

automon/integrations/xsoar/client.py

@@ -0,0 +1,52 @@
+from automon.log import logging
+from automon.integrations.requestsWrapper import RequestsClient
+
+from .config import XSOARConfig
+from .endpoints import v1
+
+logger = logging.getLogger(__name__)
+logger.setLevel(level=logging.DEBUG)
+
+
+class XSOARClient(object):
+    """XSOAR REST API client
+
+    referenc: https://cortex-panw.stoplight.io/docs/cortex-xsoar-8/kjn2q21a7yrbm-get-started-with-cortex-xsoar-8-ap-is
+    """
+
+    def __init__(
+            self,
+            host: str = None,
+            api_key: str = None,
+            api_key_id: str = None,
+            config: XSOARConfig = None
+    ):
+        self.config = config or XSOARConfig(host=host, api_key=api_key, api_key_id=api_key_id)
+        self._requests = RequestsClient()
+
+    async def is_ready(self):
+        if self.config.is_ready():
+            return True
+        return False
+
+    async def auth(self):
+        return
+
+    async def get(self, endpoint: str):
+        logger.info(dict(
+            endpoint=f'{self.config.host}/{endpoint}'
+        ))
+        return await self._requests.get(url=f'{self.config.host}/{endpoint}', headers=self.config.headers)
+
+    async def post(self, endpoint: str):
+        logger.info(dict(
+            endpoint=f'{self.config.host}/{endpoint}'
+        ))
+        return self._requests.post(url=f'{self.config.host}/{endpoint}', headers=self.config.headers)
+
+    async def reports(self):
+        reports = await self.get(endpoint=v1.Reports.reports)
+        logger.info(dict(
+            reports=self._requests.content
+        ))
+        return reports

automon/integrations/xsoar/config.py

@@ -0,0 +1,41 @@
+from automon import environ
+from automon.log import logging
+
+logger = logging.getLogger(__name__)
+logger.setLevel(level=logging.DEBUG)
+
+
+class XSOARConfig(object):
+    """XSOAR REST API client config"""
+
+    def __init__(
+            self,
+            host: str = None,
+            api_key: str = None,
+            api_key_id: str = None
+    ):
+        self.host = host or environ('XSOAR_FQDN')
+        self.api_key = api_key or environ('XSOAR_API_KEY')
+        self.api_key_id = api_key_id or environ('XSOAR_API_KEY_ID')
+
+    def is_ready(self) -> bool:
+        if not self.host:
+            logger.error(f'missing XSOAR_FQDN')
+
+        if not self.api_key:
+            logger.error(f'missing XSOAR_API_KEY')
+
+        if not self.api_key_id:
+            logger.error(f'missing XSOAR_API_KEY_ID')
+
+        if self.host and self.api_key and self.api_key_id:
+            return True
+        return False
+
+    @property
+    def headers(self):
+        return {
+            'Authorization': f'{self.api_key}',
+            'x-xdr-auth-id': f'{self.api_key_id}',
+            "Content-Type": "application/json"
+        }

automon/integrations/xsoar/endpoints/v1.py

@@ -0,0 +1,9 @@
+class V1:
+    xsoar: str = 'xsoar'
+    public: str = f'{xsoar}/public'
+    v1: str = f'{public}/v1'
+
+
+class Reports:
+    """xsoar/public/v1/reports"""
+    reports: str = f'{V1.v1}/reports'

automon/integrations/xsoar/tests/test_client_auth.py

@@ -0,0 +1,17 @@
+import asyncio
+import unittest
+
+from automon.integrations.xsoar import XSOARClient
+
+
+class MyTestCase(unittest.TestCase):
+    test = XSOARClient()
+
+    if asyncio.run(test.is_ready()):
+        def test_auth(self):
+            result = asyncio.run(self.test.reports())
+            pass
+
+
+if __name__ == '__main__':
+    unittest.main()

automon/integrations/xsoar/tests/test_config.py

@@ -0,0 +1,15 @@
+import unittest
+
+from automon.integrations.xsoar import XSOARConfig
+
+
+class MyTestCase(unittest.TestCase):
+    test = XSOARConfig()
+
+    if test.is_ready():
+        def test_config(self):
+            self.assertTrue(self.test.is_ready())
+
+
+if __name__ == '__main__':
+    unittest.main()

env-example.sh

@@ -147,3 +147,8 @@ VDS_PASSWORD=
 
 # Wdutil
 WDUTIL_PASSWORD=
+
+# XSOAR
+XSOAR_FQDN=
+XSOAR_API_KEY=
+XSOAR_API_KEY_ID=