diff --git a/AirlockTools.py b/AirlockTools.py index 0995fe8..de6960b 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -1,6 +1,10 @@ import dotenv import os import utils.getdeviceevents +import utils.allowlist +import urllib3 + +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) dotenv.load_dotenv() @@ -18,12 +22,13 @@ def apivalidation(): def menu(): print("\n--- Main Menu ---") print("1. Get All Events for Single Device") - print("2. Policy Enforcement Readiness") - print("3. Get Device with Highest Blocks in 7 Days") + print("2. Get Execution Histories for Allow List") while True: choice = input("Enter Menu Item: ") if choice == '1': utils.getdeviceevents.devicehistory(url) + if choice == '2': + utils.allowlist.allowlistexechistories(url) if __name__ == "__main__": apivalidation() \ No newline at end of file diff --git a/utils/__pycache__/allowlist.cpython-313.pyc b/utils/__pycache__/allowlist.cpython-313.pyc new file mode 100644 index 0000000..3f3e038 Binary files /dev/null and b/utils/__pycache__/allowlist.cpython-313.pyc differ diff --git a/utils/__pycache__/getdeviceevents.cpython-313.pyc b/utils/__pycache__/getdeviceevents.cpython-313.pyc index 3b24f94..f039bb8 100644 Binary files a/utils/__pycache__/getdeviceevents.cpython-313.pyc and b/utils/__pycache__/getdeviceevents.cpython-313.pyc differ diff --git a/utils/allowlist.py b/utils/allowlist.py new file mode 100644 index 0000000..bb4d75a --- /dev/null +++ b/utils/allowlist.py @@ -0,0 +1,53 @@ +import datetime +import requests +import json +import os + +def allowlistexechistories(url): + endpoint = url + '/v1/group' + print("[+] Grabbing All Policies") + payload = {} + headers = { + "X-APIKey": os.getenv('APIKEY') + } + response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) + parse_text = json.loads(response.text) + policiesnames = [] + policyids = [] + for index, list in enumerate(parse_text['response']['groups'], start=1): + print(f"{index}. {list['name']}") + policiesnames.append(list['name']) + policyids.append(list['groupid']) + choice = input("Select Policy Group: ") + choice = int(choice) - 1 + endpoint = url + '/v1/logging/exechistories' + payload_dict = { + "type":[1], + "checkpoint":"68a153c23963989b484541b4", + "policy": [policiesnames[choice]] + } + payload = json.dumps(payload_dict) + print(payload) + response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) + parse_text = json.loads(response.text) + for item in parse_text['response']['exechistories']: + print(item['checkpoint']) + print(item['datetime']) + print(item['hostname']) + print(item['filename']) + # checkpoint_stomper(item['checkpoint'], endpoint, headers, policiesnames[choice]) + +def checkpoint_stomper(checkpoint, endpoint, headers, policyname): + print(checkpoint) + payload_dict = { + "type":[1,2,6,7], + "checkpoint": checkpoint, + "policy":[policyname] + } + payload = json.dumps(payload_dict) + response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) + parse_text = json.loads(response.text) + # Send Whole JSON Response + for item in parse_text['response']['exechistories']: + print("test") + diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py new file mode 100644 index 0000000..ff57f43 --- /dev/null +++ b/utils/hashfunctions.py @@ -0,0 +1,84 @@ +import pandas as pd +import requests +import os + +def aggregateHashes(executions_json: dict) -> pd.DataFrame: + """ + Takes the executions, aggregates all the data with sha256 as primary, then returns aggregated dataframe + """ + + exechistories = executions_json.get("response", {}).get("exechistories", []) + df = pd.DataFrame(exechistories) + + if df.empty: + return df + + # Aggregate by sha256 - keep all entries in lists + agg_df = df.groupby("sha256").agg(lambda x: list(x)).reset_index() + + return agg_df + +def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame: + """ + Takes output of aggregatedHashes, queries API for those hashes, flattens response while keeping one row per hash, + aggregate applications and baselines into lists, then merges results back into agg_df to create a + """ + endpoint = url + 'v1/hash/query' + payload = agg_df['sha256'].tolist() + headers = {"X-APIKey": os.getenv('APIKEY')} + + response = requests.post(endpoint, headers=headers, json=payload, verify=False) + data = response.json() + results = data.get("response", {}).get("results", []) + + rows = [] + for res in results: + row = {"sha256": res.get("sha256"), "result": res.get("result")} + + if "data" in res: + d = res["data"] + for key in ["filename", "filepath", "description", "filesize", "md5", + "productname", "productversion", "publisher", "createtime", "modtime", + "sha128", "sha384", "sha512", "datetime"]: + row[key] = d.get(key) + + row["applications"] = d.get("applications", []) + row["baselines"] = d.get("baselines", []) + + reputation = d.get("reputation", {}) + for k, v in reputation.items(): + row[f"reputation_{k}"] = v + + rows.append(row) + + df_api = pd.DataFrame(rows) + + aug_df = agg_df.merge(df_api, on="sha256", how="left") + + return aug_df + +def categorize_hashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list): + """ + Categorize hashes into needsreview, approved, and remaining based on publisher and threat level. + + """ + if untrusted_publishers is None: + untrusted_publishers = [] + + # Flatten threatlevel from nested reputation dict + df = df.copy() + df["threatlevel"] = df["reputation"].apply(lambda x: x.get("threatlevel") if pd.notnull(x) else None) + + # Masks for each category + mask_needsreview = (df["publisher"] == "Not Signed") & (df["threatlevel"] > threat_tolerance) + mask_approved = (df["publisher"] != "Not Signed") & (~df["publisher"].isin(untrusted_publishers)) + + # Create DataFrames for each category + needsreview_df = df[mask_needsreview].drop(columns=["threatlevel"]) + approved_df = df[mask_approved].drop(columns=["threatlevel"]) + remaining_df = df[~(mask_needsreview | mask_approved)].drop(columns=["threatlevel"]) + + return needsreview_df, approved_df, remaining_df + +def approve_hashes(approved_df: pd.DataFrame): + pass \ No newline at end of file