Chris Approves

AirlockTools.py

@@ -33,8 +33,20 @@ def menu():
 			utils.allowlist.allowlistexechistories(url,False)
 		if choice == '3':
 			executionhist = utils.allowlist.allowlistexechistories(url,True)
+			print(executionhist)
 			aggregated = utils.hashfunctions.aggregateHashes(executionhist)
 			print(aggregated)
+			augmented = utils.hashfunctions.augmentAggregatedHashes(url,aggregated)
+			print(augmented)
+
+			augmented.to_html("augmentedlist.html", index=False)
+
+			badpublisherlist = []
+			categorized = utils.hashfunctions.categorizeHashes(augmented, 5, badpublisherlist)
+			categorized[0].to_html("needsreview.html", index=False)
+			categorized[1].to_html("approved.html", index=False)
+			categorized[2].to_html("remaining.html", index=False)
+
 
 if __name__ == "__main__":
     apivalidation()

utils/allowlist.py

@@ -22,15 +22,17 @@ def allowlistexechistories(url, outputjson: bool):
     choice = int(choice) - 1
     endpoint = url + '/v1/logging/exechistories'
     payload_dict = {
-        "type":[1],
+        "type":[1, 2, 6, 7],
         "checkpoint":"68a153c23963989b484541b4",
         "policy": [policiesnames[choice]]
     }
     payload = json.dumps(payload_dict)
     print(payload)
     response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
+    
     if outputjson == True:
         return response
+    
     parse_text = json.loads(response.text)
     for item in parse_text['response']['exechistories']:
         print(item['checkpoint'])

utils/hashfunctions.py

@@ -8,22 +8,6 @@ def aggregateHashes(executions_json) -> pd.DataFrame:
     """
     Takes the executions,  aggregates all the data with sha256 as primary, then returns aggregated dataframe
     """
-    """
-   old version 
-   data = executions_json.json()
-
-    df = pd.DataFrame(data["response"]["exechistories"])
-    print(df)
-    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
-    """
-
-
     data = executions_json.json()
     df = pd.DataFrame(data["response"]["exechistories"])
 
@@ -46,11 +30,15 @@ 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')}
+    endpoint = url + '/v1/hash/query'
+    payload = {
+            "hashes": agg_df['sha256'].tolist()
+    }
     
-    response = requests.post(endpoint, headers=headers, json=payload, verify=False)
+    headers = {"X-APIKey": os.getenv('APIKEY')}
+    payload = json.dumps(payload)
+  
+    response = requests.post(endpoint, headers=headers, data=payload, verify=False)
     data = response.json()  
     results = data.get("response", {}).get("results", [])
     
@@ -80,28 +68,66 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame:
 
     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.
+def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list):
+   if untrusted_publishers is None:
+       untrusted_publishers = []
+
+   df = aug_df.copy()
+   def reputationtool(row, threat_tolerance):
+        if row["reputation_scannermatch"] == "N/A":
+            return True
+        try:
+            if int(row["reputation_scannermatch"]) > threat_tolerance:
+                return True
+        except (ValueError, TypeError):
+            pass
+        return False
+
+   mask_needsreview = (df["publisher_y"] == "Not Signed") & df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1)
+   mask_approved = (df["publisher_y"] != "Not Signed") & (~df["publisher_y"].isin(untrusted_publishers))
+
+   needsreview_df = df[mask_needsreview]
+   approved_df = df[mask_approved]
+   remaining_df = df[~(mask_needsreview | mask_approved)]
+
+   return needsreview_df, approved_df, remaining_df
+
  
 """
+def categorizeHashes(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 = aug_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))
+    mask_needsreview = ((df["publisher_y"] == "Not Signed") & reputationtool(df)) 
+
+    print(mask_needsreview)
+
+    mask_approved = (df["publisher_y"] != "Not Signed") & (~df["publisher_y"].isin(untrusted_publishers))
+
+    print(mask_approved)
 
     # 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"])
+    needsreview_df = df[mask_needsreview]
+    approved_df = df[mask_approved]
+    remaining_df = df[~(mask_needsreview | mask_approved)]
 
     return needsreview_df, approved_df, remaining_df
 
-def approve_hashes(approved_df: pd.DataFrame):
+def approvehashes(approved_df: pd.DataFrame):
     pass
+
+def reputationtool(df):
+    if df["reputation_scannermatch"] == "N/A":
+        return True
+    if df["reputation_scannermatch"].astype(int) > 3:
+        return True
+    return False
+    """