[ASIM] Fix ASIM workflows

.script/tests/asimParsersTest/VerifyASimParserTemplate.py

@@ -313,7 +313,7 @@ def extract_and_check_properties(Parser_file, Union_Parser__file, FileType, ASIM
         if os.path.exists(SampleDataFilePath):
             results.append((SampleDataFile, 'Sample data file exists', 'Pass'))
         else:
-            results.append((f'{RED}Expected sample file not found{RESET}', f'{RED}Sample data file does not exist or may not be named correctly. Please include sample data file "{event_vendor}_{event_product}_{schema}_IngestedLogs.csv"{RESET}', f'{RED}Fail{RESET}'))
+            results.append((f'{RED}Expected sample file not found{RESET}', f'{RED}Sample data file does not exist or may not be named correctly. Please include sample data file "{SampleDataFile}"{RESET}', f'{RED}Fail{RESET}'))
     return results
 
 def filter_yaml_files(modified_files):

.script/tests/asimParsersTest/ingestASimSampleData.py

@@ -1,3 +1,16 @@
+# ingestASimSampleData.py
+#
+# This script ingests ASIM parser sample data into a Log Analytics workspace for testing.
+#
+# It expects the Azure-Sentinel repo to be checked out locally and works as follows:
+#   1. Detects modified ASIM parser YAML files by comparing the current branch against upstream/master.
+#   2. Reads each parser YAML from the local repo to extract schema, vendor, and product info.
+#   3. Loads the corresponding sample data and schema CSV files from "Sample Data/ASIM/".
+#   4. For custom log tables: creates the table and a Data Collection Rule (DCR), then ingests data.
+#   5. For built-in tables: creates a DCR (handling GUID column mismatches) and ingests data.
+#
+# Usage: python ingestASimSampleData.py <pr_number>
+
 import sys
 import os
 
@@ -36,46 +49,78 @@ def get_modified_files(current_directory):
         print(f"::error::Error occurred while executing the command: {e}")
         return []
 
-def get_current_commit_number():
-    cmd = "git rev-parse HEAD"
-    try:
-        return subprocess.check_output(cmd, shell=True, text=True).strip()
-    except subprocess.CalledProcessError as e:
-        print(f"::error::Error occurred while executing the command: {e}")
-        return None
-
-def read_github_yaml(url):
-    try:
-        response = requests.get(url)
-    except Exception as e:
-        print(f"::error::An error occurred while trying to get content of YAML file located at {url}: {e}")
-    return yaml.safe_load(response.text) if response.status_code == 200 else None    
-
 def filter_yaml_files(modified_files):
     # Take only the YAML files
     return [line for line in modified_files if line.endswith('.yaml')]
 
 
 def convert_schema_csv_to_json(csv_file):
+    """Reads a schema CSV file with 'ColumnName' and 'ColumnType' headers and returns a list of
+    column definitions (name/type dicts) suitable for table creation APIs. Reserved columns are
+    excluded and 'bool' types are mapped to 'boolean'."""
     data = []
-    with open(csv_file, 'r',encoding='utf-8-sig') as file:
-        reader = csv.DictReader(file)
-        for row in reader:
-            if row['ColumnName'] in reserved_columns:
-                continue
-            elif row['ColumnType'] == "bool":
-                data.append({        
-                'name': row['ColumnName'],
-                'type': "boolean",
-                })
-            else:
-                data.append({        
-                'name': row['ColumnName'],
-                'type': row['ColumnType'],
-                })                       
+    try:
+        with open(csv_file, 'r',encoding='utf-8-sig') as file:
+            reader = csv.DictReader(file)
+            for row in reader:
+                if 'ColumnName' not in row or 'ColumnType' not in row:
+                    raise KeyError(f"Required columns 'ColumnName' and/or 'ColumnType' not found in schema CSV '{csv_file}'. Available columns: {list(row.keys())}")
+                if row['ColumnName'] in reserved_columns:
+                    continue
+                elif row['ColumnType'] == "bool":
+                    data.append({        
+                    'name': row['ColumnName'],
+                    'type': "boolean",
+                    })
+                else:
+                    data.append({        
+                    'name': row['ColumnName'],
+                    'type': row['ColumnType'],
+                    })
+    except FileNotFoundError:
+        print(f"::error::Schema CSV file not found: {csv_file}")
+        raise
+    except KeyError:
+        raise
+    except Exception as e:
+        print(f"::error::Failed to parse schema CSV file '{csv_file}': {e}")
+        raise
+
+    if not data:
+        raise ValueError(f"No schema columns found in CSV file '{csv_file}'")
+    print(f"Schema data for {csv_file}: {data}")
     return data
 
+def infer_schema_from_data(data_result, datetime_keys):
+    """Infers a schema (list of name/type dicts) from sample data rows when no Schema CSV exists.
+    Inspects the first data row's values to determine column types. Reserved columns are excluded."""
+    if not data_result:
+        raise ValueError("Cannot infer schema from empty data")
+    first_row = data_result[0]
+    schema = []
+    for key, value in first_row.items():
+        if key in reserved_columns:
+            continue
+        if key in datetime_keys:
+            col_type = "datetime"
+        elif isinstance(value, bool):
+            col_type = "boolean"
+        elif isinstance(value, int):
+            col_type = "int"
+        elif isinstance(value, float):
+            col_type = "real"
+        else:
+            col_type = "string"
+        schema.append({'name': key, 'type': col_type})
+    # Ensure TimeGenerated is always present in the schema
+    if not any(col['name'] == 'TimeGenerated' for col in schema):
+        schema.append({'name': 'TimeGenerated', 'type': 'datetime'})
+    return schema
+
 def convert_data_csv_to_json(csv_file):
+    """Reads a sample data CSV file and returns its rows as a list of dicts with auto-typed values,
+    the table name extracted from the 'Type' column, and a list of column names that had timestamp
+    suffixes ('[UTC]' or '[Local Time]') stripped from their keys."""
     def convert_value(value):
         # Try to convert the value to an integer, then to a float, and keep it as a string if those fail
         try:
@@ -90,22 +135,41 @@ def convert_value(value):
                 return value
 
     data = []
-    with open(csv_file, 'r', encoding='utf-8-sig') as file:
-        reader = csv.DictReader(file)
-        for row in reader:
-            table_name = row['Type']
-            # Convert each value in the row to its appropriate type
-            processed_row = {key: convert_value(value) for key, value in row.items()}
-            data.append(processed_row)
-
-        for item in data:
-            for key in list(item.keys()):
-                # If the key matches '[UTC]' or '[Local Time]', rename it
-                if key.endswith(('[UTC]', '[Local Time]')):
-                    substring = key.split(" [")[0]
-                    item[substring] = item.pop(key)
-
-    return data, table_name
+    table_name = None
+    datetime_keys = set()
+    try:
+        with open(csv_file, 'r', encoding='utf-8-sig') as file:
+            reader = csv.DictReader(file)
+            for row in reader:
+                if 'Type' not in row:
+                    raise KeyError(f"'Type' column not found in CSV file '{csv_file}'. Available columns: {list(row.keys())}")
+                table_name = row['Type']
+                # Convert each value in the row to its appropriate type
+                processed_row = {key: convert_value(value) for key, value in row.items()}
+                data.append(processed_row)
+
+            for item in data:
+                for key in list(item.keys()):
+                    # If the key matches '[UTC]' or '[Local Time]', rename it
+                    if key.endswith(('[UTC]', '[Local Time]')):
+                        substring = key.split(" [")[0]
+                        item[substring] = item.pop(key)
+                        datetime_keys.add(substring)
+    except FileNotFoundError:
+        print(f"::error::Data CSV file not found: {csv_file}")
+        raise
+    except KeyError:
+        raise
+    except Exception as e:
+        print(f"::error::Failed to parse data CSV file '{csv_file}': {e}")
+        raise
+
+    if not data:
+        raise ValueError(f"No data rows found in CSV file '{csv_file}'")
+    if table_name is None:
+        raise ValueError(f"Could not determine table name from CSV file '{csv_file}'")
+
+    return data, table_name, list(datetime_keys)
 
 def check_for_custom_table(table_name):
     if table_name in lia_supported_builtin_table:
@@ -245,16 +309,19 @@ def hit_api(url,request,method):
 
 def senddtosentinel(immutable_id,data_result,stream_name,flag_status):
     if flag_status == 0:
-        print("DCR is not created for the table. Please create DCR first")
-        return
+        print("::error::DCR is not created for the table. Please create DCR first")
+        return False
     print("Waiting for data to be sent to sentinel (This will take atleast 20 seconds)")
     time.sleep(20)
     credential = DefaultAzureCredential()
     client = LogsIngestionClient(endpoint=endpoint_uri, credential=credential, logging_enable=True)
     try:
         client.upload(rule_id=immutable_id, stream_name=stream_name, logs=data_result)
+        print(f"Data uploaded successfully to stream '{stream_name}' via DCR '{immutable_id}'")
+        return True
     except HttpResponseError as e:
-        print(f"Upload failed: {e}")
+        print(f"::error::Upload failed: {e}")
+        return False
 
 
 def extract_event_vendor_product(parser_query,parser_file):
@@ -313,8 +380,6 @@ def convert_data_type(schema_result, data_result):
 subscriptionId = "419581d6-4853-49bd-83b6-d94bb8a77887"
 dataCollectionEndpointname = "ASIM-SchemaDataTester-GithubShared-Canary"
 endpoint_uri = "https://asim-schemadatatester-githubshared-canary-qa1f.eastus2euap-1.canary.ingest.monitor.azure.com" # logs ingestion endpoint of the DCR
-SENTINEL_REPO_RAW_URL = f'https://raw.githubusercontent.com/Azure/Azure-Sentinel'
-SAMPLE_DATA_PATH = 'Sample%20Data/ASIM/'
 dcr_directory=[]
 
 lia_supported_builtin_table = ['ADAssessmentRecommendation','ADSecurityAssessmentRecommendation','Anomalies','ASimAuditEventLogs','ASimAuthenticationEventLogs','ASimDhcpEventLogs','ASimDnsActivityLogs','ASimDnsAuditLogs','ASimFileEventLogs','ASimNetworkSessionLogs','ASimProcessEventLogs','ASimRegistryEventLogs','ASimUserManagementActivityLogs','ASimWebSessionLogs','AWSCloudTrail','AWSCloudWatch','AWSGuardDuty','AWSVPCFlow','AzureAssessmentRecommendation','CommonSecurityLog','DeviceTvmSecureConfigurationAssessmentKB','DeviceTvmSoftwareVulnerabilitiesKB','ExchangeAssessmentRecommendation','ExchangeOnlineAssessmentRecommendation','GCPAuditLogs','GoogleCloudSCC','SCCMAssessmentRecommendation','SCOMAssessmentRecommendation','SecurityEvent','SfBAssessmentRecommendation','SharePointOnlineAssessmentRecommendation','SQLAssessmentRecommendation','StorageInsightsAccountPropertiesDaily','StorageInsightsDailyMetrics','StorageInsightsHourlyMetrics','StorageInsightsMonthlyMetrics','StorageInsightsWeeklyMetrics','Syslog','UCClient','UCClientReadinessStatus','UCClientUpdateStatus','UCDeviceAlert','UCDOAggregatedStatus','UCServiceUpdateStatus','UCUpdateAlert','WindowsEvent','WindowsServerAssessmentRecommendation','NTANetAnalytics', 'AZFWNetworkRule', 'AZFWNatRule', 'AZFWApplicationRule', 'AZFWDnsQuery', 'AZFWIdspSignature', 'AZFWThreatIntel']
@@ -326,7 +391,6 @@ def convert_data_type(schema_result, data_result):
 
 parser_yaml_files = filter_yaml_files(modified_files)
 
-commit_number = get_current_commit_number()
 prnumber = sys.argv[1]
 
 for file in parser_yaml_files:
@@ -340,42 +404,51 @@ def convert_data_type(schema_result, data_result):
         print(f"Ignoring this {file} because it is a union or empty parser file")
         continue        
     print(f"Starting ingestion for sample data present in {file}")
-    asim_parser_url = f'{SENTINEL_REPO_RAW_URL}/{commit_number}/{file}'
-    print(f"Reading Asim Parser file from : {asim_parser_url}")
-    asim_parser = read_github_yaml(asim_parser_url)
+    repo_root = os.path.abspath(os.path.join(current_directory, '..', '..', '..'))
+    asim_parser_path = os.path.join(repo_root, file)
+    print(f"Reading Asim Parser file from local path: {asim_parser_path}")
+    try:
+        with open(asim_parser_path, 'r', encoding='utf-8') as f:
+            asim_parser = yaml.safe_load(f.read())
+    except Exception as e:
+        print(f"::error::An error occurred while trying to read YAML file at {asim_parser_path}: {e}")
+        continue
     parser_query = asim_parser.get('ParserQuery', '')
     normalization = asim_parser.get('Normalization', {})
     schema = normalization.get('Schema')
     equivalent_built_in_parser = asim_parser.get('EquivalentBuiltInParser')
     event_vendor, event_product, schema_name = extract_event_vendor_product(parser_query, file)
 
     SampleDataFile = f'{event_vendor}_{event_product}_{schema}_IngestedLogs.csv'
-    sample_data_url = f'{SENTINEL_REPO_RAW_URL}/{commit_number}/{SAMPLE_DATA_PATH}'
-    SampleDataUrl = sample_data_url+SampleDataFile
-    print(f"Sample data log file reading from url: {SampleDataUrl}")
-    response = requests.get(SampleDataUrl)
-    if response.status_code == 200:
-        with open('tempfile.csv', 'wb') as file:
-            file.write(response.content)
-    else:
-        print(f"::error::An error occurred while trying to get content of Sample Data file located at {SampleDataUrl}: {response.text}")
-        continue           
-    data_result,table_name = convert_data_csv_to_json('tempfile.csv')   
+    sample_data_dir = os.path.join(repo_root, 'Sample Data', 'ASIM')
+    sample_data_path = os.path.join(sample_data_dir, SampleDataFile)
+    print(f"Sample data log file reading from local path: {sample_data_path}")
+    try:
+        with open(sample_data_path, 'rb') as f:
+            with open('tempfile.csv', 'wb') as file:
+                file.write(f.read())
+    except Exception as e:
+        print(f"::error::An error occurred while trying to read Sample Data file at {sample_data_path}: {e}")
+    data_result,table_name,datetime_keys = convert_data_csv_to_json('tempfile.csv')
     print(f"Table Name : {table_name}")
     log_ingestion_supported,table_type=check_for_custom_table(table_name)
     print(f"Log ingestion supported: {log_ingestion_supported}\n Table type: {table_type}")
     if log_ingestion_supported == True and table_type =="custom_log":
         flag=0 #flag value is used to check if DCR is created for the table or not
         schema_file_name = f"{table_name}_Schema.csv"
-        schemaUrl = sample_data_url+schema_file_name
-        response = requests.get(schemaUrl)
-        if response.status_code == 200:
-            with open('tempfile.csv', 'wb') as file:
-                file.write(response.content)
+        schema_path = os.path.join(sample_data_dir, schema_file_name)
+        if os.path.exists(schema_path):
+            try:
+                with open(schema_path, 'rb') as f:
+                    with open('tempfile.csv', 'wb') as file:
+                        file.write(f.read())
+            except Exception as e:
+                print(f"::error::An error occurred while trying to read Schema file at {schema_path}: {e}")
+                continue
+            schema_result = convert_schema_csv_to_json('tempfile.csv')
         else:
-            print(f"::error::An error occurred while trying to get content of Schema file located at {schemaUrl}: {response.text}")
-            continue        
-        schema_result = convert_schema_csv_to_json('tempfile.csv')
+            print(f"Schema CSV not found at {schema_path}, inferring schema from sample data")
+            schema_result = infer_schema_from_data(data_result, datetime_keys)
         data_result = convert_data_type(schema_result, data_result)
         # conversion of datatype is needed for boolean and string values because during testing it has been observed that 
         # boolean values are consider as string and numerical value of type string are consider