Merge remote-tracking branch 'origin/RustImplementation' into RustImplementation
Build Library / Build Library (push) Successful in 5m2s

This commit is contained in:
brotoskyj
2025-12-15 14:12:30 -05:00
3 changed files with 1305 additions and 310 deletions
File diff suppressed because it is too large Load Diff
+24 -24
View File
@@ -88,9 +88,9 @@ def sortHashes(
):
working_dir = load_env("WORKING_DIR")
history_days = Selector.select_value(
prompt="Enter how many days of history to pull (1150): ",
prompt="Enter how many days of history to pull (1-365): ",
value_type=int,
valid_range=(1, 150),
valid_range=(1, 365),
)
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):
working_dir = load_env("WORKING_DIR")
section_header("Prepare to Enforce Policy ")
section_header("Prepare to Enforce Policy")
print(
colorText(
"\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:
print(colorText(" [] No policies have been chosen", "red"))
print(colorText(" [❌] No policies have been chosen", "red"))
else:
print(colorText("The following policies have been chosen:", "green"))
for policy in selected_policies:
print(colorText(f" [] {policy.name}", "green"))
print(colorText(f" [✅] {policy.name}", "green"))
# Step 2: Destination Policy and Allowlist
print(
@@ -683,22 +683,22 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
if destination_policy:
print(
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",
)
)
else:
print(colorText(" [] No destination policy has been chosen", "red"))
print(colorText(" [❌] No destination policy has been chosen", "red"))
if destination_allowlist:
print(
colorText(
f" [] {destination_allowlist[0].name} has been selected as allowlist",
f" [✅] {destination_allowlist[0].name} has been selected as allowlist",
"green",
)
)
else:
print(colorText(" [] No allowlist has been chosen", "red"))
print(colorText(" [❌] No allowlist has been chosen", "red"))
# Step 3: Data Preparation
print(
@@ -713,9 +713,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
" [] Data has been fetched"
" [✅] Data has been fetched"
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",
)
@@ -723,7 +723,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else:
print(
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(
colorText(
(
" [] Reviewed hashes have been loaded"
" [✅] Reviewed hashes have been loaded"
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",
)
@@ -766,9 +766,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
" [] Path review list created"
" [✅] Path review list created"
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",
)
@@ -776,7 +776,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else:
print(
colorText(
" [] No policies selected, cannot check reviewed hashes or path list",
" [❌] No policies selected, cannot check reviewed hashes or path list",
"red",
)
)
@@ -812,9 +812,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
" [] Reviewed path list detected"
" [✅] Reviewed path list detected"
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",
)
@@ -825,9 +825,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
" [] Preflight Path Exclusion List has been generated"
" [✅] Preflight Path Exclusion List has been generated"
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",
)
@@ -835,7 +835,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else:
print(
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"))
# Utility Options
print(colorText("F. Open Working Directory", "cyan"))
print(colorText("B. Back", "cyan"))
print(colorText("F. Open Working Directory", "cyan"))
print(colorText("B. Back", "cyan"))
+15 -15
View File
@@ -38,9 +38,9 @@ logger = logging.getLogger(__name__)
def devicehistory(api: AirlockAPIWrapper, outputjson: bool):
agents = selectAgents(api)
history_days = Selector.select_value(
prompt="Enter how many days of history to pull (1150): ",
prompt="Enter how many days of history to pull (1–365): ",
value_type=int,
valid_range=(1, 150),
valid_range=(1, 365),
)
if not agents or not history_days:
@@ -60,7 +60,7 @@ def devicehistory(api: AirlockAPIWrapper, outputjson: bool):
except Exception as e:
print(
colorText(
f" Error retrieving history for {agent.hostname}: {e}", "red"
f"❌ Error retrieving history for {agent.hostname}: {e}", "red"
)
)
continue
@@ -139,7 +139,7 @@ def findAgents(api, return_dataframe):
print(
colorText(
f"\n Matched devices exported to: {working_dir}\\{filename}",
f"\n✓ Matched devices exported to: {working_dir}\\{filename}",
"green",
)
)
@@ -148,7 +148,7 @@ def findAgents(api, return_dataframe):
def collect_device_names() -> List[str]:
print(colorText("🖥Â Device Search", "cyan"))
print(colorText("🖥��Â Device Search", "cyan"))
print(
colorText(
"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:
print(
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",
)
)
@@ -235,8 +235,8 @@ def show_unmatched(
]
if unmatched:
logger.debug(f"⚠️ No matches for: {', '.join(unmatched)}")
print(colorText(f"⚠️ No matches for: {', '.join(unmatched)}", "yellow"))
logger.debug(f"⚠️ No matches for: {', '.join(unmatched)}")
print(colorText(f"⚠️ No matches for: {', '.join(unmatched)}", "yellow"))
def enrich_agents(agents: List["Agent"], policies: List["Policy"]):
@@ -248,7 +248,7 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
device_names = collect_device_names()
if not device_names:
logger.debug("No device names entered")
print(colorText("⚠️ No device names entered.", "red"))
print(colorText("⚠️ No device names entered.", "red"))
return []
use_exact = choose_match_type()
@@ -261,11 +261,11 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
show_unmatched(device_names, matched_agents, use_exact)
if not matched_agents:
logger.debug(" No matching devices found.")
print(colorText(" No matching devices found.", "red"))
logger.debug("❌ No matching devices found.")
print(colorText("❌ No matching devices found.", "red"))
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:")
rows = (len(matched_agents) + 2) // 3 # 3 columns
for row in range(rows):
@@ -283,8 +283,8 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
)
if not matched_agents:
logger.debug(" No matching devices remain after refinement.")
print(colorText(" No matching devices remain after refinement.", "red"))
logger.debug("❌ No matching devices remain after refinement.")
print(colorText("❌ No matching devices remain after refinement.", "red"))
return []
enrich_agents(matched_agents, policies)
@@ -302,7 +302,7 @@ def moveAgentToRelatedPolicy(
Args:
api: AirlockAPIWrapper instance.
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.
"""
policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")