How to Download Files Using RTR with FalconPy SDK?

[joseraeiro]

Hello FalconPy Community,

I am currently working on a project where I need to use the FalconPy SDK to download files from a host using the RTR (Real Time Response) capabilities of CrowdStrike's Falcon platform. While I have some understanding of initiating RTR sessions and executing commands, I am specifically looking for guidance on how to correctly use the get command to retrieve files.

Here are my specific questions:

1. How do I correctly use the get command in the RTR API to retrieve a file from a host? Is there a specific method or workflow in FalconPy that facilitates this?

2. Once the get command is executed and the file is stored in the CrowdStrike cloud, what are the steps to download this file using FalconPy? Are there specific methods or endpoints I should be aware of?

3. If there are any sample scripts or code snippets that demonstrate this process, could you please share them or guide me to where I might find them?

I have gone through the FalconPy documentation but was unable to find a detailed example that fits this use case. Any guidance, examples, or pointers to relevant parts of the documentation would be greatly appreciated.

Thank you for your time and assistance.

[jshcodes]

Hi @joseraeiro -

Here are two simple variations (one for a single host and one for multiple hosts using batch commands).

Give them a try and let us know if you have questions. 😄

Download a file from a single host using RTR_ExecuteActiveResponderCommand

"""This example demonstrates downloading a single file from a single host."""
from logging import basicConfig, DEBUG
from falconpy import Hosts, RealTimeResponse

# Set these constants before executing
SHOW_API_RESPONSES = True               # Turn this off to disable debugging
hostname = ""                           # Host we are targeting, ex: "testing-host"
target_location = ""                    # File location, ex: /home/ec2-user/
target_file = "target_file.txt"         # Target file name

if SHOW_API_RESPONSES:
    basicConfig(level=DEBUG)  # Set our log level and output configuration

# Construct instances of the Service Classes we are wanting to use.
hosts = Hosts(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, debug=SHOW_API_RESPONSES)
rtr = RealTimeResponse(auth_object=hosts)
# Retrieve our target device's AID.
target_device = hosts.query_devices(filter=f"hostname:'{hostname}'")["body"]["resources"][0]
# Initialize the session
session = rtr.init_session(device_id=target_device)
session_id = session["body"]["resources"][0]["session_id"]  # Session ID
# Get the file
cloud_request_id = rtr.execute_active_responder_command(base_command="get",
                                                        device_id=target_device,
                                                        session_id=session_id,
                                                        command_string=f"get {target_location}{target_file}"
                                                        )["body"]["resources"][0]["cloud_request_id"]
# Wait for the cloud upload to complete
waiting = True
while waiting:
    download_result = rtr.check_active_responder_command_status(cloud_request_id=cloud_request_id)["body"]["resources"][0]
    if download_result["complete"] and (download_result["stdout"] or download_result["stderr"]):
        if download_result["stderr"]:
            raise SystemExit(download_result["stderr"])
        waiting = False

target_sha = None
# Retrieve a list of files uploaded during this session
file_list = rtr.list_files_v2(session_id)["body"]["resources"]
# Get the SHA256 for the file from the list files lookup
target_sha = [
    get_file["sha256"] for get_file in file_list if get_file["cloud_request_id"] == cloud_request_id
    ][0]

if target_sha:
    # Use the SHA to request this file's contents to be saved to a local 7zip archive.
    with open(f"{target_file}.7z", "wb") as save_file:
        save_file.write(rtr.get_extracted_file_contents(session_id=session_id,
                                                        sha256=target_sha,
                                                        filename=target_file)
                                                        )
# Delete the session
rtr.delete_session(session_id)

Download a file from multiple hosts using BatchGetCmd

"""This example demonstrates downloading a single file from multiple hosts."""
from logging import basicConfig, DEBUG
from falconpy import Result, Hosts,RealTimeResponse

# Set these constants before executing
SHOW_API_RESPONSES = True               # Turn this off to disable debugging
target_filter = ""                      # Host filter, ex: hostname:*'*falconpy*'
target_location = ""                    # File location, ex: /home/ec2-user/
target_file = "target_file.txt"         # Target file name

if SHOW_API_RESPONSES:
    basicConfig(level=DEBUG)  # Set our log level and output configuration

