diff --git a/AirlockTools.py b/AirlockTools.py index 4abfed9..2ff88c0 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -35,11 +35,34 @@ dotenv.load_dotenv() url = os.getenv('url') bad_publisher_list = ["Brave","Zoom", "GlavSoft", "VNC"] pups = ["logmein", "invalid" , "nmap", "VNC", "Kaseya", "Solarwinds", "mRemoteNG"] -badpathparts = ["users", "wwwroot", "windows\\task", "windows\\system32", "startup", "windows\\fonts", "Recycle.Bin", "AppData", "programdata", "Solarwinds", "kaseya", "Windows\\assembly", "WindowsPowerShell\\Modules", "windows\\temp"] +badpathparts = ["users", "wwwroot", "windows\\task", "windows\\system32", "startup", "windows\\fonts", "Recycle.Bin", "AppData", "programdata", "Solarwinds", "kaseya", ] #"Windows\\assembly", "WindowsPowerShell\\Modules", "windows\\temp"3 path_exclusion_constant = 4 min_files_for_path = 4 threat_tolerance_constant = 4 +""" +Execution Types + 0 = "Trusted Execution", + 1 = "Blocked Execution", + 2 = "Untrusted Execution [Audit]", + 3 = "Untrusted Execution [OTP]", + 4 = "Trusted Path Execution", + 5 = "Trusted Publisher Execution", + 6 = "Blocklist Execution", + 7 = "Blocklist Execution [Audit]", + 8 = "Trusted Process Execution", + 9 = "Constrained Execution", + 10 = "Trusted Metadata Execution", + 11 = "Trusted Browser Execution", + 12 = "Blocked Browser Execution", + 13 = "Untrusted Browser Execution [Audit]", + 14 = "Untrusted Browser Execution [OTP]", + 15 = "Blocklist Browser Execution [Audit]", + 16 = "Blocklist Browser Execution", + 17 = "Trusted Installer Execution", + 18 = "Trusted Browser Metadata Execution" +""" + def main(): parser = argparse.ArgumentParser(description="Your script description") @@ -149,22 +172,56 @@ def menu_otp(): print("Invalid choice. Please try again.") def menu_feature2(): - while True: - print("\n--- Submenu ---") - print("1. Sub-option A") - print("2. Sub-option B") - print("3. Return to Main Menu") - choice = input("Enter your choice: ") + while True: + print("\n--- Submenu ---") + print("1. Pull last 24 horus execution for all ATPolicys") + print("2. Generate Paths") + print("3 Merge") + print("Q. Exit") + choice = input("Enter your choice: ") - if choice == "1": - print("You selected Sub-option A") - elif choice == "2": - print("You selected Sub-option B") - elif choice == "3": - print("Returning to Main Menu...") - break - else: - print("Invalid choice. Please try again.") + if choice == "1": + + # Get the AT policies dictionary + atpolicies = utils.policyfunctions.listATPolicies(url) + + # List to hold each policy's DataFrame + all_dfs = [] + + # Loop through each policy name + for policy_name in atpolicies: + try: + # Get the policy info DataFrame + df = utils.policyfunctions.getPolicyInfo(url, policy_name, [1, 2, 6, 7], 1, False) + + # Add a column to indicate the policy name + df['PolicyName'] = policy_name + + # Append to the list + all_dfs.append(df) + except Exception as e: + print(f"Error processing policy '{policy_name}': {e}") + + # Combine all DataFrames into one + if all_dfs: + combined_df = pd.concat(all_dfs, ignore_index=True) + print("✅ Combined DataFrame created.") + else: + combined_df = pd.DataFrame() + print("⚠️ No data was retrieved.") + + combined_df.to_csv("allATPolicyExecs.csv", index=False) + + + elif choice == "2": + utils.pathfunctions.allATpaths(url,pups, bad_publisher_list, badpathparts, threat_tolerance_constant, 3, min_files_for_path) + + elif choice == "3": + utils.pathfunctions.mergeTesting() + elif choice == "Q": + break + else: + print("Invalid choice. Please try again.") def menu_prepare_to_enforce(): @@ -222,10 +279,10 @@ def menu_prepare_to_enforce(): if not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"): - utils.policyfunctions.getPolicyInfo(url, first_policy, history_days) + utils.policyfunctions.getPolicyInfo(url, first_policy, [1,2,6,7], history_days) if not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"): - utils.policyfunctions.getPolicyInfo(url, second_policy, history_days) + utils.policyfunctions.getPolicyInfo(url, second_policy, [1,2,6,7], history_days) if not os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"): utils.hashfunctions.combineHashes(url, first_policy, second_policy) diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index 01c54ff..c7d6f9a 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -14,6 +14,7 @@ # along with this program. If not, see . #Local Imports +import utils.hashfunctions as hashf import utils.pathfunctions as pathf import utils.pretty as ct from AirlockTools import tryToReadCSV @@ -193,3 +194,130 @@ def generatePathReview(first_policy, second_policy, badpathparts, path_exclusion del unique_sha_counts del lcp_not_forbidden_review + + +def allATpaths(url, pups, untrusted_publishers, badpathparts, threat_tolerance, path_exclusion_constant, min_files_for_path): + import pandas as pd + + # Load raw data + hashes = pd.read_csv("allATPolicyExecs.csv") + + # Deduplicate hashes before augmentation + deduped_hashes = hashes.drop_duplicates(subset=['sha256']).copy() + + # Save policy name mapping (before deduplication) + policyname_map = hashes[['hostname', 'PolicyName']].drop_duplicates() + + # Handle None inputs + untrusted_publishers = untrusted_publishers or [] + pups = pups or [] + + # Augment deduplicated hashes + augmented = hashf.augmentAggregatedHashes(url, deduped_hashes) + + # Merge policy names + augmented = augmented.merge(policyname_map, on='hostname', how='left') + + # Clean numeric reputation fields + for col in ['reputation_scannermatch', 'reputation_scannercount', 'reputation_threatlevel']: + if col in augmented.columns: + augmented[col] = pd.to_numeric(augmented[col].replace('N/A', pd.NA), errors='coerce') + + # Rename and select relevant columns + augmented = augmented.rename(columns={'publisher_x': 'publisher'}) + augmented = augmented[[ + 'PolicyName', 'sha256', 'publisher', 'description', 'productname', 'productversion', + 'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount', + 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', + 'reputation_timestamp' + ]].sort_values(by=['publisher', 'description', 'productname']) + + # Merge with deduplicated hashes to enrich data + final_augmented = augmented.merge(deduped_hashes, on='sha256', how='left') + + # Clean up column names before applying reputation logic + if 'publisher_y' in final_augmented.columns: + final_augmented = final_augmented.drop(columns=['publisher_y']) + if 'publisher_x' in final_augmented.columns: + final_augmented = final_augmented.rename(columns={'publisher_x': 'publisher'}) + if 'PolicyName_x' in final_augmented.columns: + final_augmented = final_augmented.rename(columns={'PolicyName_x': 'PolicyName'}) + + final_augmented.to_csv("testing.csv", index=False) + + # Reputation flag logic + def reputationtool(row): + val = row["reputation_scannermatch"] + if pd.isna(val): + return row["publisher"] == "Not Signed" + try: + return int(val) > threat_tolerance + except (ValueError, TypeError): + return row["publisher"] == "Not Signed" + + df = final_augmented.copy() + df["reputation_flag"] = df.apply(reputationtool, axis=1) + + # Filtering logic + mask_needsreview = ( + ((df["publisher"] == "Not Signed") & df["reputation_flag"]) | + (df["reputation_status"] == "UNKNOWN") + ) + + mask_approved = ( + ( + (df["publisher"] != "Not Signed") & + ~df["publisher"].str.contains(pathf.regulator(untrusted_publishers), case=False, na=False) & + ~df["reputation_status"].isna() & + ~df["description"].str.contains(pathf.regulator(pups), case=False, na=False) + ) | + ( + (df["publisher"] == "Not Signed") & + ~df["reputation_flag"] & + ~df["publisher"].str.contains(pathf.regulator(untrusted_publishers), case=False, na=False) & + ~df["reputation_status"].isna() & + ~df["description"].str.contains(pathf.regulator(pups), case=False, na=False) + ) + ) + + needsreview_df = df[mask_needsreview] + approved_df = df[mask_approved] + all_approved_hashes = pd.concat([needsreview_df, approved_df], ignore_index=True) + + print(ct.colorText("Beginning calculating longest common filepaths for path exceptions", "green")) + + # Path analysis + haslcp = pathf.split_filepaths_grouped(all_approved_hashes, "filename", path_exclusion_constant, min_files_for_path) + haslcp = haslcp.drop_duplicates() + + # Remove forbidden paths + forbidden = pathf.regulator(badpathparts, True) + lcp_not_forbidden = haslcp[~haslcp["longestcfp"].str.contains(forbidden, na=False)].copy() + + print(ct.colorText("Removing forbidden filepaths for path exceptions", "green")) + + # Reviewable paths + review_df = lcp_not_forbidden[['longestcfp', 'middle', 'filename_only', 'file_extension', 'sha256']] + sha_counts = review_df.groupby('longestcfp')['sha256'].nunique().reset_index() + sha_counts.columns = ['longestcfp', 'unique_sha256_count'] + + review_df = review_df.merge(sha_counts, on='longestcfp', how='left') + review_df = review_df[review_df['unique_sha256_count'] >= min_files_for_path] + + review_df.to_csv("ALL_AT_PATHS.csv", index=False) + + # Cleanup + del lcp_not_forbidden, sha_counts, review_df + +def mergeTesting(): + # Load the two CSVs + testing_df = pd.read_csv("testing.csv") + paths_df = pd.read_csv("ALL_AT_PATHS.csv") + + # Merge on 'sha256' with testing as the left DataFrame + merged_df = testing_df.merge(paths_df, on="sha256", how="left") + + # Save the merged result + merged_df.to_csv("merged_output.csv", index=False) + + print(f"Merged DataFrame saved with {len(merged_df)} rows.") diff --git a/utils/policyfunctions.py b/utils/policyfunctions.py index c3f16bf..ecf4e5d 100644 --- a/utils/policyfunctions.py +++ b/utils/policyfunctions.py @@ -93,16 +93,16 @@ def addPubReal(url, grouplistID, publist): print(response.text) -def getPolicyInfo(url, policy, days): +def getPolicyInfo(url, policy, type, days, parquet=True): executionhist_policy = pd.DataFrame() - exehist = pullPolicyExechistories(url, policy, days, True) + exehist = pullPolicyExechistories(url, policy, type, days, True) data = json.loads(exehist) executionhist_policy = pd.DataFrame(data["response"]["exechistories"]) if not executionhist_policy.empty: - executionhist_policyxecutionhist_policy = executionhist_policy[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']] + executionhist_policy = executionhist_policy[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']] executionhist_policy = executionhist_policy.drop_duplicates(subset=['sha256', 'filename', 'hostname']) executionhist_policy = executionhist_policy.sort_values(by=['sha256', 'filename']) - executionhist_policy.to_parquet(f"parquet\\execution_history_{policy}.parquet", index=False) + if parquet: executionhist_policy.to_parquet(f"parquet\\execution_history_{policy}.parquet", index=False) print(ct.colorText(f"Staging of Execution history for policy: {policy} is complete", "green")) del data del exehist @@ -157,7 +157,7 @@ def sendToPolicy(url, first_policy, second_policy, destination_name, destination else: print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red")) -def pullPolicyExechistories(url, policiesnames, days, outputjson: bool): +def pullPolicyExechistories(url, policiesnames, type, days, outputjson: bool): file_path = 'chunkinator.json' if not os.path.exists(file_path): with open(file_path, 'w') as file: @@ -171,7 +171,7 @@ def pullPolicyExechistories(url, policiesnames, days, outputjson: bool): with tqdm.tqdm(file=sys.stdout, leave=True, total=10000, desc=f"Checkpoint Progess: {checkpoint}", colour="blue", initial=1) as filebar: with tqdm.tqdm(file=sys.stdout, leave=True, total=100, desc=f"Total of {policiesnames} Complete: ") as pbar: while True: - json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers) + json_response_data = checkpoint_stomper(checkpoint, url, type, policiesnames, headers) histories = json_response_data['response']['exechistories'] filebar.total=len(histories) if not histories: @@ -214,11 +214,11 @@ def pullPolicyExechistories(url, policiesnames, days, outputjson: bool): os.remove(file_path) return json.dumps(final_output) if outputjson else None -def checkpoint_stomper(checkpoint, url, policy, headers): +def checkpoint_stomper(checkpoint, url, type, policy, headers): json_output = {'error': 'Success', 'response': {'exechistories': []}} endpoint = url + '/v1/logging/exechistories' payload_dict = { - "type":[1,2,6,7], + "type":[type], "checkpoint": checkpoint, "policy": [policy] } @@ -251,6 +251,37 @@ def listPolicies(url): choice = int(choice) - 1 return choice, policiesnames, policyids +def listATPolicies(url): + endpoint = url + '/v1/group' + print(ct.colorText("[+] Grabbing All Policies", "cyan")) + + headers = { + "X-APIKey": os.getenv('APIKEY') + } + + try: + response = requests.post(endpoint, headers=headers, json={}, verify=False) + response.raise_for_status() + parse_text = response.json() + + at_policies = {} + + for index, group in enumerate(parse_text.get('response', {}).get('groups', []), start=1): + name = group.get('name', '') + if "AT" in name: + print(ct.colorText(f"{index}. {name}", "yellow")) + at_policies[name] = group.get('groupid') + + return at_policies + + 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 listAllowlists(url): endpoint = url + '/v1/application' print(ct.colorText("[+] Grabbing All Allowlists", "cyan"))