Source code for windows.winobject.task_scheduler

import windows.com
import windows.generated_def as gdef


def generate_simple_getter(function, restype, extract_value=True, doc=None):
    def value_getter(self):
        res = restype()
        getattr(self, function)(res)
        if extract_value:
            return res.value
        return res
    return property(value_getter, doc=doc)


def add_simple_setter(getter, function, restype):
    @getter.setter
    def value_setter(self, value):
        resvalue = restype(value)
        return getattr(self, function)(resvalue)
    return value_setter


class TaskCollectionType(object):
    ITEM_TYPE = None

    count = generate_simple_getter("get_Count", gdef.LONG)

    def get_item_type(self):
        return self.ITEM_TYPE

    def get_item(self, index):
        """Return elements nb ``index``. Collection index starts at 1"""
        if index == 0:
            raise IndexError("<{0}> Index start as 1".format(type(self).__name__))
        index = self.get_index(index)
        res = self.get_item_type()()
        self.get_Item(index, res)
        return res

    def get_index(self, index):
        return index

    def items_generator(self):
        for i in range(self.count):
            # Start index is 1
            # https://msdn.microsoft.com/en-us/library/windows/desktop/aa446901(v=vs.85).aspx
            yield self.get_item(1 + i)

    @property
    def items(self):
        """Return the list of item in the collection

        :type: :class:`list`
        """
        return list(self.items_generator())

    def __iter__(self):
        return self.items_generator()

    def __getitem__(self, index): # Allow subclasses to only overwrite 'get_item' to rewrite both behavior
        return self.get_item(index)

# Need to-do the doc=xx tricks to have the documentation in the 'AbstractAction' subclasses

class AbstractAction(object):
    type_doc = """The type of action

    :type: :class:`~windows.generated_def.winstructs.TASK_ACTION_TYPE`
    """
    type = generate_simple_getter("get_Type", gdef.TASK_ACTION_TYPE, doc=type_doc)

    id_doc = """The action id

    :type: :class:`~windows.generated_def.winstructs.BSTR`
    """
    id = generate_simple_getter("get_Id", gdef.BSTR, doc=id_doc)


