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

Updates to Allowlist Selection

Browse Source
This commit is contained in:
Zarithas
2025-11-17 16:13:41 -05:00
parent fbb5cc396b
commit eb1d710d07
3 changed files with 75 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Loxide.py
+3 -14
View File
@@ -23,14 +23,13 @@
import logging
import os
import tempfile
import dotenv
import urllib3
from services.API import AirlockAPIWrapper
from services.security import getAPI
from TUI.TUI import run_Loxide
from utils.configmanager import get_system_value
from utils.setup import get_base_directory, setup
from utils.utils import irtang
@@ -38,24 +37,14 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def main():
if "NUITKA_ONEFILE_PARENT" in os.environ:
splash_filename = os.path.join(
tempfile.gettempdir(),
f"onefile_{int(os.environ['NUITKA_ONEFILE_PARENT'])}_splash_feedback.tmp",
)
if os.path.exists(splash_filename):
os.unlink(splash_filename)
irtang()
# Determine working directory, setup directory, configure logging, sent env, get API and URL if not already stored
setup()
base_dir = get_base_directory()
logger = logging.getLogger(__name__)
dotenv.load_dotenv(dotenv_path=base_dir / ".env")
try:
url = os.getenv("URL")
url = get_system_value("URL")
username = os.getenv("USERNAME")
if not url:
@@ -75,7 +64,7 @@ def main():
raise ValueError("API key for Loxide is missing.")
api = AirlockAPIWrapper(
base_url=str(os.getenv("URL")),
base_url=str(url),
api_key=api_key,
)
run_Loxide(api)
TUI/allowlistselectionscreen.py
+59 -15
View File
@@ -32,7 +32,7 @@ class AllowlistSelectionWidget(Static):
layout: vertical;
}
#allowlist_main {
height: 100%;
height: 1fr;
width: 100%;
}
#left_panel {
@@ -61,7 +61,8 @@ class AllowlistSelectionWidget(Static):
margin: 1 0;
}
#action_buttons {
height: 10%;
height: auto;
min-height: 3;
padding: 1;
content-align: center middle;
}
@@ -149,7 +150,7 @@ class AllowlistSelectionWidget(Static):
# Action buttons at bottom
with Horizontal(id="action_buttons"):
self.back_btn = Button("← Back", id="back_btn")
self.back_btn = Button("⬅ Back", id="back_btn")
self.add_btn = Button("➕ Add to Allowlist", id="add_to_allowlist_btn")
self.back_btn.styles.width = "50%"
@@ -176,7 +177,8 @@ class AllowlistSelectionWidget(Static):
# First, try to get the host's policy if hostname is provided
host_policy_allowlists = []
host_policy_ids = set()
policy_name = None
policy_name = "Unknown Policy" # Default value
group_id = None
if self.hostname:
try:
@@ -185,8 +187,23 @@ class AllowlistSelectionWidget(Static):
if not agents_df.empty:
# Get the policy group ID for this host
group_id = agents_df.iloc[0].get("groupid")
policy_name = agents_df.iloc[0].get(
"groupname", "Unknown Policy"
# Look up the policy name from app's cached policies
if (
group_id
and hasattr(self.app, "policies")
and self.app.policies
):
for policy in self.app.policies:
if policy.groupid == group_id:
policy_name = policy.name
logger.info(
f"Found policy name: '{policy_name}' for group_id: {group_id}"
)
break
logger.info(
f"Found host '{self.hostname}' in policy '{policy_name}' (group_id: {group_id})"
)
if group_id:
@@ -208,6 +225,33 @@ class AllowlistSelectionWidget(Static):
except Exception as e:
logger.warning(f"Could not get host's policy allowlists: {e}")
# If we still don't have a policy name, try to get it from the first allowlist or use a default
if not policy_name:
# Get all policies and try to find which one has allowlists
try:
all_policies_df = self.api.policy_find_all()
if not all_policies_df.empty:
# If we have a group_id from somewhere, use it
if group_id:
policy_row = all_policies_df[
all_policies_df["groupid"] == group_id
]
if not policy_row.empty:
policy_name = policy_row.iloc[0].get(
"groupname", "Unknown Policy"
)
else:
# Use the first policy as fallback
policy_name = all_policies_df.iloc[0].get(
"groupname", "Default Policy"
)
logger.info(f"Using first available policy: {policy_name}")
else:
policy_name = "Unknown Policy"
except Exception as e:
logger.warning(f"Could not fetch policies: {e}")
policy_name = "Unknown Policy"
# Get all allowlists
all_allowlists_df = self.api.allowlist_find_all()
@@ -243,7 +287,7 @@ class AllowlistSelectionWidget(Static):
# Add policy-associated allowlists if any
if host_policy_allowlists:
# Add section header
header_text = f"━━━ Policy: {policy_name or 'Host Policy'} ━━━"
header_text = f"=== Policy: {policy_name or 'Host Policy'} ==="
self.allowlist_table.add_row(header_text, "", "", key="header_policy")
current_row += 1
@@ -270,7 +314,7 @@ class AllowlistSelectionWidget(Static):
current_row += 1
self.allowlist_table.add_row(
"━━━ Other Available Allowlists ━━━", "", "", key="header_other"
"=== Other Available Allowlists ===", "", "", key="header_other"
)
current_row += 1
@@ -336,9 +380,9 @@ class AllowlistSelectionWidget(Static):
if found_col:
self.hash_column = found_col
preview_lines.append(f"✓ Found hash column: **{found_col}**\n")
preview_lines.append(f"✓ Found hash column: **{found_col}**\n")
else:
preview_lines.append("⚠️ **No hash column found**\n")
preview_lines.append("⚠️ **No hash column found**\n")
preview_lines.append("Available columns:\n")
for col in self.selected_data.columns:
if col != "_row_id":
@@ -420,7 +464,7 @@ class AllowlistSelectionWidget(Static):
self.selected_allowlist = self.allowlists[actual_allowlist_index]
self.add_btn.disabled = False
self.add_btn.label = (
f"➕ Add to '{self.selected_allowlist.get('name', 'Unknown')}'"
f"➕ Add to '{self.selected_allowlist.get('name', 'Unknown')}'"
)
# Update preview with selection
@@ -523,7 +567,7 @@ class AllowlistSelectionWidget(Static):
# Success notification
self.app.notify(
f"✅ Successfully added {len(self.hashes_to_add)} hashes to '{allowlist_name}'",
f"✅ Successfully added {len(self.hashes_to_add)} hashes to '{allowlist_name}'",
title="Success",
severity="information",
timeout=5,
@@ -531,7 +575,7 @@ class AllowlistSelectionWidget(Static):
# Update preview to show success
self.preview_area.text = (
f"## ✅ SUCCESS\n\n"
f"## ✅ SUCCESS\n\n"
f"Added **{len(self.hashes_to_add)} hashes** to allowlist:\n"
f"**{allowlist_name}** (ID: {app_id})\n\n"
f"### Operation Details:\n"
@@ -551,7 +595,7 @@ class AllowlistSelectionWidget(Static):
except Exception as exc:
logger.exception(f"Failed to add hashes to allowlist: {exc}")
self.app.notify(
f"❌ Failed to add hashes: {str(exc)}",
f"❌ Failed to add hashes: {str(exc)}",
title="Error",
severity="error",
timeout=10,
@@ -559,7 +603,7 @@ class AllowlistSelectionWidget(Static):
# Re-enable button
self.add_btn.disabled = False
self.add_btn.label = "➕ Retry Add to Allowlist"
self.add_btn.label = "➕ Retry Add to Allowlist"
class AllowlistSelectionScreen(Screen):
TUI/otpactivityscreen.py
+13 -10
View File
@@ -56,12 +56,13 @@ class OTPActivitiesWidget(Static):
layout: vertical;
}
#activity_preview_container {
height: 75%;
height: 1fr;
border: none;
padding: 1 1;
}
#activity_buttons {
height: 25%;
height: auto;
min-height: 3;
padding: 1 1;
content-align: center middle;
}
@@ -481,16 +482,18 @@ class ActivityDetailWidget(Static):
layout: vertical;
}
#detail_table_container {
height: 75%;
height: 1fr;
padding: 1 1;
}
#selection_buttons {
height: 10%;
height: auto;
min-height: 3;
padding: 1 1;
content-align: center middle;
}
#detail_buttons {
height: 15%;
height: auto;
min-height: 3;
padding: 1 1;
content-align: center middle;
}
@@ -649,7 +652,7 @@ class ActivityDetailWidget(Static):
logger.exception("Failed to sort by column %s: %s", column_key, exc)
return
# ✅ Only refresh rows, not columns
# ✅ Only refresh rows, not columns
await self._build_table(rebuild=False)
async def on_button_pressed(self, event) -> None:
@@ -728,13 +731,13 @@ class ActivityDetailWidget(Static):
if self.activities_df is None or self.activities_df.empty:
logger.info("No activities to export.")
await self.mount(
Static("❌ No activities to export.", classes="notification")
Static("❌ No activities to export.", classes="notification")
)
return
if not self.selected_row_ids:
logger.info("No rows selected for export.")
await self.mount(
Static("❌ No rows selected for export.", classes="notification")
Static("❌ No rows selected for export.", classes="notification")
)
return
try:
@@ -755,12 +758,12 @@ class ActivityDetailWidget(Static):
logger.exception("Failed to export detail activities: %s", exc)
await self.mount(
Static(
"❌ Failed to export activities; check logs.",
"❌ Failed to export activities; check logs.",
classes="notification",
)
)
# ✅ Helper methods
# ✅ Helper methods
def get_selected_data(self) -> pd.DataFrame:
"""Return a DataFrame of the selected rows."""
if not self.selected_row_ids: