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
    
    Args:
        task_name (CEF type: *): Name of a workbook task (Required)
        note_title (CEF type: *): Note title goes here (Optional)
        note_content (CEF type: *): Body of note goes here (Optional)
        status (CEF type: *): One of: incomplete, in_progress, complete (Optional)
        owner (CEF type: *): Assigns task to provided owner. Accepts keyword 'current" to assign task to currently running playbook user. (Optional)
        container (CEF type: phantom container id): ID of Phantom Container (Required)
    
    Returns a JSON-serializable object that implements the configured data paths:
        note_id: Returns note_id if a note was added
    """
    ############################ Custom Code Goes Below This Line #################################
    import json
    import phantom.rules as phantom
    
    outputs = {}
    
    # 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 task_name:
        task_list = phantom.get_tasks(container_id)
        task_count = 0
        for task in task_list:
            if task_name == task['data']['name']:
                task_count += 1
                if task_count > 1:
                    raise RuntimeError(f'Unable to update workbook task - multiple tasks match criteria: {task_count}')
                task_id = task['data']['id']
                task_is_note_required = task['data']['is_note_required']
                task_count += 1
                task_status = task['data']['status']
                task_notes = task['data']['notes']
                task_owner = task['data']['owner']
                
    if task_count == 0:
        raise RuntimeError(f"No task name matches input task_name: '{task_name}'")

    if task_is_note_required and (not note_content or not note_title) and status == 'complete' and task_status != 1:
        raise RuntimeError('Unable to update workbook task - The task requires a closing note and a closing title')     
    else:
        # Add Note
        if note_content:
            success, message, note_id = phantom.add_note(container=container_id, note_type='task',
                                                         task_id=task_id, title=note_title, 
                                                         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
                    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
                    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