Untersuchung logarithmischer Artefakte
Bis jetzt haben wir gesehen, wie man mit Python Artefakte in Windows erhält. In diesem Kapitel erfahren Sie mehr über die Untersuchung protokollbasierter Artefakte mit Python.
Einführung
Protokollbasierte Artefakte sind der Schatz an Informationen, die für einen Experten für digitale Forensik sehr nützlich sein können. Obwohl wir über verschiedene Überwachungssoftware zum Sammeln der Informationen verfügen, besteht das Hauptproblem beim Analysieren nützlicher Informationen darin, dass wir viele Daten benötigen.
Verschiedene log-basierte Artefakte und Untersuchungen in Python
Lassen Sie uns in diesem Abschnitt verschiedene logbasierte Artefakte und ihre Untersuchung in Python diskutieren -
Zeitstempel
Der Zeitstempel übermittelt die Daten und die Zeit der Aktivität im Protokoll. Es ist eines der wichtigen Elemente jeder Protokolldatei. Beachten Sie, dass diese Daten- und Zeitwerte in verschiedenen Formaten vorliegen können.
Das unten gezeigte Python-Skript verwendet die unformatierte Datums- und Uhrzeitangabe als Eingabe und stellt einen formatierten Zeitstempel als Ausgabe bereit.
Für dieses Skript müssen wir die folgenden Schritte ausführen:
Richten Sie zunächst die Argumente ein, die den Rohdatenwert zusammen mit der Datenquelle und dem Datentyp verwenden.
Stellen Sie jetzt eine Klasse für die Bereitstellung einer gemeinsamen Schnittstelle für Daten in verschiedenen Datumsformaten bereit.
Python-Code
Lassen Sie uns sehen, wie Python-Code für diesen Zweck verwendet wird -
Importieren Sie zunächst die folgenden Python-Module:
from __future__ import print_function
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
from datetime import datetime as dt
from datetime import timedeltaJetzt müssen wir wie üblich Argumente für den Befehlszeilen-Handler bereitstellen. Hier werden drei Argumente akzeptiert: erstens der zu verarbeitende Datumswert, zweitens die Quelle dieses Datumswerts und drittens der Typ -
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)Jetzt müssen wir eine Klasse definieren, die die Argumente für Datumswert, Datumsquelle und Werttyp akzeptiert.
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 = NoneJetzt definieren wir eine Methode, die sich wie die main () -Methode wie ein Controller verhält -
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']Jetzt müssen wir zwei Methoden definieren, die die Unix-Epochenzeit bzw. FILETIME verarbeiten -
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')Nach dem Ausführen des obigen Skripts können wir durch Angabe eines Zeitstempels den konvertierten Wert in einem einfach zu lesenden Format erhalten.
Webserver-Protokolle
Aus Sicht des Experten für digitale Forensik sind Webserverprotokolle ein weiteres wichtiges Artefakt, da sie nützliche Benutzerstatistiken sowie Informationen über den Benutzer und die geografischen Standorte enthalten können. Das folgende Python-Skript erstellt nach der Verarbeitung der Webserver-Protokolle eine Tabelle zur einfachen Analyse der Informationen.
Zunächst müssen wir die folgenden Python-Module importieren:
from __future__ import print_function
from argparse import ArgumentParser, FileType
import re
import shlex
import logging
import sys
import csv
logger = logging.getLogger(__file__)Jetzt müssen wir die Muster definieren, die aus den Protokollen analysiert werden -
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*"))]Geben Sie nun ein Argument für den Befehlszeilenhandler an. Hier werden zwei Argumente akzeptiert: erstens das zu verarbeitende IIS-Protokoll und zweitens der gewünschte CSV-Dateipfad.
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")Jetzt müssen wir die main () -Methode definieren, die das Skript für Massenprotokollinformationen verarbeitet.
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")Zuletzt müssen wir eine Methode definieren, die die Ausgabe in eine Tabelle schreibt -
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)Nach dem Ausführen des obigen Skripts erhalten wir die Webserver-basierten Protokolle in einer Tabelle.
Scannen wichtiger Dateien mit YARA
YARA (Noch ein rekursiver Algorithmus) ist ein Dienstprogramm zur Mustererkennung, das zur Identifizierung von Malware und zur Reaktion auf Vorfälle entwickelt wurde. Wir werden YARA zum Scannen der Dateien verwenden. Im folgenden Python-Skript verwenden wir YARA.
Wir können YARA mit Hilfe des folgenden Befehls installieren:
pip install YARAWir können die folgenden Schritte ausführen, um YARA-Regeln zum Scannen von Dateien zu verwenden.
Richten Sie zunächst YARA-Regeln ein und kompilieren Sie sie
Scannen Sie dann eine einzelne Datei und durchlaufen Sie die Verzeichnisse, um einzelne Dateien zu verarbeiten.
Zuletzt werden wir das Ergebnis in CSV exportieren.
Python-Code
Lassen Sie uns sehen, wie Python-Code für diesen Zweck verwendet wird -
Zuerst müssen wir die folgenden Python-Module importieren -
from __future__ import print_function
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
import os
import csv
import yaraGeben Sie als Nächstes ein Argument für den Befehlszeilenhandler an. Beachten Sie, dass hier zwei Argumente akzeptiert werden: Erstens der Pfad zu den YARA-Regeln, zweitens die zu scannende Datei.
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)Jetzt definieren wir die main () - Funktion, die den Pfad zu den zu scannenden Yara-Regeln und -Dateien akzeptiert.
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)Definieren Sie nun eine Methode, die das Verzeichnis durchläuft und das Ergebnis zur weiteren Verarbeitung an eine andere Methode weitergibt.
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_infoDefinieren Sie als Nächstes zwei Funktionen. Beachten Sie, dass wir zuerst verwenden werdenmatch() Methode zu yrulesobject und ein anderes melden diese übereinstimmenden Informationen an die Konsole, wenn der Benutzer keine Ausgabedatei angibt. Beachten Sie den unten gezeigten Code -
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)Zuletzt definieren wir eine Methode, die die Ausgabe in die CSV-Datei schreibt, wie unten gezeigt -
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)Sobald Sie das obige Skript erfolgreich ausgeführt haben, können wir entsprechende Argumente in der Befehlszeile bereitstellen und einen CSV-Bericht erstellen.