Skip to main content
Solved

Chronicle Search API to Pull UDM data

  • December 12, 2024
  • 23 replies
  • 962 views
russell_pfeifer
Forum|alt.badge.img+7

Good morning--

Looking for some feedback on the feasibility of pulling large UDM datasets out of Chronicle using a curl command. We previously did this successfully using Splunk but I'd prefer to not waste my time going down any rabbit-holes if it is simply not possible in Chronicle. 

Basically I'd like to query our Tenable vulnerability data (currently parsed out into UDM fields in Chronicle) and export 15 or so of the fields to a csv which would then be uploaded to JIRA for remediation. 

I see within the API documentation there is a way to search UDM but I don't see any documentation around pulling all events from a specific log type. 

I've been using the following documentation as a guide so far:

https://cloud.google.com/chronicle/docs/reference/search-api

Thanks in advance for any help you can provide

Best answer by mikewilusz

The UDM search endpoint is documented here: https://cloud.google.com/chronicle/docs/reference/search-api#udmsearch

You can see one of the parameters is a query, so for your example if you want to pull your Tenable data, the query would look like: 

metadata.log_type = "TENABLE_IO"

That should retrieve all the Tenable events from the time span you specify in the API request.

-mike 

Russell P

    23 replies

    mikewilusz
    Staff
    Forum|alt.badge.img+10
    • mikewilusz
    • Staff
    • Answer
    • December 12, 2024

    The UDM search endpoint is documented here: https://cloud.google.com/chronicle/docs/reference/search-api#udmsearch

    You can see one of the parameters is a query, so for your example if you want to pull your Tenable data, the query would look like: 

    metadata.log_type = "TENABLE_IO"

    That should retrieve all the Tenable events from the time span you specify in the API request.

    -mike 

      russell_pfeifer
      Forum|alt.badge.img+7

      Understood -- thank you for confirming -- glad I'm not wasting my time. 

      Russell P
        mwisener
        Forum|alt.badge.img+3
        • mwisenerBronze 1
        • Bronze 1
        • December 13, 2024

        FYI udmSearch is limited to 10k results, no paging, so you would need to break up the queries into time chunks without really knowing how much data is there all while keeping under the rate limit of 2 queries per minute.

          mwisener
          Forum|alt.badge.img+3
          • mwisenerBronze 1
          • Bronze 1
          • December 16, 2024

          FYI udmSearch is limited to 10k results, no paging, so you would need to break up the queries into time chunks without really knowing how much data is there all while keeping under the rate limit of 2 queries per minute.


          Forgot to mention the only real solution to mass export normalized data is the bigquery export option. Request google export datalake.events to your own gcp project so you can query it at your leisure. It's not real-time and you pay storage and compute costs (assuming your instance is associated with a gcp project)

           

          https://cloud.google.com/chronicle/docs/preview/cloud-integration/export-to-customer-managed-project

            russell_pfeifer
            Forum|alt.badge.img+7

            Ahh OK so this may explain why my shell script is not actually able to pull back any data. Are you saying that if I don't use the bigquery / datalake.events option in a gcp project I'm not able to simply pull UDM events from my Chronicle instance URI? This the way we were able to pull this data using Splunk.

            For example:

            curl -v --ca-native https://backstory.chronicle.security/events:udmSearch?query=metadata.log_type="TENABLE_IO" --user email\\*TOKEN*

            As of now (not using bigquery) this does not return any events from the Tenable index. 

            Thanks

             

            Russell P
            mwisener
            Forum|alt.badge.img+3
            • mwisenerBronze 1
            • Bronze 1
            • December 16, 2024

            No you can do that, it will just be limited to 10k rows. 

            See: https://cloud.google.com/chronicle/docs/reference/search-api#udmsearch

              russell_pfeifer
              Forum|alt.badge.img+7

              Understood - the documentation certainly makes it look like its possible although I'm still unable to pull any UDM data with a curl command. 

              Have you personally been able to pull UDM data using this approach? If so, did you have to integrate Google sign in before you could access googleapis.com? The following documentation makes it seem like that might be pre-requisite:

              https://developers.google.com/identity/sign-in/web/sign-in

               

              Russell P
              mwisener
              Forum|alt.badge.img+3
              • mwisenerBronze 1
              • Bronze 1
              • December 18, 2024

              You need a chronicle backstory key which support can provide you if you do not have it.

              I do query udmSearch (via python) - I do not use curl. There is authentication information on the reference docs provided. I use python so can use the examples, and google already provides libraries like google.oauth2 service_account that handles all the authentication stuff.

              https://cloud.google.com/chronicle/docs/reference/search-api#getting_api_authentication_credentials

              there is also:

              https://github.com/chronicle/api-samples-python/blob/master/search/udm_search.py

               

                russell_pfeifer
                Forum|alt.badge.img+7

                You need a chronicle backstory key which support can provide you if you do not have it.

                I do query udmSearch (via python) - I do not use curl. There is authentication information on the reference docs provided. I use python so can use the examples, and google already provides libraries like google.oauth2 service_account that handles all the authentication stuff.

                https://cloud.google.com/chronicle/docs/reference/search-api#getting_api_authentication_credentials

                there is also:

                https://github.com/chronicle/api-samples-python/blob/master/search/udm_search.py

                 


                There seems to be a lot more documentation / support for python so I am going to switch over to that for my script. Thanks for all the information!

                Russell P
                russell_pfeifer
                Forum|alt.badge.img+7

                So I've gotten the Python query up to the point where I can generate a JSON file pulling UDM events from the Tenable log type. I have also incorporated a function that writes the JSON response to a CSV. 

                Currently though - the query parameters are quite simple: 

                query = 'metadata.log_type = "TENABLE_IO"'

                Where the query seems to be failing is when I go to incorporate statistical functions into my query, specifically:

                # Simplified query parameters query =''' events: metadata.log_type = "TENABLE_IO" $pluginID = security_result.rule_id $assetID = principal.asset.product_object_id match: $pluginID, $assetID outcome: $Description = array_distinct(extensions.vulns.vulnerabilities.description) $FQDN = array_distinct(principal.hostname) '''

                Once the string is run with that query it returns this error:

                Error: 400 - { "error": { "code": 400, "message": "generic::invalid_argument: compilation error query uses a feature that is not yet allowed: invalid argument", "status": "INVALID_ARGUMENT" } }

                Is this simply a limitation of the current query parameters?

                Full python string here:

                # Set up API credentials and scopes SCOPES = ['https://www.googleapis.com/auth/chronicle-backstory'] SERVICE_ACCOUNT_FILE = r'C:\\API_keys\\OAuth_2_Keys.json' # Authenticate using the service account key credentials = service_account.Credentials.from_service_account_file( SERVICE_ACCOUNT_FILE, scopes=SCOPES ) http_session = AuthorizedSession(credentials) # API endpoint base_url = "https://backstory.googleapis.com/v1/events:udmSearch" # Simplified query parameters query =''' events: metadata.log_type = "TENABLE_IO" $pluginID = security_result.rule_id $assetID = principal.asset.product_object_id match: $pluginID, $assetID outcome: $Description = array_distinct(extensions.vulns.vulnerabilities.description) $FQDN = array_distinct(principal.hostname) ''' start_time = "2024-11-01T00:00:00Z" # Replace with desired start time end_time = "2024-12-20T01:00:00Z" # Replace with desired end time limit = 100 # Replace with desired limit # Build the query string params = { "query": query, "time_range.start_time": start_time, "time_range.end_time": end_time, "limit": limit } # Encode the query string for the URL (properly handle spaces and special characters) query_string = urlencode(params) # Full URL with query parameters url = f"{base_url}?{query_string}" # Make the GET request response = http_session.request("GET", url) # Function to write JSON response to CSV file def write_to_csv(json_data, file_path): if not json_data: print("No data to write to CSV.") return # Extract keys (column names) from the first dictionary in the list keys = json_data[0].keys() with open(file_path, 'w', newline='') as file: writer = csv.DictWriter(file, fieldnames=keys) writer.writeheader() writer.writerows(json_data) # Check the response status and write to CSV if response.status_code == 200: response_json = response.json() # Assuming the response JSON contains a list of events events = response_json.get('events', []) csv_file_path = r'C:\\temp\\Tenable\\output1.csv' # Replace with your desired file location write_to_csv(events, csv_file_path) print(f"Data written to {csv_file_path}") else: print(f"Error: {response.status_code} - {response.text}") Error: 400 - { "error": { "code": 400, "message": "generic::invalid_argument: compilation error query uses a feature that is not yet allowed: invalid argument", "status": "INVALID_ARGUMENT" } }
                Russell P
                  Z
                  Forum|alt.badge.img+1
                  • zoooz
                  • New Member
                  • December 30, 2024

                  @russell_pfeifer Hi, i encountered this problem too, did you manged to solve it? thanks

                  AymanC
                  Forum|alt.badge.img+14
                  • AymanCBronze 5
                  • Bronze 5
                  • December 30, 2024

                  So I've gotten the Python query up to the point where I can generate a JSON file pulling UDM events from the Tenable log type. I have also incorporated a function that writes the JSON response to a CSV. 

                  Currently though - the query parameters are quite simple: 

                  query = 'metadata.log_type = "TENABLE_IO"'

                  Where the query seems to be failing is when I go to incorporate statistical functions into my query, specifically:

                  # Simplified query parameters query =''' events: metadata.log_type = "TENABLE_IO" $pluginID = security_result.rule_id $assetID = principal.asset.product_object_id match: $pluginID, $assetID outcome: $Description = array_distinct(extensions.vulns.vulnerabilities.description) $FQDN = array_distinct(principal.hostname) '''

                  Once the string is run with that query it returns this error:

                  Error: 400 - { "error": { "code": 400, "message": "generic::invalid_argument: compilation error query uses a feature that is not yet allowed: invalid argument", "status": "INVALID_ARGUMENT" } }

                  Is this simply a limitation of the current query parameters?

                  Full python string here:

                  # Set up API credentials and scopes SCOPES = ['https://www.googleapis.com/auth/chronicle-backstory'] SERVICE_ACCOUNT_FILE = r'C:\\API_keys\\OAuth_2_Keys.json' # Authenticate using the service account key credentials = service_account.Credentials.from_service_account_file( SERVICE_ACCOUNT_FILE, scopes=SCOPES ) http_session = AuthorizedSession(credentials) # API endpoint base_url = "https://backstory.googleapis.com/v1/events:udmSearch" # Simplified query parameters query =''' events: metadata.log_type = "TENABLE_IO" $pluginID = security_result.rule_id $assetID = principal.asset.product_object_id match: $pluginID, $assetID outcome: $Description = array_distinct(extensions.vulns.vulnerabilities.description) $FQDN = array_distinct(principal.hostname) ''' start_time = "2024-11-01T00:00:00Z" # Replace with desired start time end_time = "2024-12-20T01:00:00Z" # Replace with desired end time limit = 100 # Replace with desired limit # Build the query string params = { "query": query, "time_range.start_time": start_time, "time_range.end_time": end_time, "limit": limit } # Encode the query string for the URL (properly handle spaces and special characters) query_string = urlencode(params) # Full URL with query parameters url = f"{base_url}?{query_string}" # Make the GET request response = http_session.request("GET", url) # Function to write JSON response to CSV file def write_to_csv(json_data, file_path): if not json_data: print("No data to write to CSV.") return # Extract keys (column names) from the first dictionary in the list keys = json_data[0].keys() with open(file_path, 'w', newline='') as file: writer = csv.DictWriter(file, fieldnames=keys) writer.writeheader() writer.writerows(json_data) # Check the response status and write to CSV if response.status_code == 200: response_json = response.json() # Assuming the response JSON contains a list of events events = response_json.get('events', []) csv_file_path = r'C:\\temp\\Tenable\\output1.csv' # Replace with your desired file location write_to_csv(events, csv_file_path) print(f"Data written to {csv_file_path}") else: print(f"Error: {response.status_code} - {response.text}") Error: 400 - { "error": { "code": 400, "message": "generic::invalid_argument: compilation error query uses a feature that is not yet allowed: invalid argument", "status": "INVALID_ARGUMENT" } }

                  Hi @russell_pfeifer,

                  That's correct - currently Statistics and Aggregates isn't supported in the '/v1/events:udmSearch endpoint'.

                   

                  Kind Regards,

                  Ayman

                    russell_pfeifer
                    Forum|alt.badge.img+7

                    Hi @russell_pfeifer,

                    That's correct - currently Statistics and Aggregates isn't supported in the '/v1/events:udmSearch endpoint'.

                     

                    Kind Regards,

                    Ayman


                    Thanks Ayman -- good to know. I wonder if there are future plans to integrate stats / aggregates. 

                    Russell P
                    russell_pfeifer
                    Forum|alt.badge.img+7

                    @russell_pfeifer Hi, i encountered this problem too, did you manged to solve it? thanks


                    hi @zoooz 

                    See Ayman's response above. I have a formal inquiry out to Google as well but I believe that stats and aggregate commands simply aren't supported at this time given the specific error message generated by the string that attempts to incorporate them.

                    Regards

                    Russell P
                    AymanC
                    Forum|alt.badge.img+14
                    • AymanCBronze 5
                    • Bronze 5
                    • December 30, 2024

                    Thanks Ayman -- good to know. I wonder if there are future plans to integrate stats / aggregates. 


                    Hi @russell_pfeifer,

                    I believe there has been mentioned of plans to / is likely.

                    Kind Regards,

                    Ayman

                      raybrian
                      Staff
                      Forum|alt.badge.img+6
                      • raybrian
                      • Staff
                      • December 30, 2024

                      Here is an example from my scratchpad which might be helpful for you: 

                      https://gist.github.com/emeryray2002/1f3a4e19726b3fe1dffdceb930f33557

                        russell_pfeifer
                        Forum|alt.badge.img+7

                        Here is an example from my scratchpad which might be helpful for you: 

                        https://gist.github.com/emeryray2002/1f3a4e19726b3fe1dffdceb930f33557


                        Thanks for sharing @raybrian -- seeing if I can integrate your code into our instance now

                        Russell P
                        K
                        Forum|alt.badge.img
                        • KithuBronze 1
                        • Bronze 1
                        • May 7, 2025

                        Thanks for sharing @raybrian -- seeing if I can integrate your code into our instance now


                        Hi @russell_pfeifer ,

                        Did it work? I am trying to connect the backstory API's as well, till no luck. getting either 404 or 400 codes.

                        @raybrian does the json file require any other details other than the Backsory keys.

                        raybrian
                        Staff
                        Forum|alt.badge.img+6
                        • raybrian
                        • Staff
                        • May 7, 2025

                        My examples are always using the Chronicle API from the BYOP project, not the legacy APIs. So you'd need to auth with Chronicle APi credentials from the correct project. LMK if you need help figuring that out.

                          K
                          Forum|alt.badge.img
                          • KithuBronze 1
                          • Bronze 1
                          • May 8, 2025

                          Hi @raybrian,

                          Thanks for clarification. Currently I have Admin, Backstor-api, Ingestion-api, forwarder-api and bigquery-api credentials of our service account.

                          From your comment, I believe we need another chronicle API or we are lacking any permission in backstory-api?


                            mwisener
                            Forum|alt.badge.img+3
                            • mwisenerBronze 1
                            • Bronze 1
                            • May 9, 2025

                            Hi @raybrian,

                            Thanks for clarification. Currently I have Admin, Backstor-api, Ingestion-api, forwarder-api and bigquery-api credentials of our service account.

                            From your comment, I believe we need another chronicle API or we are lacking any permission in backstory-api?



                            @Kithu 

                            Look at your keys, if the project field in the key has "malachite" that is what they are calling "legacy api". (these keys are associated with a google-owned project (the malachite project)

                            In your SecOps instance, if you go to Settings > Profile and there is a gcp_project associated with the instance then it's called "BYOP" (bring your own project).

                            In GCP in the project associated with the SecOps instance, you can go to "service accounts" under IAM, and create a new service-account. Give it the "Chronicle API Admin" role. 

                            You can then use this key with the code that raybrian references since their code is using the "new" or "dataplane" api (although it's still referenced as a v1alpha/beta api in the docs).

                            I suspect you are using the "old" apis. Here are the references to the difference:

                            https://cloud.google.com/chronicle/docs/reference

                            Anything under "Chronicle SIEM API" is considered "legacy"

                            Anything under the "Chronicle API" (at the bottom) is considered "new"

                            You can tell if it's new or old based on the endpoint path. If the path has a project-id/location/customer-id in the path, it's "new" since it's linked to your gcp secops project.

                            The customer-id is obtained from the Setting > Profile in SecOps "customer-id" and is a uuid.

                            HTH,

                            \\- Mike

                             

                              K
                              Forum|alt.badge.img
                              • KithuBronze 1
                              • Bronze 1
                              • May 13, 2025

                              Hi @mwisener ,

                              Thanks for the clarification. It is malachite, as it is our test env.

                              Will check with team and update here.

                              Regards,
                              Kithu

                                K
                                Forum|alt.badge.img
                                • KithuBronze 1
                                • Bronze 1
                                • May 14, 2025

                                Hi @mwisener ,

                                We have created new service account, tweaked a code a bit and we have achieved it 🙂

                                Full python code:
                                 

                                import requests import json from datetime import datetime, timedelta from google.oauth2 import service_account from google.auth.transport import requests as google_auth_requests from tabulate import tabulate import textwrap class ChronicleSearch: def __init__(self, project_id: str, instance_id: str, credentials_json: str, region: str = "europe"): self.base_url = f"https://{region}-chronicle.googleapis.com" self.project_id = project_id self.instance_id = instance_id self.credentials = service_account.Credentials.from_service_account_info( json.loads(credentials_json), scopes=["https://www.googleapis.com/auth/cloud-platform"] ) def get_access_token(self) -> str: request = google_auth_requests.Request() self.credentials.refresh(request) return self.credentials.token def get_stats(self, query: str, start_time: datetime, end_time: datetime): access_token = self.get_access_token() url = ( f"{self.base_url}/v1alpha/projects/{self.project_id}/locations/europe/" f"instances/{self.instance_id}/legacy:legacyFetchUdmSearchView" ) payload = { "baselineQuery": query, "baselineTimeRange": { "startTime": start_time.strftime("%Y-%m-%dT%H:%M:%S.000Z"), "endTime": end_time.strftime("%Y-%m-%dT%H:%M:%S.000Z") }, "caseInsensitive": True, "returnOperationIdOnly": False, "eventList": { "maxReturnedEvents": 10000 }, "generateAiOverview": False } headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "Chronicle-Search-Client/1.0" } response = requests.post(url, json=payload, headers=headers) if response.status_code != 200: raise Exception(f"Error initiating search: {response.text}") return response.json() def validate_chronicle_query(query: str, credentials_json: str, project_id: str, instance_id: str) -> dict: try: credentials = service_account.Credentials.from_service_account_info( json.loads(credentials_json), scopes=["https://www.googleapis.com/auth/cloud-platform"] ) request = google_auth_requests.Request() credentials.refresh(request) access_token = credentials.token url = f"https://europe-chronicle.googleapis.com/v1alpha/projects/{project_id}/locations/europe/instances/{instance_id}:validateQuery" headers = { "Authorization": f"Bearer {access_token}", "Accept": "*/*", "User-Agent": "Chronicle-Query-Validator/1.0" } params = { "rawQuery": query, "dialect": "DIALECT_UDM_SEARCH", "allowUnreplacedPlaceholders": "false" } response = requests.get(url, headers=headers, params=params) if response.status_code != 200: return {"error": f"API Error: {response.text}"} return response.json() except Exception as e: return {"error": str(e)} def extract_stats_rows(stats, wrap_width=100): """Extract tabular rows from Chronicle stats section with wrapped command line.""" columns = [col["column"] for col in stats["results"]] num_rows = len(stats["results"][0]["values"]) rows = [] for i in range(num_rows): row = [] for col in stats["results"]: val_entry = col["values"][i] val_obj = val_entry.get("value", val_entry) if "stringVal" in val_obj: wrapped = textwrap.fill(val_obj["stringVal"], width=wrap_width) row.append(wrapped) elif "int64Val" in val_obj: row.append(int(val_obj["int64Val"])) else: row.append(str(val_obj)) rows.append(row) return columns, rows def main(): PROJECT_ID = "<Value>" # Service account key has it INSTANCE_ID = "<Value>" # UUID customer ID value from Console with open('<location>/file_name.json', 'r') as f: credentials_json = f.read() query = """metadata.product_name = \\"Microsoft-Windows-Sysmon\\" AND src.process.file.full_path = \\"Cmd.Exe\\" match: principal.hostname, principal.process.command_line outcome: $count = count(metadata.id) order: principal.hostname asc """ end_time = datetime.utcnow() start_time = end_time - timedelta(days=1) try: validation_result = validate_chronicle_query(query, credentials_json, PROJECT_ID, INSTANCE_ID) if "error" in validation_result: print(f"⚠️ Query validation error: {validation_result['error']}") return client = ChronicleSearch(PROJECT_ID, INSTANCE_ID, credentials_json) results = client.get_stats(query, start_time, end_time) stats_found = False if isinstance(results, list): for res in results: if isinstance(res, dict) and "stats" in res: stats = res["stats"] if stats.get("results"): columns, rows = extract_stats_rows(stats) print("\\n📊 Stats Results (Wrapped View):") print(tabulate(rows, headers=columns, tablefmt="fancy_grid")) stats_found = True elif isinstance(results, dict) and "stats" in results: stats = results["stats"] if stats.get("results"): columns, rows = extract_stats_rows(stats) print("\\n📊 Stats Results (Wrapped View):") print(tabulate(rows, headers=columns, tablefmt="fancy_grid")) stats_found = True if not stats_found: print("⚠️ No stats found in response.") except Exception as e: print(f"❌ Error performing search: {str(e)}") if __name__ == "__main__": main()

                                 

                                 

                                And the result would like this:


                                Once again, thank guys 😄



                                 

                                  Terms & ConditionsCookie settingsAccessibility statement
                                  Privacy Policy Terms of Service Gen AI Policy Community Guidelines Cookie Settings

                                  © 2025 Google. All rights reserved.