ALL AT PATHS TOOL

utils/pathfunctions.py

@@ -14,6 +14,7 @@
 #    along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 #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.")