Merge branch 'master' of https://github.com/Sentinel-One/CobaltStrikeParser

.github/workflows/main.yml

@@ -0,0 +1,34 @@
+# This is a basic workflow to help you get started with Actions
+
+name: CI
+
+# Controls when the action will run. 
+on:
+  # Triggers the workflow on push or pull request events but only for the master branch
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+  # Allows you to run this workflow manually from the Actions tab
+  workflow_dispatch:
+
+# A workflow run is made up of one or more jobs that can run sequentially or in parallel
+jobs:
+  # This workflow contains a single job called "build"
+  build:
+    # The type of runner that the job will run on
+    runs-on: ubuntu-latest
+
+    # Steps represent a sequence of tasks that will be executed as part of the job
+    steps:
+      # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+        with:
+          python-version: '3.8'
+
+      - name: Install dependencies and run test
+        run: |
+          pip install -r requirements.txt
+          python -m unittest

beacon_utils.py

@@ -55,7 +55,10 @@ def get_beacon_data(url, arch):
     eicar_offset = buf.find(b'EICAR-STANDARD-ANTIVIRUS-TEST-FILE')
     if eicar_offset != -1:
         return buf
+    return decrypt_beacon(buf)
 
+
+def decrypt_beacon(buf):
     offset = buf.find(b'\xff\xff\xff')
     if offset == -1:
         _cli_print('[-] Unexpected buffer received')
@@ -74,5 +77,4 @@ def get_beacon_data(url, arch):
             b = struct.unpack_from('<I', buf, i*4+4)[0]
             с = a ^ b
             decoded_data += struct.pack('<I', с)
-
         return decoded_data

parse_beacon_config.py

@@ -472,6 +472,10 @@ class cobaltstrikeConfig:
         return None
 
 
+    def parse_encrypted_config_non_pe(self, version=None, quiet=False, as_json=False):
+        self.data = decrypt_beacon(self.data)
+        return self.parse_config(version=version, quiet=quiet, as_json=as_json)
+
     def parse_encrypted_config(self, version=None, quiet=False, as_json=False):
         '''
         Parses beacon's configuration from stager dll or memory dump
@@ -481,7 +485,11 @@ class cobaltstrikeConfig:
         :bool as_json: Whether to dump as json
         '''
 
-        pe = pefile.PE(data=self.data)
+        try:
+            pe = pefile.PE(data=self.data)
+        except pefile.PEFormatError:
+            return self.parse_encrypted_config_non_pe(version=version, quiet=quiet, as_json=as_json)
+
         data_sections = [s for s in pe.sections if s.Name.find(b'.data') != -1]
         if not data_sections:
             _cli_print("Failed to find .data section")
@@ -541,4 +549,4 @@ if __name__ == '__main__':
             exit(0)
 
     print("[-] Failed to find any beacon configuration")
-    exit(1)
+    exit(1)

test_parse_beacon_config.py

@@ -0,0 +1,65 @@
+#! /usr/bin/python3
+
+import io
+import os
+import unittest
+
+from parse_beacon_config import cobaltstrikeConfig
+
+from zipfile import ZipFile
+
+
+def decrypt_sample(zip_path):
+    with ZipFile(zip_path) as z:
+        for fn in z.namelist():
+            return io.BytesIO(z.read(fn, pwd=bytes("infected", "ascii")))
+
+
+class TestBeaconParsing(unittest.TestCase):
+    def test_non_pe_x86(self):
+        path = os.path.join(
+            os.path.dirname(__file__),
+            "samples",
+            "13e954be0b0c022c392c956e9a800201a75dab7e288230b835bcdd4a9d68253d.zip",
+        )
+        f = decrypt_sample(path)
+        parser = cobaltstrikeConfig(f)
+        conf = parser.parse_encrypted_config()
+        self.assertEqual(conf.get("HttpPostUri"), "/submit.php")
+
+    def test_encrypted_x86_64(self):
+        path = os.path.join(
+            os.path.dirname(__file__),
+            "samples",
+            "10fd211ba97ddf12aecb1e7931d92c3ba37421c362cb1490e0203c1bd88ec141.zip",
+        )
+        f = decrypt_sample(path)
+        parser = cobaltstrikeConfig(f)
+        conf = parser.parse_encrypted_config()
+        self.assertEqual(conf.get("PublicKey_MD5"), "d2c8ec15d925e2514714d619022f7cdf")
+
+    def test_encrypted_x86(self):
+        path = os.path.join(
+            os.path.dirname(__file__),
+            "samples",
+            "7773169ca4ea81203a550dfebe53f091a8c57a3a5b12386e51c5a05194fef3ff.zip",
+        )
+        f = decrypt_sample(path)
+        parser = cobaltstrikeConfig(f)
+        conf = parser.parse_encrypted_config()
+        self.assertEqual(conf.get("PublicKey_MD5"), "8ac540617dddcdf575f6dc207abb7344")
+
+    def test_trial_beacon_x86(self):
+        path = os.path.join(
+            os.path.dirname(__file__),
+            "samples",
+            "4d1d732125e4d1a3ba0571e0cd892cf8e0dce854387ee405f75df4dcfb0f616b.zip",
+        )
+        f = decrypt_sample(path)
+        parser = cobaltstrikeConfig(f)
+        conf = parser.parse_config()
+        self.assertIn('header "CGGGGG"', conf.get("HttpGet_Metadata").get("Metadata"))
+
+
+if __name__ == "__main__":
+    unittest.main()