✨ 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,
+]