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

fix(policy-prep): implement table editors and workflow improvements

Browse Source 
- Add table editors for Policy Prep workflow
- 'Add to policy' remains a placeholder
- Apply planned tweaks:
  - Replace ballot checkbox with ✓ for selection
  - Relocate loading screen text to bottom:
    'Building Path exclusions and publisher lists...
     This may take a moment for large datasets.'
  - Ensure interaction with all tables before allowing review steps
  - Move excessive logging to debug level
  - Add Step 0 to explain process before user begins

Notes:
Further discussion needed on enforcing table interaction before review.
This commit is contained in:
Zarithas
2025-12-11 16:55:12 -05:00
parent 59bb97ec4e
commit 797d0f4462
3 changed files with 1305 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TUI/Screens/policyprepworkflowscreen.py
+1266 -271
View File
File diff suppressed because it is too large Load Diff
TUI/Widgets/prepPolicy.py
+24 -24
View File
@@ -88,9 +88,9 @@ def sortHashes(
): ):
working_dir = load_env("WORKING_DIR") working_dir = load_env("WORKING_DIR")
history_days = Selector.select_value( history_days = Selector.select_value(
prompt="Enter how many days of history to pull (1–150): ", prompt="Enter how many days of history to pull (1-365): ",
value_type=int, value_type=int,
valid_range=(1, 150), valid_range=(1, 365),
) )
logger.debug(f"{history_days} day selected for history") logger.debug(f"{history_days} day selected for history")
@@ -655,7 +655,7 @@ def section_header(title):
def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist): def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist):
working_dir = load_env("WORKING_DIR") working_dir = load_env("WORKING_DIR")
section_header("Prepare to Enforce Policy ") section_header("Prepare to Enforce Policy")
print( print(
colorText( colorText(
"\nSequentially follow these steps to prepare a policy for enforcement:", "\nSequentially follow these steps to prepare a policy for enforcement:",
@@ -670,11 +670,11 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
) )
) )
if not selected_policies: if not selected_policies:
print(colorText(" [✗] No policies have been chosen", "red")) print(colorText(" [❌] No policies have been chosen", "red"))
else: else:
print(colorText("The following policies have been chosen:", "green")) print(colorText("The following policies have been chosen:", "green"))
for policy in selected_policies: for policy in selected_policies:
print(colorText(f" [✓] {policy.name}", "green")) print(colorText(f" [✅] {policy.name}", "green"))
# Step 2: Destination Policy and Allowlist # Step 2: Destination Policy and Allowlist
print( print(
@@ -683,22 +683,22 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
if destination_policy: if destination_policy:
print( print(
colorText( colorText(
f" [✓] {destination_policy[0].name} has been selected as the destination policy", f" [✅] {destination_policy[0].name} has been selected as the destination policy",
"green", "green",
) )
) )
else: else:
print(colorText(" [✗] No destination policy has been chosen", "red")) print(colorText(" [❌] No destination policy has been chosen", "red"))
if destination_allowlist: if destination_allowlist:
print( print(
colorText( colorText(
f" [✓] {destination_allowlist[0].name} has been selected as allowlist", f" [✅] {destination_allowlist[0].name} has been selected as allowlist",
"green", "green",
) )
) )
else: else:
print(colorText(" [✗] No allowlist has been chosen", "red")) print(colorText(" [❌] No allowlist has been chosen", "red"))
# Step 3: Data Preparation # Step 3: Data Preparation
print( print(
@@ -713,9 +713,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print( print(
colorText( colorText(
( (
" [✓] Data has been fetched" " [✅] Data has been fetched"
if os.path.exists(review_path) if os.path.exists(review_path)
else " [✗] Data has not been fetched" else " [❌] Data has not been fetched"
), ),
"green" if os.path.exists(review_path) else "red", "green" if os.path.exists(review_path) else "red",
) )
@@ -723,7 +723,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else: else:
print( print(
colorText( colorText(
" [✗] No policies selected, cannot check data fetch status", "red" " [❌] No policies selected, cannot check data fetch status", "red"
) )
) )
@@ -756,9 +756,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print( print(
colorText( colorText(
( (
" [✓] Reviewed hashes have been loaded" " [✅] Reviewed hashes have been loaded"
if os.path.exists(approved_path) if os.path.exists(approved_path)
else " [✗] Reviewed hashes have not been loaded" else " [❌] Reviewed hashes have not been loaded"
), ),
"green" if os.path.exists(approved_path) else "red", "green" if os.path.exists(approved_path) else "red",
) )
@@ -766,9 +766,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print( print(
colorText( colorText(
( (
" [✓] Path review list created" " [✅] Path review list created"
if os.path.exists(second_review_path) if os.path.exists(second_review_path)
else " [✗] Path review list has not been created" else " [❌] Path review list has not been created"
), ),
"green" if os.path.exists(second_review_path) else "red", "green" if os.path.exists(second_review_path) else "red",
) )
@@ -776,7 +776,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else: else:
print( print(
colorText( colorText(
" [✗] No policies selected, cannot check reviewed hashes or path list", " [❌] No policies selected, cannot check reviewed hashes or path list",
"red", "red",
) )
) )
@@ -812,9 +812,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print( print(
colorText( colorText(
( (
" [✓] Reviewed path list detected" " [✅] Reviewed path list detected"
if os.path.exists(reviewed_path) if os.path.exists(reviewed_path)
else " [✗] Path review list has not been detected" else " [❌] Path review list has not been detected"
), ),
"green" if os.path.exists(reviewed_path) else "red", "green" if os.path.exists(reviewed_path) else "red",
) )
@@ -825,9 +825,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print( print(
colorText( colorText(
( (
" [✓] Preflight Path Exclusion List has been generated" " [✅] Preflight Path Exclusion List has been generated"
if preflight_ready if preflight_ready
else " [✗] Preflight Path Exclusion List has not been generated" else " [❌] Preflight Path Exclusion List has not been generated"
), ),
"green" if preflight_ready else "red", "green" if preflight_ready else "red",
) )
@@ -835,7 +835,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else: else:
print( print(
colorText( colorText(
" [✗] No policies selected, cannot check preflight status", "red" " [❌] No policies selected, cannot check preflight status", "red"
) )
) )
@@ -866,5 +866,5 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(colorText(" Apply approved hashes to allowlist", "cyan")) print(colorText(" Apply approved hashes to allowlist", "cyan"))
# Utility Options # Utility Options
print(colorText("F. Open Working Directory", "cyan")) print(colorText("F. Open Working Directory", "cyan"))
print(colorText("B. Back", "cyan")) print(colorText("B. Back", "cyan"))
services/agenthandler.py
+15 -15
View File
@@ -38,9 +38,9 @@ logger = logging.getLogger(__name__)
def devicehistory(api: AirlockAPIWrapper, outputjson: bool): def devicehistory(api: AirlockAPIWrapper, outputjson: bool):
agents = selectAgents(api) agents = selectAgents(api)
history_days = Selector.select_value( history_days = Selector.select_value(
prompt="Enter how many days of history to pull (1–150): ", prompt="Enter how many days of history to pull (1–365): ",
value_type=int, value_type=int,
valid_range=(1, 150), valid_range=(1, 365),
) )
if not agents or not history_days: if not agents or not history_days:
@@ -60,7 +60,7 @@ def devicehistory(api: AirlockAPIWrapper, outputjson: bool):
except Exception as e: except Exception as e:
print( print(
colorText( colorText(
f"❌ Error retrieving history for {agent.hostname}: {e}", "red" f"❌ Error retrieving history for {agent.hostname}: {e}", "red"
) )
) )
continue continue
@@ -139,7 +139,7 @@ def findAgents(api, return_dataframe):
print( print(
colorText( colorText(
f"\n✓ Matched devices exported to: {working_dir}\\{filename}", f"\n✓ Matched devices exported to: {working_dir}\\{filename}",
"green", "green",
) )
) )
@@ -148,7 +148,7 @@ def findAgents(api, return_dataframe):
def collect_device_names() -> List[str]: def collect_device_names() -> List[str]:
print(colorText("🖥��Â Device Search", "cyan")) print(colorText("🖥��Â Device Search", "cyan"))
print( print(
colorText( colorText(
"Enter the device hostnames you'd like to search for, one per line.", "cyan" "Enter the device hostnames you'd like to search for, one per line.", "cyan"
@@ -185,7 +185,7 @@ def collect_device_names() -> List[str]:
else: else:
print( print(
colorText( colorText(
f"⚠️ Invalid input: '{stripped_line}' — only letters, numbers, underscores, spaces, and hyphens are allowed.", f"⚠️ Invalid input: '{stripped_line}' — only letters, numbers, underscores, spaces, and hyphens are allowed.",
"yellow", "yellow",
) )
) )
@@ -235,8 +235,8 @@ def show_unmatched(
] ]
if unmatched: if unmatched:
logger.debug(f"⚠️ No matches for: {', '.join(unmatched)}") logger.debug(f"⚠️ No matches for: {', '.join(unmatched)}")
print(colorText(f"⚠️ No matches for: {', '.join(unmatched)}", "yellow")) print(colorText(f"⚠️ No matches for: {', '.join(unmatched)}", "yellow"))
def enrich_agents(agents: List["Agent"], policies: List["Policy"]): def enrich_agents(agents: List["Agent"], policies: List["Policy"]):
@@ -248,7 +248,7 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
device_names = collect_device_names() device_names = collect_device_names()
if not device_names: if not device_names:
logger.debug("No device names entered") logger.debug("No device names entered")
print(colorText("⚠️ No device names entered.", "red")) print(colorText("⚠️ No device names entered.", "red"))
return [] return []
use_exact = choose_match_type() use_exact = choose_match_type()
@@ -261,11 +261,11 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
show_unmatched(device_names, matched_agents, use_exact) show_unmatched(device_names, matched_agents, use_exact)
if not matched_agents: if not matched_agents:
logger.debug("❌ No matching devices found.") logger.debug("❌ No matching devices found.")
print(colorText("❌ No matching devices found.", "red")) print(colorText("❌ No matching devices found.", "red"))
return [] return []
print(colorText(f"✓ Found {len(matched_agents)} matching device(s).", "green")) print(colorText(f"✓ Found {len(matched_agents)} matching device(s).", "green"))
logger.info("Matched agent hostnames:") logger.info("Matched agent hostnames:")
rows = (len(matched_agents) + 2) // 3 # 3 columns rows = (len(matched_agents) + 2) // 3 # 3 columns
for row in range(rows): for row in range(rows):
@@ -283,8 +283,8 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
) )
if not matched_agents: if not matched_agents:
logger.debug("❌ No matching devices remain after refinement.") logger.debug("❌ No matching devices remain after refinement.")
print(colorText("❌ No matching devices remain after refinement.", "red")) print(colorText("❌ No matching devices remain after refinement.", "red"))
return [] return []
enrich_agents(matched_agents, policies) enrich_agents(matched_agents, policies)
@@ -302,7 +302,7 @@ def moveAgentToRelatedPolicy(
Args: Args:
api: AirlockAPIWrapper instance. api: AirlockAPIWrapper instance.
agent: Agent object. agent: Agent object.
policy_relationship_map: Dict mapping enforcement â–€ –€™ audit. policy_relationship_map: Dict mapping enforcement â–€ –€™ audit.
mode: 'audit' to move to audit, 'enforcement' to move to enforcement. mode: 'audit' to move to audit, 'enforcement' to move to enforcement.
""" """
policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}") policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")