Recorded Future Sandbox Analyzer

analyzers/RecordedFuture_Sandbox/README.md

@@ -0,0 +1,14 @@
+# Recorded Future Sandbox analyzer
+
+Recorded Future Sandbox service is based on Triage solution, which is a commercial malware sandbox that let's users run malware in a safe way.
+
+You can read more about the underlying solutions at: https://sandbox.recordedfuture.com/
+
+Thus, this analyzer requires you to have a commercial license.
+
+# FAQ
+
+### Q: This analyzer is very close than the Triage one developped by Mikael Keri. Isn't it a duplicate or a clone?
+
+#### This analyser has NOT been developped from scratch and involves the Triage analyser code in order to be lighlty adapted to fit Recorded Future solution.
+#### Actually, Recorded Future acquired the Triage technology in July 2022 and its integration requires a different URL base and a few ajustements in the Python script.

analyzers/RecordedFuture_Sandbox/RecordedFuture_Sandbox.json

@@ -0,0 +1,57 @@
+{
+  "name": "RecordedFuture_Sandbox_Triage",
+  "author": "THA-CERT feat. Mikael Keri",
+  "license": "AGPL-V3",
+  "url": "https://github.com/TheHive-Project/Cortex-Analyzers",
+  "version": "1.0",
+  "description": "Submit artifacts to the Recorded Future sandbox service; which is Triage based. This analyzer requires a paid subscription",
+  "dataTypeList": ["ip", "url", "file"],
+  "baseConfig": "RecordedFuture_Sandbox_Triage",
+  "config": {
+      "check_tlp": true,
+      "max_tlp": 2,
+      "check_pap": true,
+      "max_pap": 2
+   },
+  "command": "RecordedFuture_Sandbox/sandbox_analyzer.py",
+  "configurationItems": [
+    {
+      "name": "api_key",
+      "description": "API key",
+      "type": "string",
+      "multi": false,
+      "required": true
+    },
+    {
+      "name": "timeout",
+      "description": "Sandbox run timeout in seconds (default: 200)",
+      "type": "string",
+      "multi": false,
+      "required": false
+    },
+    {
+      "name": "zip_pw",
+      "description": "Zip archive password",
+      "type": "string",
+      "multi": false,
+      "required": false
+    }
+   ],
+   "registration_required": true,
+   "subscription_required": true,
+   "free_subscription": false,
+   "service_homepage": "https://sandbox.recordedfuture.com/",
+   "service_logo": {"path":"assets/triage_logo.png", "caption": "logo"},
+   "screenshots": [
+     {"path":"assets/triage_cortex_settings.png",
+       "caption":"Triage analyzer cortex setting"
+     },
+     {
+       "path": "assets/triage_long_report.png",
+       "caption:":"Triage analyzer full report"
+     },
+     {
+      "path": "assets/triage_verdict.png",
+      "caption:":"Triage analyzer verdict"
+    }]
+}

analyzers/RecordedFuture_Sandbox/requirements.txt

@@ -0,0 +1,2 @@
+cortexutils
+hatching-triage

analyzers/RecordedFuture_Sandbox/sandbox_analyzer.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+# encoding: utf-8
+
+import requests
+import time
+import io
+import zipfile
+from os.path import basename
+from triage import Client
+from cortexutils.analyzer import Analyzer
+
+
+class TriageAnalyzer(Analyzer):
+    def __init__(self):
+        Analyzer.__init__(self)
+        self.app_secret = self.get_param('config.api_key', None, 'API key is missing')
+        self.zip_pw = self.get_param('config.zip_pw')
+        self.timeout = self.get_param('config.timeout')
+
+        if self.timeout:
+        # sleep some more, we all need to relax more
+            self.timeout = int(self.timeout) + 60
+        else:
+            self.timeout = 200
+
+        self.url = 'https://sandbox.recordedfuture.com/api/'
+
+    def summary(self, raw):
+         taxonomies = []
+         namespace = "RecordedFuture_Sandbox"
+
+         value = "{}/10".format(raw['result']['sample'].get('score'))
+
+         if raw['result']['sample'].get('score') == 0:
+            verdict = "safe"
+         elif raw['result']['sample'].get('score') < 5:
+            verdict = "suspicious"
+         else:
+            verdict = "malicious"
+
+         taxonomies.append(self.build_taxonomy(
+         verdict,
+         namespace,
+         'Score',
+         value
+         ))
+
+         return {"taxonomies": taxonomies}
+
+    def file_submit(self, filename, filepath):
+
+        token = self.app_secret
+        connect = Client(token, root_url=self.url)
+
+        # Check if it's a zip file and if it's password protected
+        if zipfile.is_zipfile(filepath):
+           zf = zipfile.ZipFile(filepath)
+           for zinfo in zf.infolist():
+               is_encrypted = zinfo.flag_bits & 0x1
+               if is_encrypted:
+                   password = self.zip_pw
+                   sample = open(filepath, "rb")
+                   submit = connect.submit_sample_file(filename, sample, password=password)
+               else:
+                   sample = open(filepath, "rb")
+                   submit = connect.submit_sample_file(filename, sample)
+        else:
+           # Submit
+           sample = open(filepath, "rb")
+           submit = connect.submit_sample_file(filename, sample)
+
+        # Wait
+        time.sleep(self.timeout)
+        # Enjoy
+        retrive = connect.overview_report(submit['id'])
+        return retrive
+
+    def url_submit(self, data):
+
+        # Submit
+        token = self.app_secret
+        connect = Client(token, root_url=self.url)
+        submit = connect.submit_sample_url(data)
+        # Wait
+        time.sleep(self.timeout)
+        # Enjoy
+        retrive = connect.overview_report(submit['id'])
+        return retrive
+
+    def run(self):
+
+        if self.data_type == 'ip' or self.data_type == 'url':
+            data = self.get_param('data', None, 'Data is missing')
+
+            if ':' in data:
+                result = self.url_submit(data)
+                self.report({'result': result})
+            else:
+                self.error('Schema is missing')
+
+        elif self.data_type == 'file':
+            filepath = self.get_param('file', None, 'File is missing')
+            filename = self.get_param('filename', basename(filepath))
+
+            result = self.file_submit(filename, filepath)
+
+            self.report({'result': result})
+
+        else:
+            data = self.get_param('data', None, 'Data is missing')
+
+if __name__ == '__main__':
+    TriageAnalyzer().run()