diff --git a/AirlockTools.py b/AirlockTools.py index 2ff88c0..d1b69a1 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -15,6 +15,7 @@ import argparse import dotenv import os +import re import pandas as pd import urllib3 import utils.clientfunctions @@ -133,7 +134,7 @@ def menu_main(): ct.displayIntro(); print(ct.colorText("1. Get All Events for Single Device", "yellow")) print(ct.colorText("2. OTP", "yellow")) - print(ct.colorText("3. Placeholder for Another Tool", "yellow")) + print(ct.colorText("3. Divide by Exclusion", "yellow")) print(ct.colorText("4. Prepare Policy For Enforcement", "yellow")) print(ct.colorText("Q. Quit", "yellow")) @@ -174,9 +175,11 @@ def menu_otp(): def menu_feature2(): while True: print("\n--- Submenu ---") - print("1. Pull last 24 horus execution for all ATPolicys") + print("1. Pull last 24 hours execution for all ATPolicys") print("2. Generate Paths") - print("3 Merge") + print("3. Merge") + print("4. Pull existing paths") + print("5. Divide By Excluded or not excluded by path") print("Q. Exit") choice = input("Enter your choice: ") @@ -218,6 +221,115 @@ def menu_feature2(): elif choice == "3": utils.pathfunctions.mergeTesting() + + elif choice == "4": + + + # Get the AT policies dictionary + atpolicies = utils.policyfunctions.listATPolicies(url) + + # List to hold each policy's path data + all_paths = [] + + # Loop through each policy name and group ID + for policy_name, group_id in atpolicies.items(): + try: + # Get the list of paths using the group ID + paths = utils.pathfunctions.listPaths(url, group_id) + + # Ensure each path is a string and append properly + for path in paths: + if isinstance(path, str): + all_paths.append({ + 'PolicyName': policy_name, + 'GroupID': group_id, + 'Path': path + }) + except Exception as e: + print(f"Error retrieving paths for policy '{policy_name}': {e}") + + # Convert to DataFrame + if all_paths: + paths_df = pd.DataFrame(all_paths) + print(paths_df.head()) # Optional: preview first few rows + print("✅ Paths DataFrame created.") + else: + paths_df = pd.DataFrame() + print("⚠️ No paths were retrieved.") + + # Save to CSV + paths_df.to_csv("paths.csv", index=False) + elif choice == "5": + + # Load and clean data + filenames_df = pd.read_csv('merged_output.csv') # Contains 'PolicyName' and 'filename' + exclusions_df = pd.read_csv('paths.csv') # Contains 'PolicyName' and 'Path' + + # Clean and normalize columns + filenames_df['filename'] = filenames_df['filename'].fillna('').astype(str).str.strip() + filenames_df['PolicyName'] = filenames_df['PolicyName'].fillna('').astype(str).str.strip() + exclusions_df['Path'] = exclusions_df['Path'].fillna('').astype(str).str.strip() + exclusions_df['PolicyName'] = exclusions_df['PolicyName'].fillna('').astype(str).str.strip() + + # Decode escaped backslashes in exclusion patterns + exclusions_df['Path'] = exclusions_df['Path'].apply(lambda p: p.encode('utf-8').decode('unicode_escape')) + + # Build exclusion map: {PolicyName: [compiled regex patterns]} + exclusion_map = {} + for _, row in exclusions_df.iterrows(): + policy = row['PolicyName'] + raw_pattern = row['Path'] + try: + regex = utils.pathfunctions.wildcardRegex(raw_pattern) + print(f"[EXCLUSION MAP] Policy: {policy}, Pattern: {raw_pattern} → Regex: {regex.pattern}") + exclusion_map.setdefault(policy, []).append(regex) + except Exception as e: + print(f"[ERROR] Failed to compile pattern for Policy: {policy}, Path: {raw_pattern}, Error: {e}") + + # Function to check if a filename matches any exclusion pattern for its PolicyName + def is_excluded(row): + policy = row['PolicyName'].strip() + path = os.path.normpath(row['filename'].strip()) + patterns = exclusion_map.get(policy, []) + for r in patterns: + if r.match(path): + print(f"[MATCH] {policy}: {path} matches {r.pattern}") + return True + print(f"[NO MATCH] {policy}: {path}") + return False + + # Apply matching + excluded = filenames_df[filenames_df.apply(is_excluded, axis=1)] + not_excluded = filenames_df[~filenames_df.apply(is_excluded, axis=1)] + + + # Desired column order + column_order = [ + 'PolicyName', 'filename', 'longestcfp', + 'pprocess', 'gprocess', 'sha256', 'publisher', 'description', 'productname','commandline', 'middle', 'filename_only', + 'file_extension', 'unique_sha256_count', 'hostname', 'username', 'productversion', + 'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount', + 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp' + ] + + # Reorder columns (ignore missing ones) + excluded = excluded[[col for col in column_order if col in excluded.columns]] + not_excluded = not_excluded[[col for col in column_order if col in not_excluded.columns]] + + + # Save results + excluded.to_csv('excluded_filenames.csv', index=False) + not_excluded.to_csv('non_excluded_filenames.csv', index=False) + + + elif choice == "6": + + pattern = utils.pathfunctions.wildcardRegex("C:\\Windows\\SystemTemp\\????????\\????????.dll") + test_path = "C:\\Windows\\SystemTemp\\fsq2xzua\\fsq2xzua.dll" + + print("Match:", pattern.match(test_path) is not None) + + elif choice == "Q": break else: diff --git a/merged_output.xlsx b/merged_output.xlsx new file mode 100644 index 0000000..7490ac6 Binary files /dev/null and b/merged_output.xlsx differ diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index c7d6f9a..fbaec32 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -22,11 +22,13 @@ from AirlockTools import tryToReadCSV #Standard Libary Imports: import ast import gc +import json import os import re #3rd Party Imports: import pandas as pd +import requests def split_filepaths_grouped(df, col="filename", group_parts=4, min_parts=4): def clean_split(path): @@ -321,3 +323,45 @@ def mergeTesting(): merged_df.to_csv("merged_output.csv", index=False) print(f"Merged DataFrame saved with {len(merged_df)} rows.") + + +def listPaths(url, group): + endpoint = url + '/v1/group/policies' + print(ct.colorText("[+] Grabbing All Paths", "cyan")) + + payload = { + "groupid": [group], + } + headers = { + "X-APIKey": os.getenv('APIKEY') + } + + try: + response = requests.post(endpoint, headers=headers, data=payload, verify=False) + response.raise_for_status() + parse_text = response.json() + + pathnames = [] + print(parse_text) # Optional: for debugging + + for item in parse_text.get('response', {}).get('paths', []): + path = item.get('name') + if path: + pathnames.append(path) + + return pathnames + + except requests.exceptions.RequestException as e: + print(ct.colorText(f"[!] Request failed: {e}", "red")) + return [] + except (KeyError, json.JSONDecodeError) as e: + print(ct.colorText(f"[!] Failed to parse response: {e}", "red")) + return [] + +def wildcardRegex(pattern): + pattern = pattern.replace("\\", "\\\\") + pattern = pattern.replace("**", "___RECURSIVE___") + pattern = pattern.replace("*", "[^\\\\]*") + pattern = pattern.replace("?", ".") + pattern = pattern.replace("___RECURSIVE___", ".*") + return re.compile(f"^{pattern}$", re.IGNORECASE) \ No newline at end of file diff --git a/utils/policyfunctions.py b/utils/policyfunctions.py index ecf4e5d..638bfc6 100644 --- a/utils/policyfunctions.py +++ b/utils/policyfunctions.py @@ -352,4 +352,5 @@ def sendToPolicyTest(url, first_policy, second_policy, destination_name, destina print(ct.colorText(f"These hashes would be added to {allowlist_child_name}", "yellow")) allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist() - addHash(url, allowlist_child_id, allowlist_childhashlist) \ No newline at end of file + addHash(url, allowlist_child_id, allowlist_childhashlist) +