forked from Sentinel-One/CobaltStrikeParser
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathparse_beacon_config.py
552 lines (467 loc) · 24.9 KB
/
parse_beacon_config.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
#!/usr/bin/python3
'''
Parses CobaltStrike Beacon's configuration from PE file or memory dump.
By Gal Kristal from SentinelOne ([email protected]) @gal_kristal
Inspired by https://github.com/JPCERTCC/aa-tools/blob/master/cobaltstrikescan.py
TODO:
1. Parse headers modifiers
2. Dynamic size parsing
'''
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 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']
def _cli_print(msg, end='\n'):
if __name__ == '__main__':
print(msg, end=end)
class confConsts:
MAX_SETTINGS = 64
TYPE_NONE = 0
TYPE_SHORT = 1
TYPE_INT = 2
TYPE_STR = 3
START_PATTERNS = {
3: b'\x69\x68\x69\x68\x69\x6b..\x69\x6b\x69\x68\x69\x6b..\x69\x6a',
4: b'\x2e\x2f\x2e\x2f\x2e\x2c..\x2e\x2c\x2e\x2f\x2e\x2c..\x2e'
}
START_PATTERN_DECODED = b'\x00\x01\x00\x01\x00\x02..\x00\x02\x00\x01\x00\x02..\x00'
CONFIG_SIZE = 4096
XORBYTES = {
3: 0x69,
4: 0x2e
}
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):
self.pos = pos
self.datatype = datatype
self.is_blob = isBlob
self.is_headers = isHeaders
self.is_ipaddress = isIpAddress
self.is_bool = isBool
self.is_date = isDate
self.is_malleable_stream = isMalleableStream
self.bool_false_value = boolFalseValue
self.is_transform = isProcInjectTransform
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"))
self.length = length
if datatype == confConsts.TYPE_SHORT:
self.length = 2
elif datatype == confConsts.TYPE_INT:
self.length = 4
def binary_repr(self):
"""
Param number - Type - Length - Value
"""
self_repr = bytearray(6)
self_repr[1] = self.pos
self_repr[3] = self.datatype
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'
repr_len = len(self.binary_repr())
conf_data = full_config_data[data_offset + repr_len : data_offset + repr_len + self.length]
if self.datatype == confConsts.TYPE_SHORT:
conf_data = unpack('>H', conf_data)[0]
if self.is_bool:
ret = 'False' if conf_data == self.bool_false_value else 'True'
return ret
elif self.enum:
return self.enum[conf_data]
elif self.mask:
ret_arr = []
for k,v in self.mask.items():
if k == 0 and k == conf_data:
ret_arr.append(v)
if k & conf_data:
ret_arr.append(v)
return ret_arr
else:
return conf_data
elif self.datatype == confConsts.TYPE_INT:
if self.is_ipaddress:
return inet_ntoa(conf_data)
else:
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:])
return conf_data
if self.is_blob:
if self.enum != None:
ret_arr = []
i = 0
while i < len(conf_data):
v = conf_data[i]
if v == 0:
return ret_arr
v = self.enum[v]
if v:
ret_arr.append(v)
i+=1
# Only EXECUTE_TYPE for now
else:
# Skipping unknown short value in the start
string1 = netunpack(b'I$', conf_data[i+3:])[0].decode()
string2 = netunpack(b'I$', conf_data[i+3+4+len(string1):])[0].decode()
ret_arr.append("%s:%s" % (string1.strip('\x00'),string2.strip('\x00')))
i += len(string1) + len(string2) + 11
if self.is_transform:
if conf_data == bytes(len(conf_data)):
return 'Empty'
ret_arr = []
prepend_length = unpack('>I', conf_data[0:4])[0]
prepend = conf_data[4 : 4+prepend_length]
append_length_offset = prepend_length + 4
append_length = unpack('>I', conf_data[append_length_offset : append_length_offset+4])[0]
append = conf_data[append_length_offset+4 : append_length_offset+4+append_length]
ret_arr.append(prepend)
ret_arr.append(append if append_length < 256 and append != bytes(append_length) else 'Empty')
return ret_arr
if self.is_malleable_stream:
prog = []
fh = io.BytesIO(conf_data)
while True:
op = read_dword_be(fh)
if not op:
break
if op == 1:
l = read_dword_be(fh)
prog.append("Remove %d bytes from the end" % l)
elif op == 2:
l = read_dword_be(fh)
prog.append("Remove %d bytes from the beginning" % l)
elif op == 3:
prog.append("Base64 decode")
elif op == 8:
prog.append("NetBIOS decode 'a'")
elif op == 11:
prog.append("NetBIOS decode 'A'")
elif op == 13:
prog.append("Base64 URL-safe decode")
elif op == 15:
prog.append("XOR mask w/ random key")
conf_data = prog
if self.hashBlob:
conf_data = conf_data.strip(b'\x00')
conf_data = hashlib.md5(conf_data).hexdigest()
return conf_data
if self.is_headers:
return self.parse_transformdata(conf_data)
conf_data = conf_data.strip(b'\x00').decode()
return conf_data
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"}
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)
return
self.version = version
self.settings = OrderedDict()
self.init()
def init(self):
self.settings['BeaconType'] = packedSetting(1, confConsts.TYPE_SHORT, mask=self.BEACON_TYPE)
self.settings['Port'] = packedSetting(2, confConsts.TYPE_SHORT)
self.settings['SleepTime'] = packedSetting(3, confConsts.TYPE_INT)
self.settings['MaxGetSize'] = packedSetting(4, confConsts.TYPE_INT)
self.settings['Jitter'] = packedSetting(5, confConsts.TYPE_SHORT)
self.settings['MaxDNS'] = packedSetting(6, confConsts.TYPE_SHORT)
# 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
self.settings['DNS_Idle'] = packedSetting(19, confConsts.TYPE_INT, isIpAddress=True)
self.settings['DNS_Sleep'] = packedSetting(20, confConsts.TYPE_INT)
# Options 21-25 are for SSHAgent
self.settings['SSH_Host'] = packedSetting(21, confConsts.TYPE_STR, 256)
self.settings['SSH_Port'] = packedSetting(22, confConsts.TYPE_SHORT)
self.settings['SSH_Username'] = packedSetting(23, confConsts.TYPE_STR, 128)
self.settings['SSH_Password_Plaintext'] = packedSetting(24, confConsts.TYPE_STR, 128)
self.settings['SSH_Password_Pubkey'] = packedSetting(25, confConsts.TYPE_STR, 6144)
self.settings['SSH_Banner'] = packedSetting(54, confConsts.TYPE_STR, 128)
self.settings['HttpGet_Verb'] = packedSetting(26, confConsts.TYPE_STR, 16)
self.settings['HttpPost_Verb'] = packedSetting(27, confConsts.TYPE_STR, 16)
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)
self.settings['Proxy_Password'] = packedSetting(34, confConsts.TYPE_STR, 64)
self.settings['Proxy_Behavior'] = packedSetting(35, confConsts.TYPE_SHORT, enum=self.ACCESS_TYPE)
# Option 36 is deprecated
self.settings['Watermark'] = packedSetting(37, confConsts.TYPE_INT)
self.settings['bStageCleanup'] = packedSetting(38, confConsts.TYPE_SHORT, isBool=True)
self.settings['bCFGCaution'] = packedSetting(39, confConsts.TYPE_SHORT, isBool=True)
self.settings['KillDate'] = packedSetting(40, confConsts.TYPE_INT, isDate=True)
# Inner parameter, does not seem interesting so silencing
#self.settings['textSectionEnd (0 if !sleep_mask)'] = packedSetting(41, confConsts.TYPE_INT)
#TODO: dynamic size parsing
#self.settings['ObfuscateSectionsInfo'] = packedSetting(42, confConsts.TYPE_STR, %d, isBlob=True)
self.settings['bProcInject_StartRWX'] = packedSetting(43, confConsts.TYPE_SHORT, isBool=True, boolFalseValue=4)
self.settings['bProcInject_UseRWX'] = packedSetting(44, confConsts.TYPE_SHORT, isBool=True, boolFalseValue=32)
self.settings['bProcInject_MinAllocSize'] = packedSetting(45, confConsts.TYPE_INT)
self.settings['ProcInject_PrependAppend_x86'] = packedSetting(46, confConsts.TYPE_STR, 256, isBlob=True, isProcInjectTransform=True)
self.settings['ProcInject_PrependAppend_x64'] = packedSetting(47, confConsts.TYPE_STR, 256, isBlob=True, isProcInjectTransform=True)
self.settings['ProcInject_Execute'] = packedSetting(51, confConsts.TYPE_STR, 128, isBlob=True, enum=self.EXECUTE_TYPE)
# If True then allocation is using NtMapViewOfSection
self.settings['ProcInject_AllocationMethod'] = packedSetting(52, confConsts.TYPE_SHORT, enum=self.ALLOCATION_FUNCTIONS)
# 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)
# 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):
'''
f: file path or file-like object
'''
self.data = None
if isinstance(f, str):
with open(f, 'rb') as fobj:
self.data = fobj.read()
else:
self.data = f.read()
"""Parse the CobaltStrike configuration"""
@staticmethod
def decode_config(cfg_blob, version):
return bytes([cfg_offset ^ confConsts.XORBYTES[version] for cfg_offset in cfg_blob])
def _parse_config(self, version, quiet=False, as_json=False):
'''
Parses beacon's configuration from beacon PE or memory dump.
Returns json of config is found; else it returns None.
:int version: Try a specific version (3 or 4), or leave None to try both of them
:bool quiet: Whether to print missing or empty settings
:bool as_json: Whether to dump as json
'''
re_start_match = re.search(confConsts.START_PATTERNS[version], self.data)
re_start_decoded_match = re.search(confConsts.START_PATTERN_DECODED, self.data)
if not re_start_match and not re_start_decoded_match:
return None
encoded_config_offset = re_start_match.start() if re_start_match else -1
decoded_config_offset = re_start_decoded_match.start() if re_start_decoded_match else -1
if encoded_config_offset >= 0:
full_config_data = cobaltstrikeConfig.decode_config(self.data[encoded_config_offset : encoded_config_offset + confConsts.CONFIG_SIZE], version=version)
else:
full_config_data = self.data[decoded_config_offset : decoded_config_offset + confConsts.CONFIG_SIZE]
parsed_config = {}
settings = BeaconSettings(version).settings.items()
for conf_name, packed_conf in settings:
parsed_setting = packed_conf.pretty_repr(full_config_data)
parsed_config[conf_name] = parsed_setting
if as_json:
continue
if conf_name in SILENT_CONFIGS:
continue
if parsed_setting == 'Not Found' and quiet:
continue
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))
elif parsed_setting == []:
if quiet:
continue
_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)
if as_json:
_cli_print(json.dumps(parsed_config, cls=Base64Encoder))
return parsed_config
def parse_config(self, version=None, quiet=False, as_json=False):
'''
Parses beacon's configuration from beacon PE or memory dump
Returns json of config is found; else it returns None.
:int version: Try a specific version (3 or 4), or leave None to try both of them
:bool quiet: Whether to print missing or empty settings
:bool as_json: Whether to dump as json
'''
if not version:
for ver in SUPPORTED_VERSIONS:
parsed = self._parse_config(version=ver, quiet=quiet, as_json=as_json)
if parsed:
return parsed
else:
return self._parse_config(version=version, quiet=quiet, as_json=as_json)
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
Returns json of config is found; else it returns None.
:bool quiet: Whether to print missing settings
:bool as_json: Whether to dump as json
'''
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")
return False
data = data_sections[0].get_data()
offset = 0
key_found = False
while offset < len(data):
key = data[offset:offset+4]
if key != bytes(4):
if data.count(key) >= THRESHOLD:
key_found = True
size = int.from_bytes(data[offset-4:offset], 'little')
encrypted_data_offset = offset+16 - (offset % 16)
break
offset += 4
if not key_found:
return False
# decrypt
enc_data = data[encrypted_data_offset:encrypted_data_offset+size]
dec_data = []
for i,c in enumerate(enc_data):
dec_data.append(c ^ key[i % 4])
dec_data = bytes(dec_data)
self.data = dec_data
return self.parse_config(version=version, quiet=quiet, as_json=as_json)
if __name__ == '__main__':
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 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)