[docs] class Action(gdef.IAction, AbstractAction): """Describe an action performed by a task""" ACTION_SUBTYPE = {} @property def subtype(self): """Return the :class:`Action`-subtype according to :data:`AbstractAction.type`""" subinterface = self.ACTION_SUBTYPE[self.type] # KeyError ? return self.query(subinterface)
[docs] class ExecAction(gdef.IExecAction, AbstractAction): """Represent an action of type :data:`~windows.generated_def.winstructs._TASK_ACTION_TYPE.TASK_ACTION_EXEC`""" path = generate_simple_getter("get_Path", gdef.BSTR) path = add_simple_setter(path, "put_Path", gdef.BSTR) """[R-W] The path of the programm to execute""" arguments = generate_simple_getter("get_Arguments", gdef.BSTR) arguments = add_simple_setter(arguments, "put_Arguments", gdef.BSTR) """[R-W] The arguments for the command to execute""" working_directory = generate_simple_getter("get_WorkingDirectory", gdef.BSTR) """The working directory for the command to execute"""
# Register action subtype Action.ACTION_SUBTYPE[gdef.TASK_ACTION_EXEC] = ExecAction
[docs] class ComHandlerAction(gdef.IComHandlerAction, AbstractAction): """Represent an action of type :data:`~windows.generated_def.winstructs._TASK_ACTION_TYPE.TASK_ACTION_COM_HANDLER`""" classid = generate_simple_getter("get_ClassId", gdef.BSTR) classid = add_simple_setter(classid, "put_ClassId", gdef.BSTR) """The CLSID of the COM server executed :type: :class:`~windows.generated_def.winstructs.BSTR` """ data = generate_simple_getter("get_Data", gdef.BSTR) data = add_simple_setter(data, "put_Data", gdef.BSTR) """The DATA for the COM class :type: :class:`~windows.generated_def.winstructs.BSTR` """
# Register action subtype Action.ACTION_SUBTYPE[gdef.TASK_ACTION_COM_HANDLER] = ComHandlerAction class EmailAction(gdef.IEmailAction, AbstractAction): pass Action.ACTION_SUBTYPE[gdef.TASK_ACTION_SEND_EMAIL] = EmailAction class ShowMessageAction(gdef.IShowMessageAction, AbstractAction): pass Action.ACTION_SUBTYPE[gdef.TASK_ACTION_SHOW_MESSAGE] = ShowMessageAction
[docs] class Trigger(gdef.ITrigger): """A task trigger""" type = generate_simple_getter("get_Type", gdef.TASK_TRIGGER_TYPE2) """The type of trigger :type: :class:`~windows.generated_def.winstructs.TASK_TRIGGER_TYPE2` """
[docs] class ActionCollection(gdef.IActionCollection, TaskCollectionType): ITEM_TYPE = Action
[docs] def create(self, action_type): """Create a new action of type ``action_type`` :rtype: A subclass of :class:`Action` """ res = self.ITEM_TYPE() self.Create(action_type, res) return res.subtype
[docs] def get_item(self, index): item = super(ActionCollection, self).get_item(index) # Need to Release() item ? return item.subtype
[docs] class TriggerCollection(gdef.ITriggerCollection, TaskCollectionType): ITEM_TYPE = Trigger
[docs] class TaskRegistrationInfo(gdef.IRegistrationInfo): """Provides the administrative information that can be used to describe the task. This information includes details such as a description of the task, the author of the task, the date the task is registered, and the security descriptor of the task. """ author = generate_simple_getter("get_Author", gdef.BSTR) """The author of the task""" description = generate_simple_getter("get_Description", gdef.BSTR) """The description of the task""" date = generate_simple_getter("get_Date", gdef.BSTR) """The registration date of the task""" source = generate_simple_getter("get_Source", gdef.BSTR) """Where the task originated from. For example, a task may originate from a component, service, application, or user. """ documentation = generate_simple_getter("get_Documentation", gdef.BSTR) """Any additional documentation for the task""" uri = generate_simple_getter("get_URI", gdef.BSTR) """the URI of the task.""" version = generate_simple_getter("get_Version", gdef.BSTR) """The version number of the task.""" # Return WindowsError: [Error -2147467263] Not implemented # xml = generate_simple_getter("get_XmlText", gdef.BSTR) sddl = generate_simple_getter("get_SecurityDescriptor", windows.com.Variant) @property def security_descriptor(self): sddl = self.sddl if not sddl: return None return windows.security.SecurityDescriptor.from_string(sddl)
[docs] class TaskPrincipal(gdef.IPrincipal): """Provides the security credentials for a principal. These security credentials define the security context for the tasks that are associated with the principal. """ name = generate_simple_getter("get_DisplayName", gdef.BSTR) name = add_simple_setter(name, "put_DisplayName", gdef.BSTR) """The name of the principal""" id = generate_simple_getter("get_Id", gdef.BSTR) id = add_simple_setter(id, "put_Id", gdef.BSTR) """the identifier of the principal.""" user_id = generate_simple_getter("get_UserId", gdef.BSTR) user_id = add_simple_setter(user_id, "put_UserId", gdef.BSTR) """the user identifier that is required to run the task""" group_id = generate_simple_getter("get_GroupId", gdef.BSTR) group_id = add_simple_setter(group_id, "put_GroupId", gdef.BSTR) """the user group that is required to run the task""" run_level = generate_simple_getter("get_RunLevel", gdef.TASK_RUNLEVEL_TYPE) """the privilege level that is required to run the tasks :type: :class:`~windows.generated_def.winstructs.TASK_RUNLEVEL_TYPE` """ logon_type = generate_simple_getter("get_LogonType", gdef.TASK_LOGON_TYPE) """ logon method that is required to run the task :type: :class:`~windows.generated_def.winstructs.TASK_LOGON_TYPE` """
[docs] class TaskDefinition(gdef.ITaskDefinition): """The definition of a task""" actions = generate_simple_getter("get_Actions", ActionCollection, extract_value=False) """The list of actions of the task :type: :class:`ActionCollection` """ triggers = generate_simple_getter("get_Triggers", TriggerCollection, extract_value=False) """The list of triggers of the task :type: :class:`TriggerCollection` """ registration_info = generate_simple_getter("get_RegistrationInfo", TaskRegistrationInfo, extract_value=False) """The registration information of the task :type: :class:`TaskRegistrationInfo` """ principal = generate_simple_getter("get_Principal", TaskPrincipal, extract_value=False) """The principal that provides the security credentials for the task. These security credentials define the security context for the tasks that are associated with the principal. :type: :class:`TaskPrincipal` """ xml = generate_simple_getter("get_XmlText", gdef.BSTR) """The XML representig the task definition :type: :class:`str` """
[docs] class Task(gdef.IRegisteredTask): """A scheduled task""" name = generate_simple_getter("get_Name", gdef.BSTR) """The name of the task""" path = generate_simple_getter("get_Path", gdef.BSTR) """The path of the task""" state = generate_simple_getter("get_State", gdef.TASK_STATE) """The state of the task :type: :class:`~windows.generated_def.winstructs.TASK_STATE` """ enabled = generate_simple_getter("get_Enabled", gdef.VARIANT_BOOL) """``True`` is the task is enabled""" last_runtime = generate_simple_getter("get_LastRunTime", gdef.DATE) """Gets the last time the registered task was last run.""" next_runtime = generate_simple_getter("get_NextRunTime", gdef.DATE) """Gets the next time the registered task will be run.""" definition = generate_simple_getter("get_Definition", TaskDefinition, extract_value=False) """The definition of the task :type: :class:`TaskDefinition` """ xml = generate_simple_getter("get_Xml", gdef.BSTR) """The XML representig the task :type: :class:`str` """ def run(self, params=None, flags=gdef.TASK_RUN_NO_FLAGS, sessionid=0, user=None): if params is None: params = gdef.VARIANT() # Empty variant result = gdef.IRunningTask() self.RunEx(params, flags, sessionid, user, result) return result def get_security_descriptor(self, secinfo): res = gdef.BSTR() self.GetSecurityDescriptor(secinfo, res) return res.value def __repr__(self): return """<{0} "{1}" at {2:#x}>""".format(type(self).__name__, self.name, id(self))
[docs] class TaskCollection(gdef.IRegisteredTaskCollection, TaskCollectionType): ITEM_TYPE = Task def get_index(self, index): vindex = windows.com.Variant() vindex.vt = gdef.VT_I4 vindex._VARIANT_NAME_3.lVal = index return vindex
[docs] class TaskService(gdef.ITaskService): """The task scheduler"""
[docs] def create(self, flags=0): """Create a new :class:`TaskDefinition` that can be used to create/register a new scheduled task :rtype: :class:`TaskDefinition` """ res = TaskDefinition() self.NewTask(flags, res) return res
def connect(self, server=None, user=None, domain=None, password=None): if server is None: server = gdef.VARIANT() # Empty variant if user is None: user = gdef.VARIANT() # Empty variant if domain is None: domain = gdef.VARIANT() # Empty variant if password is None: password = gdef.VARIANT() # Empty variant self.Connect(server, user, domain, password)
[docs] def folder(self, name): """Return the :class:`TaskFolder` with ``name`` :rtype: :class:`TaskFolder` """ folder = TaskFolder() self.GetFolder(name, folder) return folder
__call__ = folder # use the same 'API' than the registry """Alias for :func:`folder`""" @property def root(self): r"""The root ``\`` :class:`TaskFolder`""" return self.folder("\\")
[docs] class TaskFolder(gdef.ITaskFolder): """A folder of tasks""" path = generate_simple_getter("get_Path", gdef.BSTR) name = generate_simple_getter("get_Name", gdef.BSTR) @property def folders(self): """The list of sub-folders :type: :class:`TaskFolderCollection` """ res = TaskFolderCollection() self.GetFolders(0, res) return res
[docs] def register(self, name, taskdef, flags=gdef.TASK_CREATE, userid=None, password=None, logonType=gdef.TASK_LOGON_NONE, ssid=None): """Register the task definition ``taskdef`` as a new task with ``name`` :rtype: :class:`Task` """ new_task = Task() if userid is None: userid = gdef.VARIANT() # Empty variant if password is None: password = gdef.VARIANT() # Empty variant if ssid is None: ssid = gdef.VARIANT() # Empty variant self.RegisterTaskDefinition(name, taskdef, flags, userid, password, logonType, ssid, new_task) return new_task
@property def tasks(self, flags=gdef.TASK_ENUM_HIDDEN): """The list of tasks in the folder :type: :class:`TaskCollection` """ tasks = TaskCollection() self.GetTasks(flags, tasks) return tasks
[docs] def get_task(self, name): """Retrieve the task with ``name`` in the current folder :rtype: :class:`Task` """ res = Task() self.GetTask(name, res) return res
[docs] def delete_task(self, name): """Delete the task with ``name`` in the current folder""" return self.DeleteTask(name, 0)
[docs] def folder(self, name): """Return the :class:`TaskFolder` with ``name``""" folder = TaskFolder() self.GetFolder(name, folder) return folder
[docs] def create_folder(self, name): """Create a new sub-:class:`TaskFolder` with ``name``""" folder = TaskFolder() self.CreateFolder(name, gdef.VARIANT(), folder) return folder
[docs] def delete_folder(self, name): """Delete the sub-folder with ``name`` in the current folder""" return self.DeleteFolder(name, 0)
__getitem__ = get_task """ Alias for :func:`get_task`""" __delitem__ = delete_task """ Alias for :func:`delete_task`""" __call__ = folder # use the same 'API' than the registry """ Alias for :func:`folder`""" def __repr__(self): return """<{0} "{1}" at {2:#x}>""".format(type(self).__name__, self.path, id(self))
[docs] class TaskFolderCollection(gdef.ITaskFolderCollection, TaskCollectionType): ITEM_TYPE = TaskFolder def get_index(self, index): vindex = windows.com.Variant() vindex.vt = gdef.VT_I4 vindex._VARIANT_NAME_3.lVal = index return vindex
# windows.com.init() # clsid_task_scheduler = gdef.IID.from_string("0f87369f-a4e5-4cfc-bd3e-73e6154572dd") # x = TaskService() # emptvar = gdef.VARIANT() # windows.com.create_instance(clsid_task_scheduler, x) # x.connect() # folder = x.folder("\\") # assert folder.value # tasks = folder.tasks # for task in tasks.items: # print(task.name) # for action in task.definition.actions.items: # print(" * {0}".format(action.type)) # subtype = action.subtype # print(" * Path: {0}".format(subtype.path)) # print(" * Args: {0}".format(subtype.arguments)) # print(" * WDir: {0}".format(subtype.working_directory)) # Test creation # ntd = x.create() # actions = ntd.actions # nea = actions.create(gdef.TASK_ACTION_EXEC).subtype # nea.path = "MY_BINARY" # nea.arguments = "MY_ARGUMENTS" # folder.register("PROUT", ntd) # path = gdef.BSTR() # e.get_Path(path) # print(path)