✨ Adds initial scaffolding for accessing user data (#92)

caracara/client.py

@@ -33,6 +33,7 @@
     PreventionPoliciesApiModule,
     ResponsePoliciesApiModule,
     RTRApiModule,
+    UsersApiModule,
     MODULE_FILTER_ATTRIBUTES,
 )
 
@@ -161,6 +162,8 @@ def __init__(  # pylint: disable=R0913,R0914,R0915
         self.response_policies = ResponsePoliciesApiModule(self.api_authentication)
         self.logger.debug("Setting up the RTR module")
         self.rtr = RTRApiModule(self.api_authentication)
+        self.logger.debug("Setting up the Users module")
+        self.users = UsersApiModule(self.api_authentication)
 
         self.logger.debug("Configuring FQL filters")
         # Pre-configure the FQL modules for faster instantiation later

caracara/common/batching.py

@@ -105,6 +105,8 @@ def worker(
             resources_dict[resource['device_id']] = resource
         elif 'child_cid' in resource:
             resources_dict[resource['child_cid']] = resource
+        elif 'uuid' in resource:
+            resources_dict[resource['uuid']] = resource
         else:
             raise Exception("No ID field to build the dictionary from")
 

caracara/common/module.py

@@ -31,7 +31,7 @@ def help(self) -> str:
     def __init__(self, api_authentication: OAuth2):
         """Configure a Caracara API module with a FalconPy OAuth2 module."""
         class_name = self.__class__.__name__
-        self.logger = logging.getLogger(class_name)
+        self.logger = logging.getLogger(f"caracara.modules.{class_name}")
         self.logger.debug("Initialising API module: %s", class_name)
 
         self.api_authentication = api_authentication

caracara/common/pagination.py

@@ -18,7 +18,7 @@
 code file contains the required code to pull this data down as quickly
 as possible.
 
-Four types of paginator are in use:
+Five types of paginator are in use:
 
 Style 1 (Numerical Offset)
 Implementation function: all_pages_numbered_offset()
@@ -74,6 +74,18 @@
 this will be built.
 See the "Deep Pagination Leveraging Markers (Timestamp)" section here:
     https://falconpy.io/Usage/Response-Handling.html
+
+
+Style 5 (Generic Parallelised One-by-One Call)
+Some APIs, such as the User Management ones, include functions that accept
+only one parameter at a time (e.g., a single ID rather than a list of IDs).
+These APIs could be parallelised within individual Caracara modules / functions,
+but instead a generic implementation is provided here.
+
+Given that there are different types of IDs within the API (CIDs, AIDs, UUIDs, etc.),
+this function takes a parameter name / parameter value list pair.
+Param Name = the name of the kwarg that should be swapped out in each call
+Value List = a list of strings to be iterated over
 """
 import concurrent.futures
 import logging
@@ -280,3 +292,71 @@ def all_pages_token_offset(
     logger.debug(item_ids)
 
     return item_ids
+
+
+def _generic_parallel_list_execution_worker(
+    func: Callable[[Dict[str, Dict]], List[Dict] or List[str]] or partial,
+    logger: logging.Logger,
+    param_name: str,
+    param_value: str,
+):
+    thread_name = current_thread().name
+    if isinstance(func, partial):
+        logger.info(
+            "%s | Batch worker started with partial function: %s",
+            thread_name, func.func.__name__,
+        )
+    else:
+        logger.info(
+            "%s | Batch worker started with function: %s",
+            thread_name, func.__name__,
+        )
+
+    response = func(**{param_name: param_value})['body']
+    logger.debug("%s | %s", thread_name, response)
+    resources = response.get('resources', [])
+    logger.info("%s | Retrieved %d resources", thread_name, len(resources))
+
+    return resources
+
+
+def generic_parallel_list_execution(
+        func: Callable[[Dict[str, Dict]], List[Dict] or List[str]] or partial,
+        logger: logging.Logger,
+        param_name: str,
+        value_list: List[str],
+):
+    """Call a function many times in a thread pool based on a list of kwarg values.
+
+    This is what is described above as Pagination Style 5. A function (or partial)
+    should be provided, along with a param_name (e.g., id, cid, uuid, etc.) and
+    a list of strings to be iterated over, each of which will be assigned in turn to
+    the param named in the previous parameter.
+    """
+    logger = logger.getChild(__name__)
+    if isinstance(func, partial):
+        logger.info(
+            "Pagination Style 5: Repeatedly executing the partial function %s",
+            func.func.__name__,
+        )
+    else:
+        logger.info(
+            "Pagination Style 2: Grabbing all pages from the function %s",
+            func.__name__,
+        )
+
+    all_resources = []
+    threads = batch_data_pull_threads()
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor:
+        partial_worker = partial(_generic_parallel_list_execution_worker, func, logger, param_name)
+        completed = executor.map(
+            partial_worker,
+            value_list,
+        )
+
+    for complete in completed:
+        logger.debug("Completed a function execution against one item")
+        all_resources.extend(complete)
+
+    return all_resources

caracara/modules/__init__.py

@@ -1,7 +1,8 @@
 """
-Falcon Host module.
+Caracara Modules Package Initialisation.
 
-Exposes functions to get host data and perform actions on hosts, such as network containment.
+Proides pipework to link together the various modules within Caracara and expose them to
+the Client object at setup.
 """
 __all__ = [
     'CustomIoaApiModule',
@@ -10,6 +11,7 @@
     'PreventionPoliciesApiModule',
     'ResponsePoliciesApiModule',
     'RTRApiModule',
+    'UsersApiModule',
     'MODULE_FILTER_ATTRIBUTES',
 ]
 
@@ -25,10 +27,15 @@
     FILTER_ATTRIBUTES as rtr_filter_attributes,
     RTRApiModule,
 )
+from caracara.modules.users import (
+    FILTER_ATTRIBUTES as users_filter_attributes,
+    UsersApiModule,
+)
 
 # Build up a list with all filter attributes from the includes modules.
 # This makes makes for much easier importing when setting up Falcon Filters.
 MODULE_FILTER_ATTRIBUTES = [
     *hosts_filter_attributes,
     *rtr_filter_attributes,
+    *users_filter_attributes,
 ]

caracara/modules/users/__init__.py

@@ -0,0 +1,7 @@
+"""Caracara User Management module."""
+__all__ = [
+    'FILTER_ATTRIBUTES',
+    'UsersApiModule',
+]
+from caracara.modules.users.users import UsersApiModule
+from caracara.modules.users.users_filters import FILTER_ATTRIBUTES

caracara/modules/users/users.py

@@ -0,0 +1,126 @@
+"""Falcon Users API."""
+from functools import partial
+from typing import Dict, List
+
+from falconpy import (
+    OAuth2,
+    UserManagement,
+)
+
+from caracara.common.batching import batch_get_data
+from caracara.common.module import FalconApiModule
+from caracara.common.pagination import (
+    all_pages_numbered_offset_parallel,
+    generic_parallel_list_execution,
+)
+from caracara.filters import FalconFilter
+from caracara.filters.decorators import filter_string
+
+
+class UsersApiModule(FalconApiModule):
+    """
+    Users API module.
+
+    This module provides the logic to interact with the Falcon User Management
+    APIs. With the functions provided herein, one can list and create users, and
+    assign access roles.
+    """
+
+    name = "CrowdStrike User Management API Module"
+    help = "Describe, create, delete and edit users in a Falcon tenant"
+
+    def __init__(self, api_authentication: OAuth2):
+        """Construct an instance of the UsersApiModule class."""
+        super().__init__(api_authentication)
+        self.logger.debug("Configuring the FalconPy User Management API")
+        self.user_management_api = UserManagement(auth_object=self.api_authentication)
+
+    @filter_string
+    def get_user_uuids(self, filters: FalconFilter or str = None) -> List[str]:
+        """Get a list of IDs of users (UUIDs) configured in the Falcon tenant."""
+        self.logger.info("Obtaining a list of all users in the Falcon tenant")
+
+        query_users_func = partial(self.user_management_api.query_users, filter=filters)
+        user_uuids: List[str] = all_pages_numbered_offset_parallel(
+            query_users_func,
+            self.logger,
+            500,
+        )
+        return user_uuids
+
+    def get_user_data(self, user_uuids: List[str]) -> Dict[str, Dict]:
+        """Fetch a dictionary of data for each of the User IDs (UUIDs) passed into the function."""
+        self.logger.info("Obtaining data for the %d User IDs provided", len(user_uuids))
+
+        user_data: Dict[str, Dict] = batch_get_data(
+            user_uuids,
+            self.user_management_api.retrieve_users,
+        )
+        return user_data
+
+    @filter_string
+    def describe_users(self, filters: FalconFilter or str = None) -> Dict[str, Dict]:
+        """Describe the users in a Falcon tenant."""
+        self.logger.info("Describing users")
+
+        user_uuids = self.get_user_uuids(filters=filters)
+        user_data = self.get_user_data(user_uuids)
+        user_roles = self.get_assigned_user_roles(user_uuids)
+
+        # Set up an empty set to contain all roles assigned to each user
+        for user_data_dict in user_data.values():
+            user_data_dict['roles'] = set()
+
+        # Iterate over all roles in the CID and add them to each user's role set
+        for user_role in user_roles:
+            if user_role['uuid'] in user_data:
+                user_data[user_role['uuid']]['roles'].add(user_role['role_id'])
+
+        # Convert the sets to sorted list objects so that they can be JSON serialised
+        for user_data_dict in user_data.values():
+            user_data_dict['roles'] = sorted(list(user_data_dict['roles']))
+
+        return user_data
+
+    def get_available_role_ids(self) -> List[str]:
+        """Obtain a list of role IDs enabled on the Falcon tenant."""
+        self.logger.info("Fetching a list of role IDs")
+
+        # Endpoint does not support offsets, so we do not need to parallelise here
+        role_ids: List[str] = self.user_management_api.get_available_role_ids()['body']['resources']
+        return role_ids
+
+    def get_role_information(self, role_ids: List[str]) -> Dict[str, Dict]:
+        """Fetch a dictionary describing each of the Role IDs passed into the function."""
+        self.logger.info("Getting information on the %d role IDs provided", len(role_ids))
+
+        role_info: Dict[str, Dict] = batch_get_data(
+            role_ids,
+            self.user_management_api.get_roles_mssp,
+        )
+        return role_info
+
+    def describe_available_roles(self) -> Dict[str, Dict]:
+        """Describe the roles that are available within the current Falcon tenant."""
+        self.logger.info("Describing available roles")
+
+        role_ids = self.get_available_role_ids()
+        role_info = self.get_role_information(role_ids)
+        return role_info
+
+    def get_assigned_user_roles(self, user_uuids: List[str]) -> Dict[str, Dict]:
+        """Retrieve a list of roles assigned to a list of User UUIDs."""
+        self.logger.info("Retrieving roles for %d User IDs", len(user_uuids))
+
+        get_user_grants_func = partial(
+            self.user_management_api.get_user_grants,
+            direct_only=False,
+            sort="cid",
+        )
+        user_roles = generic_parallel_list_execution(
+            get_user_grants_func,
+            logger=self.logger,
+            param_name="user_uuid",
+            value_list=user_uuids,
+        )
+        return user_roles

caracara/modules/users/users_filters.py

@@ -0,0 +1,76 @@
+"""User Management-Specific FQL Filters."""
+from caracara.filters.fql import FalconFilterAttribute
+
+
+class UsersAssignedCIDsFilterAttribute(FalconFilterAttribute):
+    """Filter by CIDs assigned to a user."""
+
+    name = "AssignedCIDs"
+    fql = "assigned_cids"
+
+    def example(self) -> str:
+        """Show filter example."""
+        return (
+            "This filter accepts a CID that the user should be assigned to."
+        )
+
+
+class UsersCIDFilterAttribute(FalconFilterAttribute):
+    """Filter by a user's home CID."""
+
+    name = "CID"
+    fql = "cid"
+
+    def example(self) -> str:
+        """Show filter example."""
+        return (
+            "This filter accepts a CID that would represent the user's home CID."
+        )
+
+
+class UsersFirstNameFilterAttribute(FalconFilterAttribute):
+    """Filter by a user's first name."""
+
+    name = "FirstName"
+    fql = "first_name"
+
+    def example(self) -> str:
+        """Show filter example."""
+        return (
+            "This filter accepts a user's first name, such as John."
+        )
+
+
+class UsersLastNameFilterAttribute(FalconFilterAttribute):
+    """Filter by a user's last name."""
+
+    name = "LastName"
+    fql = "last_name"
+
+    def example(self) -> str:
+        """Show filter example."""
+        return (
+            "This filter accepts a user's last name, such as Smith."
+        )
+
+
+class UsersNameFilterAttribute(FalconFilterAttribute):
+    """Filter by a user's name."""
+
+    name = "Name"
+    fql = "name"
+
+    def example(self) -> str:
+        """Show filter example."""
+        return (
+            "This filter accepts a user's name, such as John Smith."
+        )
+
+
+FILTER_ATTRIBUTES = [
+    UsersAssignedCIDsFilterAttribute,
+    UsersCIDFilterAttribute,
+    UsersFirstNameFilterAttribute,
+    UsersLastNameFilterAttribute,
+    UsersNameFilterAttribute,
+]