codex-admin/autopsy
1
0
Fork 0
mirror of https://github.com/elisspace/autopsy.git synced 2026-08-31 00:22:00 +00:00
Code Issues Releases Wiki Activity

Fully implemented line parser

Browse Source
This commit is contained in:
U-BASIS\dsmyda
2019-09-15 22:04:27 -04:00
parent 8edaab679e
commit 8259882b71
2 changed files with 136 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
InternalPythonModules/android/TskCallLogsParser.py
View File

@@ -38,6 +38,7 @@ class TskCallLogsParser(ResultSetIterator):
self._DEFAULT_DIRECTION = AppDBParserHelper.CommunicationDirection.UNKNOWN
self._DEFAULT_ADDRESS = None
self._DEFAULT_CALL_TYPE = AppDBParserHelper.CallMediaType.UNKNOWN
self._DEFAULT_LONG = -1
self.INCOMING_CALL = AppDBParserHelper.CommunicationDirection.INCOMING
self.OUTGOING_CALL = AppDBParserHelper.CommunicationDirection.OUTGOING

211
InternalPythonModules/android/line.py
View File

@@ -31,8 +31,8 @@ from org.apache.commons.codec.binary import Base64
from org.sleuthkit.autopsy.casemodule import Case
from org.sleuthkit.autopsy.coreutils import Logger
from org.sleuthkit.autopsy.coreutils import MessageNotifyUtil
from org.sleuthkit.autopsy.coreutils import AppSQLiteDB as SQLiteUtil
from org.sleuthkit.autopsy.coreutils import AppDBParserHelper as BlackboardUtil
from org.sleuthkit.autopsy.coreutils import AppSQLiteDB
from org.sleuthkit.autopsy.coreutils import AppDBParserHelper
from org.sleuthkit.autopsy.datamodel import ContentUtils
from org.sleuthkit.autopsy.ingest import IngestJobContext
from org.sleuthkit.datamodel import AbstractFile
@@ -57,18 +57,22 @@ class LineAnalyzer(general.AndroidComponentAnalyzer):
self._logger = Logger.getLogger(self.__class__.__name__)
self._LINE_PACKAGE_NAME = "jp.naver.line.android"
self._PARSER_NAME = "Line Parser"
self._VERSION = "9.15.1"
def analyze(self, dataSource, fileManager, context):
try:
contact_and_message_dbs = SQLiteUtil.findAppDatabases(dataSource, "naver_line.db", True, self._LINE_PACKAGE_NAME)
calllog_dbs = SQLiteUtil.findAppDatabases(dataSource, "call_history", True, self._LINE_PACKAGE_NAME)
contact_and_message_dbs = AppSQLiteDB.findAppDatabases(dataSource,
"naver_line", True, self._LINE_PACKAGE_NAME)
calllog_dbs = AppSQLiteDB.findAppDatabases(dataSource,
"call_history", True, self._LINE_PACKAGE_NAME)
for contact_and_message_db in contact_and_message_dbs:
blackboard_util = BlackboardUtil(self._PARSER_NAME, contact_and_message_db.getDBFile(), Account.Type.LINE)
helper = AppDBParserHelper(self._PARSER_NAME,
contact_and_message_db.getDBFile(), Account.Type.LINE)
contacts_parser = LineContactsParser(contact_and_message_db)
while contacts_parser.next():
blackboard_util.addContact(
helper.addContact(
contacts_parser.get_account_name(),
contacts_parser.get_contact_name(),
contacts_parser.get_phone(),
@@ -77,11 +81,10 @@ class LineAnalyzer(general.AndroidComponentAnalyzer):
contacts_parser.get_email()
)
contacts_parser.close()
"""
messages_parser = LineMessagesParser(line_db)
messages_parser = LineMessagesParser(contact_and_message_db)
while messages_parser.next():
blackboard_util.addMessage(
messages_parser.get_account_id(),
helper.addMessage(
messages_parser.get_message_type(),
messages_parser.get_message_direction(),
messages_parser.get_phone_number_from(),
@@ -93,36 +96,27 @@ class LineAnalyzer(general.AndroidComponentAnalyzer):
messages_parser.get_thread_id()
)
messages_parser.close()
"""
contact_and_message_db.close()
for calllog_db in calllog_dbs:
blackboard_util = BlackboardUtil(self._PARSER_NAME, calllog_db.getDBFile(), Account.Type.LINE)
calllog_db.attachDatabase(dataSource, "naver_line.db", True, calllog_db.getDBFile().getParentPath(), "naver")
helper = AppDBParserHelper(self._PARSER_NAME,
calllog_db.getDBFile(), Account.Type.LINE)
calllog_db.attachDatabase(dataSource,
"naver_line", calllog_db.getDBFile().getParentPath(), "naver")
calllog_parser = LineCallLogsParser(calllog_db)
while calllog_parser.next():
print(calllog_parser.get_account_name())
print(calllog_parser.get_contact_name())
print(calllog_parser.get_call_direction())
print(calllog_parser.get_phone_number_from())
print(calllog_parser.get_phone_number_to())
print(calllog_parser.get_call_start_date_time())
print(calllog_parser.get_call_end_date_time())
blackboard_util.addCalllog(
calllog_parser.get_account_name(),
helper.addCalllog(
calllog_parser.get_call_direction(),
calllog_parser.get_phone_number_from(),
calllog_parser.get_phone_number_to(),
calllog_parser.get_call_start_date_time(),
calllog_parser.get_call_end_date_time(),
calllog_parser.get_contact_name()
calllog_parser.get_call_type()
)
calllog_db.detachDatabase("naver")
calllog_parser.close()
calllog_db.close()
calllog_db.close()
except (SQLException, TskCoreException) as ex:
# Error parsing Line databases.
self._logger.log(Level.WARNING, "Error parsing the Line App Databases", ex)
@@ -141,47 +135,62 @@ class LineCallLogsParser(TskCallLogsParser):
SELECT substr(CallH.call_type, -1) AS direction,
CallH.start_time AS start_time,
CallH.end_time AS end_time,
ConT.server_name AS account_name,
ConT.name AS contact_name
FROM call_history AS CallH
JOIN naver.contacts AS ConT
ON CallH.caller_mid = ConT.m_id
ConT.server_name AS name,
CallH.voip_type AS call_type,
ConT.m_id
FROM call_history AS CallH
JOIN naver.contacts AS ConT
ON CallH.caller_mid = ConT.m_id
"""
)
)
self._OUTGOING_CALL = "O"
self._INCOMING_CALL = "I"
self._OUTGOING_CALL_TYPE = "O"
self._INCOMING_CALL_TYPE = "I"
self._had_error = False
self._VIDEO_CALL_TYPE = "V"
self._AUDIO_CALL_TYPE = "A"
def get_call_direction(self):
direction = self.result_set.getString("direction")
if direction == self._OUTGOING_CALL:
return self.OUTGOING_MSG_STRING
return self.INCOMING_MSG_STRING
def get_call_direction(self):
direction = self.result_set.getString("direction")
if direction == self._OUTGOING_CALL_TYPE:
return self.OUTGOING_CALL
return self.INCOMING_CALL
def get_call_start_date_time(self):
start_time = self.result_set.getString("start_time")
try:
return long(start_time) / 1000
except ValueError as ve:
self._had_error = True
print("bad_conversion")
return super(LineCallLogsParser, self).get_call_start_date_time()
def get_call_start_date_time(self):
try:
return long(self.result_set.getString("start_time")) / 1000
except ValueError as ve:
self._had_error = True
return super(LineCallLogsParser, self).get_call_start_date_time()
def get_call_end_date_time(self):
end_time = self.result_set.getString("end_time")
try:
return long(end_time) / 1000
except ValueError as ve:
self._had_error = True
print("bad conversion")
return super(LineCallLogsParser, self).get_call_end_date_time()
def get_call_end_date_time(self):
try:
return long(self.result_set.getString("end_time")) / 1000
except ValueError as ve:
self._had_error = True
return super(LineCallLogsParser, self).get_call_end_date_time()
def get_phone_number_to(self):
if self.get_call_direction() == self.OUTGOING_CALL:
return Account.Address(self.result_set.getString("m_id"),
self.result_set.getString("name"))
return super(LineCallLogsParser, self).get_phone_number_to()
def get_account_name(self):
return self.result_set.getString("account_name")
def get_phone_number_from(self):
if self.get_call_direction() == self.INCOMING_CALL:
return Account.Address(self.result_set.getString("m_id"),
self.result_set.getString("name"))
return super(LineCallLogsParser, self).get_phone_number_from()
def get_contact_name(self):
return self.result_set.getString("contact_name")
def get_call_type(self):
if self.result_set.getString("call_type") == self._VIDEO_CALL_TYPE:
return self.VIDEO_CALL
if self.result_set.getString("call_type") == self._AUDIO_CALL_TYPE:
return self.AUDIO_CALL
return super(LineCallLogsParser, self).get_call_type()
def has_incomplete_results(self):
return self._had_error
class LineContactsParser(TskContactsParser):
"""
@@ -193,17 +202,17 @@ class LineContactsParser(TskContactsParser):
def __init__(self, contact_db):
super(LineContactsParser, self).__init__(contact_db.runQuery(
"""
SELECT name,
SELECT m_id,
server_name
FROM contacts
"""
)
)
def get_account_name(self):
return self.result_set.getString("server_name")
return self.result_set.getString("m_id")
def get_contact_name(self):
return self.result_set.getString("name")
return self.result_set.getString("server_name")
class LineMessagesParser(TskMessagesParser):
"""
@@ -213,23 +222,45 @@ class LineMessagesParser(TskMessagesParser):
"""
def __init__(self, message_db):
super().__init__(message_db.runQuery(
"""SELECT created_time, content, contacts.server_name AS server_name, read_count
FROM chat_history
JOIN contacts
ON chat_history.from_mid = contacts.m_id"""
))
super(LineMessagesParser, self).__init__(message_db.runQuery(
"""
SELECT all_contacts.name,
all_contacts.id,
all_contacts.members,
CH.from_mid,
CH.content,
CH.created_time,
CH.attachement_type,
CH.attachement_local_uri,
CH.status
FROM (SELECT G.name,
group_members.id,
group_members.members
FROM (SELECT id,
group_concat(m_id) AS members
FROM membership
GROUP BY id) AS group_members
JOIN groups AS G
ON G.id = group_members.id
UNION
SELECT server_name,
m_id,
NULL
FROM contacts) AS all_contacts
JOIN chat_history AS CH
ON CH.chat_id = all_contacts.id
"""
)
)
self._LINE_MESSAGE_TYPE = "Line Message"
#From the limited test data, it appeared that incoming
#was only associated with a 1 status. Status # 3 and 7
#was only associated with outgoing.
self._INCOMING_MESSAGE_TYPE = 1
self._had_error = False
def get_account_id(self):
return self.result_set.getString("server_name")
def get_message_type(self):
return self.LINE_MESSAGE_TYPE
def get_phone_number_from(self):
return self.result_set("server_name")
return self._LINE_MESSAGE_TYPE
def get_message_date_time(self):
created_time = self.result_set.getString("created_time")
@@ -242,6 +273,34 @@ class LineMessagesParser(TskMessagesParser):
def get_message_text(self):
content = self.result_set.getString("content")
if not LineContentUtil.is_text_message(content):
return ""
return content
def get_message_direction(self):
if self.result_set.getInt("status") == self._INCOMING_MESSAGE_TYPE:
return self.INCOMING
return self.OUTGOING
def get_phone_number_from(self):
if self.get_message_direction() == self.INCOMING:
group = self.result_set.getString("members")
if group is None:
return Account.Address(self.result_set.getString("from_mid"),
self.result_set.getString("name"))
return Account.Address(self.result_set.getString("from_mid"),
self.result_set.getString("name"))
return super(LineMessagesParser, self).get_phone_number_from()
def get_phone_number_to(self):
if self.get_message_direction() == self.OUTGOING:
group = self.result_set.getString("members")
if group is None:
return Account.Address(self.result_set.getString("id"),
self.result_set.getString("name"))
return Account.Address(group, self.result_set.getString("name"))
return super(LineMessagesParser, self).get_phone_number_to()
def get_thread_id(self):
members = self.result_set.getString("members")
if members is not None:
return self.result_set.getString("id")
return super(LineMessagesParser, self).get_thread_id()