# Construct instances of the Service Classes we are wanting to use.
hosts = Hosts(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, debug=SHOW_API_RESPONSES)
rtr = RealTimeResponse(auth_object=hosts)
# Retrieve our target device AIDs.
target_devices = hosts.query_devices_by_filter_scroll(filter=target_filter)["body"]["resources"]
# Initialize a session with the host batch.
session_init: Result = rtr.batch_init_sessions(host_ids=target_devices)
batch_id = session_init["body"]["batch_id"]  # Grab the batch ID
# Issue a batch get command
result = rtr.batch_get_command(batch_id=batch_id, file_path=f"{target_location}{target_file}")
# Grab the batch ID
batch_req_id = result["body"]["batch_get_cmd_req_id"]
# Check the status of the batch command
status = rtr.batch_get_command_status(batch_req_id)
# Loop thru the results
for device_id, device_result in status["body"]["resources"].items():
    session_id = device_result["session_id"]            # Session ID
    file_sha = device_result["sha256"]                  # File SHA256
    with open(f"{device_id}.7z", "wb") as save_file:    # Save to a file named after the device ID
        save_file.write(rtr.get_extracted_file_contents(session_id=session_id,
                                                        sha256=file_sha,
                                                        filename=target_file,
                                                        ))
        rtr.delete_session(session_id)

There are also a few different RTR examples located in the Sample Library.

[jshcodes]

Hi @joseraeiro -

I'm able to use the single file download example above on a Windows host successfully.

I've made the following two changes:

• Updated target_location:

  target_location = "C:\\Users\\Administrator\\Desktop\\"

