# Copyright (C) 2025 James Brotosky, Brandon Wickline # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published # by the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import argparse import dotenv import os import urllib3 import utils.clientfunctions import utils.localapproval as la import utils.otpfunctions import utils.policyfunctions import utils.utils as ct import pandas as pd from utils.perstscheduler import register_function, run_once_job, recurring_job, reload_jobs, start_scheduler urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) dotenv.load_dotenv() #Constants 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"3 path_exclusion_constant = 4 min_files_for_path = 4 threat_tolerance_constant = 4 policy_relationship_map ={ #Enforcement : Audit "bf0b1f9b-bfea-4f44-97c0-80e27ff61712" : "538d3218-92f4-4943-a6ee-db9267ab62d8", #AT Servers General, AT Servers General Audit "31ababac-65de-4c6a-86dd-6691d7e3ee3b" : "fc05b42a-b846-4e72-88ca-c35d416e699f", #AT Epic, #AT Epic Audit "d1f58960-f866-49e0-848a-a5b09fffd4cd" : "d55c03a6-c376-4391-8626-4f843b882a7c", #AT DMZ Enforced, #AT DMZ Audit "504dd011-86b6-489a-b78f-eff589cef8aa" : "88a1cfdc-3b30-448b-b309-be16fe437ca3", #AT Workstations BCA, #AT Workstations BCA Audit "d126db36-72ed-4937-adc7-d88b7509a5b5" : "5aebf6a0-1d67-47b4-9c5f-2866ffca5671" #AT Testing, AT Testing } def main(): parser = argparse.ArgumentParser(description="Your script description") parser.add_argument('--monitorOTP', action='store_true', help='Run in non-interactive mode') # Add other arguments as needed args = parser.parse_args() if args.monitorOTP: # Non-interactive logic print(f"Running non-interactively to start monitoring OTP") os.makedirs("scheduling", exist_ok=True) os.makedirs("OTP/HTML", exist_ok=True) os.makedirs("OTP/PARQ", exist_ok=True) os.makedirs("Local_Approval/HTML", exist_ok=True) os.makedirs("Local_Approval/PARQ", exist_ok=True) ct.apivalidation() register_function("monitorOTP", utils.otpfunctions.monitorOTP) register_function("monitorLA", la.scheduleAddingLAHashes) register_function("updateAuditPolicies", utils.policyfunctions.updateAuditPoliciesFromEnforcementPolices) if not os.path.exists("scheduling\\jobs.json"): recurring_job("monitorOTP", "monitorOTP", interval=60, unit="seconds", args=[url, pups]) recurring_job("monitorLA", "monitorLA", interval=50, unit="seconds", args=[url, policy_relationship_map, bad_publisher_list, pups, threat_tolerance_constant]) recurring_job("updateAuditPolicies", "updateAudit", interval=5, unit="minutes", args=[url, policy_relationship_map]) else: reload_jobs() start_scheduler() else: # Interactive logic ct.apivalidation() menu_main() def menu_main(): while True: ct.displayIntro(); print(ct.colorText("1. 🖥️ - Get All Events for Single Device", "yellow")) print(ct.colorText("2. 🎫 - OTP", "yellow")) print(ct.colorText("3. 🔇 - Find Quiet Hosts", "yellow")) print(ct.colorText("4. 🔒 - Prepare Policy For Enforcement", "yellow")) print(ct.colorText("5. 🔄 - Update Audit Policies from Enforcement Policies", "yellow")) print(ct.colorText("6. 🔍 - Device Search", "yellow")) print(ct.colorText("7. ➡️ - Move Devices to Local Approval", "yellow")) print(ct.colorText("Q. 🔚 - Quit", "yellow")) choice = input(ct.colorText("\nEnter Menu Item: ", "white")) if choice == '1': utils.clientfunctions.devicehistory(url,False) elif choice == "2": menu_otp() elif choice == "3": utils.clientfunctions.findQuietAgents(url) elif choice == "4": utils.policyfunctions.prepare_to_enforce(url, bad_publisher_list, pups, badpathparts, threat_tolerance_constant, path_exclusion_constant, min_files_for_path) elif choice == "5": ct.areYouSure() confirmation = input(ct.colorText("Type 'I AGREE' to continue: ","white")) if confirmation.strip().upper() == "I AGREE": utils.policyfunctions.updateAuditPoliciesFromEnforcementPolices(url, policy_relationship_map) elif choice == "6": devicelist = utils.clientfunctions.promptForDevices() utils.clientfunctions.findAgents(url, devicelist, False) elif choice == "7": la.moveToLocalApproval(url, policy_relationship_map) elif choice == "8": las = la.getLocalApprovals(url) las.to_csv("la.csv", index=False) elif choice == "9": pass elif choice == "10": utils.clientfunctions.moveAgentToAudit(url,"6a221ece-0c10-4eb8-b1e5-06a1000a5696",policy_relationship_map) elif choice == "11": utils.clientfunctions.moveAgentToEnforcement(url,"6a221ece-0c10-4eb8-b1e5-06a1000a5696",policy_relationship_map) elif choice == "Q": break else: print(ct.colorText("Invalid choice. Please try again.","red")) def menu_otp(): while True: print(ct.colorText("\n--- 🎫 OTP Submenu 🎫 ---","cyan")) print(ct.colorText("1. Generate OTP","cyan")) #print(ct.colorText("2. Sub-option B","cyan")) print(ct.colorText("Q. Return to Main Menu","cyan")) choice = input("Enter your choice: ") if choice == "1": utils.otpfunctions.generateOTP(url, utils.clientfunctions.findAgentID(url)) break elif choice == "2": print("You selected Sub-option B") elif choice == "Q": print("Returning to Main Menu...") break else: print("Invalid choice. Please try again.") if __name__ == "__main__": main()