comm and improvements

README.md

@@ -1,26 +1,31 @@
 # CobaltStrikeParser
 Python parser for CobaltStrike Beacon's configuration
 
-## Background
-Use `parse_beacon_config.py` for stageless beacons or on memory dumps.
-
-Many stageless beacons are PEs where the beacon code itself is stored in the `.data` section and xored with 4-byte key.
+## Description
+Use `parse_beacon_config.py` for stageless beacons, memory dumps or C2 urls with metasploit compatibility mode (default true).  
+Many stageless beacons are PEs where the beacon code itself is stored in the `.data` section and xored with 4-byte key.  
 The script tries to find the xor key and data heuristically, decrypt the data and parse the configuration from it.
 
 This is designed so it can be used as a library too.
 
+The repo now also includes a small commuincation module (comm.py) that can help with communcating to a C2 server as a beacon.  
+
 ## Usage
 ```
-usage: parse_beacon_config.py [-h] [--json] [--quiet] [--version VERSION] path
+usage: parse_beacon_config.py [-h] [--json] [--quiet] [--version VERSION] beacon
 
-Parses CobaltStrike Beacon's configuration from PE or memory dump.
+Parses CobaltStrike Beacon's configuration from PE, memory dump or URL.
 
 positional arguments:
-  path               Beacon file path
+  beacon             This can be a file path or a url (if started with http/s)
 
 optional arguments:
   -h, --help         show this help message and exit
   --json             Print as json
   --quiet            Do not print missing or empty settings
   --version VERSION  Try as specific cobalt version (3 or 4). If not specified, tries both.
-```
+```
+
+## Extra
+To use the communication poc copy it to the main folder and run it from there.
+For installing the M2Crypto library (a requirement for the poc) on Windows, it's easiest with installers found online, and not through pip.

beacon_utils.py

@@ -0,0 +1,78 @@
+#!/usr/bin/python3
+'''
+By Gal Kristal from SentinelOne (gkristal.w@gmail.com) @gal_kristal
+Refs: 
+    https://github.com/RomanEmelyanov/CobaltStrikeForensic/blob/master/L8_get_beacon.py
+    https://github.com/nccgroup/pybeacon
+'''
+
+import requests, struct, urllib3
+import argparse
+from urllib.parse import urljoin
+import socket
+import json
+from base64 import b64encode
+from struct import unpack, unpack_from
+
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+EMPTY_UA_HEADERS = {"User-Agent":""}
+URL_PATHS = {'x86':'ab2g', 'x64':'ab2h'}
+
+class Base64Encoder(json.JSONEncoder):
+    def default(self, o):
+        if isinstance(o, bytes):
+            return b64encode(o).decode()
+        return json.JSONEncoder.default(self, o)
+
+
+def _cli_print(msg, end='\n'):
+    if __name__ == '__main__':
+        print(msg, end=end)
+
+
+def read_dword_be(fh):
+    data = fh.read(4)
+    if not data or len(data) != 4:
+        return None
+    return unpack(">I",data)[0]
+
+
+def get_beacon_data(url, arch):
+    full_url = urljoin(url, URL_PATHS[arch])
+    try:
+        resp = requests.get(full_url, timeout=30, headers=EMPTY_UA_HEADERS, verify=False)
+    except requests.exceptions.RequestException as e:
+        _cli_print('[-] Connection error: ', e)
+        return
+
+    if resp.status_code != 200:
+        _cli_print('[-] Failed with HTTP status code: ', resp.status_code)
+        return
+
+    buf = resp.content
+
+    # Check if it's a Trial beacon, therefore not xor encoded (not tested)
+    eicar_offset = buf.find(b'EICAR-STANDARD-ANTIVIRUS-TEST-FILE')
+    if eicar_offset != -1:
+        return buf
+
+    offset = buf.find(b'\xff\xff\xff')
+    if offset == -1:
+        _cli_print('[-] Unexpected buffer received')
+        return
+    offset += 3
+    key = struct.unpack_from('<I', buf, offset)[0]
+    size = struct.unpack_from('<I', buf, offset+4)[0] ^ key
+    head_enc = struct.unpack_from('<I', buf, offset+8)[0] ^ key
+    head = head_enc & 0xffff
+
+    # Taken directly from L8_get_beacon.py
+    if head == 0x5a4d or head ==0x9090:
+        decoded_data = b''
+        for i in range(2+offset//4, len(buf)//4-4):
+            a = struct.unpack_from('<I', buf, i*4)[0]
+            b = struct.unpack_from('<I', buf, i*4+4)[0]
+            с = a ^ b
+            decoded_data += struct.pack('<I', с)
+
+        return decoded_data

comm.py

@@ -0,0 +1,253 @@
+#!/usr/bin/python3
+'''
+By Gal Kristal from SentinelOne (gkristal.w@gmail.com) @gal_kristal
+These were awesome reference material:
+    https://research.nccgroup.com/2020/06/15/striking-back-at-retired-cobalt-strike-a-look-at-a-legacy-vulnerability/
+    https://github.com/nccgroup/pybeacon
+'''
+
+import struct
+import base64
+import hashlib
+import random
+import os
+import string
+import M2Crypto
+import requests
+
+PUBLIC_KEY_TEMPLATE = "-----BEGIN PUBLIC KEY-----\n{}\n-----END PUBLIC KEY-----"
+
+class Metadata(object):
+    """
+    Class to represent a beacon's metadata.
+    This is specific to Cobalt 4 and up
+    """
+    def __init__(self, public_key, aes_source_bytes):
+        """
+        Generates a random beacon entry
+        Args:
+            public_key (bytes): The extracted public key from beacon configuration
+            aes_source_bytes (bytes): 16 bytes used to generate AES keys from
+        """
+        self.public_key = public_key
+        self.port = random.randint(40000,50000)
+        self.ciphertext = ""
+        self.charset = 20273
+        self.ver = random.randint(1,10)
+        self.ip = os.urandom(4)
+        self.comp = ''.join(random.choices(string.ascii_uppercase + string.digits, k=7))
+        self.user = ''.join(random.choices(string.ascii_uppercase + string.digits, k=7))
+        self.pid = random.randint(1,50000) * 4 - 2  # ;)
+        self.bid = random.randint(1,1000000) * 2
+        self.barch = 1
+        self.is64 = False
+        self.high_integrity = False
+        self.aes_source_bytes = aes_source_bytes
+        self.junk = os.urandom(14)
+        d = hashlib.sha256(aes_source_bytes).digest()        
+        self.aes_key = d[0:16]
+        self.hmac_key = d[16:]
+
+
+    def rsa_encrypt(self, data):
+        """Encrypt given data the way Cobalt's server likes
+        
+        Args:
+            data (bytes): Data to encrypt
+        
+        Returns:
+            bytes: 
+        """
+        bio = M2Crypto.BIO.MemoryBuffer(PUBLIC_KEY_TEMPLATE.format(base64.b64encode(self.public_key).decode()).encode())
+        pubkey = M2Crypto.RSA.load_pub_key_bio(bio)
+        # Format is: magic + dataLength + data
+        # beef is the magic used by the server
+        packed_data = b'\x00\x00\xBE\xEF' + struct.pack('>I', len(data)) + data
+        return pubkey.public_encrypt(packed_data, M2Crypto.RSA.pkcs1_padding)
+
+
+    def pack(self):
+        data = self.aes_source_bytes + struct.pack('>hhIIHBH', self.charset, self.charset, self.bid, self.pid, self.port, self.is64, self.ver) + self.junk
+        data += struct.pack('4s', self.ip)
+        data += b'\x00' * (51 - len(data))
+        data += '\t'.join([self.comp, self.user]).encode()
+        return self.rsa_encrypt(data)
+
+
+TERMINATION_STEPS = ['header', 'parameter', 'print']
+TSTEPS = {1: "append", 2: "prepend", 3: "base64", 4: "print", 5: "parameter", 6: "header", 7: "build", 8: "netbios", 9: "const_parameter", 10: "const_header", 11: "netbiosu", 12: "uri_append", 13: "base64url", 14: "strrep", 15: "mask", 16: "const_host_header"}
+
+# Could probably just be b'\x00'*4 + data
+def mask(arg, data):
+    key = os.urandom(4)
+    data = data.encode('latin-1')
+    return key.decode('latin-1') + ''.join(chr(c ^ key[i%4]) for i, c in enumerate(data))
+    
+def demask(arg, data):
+    key = data[:4].encode('latin-1')
+    data = data.encode('latin-1')
+    return ''.join(chr(c ^ key[i%4]) for i, c in enumerate(data[4:]))
+
+def netbios_decode(name, case):
+    i = iter(name.upper())
+    try:
+        return ''.join([chr(((ord(c)-ord(case))<<4)+((ord(next(i))-ord(case))&0xF)) for c in i])
+    except:
+        return ''
+
+func_dict_encode = {"append": lambda arg, data: data + arg,
+"prepend": lambda arg, data: arg + data,
+"base64": lambda arg, data: base64.b64encode(data),
+"netbios": lambda arg, data: ''.join([chr((ord(c)>>4) + ord('a')) + chr((ord(c)&0xF) + ord('a')) for c in data]),
+"netbiosu": lambda arg, data: ''.join([chr((ord(c)>>4) + ord('A')) + chr((ord(c)&0xF) + ord('A')) for c in data]),
+"base64": lambda arg, data: base64.b64encode(data.encode('latin-1')).decode('latin-1'),
+"base64url": lambda arg, data: base64.urlsafe_b64encode(data.encode('latin-1')).decode('latin-1').strip('='),
+"mask": mask,
+}
+
+func_dict_decode = {"append": lambda arg, data: data[:-len(arg)],
+"prepend": lambda arg, data: data[len(arg):],
+"base64": lambda arg, data: base64.b64decode(data),
+"netbios": lambda arg, data: netbios_decode(data, 'a'),
+"netbiosu": lambda arg, data: netbios_decode(data, 'A'),
+"base64": lambda arg, data: base64.b64decode(data.encode('latin-1')).decode('latin-1'),
+"base64url": lambda arg, data: base64.urlsafe_b64decode(data.encode('latin-1')).decode('latin-1').strip('='),
+"mask": demask,
+}
+
+
+class Transform(object):
+    def __init__(self, trans_dict):
+        """An helper class to tranform data according to cobalt's malleable profile
+        
+        Args:
+            trans_dict (dict): A dictionary that came from packedSetting data. It's in the form of:
+                                {'ConstHeaders':[], 'ConstParams': [], 'Metadata': [], 'SessionId': [], 'Output': []}
+        """
+        self.trans_dict = trans_dict
+
+    def encode(self, metadata, output, sessionId):
+        """
+        
+        Args:
+            metadata (str): The metadata of a Beacon, usually given from Metadata.pack()
+            output (str): If this is for a Beacon's response, then this is the response's data
+            sessionId (str): the Beacon's ID
+        
+        Returns:
+            (str, dict, dict): This is to be used in an HTTP request. The tuple is (request_body, request_headers, request_params)
+        """
+        params = {}
+        headers = {}
+        body = ''
+        for step in self.trans_dict['Metadata']:
+            action = step.split(' ')[0].lower()
+            arg = step.lstrip(action).strip().strip('"')
+            if action in TERMINATION_STEPS:
+                if action == "header":
+                    headers[arg] = metadata
+                elif action == "parameter":
+                    params[arg] = metadata
+                elif action == "print":
+                    body = metadata
+            else:
+                metadata = func_dict_encode[action](arg, metadata)
+
+        for step in self.trans_dict['Output']:
+            action = step.split(' ')[0].lower()
+            arg = step.lstrip(action).strip().strip('"')
+            if action in TERMINATION_STEPS:
+                if action == "header":
+                    headers[arg] = output
+                elif action == "parameter":
+                    params[arg] = output
+                elif action == "print":
+                    body = output
+            else:
+                output = func_dict_encode[action](arg, output)
+
+        for step in self.trans_dict['SessionId']:
+            action = step.split(' ')[0].lower()
+            arg = step.lstrip(action).strip().strip('"')
+            if action in TERMINATION_STEPS:
+                if action == "header":
+                    headers[arg] = sessionId
+                elif action == "parameter":
+                    params[arg] = sessionId
+                elif action == "print":
+                    body = sessionId
+            else:
+                sessionId = func_dict_encode[action](arg, sessionId)
+
+        for step in self.trans_dict['ConstHeaders']:
+            offset = step.find(': ')
+            header, value = step[:offset], step[offset+2:]
+            headers[header] = value
+
+        for step in self.trans_dict['ConstParams']:
+            offset = step.find('=')
+            param, value = step[:offset], step[offset+1:]
+            params[param] = value
+
+        return body, headers, params
+
+    def decode(self, body, headers, params):
+        """
+        Parses beacon's communication data from an HTTP request
+        Args:
+            body (str): The body of an HTTP request
+            headers (dict): Headers dict from the HTTP request
+            params (dict): Params dict from the HTTP request
+        
+        Returns:
+            (str, str, str): The tuple is (metadata, output, sessionId)
+        """
+        metadata = ''
+        output = ''
+        sessionId = ''
+        for step in self.trans_dict['Metadata'][::-1]:
+            action = step.split(' ')[0].lower()
+            arg = step.lstrip(action).strip().strip('"')
+            if action in TERMINATION_STEPS:
+                if action == "header":
+                    metadata = headers[arg]
+                elif action == "parameter":
+                    metadata = params[arg]
+                elif action == "print":
+                    metadata = body
+            else:
+                metadata = func_dict_decode[action](arg, metadata)
+
+        for step in self.trans_dict['Output'][::-1]:
+            action = step.split(' ')[0].lower()
+            arg = step.lstrip(action).strip().strip('"')
+            if action in TERMINATION_STEPS:
+                if action == "header":
+                    output = headers[arg]
+                elif action == "parameter":
+                    output = params[arg]
+                elif action == "print":
+                    output = body
+            else:
+                output = func_dict_decode[action](arg, output)
+
+        for step in self.trans_dict['SessionId'][::-1]:
+            action = step.split(' ')[0].lower()
+            arg = step.lstrip(action).strip().strip('"')
+            if action in TERMINATION_STEPS:
+                if action == "header":
+                    sessionId = headers[arg]
+                elif action == "parameter":
+                    sessionId = params[arg]
+                elif action == "print":
+                    sessionId = body
+            else:
+                sessionId = func_dict_decode[action](arg, sessionId)
+
+        return metadata, output, sessionId
+
+
+
+
+
+        

extra/communication_poc.py

@@ -0,0 +1,131 @@
+#!/usr/bin/python3
+'''
+By Gal Kristal from SentinelOne (gkristal.w@gmail.com) @gal_kristal
+Refs: 
+    https://github.com/RomanEmelyanov/CobaltStrikeForensic/blob/master/L8_get_beacon.py
+    https://github.com/nccgroup/pybeacon
+'''
+
+import requests, struct, sys, os, urllib3
+import argparse
+from parse_beacon_config import cobaltstrikeConfig
+from urllib.parse import urljoin
+from io import BytesIO
+from Crypto.Cipher import AES
+import hmac
+import urllib
+import socket
+from comm import *
+
+HASH_ALGO = hashlib.sha256
+SIG_SIZE = HASH_ALGO().digest_size
+CS_FIXED_IV = b"abcdefghijklmnop"
+
+EMPTY_UA_HEADERS = {"User-Agent":""}
+URL_PATHS = {'x86':'ab2g', 'x64':'ab2h'}
+
+def get_beacon_data(url, arch):
+    full_url = urljoin(url, URL_PATHS[arch])
+    try:
+        resp = requests.get(full_url, timeout=30, headers=EMPTY_UA_HEADERS, verify=False)
+    except requests.exceptions.RequestException as e:
+        print('[-] Connection error: ', e)
+        return
+
+    if resp.status_code != 200:
+        print('[-] Failed with HTTP status code: ', resp.status_code)
+        return
+
+    buf = resp.content
+
+    # Check if it's a Trial beacon, therefore not xor encoded (not tested)
+    eicar_offset = buf.find(b'EICAR-STANDARD-ANTIVIRUS-TEST-FILE')
+    if eicar_offset != -1:
+        return cobaltstrikeConfig(BytesIO(buf)).parse_config()
+
+    offset = buf.find(b'\xff\xff\xff')
+    if offset == -1:
+        print('[-] Unexpected buffer received')
+        return
+    offset += 3
+    key = struct.unpack_from('<I', buf, offset)[0]
+    size = struct.unpack_from('<I', buf, offset+4)[0] ^ key
+    head_enc = struct.unpack_from('<I', buf, offset+8)[0] ^ key
+    head = head_enc & 0xffff
+
+    # Taken directly from L8_get_beacon.py
+    if head == 0x5a4d or head ==0x9090:
+        decoded_data = b''
+        for i in range(2+offset//4, len(buf)//4-4):
+            a = struct.unpack_from('<I', buf, i*4)[0]
+            b = struct.unpack_from('<I', buf, i*4+4)[0]
+            с = a ^ b
+            decoded_data += struct.pack('<I', с)
+
+        return cobaltstrikeConfig(BytesIO(decoded_data)).parse_config()
+
+
+def register_beacon(conf):
+    """Registers a random beacon and sends a task data.
+    This is a POC that shows how a beacon send its metadata and task results with Malleable profiles
+    
+    Args:
+        conf (dict): Beacon configuration dict, from cobaltstrikeConfig parser
+    """
+    # Register new random beacon
+    urljoin('http://'+conf['C2Server'].split(',')[0], conf['C2Server'].split(',')[1])
+    aes_source = os.urandom(16)
+    m = Metadata(conf['PublicKey'], aes_source)
+    t = Transform(conf['HttpGet_Metadata'])
+    body, headers, params = t.encode(m.pack().decode('latin-1'), '', str(m.bid))
+
+    print('[+] Registering new random beacon: comp=%s user=%s' % (m.comp, m.user))
+    try:
+        req = requests.request('GET', urljoin('http://'+conf['C2Server'].split(',')[0], conf['C2Server'].split(',')[1]), params=params, data=body, headers=dict(**headers, **{'User-Agent':''}), timeout=5)
+    except Exception as e:
+        print('[-] Got excpetion from server: %s' % e)
+        return
+
+    # This is how to properly encrypt a task:
+    # Tasks are encrypted with the session's aes key, decided and negotiated when we registered the beacon (it's part of the metadata)
+    ## Here is where you'll build a proper task struct ##
+    random_data = os.urandom(50)
+    # session counter = 1
+    data = struct.pack('>II', 1, len(random_data)) + random_data
+    pad_size = AES.block_size - len(data) % AES.block_size
+    data = data + pad_size * b'\x00'
+
+    # encrypt the task data and wrap with hmac sig and encrypted data length
+    cipher = AES.new(m.aes_key, AES.MODE_CBC, CS_FIXED_IV)
+    enc_data = cipher.encrypt(data)
+    sig = hmac.new(m.hmac_key, enc_data, HASH_ALGO).digest()[0:16]
+    enc_data += sig
+    enc_data = struct.pack('>I', len(enc_data)) + enc_data
+
+    # task data is POSTed so we need to take the transformation steps of http-post.client
+    t = Transform(conf['HttpPost_Metadata'])
+    body, headers, params = t.encode(m.pack().decode('latin-1'), enc_data.decode('latin-1'), str(m.bid))
+
+    print('[+] Sending task data')
+    
+    try:
+        req = requests.request('POST', urljoin('http://'+conf['C2Server'].split(',')[0], conf['HttpPostUri'].split(',')[0]), params=params, data=body, headers=dict(**headers, **{'User-Agent':''}), timeout=5)
+    except Exception as e:
+        print('[-] Got excpetion from server while sending task: %s' % e)
+
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description="Parse CobaltStrike Beacon's configuration from C2 url and registers a beacon with it")
+    parser.add_argument("url", help="Cobalt C2 server (e.g. http://1.1.1.1)")
+    args = parser.parse_args()
+
+    x86_beacon_conf = get_beacon_data(args.url, 'x86')
+    x64_beacon_conf = get_beacon_data(args.url, 'x64')
+    if not x86_beacon_conf and not x64_beacon_conf:
+        print("[-] Failed finding any beacon configuration")
+        exit(1)
+
+    print("[+] Got beacon configuration successfully")
+    conf = x86_beacon_conf or x64_beacon_conf
+    register_beacon(conf)

parse_beacon_config.py

@@ -1,6 +1,7 @@
+#!/usr/bin/python3
 '''
 Parses CobaltStrike Beacon's configuration from PE file or memory dump.
-By Gal Kristal from SentinelOne (gkristal.w@gmail.com)
+By Gal Kristal from SentinelOne (gkristal.w@gmail.com) @gal_kristal
 
 Inspired by https://github.com/JPCERTCC/aa-tools/blob/master/cobaltstrikescan.py
 
@@ -9,35 +10,28 @@ TODO:
  2. Dynamic size parsing
 '''
 
-from struct import unpack
+from beacon_utils import *
+from struct import unpack, unpack_from
 from socket import inet_ntoa
 from collections import OrderedDict
 from netstruct import unpack as netunpack
-import json
-from base64 import b64encode
-from sys import argv
 import argparse
 import io
 import re
 import pefile
+import os
 import hashlib
+from io import BytesIO
 
 THRESHOLD = 1100
 COLUMN_WIDTH = 35
 SUPPORTED_VERSIONS = (3, 4)
+SILENT_CONFIGS = ['PublicKey', 'ProcInject_Stub', 'smbFrameHeader', 'tcpFrameHeader', 'SpawnTo']
 
-class Base64Encoder(json.JSONEncoder):
-    def default(self, o):
-        if isinstance(o, bytes):
-            return b64encode(o).decode()
-        return json.JSONEncoder.default(self, o)
-
-
-def cli_print(msg, end='\n'):
+def _cli_print(msg, end='\n'):
     if __name__ == '__main__':
         print(msg, end=end)
 
-
 class confConsts:
     MAX_SETTINGS = 64
     TYPE_NONE = 0
@@ -56,12 +50,6 @@ class confConsts:
     4: 0x2e
     }
 
-def read_dword_be(fh):
-    data = fh.read(4)
-    if not data or len(data) != 4:
-        return None
-    return unpack(">I",data)[0]
-
 class packedSetting:
 
     def __init__(self, pos, datatype, length=0, isBlob=False, isHeaders=False, isIpAddress=False, isBool=False, isDate=False, boolFalseValue=0, isProcInjectTransform=False, isMalleableStream=False, hashBlob=False, enum=None, mask=None):
@@ -78,6 +66,8 @@ class packedSetting:
         self.hashBlob = hashBlob
         self.enum = enum
         self.mask = mask
+        self.transform_get = None
+        self.transform_post = None
         if datatype == confConsts.TYPE_STR and length == 0:
             raise(Exception("if datatype is TYPE_STR then length must not be 0"))
 
@@ -98,8 +88,62 @@ class packedSetting:
         self_repr[4:6] = self.length.to_bytes(2, 'big')
         return self_repr
 
+    def parse_transformdata(self, data):
+        '''
+        Args:
+            data (bytes): Raw communication transforam data
+        
+        Returns:
+            dict: Dict of transform commands that should be convenient for communication forging
+        
+        '''
+        dio = io.BytesIO(data)
+        trans = {'ConstHeaders':[], 'ConstParams': [], 'Metadata': [], 'SessionId': [], 'Output': []}
+        current_category = 'Constants'
+
+        # TODO: replace all magic numbers here with enum
+        while True:
+            tstep = read_dword_be(dio)
+            if tstep == 7:
+                name = read_dword_be(dio)
+                if self.pos == 12: # GET
+                    current_category = 'Metadata'
+                else: # POST
+                    current_category = 'SessionId' if name == 0 else 'Output'
+            elif tstep in (1, 2, 5, 6):
+                length = read_dword_be(dio)
+                step_data = dio.read(length).decode()
+                trans[current_category].append(BeaconSettings.TSTEPS[tstep] + ' "' + step_data + '"')
+            elif tstep in (10, 16, 9):
+                length = read_dword_be(dio)
+                step_data = dio.read(length).decode()
+                if tstep == 9:
+                    trans['ConstParams'].append(step_data)
+                else:
+                    trans['ConstHeaders'].append(step_data)
+            elif tstep in (3, 4, 13, 8, 11, 12, 15):
+                trans[current_category].append(BeaconSettings.TSTEPS[tstep])
+            else:
+                break
+
+        if self.pos == 12:
+            self.transform_get = trans
+        else:
+            self.transform_post = trans
+
+        return trans
+
+
     def pretty_repr(self, full_config_data):
         data_offset = full_config_data.find(self.binary_repr())
+        if data_offset < 0 and self.datatype == confConsts.TYPE_STR:
+            self.length = 16
+            while self.length < 2048:
+                data_offset = full_config_data.find(self.binary_repr())
+                if data_offset > 0:
+                    break
+                self.length *= 2
+
         if data_offset < 0:
             return 'Not Found'
 
@@ -128,7 +172,7 @@ class packedSetting:
                 return inet_ntoa(conf_data)
                 
             else:
-                conf_data = unpack('>I', conf_data)[0]
+                conf_data = unpack('>i', conf_data)[0]
                 if self.is_date and conf_data != 0:
                     fulldate = str(conf_data)
                     return "%s-%s-%s" % (fulldate[0:4], fulldate[4:6], fulldate[6:])
@@ -203,9 +247,7 @@ class packedSetting:
             return conf_data
 
         if self.is_headers:
-            conf_data = conf_data.strip(b'\x00')
-            conf_data = [chunk[1:].decode() for chunk in conf_data.split(b'\x00') if len(chunk) > 1]
-            return conf_data
+            return self.parse_transformdata(conf_data)
 
         conf_data = conf_data.strip(b'\x00').decode()
         return conf_data
@@ -216,13 +258,13 @@ class BeaconSettings:
     BEACON_TYPE = {0x0: "HTTP", 0x1: "Hybrid HTTP DNS", 0x2: "SMB", 0x4: "TCP", 0x8: "HTTPS", 0x10: "Bind TCP"}
     ACCESS_TYPE = {0x1: "Use direct connection", 0x2: "Use IE settings", 0x4: "Use proxy server"}
     EXECUTE_TYPE = {0x1: "CreateThread", 0x2: "SetThreadContext", 0x3: "CreateRemoteThread", 0x4: "RtlCreateUserThread", 0x5: "NtQueueApcThread", 0x6: None, 0x7: None, 0x8: "NtQueueApcThread-s"}
-    #TRANSFORMSTEP = {1: "append", 2: "prepend", 3: "base64", 4: "print", 5: "parameter", 6: "header", 7: "build", 8: "netbios", 9: "_parameter", 10: "_header", 
-    #                11: "netbiosu", 12: "uri_append",  13: "base64_url", 14: "strrep", 15: "mask"}
     ALLOCATION_FUNCTIONS = {0: "VirtualAllocEx", 1: "NtMapViewOfSection"}
+    TSTEPS = {1: "append", 2: "prepend", 3: "base64", 4: "print", 5: "parameter", 6: "header", 7: "build", 8: "netbios", 9: "const_parameter", 10: "const_header", 11: "netbiosu", 12: "uri_append", 13: "base64url", 14: "strrep", 15: "mask", 16: "const_host_header"}
+    ROTATE_STRATEGY = ["round-robin", "random", "failover", "failover-5x", "failover-50x", "failover-100x", "failover-1m", "failover-5m", "failover-15m", "failover-30m", "failover-1h", "failover-3h", "failover-6h", "failover-12h", "failover-1d", "rotate-1m", "rotate-5m", "rotate-15m", "rotate-30m", "rotate-1h", "rotate-3h", "rotate-6h", "rotate-12h", "rotate-1d" ]
 
     def __init__(self, version):
         if version not in SUPPORTED_VERSIONS:
-            cli_print("Error: Only supports version 3 and 4, not %d" % version)
+            _cli_print("Error: Only supports version 3 and 4, not %d" % version)
             return
         self.version = version
         self.settings = OrderedDict()
@@ -235,17 +277,23 @@ class BeaconSettings:
         self.settings['MaxGetSize'] = packedSetting(4, confConsts.TYPE_INT)
         self.settings['Jitter'] = packedSetting(5, confConsts.TYPE_SHORT)
         self.settings['MaxDNS'] = packedSetting(6, confConsts.TYPE_SHORT)
-        # Silencing for now
-        #self.settings['PublicKey'] = packedSetting(7, confConsts.TYPE_STR, 256, isBlob=True)
+        # Silenced config
+        self.settings['PublicKey'] = packedSetting(7, confConsts.TYPE_STR, 256, isBlob=True)
         self.settings['PublicKey_MD5'] = packedSetting(7, confConsts.TYPE_STR, 256, isBlob=True, hashBlob=True)
         self.settings['C2Server'] = packedSetting(8, confConsts.TYPE_STR, 256)
         self.settings['UserAgent'] = packedSetting(9, confConsts.TYPE_STR, 128)
+        # TODO: Concat with C2Server?
         self.settings['HttpPostUri'] = packedSetting(10, confConsts.TYPE_STR, 64)
 
+        # This is how the server transforms its communication to the beacon
         # ref: https://www.cobaltstrike.com/help-malleable-c2 | https://usualsuspect.re/article/cobalt-strikes-malleable-c2-under-the-hood
+        # TODO: Switch to isHeaders parser logic
         self.settings['Malleable_C2_Instructions'] = packedSetting(11, confConsts.TYPE_STR, 256, isBlob=True,isMalleableStream=True)
+        # This is the way the beacon transforms its communication to the server
+        # TODO: Change name to HttpGet_Client and HttpPost_Client
         self.settings['HttpGet_Metadata'] = packedSetting(12, confConsts.TYPE_STR, 256, isHeaders=True)
         self.settings['HttpPost_Metadata'] = packedSetting(13, confConsts.TYPE_STR, 256, isHeaders=True)
+
         self.settings['SpawnTo'] = packedSetting(14, confConsts.TYPE_STR, 16, isBlob=True)
         self.settings['PipeName'] = packedSetting(15, confConsts.TYPE_STR, 128)
         # Options 16-18 are deprecated in 3.4
@@ -264,6 +312,7 @@ class BeaconSettings:
         self.settings['HttpPostChunk'] = packedSetting(28, confConsts.TYPE_INT)
         self.settings['Spawnto_x86'] = packedSetting(29, confConsts.TYPE_STR, 64)
         self.settings['Spawnto_x64'] = packedSetting(30, confConsts.TYPE_STR, 64)
+        # Whether the beacon encrypts his communication, should be always on (1) in beacon 4
         self.settings['CryptoScheme'] = packedSetting(31, confConsts.TYPE_SHORT)
         self.settings['Proxy_Config'] = packedSetting(32, confConsts.TYPE_STR, 128)
         self.settings['Proxy_User'] = packedSetting(33, confConsts.TYPE_STR, 64)
@@ -288,16 +337,29 @@ class BeaconSettings:
         # If True then allocation is using NtMapViewOfSection
         self.settings['ProcInject_AllocationMethod'] = packedSetting(52, confConsts.TYPE_SHORT, enum=self.ALLOCATION_FUNCTIONS)
         
-        # Unknown data, silencing for now
-        #self.settings['ProcInject_Stub'] = packedSetting(53, confConsts.TYPE_STR, 16, isBlob=True)
+        # Unknown data, silenced for now
+        self.settings['ProcInject_Stub'] = packedSetting(53, confConsts.TYPE_STR, 16, isBlob=True)
         self.settings['bUsesCookies'] = packedSetting(50, confConsts.TYPE_SHORT, isBool=True)
         self.settings['HostHeader'] = packedSetting(54, confConsts.TYPE_STR, 128)
 
-        # Silencing as I've yet to test it on a sample with those options
-        #self.settings['smbFrameHeader'] = packedSetting(57, confConsts.TYPE_STR, 128, isBlob=True)
-        #self.settings['tcpFrameHeader'] = packedSetting(58, confConsts.TYPE_STR, 128, isBlob=True)
+        # Silenced as I've yet to test it on a sample with those options
+        self.settings['smbFrameHeader'] = packedSetting(57, confConsts.TYPE_STR, 128, isBlob=True)
+        self.settings['tcpFrameHeader'] = packedSetting(58, confConsts.TYPE_STR, 128, isBlob=True)
         self.settings['headersToRemove'] = packedSetting(59, confConsts.TYPE_STR, 64)
 
+        # DNS Beacon
+        self.settings['DNS_Beaconing'] = packedSetting(60, confConsts.TYPE_STR, 33)
+        self.settings['DNS_get_TypeA'] = packedSetting(61, confConsts.TYPE_STR, 33)
+        self.settings['DNS_get_TypeAAAA'] = packedSetting(62, confConsts.TYPE_STR, 33)
+        self.settings['DNS_get_TypeTXT'] = packedSetting(63, confConsts.TYPE_STR, 33)
+        self.settings['DNS_put_metadata'] = packedSetting(64, confConsts.TYPE_STR, 33)
+        self.settings['DNS_put_output'] = packedSetting(65, confConsts.TYPE_STR, 33)
+        self.settings['DNS_resolver'] = packedSetting(66, confConsts.TYPE_STR, 15)
+        self.settings['DNS_strategy'] = packedSetting(67, confConsts.TYPE_SHORT, enum=self.ROTATE_STRATEGY)
+        self.settings['DNS_strategy_rotate_seconds'] = packedSetting(68, confConsts.TYPE_INT)
+        self.settings['DNS_strategy_fail_x'] = packedSetting(69, confConsts.TYPE_INT)
+        self.settings['DNS_strategy_fail_seconds'] = packedSetting(70, confConsts.TYPE_INT)
+
 
 class cobaltstrikeConfig:
     def __init__(self, f):
@@ -348,24 +410,45 @@ class cobaltstrikeConfig:
             if as_json:
                 continue
 
+            if conf_name in SILENT_CONFIGS:
+                continue
+
             if parsed_setting == 'Not Found' and quiet:
                 continue
-            if type(parsed_setting) != list:
-                if quiet and type(parsed_setting) == str and parsed_setting.strip() == '':
+
+            conf_type = type(parsed_setting)
+            if conf_type in (str, int, bytes):
+                if quiet and conf_type == str and parsed_setting.strip() == '':
                     continue
-                cli_print("{: <{width}} - {val}".format(conf_name, width=COLUMN_WIDTH-3, val=parsed_setting))
+                _cli_print("{: <{width}} - {val}".format(conf_name, width=COLUMN_WIDTH-3, val=parsed_setting))
+            
             elif parsed_setting == []:
                 if quiet:
                     continue
-                cli_print("{: <{width}} - {val}".format(conf_name, width=COLUMN_WIDTH-3, val='Empty'))
-            else:
-                cli_print("{: <{width}} - {val}".format(conf_name, width=COLUMN_WIDTH-3, val=parsed_setting[0]))
+                _cli_print("{: <{width}} - {val}".format(conf_name, width=COLUMN_WIDTH-3, val='Empty'))
+            
+            elif conf_type == dict: # the beautifulest code
+                conf_data = []
+                for k in parsed_setting.keys():
+                    if parsed_setting[k]:
+                        conf_data.append(k)
+                        for v in parsed_setting[k]:
+                            conf_data.append('\t' + v)
+                if not conf_data:
+                    continue
+                _cli_print("{: <{width}} - {val}".format(conf_name, width=COLUMN_WIDTH-3, val=conf_data[0]))
+                for val in conf_data[1:]:
+                    _cli_print(' ' * COLUMN_WIDTH, end='')
+                    _cli_print(val)        
+            
+            elif conf_type == list: # list
+                _cli_print("{: <{width}} - {val}".format(conf_name, width=COLUMN_WIDTH-3, val=parsed_setting[0]))
                 for val in parsed_setting[1:]:
-                    cli_print(' ' * COLUMN_WIDTH, end='')
-                    cli_print(val)
+                    _cli_print(' ' * COLUMN_WIDTH, end='')
+                    _cli_print(val)
 
         if as_json:
-            cli_print(json.dumps(parsed_config, cls=Base64Encoder))
+            _cli_print(json.dumps(parsed_config, cls=Base64Encoder))
 
         return parsed_config
 
@@ -401,7 +484,7 @@ class cobaltstrikeConfig:
         pe = pefile.PE(data=self.data)
         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")
+            _cli_print("Failed to find .data section")
             return False
         data = data_sections[0].get_data()
 
@@ -433,12 +516,29 @@ class cobaltstrikeConfig:
 
 
 if __name__ == '__main__':
-    parser = argparse.ArgumentParser(description="Parses CobaltStrike Beacon's configuration from PE or memory dump.")
-    parser.add_argument("path", help="Beacon file path")
+    parser = argparse.ArgumentParser(description="Parses CobaltStrike Beacon's configuration from PE, memory dump or URL.")
+    parser.add_argument("beacon", help="This can be a file path or a url (if started with http/s)")
     parser.add_argument("--json", help="Print as json", action="store_true", default=False)
     parser.add_argument("--quiet", help="Do not print missing or empty settings", action="store_true", default=False)
     parser.add_argument("--version", help="Try as specific cobalt version (3 or 4). If not specified, tries both.", type=int)
     args = parser.parse_args()
-    if not cobaltstrikeConfig(args.path).parse_config(version=args.version, quiet=args.quiet, as_json=args.json):
-        if not cobaltstrikeConfig(args.path).parse_encrypted_config(version=args.version, quiet=args.quiet, as_json=args.json):
-            print("Failed to find any beacon configuration")
+
+    if os.path.isfile(args.beacon):
+        if cobaltstrikeConfig(args.beacon).parse_config(version=args.version, quiet=args.quiet, as_json=args.json) or \
+        cobaltstrikeConfig(args.beacon).parse_encrypted_config(version=args.version, quiet=args.quiet, as_json=args.json):
+            exit(0)
+
+    elif args.beacon.lower().startswith('http'):
+        x86_beacon_data = get_beacon_data(args.beacon, 'x86')
+        x64_beacon_data = get_beacon_data(args.beacon, 'x64')
+        if not x86_beacon_data and not x64_beacon_data:
+            print("[-] Failed to find any beacon configuration")
+            exit(1)
+
+        conf_data = x86_beacon_data or x64_beacon_data
+        if cobaltstrikeConfig(BytesIO(conf_data)).parse_config(version=args.version, quiet=args.quiet, as_json=args.json) or \
+        cobaltstrikeConfig(BytesIO(conf_data)).parse_encrypted_config(version=args.version, quiet=args.quiet, as_json=args.json):
+            exit(0)
+
+    print("[-] Failed to find any beacon configuration")
+    exit(1)

requirements.txt

@@ -1,2 +1,4 @@
 netstruct==1.1.2
 pefile==2019.4.18
+#M2Crypto==0.37.1
+#pycryptodome==3.10.1