splunk-security_content/playbooks/custom_functions/workbook_task_update.py

def workbook_task_update(task_name=None, note_title=None, note_content=None, status=None, owner=None, container=None, **kwargs):
    """
    Update a workbook task by task name or the task where the currently running playbook appears. Requires a task_name, container_id, and a note_title, note_content, owner, or status.
    
    Args:
        task_name (CEF type: *): Name of a workbook task or keyword 'playbook' to update the task where the currently running playbook appears. (Required)
        note_title (CEF type: *): Note title. (Optional)
        note_content (CEF type: *): Note content. (Optional)
        status (CEF type: *): Accepts 'incomplete', 'in_progress, or 'complete'. (Optional)
        owner (CEF type: *): A user to assign as the task owner or keyword 'current" to assign the task to the user that launched the playbook. (Optional)
        container (CEF type: phantom container id): The ID of a SOAR Container. (Required)
    
    Returns a JSON-serializable object that implements the configured data paths:
        note_id: Returns note_id if a note was added. A closing note on a required task note will not produce a note_id.
    """
    ############################ Custom Code Goes Below This Line #################################
    import json
    import phantom.rules as phantom
    
    outputs = {}
    closing_note = False # special variable to keep track of if this note should be treated as a closing note
    
    # Ensure valid container input
    if isinstance(container, dict) and container.get('id'):
        container_id = container['id']
    elif isinstance(container, int):
        container_id = container
    else:
        raise TypeError("The input 'container' is neither a container dictionary nor an int, so it cannot be used")
    
    if not any([note_title, note_content, owner, status]):
        raise RuntimeError("Workbook task update requires a note_title, note_content, owner, or status.")
    
    task_id = None
    if task_name:
        task_list = phantom.get_tasks(container_id)
        task_count = 0
        current_playbook = ""
        
        if task_name == 'playbook':
            current_playbook = (phantom.get_playbook_info()[0]['repo_name'], phantom.get_playbook_info()[0]['name'])
            
        for task in task_list:
            if task['success'] and task.get('data'):
                task_data = task['data']
                match = False
                if task_name == 'playbook':
                    playbooks = task_data['suggestions'].get('playbooks', [])
                    for playbook in playbooks:
                        if playbook['scm'] == current_playbook[0] and playbook['playbook'] == current_playbook[1]:
                            match = True
                            task_count += 1
                elif task_name == task_data['name']:
                    match = True
                    task_count += 1

                if match == True:
                    task_id = task_data['id']
                    task_is_note_required = task_data['is_note_required']
                    task_status = task_data['status']
                    task_notes = task_data['notes']
                    task_owner = task_data['owner']
                    if task_is_note_required and status == 'complete' and task_status != 1:
                        closing_note = True
                
    elif not task_name:
        raise RuntimeError("Missing input for task_name.")
    
    if task_count > 1:
        raise RuntimeError(f"'Unable to update workbook task. {task_count} match criteria '{task_name}'")
    elif task_count == 0:
        if task_name == "playbook" and current_playbook:
            raise RuntimeError(f"No task contains the current playbook: '{current_playbook}'")
        else:
            raise RuntimeError(f"No task name matches input task_name: '{task_name}'")

    if task_is_note_required and not note_content and status == 'complete' and task_status != 1:
        raise RuntimeError('Unable to update workbook task - The task requires a closing note and a closing title')     

    # Use add note if this is not a closing note otherwise add note during status change.
    if note_content and not closing_note:
        _, _, note_id = phantom.add_note(
            container=container_id, 
            note_type='task',
            task_id=task_id, 
            title=note_title if note_title else "", 
            content=note_content,
            note_format='markdown'
        )
        outputs['note_id'] = str(note_id)

    # Set owner
    if owner:
        owner_dict = {}
        # If keyword 'current' entered then translate effective_user id to a username
        if owner.lower() == 'current':
            owner_dict['owner_id'] = phantom.get_effective_user()
        else:
            # Attempt to translate name to owner_id
            url = phantom.build_phantom_rest_url('ph_user') + f'?_filter_username="{owner}"'
            data = phantom.requests.get(url, verify=False).json().get('data')
            if data and len(data) == 1:
                owner_dict['owner_id'] = data[0]['id']
            elif data and len(data) > 1:
                raise RuntimeError(f'Multiple matches for owner "{owner}"')
            else:
                # Attempt to translate name to role_id
                url = phantom.build_phantom_rest_url('role') + f'?_filter_name="{owner}"'
                data = phantom.requests.get(url, verify=False).json().get('data')
                if data and len(data) == 1:
                    owner_dict['role_id'] = data[0]['id']
                elif data and len(data) > 1:
                    raise RuntimeError(f'Multiple matches for owner "{owner}"')
                else:
                    raise RuntimeError(f'"{owner}" is not a valid username or role')

        url = phantom.build_phantom_rest_url('workbook_task') + '/{}'.format(task_id)
        response = phantom.requests.post(url, data=json.dumps(owner_dict), verify=False).json()             
        if not response.get('success'):
            raise RuntimeError(f'Error setting "{owner}" - {response}')

    # Set Status
    if isinstance(status, str):
        status = status.lower()
        url = phantom.build_phantom_rest_url('workbook_task') + '/{}'.format(task_id)
        if status == 'complete' and task_status == 0:
            # Move to in progress
            data = {'status': 2}
            response = phantom.requests.post(url, data=json.dumps(data), verify=False).json()
            if not response.get('success'):
                raise RuntimeError(f'Error setting status "{status}" - {response}')
                # Then move to close
            data = {'status': 1}
            if task_is_note_required and note_content:
                data['note'] = note_content
                data['title'] = note_title if note_title else ""
                data['note_format'] = 'markdown'
            response = phantom.requests.post(url, data=json.dumps(data), verify=False).json()
            if not response.get('success'):
                raise RuntimeError(f'Error setting status "{status}" - {response}')
        elif (status == 'in progress' or status == 'in_progress') and task_status != 2:
            data = {'status': 2}
            # Move to in progress
            response = phantom.requests.post(url, data=json.dumps(data), verify=False).json()
            if not response.get('success'):
                raise RuntimeError(f'Error setting status "{status}" - {response}')
        elif status == 'incomplete' and task_status != 0:
            data = {'status': 0}
            # Move to incomplete
            response = phantom.requests.post(url, data=json.dumps(data), verify=False).json()
            if not response.get('success'):
                raise RuntimeError(f'Error setting status "{status}" - {response}')
        elif status == 'complete' and task_status != 1:
            data = {'status': 1}
            # Move to complete
            if task_is_note_required and note_content:
                data['note'] = note_content
                data['title'] = note_title if note_title else ""
                data['note_format'] = 'markdown'
            response = phantom.requests.post(url, data=json.dumps(data), verify=False).json()
            if not response.get('success'):
                raise RuntimeError(f'Error setting status "{status}" - {response}')

        
    # Return a JSON-serializable object
    assert json.dumps(outputs)  # Will raise an exception if the :outputs: object is not JSON-serializable
    return outputs