Query bulk user roles #1070
-
|
Hello community. I'm developing a health-check script that would get all user's granted role and check how many users are falcon admin and how many are RTR admin. I use the user_management.get_user_grants to query the user granted roles. [https://falconpy.io/Service-Collections/User-Management.html#combineduserrolesv1] The problem here I'm facing is that this method could only retrieve roles for 1 user at a time. Each API call would run for around 5 seconds which means it would become very long if I was to query 50+ users. Is there a way I can bulk query all users using 1 API call instead of N-user times? |
Beta Was this translation helpful? Give feedback.
Replies: 1 comment 5 replies
-
|
Hi @dkang-firmus - I've confirmed this API operation only accepts one UUID per request. Adding support for multiple IDs is an excellent idea submission. In the interim, I've put together a sample for you that leverages A few notes:
Let us know if there are any questions! 😄 r"""Threaded user grant lookup example.
______ __ _______ __ __ __
| |.----.-----.--.--.--.--| | __| |_.----.|__| |--.-----.
| ---|| _| _ | | | | _ |__ | _| _|| | <| -__|
|______||__| |_____|________|_____|_______|____|__| |__|__|__|_____|
___ ___ ___ ___ __
| Y .-----.-----.----.| Y .---.-.-----.---.-.-----.-----.--------.-----.-----| |_
|. | |__ --| -__| _||. | _ | | _ | _ | -__| | -__| | _|
|. | |_____|_____|__| |. \_/ |___._|__|__|___._|___ |_____|__|__|__|_____|__|__|____|
|: 1 | |: | | |_____|
|::.. . | |::.|:. | FalconPy v1.3.3
`-------' `--- ---'
Asynchronously retrieve all user grants for every user defined within the tenant and output
the results to a comma-delimited text file. When not specified, this file is named user_grants.csv.
Creation date: 11.13.2023 - jshcodes@CrowdStrike
"""
import os
import csv
import logging
from argparse import ArgumentParser, RawTextHelpFormatter
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
try:
from falconpy import UserManagement
except ImportError as no_falconpy:
raise ImportError("The CrowdStrike FalconPy library must be installed.") from no_falconpy
def consume_arguments():
"""Retrieve any provided command line arguments."""
parser = ArgumentParser(description=__doc__, formatter_class=RawTextHelpFormatter)
parser.add_argument("-d", "--debug", help="Enable debug.", default=False, action="store_true")
parser.add_argument("-o", "--output", help="CSV output filename.", default="user_grants.csv")
auth = parser.add_argument_group("Authentication arguments "
"(not required if using environment authentication)")
auth.add_argument("-k", "--client_id",
help="Falcon API client ID",
default=os.getenv("FALCON_CLIENT_ID")
)
auth.add_argument("-s", "--client_secret",
help="Falcon API client secret",
default=os.getenv("FALCON_CLIENT_SECRET")
)
return parser.parse_args(), parser
def get_grants(interface: UserManagement, uuid: str):
"""Retrieve all grants for the user UUID in question."""
print(f" Retrieving grant detail for {uuid}", end="\r")
return interface.get_user_grants(uuid).data
def get_grant_data(sdk: UserManagement):
"""Retrieve all user UUIDs within the tenant and then retrieve the grants for each."""
running = True
user_ids = []
user_grants = []
offset = None
while running:
user_lookup = sdk.query_users(limit=500, offset=offset)
user_ids.extend(user_lookup.data)
total = user_lookup.total
offset = len(user_ids)
with ThreadPoolExecutor() as executor:
futures = {
executor.submit(get_grants, sdk, user_id) for user_id in user_lookup.data
}
for fut in futures:
user_grants.extend(fut.result())
if len(user_ids) >= total:
running = False
return user_ids, user_grants
def write_grant_results(user_grants: list, output_file: str):
"""Write grant details to the specified CSV file."""
with open(output_file, "w", newline="", encoding="utf-8") as csv_file:
writer = csv.writer(csv_file)
writer.writerow(user_grants[0].keys()) # Header row
for grants in user_grants:
# Data rows
writer.writerow(grants.values())
if __name__ == "__main__":
begin = datetime.now().timestamp()
parsed, handler = consume_arguments()
if not parsed.client_id or not parsed.client_secret:
handler.print_help()
raise SystemExit(
"\nYou must provide API credentials via the environment variables\n"
"FALCON_CLIENT_ID and FALCON_CLIENT_SECRET or you must provide\n"
"them using the '-k' and '-s' command line arguments."
)
if parsed.debug:
logging.basicConfig(level=logging.DEBUG)
users = UserManagement(client_id=parsed.client_id,
client_secret=parsed.client_secret,
debug=parsed.debug,
pythonic=True
)
id_list, grant_list = get_grant_data(users)
write_grant_results(grant_list, parsed.output)
print(" " * 80) # Clear the last status update line
print(f"{len(id_list):,} total users identified.")
print(f"{len(grant_list):,} total grants retrieved.")
print(f"Results saved to {parsed.output}.")
print(f"Total processing time: {datetime.now().timestamp() - begin:.2f} seconds.") |
Beta Was this translation helpful? Give feedback.
-
|
Hi @dkang-firmus - This solution was expanded to speak to MSSP scenarios and is now available as a sample! 🎉 |
Beta Was this translation helpful? Give feedback.
-
|
Hello jshcodes, thank you for assisting me out on this. I need to review on your code and see why it is not working in my environment. Perhaps some system dependencies are not met. Previously I thought about asynchronicity but it doesn't seem to work in my environment using asyncio library. The outcome was, multiple API calls are sent asynchronously, but the response I get is queued back-to-back and not at the same time. I also tried running your code but getting output related to network connectivity issue. I'm using US-1. Here's the error code for documentation purpose. Besides, I'm using a .env to import my credentials. I needed to add in a couple of lines of code 'load_dotenv()' to import them. Could you verify if this is required? |
Beta Was this translation helpful? Give feedback.
-
|
I agree, these look like connectivity errors. I would definitely look into the asyncio portion of the code being executed. (One reason I tend to lean on You can confirm general network availability to the API with the following code: import logging
from falconpy import Hosts
logging.basicConfig(level=logging.DEBUG)
hosts = Hosts(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, debug=True)
print(hosts.query_devices_by_filter_scroll())Regarding using the sample code I provided in the previous response; you don't have to provide credentials using the environment, you can use the keyword arguments to provide them instead. |
Beta Was this translation helpful? Give feedback.
-
|
Hey Josh, I'm sorry for not getting back to you after a month time. Work has been so packed up and I could only continue this mini project once settling down everything. But I will finish this since I've brought this up. Your suggestion in using the logging library does help me a lot. I've found the reason of getting the 'connection' failure. Seems like my computer wasn't able to 'multithread' that many requests. Refer below: with ThreadPoolExecutor(max_workers=8) as executor However, seems like the program didn't send the request concurrently despite using concurrent.futures library. Refer below output: PS C:\REDACTED> python .\joshs.py -k REDACTED -s REDACTED 15 total users identified. I compare the result with my original code (requests are not concurrent, but queued back-to-back), the time to collect all user grants is also around 75 seconds. Therefore, this has proven that the requests weren't sent concurrently. I tried using ThreadPoolExecutor in running some simple program and the multithreading works just fine. Just when it's used for calling API requests, it gets queued back-to-back. Could you share with me if there's any dependencies need to be met before multithread calling the API requests? |
Beta Was this translation helpful? Give feedback.
-
|
Hello Josh, it's me again. I've setup a new Linux environment. Previously was running on host Windows 10. After trying on the new Linux env, everything has been very fast. I supposed the RCA was coming from my hardware instead. Thanks for the assistance. |
Beta Was this translation helpful? Give feedback.
Hi @dkang-firmus -
This solution was expanded to speak to MSSP scenarios and is now available as a sample! 🎉