Исследование артефактов на основе журналов
До сих пор мы видели, как получать артефакты в Windows с помощью Python. В этой главе давайте узнаем об исследовании артефактов на основе журналов с помощью Python.
Введение
Артефакты на основе журналов - это кладезь информации, которая может быть очень полезна эксперту в области цифровой криминалистики. Хотя у нас есть различное программное обеспечение для мониторинга для сбора информации, основная проблема для извлечения из них полезной информации заключается в том, что нам нужно много данных.
Различные артефакты на основе журналов и исследования в Python
В этом разделе давайте обсудим различные артефакты на основе журналов и их исследование в Python.
Отметки времени
Отметка времени передает данные и время активности в журнале. Это один из важных элементов любого файла журнала. Обратите внимание, что эти значения данных и времени могут иметь различные форматы.
Скрипт Python, показанный ниже, примет необработанные данные о дате и времени в качестве входных данных и предоставит отформатированную метку времени в качестве вывода.
Для этого скрипта нам нужно выполнить следующие шаги -
Сначала настройте аргументы, которые будут принимать значение необработанных данных вместе с источником данных и типом данных.
Теперь предоставьте класс для предоставления общего интерфейса для данных в разных форматах даты.
Код Python
Давайте посмотрим, как использовать код Python для этой цели -
Сначала импортируйте следующие модули Python -
from __future__ import print_function
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
from datetime import datetime as dt
from datetime import timedeltaТеперь, как обычно, нам нужно предоставить аргумент для обработчика командной строки. Здесь он примет три аргумента: первый - это значение даты, которое нужно обработать, второй - источник этого значения даты, а третий - его тип -
if __name__ == '__main__':
parser = ArgumentParser('Timestamp Log-based artifact')
parser.add_argument("date_value", help="Raw date value to parse")
parser.add_argument(
"source", help = "Source format of date",choices = ParseDate.get_supported_formats())
parser.add_argument(
"type", help = "Data type of input value",choices = ('number', 'hex'), default = 'int')
args = parser.parse_args()
date_parser = ParseDate(args.date_value, args.source, args.type)
date_parser.run()
print(date_parser.timestamp)Теперь нам нужно определить класс, который будет принимать аргументы для значения даты, источника даты и типа значения -
class ParseDate(object):
def __init__(self, date_value, source, data_type):
self.date_value = date_value
self.source = source
self.data_type = data_type
self.timestamp = NoneТеперь мы определим метод, который будет действовать как контроллер, как и метод main () -
def run(self):
if self.source == 'unix-epoch':
self.parse_unix_epoch()
elif self.source == 'unix-epoch-ms':
self.parse_unix_epoch(True)
elif self.source == 'windows-filetime':
self.parse_windows_filetime()
@classmethod
def get_supported_formats(cls):
return ['unix-epoch', 'unix-epoch-ms', 'windows-filetime']Теперь нам нужно определить два метода, которые будут обрабатывать время эпохи Unix и FILETIME соответственно:
def parse_unix_epoch(self, milliseconds=False):
if self.data_type == 'hex':
conv_value = int(self.date_value)
if milliseconds:
conv_value = conv_value / 1000.0
elif self.data_type == 'number':
conv_value = float(self.date_value)
if milliseconds:
conv_value = conv_value / 1000.0
else:
print("Unsupported data type '{}' provided".format(self.data_type))
sys.exit('1')
ts = dt.fromtimestamp(conv_value)
self.timestamp = ts.strftime('%Y-%m-%d %H:%M:%S.%f')
def parse_windows_filetime(self):
if self.data_type == 'hex':
microseconds = int(self.date_value, 16) / 10.0
elif self.data_type == 'number':
microseconds = float(self.date_value) / 10
else:
print("Unsupported data type '{}' provided".format(self.data_type))
sys.exit('1')
ts = dt(1601, 1, 1) + timedelta(microseconds=microseconds)
self.timestamp = ts.strftime('%Y-%m-%d %H:%M:%S.%f')После выполнения вышеуказанного скрипта, указав временную метку, мы можем получить преобразованное значение в удобном для чтения формате.
Журналы веб-сервера
С точки зрения эксперта в области цифровой криминалистики, журналы веб-сервера являются еще одним важным артефактом, потому что они могут получать полезную статистику пользователей вместе с информацией о пользователе и географическом местоположении. Ниже приведен сценарий Python, который после обработки журналов веб-сервера создаст электронную таблицу для облегчения анализа информации.
Прежде всего нам нужно импортировать следующие модули Python -
from __future__ import print_function
from argparse import ArgumentParser, FileType
import re
import shlex
import logging
import sys
import csv
logger = logging.getLogger(__file__)Теперь нам нужно определить шаблоны, которые будут анализироваться из журналов -
iis_log_format = [
("date", re.compile(r"\d{4}-\d{2}-\d{2}")),
("time", re.compile(r"\d\d:\d\d:\d\d")),
("s-ip", re.compile(
r"((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\.|$)){4}")),
("cs-method", re.compile(
r"(GET)|(POST)|(PUT)|(DELETE)|(OPTIONS)|(HEAD)|(CONNECT)")),
("cs-uri-stem", re.compile(r"([A-Za-z0-1/\.-]*)")),
("cs-uri-query", re.compile(r"([A-Za-z0-1/\.-]*)")),
("s-port", re.compile(r"\d*")),
("cs-username", re.compile(r"([A-Za-z0-1/\.-]*)")),
("c-ip", re.compile(
r"((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\.|$)){4}")),
("cs(User-Agent)", re.compile(r".*")),
("sc-status", re.compile(r"\d*")),
("sc-substatus", re.compile(r"\d*")),
("sc-win32-status", re.compile(r"\d*")),
("time-taken", re.compile(r"\d*"))]Теперь предоставьте аргумент для обработчика командной строки. Здесь он будет принимать два аргумента: первый - это журнал IIS, который нужно обработать, второй - желаемый путь к файлу CSV.
if __name__ == '__main__':
parser = ArgumentParser('Parsing Server Based Logs')
parser.add_argument('iis_log', help = "Path to IIS Log",type = FileType('r'))
parser.add_argument('csv_report', help = "Path to CSV report")
parser.add_argument('-l', help = "Path to processing log",default=__name__ + '.log')
args = parser.parse_args()
logger.setLevel(logging.DEBUG)
msg_fmt = logging.Formatter(
"%(asctime)-15s %(funcName)-10s ""%(levelname)-8s %(message)s")
strhndl = logging.StreamHandler(sys.stdout)
strhndl.setFormatter(fmt = msg_fmt)
fhndl = logging.FileHandler(args.log, mode = 'a')
fhndl.setFormatter(fmt = msg_fmt)
logger.addHandler(strhndl)
logger.addHandler(fhndl)
logger.info("Starting IIS Parsing ")
logger.debug("Supplied arguments: {}".format(", ".join(sys.argv[1:])))
logger.debug("System " + sys.platform)
logger.debug("Version " + sys.version)
main(args.iis_log, args.csv_report, logger)
iologger.info("IIS Parsing Complete")Теперь нам нужно определить метод main (), который будет обрабатывать скрипт для информации массового журнала -
def main(iis_log, report_file, logger):
parsed_logs = []
for raw_line in iis_log:
line = raw_line.strip()
log_entry = {}
if line.startswith("#") or len(line) == 0:
continue
if '\"' in line:
line_iter = shlex.shlex(line_iter)
else:
line_iter = line.split(" ")
for count, split_entry in enumerate(line_iter):
col_name, col_pattern = iis_log_format[count]
if col_pattern.match(split_entry):
log_entry[col_name] = split_entry
else:
logger.error("Unknown column pattern discovered. "
"Line preserved in full below")
logger.error("Unparsed Line: {}".format(line))
parsed_logs.append(log_entry)
logger.info("Parsed {} lines".format(len(parsed_logs)))
cols = [x[0] for x in iis_log_format]
logger.info("Creating report file: {}".format(report_file))
write_csv(report_file, cols, parsed_logs)
logger.info("Report created")Наконец, нам нужно определить метод, который будет записывать вывод в электронную таблицу -
def write_csv(outfile, fieldnames, data):
with open(outfile, 'w', newline="") as open_outfile:
csvfile = csv.DictWriter(open_outfile, fieldnames)
csvfile.writeheader()
csvfile.writerows(data)После запуска вышеуказанного сценария мы получим журналы веб-сервера в электронной таблице.
Сканирование важных файлов с помощью YARA
YARA (еще один рекурсивный алгоритм) - это утилита сопоставления шаблонов, разработанная для идентификации вредоносных программ и реагирования на инциденты. Мы будем использовать YARA для сканирования файлов. В следующем скрипте Python мы будем использовать YARA.
Мы можем установить YARA с помощью следующей команды -
pip install YARAМы можем выполнить следующие шаги, чтобы использовать правила YARA для сканирования файлов:
Сначала настройте и скомпилируйте правила YARA
Затем просканируйте один файл и затем переберите каталоги для обработки отдельных файлов.
Наконец, мы экспортируем результат в CSV.
Код Python
Давайте посмотрим, как использовать код Python для этой цели -
Во-первых, нам нужно импортировать следующие модули Python -
from __future__ import print_function
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
import os
import csv
import yaraЗатем укажите аргумент для обработчика командной строки. Обратите внимание, что здесь он принимает два аргумента: первый - это путь к правилам YARA, второй - это файл, который нужно просканировать.
if __name__ == '__main__':
parser = ArgumentParser('Scanning files by YARA')
parser.add_argument(
'yara_rules',help = "Path to Yara rule to scan with. May be file or folder path.")
parser.add_argument('path_to_scan',help = "Path to file or folder to scan")
parser.add_argument('--output',help = "Path to output a CSV report of scan results")
args = parser.parse_args()
main(args.yara_rules, args.path_to_scan, args.output)Теперь мы определим функцию main (), которая будет принимать путь к правилам yara и файлу для сканирования -
def main(yara_rules, path_to_scan, output):
if os.path.isdir(yara_rules):
yrules = yara.compile(yara_rules)
else:
yrules = yara.compile(filepath=yara_rules)
if os.path.isdir(path_to_scan):
match_info = process_directory(yrules, path_to_scan)
else:
match_info = process_file(yrules, path_to_scan)
columns = ['rule_name', 'hit_value', 'hit_offset', 'file_name',
'rule_string', 'rule_tag']
if output is None:
write_stdout(columns, match_info)
else:
write_csv(output, columns, match_info)Теперь определите метод, который будет перебирать каталог и передавать результат другому методу для дальнейшей обработки -
def process_directory(yrules, folder_path):
match_info = []
for root, _, files in os.walk(folder_path):
for entry in files:
file_entry = os.path.join(root, entry)
match_info += process_file(yrules, file_entry)
return match_infoЗатем определите две функции. Обратите внимание, что сначала мы будем использоватьmatch() способ yrulesобъект, а другой сообщит об этой совпадающей информации в консоль, если пользователь не укажет какой-либо выходной файл. Обратите внимание на код, показанный ниже -
def process_file(yrules, file_path):
match = yrules.match(file_path)
match_info = []
for rule_set in match:
for hit in rule_set.strings:
match_info.append({
'file_name': file_path,
'rule_name': rule_set.rule,
'rule_tag': ",".join(rule_set.tags),
'hit_offset': hit[0],
'rule_string': hit[1],
'hit_value': hit[2]
})
return match_info
def write_stdout(columns, match_info):
for entry in match_info:
for col in columns:
print("{}: {}".format(col, entry[col]))
print("=" * 30)Наконец, мы определим метод, который будет записывать вывод в файл CSV, как показано ниже -
def write_csv(outfile, fieldnames, data):
with open(outfile, 'w', newline="") as open_outfile:
csvfile = csv.DictWriter(open_outfile, fieldnames)
csvfile.writeheader()
csvfile.writerows(data)После успешного выполнения вышеуказанного сценария мы можем предоставить соответствующие аргументы в командной строке и сгенерировать отчет в формате CSV.