Lobshot
Lobshot a basic hVNC bot
Background
LOBSHOT is the internal name given to an hVNC based bot by the Elastic Security Labs team. The public name of this malware is currently unknown. The primary capability of the malware is acting as both a stealer and hidden VNC.
References
Samples
Analysis
DarkVNC Overlap
On VT the sample is being flagged as "DarkVNC" by Kaspersy this is possibly just an overlap with hVNC code from DarkVNC.
C2 Request
According to Elastic the C2 request contains some hardcoded data that can be used as a signature to hunt for the samples.
Searching for the above mov instruction paired with the first DWORD of the hardcoded value (C7 06 25 56 0A DC) shows over 550 samples in VirusTotal within the last year. With some of the first samples showing up in late July 2022. The prevalence of these hardcoded values shows that it has been actively used and under development for a long period of time, and will likely continue to be used in the future.
There appears to also be a version number at the top of this struct
Config Extractor
This has been copied from the original Elastic Github
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under
# one or more contributor license agreements. Licensed under the Elastic License 2.0;
# you may not use this file except in compliance with the Elastic License 2.0.
Notes
- The original sample used a convoluted way to obtain a static seed value used for the string decryption in other versions they realized this was silly and just hard coded a value
- There are different seed values for different variants
import pefile
import re
IP_ADDRESS_REGEX = r"^([0-9]{1,3}\.){3}[0-9]{1,3}$"
PORT_REGEX = r"^[0-9]{1,5}$"
def is_ascii(s):
return all((c < 128 and c > 39) for c in s)
def perform_extraction_file(file_path):
rdata = parse_file(file_path)
if not rdata:
raise RuntimeError(".rdata section not found")
candidates = generate_candidates(rdata)
strings = get_encrypted_strings(candidates)
#print(strings)
for s in strings:
if re.search(IP_ADDRESS_REGEX, s):
result_ip = s
if re.search(PORT_REGEX, s):
result_port = int(s)
if not result_ip:
raise RuntimeError("Configuration unsuccessful, could not extract IP/Port \n")
print(f"{result_ip}:{result_port}")
def string_decryption(encrypted_data, seed):
buffer = bytearray([0 for _ in range(len(encrypted_data) // 2)])
flag = False
z = 0
index = 0
index2 = 2
for i, x in enumerate(encrypted_data):
try:
y = encrypted_data[index + 1] - 0x61
index += 2
if flag:
buffer[i] = seed ^ z ^ (y | (16 * encrypted_data[index2] - 16)) & 0xFF
index2 += 2
else:
flag = True
z = y | (16 * (x - 1)) & 0xFF
except (IndexError, ValueError):
continue
buffer = buffer[1:]
return buffer
def parse_file(file_path):
with open(file_path, "rb") as data:
data = data.read()
pe = pefile.PE(data=data)
for section in pe.sections:
if b"rdata" in section.Name:
rdata = section.get_data()
return rdata
else:
print(".rdata section not found in {}\n".format(file_path.name))
return None
def generate_candidates(section_data):
candidates: list[bytes] = list()
blocks = section_data.split(b"\x00")
blocks = [x for x in blocks if x != b""]
for block in blocks:
if len(block) > 3 and not b"\\" in block:
candidates.append(block)
return candidates
def get_encrypted_strings(candidates):
string_table = []
seeds = 'abcdefghijklmnopqrstuvwxyz'
seeds += seeds.upper()
for s in seeds:
seed = ord(s)
tmp_out = []
for string in candidates:
decrypted_string = string_decryption(string, seed)
decrypted_string = decrypted_string.replace(b'\x00', b'')
if len(decrypted_string) < 3 or not is_ascii(decrypted_string):
continue
tmp_out.append(decrypted_string.decode('utf-8'))
if 'connect' in tmp_out or 'msedge.exe' in tmp_out or 'kernel32.dll' in tmp_out:
string_table = tmp_out
break
return string_table
def decrypt_candidates(candidates):
result_ip = None
result_port = None
for string in candidates:
decrypted_string = string_decryption(string, ord('S'))
print(decrypted_string)
if re.search(IP_ADDRESS_REGEX, decrypted_string):
result_ip = decrypted_string
if re.search(PORT_REGEX, decrypted_string):
result_port = int(decrypted_string)
if not result_ip:
raise RuntimeError("Configuration unsuccessful, could not extract IP/Port \n")
return result_ip.decode('utf-8'), result_port
def display_results(result):
print("FILE: {}".format(result[2].name))
print("IP: {}".format(result[0].decode("utf-8")))
print("Port: {}\n".format(result[1]))
FILE_PATH = '/tmp/e4ea88887753a936eaf3361dcc00380b88b0c210dcbde24f8f7ce27991856bf6'
perform_extraction_file(FILE_PATH)
chr(0x53)
import os
# assign directory
directory = '/tmp/lobshots'
# iterate over files in
# that directory
for filename in os.listdir(directory):
f = os.path.join(directory, filename)
# checking if it is a file
if os.path.isfile(f):
try:
print(f)
perform_extraction_file(f)
except:
print('\t fail!')