Skip to content

fix(googleChat): context, media and linter cleanup #1669

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
stark4n6 merged 1 commit into from
Jul 8, 2026
+123 −120
Merged

fix(googleChat): context, media and linter cleanup #1669

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
243 changes: 123 additions & 120 deletions scripts/artifacts/googleChat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,15 @@
__artifacts_v2__ = {
"get_googleChat": { # This should match the function name exactly
"google_chat": { # This should match the function name exactly
"name": "Google Chat",
"description": "Parses google chats",
"author": "@AlexisBrignoni",
"version": "0.1",
"date": "2023-09-03",
"creation_date": "2023-09-03",
"last_update_date": "2025-06-04",
"requirements": "none",
"category": "Google Chats",
"notes": "",
"paths": ('*/Documents/user_accounts/*/dynamite.db*',),
"paths": ('*/Documents/user_accounts/*/dynamite.db*',
'*/Documents/user_accounts/*/tmp/*'),
"output_types": "all", # or ["html", "tsv", "timeline", "lava"]
"data_views": {
"conversation": {
Expand All @@ -30,26 +31,37 @@
import re
from io import BytesIO

from scripts.ilapfuncs import artifact_processor
from scripts.ilapfuncs import open_sqlite_db_readonly, utf8_in_extended_ascii, media_to_html, logfunc, convert_ts_human_to_timezone_offset
from scripts.ilapfuncs import (
artifact_processor,
open_sqlite_db_readonly,
utf8_in_extended_ascii,
check_in_media,
convert_ts_human_to_timezone_offset,
)

_FLATTEN_ERRORS = (TypeError, AttributeError, KeyError, ValueError)


class Tuppsub(tuple):
pass



class ProtectedTuple(tuple):
pass


class ProtectedList(list):
pass


class ProtectedDict(dict):
pass


class ProtectedSet(set):
pass


def aa_flatten_dict_tu(
v,
listitem,
Expand All @@ -75,7 +87,6 @@ def aa_flatten_dict_tu(
# only get the keys back.
for k, v2 in v.items():
newtu = listitem + (k,) # we accumulate all keys in a tuple

# and check if there are more dicts (nested) in this dict
yield from aa_flatten_dict_tu(
v2, listitem=newtu, forbidden=forbidden, allowed=allowed
Expand All @@ -84,9 +95,7 @@ def aa_flatten_dict_tu(
v, forbidden
): # if we have an iterable without keys (list, tuple, set, frozenset) we have to enumerate them to be able to
# access the original dict values later: di['blabla'][0] instead of di['blabla']

for indi, v2 in enumerate(v):

if isinstance(v2, allowed):
yield v2, listitem
# if the value is not in our allowed data types, we have to check if it is an iterable
Expand All @@ -100,31 +109,28 @@ def aa_flatten_dict_tu(
elif isinstance(v, allowed):
# if the datatype is allowed, we yield it
yield Tuppsub((v, listitem))

# Brute force to check if we have an iterable. We have to get all iterables!
else:
try:
for indi2, v2 in enumerate(v):

try:
if isinstance(v2, allowed):
yield v2, listitem

else:
yield aa_flatten_dict_tu(
v2,
listitem=(listitem + (indi2,)),
forbidden=forbidden,
allowed=allowed,
)
except Exception: # pylint: disable=broad-exception-caught
except _FLATTEN_ERRORS:
# if there is an exception, it is probably not an iterable, so we yield it
yield v2, listitem
except Exception: # pylint: disable=broad-exception-caught
except _FLATTEN_ERRORS:
# if there is an exception, it is probably not an iterable, so we yield it
yield v, listitem


def fla_tu(
item,
walkthrough=(), # accumulate nested keys
Expand Down Expand Up @@ -163,16 +169,14 @@ def fla_tu(
allowed=allowed,
dict_variation=dict_variation,
) # if we have an iterable, we check recursively for other iterables

except Exception: # pylint: disable=broad-exception-caught

except _FLATTEN_ERRORS:
yield xaa, Tuppsub(
(walkthrough + Tuppsub((ini,)))
) # we just yield the value (value, (key1,key2,...)) because it is probably not an iterable
elif isinstance(
item, dict
): # we need to pass dicts to aa_flatten_dict_tu(), they need a special treatment, if not, we only get the keys from the dict back

): # we need to pass dicts to aa_flatten_dict_tu(), they need a special treatment,
# if not, we only get the keys from the dict back
yield from aa_flatten_dict_tu(
item, listitem=walkthrough, forbidden=forbidden, allowed=allowed
)
Expand All @@ -186,10 +190,8 @@ def fla_tu(
yield from aa_flatten_dict_tu(
dict(item), listitem=walkthrough, forbidden=forbidden, allowed=allowed
)

# isinstance(item, pd.DataFrame) maybe better?
elif "DataFrame" in str(type(item)):

yield from aa_flatten_dict_tu(
item.copy().to_dict(),
# pandas needs to be converted to dict first, if not, we only get the columns back. Copying might not be
Expand All @@ -198,15 +200,13 @@ def fla_tu(
forbidden=forbidden,
allowed=allowed,
)

# # many iterables are hard to identify using isinstance() / type(), so we have to use brute force to check if it is
# an iterable. If one iterable escapes, we are screwed!
else:
try:
for ini2, xaa in enumerate(item):
try:
if isinstance(xaa, allowed): # yield only for allowed data types

yield xaa, Tuppsub(
(walkthrough + (ini2,))
) # yields (value, (key1,key2,...)) -> always same format -> first value, then all keys in
Expand All @@ -221,31 +221,41 @@ def fla_tu(
allowed=allowed,
dict_variation=dict_variation,
)
except Exception: # pylint: disable=broad-exception-caught

except _FLATTEN_ERRORS:
yield xaa, Tuppsub(
(walkthrough + (ini2,))
) # in case of an exception, we yield (value, (key1,key2,...))
except Exception: # pylint: disable=broad-exception-caught

except _FLATTEN_ERRORS:
yield item, Tuppsub(
(walkthrough + Tuppsub(item, ))
) # in case of an exception, we yield (value, (key1,key2,...))


@artifact_processor
def get_googleChat(files_found, report_folder, seeker, _wrap_text, timezone_offset):
for file_found in files_found:
def google_chat(context):
data_headers = (('Timestamp', 'datetime'), 'Group Type', 'Conversation Name', 'Message Author', 'Message',
'Is Sent', 'Filename', ('Media', 'media'), 'Reaction', 'Reaction User', 'Account ID')
data_list = []
source_path = ''
timezone_offset = context.get_output_params().timezone_offset

for file_found in context.get_files_found():
file_found = str(file_found)

if file_found.endswith('dynamite.db'):
result = re.findall(r"([0-9A-F]{8}-[0-9A-F]{4}-[4][0-9A-F]{3}-[89AB][0-9A-F]{3}-[0-9A-F]{12}+)", file_found)
guid = result[0]
account_id = os.path.basename(os.path.dirname(file_found))

db = open_sqlite_db_readonly(file_found)
cursor = db.cursor()
cursor.execute(f'''
SELECT

if not file_found.endswith('dynamite.db'):
continue

source_path = file_found
result = re.findall(
r"([0-9A-F]{8}-[0-9A-F]{4}-[4][0-9A-F]{3}-[89AB][0-9A-F]{3}-[0-9A-F]{12}+)",
file_found)
guid = result[0]
account_id = os.path.basename(os.path.dirname(file_found))

db = open_sqlite_db_readonly(file_found)
cursor = db.cursor()
cursor.execute(f'''
SELECT
datetime(topic_messages.create_time/1000000,'unixepoch') AS "Message Time",
CASE
WHEN group_type=1 THEN "Group Message"
Expand All @@ -261,77 +271,70 @@ def get_googleChat(files_found, report_folder, seeker, _wrap_text, timezone_offs
WHEN topic_messages.creator_id = '{account_id}' THEN 1
ELSE 0
END is_sent
FROM
topic_messages
JOIN users ON users.user_id=topic_messages.creator_id
JOIN Groups ON Groups.group_id=topic_messages.group_id
ORDER BY "Message Time" ASC
''')

all_rows = cursor.fetchall()
usageentries = len(all_rows)
if usageentries > 0:
data_list = []

for row in all_rows:
thumb = ''
protobufreactions = row[5]
checkforempty = BytesIO(protobufreactions)
check = checkforempty.read(3)

if check == b'\xfe\xff\x00':
reaction = ''
reactionuser = ''
FROM
topic_messages
JOIN users ON users.user_id=topic_messages.creator_id
JOIN Groups ON Groups.group_id=topic_messages.group_id
ORDER BY "Message Time" ASC
''')

for row in cursor.fetchall():
media_ref_id = ''
protobufreactions = row[5]
checkforempty = BytesIO(protobufreactions)
check = checkforempty.read(3)

if check == b'\xfe\xff\x00':
reaction = ''
reactionuser = ''
else:
protostuff, _types = blackboxprotobuf.decode_message(protobufreactions)
reaction = (protostuff['1']['1']['1']['1']).decode()
reaction = (utf8_in_extended_ascii(reaction))[1]

reactionuser = protostuff['1'].get('2')
if reactionuser is not None:
reactionuser = protostuff['1']['2']['1']
reactionuser = (reactionuser.decode())
else:
reactionuser = ''

protobufmedia = row[6]
checkforempty = BytesIO(protobufmedia)
check = checkforempty.read(3)

if check == b'\xfe\xff\x00':
mediafilename = ''
else:
protostuff, _ = blackboxprotobuf.decode_message(protobufmedia)
aggregator = []
if isinstance(protostuff['1'], list):
nested_whatever = list(fla_tu(protostuff['1']))
for x in nested_whatever:
if isinstance(x[0], bytes):
aggregator.append(x[0].decode())
mediafilename = 'Group List'
media_ref_id = ", ".join(aggregator)
else:
checkkeyten = (protostuff['1'].get('10'))
if checkkeyten is not None:
mediafilename = (protostuff['1']['10']['3'])
mediafilename = (mediafilename.decode())
attachment = context.get_seeker().search(
'*/' + guid + '/tmp/' + mediafilename, return_on_first_hit=True)

if attachment:
media_ref_id = check_in_media(attachment, name=mediafilename)
else:
protostuff, _types = blackboxprotobuf.decode_message(protobufreactions)
reaction = (protostuff['1']['1']['1']['1']).decode()
reaction = (utf8_in_extended_ascii(reaction))[1]

_timestampofreaction = protostuff['1']['1']['4']
reactionuser = protostuff['1'].get('2')
if reactionuser is not None:
reactionuser = protostuff['1']['2']['1']
reactionuser = (reactionuser.decode())
else:
reactionuser = ''

protobufmedia = row[6]
checkforempty = BytesIO(protobufmedia)
check = checkforempty.read(3)

if check == b'\xfe\xff\x00':
mediafilename = ''
else:
protostuff, _ = blackboxprotobuf.decode_message(protobufmedia)
aggregator = []
if isinstance(protostuff['1'], list):
nested_whatever=list(fla_tu(protostuff['1']))
for x in nested_whatever:
if isinstance(x[0], bytes):
aggregator.append(x[0].decode())
mediafilename = 'Group List'
thumb = ", ".join(aggregator)
else:
checkkeyten = (protostuff['1'].get('10'))
if checkkeyten is not None:
mediafilename = (protostuff['1']['10']['3'])
mediafilename = (mediafilename.decode())
attachment = seeker.search('*/'+guid+'/tmp/'+mediafilename, return_on_first_hit=True)

if len(attachment) < 1:
thumb = ''
else:
thumb = media_to_html(attachment, (attachment,), report_folder)
else:
mediafilename = ''
timestamp = convert_ts_human_to_timezone_offset(row[0], timezone_offset)

data_list.append((timestamp, row[1], row[2], row[3], row[4], row[7], mediafilename, thumb, reaction, reactionuser, account_id))
mediafilename = thumb = reaction = reactionuser = ''

data_headers = (('Timestamp', 'datetime'), 'Group Type', 'Conversation Name', 'Message Author', 'Message',
'Is Sent', 'Filename', 'Media', 'Reaction', 'Reaction User', 'Account ID')

return data_headers, data_list, file_found

logfunc('No Google Chat data available')

timestamp = convert_ts_human_to_timezone_offset(row[0], timezone_offset)

data_list.append((
timestamp, row[1], row[2], row[3], row[4], row[7],
mediafilename, media_ref_id, reaction, reactionuser, account_id,
))

break

return data_headers, data_list, source_path
Loading