• Changed how target_sha was calculated to look at cloud_request_id instead (I've updated the example above as well):

  target_sha = [
      get_file["sha256"] for get_file in file_list if get_file["cloud_request_id"] == cloud_request_id
      ][0]

The error above states "Check your filename. Couldn't find 'C:\\known_hosts'\n".

An apostrophe (') and a line feed (\n) appear to be present within the target_location or target_file variables.

[joseraeiro]

But they aren't present! I just made a minor change to the script: used the device_id directly instead of fetching it by the hostname, which shouldn't interfere with this step... I even tried using the same directory structure as you did.

But nevertheless, I'm getting the same result:

DEBUG:urllib3.connectionpool:Starting new HTTPS connection (1): api.eu-1.crowdstrike.com:443
DEBUG:urllib3.connectionpool:https://api.eu-1.crowdstrike.com:443 "GET /real-time-response/entities/active-responder-command/v1?sequence_id=0&cloud_request_id=9c0b39bb-6b72-4871-ae5d-5ba690a0d728 HTTP/1.1" 200 344
DEBUG:falconpy._auth_object._falcon_interface:RECEIVED: Content returned in application/json format
DEBUG:falconpy._auth_object._falcon_interface:STATUS CODE: 200
DEBUG:falconpy._auth_object._falcon_interface:RESULT: {'status_code': 200, 'headers': {'Server': 'nginx', 'Date': 'Thu, 23 Nov 2023 09:36:44 GMT', 'Content-Type': 'application/json', 'Content-Length': '344', 'Connection': 'keep-alive', 'Content-Encoding': 'gzip', 'Strict-Transport-Security': 'max-age=15724800; includeSubDomains, max-age=31536000; includeSubDomains', 'X-Cs-Region': 'eu-1', 'X-Cs-Traceid': '801feaed-6b3b-4904-9f9e-120b8b54f609', 'X-Ratelimit-Limit': '6000', 'X-Ratelimit-Remaining': '5979'}, 'body': {'meta': {'query_time': 0.311482883, 'powered_by': 'empower-api', 'trace_id': '801feaed-6b3b-4904-9f9e-120b8b54f609'}, 'resources': [{'session_id': 'c01e7d15-aabf-456c-8425-b16d9dd525da', 'task_id': '9c0b39bb-6b72-4871-ae5d-5ba690a0d728', 'complete': True, 'stdout': '', 'stderr': "Check your filename. Couldn't find 'C:\\Users\\Administrator\\Desktop\\desktop.ini'\n", 'base_command': 'get'}], 'errors': []}}
DEBUG:falconpy._auth_object._falcon_interface:OPERATION: RTR_CheckActiveResponderCommandStatus
DEBUG:falconpy._auth_object._falcon_interface:ENDPOINT: https://api.eu-1.crowdstrike.com/real-time-response/entities/active-responder-command/v1 (GET)
DEBUG:falconpy._auth_object._falcon_interface:HEADERS: {'Authorization': 'Bearer REDACTED', 'User-Agent': 'crowdstrike-falconpy/1.3.0', 'CrowdStrike-SDK': 'crowdstrike-falconpy/1.3.0'}
DEBUG:falconpy._auth_object._falcon_interface:PARAMETERS: {'sequence_id': 0, 'cloud_request_id': '9c0b39bb-6b72-4871-ae5d-5ba690a0d728'}
DEBUG:falconpy._auth_object._falcon_interface:BODY: None
DEBUG:falconpy._auth_object._falcon_interface:DATA: None

What's going on here, you think?

"""This example demonstrates downloading a single file from a single host."""
from logging import basicConfig, DEBUG
from falconpy import Hosts, RealTimeResponse

# Set these constants before executing
SHOW_API_RESPONSES = True               # Turn this off to disable debugging
hostname = ""
target_location = "C:\\Users\\Administrator\\Desktop\\"      # File location, ex: /home/ec2-user/
target_file = "desktop.ini"         # Target file name
device_id = ""
CLIENT_ID = ""
CLIENT_SECRET = ""
MEMBER_CID = ""
if SHOW_API_RESPONSES:
    basicConfig(level=DEBUG)  # Set our log level and output configuration

# Construct instances of the Service Classes we are wanting to use.
hosts = Hosts(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, member_cid=MEMBER_CID, debug=SHOW_API_RESPONSES)
rtr = RealTimeResponse(auth_object=hosts)
# Retrieve our target device's AID.
#target_device = hosts.query_devices(filter=f"hostname:'{hostname}'")["body"]["resources"][0]
# Initialize the session
session = rtr.init_session(device_id=device_id)
session_id = session["body"]["resources"][0]["session_id"]  # Session ID
# Get the file
cloud_request_id = rtr.execute_active_responder_command(base_command="get",
                                                        device_id=device_id,
                                                        session_id=session_id,
                                                        command_string=f"get {target_location}{target_file}"
                                                        )["body"]["resources"][0]["cloud_request_id"]
# Wait for the cloud upload to complete
waiting = True
while waiting:
    download_result = rtr.check_active_responder_command_status(cloud_request_id=cloud_request_id)["body"]["resources"][0]
    if download_result["complete"] and download_result["stdout"]:
        waiting = False

target_sha = None
# Retrieve a list of files uploaded during this session
file_list = rtr.list_files_v2(session_id)["body"]["resources"]
# Get the SHA256 for the file from the list files lookup
target_sha = [
    get_file["sha256"] for get_file in file_list if get_file["cloud_request_id"] == cloud_request_id
    ][0]

if target_sha:
    # Use the SHA to request this file's contents to be saved to a local 7zip archive.
    with open(f"{target_file}.7z", "wb") as save_file:
        save_file.write(rtr.get_extracted_file_contents(session_id=session_id,
                                                        sha256=target_sha,
                                                        filename=target_file)
                                                        )
# Delete the session
rtr.delete_session(session_id)

[joseraeiro]

I even tried adding the following lines just to check:

target_location = target_location.strip()
target_file = target_file.strip()
print(f"Final command string: get {target_location}{target_file}")

Which produces the following output:

Final command string: get C:\Users\Administrator\Desktop\desktop.ini

But the same result. Not sure where that '\n is coming from.

[jshcodes]

Looks like the apostrophes and the \n are added as part of the error response. (I asked for a file that definitely did not exist to confirm this.)

What if you try the other example? It should work with only one host as the target. Also confirm your API keys have READ and WRITE for all Real Time Response scopes.

Side note: I updated our simple example above to check for stderr and exit.

[joseraeiro]

It works now! Thank you very much for all your kind help and support!

[RoemIko]

Hi @jshcodes,

In addition to this question.
How would you handle slow download speeds/large files over multiple hosts?

In this part for example:

status = rtr.batch_get_command_status(batch_req_id)
# Loop thru the results
for device_id, device_result in status["body"]["resources"].items():
    session_id = device_result["session_id"]            # Session ID
    file_sha = device_result["sha256"]                  # File SHA256
    with open(f"{device_id}.7z", "wb") as save_file:    # Save to a file named after the device ID
        save_file.write(rtr.get_extracted_file_contents(session_id=session_id,
                                                        sha256=file_sha,
                                                        filename=target_file,
                                                        ))
        rtr.delete_session(session_id)

I noticed that when a file is large or there is slow internet the status["body"]["resources"] value is empty, so the API thinks there is no file to collect, however in the UI is see that its at 1% for example

[joseraeiro]

I have tried to implement your solution to asynchronously wait for a file to complete the get process, but to download a file from a single host.

Code so far:

import logging
from falconpy import Hosts, RealTimeResponse
from flask import Flask, request, jsonify
import time
import re
import os
from concurrent.futures import ThreadPoolExecutor, as_completed

# Configure logging
logging.basicConfig(filename="application.log", filemode='a', level=logging.DEBUG,
                    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S')

app = Flask(__name__)

def extract_last_segment(parent_string):
    return parent_string.split('\\')[-1]

def convert_file_path(path):
    match = re.match(r"\\Device\\HarddiskVolume(\d+)\\(.+)", path)
    if match:
        drive_letter = chr(67 + int(match.group(1)) - 1)
        new_path = f"{drive_letter}:\\{match.group(2)}"
        return new_path
    return path

def get_hosts(device_id, client_id, client_secret):
    BASE_URL = "https://api.eu-1.crowdstrike.com"
    falcon = Hosts(base_url=BASE_URL, client_id=client_id, client_secret=client_secret)
    logging.info("Attempting to retrieve MEMBER_CID for device_id: %s", device_id)
    response = falcon.get_device_details_v2(ids=device_id)
    member_cid = response.get("body", {}).get("resources", [{}])[0].get("cid")
    if member_cid:
        logging.info("Successfully retrieved MEMBER_CID: %s", member_cid)
        return member_cid
    else:
        logging.error("Unable to retrieve MEMBER_CID from the response.")
        return None

def wait_for_file(session_id, cloud_request_id, modified_path, hosts):
    """Asynchronous function to wait for a file to complete the get process.
    After processing completes, delete the session."""
    rtr = RealTimeResponse(auth_object=hosts)
    waiting = True
    while waiting:
        file_list = rtr.list_files_v2(session_id=session_id)["body"]["resources"]
        for item in [f for f in file_list if f["cloud_request_id"] == cloud_request_id]:
            if item["complete"]:
                waiting = False
                file_download = rtr.get_extracted_file_contents(session_id=session_id, sha256=item["sha256"], filename=extract_last_segment(modified_path))
                if isinstance(file_download, bytes):
                    target_directory = "/home/user/RTR_testes/downloaded_files"
                    file_name = f"{item['sha256']}_{extract_last_segment(modified_path)}.7z"
                    full_file_path = os.path.join(target_directory, file_name)
                    with open(full_file_path, "wb") as save_file:
                        save_file.write(file_download)
                    logging.info("File downloaded successfully.")
                    rtr.delete_session(session_id)  # Close this specific host session
                    return True
                else:
                    logging.error("Failed to retrieve file contents.")
                    rtr.delete_session(session_id)  # Close this specific host session
                    return False
            else:
                time.sleep(1)
    rtr.delete_session(session_id)  # Close this specific host session
    return False

@app.route('/webhook6', methods=['POST'])
def webhook():
    data = request.json
    logging.info("Received data: %s", data)

    original_file_path = data.get('data', {}).get('detections.file_path', "C:\\Users\\Administrator\\Desktop\\desktop.ini")
    device_id = data.get('data', {}).get('detections.sensor_id', "default_sensor_id")

    target_location = convert_file_path(original_file_path)

    CLIENT_ID = ""
    CLIENT_SECRET = ""

    member_cid = get_hosts(device_id, CLIENT_ID, CLIENT_SECRET)

    BASE_URL = "https://api.eu-1.crowdstrike.com"

    hosts = Hosts(base_url=BASE_URL, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, member_cid=member_cid)
    rtr = RealTimeResponse(auth_object=hosts)

    success = False
    session_id = None

    for drive_letter in ['C:', 'D:', 'E:', 'F:', 'G:', 'H:', 'I:', 'J:']:
        modified_path = re.sub(r"^[A-Z]:", drive_letter, target_location)
        logging.info(f"Trying {modified_path}")

        session = rtr.init_session(device_id=device_id)
        session_id = session["body"]["resources"][0]["session_id"]
        logging.info("Session initialized. Session ID: %s", session_id)

        cloud_request_id = rtr.execute_active_responder_command(
            base_command="get",
            device_id=device_id,
            session_id=session_id,
            command_string=f'get "{modified_path}"'
        )["body"]["resources"][0]["cloud_request_id"]
        time.sleep(1)

        with ThreadPoolExecutor() as executor:
            future = executor.submit(wait_for_file, session_id, cloud_request_id, modified_path, hosts)
            try:
                success = future.result()
                if success:
                    break
            except Exception as exc:
                logging.error(f"{exc} occurred during processing on {drive_letter} drive.")

    logging.info("Webhook processing complete.")
    return jsonify({'status': 'success', 'message': 'Webhook processed'})

if __name__ == '__main__':
    app.run(debug=True)

This code seems to be working, except when it doesn't find the file on drive C: It keeps trying until the Gunicorn instance times out.

What would be the best way to handle this, in order to be able to iterate through all the specified drives, you think?

[joseraeiro]

I thought about counting the number of times it tries to retrieve the file and/or limit them by time. I would have to count the maximum tries/time that it used to retrieve a file and then add some more?

I was in the hopes for a more elegant solution to this problem. Can anyone suggest anything?

Other than that, it's working flawlessly.

It never misses a single file (except in aforementioned case), whereas the other approach would fail many times and without any obvious pattern.

💪💪💪

[jshcodes]

We're not seeing the stdout response for the get request like we do when we call BatchGetCmd.

Instead, we can take the cloud_request_id and provide it to the RTR_CheckActiveResponderCommandStatus operation. This will return a value in stderr if the file was not found.

{
    "status_code": 200,
    "headers": {
        "Server": "nginx",
        "Date": "Fri, 12 Jan 2024 19:27:46 GMT",
        "Content-Type": "application/json",
        "Content-Length": "311",
        "Connection": "keep-alive",
        "Content-Encoding": "gzip",
        "X-Cs-Region": "us-1",
        "X-Cs-Traceid": "REDACTED",
        "X-Ratelimit-Limit": "6000",
        "X-Ratelimit-Remaining": "5998",
        "Strict-Transport-Security": "max-age=31536000; includeSubDomains"
    },
    "body": {
        "meta": {
            "query_time": 0.312158753,
            "powered_by": "empower-api",
            "trace_id": "REDACTED"
        },
        "resources": [
            {
                "session_id": "REDACTED",
                "task_id": "REDACTED",
                "complete": true,
                "stdout": "",
                "stderr": "get: test.txt: No such file or directory\n",
                "base_command": "get"
            }
        ],
        "errors": []
    }
}

[joseraeiro]

I have tried what's basically the repetition of the method previously used to check if the file had been downloaded, before trying to retrieve it.

It works perfectly, thank you very much, I just had to tweak it to wait to finish. And so, I'm perfectly able to check other drives if it's not found on C:[path].

There isn't any way of guessing which drive Crowdstrike refers to when it uses the number at the end of HarddiskVolume, or is there?

[jshcodes]

That's dependent upon where Window's has the drive mounted. You could try pulling it from the registry using the reg command within the RTR_ExecuteAdminCommand operation. (HKEY_LOCAL_MACHINE\SYSTEM\MountedDevices may be a good place to start.)

You could also make a -Raw runscript call using RTR_ExecuteAdminCommand. Then you could execute diskpart (may be kind of ugly, and will require parsing) or a PowerShell script. Since this had a PowerShell option, I hit up @bk-cs to get his solution. 😃

@bk-cs's Convert-Path PowerShell function

function Convert-Path([Parameter(Mandatory=$true)][string]$String){
  $sb=New-Object System.Text.StringBuilder(65536)
  $k32=Add-Type -MemberDefinition @'
[DllImport("kernel32.dll",SetLastError=true)]
public static extern uint QueryDosDevice(string lpDeviceName,System.Text.StringBuilder lpTargetPath,uint ucchMax);
'@ -Name Kernel32 -Namespace Win32 -PassThru
  gwmi -Query "Select DriveLetter FROM Win32_Volume WHERE DriveLetter LIKE '%'"|
  select -ExpandProperty DriveLetter|%{
    [void]$k32::QueryDosDevice($_,$sb,65536)
    [string]$ntp=[regex]::Escape($sb.ToString())
    if($String -match $ntp){$String -replace $ntp,$_}
  }
}

Call this function like this: if($Path -match '\\HarddiskVolume\d\\'){$Path=Convert-Path $Path}. You may have to upload this as a Cloud File, and execute it as a script where you provide the volume name as an argument.