From 57ef3f587eae3bb0eb46e760e64ebdc9fbd36394 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Mon, 30 Sep 2024 14:51:42 -0500 Subject: [PATCH] Windows: Adds scheduled tasks plugin This adds a plugin 'ScheduledTasks' that can decode binary-encoded scheduled tasks from the Windows registry's SOFTWARE hive using a custom reader that extends the `io.BytesIO` class. Decoding operations are intended to be as fault tolerant as possible, swallowing exceptions and returning `None` to account for smear or missing data. Because each task can have mulitple triggers and multiple actions, a single entry is generated for each trigger + action pair. In the event that the either the actions could not be parsed or the triggers could not be parsed due to missing or smeared data, an entry will still be generated using the available information from the other registry value, since trigger and action data is stored separately. Much more information is decoded than is rendered, this was done intentionally to avoid overpopulating the TreeGrid with less pertinent data and to avoid an explosion of trigger and action-specific fields that may not apply to most other entries. --- volatility3/framework/layers/registry.py | 3 + .../plugins/windows/scheduled_tasks.py | 1796 +++++++++++++++++ volatility3/framework/renderers/conversion.py | 21 + 3 files changed, 1820 insertions(+) create mode 100644 volatility3/framework/plugins/windows/scheduled_tasks.py diff --git a/volatility3/framework/layers/registry.py b/volatility3/framework/layers/registry.py index 471d19f68..609832886 100644 --- a/volatility3/framework/layers/registry.py +++ b/volatility3/framework/layers/registry.py @@ -170,6 +170,9 @@ class RegistryHive(linear.LinearlyMappedLayer): return_list specifies whether the return result will be a single node (default) or a list of nodes from root to the current node (if return_list is true). + + Raises RegistryFormatException if an invalid structure is encountered + Raises KeyError if the key is not found """ root_node = self.get_node(self.root_cell_offset) if not root_node.vol.type_name.endswith(constants.BANG + "_CM_KEY_NODE"): diff --git a/volatility3/framework/plugins/windows/scheduled_tasks.py b/volatility3/framework/plugins/windows/scheduled_tasks.py new file mode 100644 index 000000000..6d3ee586d --- /dev/null +++ b/volatility3/framework/plugins/windows/scheduled_tasks.py @@ -0,0 +1,1796 @@ +# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# + +import base64 +import binascii +import dataclasses +import datetime +import enum +import io +import itertools +import logging +import struct +import traceback +import unittest +from typing import Dict, Iterator, List, Optional, Tuple, Union + +from volatility3.framework import exceptions, interfaces, renderers +from volatility3.framework.configuration import requirements +from volatility3.framework.layers import registry +from volatility3.framework.renderers import conversion +from volatility3.framework.symbols.windows.extensions import registry as reg_extensions +from volatility3.plugins import timeliner +from volatility3.plugins.windows.registry import hivelist + +vollog = logging.getLogger(__name__) + +# Reference: https://cyber.wtf/2022/06/01/windows-registry-analysis-todays-episode-tasks/ + + +class TimeMode(enum.Enum): + """ + Enumeration containing the different time modes that a 'Time' trigger can be configured to run in. + """ + + Once = "Once" + + # run at and repeat every days + Daily = "Daily" + + # run on days of week <(data2 as day_of_week bitmap)> every weeks starting at + Weekly = "Weekly" + + # run in months <(data3 as months bitmap> on days <(data2:data1 as day in month bitmap)> + # starting at + DaysInMonths = "Days In Months" + + # run in months <(data3 as months bitmap> in weeks <(data2 as week bitmap)> + # on days <(data1 as day_of_week bitmap)> starting at + DaysInWeeksInMonths = "Days In Weeks in Months" + + Unknown = "Unknown" + + +class ActionType(enum.Enum): + """ + Enumeration that maps action types to their magic number encodings + """ + + Exe = 0x6666 + ComHandler = 0x7777 + Email = 0x8888 + MessageBox = 0x9999 + + +class TriggerType(enum.Enum): + """ + Enumeration that maps trigger types to their magic number encodings + """ + + WindowsNotificationFacility = 0x6666 + Session = 0x7777 + Registration = 0x8888 + Logon = 0xAAAA + Event = 0xCCCC + Time = 0xDDDD + Idle = 0xEEEE + Boot = 0xFFFF + + +class Weekday(enum.Enum): + """ + Enumeration that contains bitwise values for days of the week. + """ + + Sunday = 0x1 + Monday = 0x2 + Tuesday = 0x4 + Wednesday = 0x8 + Thursday = 0x10 + Friday = 0x20 + Saturday = 0x40 + + +class Months(enum.Enum): + """ + Enumeration that contains bitwise values for months of the year. + """ + + January = 0x1 + February = 0x2 + March = 0x4 + April = 0x8 + May = 0x10 + June = 0x20 + July = 0x40 + August = 0x80 + September = 0x100 + October = 0x200 + November = 0x400 + December = 0x800 + + +class SidType(enum.Enum): + """ + Enumeration that maps SID types to their encoded integer values + """ + + User = 1 + Group = 2 + Domain = 3 + Alias = 4 + WellKnownGroup = 5 + DeletedAccount = 6 + Invalid = 7 + Unknown = 8 + Computer = 9 + Label = 10 + LogonSession = 11 + + +@dataclasses.dataclass +class TaskSchedulerTime: + """ + A class containing datetime information about when a task will run + """ + + is_localized: bool + filetime: Optional[datetime.datetime] + + +@dataclasses.dataclass +class TaskSchedulerTimePeriod: + """ + Class containing information delimiting time periods within scheduled tasks. + """ + + years: int + months: int + weeks: int + days: int + hours: int + minutes: int + seconds: int + + +JOB_BUCKET_FLAGS = { + 0x2: "Run only if idle", + 0x4: "Restart on idle", + 0x8: "Stop on idle end", + 0x10: "Disallow start if on batteries", + 0x20: "Stop if going on batteries", + 0x40: "Start when available", + 0x80: "Run only if network available", + 0x100: "Allow start on demand", + 0x200: "Wake to run", + 0x400: "Execute parallel", + 0x800: "Execute stop existing", + 0x1000: "Execute queue", + 0x2000: "Execute ignore new", + 0x4000: "Logon type s4u", + 0x10000: "Logon type InteractiveToken", + 0x40000: "Logon type Password", + 0x80000: "Logon type InteractiveTokenOrPassword", + 0x400000: "Enabled", + 0x800000: "Hidden", + 0x1000000: "Runlevel highest available", + 0x2000000: "Task", + 0x4000000: "Version", + 0x8000000: "Token SID type none", + 0x10000000: "Token SID type unrestricted", + 0x20000000: "Interval", + 0x40000000: "Allow hard terminate", +} + +NULL = "\u0000" + + +class _ScheduledTasksReader(io.BytesIO): + + def read_task_scheduler_time(self) -> Optional[Tuple[bool, datetime.datetime]]: + is_localized = bool(self.read_aligned_u1()) + filetime = self.decode_filetime() + if filetime is None: + return None + return is_localized, filetime + + def read_filetime(self) -> Optional[datetime.datetime]: + return datetime.datetime.now() + + def seek_relative(self, offset) -> int: + return self.seek(offset, io.SEEK_CUR) + + def read_bool(self, aligned=False) -> Optional[bool]: + try: + val = struct.unpack("?", self.read(1))[0] + if aligned: + self.seek(7) + return val + except struct.error: + return None + + def decode_filetime(self) -> Optional[datetime.datetime]: + filetime = self.read_u8() + if filetime is None: + return None + + if filetime == 0 or filetime == 0xFFFFFFFFFFFFFFFF: + return None + filetime = conversion.wintime_to_datetime(filetime) + if isinstance(filetime, datetime.datetime): + return filetime + else: + return None + + def _read_uint( + self, size: int, format: str, aligned: bool = False + ) -> Optional[int]: + try: + val = struct.unpack(format, self.read(size))[0] + if aligned: + self.seek(8 - size, io.SEEK_CUR) + return val + except struct.error: + return None + + def read_aligned_u1(self) -> Optional[int]: + return self._read_uint(1, "B", True) + + def read_u2(self) -> Optional[int]: + return self._read_uint(2, " Optional[int]: + return self._read_uint(2, " Optional[int]: + return self._read_uint(4, " Optional[int]: + return self._read_uint(8, " Optional[int]: + return self._read_uint(4, " Optional[bytes]: + count = self.read_u4() if not aligned else self.read_aligned_u4() + if count is None: + return None + data = self.read(count) + if aligned: + self.seek((8 - (count % 8)) % 8, io.SEEK_CUR) + return data + + def read_bstring(self, aligned=False) -> Optional[str]: + size = self.read_u4() if not aligned else self.read_aligned_u4() + if size is None: + return None + try: + raw = self.read(size) + val = raw.decode("utf-16le", errors="replace").rstrip(NULL) or None + except UnicodeDecodeError: + val = None + + if aligned: + self.seek((8 - (size % 8)) % 8, io.SEEK_CUR) + + return val + + def read_aligned_bstring_expand_sz(self) -> Optional[str]: + # type: () -> Optional[str] + sz = self.read_aligned_u4() + if sz is None: + return None + byte_count = sz * 2 + 2 + + if sz == 0: + return None + + try: + content = self.read(byte_count).decode("utf-16le") + except UnicodeDecodeError: + content = None + + self.seek((8 - (byte_count % 8)) % 8, io.SEEK_CUR) + return content.rstrip("\x00") if content is not None else None + + def read_tstimeperiod(self) -> Optional[TaskSchedulerTimePeriod]: + values = ( + self.read_u2(), + self.read_u2(), + self.read_u2(), + self.read_u2(), + self.read_u2(), + self.read_u2(), + self.read_u2(), + ) + + if any(value is None for value in values): + return None + + return TaskSchedulerTimePeriod(*values) + + +def _build_guid_name_map(key: reg_extensions.CM_KEY_NODE) -> Dict[str, str]: + mapping = {} + task_id_value = None + for value in key.get_values(): + try: + if value.get_name() == "Id": + task_id_value = value + break + except exceptions.InvalidAddressException: + continue + + if ( + task_id_value is not None + and task_id_value.get_type() == reg_extensions.RegValueTypes.REG_SZ + ): + try: + id_str = task_id_value.decode_data() + except exceptions.InvalidAddressException: + id_str = None + + if isinstance(id_str, bytes): + mapping[id_str.decode("utf-16le", errors="replace").rstrip(NULL)] = str( + key.get_name() + ) + + for subkey in key.get_subkeys(): + mapping.update(_build_guid_name_map(subkey)) + return mapping + + +@dataclasses.dataclass +class TaskAction: + action_type: ActionType + action: str + action_args: Optional[str] + working_directory: Optional[str] + + @classmethod + def decode_messagebox_action( + cls, reader: _ScheduledTasksReader + ) -> Optional["TaskAction"]: + caption, content = reader.read_bstring(), reader.read_bstring() + return cls( + ActionType.MessageBox, + f'"{caption or ""}": {content or ""}', + None, + None, + ) + + @classmethod + def _decode_exe_action( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskAction"]: + command = reader.read_bstring() + args = reader.read_bstring() + if command is None or args is None: + return None + + workdir = reader.read_bstring() + if version == 3: + _flags = reader.read_u2() + + return cls(ActionType.Exe, command, args, workdir) + + @classmethod + def _decode_email_action( + cls, reader: _ScheduledTasksReader + ) -> Optional["TaskAction"]: + props = { + "From": reader.read_bstring(), + "To": reader.read_bstring(), + "Cc": reader.read_bstring(), + "Bcc": reader.read_bstring(), + "Reply_to": reader.read_bstring(), + "Server": reader.read_bstring(), + "Subject": reader.read_bstring(), + "Body": reader.read_bstring(), + } + + num_attachment_filenames = reader.read_u4() + if num_attachment_filenames is not None: + + attachment_filenames = [ + reader.read_bstring() for _ in range(num_attachment_filenames) + ] + + props["Attachments"] = ( + "<" + + ", ".join( + filename + for filename in attachment_filenames + if filename is not None + ) + + ">" + ) + + num_headers = reader.read_u4() + if num_headers is not None: + headers = [ + (reader.read_bstring(), reader.read_bstring()) + for _ in range(num_headers) + ] + + props["Headers"] = ( + "<" + + ", ".join( + f"{field}: {value}" + for field, value in headers + if field is not None and value is not None + ) + + ">" + ) + + cls( + ActionType.Email, + ", ".join( + f"{key}: {value}" for key, value in props.items() if value is not None + ), + None, + None, + ) + + @classmethod + def _decode_comhandler_action( + cls, reader: _ScheduledTasksReader + ) -> Optional["TaskAction"]: + guid_raw = reader.read(16) + if not guid_raw and len(guid_raw) == 16: + return None + clsid = conversion.windows_bytes_to_guid(guid_raw) + args = reader.read_bstring() + + return cls(ActionType.ComHandler, clsid, args, None) + + +@dataclasses.dataclass +class _ScheduledTaskEntry: + name: Union[str, interfaces.renderers.BaseAbsentValue] + principal_id: Union[str, interfaces.renderers.BaseAbsentValue] + display_name: Union[str, interfaces.renderers.BaseAbsentValue] + enabled: Union[bool, interfaces.renderers.BaseAbsentValue] + creation_time: Union[datetime.datetime, interfaces.renderers.BaseAbsentValue] + last_run_time: Union[datetime.datetime, interfaces.renderers.BaseAbsentValue] + last_successful_run_time: Union[ + datetime.datetime, interfaces.renderers.BaseAbsentValue + ] + trigger_type: Union[str, interfaces.renderers.BaseAbsentValue] + trigger_description: Union[str, interfaces.renderers.BaseAbsentValue] + action_type: Union[str, interfaces.renderers.BaseAbsentValue] + action_description: Union[str, interfaces.renderers.BaseAbsentValue] + action_args: Union[str, interfaces.renderers.BaseAbsentValue] + action_context: Union[str, interfaces.renderers.BaseAbsentValue] + working_directory: Union[str, interfaces.renderers.BaseAbsentValue] + guid: str + + +@dataclasses.dataclass +class _JobSchedule: + start_boundary: Optional[datetime.datetime] + end_boundary: Optional[datetime.datetime] + repetition_interval_secs: Optional[int] + repetition_duration_secs: Optional[int] + execution_time_limit_secs: Optional[int] + mode: Optional[TimeMode] + data1: Optional[int] + data2: Optional[int] + data3: Optional[int] + stop_tasks_at_duration_end: Optional[int] + is_enabled: Optional[bool] + max_delay_seconds: Optional[int] + + def get_description(self) -> Optional[str]: + if self.mode == TimeMode.Once: + return "Run one time starting at {}".format( + self.start_boundary.isoformat() + if self.start_boundary is not None + else "" + ) + + elif self.mode == TimeMode.Daily: + if self.data1 is None: + return None + return "Run at {} and repeat every {} days".format( + ( + self.start_boundary.isoformat() + if self.start_boundary is not None + else "" + ), + self.data1, + ) + + elif self.mode == TimeMode.Weekly: + if self.data2 is None: + return None + + days = [k.name for k in Weekday if k.value & self.data2] + return "Run on {} every {} weeks starting at {}".format( + ", ".join(days), + self.data1, + ( + self.start_boundary.isoformat() + if self.start_boundary is not None + else "" + ), + ) + elif self.mode == TimeMode.DaysInMonths: + if self.data2 is None or self.data1 is None or self.data3 is None: + return None + months = [month.name for month in Months if month.value & self.data3] + days_bitmap = (self.data2 << 16) + self.data1 + days = [str(v + 1) for v in range(31) if (1 << v) & days_bitmap] + return "Run in months {} on days {} starting at {}".format( + ", ".join(months), + ", ".join(days), + ( + self.start_boundary.isoformat() + if self.start_boundary is not None + else "" + ), + ) + elif self.mode == TimeMode.DaysInWeeksInMonths: + if self.data1 is None or self.data2 is None or self.data3 is None: + return None + + months = [month.name for month in Months if month.value & self.data3] + weeks = [str(v + 1) for v in range(5) if (v << 1) & self.data2] + days = [day.name for day in Weekday if day.value & self.data1] + return "Run in months {} in weeks {} on days {} starting at {}".format( + ", ".join(months), + ", ".join(weeks), + ", ".join(days), + ( + self.start_boundary.isoformat() + if self.start_boundary is not None + else "" + ), + ) + else: + return None + + @classmethod + def decode(cls, reader: _ScheduledTasksReader) -> Optional["_JobSchedule"]: + start_boundary = reader.read_task_scheduler_time() + end_boundary = reader.read_task_scheduler_time() + + _ = reader.read_task_scheduler_time() + repetition_interval_secs = reader.read_u4() + repetition_duration_secs = reader.read_u4() + execution_time_limit_secs = reader.read_u4() + mode_index = reader.read_u4() + try: + if mode_index is not None: + mode = list(TimeMode)[mode_index] + else: + mode = None + except IndexError: + mode = TimeMode.Unknown + data1 = reader.read_u2() + data2 = reader.read_u2() + data3 = reader.read_u2() + + reader.seek(2, io.SEEK_CUR) # pad + stop_tasks_at_duration_end = reader.read_bool() + is_enabled = reader.read_bool() + reader.seek(6, io.SEEK_CUR) # pad (2) + unknown (4) + max_delay_seconds = reader.read_u4() + reader.seek(4, io.SEEK_CUR) # pad + + return cls( + start_boundary[1] if start_boundary is not None else None, + end_boundary[1] if end_boundary is not None else None, + repetition_interval_secs, + repetition_duration_secs, + execution_time_limit_secs, + mode, + data1, + data2, + data3, + stop_tasks_at_duration_end, + is_enabled, + max_delay_seconds, + ) + + +def decode_sid(data: bytes) -> Optional[str]: + """ + Decodes a windows SID from variable-length raw bytes + + Returns the string representation of the SID if decoding was successful, or None + if the data could not be parsed due to an insufficent number of bytes. + """ + try: + revision, subid_count, id_authority = struct.unpack( + ">BBQ", data[:2] + b"\x00\x00" + data[2:8] + ) + subauthorities = struct.unpack( + "<" + "I" * subid_count, data[8 : 8 + subid_count * 4] + ) + sid_string = "S-" + "-".join( + [str(item) for item in [revision, id_authority] + list(subauthorities)] + ) + except struct.error: + return None + + return sid_string + + +@dataclasses.dataclass +class UserInfo: + sid_type: Optional[SidType] + sid: Optional[str] + username: Optional[str] + + @classmethod + def _decode(cls, reader: _ScheduledTasksReader) -> Optional["UserInfo"]: + skip_user = reader.read_aligned_u1() != 0 + if not skip_user: + skip_sid = reader.read_aligned_u1() != 0 + else: + skip_sid = None + + sid_type = None + sid = None + if not skip_user and not skip_sid: + try: + sid_type = SidType(reader.read_aligned_u4()) + except ValueError: + sid_type = SidType.Unknown + + sid_raw = reader.read_buffer(aligned=True) + if sid_raw is None: + return None + sid = decode_sid(sid_raw) + + username = reader.read_bstring(aligned=True) if not skip_user else None + + return UserInfo(sid_type, sid, username) + + +@dataclasses.dataclass +class OptionalSettings: + IdleDurationSeconds: int + idleWaitTimeoutSeconds: int + ExecutionTimeLimitSeconds: int + DeleteExpiredTaskAfter: int + Priority: int + RestartOnFailureDelay: int + RestartOnFailureRetries: int + NetworkId: bytes + Privileges: Optional[List[str]] + Periodicity: Optional[TaskSchedulerTimePeriod] + Deadline: Optional[TaskSchedulerTimePeriod] + Exclusive: Optional[bool] + + @classmethod + def _decode(cls, reader: _ScheduledTasksReader) -> Optional["OptionalSettings"]: + LEN_WITH_PRIVILEGES = 0x38 + LEN_WITH_TIME_PERIODS = 0x58 + length = reader.read_aligned_u4() + if length == 0: + return None + + base_values = ( + reader.read_u4(), + reader.read_u4(), + reader.read_u4(), + reader.read_u4(), + reader.read_u4(), + reader.read_u4(), + reader.read_u4(), + binascii.hexlify(reader.read(16)), + ) + + if any(value is None for value in base_values): + return None + + reader.seek(4, io.SEEK_CUR) # padding + + privileges = None + periodicity = None + deadline = None + exclusive = None + if length == LEN_WITH_PRIVILEGES or length == LEN_WITH_TIME_PERIODS: + privileges_raw = reader.read_u8() + if privileges_raw is None: + return None + privileges = [ + priv.name for priv in Privileges if priv.value & privileges_raw + ] + if length == LEN_WITH_TIME_PERIODS: + periodicity = reader.read_tstimeperiod() + deadline = reader.read_tstimeperiod() + exclusive = reader.read_bool() + reader.seek(3, io.SEEK_CUR) # padding + + return OptionalSettings( + *base_values, privileges, periodicity, deadline, exclusive + ) + + +@dataclasses.dataclass +class JobBucket: + flags: List[str] + crc32: int + principal_id: Optional[str] + display_name: Optional[str] + user_info: Optional[UserInfo] + optional_settings: Optional[OptionalSettings] + + @classmethod + def _decode( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["JobBucket"]: + flags_raw = reader.read_aligned_u4() + if flags_raw is None: + return None + flags = [y for x, y in JOB_BUCKET_FLAGS.items() if x & flags_raw] + crc32 = reader.read_aligned_u4() + if crc32 is None: + return None + + principal_id = None + display_name = None + if version >= 0x16: + principal_id = reader.read_bstring(aligned=True) + if version >= 0x17: + display_name = reader.read_bstring(aligned=True) + + user_info = UserInfo._decode(reader) + optional_settings = OptionalSettings._decode(reader) + + return JobBucket( + flags, crc32, principal_id, display_name, user_info, optional_settings + ) + + +class Privileges(enum.Enum): + SeCreateTokenPrivilege = 0x4 + SeAssignPrimaryTokenPrivilege = 0x8 + SeLockMemoryPrivilege = 0x10 + SeIncreaseQuotaPrivilege = 0x20 + SeMachineAccountPrivilege = 0x40 + SeTcbPrivilege = 0x80 + SeSecurityPrivilege = 0x100 + SeTakeOwnershipPrivilege = 0x200 + SeLoadDriverPrivilege = 0x400 + SeSystemProfilePrivilege = 0x800 + SeSystemtimePrivilege = 0x1000 + SeProfileSingleProcessPrivilege = 0x2000 + SeIncreaseBasePriorityPrivilege = 0x4000 + SeCreatePagefilePrivilege = 0x8000 + SeCreatePermanentPrivilege = 0x10000 + SeBackupPrivilege = 0x20000 + SeRestorePrivilege = 0x40000 + SeShutdownPrivilege = 0x80000 + SeDebugPrivilege = 0x100000 + SeAuditPrivilege = 0x200000 + SeSystemEnvironmentPrivilege = 0x400000 + SeChangeNotifyPrivilege = 0x800000 + SeRemoteShutdownPrivilege = 0x1000000 + SeUndockPrivilege = 0x2000000 + SeSyncAgentPrivilege = 0x4000000 + SeEnableDelegationPrivilege = 0x8000000 + SeManageVolumePrivilege = 0x10000000 + SeImpersonatePrivilege = 0x20000000 + SeCreateGlobalPrivilege = 0x40000000 + SeTrustedCredManAccessPrivilege = 0x80000000 + SeRelabelPrivilege = 0x100000000 + SeIncreaseWorkingSetPrivilege = 0x200000000 + SeTimeZonePrivilege = 0x400000000 + SeCreateSymbolicLinkPrivilege = 0x800000000 + SeDelegateSessionUserImpersonatePrivilege = 0x1000000000 + + +class SessionState(enum.Enum): + ConsoleConnect = 1 + ConsoleDisconnect = 2 + RemoteConnect = 3 + RemoteDisconnect = 4 + SessionLock = 5 + SessionUnlock = 6 + Unknown = "Unknown" + + +@dataclasses.dataclass +class TaskTrigger: + start_boundary: Optional[datetime.datetime] + end_boundary: Optional[datetime.datetime] + repetition_interval_seconds: Optional[int] + enabled: Optional[bool] + trigger_type: TriggerType + description: Optional[str] + + @classmethod + def _decode_generic_trigger( + cls, reader: _ScheduledTasksReader, version: int, trigger_type: TriggerType + ) -> Optional["TaskTrigger"]: + start_boundary = reader.read_task_scheduler_time() + end_boundary = reader.read_task_scheduler_time() + + _ = reader.read_u4() # delay seconds + _ = reader.read_u4() # timeout seconds + + repetition_interval_secs = reader.read_u4() + _ = reader.read_u4() # reptition duration seconds + _ = reader.read_u4() # repetition duration seconds 2 + + _ = reader.read_bool() # stop at duration end + reader.seek(3, io.SEEK_CUR) + trigger_enabled = bool(reader.read_aligned_u1()) + reader.seek(8, io.SEEK_CUR) # unknown field + + if version >= 0x16: + cur = reader.tell() + _ = reader.read_bstring() # trigger id + reader.seek((8 - (reader.tell() - cur)) % 8, io.SEEK_CUR) # pad to block + + return cls( + start_boundary[1] if start_boundary is not None else None, + end_boundary[1] if end_boundary is not None else None, + repetition_interval_secs, + trigger_enabled, + trigger_type, + f"{trigger_type.name} trigger", + ) + + @classmethod + def _decode_logon_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + base = cls._decode_generic_trigger(reader, version, TriggerType.Logon) + if base is None: + return None + + user = UserInfo._decode(reader) + if user is not None and user.username is not None: + base.description = f"{user.username}: {user.sid} ({user.sid_type})" + + return base + + @classmethod + def _decode_session_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + base = cls._decode_generic_trigger(reader, version, TriggerType.Session) + if base is None: + return None + session_type_raw = reader.read_u4() + reader.seek(4, io.SEEK_CUR) + + try: + session_type = SessionState(session_type_raw) + except ValueError: + session_type = SessionState.Unknown + + user_info = UserInfo._decode(reader) + if user_info is not None and user_info.username is not None: + base.description = f"{session_type.name} for user {user_info.username}" + else: + base.description = session_type.name + + return base + + @classmethod + def _decode_time_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + job_schedule = _JobSchedule.decode(reader) + if job_schedule is None: + return None + + if version >= 0x16: + cur = reader.tell() + _ = reader.read_bstring() # trigger id + reader.seek((8 - (reader.tell() - cur)) % 8, io.SEEK_CUR) # pad to block + + return cls( + job_schedule.start_boundary, + job_schedule.end_boundary, + job_schedule.repetition_interval_secs, + job_schedule.is_enabled, + TriggerType.Time, + job_schedule.get_description() or None, + ) + + @classmethod + def _decode_event_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + base = cls._decode_generic_trigger(reader, version, TriggerType.Event) + if base is None: + return base + + subscription = reader.read_aligned_bstring_expand_sz() + reader.seek(8, io.SEEK_CUR) # 2 4-byte unknown fields + reader.read_aligned_bstring_expand_sz() # another unknown field + len_value_queries = reader.read_aligned_u4() + + if len_value_queries is None: + return base + + queries = [ + ( + reader.read_aligned_bstring_expand_sz(), + reader.read_aligned_bstring_expand_sz(), + ) + for _ in range(len_value_queries) + ] + valid = [(k, v) for (k, v) in queries if k is not None and v is not None] + if base.description is None: + base.description = "Event Trigger" + base.description += f": Subscription: {subscription}, Queries: {str(valid)}" + return base + + @classmethod + def _decode_boot_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + return cls._decode_generic_trigger(reader, version, TriggerType.Boot) + + @classmethod + def _decode_wnf_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + base = cls._decode_generic_trigger( + reader, version, TriggerType.WindowsNotificationFacility + ) + if base is None: + return None + + state_name = binascii.hexlify(reader.read(8)).decode("ascii") + datalen = reader.read_aligned_u4() + _ = base64.b64encode(reader.read(datalen)) # state binary data + base.description = f"WNF state {state_name}" + return base + + @classmethod + def _decode_idle_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + return cls._decode_generic_trigger(reader, version, TriggerType.Logon) + + @classmethod + def _decode_registration_trigger( + cls, reader: _ScheduledTasksReader, version: int + ) -> Optional["TaskTrigger"]: + return cls._decode_generic_trigger(reader, version, TriggerType.Logon) + + +@dataclasses.dataclass +class TriggerSet: + job_bucket: JobBucket + triggers: List[TaskTrigger] + + @classmethod + def decode(cls, data) -> Optional["TriggerSet"]: + reader = _ScheduledTasksReader(data) + + version = reader.read_aligned_u1() + _ = reader.read_task_scheduler_time() # start boundary + _ = reader.read_task_scheduler_time() # end_boundary + + if version is None: + return None + + job_bucket = JobBucket._decode(reader, version) + if job_bucket is None: + return None + + triggers = [] + + while True: + magic = reader.read_aligned_u4() + if magic is None: + break + try: + trigger_type = TriggerType(magic) + except ValueError: + vollog.warning(f"Invalid trigger magic {hex(magic)}") + break + + if trigger_type == TriggerType.Logon: + trigger = TaskTrigger._decode_logon_trigger(reader, version) + elif trigger_type == TriggerType.Session: + trigger = TaskTrigger._decode_session_trigger(reader, version) + elif trigger_type == TriggerType.WindowsNotificationFacility: + trigger = TaskTrigger._decode_wnf_trigger(reader, version) + elif trigger_type == TriggerType.Boot: + trigger = TaskTrigger._decode_boot_trigger(reader, version) + elif trigger_type == TriggerType.Registration: + trigger = TaskTrigger._decode_registration_trigger(reader, version) + elif trigger_type == TriggerType.Event: + trigger = TaskTrigger._decode_event_trigger(reader, version) + elif trigger_type == TriggerType.Idle: + trigger = TaskTrigger._decode_idle_trigger(reader, version) + elif trigger_type == TriggerType.Time: + trigger = TaskTrigger._decode_time_trigger(reader, version) + else: + vollog.warning( + f"Invalid trigger magic {hex(magic)} encountered at offset {hex(reader.tell() - 8)}, stopping parsing" + ) + break + triggers.append(trigger) + + return cls(job_bucket, triggers) + + +@dataclasses.dataclass +class ActionSet: + actions: List[TaskAction] + context: Optional[str] + + @classmethod + def decode(cls, data: bytes) -> Optional["ActionSet"]: + reader = _ScheduledTasksReader(data) + actions = [] + + version = reader.read_u2() + if version is None: + return None + + if version in [2, 3]: + action_context = reader.read_bstring() + else: + action_context = None + + while True: + magic = reader.read_u2() + if magic is None: + break + + _ = ( + reader.read_bstring() + ) # action identifier, usually (but not always) empty + + if magic == ActionType.Email.value: + action = TaskAction._decode_email_action(reader) + elif magic == ActionType.Exe.value: + action = TaskAction._decode_exe_action(reader, version) + elif magic == ActionType.ComHandler.value: + action = TaskAction._decode_comhandler_action(reader) + elif magic == ActionType.MessageBox.value: + action = TaskAction.decode_messagebox_action(reader) + else: + break + actions.append(action) + + return cls(actions, action_context) + + +@dataclasses.dataclass +class DynamicInfo: + """ + Contains information about execution history for this task, + including timestamps and the last error code + """ + + creation_time: Optional[datetime.datetime] + last_run_time: Optional[datetime.datetime] + last_successful_run_time: Optional[datetime.datetime] + last_error_code: Optional[int] + + @classmethod + def decode(cls, data: bytes) -> Optional["DynamicInfo"]: + """ + Decodes a DynamicInfo structure from RegBin value data. + Raises a `ScheduledTaskDecodingError` if the magic bytes are invalid, but otherwise + attempts to decode as much as possible without returning an error. + """ + DYNAMICINFO_MAGIC = 3 + + reader = _ScheduledTasksReader(data) + magic = reader.read_u4() + if magic != DYNAMICINFO_MAGIC: + return None + + creation_time = reader.decode_filetime() + last_run_time = reader.decode_filetime() + + reader.seek(4, io.SEEK_CUR) # deprecated field 'TaskState' + + last_error_code = reader.read_u4() + last_success_time = reader.decode_filetime() + + vollog.debug((creation_time, last_run_time, last_success_time)) + + return cls( + last_run_time, + creation_time, + last_success_time, + last_error_code, + ) + + +class ScheduledTasks(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): + """Decodes scheduled task information from the Windows registry, including + information about triggers, actions, run times, and creation times. + """ + + _required_framework_version = (2, 11, 0) + _version = (1, 0, 0) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + # Since we're calling the plugin, make sure we have the plugin's requirements + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel33", "Intel64"], + ), + requirements.PluginRequirement( + name="hivelist", plugin=hivelist.HiveList, version=(1, 0, 0) + ), + ] + + def generate_timeline( + self, + ) -> Iterator[Tuple[str, timeliner.TimeLinerType, datetime.datetime]]: + for _, task in self._generator(): + if isinstance(task.last_run_time, datetime.datetime): + yield f"ScheduledTasks: task action {task.action_description} with trigger {task.trigger_description} ran", timeliner.TimeLinerType.ACCESSED, task.last_run_time + if isinstance(task.last_successful_run_time, datetime.datetime): + yield f"ScheduledTasks: task action {task.action_description} with trigger {task.trigger_description} ran successfully", timeliner.TimeLinerType.ACCESSED, task.last_successful_run_time + if isinstance(task.creation_time, datetime.datetime): + yield f"ScheduledTasks: Creation Time for task {task.guid} with trigger {task.trigger_description or ''}", timeliner.TimeLinerType.CREATED, task.creation_time + + @classmethod + def get_software_hive( + cls, + context: interfaces.context.ContextInterface, + config_path: str, + kernel: interfaces.context.ModuleInterface, + ) -> Optional[registry.RegistryHive]: + """Retrieves the `Amcache.hve` registry hive from the kernel module, if it can be located.""" + return next( + hivelist.HiveList.list_hives( + context=context, + base_config_path=interfaces.configuration.path_join( + config_path, "hivelist" + ), + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, + filter_string="SOFTWARE", + ), + None, + ) + + @classmethod + def parse_actions_value( + cls, actions_value: reg_extensions.CM_KEY_VALUE + ) -> Optional[ActionSet]: + """Parses File entries from the Windows 8 `Root\\File` key. + + :param programs_key: The `Root\\File` registry key. + + :return: An iterator of tuples, where the first member is the program ID string for + correlating `Root\\Program` entries, and the second member is the `AmcacheEntry`. + """ + try: + data = actions_value.decode_data() + except exceptions.InvalidAddressException: + data = None + + if not isinstance(data, bytes): + return None + + return ActionSet.decode(data) + + @classmethod + def parse_triggers_value( + cls, triggers_value: reg_extensions.CM_KEY_VALUE + ) -> Optional[TriggerSet]: + try: + data = triggers_value.decode_data() + except exceptions.InvalidAddressException: + data = None + + if not isinstance(data, bytes): + return None + + return TriggerSet.decode(data) + + @classmethod + def parse_dynamic_info_value( + cls, dyn_info_value: reg_extensions.CM_KEY_VALUE + ) -> Optional[DynamicInfo]: + + try: + data = dyn_info_value.decode_data() + except exceptions.InvalidAddressException: + data = None + + if not isinstance(data, bytes): + return None + + return DynamicInfo.decode(data) + + @classmethod + def _get_task_keys( + cls, software_hive: reg_extensions.RegistryHive + ) -> Tuple[ + Optional[reg_extensions.CM_KEY_NODE], Optional[reg_extensions.CM_KEY_NODE] + ]: + try: + task_key = software_hive.get_key( + "Microsoft\\Windows NT\\CurrentVersion\\Schedule\\TaskCache\\Tasks" + ) + except (KeyError, registry.RegistryFormatException): + task_key = None + + try: + task_tree = software_hive.get_key( + "Microsoft\\Windows NT\\CurrentVersion\\Schedule\\TaskCache\\Tree" + ) + except (KeyError, registry.RegistryFormatException): + task_tree = None + + return (task_key, task_tree) # type: ignore + + @classmethod + def _parse_task_key( + cls, key: reg_extensions.CM_KEY_NODE, guid_mapping: Dict[str, str] + ) -> Iterator[_ScheduledTaskEntry]: + values = {} + for value in key.get_values(): + try: + name = str(value.get_name()) + except exceptions.InvalidAddressException: + continue + + if name in ["Actions", "Triggers", "DynamicInfo"]: + values[name] = value + + task_name = guid_mapping.get(str(key.get_name()), renderers.NotAvailableValue()) + + try: + action_set = cls.parse_actions_value(values["Actions"]) + except KeyError: + vollog.debug("Failed to get Actions value") + action_set = None + + try: + triggers_value = values["Triggers"] + trigger_set = cls.parse_triggers_value(triggers_value) + except KeyError: + vollog.debug("Failed to get Triggers value") + trigger_set = None + + if trigger_set is not None: + vollog.debug("Parsed triggers successfully") + + principal_id = ( + trigger_set.job_bucket.principal_id or renderers.NotAvailableValue() + ) + display_name = ( + trigger_set.job_bucket.display_name or renderers.NotAvailableValue() + ) + else: + vollog.debug("Failed to parse triggers") + + principal_id = renderers.NotAvailableValue() + display_name = renderers.NotAvailableValue() + + try: + dynamic_info = cls.parse_dynamic_info_value(values["DynamicInfo"]) + except KeyError: + vollog.debug("DynamicInfo value not found") + dynamic_info = None + + vollog.debug(dynamic_info) + + creation_time = dynamic_info.creation_time if dynamic_info is not None else None + last_run_time = dynamic_info.last_run_time if dynamic_info is not None else None + last_successful_run_time = ( + dynamic_info.last_successful_run_time if dynamic_info is not None else None + ) + + all_triggers = ( + trigger_set.triggers or [None] if trigger_set is not None else [None] + ) + + all_actions = action_set.actions or [None] if action_set is not None else [None] + + for action, trigger in itertools.product(all_actions, all_triggers): + + if action is not None: + if action.action_type in ( + ActionType.Exe, + ActionType.ComHandler, + ): + if action.action_args is None: + args = renderers.NotAvailableValue() + else: + args = action.action_args + else: + args = renderers.NotApplicableValue() + + if action.action_type == ActionType.Exe: + working_directory = ( + action.working_directory or renderers.NotAvailableValue() + ) + else: + working_directory = renderers.NotApplicableValue() + + else: + args = renderers.NotAvailableValue() + working_directory = renderers.NotAvailableValue() + + if trigger is not None and trigger.enabled is not None: + enabled = trigger.enabled + else: + enabled = renderers.NotAvailableValue() + + yield _ScheduledTaskEntry( + task_name, + principal_id, + display_name, + enabled, + creation_time or renderers.NotAvailableValue(), + last_run_time or renderers.NotAvailableValue(), + last_successful_run_time or renderers.NotAvailableValue(), + ( + trigger.trigger_type.name + if trigger is not None + else renderers.NotAvailableValue() + ), + ( + trigger.description or renderers.NotAvailableValue() + if trigger is not None + else renderers.NotAvailableValue() + ), + ( + action.action_type.name + if action is not None + else renderers.NotAvailableValue() + ), + ( + action.action + if action is not None + else renderers.NotAvailableValue() + ), + args, + ( + action_set.context + if action_set is not None + else renderers.NotAvailableValue() + ), + working_directory, + str(key.get_name()), + ) + + def _generator(self) -> Iterator[Tuple[int, _ScheduledTaskEntry]]: + kernel = self.context.modules[self.config["kernel"]] + + # Building the dictionary ahead of time is much better for performance + # vs looking up each service's DLL individually. + software_hive = self.get_software_hive(self.context, self.config_path, kernel) + if software_hive is None: + vollog.warning("Failed to get SOFTWARE hive") + return + + task_key_root, task_tree = self._get_task_keys(software_hive) + if task_key_root is None: + vollog.warning("Failed to get 'Tasks' key") + return + + if task_tree is not None: + task_name_map = _build_guid_name_map(task_tree) + else: + vollog.info("'Tree' key not found, can't map GUIDs to task names") + task_name_map = {} + + for key in task_key_root.get_subkeys(): + for task in self._parse_task_key(key, task_name_map): + yield 0, task + + def run(self): + return renderers.TreeGrid( + [ + ("Task Name", str), + ("Principal ID", str), + ("Display Name", str), + ("Enabled", bool), + ("Creation Time", datetime.datetime), + ("Last Run Time", datetime.datetime), + ("Last Successful Run Time", datetime.datetime), + ("Trigger Type", str), + ("Trigger Description", str), + ("Action Type", str), + ("Action", str), + ("Action Arguments", str), + ("Action Context", str), + ("Working Directory", str), + ("Key Name", str), + ], + ( + (indent, dataclasses.astuple(entry)) + for indent, entry in self._generator() + ), + ) + + +class TestActionsDecoding(unittest.TestCase): + def test_decode_exe_action(self): + # fmt: off + buf = struct.pack( + "512B", + *[ + 0x03, 0x00, 0x16, 0x00, 0x00, 0x00, 0x4c, 0x00, + 0x6f, 0x00, 0x63, 0x00, 0x61, 0x00, 0x6c, 0x00, + 0x53, 0x00, 0x79, 0x00, 0x73, 0x00, 0x74, 0x00, + 0x65, 0x00, 0x6d, 0x00, 0x66, 0x66, 0x00, 0x00, + 0x00, 0x00, 0x6e, 0x00, 0x00, 0x00, 0x25, 0x00, + 0x77, 0x00, 0x69, 0x00, 0x6e, 0x00, 0x64, 0x00, + 0x69, 0x00, 0x72, 0x00, 0x25, 0x00, 0x5c, 0x00, + 0x73, 0x00, 0x79, 0x00, 0x73, 0x00, 0x74, 0x00, + 0x65, 0x00, 0x6d, 0x00, 0x33, 0x00, 0x32, 0x00, + 0x5c, 0x00, 0x57, 0x00, 0x69, 0x00, 0x6e, 0x00, + 0x64, 0x00, 0x6f, 0x00, 0x77, 0x00, 0x73, 0x00, + 0x50, 0x00, 0x6f, 0x00, 0x77, 0x00, 0x65, 0x00, + 0x72, 0x00, 0x53, 0x00, 0x68, 0x00, 0x65, 0x00, + 0x6c, 0x00, 0x6c, 0x00, 0x5c, 0x00, 0x76, 0x00, + 0x31, 0x00, 0x2e, 0x00, 0x30, 0x00, 0x5c, 0x00, + 0x70, 0x00, 0x6f, 0x00, 0x77, 0x00, 0x65, 0x00, + 0x72, 0x00, 0x73, 0x00, 0x68, 0x00, 0x65, 0x00, + 0x6c, 0x00, 0x6c, 0x00, 0x2e, 0x00, 0x65, 0x00, + 0x78, 0x00, 0x65, 0x00, 0x62, 0x01, 0x00, 0x00, + 0x2d, 0x00, 0x45, 0x00, 0x78, 0x00, 0x65, 0x00, + 0x63, 0x00, 0x75, 0x00, 0x74, 0x00, 0x69, 0x00, + 0x6f, 0x00, 0x6e, 0x00, 0x50, 0x00, 0x6f, 0x00, + 0x6c, 0x00, 0x69, 0x00, 0x63, 0x00, 0x79, 0x00, + 0x20, 0x00, 0x55, 0x00, 0x6e, 0x00, 0x72, 0x00, + 0x65, 0x00, 0x73, 0x00, 0x74, 0x00, 0x72, 0x00, + 0x69, 0x00, 0x63, 0x00, 0x74, 0x00, 0x65, 0x00, + 0x64, 0x00, 0x20, 0x00, 0x2d, 0x00, 0x4e, 0x00, + 0x6f, 0x00, 0x6e, 0x00, 0x49, 0x00, 0x6e, 0x00, + 0x74, 0x00, 0x65, 0x00, 0x72, 0x00, 0x61, 0x00, + 0x63, 0x00, 0x74, 0x00, 0x69, 0x00, 0x76, 0x00, + 0x65, 0x00, 0x20, 0x00, 0x2d, 0x00, 0x4e, 0x00, + 0x6f, 0x00, 0x50, 0x00, 0x72, 0x00, 0x6f, 0x00, + 0x66, 0x00, 0x69, 0x00, 0x6c, 0x00, 0x65, 0x00, + 0x20, 0x00, 0x2d, 0x00, 0x57, 0x00, 0x69, 0x00, + 0x6e, 0x00, 0x64, 0x00, 0x6f, 0x00, 0x77, 0x00, + 0x53, 0x00, 0x74, 0x00, 0x79, 0x00, 0x6c, 0x00, + 0x65, 0x00, 0x20, 0x00, 0x48, 0x00, 0x69, 0x00, + 0x64, 0x00, 0x64, 0x00, 0x65, 0x00, 0x6e, 0x00, + 0x20, 0x00, 0x22, 0x00, 0x26, 0x00, 0x20, 0x00, + 0x25, 0x00, 0x77, 0x00, 0x69, 0x00, 0x6e, 0x00, + 0x64, 0x00, 0x69, 0x00, 0x72, 0x00, 0x25, 0x00, + 0x5c, 0x00, 0x73, 0x00, 0x79, 0x00, 0x73, 0x00, + 0x74, 0x00, 0x65, 0x00, 0x6d, 0x00, 0x33, 0x00, + 0x32, 0x00, 0x5c, 0x00, 0x57, 0x00, 0x69, 0x00, + 0x6e, 0x00, 0x64, 0x00, 0x6f, 0x00, 0x77, 0x00, + 0x73, 0x00, 0x50, 0x00, 0x6f, 0x00, 0x77, 0x00, + 0x65, 0x00, 0x72, 0x00, 0x53, 0x00, 0x68, 0x00, + 0x65, 0x00, 0x6c, 0x00, 0x6c, 0x00, 0x5c, 0x00, + 0x76, 0x00, 0x31, 0x00, 0x2e, 0x00, 0x30, 0x00, + 0x5c, 0x00, 0x4d, 0x00, 0x6f, 0x00, 0x64, 0x00, + 0x75, 0x00, 0x6c, 0x00, 0x65, 0x00, 0x73, 0x00, + 0x5c, 0x00, 0x53, 0x00, 0x6d, 0x00, 0x62, 0x00, + 0x53, 0x00, 0x68, 0x00, 0x61, 0x00, 0x72, 0x00, + 0x65, 0x00, 0x5c, 0x00, 0x44, 0x00, 0x69, 0x00, + 0x73, 0x00, 0x61, 0x00, 0x62, 0x00, 0x6c, 0x00, + 0x65, 0x00, 0x55, 0x00, 0x6e, 0x00, 0x75, 0x00, + 0x73, 0x00, 0x65, 0x00, 0x64, 0x00, 0x53, 0x00, + 0x6d, 0x00, 0x62, 0x00, 0x31, 0x00, 0x2e, 0x00, + 0x70, 0x00, 0x73, 0x00, 0x31, 0x00, 0x20, 0x00, + 0x2d, 0x00, 0x53, 0x00, 0x63, 0x00, 0x65, 0x00, + 0x6e, 0x00, 0x61, 0x00, 0x72, 0x00, 0x69, 0x00, + 0x6f, 0x00, 0x20, 0x00, 0x43, 0x00, 0x6c, 0x00, + 0x69, 0x00, 0x65, 0x00, 0x6e, 0x00, 0x74, 0x00, + 0x22, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + ] + ) + + try: + actions = ActionSet.decode(buf) # type: ignore + self.assertEqual(len(actions), 1) + self.assertEqual(actions[0].action_type, ActionType.Exe) + except Exception: + self.fail( + "ActionDecoder.decode should not raise exception:\n%s" + % traceback.format_exc() + ) + + +class TestTriggersDecoding(unittest.TestCase): + def test_decode_all_triggers(self): + """ + Tests decoding a set of all triggers that can be constructed via the + Task Scheduler GUI interface. Ensures that the correct number of bytes + is being consumed for each trigger structure. + """ + buf = struct.pack( + "1808B", + # fmt: off + *[ + 0x17, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x38, 0x21, 0x41, 0x42, 0x48, 0x48, 0x48, 0x48, + 0xa0, 0x12, 0xa0, 0xa4, 0x48, 0x48, 0x48, 0x48, + 0x0e, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x41, 0x00, 0x75, 0x00, 0x74, 0x00, 0x68, 0x00, + 0x6f, 0x00, 0x72, 0x00, 0x00, 0x00, 0x48, 0x48, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x1c, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x05, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, + 0x15, 0x00, 0x00, 0x00, 0x69, 0xce, 0x28, 0x2a, + 0xce, 0xd8, 0x1f, 0x77, 0x37, 0x9c, 0xe2, 0x44, + 0xf4, 0x01, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x40, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x44, 0x00, 0x45, 0x00, 0x53, 0x00, 0x4b, 0x00, + 0x54, 0x00, 0x4f, 0x00, 0x50, 0x00, 0x2d, 0x00, + 0x45, 0x00, 0x33, 0x00, 0x38, 0x00, 0x38, 0x00, + 0x44, 0x00, 0x38, 0x00, 0x50, 0x00, 0x5c, 0x00, + 0x41, 0x00, 0x64, 0x00, 0x6d, 0x00, 0x69, 0x00, + 0x6e, 0x00, 0x69, 0x00, 0x73, 0x00, 0x74, 0x00, + 0x72, 0x00, 0x61, 0x00, 0x74, 0x00, 0x6f, 0x00, + 0x72, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x2c, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x80, 0xf4, 0x03, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0xdd, 0xdd, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x01, 0x07, 0x0a, 0x00, 0x00, 0x00, 0x09, 0x00, + 0x80, 0x48, 0x11, 0xf8, 0x36, 0x1a, 0xdb, 0x01, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x01, 0x2e, 0xe2, 0x01, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0xc2, 0x31, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0xaa, 0xaa, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0xee, 0xee, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xda, 0xaf, 0x8d, 0x09, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0xcc, 0xcc, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x01, 0x00, 0x65, 0x00, 0x78, 0x00, 0x65, 0x00, + 0x22, 0x00, 0x20, 0x00, 0x53, 0x00, 0x74, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x84, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x3c, 0x00, 0x51, 0x00, 0x75, 0x00, 0x65, 0x00, + 0x72, 0x00, 0x79, 0x00, 0x4c, 0x00, 0x69, 0x00, + 0x73, 0x00, 0x74, 0x00, 0x3e, 0x00, 0x3c, 0x00, + 0x51, 0x00, 0x75, 0x00, 0x65, 0x00, 0x72, 0x00, + 0x79, 0x00, 0x20, 0x00, 0x49, 0x00, 0x64, 0x00, + 0x3d, 0x00, 0x22, 0x00, 0x30, 0x00, 0x22, 0x00, + 0x20, 0x00, 0x50, 0x00, 0x61, 0x00, 0x74, 0x00, + 0x68, 0x00, 0x3d, 0x00, 0x22, 0x00, 0x49, 0x00, + 0x6e, 0x00, 0x74, 0x00, 0x65, 0x00, 0x72, 0x00, + 0x6e, 0x00, 0x65, 0x00, 0x74, 0x00, 0x20, 0x00, + 0x45, 0x00, 0x78, 0x00, 0x70, 0x00, 0x6c, 0x00, + 0x6f, 0x00, 0x72, 0x00, 0x65, 0x00, 0x72, 0x00, + 0x22, 0x00, 0x3e, 0x00, 0x3c, 0x00, 0x53, 0x00, + 0x65, 0x00, 0x6c, 0x00, 0x65, 0x00, 0x63, 0x00, + 0x74, 0x00, 0x20, 0x00, 0x50, 0x00, 0x61, 0x00, + 0x74, 0x00, 0x68, 0x00, 0x3d, 0x00, 0x22, 0x00, + 0x49, 0x00, 0x6e, 0x00, 0x74, 0x00, 0x65, 0x00, + 0x72, 0x00, 0x6e, 0x00, 0x65, 0x00, 0x74, 0x00, + 0x20, 0x00, 0x45, 0x00, 0x78, 0x00, 0x70, 0x00, + 0x6c, 0x00, 0x6f, 0x00, 0x72, 0x00, 0x65, 0x00, + 0x72, 0x00, 0x22, 0x00, 0x3e, 0x00, 0x2a, 0x00, + 0x5b, 0x00, 0x53, 0x00, 0x79, 0x00, 0x73, 0x00, + 0x74, 0x00, 0x65, 0x00, 0x6d, 0x00, 0x5b, 0x00, + 0x45, 0x00, 0x76, 0x00, 0x65, 0x00, 0x6e, 0x00, + 0x74, 0x00, 0x49, 0x00, 0x44, 0x00, 0x3d, 0x00, + 0x32, 0x00, 0x5d, 0x00, 0x5d, 0x00, 0x3c, 0x00, + 0x2f, 0x00, 0x53, 0x00, 0x65, 0x00, 0x6c, 0x00, + 0x65, 0x00, 0x63, 0x00, 0x74, 0x00, 0x3e, 0x00, + 0x3c, 0x00, 0x2f, 0x00, 0x51, 0x00, 0x75, 0x00, + 0x65, 0x00, 0x72, 0x00, 0x79, 0x00, 0x3e, 0x00, + 0x3c, 0x00, 0x2f, 0x00, 0x51, 0x00, 0x75, 0x00, + 0x65, 0x00, 0x72, 0x00, 0x79, 0x00, 0x4c, 0x00, + 0x69, 0x00, 0x73, 0x00, 0x74, 0x00, 0x3e, 0x00, + 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x88, 0x88, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x77, 0x77, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, + 0x01, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x01, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x77, 0x77, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x1c, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x05, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, + 0x15, 0x00, 0x00, 0x00, 0x69, 0xce, 0x28, 0x2a, + 0xce, 0xd8, 0x1f, 0x77, 0x37, 0x9c, 0xe2, 0x44, + 0xf4, 0x01, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x40, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x44, 0x00, 0x45, 0x00, 0x53, 0x00, 0x4b, 0x00, + 0x54, 0x00, 0x4f, 0x00, 0x50, 0x00, 0x2d, 0x00, + 0x45, 0x00, 0x33, 0x00, 0x38, 0x00, 0x38, 0x00, + 0x44, 0x00, 0x38, 0x00, 0x50, 0x00, 0x5c, 0x00, + 0x41, 0x00, 0x64, 0x00, 0x6d, 0x00, 0x69, 0x00, + 0x6e, 0x00, 0x69, 0x00, 0x73, 0x00, 0x74, 0x00, + 0x72, 0x00, 0x61, 0x00, 0x74, 0x00, 0x6f, 0x00, + 0x72, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x77, 0x77, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, + 0x01, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x01, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x77, 0x77, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, + 0x01, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x1c, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x05, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, + 0x15, 0x00, 0x00, 0x00, 0x69, 0xce, 0x28, 0x2a, + 0xce, 0xd8, 0x1f, 0x77, 0x37, 0x9c, 0xe2, 0x44, + 0xf4, 0x01, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x40, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x44, 0x00, 0x45, 0x00, 0x53, 0x00, 0x4b, 0x00, + 0x54, 0x00, 0x4f, 0x00, 0x50, 0x00, 0x2d, 0x00, + 0x45, 0x00, 0x33, 0x00, 0x38, 0x00, 0x38, 0x00, + 0x44, 0x00, 0x38, 0x00, 0x50, 0x00, 0x5c, 0x00, + 0x41, 0x00, 0x64, 0x00, 0x6d, 0x00, 0x69, 0x00, + 0x6e, 0x00, 0x69, 0x00, 0x73, 0x00, 0x74, 0x00, + 0x72, 0x00, 0x61, 0x00, 0x74, 0x00, 0x6f, 0x00, + 0x72, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 + ] + # fmt: on + ) + triggers = TriggerSet.decode(buf) + self.assertIsNotNone(triggers) + + def test_decode_triggers(self): + # fmt: off + buf = struct.pack( + "320B", + *[ + 0x17, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xB9, 0x61, 0x1A, 0xA8, 0xB9, 0x61, 0x1A, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xB9, 0x61, 0x1A, 0xA8, 0xB9, 0x61, 0x1A, + 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, + 0x08, 0xA1, 0x40, 0x42, 0x48, 0x48, 0x48, 0x48, + 0x7A, 0x7F, 0x59, 0xDC, 0x48, 0x48, 0x48, 0x48, + 0x22, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x49, 0x00, 0x6E, 0x00, 0x74, 0x00, 0x65, 0x00, + 0x72, 0x00, 0x61, 0x00, 0x63, 0x00, 0x74, 0x00, + 0x69, 0x00, 0x76, 0x00, 0x65, 0x00, 0x55, 0x00, + 0x73, 0x00, 0x65, 0x00, 0x72, 0x00, 0x73, 0x00, + 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + 0x05, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x0C, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, + 0x04, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x2C, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x00, 0x00, 0x00, 0x00, 0xFF, 0xFF, 0xFF, 0xFF, + 0x80, 0x51, 0x01, 0x00, 0xFF, 0xFF, 0xFF, 0xFF, + 0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0xAA, 0xAA, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xB9, 0x61, 0x1A, 0xA8, 0xB9, 0x61, 0x1A, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0xB9, 0x61, 0x1A, 0xA8, 0xB9, 0x61, 0x1A, + 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, + 0x2C, 0x01, 0x00, 0x00, 0xFF, 0xFF, 0xFF, 0xFF, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0xC1, 0xD9, 0x04, + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x0F, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x48, 0x48, 0x48, 0x48, + 0x01, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, 0x48, + ] + ) + # fmt: on + triggers = TriggerSet.decode(buf) + self.assertIsNotNone(triggers) + if not triggers: + return + self.assertGreater(len(triggers.triggers), 0) diff --git a/volatility3/framework/renderers/conversion.py b/volatility3/framework/renderers/conversion.py index c8ddc19fd..e48684b31 100644 --- a/volatility3/framework/renderers/conversion.py +++ b/volatility3/framework/renderers/conversion.py @@ -28,6 +28,27 @@ def wintime_to_datetime( return renderers.UnparsableValue() +def windows_bytes_to_guid(buf: bytes) -> str: + """ + Converts 16 raw bytes to a windows GUID. + + Raises ValueError if the provided buffer is not exactly 16 bytes. + """ + if len(buf) != 16: + raise ValueError("Expected 16 bytes for GUID") + + head_components = [format(v, "x") for v in struct.unpack("HQ", + buf[8:10] + b"\x00\x00" + buf[10:16], + ) + ] + combined = head_components + tail_component + return "{" + "-".join(combined) + "}" + + def unixtime_to_datetime( unixtime: int, ) -> Union[interfaces.renderers.BaseAbsentValue, datetime.datetime]: