Mass triaging unmanaged assets

[AAuraa]

Context

Recently within my environment, I have been working on getting a bearing on our unmanaged assets pulled from AD that reportedly have no sensor. After digging around, I realized many of the devices listed actually had an entry in the managed assets section, meaning they did have a sensor, so I could set the status of the corresponding unmanaged asset to reviewed. However, there was no good way for me to mass sort these by hostname, and with around 800, manually doing so was not a great option.

I decided to try and find a better method to approach the issue, and using direct API calls, still had an issue when I was trying to use the exposure-management calls with an internal_id. So I decided to use FalconPy to automate this process from a CSV of hostnames.

This project also requires a workflow component, since I believe the API call that allows for triage of assets via exposure management is either broken, or I am just using it entirely wrong. Whatever the case, the workflow for asset triage works perfectly, I just need to get the right internal_id, and management type.

Idea link

https://us-2.ideas.crowdstrike.com/ideas/IDEA-I-14230

Python script

mass_unmanaged_host_triage.py

""".
Mass Triage Unmanaged Hosts

FalconPy v1.4.4

When provided a CSV file of hostnames as listed
in the unmanaged asset menu in exposure management
loop through them all and set to the desired triage
settings based on the configuration in your workflow.

The workflow currently sends an email to your desired email
after completing the iterations

Known bug: Sometimes on iteration failure, the workflow
sends an email despite the loop not exiting

You must provide your API credentials to this application via the command line
As well as your workflow ID, and the file path to the hostname list

Creation date: 07.30.24 - IanG@PISD
"""

from csv import reader
from argparse import ArgumentParser, RawTextHelpFormatter
import time
from falconpy import Discover, Workflows

# Much of the code taken from the FalconPy quick-start sample
# Many thanks jshcodes, wozboz, and others

parser = ArgumentParser(description=__doc__, formatter_class=RawTextHelpFormatter)
req = parser.add_argument_group("required arguments")
req.add_argument("-k", "--falcon_client_id",
                 help="CrowdStrike Falcon API Client ID",
                 required=True
                 )
req.add_argument("-s", "--falcon_client_secret",
                 help="CrowdStrike Falcon API Client Secret",
                 required=True
                 )
parser.add_argument("-b", "--base_url",
                    help="CrowdStrike base URL (only required for GovCloud, pass usgov1)",
                    required=False,
                    default="auto"
                    )
req.add_argument("-c", "--falcon_customer_id",
                 help="CrowdStrike Customer ID (cid)",
                 required=True
                 )
req.add_argument("-w", "--fusion_workflow_id",
                 help="ID of the Fusion Workflow associated with this script",
                 required=True
                 )
req.add_argument("-f", "--file",
                 help="File with hostnames to use as input",
                 required=True,
                 )
parser.add_argument("-t", "--entity_type",
                    help="Entity management type (unmanaged, managed, unsupported)",
                    required=False,
                    default="unmanaged"
                    )
parser.add_argument("-d", "--debug",
                    help="Enable API debugging",
                    default=False
                    )

args = parser.parse_args()

print("Welcome to the mass triage unmanaged assets tool")

# Authenticate with our environment so we can make calls
# Don't hardcode these values!!!!
discover_agent = Discover(client_id=args.falcon_client_id,
                          client_secret=args.falcon_client_secret,
                          base_url=args.base_url, debug=args.debug
                          )

# Start our stopwatch
start_time = time.time()

# Make another agent using the same auth_object
# So we can make calls to the workflows API
workflows_user_agent = Workflows(auth_object=discover_agent)

# Pull the list of hostnames from our CSV file
serial_data = []
with open(args.file, mode="r", encoding="UTF-8") as host_list:
    line_reader = reader(host_list)
    for row in line_reader:
        serial_data.append(row[0])

print(f"Located {len(serial_data)} hostnames")

unmanaged_ids = []
workflow_schema = {"ListOfDevices": []}

print("Beginning serial scan")

# Locate the internal IDs associated with our entities (hosts/assets)
# Then store them for use in the workflow
for serial in serial_data:
    if args.debug is True:
        print("Checking " + serial)

    # Retrieve a list of hosts matching our search filter
    # Note the entity type
    filt = "hostname:'"+serial+"'+entity_type:'"+args.entity_type.lower()+"'"
    results = discover_agent.query_hosts(filter=filt)

    # Our request succeeded
    # Grab the response data and put what we need into our variables
    if results["status_code"] == 200:
        hosts_found = results["body"]["resources"]
        for host_found in hosts_found:
            unmanaged_ids.append({"InternalID": host_found, "Serial": serial})
            wf_data = {"EntityID": host_found, "ManagementType": "Unmanaged"}
            workflow_schema["ListOfDevices"].append(wf_data)
    # We encountered some undesirable response code
    # Output the errors
    else:
        error_detail = results["body"]["errors"]
        for error in error_detail:
            error_code = error["code"]
            error_message = error["message"]
            print(f"[Error {error_code}] {error_message}")


print("Finished scanning for internal IDs")

if args.debug is True:
    print(unmanaged_ids)
    print()
    print(workflow_schema)

# Begin the longer part!
print("Beginning workflow running")
workflow_attempt = workflows_user_agent.execute(
    definition_id=args.fusion_workflow_id,
    execution_cid=args.falcon_customer_id,
    body=workflow_schema
)

if args.debug is True:
    print(workflow_attempt)

# No errors, celebrate!
if workflow_attempt["status_code"] == 200:
    print("Workflow execution request sent with no errors")
# Errors? Uhhh, check what happened then
else:
    error_detail = workflow_attempt["body"]["errors"]
    for error in error_detail:
        error_code = error["code"]
        error_message = error["message"]
        print(f"[Error {error_code}] {error_message}")


# Get the time taken to execute and output to the user
end_time = time.time()
total_time = round(end_time - start_time, 2)

print(f"The tool has finished running in {total_time} seconds")

Workflow

The workflow can be found below in .txt format, make sure you change it to .yml and when you import it, change the settings to match what you want (edit email fields, etc.).
mass_triage_unmanaged_assets.txt

Runtime

I found when running this with 800 or so assets it took about 7 hours for the workflow iterations to all finish, but when I ran it again with 75 devices, it took about 2 minutes. So your experience may vary, but you can monitor the execution status to ensure its doing something.

If you plan to run this, make sure to read the required arguments for the command-line so it executes properly.
Also, for file input file, a CSV of one hostname per line with no headers is the correct format. See below for example.
hosttest.csv

API Scopes

Lastly, the API client scopes needed to perform these are:
hosts:read
assets:read
workflow:read
workflow:write

Hopefully this provides enough info for anyone looking to do this for their own environment. I do not plan to support this fully, so sorry if you need help, I will occasionally check in though if people do post issues.

[jshcodes]

Hi @AAuraa -

Thank you for sharing your solution! 😃

Do you mind if we add this to our sample library?

[AAuraa]

Of course, go right ahead! I was planning to write tests and get it ready for a pull request but I have not found the time and I am new to contributions, so I figured in the meantime I could just post as a show and tell entry.

[jshcodes]

We're happy to assist you with a PR from your fork or we can post it on your behalf, whichever you prefer.

Either solution will get you added to our authors page as a contributor. 🚀 😄

[AAuraa]

You are welcome to post on my behalf, I am not as experienced with Git as I should be admittedly... so currently much of my code is revised without a formal repository. 😅

[jshcodes]

You got it! We'll get this posted ASAP.

Thank you again for your contribution!! 🙇