mysticmomba/AirlockTools
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

LocalApproval in progress

Browse Source
This commit is contained in:
Zarithas
2025-09-29 21:21:27 -04:00
parent 7e669ea14f
commit 789a8ed975
5 changed files with 298 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/clientfunctions.py
+89 -115
View File
@@ -29,6 +29,8 @@ import re
import pandas as pd
import requests
def findAgentID(url):
print(ct.colorText("WARNING: Device Name is Case Sensitive", "red"))
@@ -65,16 +67,9 @@ def getDestAllowlistFromClientID(url, clientid):
result = json.loads(response.text)
data = pd.DataFrame(result["response"]["agents"])
allowlists = getPolicyAllowlists(url,data.loc[0, "groupid"])
allowlist = getDestAllowlist(url,data.loc[0, "groupid"])
matches = allowlists.loc[
allowlists['name'].str.contains('local', case=False, na=False) &
allowlists['name'].str.contains('approval', case=False, na=False),
'applicationid'
].values
app_id = matches[0] if len(matches) > 0 else None
return app_id
return allowlist
def getPolicyFromClientID(url, clientid):
@@ -91,24 +86,11 @@ def getPolicyFromClientID(url, clientid):
result = json.loads(response.text)
data = pd.DataFrame(result["response"]["agents"])
policy = getPolicyName(url,data.loc[0, "groupid"])
return policy
policy_name = getPolicyName(url,data.loc[0, "groupid"])
policy_id = data.loc[0, "groupid"]
return policy_name, policy_id
def getPolicyAllowlists(url, groupid):
endpoint = url + '/v1/group/policies'
payload = {
"groupid" : f"{groupid}"
}
headers = {"X-APIKey": os.getenv('APIKEY')}
payload = json.dumps(payload)
response = requests.post(endpoint, headers=headers, data=payload, verify=False)
result = json.loads(response.text)
data = pd.DataFrame(result["response"]["applications"])
return data
def getPolicyName(url, groupid):
endpoint = url + '/v1/group/'
@@ -272,90 +254,9 @@ def promptForDevices():
return device_input_str
def returnToEnforcement(url, device_df, policy_relationship_map, bad_publisher_list, pups, badpathparts, threat_tolerance_constant, path_exclusion_constant, min_files_for_path):
policylist = sorted(device_df['policy_name'].unique().tolist())
type = [1, 2, 6, 7]
parq_base_dir = "prepare_policy\\parquet\\"
appr_base_dir = "prepare_policy\\approved\\"
needappr_base_dir = "prepare_policy\\needs_approved"
pflight_base_dir = "prepare_policy\\preflight"
#If the directorys where we're going to store our output dont exist, make them.
os.makedirs(parq_base_dir, exist_ok=True)
os.makedirs(needappr_base_dir, exist_ok=True)
os.makedirs(appr_base_dir, exist_ok=True)
os.makedirs(pflight_base_dir, exist_ok=True)
while True:
#ct.printDeviceEnforceChecklist()
choice = input(ct.colorText("\nEnter your choice: ", "white"))
if choice == "1":
policyf.buildExecHistory(url,
policylist,
parq_base_dir,
needappr_base_dir,
type,
threat_tolerance_constant,
bad_publisher_list,
pups,
)
elif choice == "2":
pathf.generateApprovables(needappr_base_dir, appr_base_dir, parq_base_dir, badpathparts, bad_publisher_list, path_exclusion_constant, min_files_for_path, True)
elif choice == "3":
policyf.savePreFlights(appr_base_dir, parq_base_dir, pflight_base_dir)
"""
elif choice == "4":
if os.path.exists(f"preflight\\final_path_exclusions.html") and os.path.exists(f"preflight\\final_hash_approvals.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ":
sendToPolicyTest(
url,
first_policy,
second_policy,
destination_name,
destination_id,
allowlist_parent_name,
allowlist_parent_id,
allowlist_child_name,
allowlist_child_id
)
elif choice == "5":
if os.path.exists(f"preflight\\final_path_exclusions.html") and os.path.exists(f"preflight\\final_hash_approvals.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ":
sendToPolicy(
url,
first_policy,
second_policy,
destination_name,
destination_id,
allowlist_parent_name,
allowlist_parent_id,
allowlist_child_name,
allowlist_child_id
)
elif choice == "R":
pathf.clean_folders(enforcement_prep)
elif choice == "Q":
break
else:
print(ct.colorText("Invalid choice. Please try again.", "red"))
"""
def findQuietAgents(url):
# Get policy selection and agent list
choice, policynames, policyid = policyf.listPolicies(url)
choice, policynames, policyid = policyf.choosePolicies(url)
policy = policynames[choice]
groupid = policyid[choice]
agents = findGroupAgents(url, groupid)
@@ -418,15 +319,18 @@ def findQuietAgents(url):
print(ct.colorText(f"Saving CSV to {policy}_agents_last_{history_days}_days.csv", "green"))
agents.to_csv(f"{policy}_agents_last_{history_days}_days.csv", index=False)
# Summary stats
zero_count = (agents['execution_count'] == 0).sum()
total_hosts = len(agents)
zero_percentage = (zero_count / total_hosts) * 100
print(f"Number of hosts in policy: {total_hosts}")
print(f"Number of hosts with execution_count = 0: {zero_count}")
print(f"Percentage of hosts with execution_count = 0: {zero_percentage:.2f}%")
# Summary statistics
total_agents = len(agents)
ready_agents = agents['enforce_ready'].sum()
not_ready_agents = total_agents - ready_agents
ready_percentage = (ready_agents / total_agents) * 100
# Print results
print(f"Total agents: {total_agents}")
print(f"Agents marked as 'enforce_ready': {ready_agents}")
print(f"Agents not ready: {not_ready_agents}")
print(f"Percentage ready for enforcement: {ready_percentage:.2f}%")
def findGroupAgents(url, groupid):
endpoint = url + '/v1/agent/find'
@@ -456,4 +360,74 @@ def findGroupAgents(url, groupid):
data['status'] = data['status'].map(status_map)
return(data)
return(data)
def moveAgentToAudit(url, agentid, policy_relationship_map):
policy_name, policy_id = getPolicyFromClientID(url, agentid)
print(policy_id)
if policy_id in policy_relationship_map:
target_policy = policy_relationship_map[policy_id]
elif policy_id in policy_relationship_map.values():
print(f"Agent {agentid} is already in an audit group. No action needed.")
return
else:
print(f"Error: No corresponding audit policy found for policy: {policy_name} - {policy_id}.")
return
moveAgent(url, agentid, target_policy, "audit")
def moveAgentToEnforcement(url, agentid, policy_relationship_map):
policy_name, policy_id = getPolicyFromClientID(url, agentid)
# Invert the map for audit → enforcement
inverse_map = {v: k for k, v in policy_relationship_map.items()}
if policy_id in inverse_map:
target_policy = inverse_map[policy_id]
elif policy_id in inverse_map.values():
print(f"Agent {agentid} is already in an enforcement group. No action needed.")
return
else:
print(f"Error: No corresponding enforcement policy found for policy: {policy_name} - {policy_id}.")
return
moveAgent(url, agentid, target_policy, "enforcement")
def moveAgent(url, agentid, target_policy, direction):
endpoint = f"{url}/v1/agent/move"
payload = {
"groupid": target_policy,
"agentid": agentid
}
headers = {"X-APIKey": os.getenv('APIKEY')}
response = None # Initialize to avoid unbound errors
try:
response = requests.post(endpoint, headers=headers, data=json.dumps(payload), verify=False)
response.raise_for_status() # Raises HTTPError for bad status codes
result = response.json()
# Check if 'error' key exists and if it's not a success message
if "error" in result and result["error"].lower() != "success":
print(f"API returned an error: {result['error']}")
else:
print(f"✅ Agent {agentid} successfully moved to {direction} group {target_policy}.")
except requests.exceptions.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
if response is not None:
print("Raw response:", response.text)
except requests.exceptions.RequestException as req_err:
print(f"Request error occurred: {req_err}")
except ValueError:
print("Failed to parse JSON response.")
if response is not None:
print("Raw response:", response.text)
except Exception as e:
print(f"Unexpected error: {e}")
if response is not None:
print("Raw response:", response.text)