Compare commits

...

34 Commits

Author SHA1 Message Date
brotoskyj 1aae27b0f4 Changed File so Build Activates
Build Library / Build Library (push) Failing after 1m22s
2026-01-05 10:12:43 -05:00
brotoskyj 663dbc3cc2 Changed YAML Build File 2026-01-05 10:11:50 -05:00
brotoskyj d2d181de4a Removed old crate imports 2026-01-05 10:08:41 -05:00
brotoskyj 31d45ca1db Styling - Progress Bar
Changed glyphs in progress bar for better printing to console
closes #51
2025-12-22 12:42:59 -05:00
brotoskyj a086956b48 Features - Optional Policy Name in Execution Histories
closes #48
2025-12-22 12:42:59 -05:00
brotoskyj 76cfe62f08 Styling - Progress Bar
Changed progress bar indicators and added policy name to the progress bar
closes #50
2025-12-22 12:42:59 -05:00
Zarithas 5b53612644 Readme update 2025-12-22 12:11:34 -05:00
Zarithas 2ed1d2e4cd Update docs/01_Software_Requirements_Specification.md 2025-12-22 11:14:29 -05:00
Zarithas 040a513bfb Merge branch 'feature-version-checker' of https://git.racooncity.org/brotoskyj/Airlocktools into feature-version-checker 2025-12-22 10:51:34 -05:00
Zarithas 423e9e8208 feat(release): Loxide 1.0 RC
- Added comprehensive documentation:
  - System Design Requirements (SDR)
  - System Design Specification (SDS)
  - API Reference
  - User Stories & Use Cases
- Fixed minor UI issues related to double encoding
2025-12-22 10:45:36 -05:00
Zarithas 7f4edcba46 Update utils/versionchecker.py
Corrected version number change in place for testing.
2025-12-19 16:39:53 -05:00
Zarithas e1e0cb7ac7 feat: add version updater and statistics enhancements (fixes #29)
- Implemented version checking system with update notifications
- Integrated Git for fetching and downloading the latest version
- Added statistics updates
- Removed unused code across the project
- Condensed project structure
- Updated README
- Cleaned up UI
2025-12-19 16:31:36 -05:00
brotoskyj 09d2c125cd Testing None type for policy 2025-12-18 09:41:37 -05:00
Zarithas fbd5b8b4b8 Initial commit for statistics feature 2025-12-17 20:17:32 -05:00
Zarithas a7b659c951 policyprepworkflow: enhancements and fixes
- Added row copy functionality (Ctrl+C)
- Improved row selection visual contrast
- Fixed data state issues when navigating between stages

Server Log tab
- Added new Server Log tab to the main application
- Implemented live filtering with wildcard support
- Enabled auto-refresh capability

Multiagent selector
- Added file loading support for device lists
2025-12-17 12:38:31 -05:00
brotoskyj 53f0b548b0 Bug Fixes
Build Library / Build Library (push) Successful in 4m51s
Changed days to days.to_string() to fix type confusion
closes #47
2025-12-17 11:02:13 -05:00
Zarithas 66eb101c5d Fixed Bitbake 2025-12-16 17:07:35 -05:00
Zarithas 22101c1eba Merge branch 'Zar-Branch' of https://git.racooncity.org/brotoskyj/Airlocktools into Zar-Branch
git commit -m "feat: Multiple UI improvements and new server log functionality

- Add Server Log tab with DataTable display of server activity logs

- Fix keyboard navigation bug in agents tab

- Add execution history viewer for selected agents

- Improve policy tree widget functionality by adding single device operations

- Integrate logging notifications into TUI
  - Add TextualNotificationHandler to setup.py
  - Display ERROR/WARNING/CRITICAL logs as toast notifications
  - Remove terminal output to prevent interference with TUI
  - Logs still written to Loxide.log file

closes #45"
2025-12-16 16:17:25 -05:00
Zarithas fc17c869fc feat: Multiple UI improvements and new server log functionality
- Add Server Log tab with DataTable display of server activity logs

- Fix keyboard navigation bug in agents tab

- Add execution history viewer for selected agents

- Improve policy tree widget functionality by adding single device operations

- Integrate logging notifications into TUI
  - Add TextualNotificationHandler to setup.py
  - Display ERROR/WARNING/CRITICAL logs as toast notifications
  - Remove terminal output to prevent interference with TUI
  - Logs still written to Loxide.log file

closes #45
2025-12-16 16:16:49 -05:00
brotoskyj 1dbbcff5d5 Merge remote-tracking branch 'origin/RustImplementation' into RustImplementation
Build Library / Build Library (push) Successful in 5m18s
2025-12-15 17:04:32 -05:00
brotoskyj f080b0034f Changes:
Features
Moved the init_tracer() function to an implementation in TelemetryConfig for cleaner main file
closes #43

Bug Fixes
Telemetry is now opt-in again. Due to changes in opentelemetry, this required a major overhaul of the telemetryconfig function.
closes #44
2025-12-15 17:04:17 -05:00
Zarithas 57d0f12000 feat: Complete Policy Prep Workflow with UX upgrades, Liftoff API, and TUI merge
- Added intro screen with workflow overview, time estimate, and onboarding controls
- Improved visuals: cleaner checkboxes (/), better loading screen layout
- Enforced mandatory tab reviews for critical steps with warnings and blocked navigation
- Optimized logging: INFO for milestones, DEBUG for internals; cleaner production logs
- Implemented Liftoff API integration: paths, publishers, hashes with granular error handling
- Color-coded completion feedback ( success,  failure,  partial) and detailed summaries
- Consolidated architecture: merged TUI.py into Loxide.py (single entry point, no circular imports)
- Fixed race condition in table creation with concurrency locks
2025-12-15 17:01:58 -05:00
brotoskyj 0dbc744471 Merge remote-tracking branch 'origin/RustImplementation' into RustImplementation 2025-12-15 14:14:16 -05:00
brotoskyj 7a912bddab Bug Fixes
Security Vulnerabilities Patched
RUSTSEC-2025-0009 - Some AES functions may panic when overflow checking is enabled
RUSTSEC-2024-0336 - rustls::Connection::Common::complete_io could fall into an infinite loop based on network input
closes #47
2025-12-15 14:13:57 -05:00
brotoskyj 24211c318b Merge remote-tracking branch 'origin/RustImplementation' into RustImplementation
Build Library / Build Library (push) Successful in 5m2s
2025-12-15 14:12:30 -05:00
brotoskyj 630e0a3cdf Bug Fixes
Security Vulnerabilities Patched
RUSTSEC-2025-0009 - Some AES functions may panic when overflow checking is enabled
RUSTSEC-2024-0336 - rustls::Connection::Common::complete_io could fall into an infinite loop based on network input
2025-12-15 14:12:03 -05:00
Zarithas 797d0f4462 fix(policy-prep): implement table editors and workflow improvements
- 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.
2025-12-11 16:55:12 -05:00
brotoskyj 59bb97ec4e Refactored LoxideLibs
Build Library / Build Library (push) Successful in 5m30s
1. Added compatibility check, LoxideLibs will now abort the entire program if OS is not linux or windows.
2. Changed the python data extraction compatibility layer, LoxideLibs was calling the extract data function twice, causing very slight overhead. I have now changed this so that the function returns a Struct that is now easily extractable via dot method notation.
2025-12-11 11:47:50 -05:00
brotoskyj 0ac3b54d89 Refactored Progress Bar
Build Library / Build Library (push) Successful in 5m54s
Refactored Progress Bar to remove multiprogress bar and only draw one instance. #36 is still open and not fixed with this push, but I believe this is the way to fix the issue.
Also implemented an Arc Mutex on the progress bar so it can be controlled via different threads.
2025-12-05 17:19:40 -05:00
Zarithas 154a7efcc8 Bug fix for Revoke OTP resolved, no longer crashes when select all is chosen when there are no active sessions, Swapped Revoke OTP and Quiet Hosts locations on menus 2025-12-05 15:30:24 -05:00
Zarithas ab5f00d8e7 Merge branch 'RustImplementation' of https://git.racooncity.org/brotoskyj/Airlocktools into RustImplementation 2025-12-05 15:06:06 -05:00
Zarithas 3ab803c12e Quiet Agent UI improvements 2025-12-05 15:05:46 -05:00
Zarithas 98cb23e5ea Bugfix for Issue 39.
Fixes:
brotoskyj/AirlockTools#39
2025-12-05 14:48:23 -05:00
brotoskyj 0aabbfd36e Style Change
Build Library / Build Library (push) Successful in 6m25s
Cleaned up services.rs file and removed whitespaces
Specified Data Types were needed instead of allowing the compiler to select types
2025-12-05 11:12:29 -05:00
47 changed files with 10562 additions and 5381 deletions
+2 -2
View File
@@ -2,8 +2,8 @@ name: Build Library
run-name: ${{ gitea.actor }} run-name: ${{ gitea.actor }}
on: on:
push: push:
branches: branches-ignore:
- RustImplementation - master
paths: paths:
- airlock_libs/** - airlock_libs/**
+3
View File
@@ -14,3 +14,6 @@ securitytest.py
system_config.json system_config.json
Development/ Development/
AirlockTools_client*/ AirlockTools_client*/
*.build/
*.dist/
*.onefile-build/
+1226 -6
View File
File diff suppressed because it is too large Load Diff
+639
View File
@@ -0,0 +1,639 @@
# 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 <https://www.gnu.org/licenses/>.
from datetime import datetime, timedelta
import logging
import os
from typing import List
import pandas as pd
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical
from textual.screen import Screen
from textual.widgets import (
Button,
DataTable,
Footer,
Header,
Label,
Select,
Static,
)
from models.agent import Agent
from models.execution import ExecutionHistoryRecord
from utils.configmanager import load_env
logger = logging.getLogger(__name__)
class ExecutionHistoryScreen(Screen):
"""
A screen for viewing and exporting execution history for selected agents.
This screen allows users to:
1. Select a start date and end date using dropdown selects
2. Fetch execution history for all selected agents
3. View the results in a DataTable
4. Export the results to CSV using a keybinding
Attributes:
agents (List[Agent]): List of agents to fetch execution history for
execution_data (pd.DataFrame): Combined execution history data
working_dir (str): Directory for CSV exports
"""
DEFAULT_CSS = """
ExecutionHistoryScreen {
align: center top;
}
#main_container {
width: 95%;
height: 1fr;
border: solid $primary;
padding: 1;
}
#title {
text-style: bold;
color: $text;
text-align: center;
margin-bottom: 1;
}
#date_container {
height: auto;
margin-bottom: 1;
}
#start_date_row, #end_date_row {
height: auto;
align-horizontal: left;
margin-bottom: 1;
}
.date_label {
width: 8;
margin-right: 1;
}
.date_selector {
width: 18;
margin: 0 1;
}
#quick_buttons_row {
height: auto;
align-horizontal: center;
margin-bottom: 1;
}
.quick_select_btn {
margin: 0 1;
}
#button_row {
height: auto;
align-horizontal: center;
margin-top: 1;
margin-bottom: 1;
}
Button {
margin: 0 1;
}
#status_label {
text-align: center;
color: $accent;
margin-bottom: 1;
}
#results_container {
height: 1fr;
display: none;
}
#results_button_row {
height: auto;
align-horizontal: center;
margin-bottom: 1;
}
#history_table {
height: 1fr;
border: solid $primary;
}
DataTable > .datatable--header {
text-style: bold;
background: $primary 20%;
}
"""
BINDINGS = [
Binding("escape", "close_screen", "Close"),
Binding("e", "export_csv", "Export CSV"),
Binding("q", "close_screen", "Quit"),
]
def __init__(self, agents: List[Agent]):
"""
Initialize the ExecutionHistoryScreen.
Args:
agents (List[Agent]): List of agents to fetch execution history for
"""
super().__init__()
self.agents = agents
self.execution_data = pd.DataFrame()
self.working_dir = load_env("WORKING_DIR") or os.getcwd()
# Generate dropdown options
today = datetime.now().date()
# Month options - format is (display_text, value)
self.month_options = [
("January", "01"),
("February", "02"),
("March", "03"),
("April", "04"),
("May", "05"),
("June", "06"),
("July", "07"),
("August", "08"),
("September", "09"),
("October", "10"),
("November", "11"),
("December", "12"),
]
# Day options (1-31) - format is (display_text, value)
self.day_options = [(f"{i}", f"{i:02d}") for i in range(1, 32)]
# Year options (current year back 5 years) - format is (display_text, value)
current_year = today.year
self.year_options = [
(str(year), str(year)) for year in range(current_year, current_year - 6, -1)
]
# Default dates: last 30 days
start_date = today - timedelta(days=30)
self.start_month = f"{start_date.month:02d}"
self.start_day = f"{start_date.day:02d}"
self.start_year = str(start_date.year)
self.end_month = f"{today.month:02d}"
self.end_day = f"{today.day:02d}"
self.end_year = str(today.year)
def compose(self) -> ComposeResult:
"""Build the UI layout."""
yield Header(show_clock=True, icon="📊")
with Vertical(id="main_container"):
title_text = f"Execution History - {len(self.agents)} Agent(s)"
yield Static(title_text, id="title")
# Date selection area
with Vertical(id="date_container"):
yield Label("Select Date Range:")
# Start date row
with Horizontal(id="start_date_row"):
yield Label("From:", classes="date_label")
yield Select(
options=self.month_options,
value=self.start_month,
id="start_month_select",
classes="date_selector",
)
yield Select(
options=self.day_options,
value=self.start_day,
id="start_day_select",
classes="date_selector",
)
yield Select(
options=self.year_options,
value=self.start_year,
id="start_year_select",
classes="date_selector",
)
# End date row
with Horizontal(id="end_date_row"):
yield Label("To:", classes="date_label")
yield Select(
options=self.month_options,
value=self.end_month,
id="end_month_select",
classes="date_selector",
)
yield Select(
options=self.day_options,
value=self.end_day,
id="end_day_select",
classes="date_selector",
)
yield Select(
options=self.year_options,
value=self.end_year,
id="end_year_select",
classes="date_selector",
)
# Quick select buttons
with Horizontal(id="quick_buttons_row"):
yield Button(
"1 Day",
id="quick_1day",
classes="quick_select_btn",
variant="default",
)
yield Button(
"1 Week",
id="quick_1week",
classes="quick_select_btn",
variant="default",
)
yield Button(
"30 Days",
id="quick_30days",
classes="quick_select_btn",
variant="default",
)
# Buttons
with Horizontal(id="button_row"):
yield Button("Fetch History", id="fetch_btn", variant="primary")
yield Button("Close", id="close_btn", variant="error")
# Status
yield Static(
"Select date range and click 'Fetch History'", id="status_label"
)
# Results container (hidden initially, shown after fetch)
with Vertical(id="results_container"):
with Horizontal(id="results_button_row"):
yield Button("Export CSV", id="export_btn", variant="success")
yield Button("Back", id="back_btn", variant="default")
yield DataTable(id="history_table")
yield Footer()
def on_mount(self) -> None:
"""Initialize the table when screen is mounted."""
table = self.query_one("#history_table", DataTable)
table.cursor_type = "row"
table.zebra_stripes = True
# Initially empty - will populate after fetch
logger.info(f"ExecutionHistoryScreen mounted with {len(self.agents)} agents")
def on_select_changed(self, event: Select.Changed) -> None:
"""Handle date selection changes."""
select_id = event.select.id
if select_id == "start_month_select":
self.start_month = event.value
logger.debug(f"Start month changed to: {self.start_month}")
elif select_id == "start_day_select":
self.start_day = event.value
logger.debug(f"Start day changed to: {self.start_day}")
elif select_id == "start_year_select":
self.start_year = event.value
logger.debug(f"Start year changed to: {self.start_year}")
elif select_id == "end_month_select":
self.end_month = event.value
logger.debug(f"End month changed to: {self.end_month}")
elif select_id == "end_day_select":
self.end_day = event.value
logger.debug(f"End day changed to: {self.end_day}")
elif select_id == "end_year_select":
self.end_year = event.value
logger.debug(f"End year changed to: {self.end_year}")
def _set_quick_date_range(self, days: int) -> None:
"""Set the date range based on quick select button."""
today = datetime.now().date()
start_date = today - timedelta(days=days)
# Update internal values
self.start_month = f"{start_date.month:02d}"
self.start_day = f"{start_date.day:02d}"
self.start_year = str(start_date.year)
self.end_month = f"{today.month:02d}"
self.end_day = f"{today.day:02d}"
self.end_year = str(today.year)
# Update the Select widgets
try:
self.query_one("#start_month_select", Select).value = self.start_month
self.query_one("#start_day_select", Select).value = self.start_day
self.query_one("#start_year_select", Select).value = self.start_year
self.query_one("#end_month_select", Select).value = self.end_month
self.query_one("#end_day_select", Select).value = self.end_day
self.query_one("#end_year_select", Select).value = self.end_year
self.app.notify(
f"Date range set to last {days} day(s)",
severity="information",
timeout=2,
)
logger.info(f"Quick select: Set date range to last {days} days")
except Exception as e:
logger.error(f"Failed to update date selects: {e}")
def _show_date_selection(self) -> None:
"""Show the date selection view and hide results."""
try:
self.query_one("#date_container").styles.display = "block"
self.query_one("#button_row").styles.display = "block"
self.query_one("#status_label").styles.display = "block"
self.query_one("#results_container").styles.display = "none"
except Exception as e:
logger.error(f"Failed to show date selection: {e}")
def _show_results(self) -> None:
"""Hide date selection view and show results."""
try:
self.query_one("#date_container").styles.display = "none"
self.query_one("#button_row").styles.display = "none"
self.query_one("#status_label").styles.display = "none"
self.query_one("#results_container").styles.display = "block"
except Exception as e:
logger.error(f"Failed to show results: {e}")
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button clicks."""
if event.button.id == "fetch_btn":
self._fetch_execution_history()
elif event.button.id == "export_btn":
self._export_to_csv()
elif event.button.id == "close_btn":
self.app.pop_screen()
elif event.button.id == "back_btn":
self._show_date_selection()
elif event.button.id == "quick_1day":
self._set_quick_date_range(days=1)
elif event.button.id == "quick_1week":
self._set_quick_date_range(days=7)
elif event.button.id == "quick_30days":
self._set_quick_date_range(days=30)
def _fetch_execution_history(self) -> None:
"""Fetch execution history for all selected agents."""
status_label = self.query_one("#status_label", Static)
status_label.update("⏳ Fetching execution history...")
# Disable buttons during fetch
fetch_btn = self.query_one("#fetch_btn", Button)
export_btn = self.query_one("#export_btn", Button)
fetch_btn.disabled = True
export_btn.disabled = True
api = self.app.api
all_history = []
try:
# Construct dates from dropdowns
start_date_str = f"{self.start_year}-{self.start_month}-{self.start_day}"
end_date_str = f"{self.end_year}-{self.end_month}-{self.end_day}"
# Validate dates
try:
start_dt = datetime.strptime(start_date_str, "%Y-%m-%d")
end_dt = datetime.strptime(end_date_str, "%Y-%m-%d")
except ValueError as e:
status_label.update(f"❌ Invalid date: {str(e)}")
fetch_btn.disabled = False
export_btn.disabled = False
self.app.notify(f"Invalid date selected: {str(e)}", severity="error")
return
if start_dt > end_dt:
status_label.update("❌ Error: Start date must be before end date")
fetch_btn.disabled = False
export_btn.disabled = False
return
# Fetch history for each agent
for i, agent in enumerate(self.agents):
try:
status_label.update(
f"⏳ Fetching history for {agent.hostname} ({i+1}/{len(self.agents)})..."
)
# Call API - note the API expects 'dateto' first, then 'datefrom'
history = api.history_execution(
today=end_date_str,
date_selected=start_date_str,
agent_name=agent.hostname,
)
if history:
# Add agent hostname to each record for identification
for record in history:
record["agent_hostname"] = agent.hostname
all_history.extend(history)
logger.info(
f"Fetched {len(history)} records for {agent.hostname}"
)
else:
logger.info(f"No history found for {agent.hostname}")
except Exception as e:
logger.error(f"Failed to fetch history for {agent.hostname}: {e}")
self.app.notify(
f"Warning: Failed to fetch history for {agent.hostname}",
severity="warning",
)
# Convert to DataFrame
if all_history:
status_label.update(
"⏳ Enriching execution data with hash information..."
)
# Normalize field names (handle API typos)
for record in all_history:
if "policver" in record and "policyver" not in record:
record["policyver"] = record.pop("policver")
# Convert dict records to ExecutionHistoryRecord objects
execution_records = []
for record in all_history:
try:
execution_records.append(ExecutionHistoryRecord(**record))
except TypeError as e:
logger.warning(f"Failed to create ExecutionHistoryRecord: {e}")
# If it fails, just keep the dict
continue
# Enrich with hash data if we have ExecutionHistoryRecord objects
if execution_records:
try:
enriched_records = ExecutionHistoryRecord.enrich_with_hashes(
api, execution_records
)
logger.info(
f"Enriched {len(enriched_records)} records with hash data"
)
# Convert back to DataFrame
self.execution_data = pd.DataFrame(
[r.__dict__ for r in enriched_records]
)
# Flatten hash_obj if present
if (
not self.execution_data.empty
and "hash_obj" in self.execution_data.columns
):
hash_df = self.execution_data["hash_obj"].apply(
lambda h: (
h.to_dict() if h and hasattr(h, "to_dict") else {}
)
)
self.execution_data = pd.concat(
[
self.execution_data.drop(columns=["hash_obj"]),
hash_df,
],
axis=1,
)
except Exception as e:
logger.warning(f"Failed to enrich with hashes: {e}")
# Fall back to plain DataFrame
self.execution_data = pd.DataFrame(all_history)
else:
# If we couldn't create any ExecutionHistoryRecord objects, just use raw data
self.execution_data = pd.DataFrame(all_history)
self._populate_table()
self._show_results() # Switch to results view
self.app.notify(
f"Successfully loaded {len(self.execution_data)} records",
severity="information",
)
else:
status_label.update(
"ℹ️ No execution history found for selected agents/dates"
)
self.app.notify("No execution history found", severity="information")
self.execution_data = pd.DataFrame()
except Exception as e:
logger.error(f"Error fetching execution history: {e}")
status_label.update(f"❌ Error: {str(e)}")
self.app.notify(f"Failed to fetch history: {str(e)}", severity="error")
finally:
# Re-enable buttons
fetch_btn.disabled = False
export_btn.disabled = False
def _populate_table(self) -> None:
"""Populate the DataTable with execution history data."""
table = self.query_one("#history_table", DataTable)
table.clear(columns=True)
if self.execution_data.empty:
return
# Define preferred column order (your specified order)
preferred_order = [
"policyname",
"policyver",
"hostname",
"username",
"publisher",
"filename",
"pprocess",
"gprocess",
"sha256",
"commandline",
"agent_hostname", # Our custom field
]
# Get available columns in preferred order, then add any remaining columns
available_cols = []
for col in preferred_order:
if col in self.execution_data.columns:
available_cols.append(col)
# Add any remaining columns not in preferred order
for col in self.execution_data.columns:
if col not in available_cols:
available_cols.append(col)
# Add columns to table
for col in available_cols:
table.add_column(col, key=col)
# Add rows
for idx, row in self.execution_data.iterrows():
row_data = []
for col in available_cols:
value = row[col]
# Convert to string, handle None/NaN
if pd.isna(value):
row_data.append("")
else:
row_data.append(str(value))
table.add_row(*row_data, key=str(idx))
logger.info(f"Populated table with {len(self.execution_data)} rows")
def _export_to_csv(self) -> None:
"""Export the current execution data to CSV."""
if self.execution_data.empty:
self.app.notify("No data to export", severity="warning")
return
try:
# Create filename with timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"execution_history_{timestamp}.csv"
filepath = os.path.join(self.working_dir, filename)
# Export to CSV
self.execution_data.to_csv(filepath, index=False, encoding="utf-8-sig")
self.app.notify(
f"✅ Exported {len(self.execution_data)} records to: {filepath}",
severity="information",
timeout=5,
)
logger.info(f"Exported execution history to: {filepath}")
except Exception as e:
logger.error(f"Failed to export CSV: {e}")
self.app.notify(f"Failed to export CSV: {str(e)}", severity="error")
def action_export_csv(self) -> None:
"""Keybinding action to export CSV."""
self._export_to_csv()
def action_close_screen(self) -> None:
"""Close this screen and return to previous."""
self.app.pop_screen()
+30 -31
View File
@@ -58,9 +58,14 @@ class OTPRevokeWidget(Static):
} }
#button_container { #button_container {
height: auto; height: auto;
width: 100%;
padding: 1; padding: 1;
align: center middle; align: center middle;
} }
#button_container Button {
min-width: 16;
margin: 0 1;
}
#result_container { #result_container {
height: auto; height: auto;
max-height: 10; max-height: 10;
@@ -87,29 +92,10 @@ class OTPRevokeWidget(Static):
# Action buttons # Action buttons
with Horizontal(id="button_container"): with Horizontal(id="button_container"):
self.refresh_button = Button("🔄 Refresh", id="refresh_btn") yield Button("Refresh", id="refresh_btn")
self.refresh_button.styles.width = "15%" yield Button("Select All", id="select_all_btn")
self.refresh_button.styles.margin = (1, 1, 1, 1) yield Button("Clear Selection", id="select_none_btn")
yield self.refresh_button yield Button("Revoke Selected", id="revoke_btn", variant="error")
self.select_all_button = Button("☑️ Select All", id="select_all_btn")
self.select_all_button.styles.width = "15%"
self.select_all_button.styles.margin = (1, 1, 1, 1)
yield self.select_all_button
self.select_none_button = Button(
"❌ Clear Selection", id="select_none_btn"
)
self.select_none_button.styles.width = "20%"
self.select_none_button.styles.margin = (1, 1, 1, 1)
yield self.select_none_button
self.revoke_button = Button(
"🛑 Revoke Selected", id="revoke_btn", variant="error"
)
self.revoke_button.styles.width = "20%"
self.revoke_button.styles.margin = (1, 1, 1, 1)
yield self.revoke_button
# Results display # Results display
with Vertical(id="result_container"): with Vertical(id="result_container"):
@@ -122,7 +108,7 @@ class OTPRevokeWidget(Static):
# Configure sessions table # Configure sessions table
self.sessions_table.clear() self.sessions_table.clear()
self.sessions_table.add_columns( self.sessions_table.add_columns(
"", "OTP ID", "Hostname", "Status", "Purpose", "Granted" "", "OTP ID", "Hostname", "Status", "Purpose", "Granted"
) )
# Enable row selection with checkbox column # Enable row selection with checkbox column
@@ -213,7 +199,11 @@ class OTPRevokeWidget(Static):
elif btn.id == "select_all_btn": elif btn.id == "select_all_btn":
# Select all visible rows # Select all visible rows
if self._filtered_df is not None: if (
self._filtered_df is not None
and not self._filtered_df.empty
and "otpid" in self._filtered_df.columns
):
self._selected_otpids = set(str(x) for x in self._filtered_df["otpid"]) self._selected_otpids = set(str(x) for x in self._filtered_df["otpid"])
await self._refresh_table() await self._refresh_table()
@@ -235,7 +225,12 @@ class OTPRevokeWidget(Static):
# Get the row index from the cursor row # Get the row index from the cursor row
row_index = self.sessions_table.cursor_row row_index = self.sessions_table.cursor_row
if self._filtered_df is not None and row_index < len(self._filtered_df): if (
self._filtered_df is not None
and not self._filtered_df.empty
and "otpid" in self._filtered_df.columns
and row_index < len(self._filtered_df)
):
# Get the OTP ID for this row # Get the OTP ID for this row
otpid = str(self._filtered_df.iloc[row_index]["otpid"]) otpid = str(self._filtered_df.iloc[row_index]["otpid"])
@@ -257,12 +252,12 @@ class OTPRevokeWidget(Static):
async def _revoke_selected(self) -> None: async def _revoke_selected(self) -> None:
"""Revoke the selected OTP sessions.""" """Revoke the selected OTP sessions."""
if not self._selected_otpids: if not self._selected_otpids:
self.results_display.update("No sessions selected for revocation") self.results_display.update("No sessions selected for revocation")
return return
api = getattr(self.app, "api", None) api = getattr(self.app, "api", None)
if not api: if not api:
self.results_display.update("API not available") self.results_display.update("API not available")
return return
# Collect results # Collect results
@@ -303,13 +298,13 @@ class OTPRevokeWidget(Static):
else "No response" else "No response"
) )
results.append( results.append(
f"Failed to revoke OTP {otpid} for {hostname}: {error_msg}" f"Failed to revoke OTP {otpid} for {hostname}: {error_msg}"
) )
logger.error(f"Failed to revoke OTP {otpid}: {error_msg}") logger.error(f"Failed to revoke OTP {otpid}: {error_msg}")
except Exception as e: except Exception as e:
failure_count += 1 failure_count += 1
results.append(f"Error revoking OTP {otpid}: {str(e)}") results.append(f"Error revoking OTP {otpid}: {str(e)}")
logger.exception(f"Exception revoking OTP {otpid}: {e}") logger.exception(f"Exception revoking OTP {otpid}: {e}")
# Update results display # Update results display
@@ -368,7 +363,11 @@ class OTPRevokeScreen(Screen):
async def action_select_all(self) -> None: async def action_select_all(self) -> None:
"""Select all visible sessions.""" """Select all visible sessions."""
if self.widget._filtered_df is not None: if (
self.widget._filtered_df is not None
and not self.widget._filtered_df.empty
and "otpid" in self.widget._filtered_df.columns
):
self.widget._selected_otpids = set( self.widget._selected_otpids = set(
str(x) for x in self.widget._filtered_df["otpid"] str(x) for x in self.widget._filtered_df["otpid"]
) )
File diff suppressed because it is too large Load Diff
+185 -112
View File
@@ -34,13 +34,13 @@ from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical from textual.containers import Horizontal, Vertical
from textual.reactive import reactive from textual.reactive import reactive
from textual.screen import Screen from textual.screen import Screen
from textual.widgets import Button, DataTable, Footer, Header, Static from textual.widgets import Button, DataTable, Footer, Header, Input, Static
from models.policy import Policy from models.policy import Policy
from services.API import AirlockAPIWrapper from services.API import AirlockAPIWrapper
from services.policyhandler import getPolicyInfo
from TUI.Widgets.policyselector import PolicySelector from TUI.Widgets.policyselector import PolicySelector
from utils.configmanager import load_env from utils.configmanager import load_env
from utils.executionfetcher import getExecutions
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -51,16 +51,17 @@ class QuietAgentWorkflowScreen(Screen):
This screen provides a multi-step workflow: This screen provides a multi-step workflow:
1. Select initial policy to analyze 1. Select initial policy to analyze
2. View categorized agents (enforce ready vs. non-enforce ready) 2. Configure analysis parameters (history period and quiet time period)
3. Select target policies for each category 3. View categorized agents (enforce ready vs. non-enforce ready)
4. Execute agent migrations 4. Select target policies for each category
5. Execute agent migrations
Attributes: Attributes:
api (AirlockAPIWrapper): API wrapper for Airlock operations api (AirlockAPIWrapper): API wrapper for Airlock operations
policies (List[Policy]): List of all available policies policies (List[Policy]): List of all available policies
selected_policy (Optional[Policy]): The initially selected policy to analyze selected_policy (Optional[Policy]): The initially selected policy to analyze
history_days (int): Number of days of history to pull (default: 150) history_days (int): Number of days of history to pull (default: 150, range: 1-365)
quiet_days (int): Number of days without execution to be considered quiet (default: 45) quiet_days (int): Number of days without execution to be considered quiet (default: 45, range: 1-365)
agents_df (Optional[pd.DataFrame]): DataFrame of all agents with analysis results agents_df (Optional[pd.DataFrame]): DataFrame of all agents with analysis results
enforce_ready_df (Optional[pd.DataFrame]): DataFrame of agents ready for enforcement enforce_ready_df (Optional[pd.DataFrame]): DataFrame of agents ready for enforcement
non_enforce_ready_df (Optional[pd.DataFrame]): DataFrame of agents not ready for enforcement non_enforce_ready_df (Optional[pd.DataFrame]): DataFrame of agents not ready for enforcement
@@ -86,7 +87,7 @@ class QuietAgentWorkflowScreen(Screen):
self.api = api self.api = api
self.policies = policies self.policies = policies
self.selected_policy: Optional[Policy] = None self.selected_policy: Optional[Policy] = None
self.history_days = 150 # Fixed as per requirements self.history_days = 150 # Default value, user-selectable
self.quiet_days = 45 # Default value self.quiet_days = 45 # Default value
self.agents_df: Optional[pd.DataFrame] = None self.agents_df: Optional[pd.DataFrame] = None
self.enforce_ready_df: Optional[pd.DataFrame] = None self.enforce_ready_df: Optional[pd.DataFrame] = None
@@ -130,7 +131,7 @@ class QuietAgentWorkflowScreen(Screen):
stage_messages = { stage_messages = {
"select_policy": "Step 1: Select Policy to Analyze", "select_policy": "Step 1: Select Policy to Analyze",
"select_quiet_days": "Step 2: Select Quiet Time Period", "select_history_days": "Step 2: Configure Analysis Parameters",
"analyzing": "Analyzing agent activity...", "analyzing": "Analyzing agent activity...",
"view_results": "Step 3: Review Categorized Agents", "view_results": "Step 3: Review Categorized Agents",
"select_enforce_target": "Step 4: Select Target Policy for Enforce Ready Agents", "select_enforce_target": "Step 4: Select Target Policy for Enforce Ready Agents",
@@ -161,7 +162,7 @@ class QuietAgentWorkflowScreen(Screen):
# Initial policy selection for analysis # Initial policy selection for analysis
self.selected_policy = message.policy self.selected_policy = message.policy
logger.info(f"Selected policy for analysis: {self.selected_policy.name}") logger.info(f"Selected policy for analysis: {self.selected_policy.name}")
self._show_quiet_days_selection() self._show_history_days_selection()
elif self.workflow_stage == "select_enforce_target": elif self.workflow_stage == "select_enforce_target":
# Target policy selection for enforce ready agents # Target policy selection for enforce ready agents
self.enforce_ready_target_policy = message.policy self.enforce_ready_target_policy = message.policy
@@ -177,48 +178,167 @@ class QuietAgentWorkflowScreen(Screen):
) )
self._show_migration_confirmation() self._show_migration_confirmation()
def _show_quiet_days_selection(self) -> None: def _show_history_days_selection(self) -> None:
"""Show the quiet days selection screen.""" """Show the history days and quiet days selection screen."""
self.workflow_stage = "select_quiet_days" self.workflow_stage = "select_history_days"
content = self.query_one("#content_area", Vertical) content = self.query_one("#content_area", Vertical)
content.remove_children() content.remove_children()
# Create info text # Create info text
info_widget = Static( info_widget = Static(
f"Policy Selected: {self.selected_policy.name}\n\n" f"Policy Selected: {self.selected_policy.name}\n\n"
f"History Period: {self.history_days} days\n\n" "Configure Analysis Parameters:",
"Select quiet time period (days without untrusted execution):", id="analysis_params_info",
id="quiet_days_info",
) )
info_widget.styles.margin = (0, 0, 2, 0) info_widget.styles.margin = (0, 0, 2, 0)
content.mount(info_widget) content.mount(info_widget)
# Create button container and mount it first # Create input container
button_container = Vertical(id="quiet_days_buttons") input_container = Vertical(id="analysis_params_input_container")
button_container.styles.height = "auto" input_container.styles.height = "auto"
content.mount(button_container) content.mount(input_container)
# Now add buttons to the mounted container # History days label
for days in [15, 30, 45, 60]: history_label = Static("History Period (days of execution history to pull):")
btn = Button( history_label.styles.margin = (0, 0, 1, 0)
f"{days} days {'(Default)' if days == 45 else ''}", input_container.mount(history_label)
id=f"quiet_days_{days}",
classes="quiet_day_btn", # Add history days input field
history_input = Input(
placeholder="Enter days (1-365, default: 150)",
value="150",
id="history_days_input",
)
history_input.styles.width = "50"
history_input.styles.margin = (0, 0, 2, 0)
input_container.mount(history_input)
# Quiet days label
quiet_label = Static(
"Quiet Time Period (days without execution to be considered quiet):"
)
quiet_label.styles.margin = (0, 0, 1, 0)
input_container.mount(quiet_label)
# Add quiet days input field
quiet_input = Input(
placeholder="Enter days (1-365, default: 45)",
value="45",
id="quiet_days_input",
)
quiet_input.styles.width = "50"
quiet_input.styles.margin = (0, 0, 2, 0)
input_container.mount(quiet_input)
# Add submit button
submit_btn = Button(
"Continue",
id="analysis_params_submit",
variant="primary",
)
submit_btn.styles.width = "50"
submit_btn.styles.margin = (1, 0, 0, 0)
input_container.mount(submit_btn)
# Focus the first input field
history_input.focus()
def _validate_and_submit_history_days(self) -> None:
"""Validate and submit the history days and quiet days inputs."""
try:
history_input = self.query_one("#history_days_input", Input)
quiet_input = self.query_one("#quiet_days_input", Input)
history_value = history_input.value.strip()
quiet_value = quiet_input.value.strip()
# Validate history days
if not history_value:
self.app.notify(
"Please enter a history period value", severity="error", timeout=3
)
history_input.focus()
return
try:
history_days = int(history_value)
except ValueError:
self.app.notify(
"Please enter a valid number for history period",
severity="error",
timeout=3,
)
history_input.focus()
return
if history_days < 1 or history_days > 365:
self.app.notify(
"History period must be between 1 and 365 days",
severity="error",
timeout=3,
)
history_input.focus()
return
# Validate quiet days
if not quiet_value:
self.app.notify(
"Please enter a quiet time period value",
severity="error",
timeout=3,
)
quiet_input.focus()
return
try:
quiet_days = int(quiet_value)
except ValueError:
self.app.notify(
"Please enter a valid number for quiet time period",
severity="error",
timeout=3,
)
quiet_input.focus()
return
if quiet_days < 1 or quiet_days > 365:
self.app.notify(
"Quiet time period must be between 1 and 365 days",
severity="error",
timeout=3,
)
quiet_input.focus()
return
# Check that quiet days doesn't exceed history days
if quiet_days > history_days:
self.app.notify(
"Quiet time period cannot exceed history period",
severity="error",
timeout=3,
)
quiet_input.focus()
return
# All validation passed
self.history_days = history_days
self.quiet_days = quiet_days
logger.info(
f"Selected history days: {history_days}, quiet days: {quiet_days}"
) )
btn.styles.width = "100%" self._start_analysis()
btn.styles.margin = (0, 0, 1, 0)
button_container.mount(btn) except Exception as e:
logger.error(f"Error validating analysis parameters: {e}")
self.app.notify(f"Error: {str(e)}", severity="error", timeout=3)
def on_button_pressed(self, event: Button.Pressed) -> None: def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button press events.""" """Handle button press events."""
button_id = event.button.id button_id = event.button.id
# Quiet days selection buttons # Analysis parameters submit button
if button_id and button_id.startswith("quiet_days_"): if button_id == "analysis_params_submit":
days = int(button_id.split("_")[-1]) self._validate_and_submit_history_days()
self.quiet_days = days
logger.info(f"Selected quiet days: {days}")
self._start_analysis()
return return
# Navigation buttons # Navigation buttons
@@ -258,46 +378,44 @@ class QuietAgentWorkflowScreen(Screen):
self._show_policy_selection() self._show_policy_selection()
return return
def on_input_submitted(self, event: Input.Submitted) -> None:
"""Handle input submission (Enter key pressed)."""
if event.input.id in ["history_days_input", "quiet_days_input"]:
self._validate_and_submit_history_days()
def _start_analysis(self) -> None: def _start_analysis(self) -> None:
"""Start the agent activity analysis.""" """Start the agent activity analysis."""
self.workflow_stage = "analyzing" # Show notification that analysis is starting
content = self.query_one("#content_area", Vertical)
content.remove_children()
# Show analyzing message with detailed steps
analyzing_msg = Static(
f"Analyzing Agent Activity\n"
f"{'=' * 50}\n\n"
f"Policy: {self.selected_policy.name}\n"
f"History Period: {self.history_days} days\n"
f"Quiet Threshold: {self.quiet_days} days\n\n"
f"Progress:\n"
f"Step 1/4: Fetching agents from policy...\n"
f"Step 2/4: Pulling execution history (this may take a moment)...\n"
f"Step 3/4: Analyzing activity patterns...\n"
f"Step 4/4: Categorizing agents...\n\n"
f"Please wait - this operation cannot be cancelled.",
id="analyzing_message",
)
analyzing_msg.styles.margin = (2, 1)
content.mount(analyzing_msg)
# Show notification
self.app.notify( self.app.notify(
"Starting analysis - this may take several minutes for large policies", "Starting analysis - this may take several minutes for large policies",
severity="information", severity="information",
timeout=5, timeout=5,
) )
# Perform the analysis asynchronously # Clear the screen to provide a blank canvas for Rust progress output
self.call_later(self._perform_analysis) # (Rust output displays over the TUI, so we clear everything except header/footer)
try:
# Clear title
title_widget = self.query_one("#workflow_title", Static)
title_widget.update("")
def _perform_analysis(self) -> None: # Clear status
status_widget = self.query_one("#workflow_status", Static)
status_widget.update("")
# Clear content area
content = self.query_one("#content_area", Vertical)
content.remove_children()
except Exception as e:
logger.debug(f"Could not clear screen for analysis: {e}")
# Delay the analysis start to ensure UI refresh completes first
# This prevents Rust output from starting before the screen is cleared
self.set_timer(0.5, self._perform_analysis_worker)
def _perform_analysis_worker(self) -> None:
"""Perform the actual agent activity analysis.""" """Perform the actual agent activity analysis."""
try: try:
# Update status: Fetching agents
self._update_analysis_status("Step 1/4: Fetching agents from policy...")
# Get agents in the selected policy # Get agents in the selected policy
agents = self.api.agents_find_by_group(self.selected_policy.groupid) agents = self.api.agents_find_by_group(self.selected_policy.groupid)
@@ -310,32 +428,11 @@ class QuietAgentWorkflowScreen(Screen):
self._show_policy_selection() self._show_policy_selection()
return return
agent_count = len(agents)
self.app.notify(
f"Found {agent_count} agents - fetching execution history...",
severity="information",
timeout=3,
)
# Update status: Pulling execution history
self._update_analysis_status(
f"Step 2/4: Pulling execution history for {agent_count} agents...\n"
f"(This may take several minutes - progress shown in terminal)"
)
# Get execution history (this shows progress bars in terminal via airlock_libs) # Get execution history (this shows progress bars in terminal via airlock_libs)
policy_exec_history = getPolicyInfo( policy_exec_history = getExecutions(
self.api, self.selected_policy, [1, 2, 6, 7], self.history_days self.api, self.selected_policy, [1, 2, 6, 7], self.history_days
) )
# Update status: Analyzing patterns
self._update_analysis_status("Step 3/4: Analyzing activity patterns...")
self.app.notify(
"History retrieved - analyzing patterns...",
severity="information",
timeout=2,
)
if policy_exec_history.empty: if policy_exec_history.empty:
logger.info( logger.info(
"No execution history found for the selected policy and time range." "No execution history found for the selected policy and time range."
@@ -381,9 +478,6 @@ class QuietAgentWorkflowScreen(Screen):
lambda x: True if pd.isna(x) or x > self.quiet_days else False lambda x: True if pd.isna(x) or x > self.quiet_days else False
) )
# Update status: Categorizing
self._update_analysis_status("Step 4/4: Categorizing agents...")
# Sort agents # Sort agents
agents = agents.sort_values( agents = agents.sort_values(
by=["execution_count", "hostname"], ascending=[True, True] by=["execution_count", "hostname"], ascending=[True, True]
@@ -394,7 +488,7 @@ class QuietAgentWorkflowScreen(Screen):
# Categorize agents into DataFrames # Categorize agents into DataFrames
self.enforce_ready_df = agents[agents["enforce_ready"]].copy() self.enforce_ready_df = agents[agents["enforce_ready"]].copy()
self.non_enforce_ready_df = agents[not agents["enforce_ready"]].copy() self.non_enforce_ready_df = agents[~agents["enforce_ready"]].copy()
logger.info( logger.info(
f"Analysis complete: {len(self.enforce_ready_df)} enforce ready, " f"Analysis complete: {len(self.enforce_ready_df)} enforce ready, "
@@ -416,27 +510,6 @@ class QuietAgentWorkflowScreen(Screen):
self.app.notify(f"Analysis failed: {str(e)}", severity="error", timeout=5) self.app.notify(f"Analysis failed: {str(e)}", severity="error", timeout=5)
self._show_policy_selection() self._show_policy_selection()
def _update_analysis_status(self, status_text: str) -> None:
"""Update the analysis status message."""
try:
analyzing_msg = self.query_one("#analyzing_message", Static)
# Build updated message
updated_text = (
f"Analyzing Agent Activity\n"
f"{'=' * 50}\n\n"
f"Policy: {self.selected_policy.name}\n"
f"History Period: {self.history_days} days\n"
f"Quiet Threshold: {self.quiet_days} days\n\n"
f"Progress:\n"
f"{status_text}\n\n"
f"Please wait - this operation cannot be cancelled."
)
analyzing_msg.update(updated_text)
except Exception as e:
logger.debug(f"Could not update analysis status: {e}")
def _show_results(self) -> None: def _show_results(self) -> None:
"""Show the categorized results.""" """Show the categorized results."""
self.workflow_stage = "view_results" self.workflow_stage = "view_results"
@@ -821,7 +894,7 @@ class QuietAgentWorkflowScreen(Screen):
# Depending on stage, go back to previous stage or exit # Depending on stage, go back to previous stage or exit
if self.workflow_stage in ["select_policy", "view_results", "complete"]: if self.workflow_stage in ["select_policy", "view_results", "complete"]:
self.app.pop_screen() self.app.pop_screen()
elif self.workflow_stage == "select_quiet_days": elif self.workflow_stage == "select_history_days":
self._show_policy_selection() self._show_policy_selection()
elif self.workflow_stage == "select_enforce_target": elif self.workflow_stage == "select_enforce_target":
self._show_results() self._show_results()
-444
View File
@@ -1,444 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
import logging
import os
from typing import Optional
import dotenv
from textual.app import App, ComposeResult
from textual.containers import Vertical
from textual.message import Message
from textual.reactive import reactive
from textual.screen import Screen
from textual.widgets import (
Button,
DirectoryTree,
Footer,
Header,
Static,
Tab,
Tabs,
)
from models.agent import Agent
from models.policy import Policy
from services.API import AirlockAPIWrapper
from TUI.Screens.moveagentworkflowscreen import MoveAgentWorkflowScreen
from TUI.Screens.otpactivityscreen import OTPActivitiesScreen
from TUI.Screens.otprevokescreen import OTPRevokeScreen
from TUI.Screens.otpworkflowscreen import OTPWorkflowScreen
from TUI.Screens.policyprepworkflowscreen import PolicyPrepWorkflowScreen
from TUI.Screens.quietagentworkflowscreen import QuietAgentWorkflowScreen
from TUI.Themes.theme_amber_terminal import get_amber_terminal_theme
from TUI.Themes.theme_retro_terminal import get_retro_terminal_theme
from TUI.Themes.themeselector import ThemeSelector
from TUI.Widgets.agentmoveoperations import AgentMoveOperations
from TUI.Widgets.multiagentselector import MultiAgentSelector
from TUI.Widgets.policytreewidget import PolicyTreeWidget
from TUI.Widgets.resultsdisplay import ResultsDisplay
from utils.configmanager import get_user_value, load_env, save_user_config
from utils.setup import get_base_directory
from utils.utils import open_directory
dotenv.load_dotenv()
# ---------------------------------------------------------------------------
# GLOBAL STASH
# ---------------------------------------------------------------------------
_APP_RESTART_REASON = None
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# helper to persist TEXTUAL_THEME to *user* config and mirror to .env
# ---------------------------------------------------------------------------
def _persist_user_theme(theme_name: str) -> None:
"""
Store the chosen Textual theme in the user's config using the config manager.
No need to touch .env - config manager handles everything.
"""
base_dir = get_base_directory()
config_dir = base_dir / "config"
try:
save_user_config(config_dir, {"TEXTUAL_THEME": theme_name})
logger.debug("Updated user config with TEXTUAL_THEME=%s", theme_name)
except Exception as exc:
logger.error("Failed to save TEXTUAL_THEME: %s", exc)
# ---------------------------------------------------------------------------
# 1) SCREEN
# ---------------------------------------------------------------------------
class MainMenuScreen(Screen):
api: AirlockAPIWrapper
current_tab = reactive("")
BUTTON_DEFS = {
"agent_actions": [
(
"🖥️ - Find, Move, or Generate OTP for Agents",
"move_agent_workflow_button",
),
("🎫 - Review and appove OTP Activities", "otp_activities_button"),
("🔕 - Find and Move Quiet Hosts to Enforcement", "find_quiet_button"),
],
"policy": [
("⚖️ - Prepare Policy For Enforcement", "policy_prep_button"),
("🛑 - Revoke OTPs", "otp_revoke_button"),
],
}
def __init__(self) -> None:
super().__init__()
self.extras = get_user_value("EXTRAS", str, "NOTTODAY")
wd = load_env("WORKING_DIR") or os.getcwd()
if not os.path.isdir(wd):
wd = os.getcwd()
self.working_dir = wd
def _make_buttons_for(self, tab_id: str) -> Vertical:
defs = self.BUTTON_DEFS.get(tab_id, [])
buttons = []
for label, btn_id in defs:
btn = Button(label, id=btn_id)
btn.styles.width = "100%"
buttons.append(btn)
return Vertical(*buttons)
def compose(self) -> ComposeResult:
yield Header(show_clock=True, icon="")
tabs = [
Tab("Tree View", id="p_tree"),
Tab("Agents", id="agent_actions"),
Tab("Directory", id="dir"),
Tab("Settings", id="settings"),
]
if self.extras == "POLICYPREP":
tabs.insert(2, Tab("Policy Prep", id="policy"))
yield Tabs(*tabs, id="tabs")
yield Vertical(id="content")
yield Footer()
def on_mount(self) -> None:
self.switch_tab("agent_actions")
# focus helpers
def _get_content_buttons(self) -> list[Button]:
content = self.query_one("#content", Vertical)
return list(content.query(Button))
def _focus_first_button(self) -> None:
buttons = self._get_content_buttons()
if buttons:
buttons[0].focus()
def _focus_tabs(self) -> None:
tabs = self.query_one("#tabs", Tabs)
tabs.focus()
def _focus_nearby_button(self, direction: int) -> None:
buttons = self._get_content_buttons()
if not buttons:
return
try:
current = next(i for i, b in enumerate(buttons) if b.has_focus)
except StopIteration:
if direction > 0:
buttons[0].focus()
else:
buttons[-1].focus()
return
if direction < 0 and current == 0:
self._focus_tabs()
return
new_index = current + direction
if 0 <= new_index < len(buttons):
buttons[new_index].focus()
def switch_tab(self, tab_id: str) -> None:
self.current_tab = tab_id
content = self.query_one("#content", Vertical)
content.remove_children()
if tab_id in self.BUTTON_DEFS:
content.mount(self._make_buttons_for(tab_id))
self.call_later(self._focus_first_button)
elif tab_id == "dir":
content.mount(DirectoryTree(self.working_dir, id="dir_tree"))
elif tab_id == "p_tree":
content.mount(PolicyTreeWidget(self.app.policies, self.app.devices))
elif tab_id == "settings":
content.mount(ThemeSelector())
else:
content.mount(Static(f"Unknown tab: {tab_id}"))
def on_tabs_tab_activated(self, event: Tabs.TabActivated) -> None:
self.switch_tab(event.tab.id)
def on_multi_agent_selector_agents_selected(
self, message: MultiAgentSelector.AgentsSelected
) -> None:
"""Handle selected agents from AgentSelector."""
global _APP_RESTART_REASON
selected_agents = message.selected_agents
logger.info("Selected agents: %s", selected_agents)
# TODO: Implement actual handling of selected agents
_APP_RESTART_REASON = ("multi_agent_action", selected_agents)
self.app.exit()
def on_theme_selector_theme_selected(
self, message: ThemeSelector.ThemeSelected
) -> None:
"""Handle theme selection from ThemeSelector."""
global _APP_RESTART_REASON
_persist_user_theme(message.theme_name)
_APP_RESTART_REASON = ("restart",)
self.app.exit()
def on_agent_move_operations_operation_complete(
self, message: AgentMoveOperations.OperationComplete
) -> None:
"""Handle completion of agent move operation - show results."""
logger.info(
"Agent move operation completed: %s, %d successful, %d unsuccessful",
message.operation,
len(message.successful),
len(message.unsuccessful),
)
# Format results for display
successful_text = "\n".join(
[f"{agent.hostname}" for agent, _ in message.successful]
)
unsuccessful_text = "\n".join(
[f"{agent.hostname}: {error}" for agent, error in message.unsuccessful]
)
# Remove the operations widget
try:
ops_widget = self.query_one(AgentMoveOperations)
ops_widget.remove()
except Exception:
pass
# Show results
self.query_one("#content", Vertical).mount(
ResultsDisplay(message.operation, successful_text, unsuccessful_text)
)
def on_results_display_go_back(self, message: ResultsDisplay.GoBack) -> None:
"""Handle back button from results display."""
try:
results_widget = self.query_one(ResultsDisplay)
results_widget.remove()
except Exception:
pass
# Return to main menu
self.app.pop_screen()
def on_directory_tree_file_selected(
self, event: DirectoryTree.FileSelected
) -> None:
path = event.path
logger.debug("Directory file selected: %s", path)
try:
open_directory(str(path))
except Exception as exc:
logger.error("Failed to open %s: %s", path, exc)
self.app.bell()
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
logger.debug("Button pressed: %s", button_id)
match button_id:
case "move_agent_workflow_button":
self.app.push_screen(MoveAgentWorkflowScreen(self.app.devices))
event.stop()
case "otp_generate_button":
self.app.push_screen(OTPWorkflowScreen(self.app.devices))
event.stop()
case "find_quiet_button":
self.app.push_screen(
QuietAgentWorkflowScreen(self.app.api, self.app.policies)
)
event.stop()
return
case "otp_activities_button":
self.app.push_screen(OTPActivitiesScreen())
event.stop()
return
case "otp_revoke_button":
self.app.push_screen(OTPRevokeScreen())
event.stop()
return
case "policy_prep_button":
# Use the new TUI workflow screen instead of legacy
self.app.push_screen(
PolicyPrepWorkflowScreen(self.app.api, self.app.policies)
)
event.stop()
return
case _:
self.app.bell()
logger.warning("Unknown button pressed: %s", button_id)
return
# ---------------------------------------------------------------------------
# 2) APP
# ---------------------------------------------------------------------------
class Loxide(App[Message]):
api: AirlockAPIWrapper
working_dir: str
policies: Optional[list[Policy]]
devices: Optional[list[Agent]]
CSS = """
#logo {
width: 100%;
content-align: center middle;
text-align: center;
}
"""
BINDINGS = [
("q", "quit", "Quit"),
("f", "open_fe", "Launch Explorer"),
("r", "refresh", "Refresh"),
]
def __init__(self, api: AirlockAPIWrapper):
self._textual_theme = get_user_value("TEXTUAL_THEME", str, "textual-dark")
super().__init__()
self.api = api
wd = load_env("WORKING_DIR") or os.getcwd()
if not os.path.isdir(wd):
wd = os.getcwd()
self.working_dir = wd
# Initial data load
self.refresh_data()
def refresh_data(self) -> None:
"""Public method to refresh policies and devices from the API."""
try:
self.policies = [
Policy(**row.to_dict())
for _, row in self.api.policy_find_all().iterrows()
]
self.devices = [
Agent(**row.to_dict())
for _, row in self.api.agent_find_all().iterrows()
]
if self.policies and self.devices:
for agent in self.devices:
agent.enrich_with_policies(self.policies)
logger.debug(
f"Enriched {len(self.devices)} agents with policy information"
)
except Exception as exc:
logger.error("Failed to load policies/devices: %s", exc)
self.policies = None
self.devices = None
def on_mount(self, api: AirlockAPIWrapper) -> None:
self.register_theme(get_retro_terminal_theme())
self.register_theme(get_amber_terminal_theme())
self.theme = self._textual_theme
self.push_screen(MainMenuScreen())
def action_refresh(self) -> None:
self.refresh_data()
def action_quit(self) -> None:
global _APP_RESTART_REASON
_APP_RESTART_REASON = None
self.exit()
def action_open_fe(self) -> None:
"""Open the working directory in the OS file manager (footer binding)."""
path_to_open = self.working_dir or os.getcwd()
try:
open_directory(path_to_open)
except Exception as exc:
logger.error("Failed to open directory %s: %s", path_to_open, exc)
self.bell() # optional feedback
# ---------------------------------------------------------------------------
# 3) PUBLIC ENTRYPOINT
# ---------------------------------------------------------------------------
def run_Loxide(api: AirlockAPIWrapper) -> None:
global _APP_RESTART_REASON
base_dir = get_base_directory()
env_path = base_dir / ".env"
dotenv.load_dotenv(dotenv_path=env_path, override=True)
max_attempts = 5
attempts = 0
while attempts < max_attempts:
attempts += 1
logger.debug("Starting app loop iteration (attempt %d)", attempts)
_APP_RESTART_REASON = None
app = Loxide(api)
try:
app.run()
except SystemExit as exc:
if exc.code != 0:
logger.debug("Caught SystemExit from Textual: %s", exc)
raise
reason = _APP_RESTART_REASON
logger.debug("After app.run(), _APP_RESTART_REASON = %r", reason)
if not reason:
logger.debug("No restart reason, exiting loop")
break
if reason[0] == "restart":
logger.debug("Restarting app loop")
continue
if reason[0] == "multi_agent_action":
logger.info("Multi-agent action with selected agents: %s", reason[1])
continue
logger.error("Unknown restart reason: %r", reason)
break
# ---------------------------------------------------------------------------
# 4) DEV
# ---------------------------------------------------------------------------
if __name__ == "__main__":
api = AirlockAPIWrapper()
run_Loxide(api)
-50
View File
@@ -1,50 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
from textual.color import Color
from textual.theme import Theme
def get_amber_terminal_theme():
"""Amber CRT theme with compensated brightness for blending."""
return Theme(
name="amber-terminal",
background=Color.parse("#000000"), # pure black
primary=Color.parse("#ffb733"), # bright amber
secondary=Color.parse("#e69500"), # strong amber
success=Color.parse("#ffb733"),
warning=Color.parse("#ffff66"),
error=Color.parse("#ff3300"),
surface=Color.parse("#49331a"), # brighter brown for blending
)
AMBER_TERMINAL_CSS = """
Screen {
align: center middle;
background: #000000; /* force black */
color: #ffb733; /* force amber text */
}
.widget {
border: tall #ffb733; /* force amber border */
background: #3a1f00; /* compensated surface */
width: 80%;
}
* {
font-family: "Courier New", monospace;
}
"""
-53
View File
@@ -1,53 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
from textual.color import Color
def get_retro_terminal_theme():
from textual.theme import Theme
return Theme(
name="retro-terminal",
background=Color.parse("#000000"),
primary=Color.parse("#00ff00"),
secondary=Color.parse("#00aa00"),
success=Color.parse("#00ff00"),
warning=Color.parse("#ffff00"),
error=Color.parse("#ff0000"),
surface=Color.parse("#071802"),
)
RETRO_TERMINAL_CSS = """
/* Retro terminal CRT effect */
Screen {
align: center middle;
background: $background;
color: $text;
}
/* Blocky, pixelated widgets */
.widget {
border: tall $primary;
background: $surface;
width: 80%;
}
/* Monospaced font */
* {
font-family: "Courier New", monospace;
}
"""
-1
View File
@@ -43,7 +43,6 @@ class OTPGenerator(Widget):
# Reactive properties to track form completion # Reactive properties to track form completion
requestor_filled = reactive(False) requestor_filled = reactive(False)
reasoning_filled = reactive(False) reasoning_filled = reactive(False)
duration_selected = reactive(True) # Default is selected
otp_generated = reactive(False) otp_generated = reactive(False)
class OTPInfo(Message): class OTPInfo(Message):
+103 -5
View File
@@ -29,13 +29,71 @@ from textual.widget import Widget
from textual.widgets import Button, DataTable, Footer, Header, Static, TextArea from textual.widgets import Button, DataTable, Footer, Header, Static, TextArea
from models.agent import Agent from models.agent import Agent
from services.API import AirlockAPIWrapper
from TUI.Screens.executionhistoryscreen import ExecutionHistoryScreen
from TUI.Screens.otpworkflowscreen import OTPWorkflowScreen from TUI.Screens.otpworkflowscreen import OTPWorkflowScreen
from TUI.Screens.policyselectorscreen import PolicySelectorScreen from TUI.Screens.policyselectorscreen import PolicySelectorScreen
from TUI.Widgets.OTP_generate import OTPGenerator from TUI.Widgets.OTP_generate import OTPGenerator
from utils.configmanager import get_system_json
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def moveAgentToRelatedPolicy(
api: AirlockAPIWrapper,
agent: Agent,
mode: str = "audit",
):
"""
Moves an agent between audit and enforcement policies based on the mode.
Args:
api: AirlockAPIWrapper instance.
agent: Agent object.
mode: 'audit' to move to audit, 'enforcement' to move to enforcement.
"""
policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
if mode == "audit":
if agent.groupid in policy_relationship_map:
target_policy = policy_relationship_map[agent.groupid]
elif agent.groupid in policy_relationship_map.values():
logger.debug(
f"Agent {agent.hostname} is already in an audit group. No action needed."
)
print(
f"Agent {agent.hostname} is already in an audit group. No action needed."
)
return
else:
logger.warning(
f"Error: No corresponding audit policy found for groupid: {agent.groupid}."
)
return
elif mode == "enforcement":
inverse_map = {v: k for k, v in policy_relationship_map.items()}
if agent.groupid in inverse_map:
target_policy = inverse_map[agent.groupid]
elif agent.groupid in inverse_map.values():
logger.info(
f"Agent {agent.hostname} is already in an enforcement group. No action needed."
)
return
else:
logger.warning(
f"Error: No corresponding enforcement policy found for groupid: {agent.groupid}."
)
return
else:
logger.error(f"Unknown mode '{mode}'. Use 'audit' or 'enforcement'.")
return
result = api.agent_move(agent.agentid, target_policy)
return result
class AgentMoveOperations(Widget): class AgentMoveOperations(Widget):
""" """
A Textual widget for managing bulk agent operations and policy migrations. A Textual widget for managing bulk agent operations and policy migrations.
@@ -140,6 +198,7 @@ class AgentMoveOperations(Widget):
toggle_enforcement_btn = self.query_one("#toggle_enforcement_btn", Button) toggle_enforcement_btn = self.query_one("#toggle_enforcement_btn", Button)
other_policy_btn = self.query_one("#other_policy_btn", Button) other_policy_btn = self.query_one("#other_policy_btn", Button)
otp_gen_btn = self.query_one("#otp_gen_btn", Button) otp_gen_btn = self.query_one("#otp_gen_btn", Button)
exec_history_btn = self.query_one("#exec_history_btn", Button)
# If operation in progress, disable all # If operation in progress, disable all
if self.operation_in_progress: if self.operation_in_progress:
@@ -148,6 +207,7 @@ class AgentMoveOperations(Widget):
local_approval_btn.disabled = True local_approval_btn.disabled = True
toggle_enforcement_btn.disabled = True toggle_enforcement_btn.disabled = True
other_policy_btn.disabled = True other_policy_btn.disabled = True
exec_history_btn.disabled = True
else: else:
# If an operation was selected, disable # If an operation was selected, disable
if self.selected_operation: if self.selected_operation:
@@ -162,6 +222,9 @@ class AgentMoveOperations(Widget):
other_policy_btn.disabled = ( other_policy_btn.disabled = (
self.selected_operation == "other_policy" self.selected_operation == "other_policy"
) )
exec_history_btn.disabled = (
self.selected_operation == "exec_history"
)
else: else:
# Enable all buttons # Enable all buttons
otp_gen_btn = False otp_gen_btn = False
@@ -169,6 +232,7 @@ class AgentMoveOperations(Widget):
local_approval_btn.disabled = False local_approval_btn.disabled = False
toggle_enforcement_btn.disabled = False toggle_enforcement_btn.disabled = False
other_policy_btn.disabled = False other_policy_btn.disabled = False
exec_history_btn.disabled = False
except NoMatches: except NoMatches:
pass pass
@@ -316,6 +380,13 @@ class AgentMoveOperations(Widget):
other_policy_btn.styles.margin = (0, 0, 1, 0) other_policy_btn.styles.margin = (0, 0, 1, 0)
yield other_policy_btn yield other_policy_btn
exec_history_btn = Button(
"📊 View Execution History", id="exec_history_btn"
)
exec_history_btn.styles.width = "100%"
exec_history_btn.styles.margin = (0, 0, 1, 0)
yield exec_history_btn
# Status label # Status label
status_label = Static("", id="status_label") status_label = Static("", id="status_label")
status_label.styles.margin = (2, 0, 0, 0) status_label.styles.margin = (2, 0, 0, 0)
@@ -407,6 +478,9 @@ class AgentMoveOperations(Widget):
elif btn_id == "otp_gen_btn": elif btn_id == "otp_gen_btn":
self._start_OTP_gen_operation() self._start_OTP_gen_operation()
event.stop() event.stop()
elif btn_id == "exec_history_btn":
self._start_execution_history_operation()
event.stop()
def _start_local_approval_operation(self) -> None: def _start_local_approval_operation(self) -> None:
""" """
@@ -446,8 +520,6 @@ class AgentMoveOperations(Widget):
try: try:
import time import time
from services.agenthandler import moveAgentToRelatedPolicy
# Generate batch ID # Generate batch ID
batch = int(time.time()) batch = int(time.time())
duration = 360 # Default 6 hours, could make this configurable duration = 360 # Default 6 hours, could make this configurable
@@ -573,9 +645,6 @@ class AgentMoveOperations(Widget):
unsuccessful = [] unsuccessful = []
try: try:
from services.agenthandler import moveAgentToRelatedPolicy
from utils.configmanager import get_system_json
policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}") policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
for agent in self.agents: for agent in self.agents:
@@ -681,6 +750,35 @@ class AgentMoveOperations(Widget):
self.app.push_screen(OTPWorkflowScreen(self.agents)) self.app.push_screen(OTPWorkflowScreen(self.agents))
def _start_execution_history_operation(self) -> None:
"""
Launch the execution history viewer for selected agents.
This operation opens a new screen that allows the user to:
1. Select a date range for execution history
2. Fetch execution logs for all selected agents
3. View the results in a table
4. Export the results to CSV
The screen is pushed onto the screen stack, allowing the user to return
to this screen when done.
"""
status_label = self.query_one("#status_label", Static)
status_label.update("Opening execution history viewer...")
try:
# Push the execution history screen
self.app.push_screen(ExecutionHistoryScreen(self.agents))
logger.info(
f"Opened execution history viewer for {len(self.agents)} agents"
)
except Exception as e:
logger.error(f"Failed to open execution history viewer: {e}")
status_label.update(f"❌ Error: {str(e)}")
self.app.notify(
f"Failed to open execution history: {str(e)}", severity="error"
)
def _execute_move_to_policy(self, target_policy) -> None: def _execute_move_to_policy(self, target_policy) -> None:
""" """
Execute the actual move of agents to the selected policy. Execute the actual move of agents to the selected policy.
+128 -4
View File
@@ -14,6 +14,7 @@
# along with this program. If not, see <https://www.gnu.org/licenses/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
import difflib import difflib
from pathlib import Path
import re import re
from typing import List, Optional from typing import List, Optional
@@ -57,7 +58,7 @@ class MultiAgentSelector(Widget):
def compose(self): def compose(self):
yield Header(show_clock=True, icon="") yield Header(show_clock=True, icon="")
title_text = Static("🖥️ Agent Selector", id="selector_title") title_text = Static("🖥️ Agent Selector", id="selector_title")
title_text.styles.margin = (0, 0, 0, 1) title_text.styles.margin = (0, 0, 0, 1)
yield title_text yield title_text
@@ -77,7 +78,7 @@ class MultiAgentSelector(Widget):
text_area.styles.overflow_y = "auto" text_area.styles.overflow_y = "auto"
yield text_area yield text_area
with Horizontal(id="switch_search_container"): with Horizontal(id="switch_container"):
switch = Switch(value=False, id="match_switch") switch = Switch(value=False, id="match_switch")
switch.styles.width = "auto" switch.styles.width = "auto"
switch.styles.margin = (1, 0, 0, 0) switch.styles.margin = (1, 0, 0, 0)
@@ -89,8 +90,13 @@ class MultiAgentSelector(Widget):
switch_label.styles.margin = (2, 1, 0, 0) switch_label.styles.margin = (2, 1, 0, 0)
yield switch_label yield switch_label
with Horizontal(id="action_buttons_container"):
load_file = Button("📂 Load File", id="load_file_button")
load_file.styles.margin = (1, 1, 0, 1)
yield load_file
search = Button("🔍 Search", id="search_button") search = Button("🔍 Search", id="search_button")
search.styles.margin = (1, 0, 0, 0) search.styles.margin = (1, 0, 0, 1)
yield search yield search
with Horizontal() as select_buttons: with Horizontal() as select_buttons:
@@ -152,6 +158,9 @@ class MultiAgentSelector(Widget):
] ]
self.post_message(self.AgentsSelected(selected_agents)) self.post_message(self.AgentsSelected(selected_agents))
event.stop() event.stop()
elif btn_id == "load_file_button":
self._load_from_file()
event.stop()
elif btn_id == "search_button": elif btn_id == "search_button":
self.update_matches() self.update_matches()
event.stop() event.stop()
@@ -166,7 +175,7 @@ class MultiAgentSelector(Widget):
match_list.add_option((name, name)) match_list.add_option((name, name))
unmatched_label = self.query_one("#unmatched_label", Static) unmatched_label = self.query_one("#unmatched_label", Static)
if unmatched: if unmatched:
unmatched_label.update(f"⚠️ No matches for: {', '.join(unmatched)}") unmatched_label.update(f"⚠️ No matches for: {', '.join(unmatched)}")
else: else:
unmatched_label.update("") unmatched_label.update("")
@@ -214,3 +223,118 @@ class MultiAgentSelector(Widget):
else: else:
unmatched.append(name) unmatched.append(name)
return sorted(matched), unmatched return sorted(matched), unmatched
def _load_from_file(self):
"""Safely load device names from a text file."""
try:
# Import here to avoid issues if tkinter isn't available
import tkinter as tk
from tkinter import filedialog
# Create file dialog
root = tk.Tk()
root.withdraw()
file_path = filedialog.askopenfilename(
title="Select device list file",
filetypes=[
("Text files", "*.txt"),
("CSV files", "*.csv"),
("All files", "*.*"),
],
)
if not file_path:
# User cancelled
return
# Validate file path
path_obj = Path(file_path)
if not path_obj.exists():
self.app.notify("File does not exist", severity="error", timeout=3)
return
if not path_obj.is_file():
self.app.notify(
"Selected path is not a file", severity="error", timeout=3
)
return
# Check file size (limit to 1 MB for safety)
file_size = path_obj.stat().st_size
if file_size > 1_000_000: # 1 MB
self.app.notify(
f"File too large ({file_size:,} bytes). Maximum 1 MB.",
severity="error",
timeout=5,
)
return
# Read file with proper encoding to preserve emojis
try:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
except UnicodeDecodeError:
# Try with different encoding if UTF-8 fails
try:
with open(file_path, "r", encoding="latin-1") as f:
content = f.read()
self.app.notify(
"File loaded with Latin-1 encoding (UTF-8 failed)",
severity="warning",
timeout=3,
)
except Exception as e:
self.app.notify(
f"Error reading file: {str(e)}", severity="error", timeout=5
)
return
# Validate and sanitize content
lines = content.split("\n")
valid_lines = []
invalid_count = 0
# Pattern for valid hostnames/device names
# Allows: letters, numbers, hyphens, underscores, periods, and Unicode chars
hostname_pattern = re.compile(r"^[\w\-\.\u0080-\uFFFF]+$", re.UNICODE)
for line in lines:
line = line.strip()
if not line:
continue # Skip empty lines
# Check if line looks like a valid hostname/device name
if hostname_pattern.match(line):
valid_lines.append(line)
else:
invalid_count += 1
# Log but don't add invalid entries
if not valid_lines:
self.app.notify(
"No valid device names found in file", severity="warning", timeout=3
)
return
# Update text area with validated content
text_area = self.query_one("#device_input", TextArea)
text_area.text = "\n".join(valid_lines)
# Show notification
msg = f"✅ Loaded {len(valid_lines)} devices from file"
if invalid_count > 0:
msg += f" ({invalid_count} invalid entries skipped)"
self.app.notify(msg, severity="information", timeout=5)
except ImportError:
self.app.notify(
"tkinter not available - cannot open file dialog",
severity="error",
timeout=3,
)
except Exception as e:
self.app.notify(
f"Error loading file: {str(e)}", severity="error", timeout=5
)
+100 -1
View File
@@ -18,8 +18,9 @@ import logging
from rich.text import Text from rich.text import Text
from textual.containers import Horizontal, Vertical from textual.containers import Horizontal, Vertical
from textual.message import Message
from textual.widget import Widget from textual.widget import Widget
from textual.widgets import Input, OptionList, Static, Switch, Tree from textual.widgets import Button, Input, OptionList, Static, Switch, Tree
from textual.widgets.option_list import Option from textual.widgets.option_list import Option
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -28,6 +29,27 @@ logger = logging.getLogger(__name__)
class PolicyTreeWidget(Widget): class PolicyTreeWidget(Widget):
"""Widget for displaying and searching a hierarchical policy tree.""" """Widget for displaying and searching a hierarchical policy tree."""
class ViewExecutionHistory(Message):
"""Message sent when user wants to view execution history for a device."""
def __init__(self, device):
super().__init__()
self.device = device
class GenerateOTP(Message):
"""Message sent when user wants to generate OTP for a device."""
def __init__(self, device):
super().__init__()
self.device = device
class ToggleEnforcement(Message):
"""Message sent when user wants to toggle audit/enforcement for a device."""
def __init__(self, device):
super().__init__()
self.device = device
def __init__(self, policies, devices): def __init__(self, policies, devices):
super().__init__() super().__init__()
self.policies = policies self.policies = policies
@@ -35,6 +57,7 @@ class PolicyTreeWidget(Widget):
self.last_highlighted_node = None self.last_highlighted_node = None
self.leaf_counts = defaultdict(int) self.leaf_counts = defaultdict(int)
self.match_type = "Count" # Default to sorting by count self.match_type = "Count" # Default to sorting by count
self.selected_device = None # Track currently selected device
def compose(self): def compose(self):
# Create the switch and its label # Create the switch and its label
@@ -56,6 +79,18 @@ class PolicyTreeWidget(Widget):
search_box = Input( search_box = Input(
placeholder="Search policies or devices...", id="tree_search" placeholder="Search policies or devices...", id="tree_search"
) )
exec_history_button = Button(
"📊 Execution History", id="view_exec_history_button", disabled=True
)
exec_history_button.styles.margin = (0, 1, 0, 0) # Right margin
otp_button = Button("🎫 Generate OTP", id="generate_otp_button", disabled=True)
otp_button.styles.margin = (0, 1, 0, 0) # Right margin
toggle_enforcement_button = Button(
"🔄 Toggle Enforcement/Audit", id="toggle_enforcement_button", disabled=True
)
# No right margin on last button
details_pane = Static("", id="details_pane") details_pane = Static("", id="details_pane")
# Layout the UI # Layout the UI
@@ -73,6 +108,12 @@ class PolicyTreeWidget(Widget):
# Add the search box and details pane # Add the search box and details pane
yield label yield label
yield search_box yield search_box
# Action buttons in a horizontal row
with Horizontal() as button_row:
button_row.styles.height = "auto"
yield exec_history_button
yield otp_button
yield toggle_enforcement_button
yield details_pane yield details_pane
def on_mount(self) -> None: def on_mount(self) -> None:
@@ -89,6 +130,32 @@ class PolicyTreeWidget(Widget):
# Expand the root node # Expand the root node
policy_tree.root.expand() policy_tree.root.expand()
def refresh_data(self, policies, devices):
"""Refresh the widget with new data and rebuild the tree."""
self.policies = policies
self.devices = devices
self.selected_device = None
# Disable all buttons since selection is lost
try:
self.query_one("#view_exec_history_button", Button).disabled = True
self.query_one("#generate_otp_button", Button).disabled = True
self.query_one("#toggle_enforcement_button", Button).disabled = True
except:
pass
# Rebuild tree with new data
self._precompute_leaf_counts()
total_leaves = sum(
self.leaf_counts.get(policy.groupid, 0)
for policy in self.policies
if policy.parent == "global-policy-settings"
)
policy_tree = self.query_one("#policy_tree", Tree)
policy_tree.root.set_label(f"Agents in Policies: ({total_leaves})")
self._build_tree()
policy_tree.root.expand()
def _precompute_leaf_counts(self): def _precompute_leaf_counts(self):
"""Precompute leaf counts for each policy group.""" """Precompute leaf counts for each policy group."""
device_counts = defaultdict(int) device_counts = defaultdict(int)
@@ -184,6 +251,9 @@ class PolicyTreeWidget(Widget):
node = message.node node = message.node
data = node.data data = node.data
details_pane = self.query_one("#details_pane", Static) details_pane = self.query_one("#details_pane", Static)
exec_history_button = self.query_one("#view_exec_history_button", Button)
otp_button = self.query_one("#generate_otp_button", Button)
toggle_enforcement_button = self.query_one("#toggle_enforcement_button", Button)
if self.last_highlighted_node is not None: if self.last_highlighted_node is not None:
original_label = str(self.last_highlighted_node.label).strip() original_label = str(self.last_highlighted_node.label).strip()
@@ -198,6 +268,20 @@ class PolicyTreeWidget(Widget):
node.set_label(highlighted_label) node.set_label(highlighted_label)
self.last_highlighted_node = node self.last_highlighted_node = node
# Check if selected node is a device (has Agent data)
from models.agent import Agent
if data and isinstance(data, Agent):
self.selected_device = data
exec_history_button.disabled = False
otp_button.disabled = False
toggle_enforcement_button.disabled = False
else:
self.selected_device = None
exec_history_button.disabled = True
otp_button.disabled = True
toggle_enforcement_button.disabled = True
if data: if data:
details = "\n".join( details = "\n".join(
f"{key}: {value}" for key, value in data.__dict__.items() f"{key}: {value}" for key, value in data.__dict__.items()
@@ -287,3 +371,18 @@ class PolicyTreeWidget(Widget):
option_list.remove() option_list.remove()
except: except:
pass pass
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
if event.button.id == "view_exec_history_button":
if self.selected_device:
self.post_message(self.ViewExecutionHistory(self.selected_device))
event.stop()
elif event.button.id == "generate_otp_button":
if self.selected_device:
self.post_message(self.GenerateOTP(self.selected_device))
event.stop()
elif event.button.id == "toggle_enforcement_button":
if self.selected_device:
self.post_message(self.ToggleEnforcement(self.selected_device))
event.stop()
-870
View File
@@ -1,870 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
import logging
import os
import os.path
import re
from typing import List
import dotenv
import pandas as pd
from models.execution import ExecutionHistoryRecord
from models.policy import Allowlist, Policy
from services.API import AirlockAPIWrapper
from utils.configmanager import get_system_list, get_system_value, load_env
from utils.selector import Selector
from utils.utils import (
areYouSure,
clear_screen,
colorText,
formatHTML,
get_sanitized_input,
locked,
open_directory,
print_x_wide,
regulator,
)
logger = logging.getLogger(__name__)
dotenv.load_dotenv()
def selectPolicies(api: AirlockAPIWrapper, allow_multiple=True) -> List[Policy]:
policies = [Policy(**row.to_dict()) for _, row in api.policy_find_all().iterrows()]
logger.debug("Prompting for Policies")
print(colorText("Please select policy/policies", "white"))
selected = Selector.select_objects(policies, allow_multiple, prompt_each=True)
if selected is None:
return []
# Normalize to always return a list
logger.debug("Returning {selected.dict}")
return selected if isinstance(selected, list) else [selected]
def selectAllowlists(
api: AirlockAPIWrapper, policy=all, allow_multiple=True
) -> List[Allowlist]:
if policy == "all":
allowlists = [
Allowlist(**row.to_dict()) for _, row in api.allowlist_find_all().iterrows()
]
else:
allowlists = [
Allowlist(**row.to_dict())
for _, row in api.policy_list_allowlists(policy[0].groupid).iterrows()
]
logger.debug("Prompting for Allowlist(s)")
print(colorText("Please select allowlist(s)", "white"))
selected = Selector.select_objects(allowlists, allow_multiple, prompt_each=True)
if selected is None:
return []
# Normalize to always return a list
logger.debug(f"Returning {selected}")
return selected if isinstance(selected, list) else [selected]
def sortHashes(
api: AirlockAPIWrapper, selected_policies: List[Policy], type=[1, 2, 6, 7]
):
working_dir = load_env("WORKING_DIR")
history_days = Selector.select_value(
prompt="Enter how many days of history to pull (1150): ",
value_type=int,
valid_range=(1, 150),
)
logger.debug(f"{history_days} day selected for history")
if history_days is None:
logging.warning("No history range selected. Aborting.")
return
policy_executions = ExecutionHistoryRecord.from_policies(
api, selected_policies, type_=type, history_days=history_days
)
logger.debug(f"Executions contains {policy_executions}")
enriched_executions = ExecutionHistoryRecord.enrich_with_hashes(
api, policy_executions
)
categorized_executions = (
ExecutionHistoryRecord.categorize_executions_by_hash_decision(
enriched_executions
)
)
approved, unapproved, needs_review, unknown = (
ExecutionHistoryRecord.sort_by_hash_decision(categorized_executions)
)
categories = {
"needs_review": needs_review,
"approved": approved,
"unapproved": unapproved,
"leftover": unknown,
}
for label, records in categories.items():
if not records:
continue # Skip empty or falsy categories
csv_path = f"{working_dir}\\Needs_Review\\Review_First\\{selected_policies[0].name}_{label}_executions.csv"
html_path = f"{working_dir}\\Needs_Review\\HTML\\{selected_policies[0].name}_{label}.html"
# Convert ExecutionHistoryRecord objects to dictionaries
df = pd.DataFrame([r.__dict__ for r in records])
# Optional: flatten hash_obj if needed
if not df.empty and "hash_obj" in df.columns:
hash_df = df["hash_obj"].apply(lambda h: h.to_dict() if h else {})
df = pd.concat([df.drop(columns=["hash_obj"]), hash_df], axis=1)
# Save to CSV
df.to_csv(csv_path, index=False)
logger.info(f"Saved {label} executions to {csv_path}")
# Generate HTML
formatHTML(df, html_path)
logger.info(f"Generated HTML report at {html_path}")
def buildPathsandPublishers(selected_policies: List[Policy], split):
working_dir = load_env("WORKING_DIR")
df1 = pd.DataFrame()
df2 = pd.DataFrame()
all_approved_hashes = pd.DataFrame()
path1 = (
f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv"
)
path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_needs_review_executions.csv"
path_exclusion_constant = get_system_value("PATH_EXCLUSION_CONST", cast_type=int)
if os.path.exists(path1):
df1 = pd.read_csv(path1)
else:
logger.warning(f"File not found: {path1}")
if os.path.exists(path2):
df2 = pd.read_csv(path2)
else:
logger.warning(f"File not found: {path2}")
if df1.empty and df2.empty:
logger.warning("Both DataFrames are empty. Skipping sort.")
all_approved_hashes = pd.DataFrame()
logger.debug(all_approved_hashes.head)
else:
all_approved_hashes = pd.concat([df1, df2], ignore_index=True)
if "filename" in all_approved_hashes.columns:
all_approved_hashes = all_approved_hashes.sort_values(by="filename")
else:
logger.warning(
"Warning: 'filename' column not found in concatenated DataFrame."
)
if not all_approved_hashes.empty and path_exclusion_constant:
primary_path_exclusions = calculatePath(
all_approved_hashes,
path_exclusion_constant,
split,
)
remaining_hashes = all_approved_hashes[
~all_approved_hashes["sha256"].isin(primary_path_exclusions["sha256"])
]
secondary_path_exclusions = calculatePath(
remaining_hashes, (path_exclusion_constant - 1), split
)
remaining_hashes = remaining_hashes[
~remaining_hashes["sha256"].isin(secondary_path_exclusions["sha256"])
]
dataframes = {
"all_approved_hashes": all_approved_hashes,
"primary_Paths": primary_path_exclusions,
"secondary_Paths": secondary_path_exclusions,
"hashes_not_approvable_by_path": remaining_hashes,
}
logger.debug("Preparing to sort dataframes")
for name, df in dataframes.items():
logger.debug(f" DataFrame headers: {list(df.columns)}")
if "hashes" in name:
df.sort_values(by="filename", inplace=True)
else:
df.sort_values(by="longestcfp", inplace=True)
df.to_csv(
f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_{name}.csv",
index=False,
)
formatHTML(
df,
f"{working_dir}\\Needs_Review\\HTML\\{selected_policies[0].name}_{name}.html",
)
if not all_approved_hashes.empty:
# Drop all not signed, only keep unique values
publist = all_approved_hashes[
all_approved_hashes["publisher"] != "Not Signed"
].drop_duplicates(subset=["publisher"])
# Remove Bad publisher if somehow they made it this far
pattern = regulator(get_system_list("BAD_PUBLISHERS"))
publist = publist[~publist["publisher"].str.contains(pattern, na=False)]
publist = publist[["publisher"]]
publist.sort_values(by="publisher", inplace=True)
publist.to_csv(
f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_publishers.csv",
index=False,
)
else:
logger.debug("Approved Hashes list appears empty")
def buildPreflights(selected_policies: List[Policy]):
working_dir = load_env("WORKING_DIR")
df1 = pd.DataFrame()
df2 = pd.DataFrame()
approved_hashes = pd.DataFrame()
approved_publishers = pd.DataFrame()
hash = f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_all_approved_hashes.csv"
path1 = f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv"
path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_secondary_Paths.csv"
publishers = f"{working_dir}\\Approved\\{selected_policies[0].name}_publishers.csv"
# Read in and combine the two path generations
if os.path.exists(path1):
df1 = pd.read_csv(path1)
else:
logger.warning(f"File not found: {path1}")
if os.path.exists(path2):
df2 = pd.read_csv(path2)
else:
logger.warning(f"File not found: {path2}")
if df1.empty and df2.empty:
logger.warning("Both DataFrames are empty. Skipping sort.")
approved_paths = pd.DataFrame()
else:
approved_paths = pd.concat([df1, df2], ignore_index=True)
approved_paths = approved_paths.drop_duplicates(subset="longestcfp", keep="first")
# We create a list of hashes that are left over if we exclude the ones that are covered by the path exclusions.
if os.path.exists(hash):
hashes = pd.read_csv(hash)
approved_hashes = hashes[~hashes["filename"].isin(approved_paths["longestcfp"])]
approved_hashes = approved_hashes.drop_duplicates(subset="sha256", keep="first")
else:
logger.warning(f"File not found: {hash}")
if os.path.exists(publishers):
approved_publishers = pd.read_csv(publishers)
else:
logger.warning(f"File not found: {publishers}")
dataframes = {
"approved_paths": approved_paths,
"approved_hashes": approved_hashes,
"approved_publishers": approved_publishers,
}
for name, df in dataframes.items():
logger.debug(f" DataFrame headers: {list(df.columns)}")
if name == "approved_paths":
df.sort_values(by="longestcfp", inplace=True)
elif name == "approved_hashes":
df.sort_values(by="filename", inplace=True)
elif name == "approved_publishers":
df.sort_values(by="publisher", inplace=True)
df.to_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_{name}.csv",
index=False,
)
formatHTML(
df,
f"{working_dir}\\Preflight\\HTML\\{selected_policies[0].name}_{name}.html",
)
def splitFilepathsGrouped(df, path_exclusion_constant, col="filename"):
min_files_for_path = get_system_value("MIN_FILES_FOR_PATH", cast_type=int)
def clean_split(path):
if not isinstance(path, (str, bytes, os.PathLike)):
return []
parts = str(os.path.normpath(path)).split(os.sep)
parts = [p for p in parts if p] # Remove empty strings
return parts
# Diagnostic: log any non-string entries
non_string_entries = df[
~df[col].apply(lambda x: isinstance(x, (str, bytes, os.PathLike)))
]
if not non_string_entries.empty:
print(f"[WARNING] Non-string entries found in column '{col}':")
print(non_string_entries)
df = df.copy()
split_paths = df[col].apply(clean_split)
if min_files_for_path is not None:
df = df[
split_paths.apply(lambda parts: len(parts) >= min_files_for_path)
].copy()
split_paths = split_paths[df.index]
df["group_key"] = split_paths.apply(
lambda parts: os.sep.join(parts[:path_exclusion_constant])
)
grouped = df.groupby("group_key")
new_rows = []
for _, group_df in grouped:
paths = group_df[col].tolist()
split_parts = [clean_split(p) for p in paths]
def longest_common_prefix(paths):
if not paths:
return []
prefix = paths[0]
for path in paths[1:]:
prefix = [a for a, b in zip(prefix, path) if a == b]
if not prefix:
break
return prefix
common_prefix = longest_common_prefix(split_parts)
prefix_str = os.sep.join(common_prefix)
for i, parts in enumerate(split_parts):
filename = parts[-1]
middle = (
os.sep.join(parts[len(common_prefix) : -1])
if len(parts) > len(common_prefix) + 1
else ""
)
row = group_df.iloc[i].copy()
row["longestcfp"] = prefix_str
row["middle"] = middle
row["filename_only"] = filename
row["file_extension"] = os.path.splitext(filename)[1].lower()
new_rows.append(row)
return pd.DataFrame(new_rows).drop(columns=["group_key"])
def calculatePath(approved_hashes, path_exclusion_constant, split):
if split:
dfs_by_policy = [group for _, group in approved_hashes.groupby("policy")]
else:
dfs_by_policy = [approved_hashes]
badpathparts = get_system_list("BAD_PATH_PARTS")
min_files_for_path = get_system_value("MIN_FILES_FOR_PATH", cast_type=int)
processed_dfs = []
for df in dfs_by_policy:
haslcp = splitFilepathsGrouped(df, path_exclusion_constant, "filename")
haslcp = haslcp.drop_duplicates()
forbidden = regulator(badpathparts, True)
forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False)
logger.debug("Removing forbidden filepaths for path exceptions")
print(colorText("Removing forbidden filepaths for path exceptions", "green"))
lcp_not_forbidden = haslcp[~forbidden_lcfp].copy()
lcp_not_forbidden_review = lcp_not_forbidden[
[
"policyname",
"longestcfp",
"middle",
"filename_only",
"file_extension",
"sha256",
]
]
unique_sha_counts = (
lcp_not_forbidden_review.groupby("longestcfp")["sha256"]
.nunique()
.reset_index()
)
unique_sha_counts.columns = ["longestcfp", "unique_sha256_count"]
lcp_not_forbidden_review = lcp_not_forbidden_review.merge(
unique_sha_counts, on="longestcfp", how="left"
)
lcp_not_forbidden_review = lcp_not_forbidden_review[
lcp_not_forbidden_review["unique_sha256_count"] >= min_files_for_path
]
processed_dfs.append(lcp_not_forbidden_review)
pathExclusions = pd.concat(processed_dfs, ignore_index=True)
return pathExclusions
def testChange(selected_policies, destination_policy, destination_allowlist):
working_dir = load_env("WORKING_DIR")
logger.info("These path exclusions would be added to:")
logger.info(destination_policy)
pathexclusions = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"
)
hashes = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv"
)
unique_combinations = pathexclusions[
["longestcfp", "file_extension"]
].drop_duplicates()
drive_letter_pattern = re.compile(r"^[a-zA-Z]:\\")
processed_paths = [
(path if drive_letter_pattern.match(path) else f"\\\\{path}") + f"\\**{ext}"
for path, ext in unique_combinations.itertuples(index=False, name=None)
]
for path in processed_paths:
logger.info(path)
print(colorText("These publishers would added", "yellow"))
processed_publishers = []
if os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv"
):
publishers = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv"
)
if publishers.empty:
print(colorText("The publishers list is empty.", "red"))
else:
processed_publishers = (
publishers[publishers["publisher"] != "Not Signed"]["publisher"]
.drop_duplicates()
.tolist()
)
for publisher in processed_publishers:
print(publisher)
print(colorText("These hashes would be added to:", "yellow"))
print(destination_allowlist)
processed_hashes = hashes["sha256"].unique().tolist()
print_x_wide(processed_hashes, 3)
return processed_paths, processed_hashes, processed_publishers
def menu_policy_enforce(
api: AirlockAPIWrapper,
): # TODO Need to clean up 6 and 7 into functions
selected_policies = []
destination_policy = []
destination_allowlist = []
processed_paths = []
processed_hashes = []
processed_publishers = []
working_dir = load_env("WORKING_DIR")
while True:
printEnforceChecklist(
selected_policies, destination_policy, destination_allowlist
)
choice = get_sanitized_input("\nEnter your choice: ")
if choice == "1":
clear_screen()
selected_policies = selectPolicies(api, True)
elif choice == "2":
clear_screen()
print(
colorText(
"Please choose destination_name Policy for Path Exclusions", "white"
)
)
destination_policy = selectPolicies(api, False)
print(colorText("Please choose Allowlist for Hashes", "white"))
destination_allowlist = selectAllowlists(api, destination_policy, False)
elif choice == "3":
clear_screen()
sortHashes(
api,
selected_policies,
type=[1, 2, 6, 7],
)
elif choice == "4":
clear_screen()
if os.path.exists(
f"{working_dir}\\Needs_Review\\Review_First\\{selected_policies[0].name}_approved_executions.csv"
):
buildPathsandPublishers(selected_policies, False)
else:
print(
"File not found. Please make sure it's saved correctly and try again."
)
elif choice == "5":
clear_screen()
if os.path.exists(
f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv"
) and os.path.exists(
f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv"
):
buildPreflights(selected_policies)
else:
print(
"File not found. Please make sure it's saved correctly and try again."
)
elif choice == "6":
clear_screen()
if (
os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"
)
and os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv"
)
and destination_policy
and destination_allowlist
):
processed_paths, processed_hashes, processed_publishers = testChange(
selected_policies, destination_policy, destination_allowlist
)
else:
# Log which condition(s) failed
missing_items = []
if not os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"
):
missing_items.append("approved_paths.csv not found")
if not os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv"
):
missing_items.append("approved_hashes.csv not found")
if not destination_policy:
missing_items.append("destination_policy is empty or None")
if not destination_allowlist:
missing_items.append("destination_allowlist is empty or None")
logger.error("Preflight check failed due to the following:")
for item in missing_items:
logger.error(f" - {item}")
elif choice == "7":
clear_screen()
areYouSure()
confirmation = get_sanitized_input("Type 'I AGREE' to continue: ")
if (
processed_paths
and processed_hashes
and processed_publishers
and destination_policy
and destination_allowlist
and confirmation.strip() == "I AGREE"
):
print(colorText("Proceeding with the code...", "yellow"))
api.hash_add_to_allowlist(
destination_allowlist[0].applicationid, processed_hashes
)
api.policy_add_path_exclusions(
destination_policy[0].groupid, processed_paths
)
if processed_publishers:
api.policy_add_publishers(
destination_policy[0].groupid, processed_publishers
)
locked()
else:
logger.error("Confirmation block failed. Reasons:")
if not processed_publishers or processed_hashes or processed_paths:
logger.error(" - Test not performed.")
if not destination_policy:
logger.error(" - `destination_policy` is missing or invalid.")
if not destination_allowlist:
logger.error(" - `destination_allowlist` is missing or invalid.")
if confirmation.strip() != "I AGREE":
logger.error(
" - User did not confirm with 'I AGREE'. Received: '%s'",
confirmation.strip(),
)
elif choice.upper() == "F":
open_directory(working_dir)
elif choice.upper() == "B":
break
else:
print(colorText("Invalid choice. Please try again.", "red"))
def section_header(title):
print(
colorText(
"\n --------------------------------------------------------------------",
"cyan",
)
)
print(colorText(f" ------------- {title} -------------", "cyan"))
print(
colorText(
" --------------------------------------------------------------------",
"cyan",
)
)
def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist):
working_dir = load_env("WORKING_DIR")
section_header("Prepare to Enforce Policy ")
print(
colorText(
"\nSequentially follow these steps to prepare a policy for enforcement:",
"white",
)
)
# Step 1: Originating Policies
print(
colorText(
"\n1. Choose which policy or policies to gather execution info from", "cyan"
)
)
if not selected_policies:
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"))
# Step 2: Destination Policy and Allowlist
print(
colorText("2. Choose the destination policy and associated allowlist", "cyan")
)
if destination_policy:
print(
colorText(
f" [✓] {destination_policy[0].name} has been selected as the destination policy",
"green",
)
)
else:
print(colorText(" [✗] No destination policy has been chosen", "red"))
if destination_allowlist:
print(
colorText(
f" [✓] {destination_allowlist[0].name} has been selected as allowlist",
"green",
)
)
else:
print(colorText(" [✗] No allowlist has been chosen", "red"))
# Step 3: Data Preparation
print(
colorText(
f"3. Select to begin pulling execution history. The executions will be sorted and placed in {working_dir}\\data\\Needs_Review",
"cyan",
)
)
if selected_policies:
policy_id = selected_policies[0].name
review_path = f"{working_dir}\\Needs_Review\\Review_First\\{policy_id}_approved_executions.csv"
print(
colorText(
(
" [✓] Data has been fetched"
if os.path.exists(review_path)
else " [✗] Data has not been fetched"
),
"green" if os.path.exists(review_path) else "red",
)
)
else:
print(
colorText(
" [✗] No policies selected, cannot check data fetch status", "red"
)
)
# Step 4: Manual Review
print(colorText("4. Manually review the files:", "cyan"))
print(
colorText(
" Remove the rows containing hashes you do not approve of", "cyan"
)
)
print(
colorText(
f" When complete, save both csv files to {working_dir}\\data\\Approved and choose this option.",
"cyan",
)
)
print(
colorText(
" This will start the process to generate possible filepath approvals",
"cyan",
)
)
if selected_policies:
policy_id = selected_policies[0].name
approved_path = f"{working_dir}\\Approved\\{policy_id}_approved_executions.csv"
second_review_path = (
f"{working_dir}\\Needs_Review\\Review_Second\\{policy_id}_primary_Paths.csv"
)
print(
colorText(
(
" [✓] Reviewed hashes have been loaded"
if os.path.exists(approved_path)
else " [✗] Reviewed hashes have not been loaded"
),
"green" if os.path.exists(approved_path) else "red",
)
)
print(
colorText(
(
" [✓] Path review list created"
if os.path.exists(second_review_path)
else " [✗] Path review list has not been created"
),
"green" if os.path.exists(second_review_path) else "red",
)
)
else:
print(
colorText(
" [✗] No policies selected, cannot check reviewed hashes or path list",
"red",
)
)
# Step 5: Path Review
print(
colorText(
f"5. Manually review the files in {working_dir}\\Needs_Review\\Review_Second\\",
"cyan",
)
)
print(
colorText(
" Remove the rows containing path exclusions or publishers you do not approve of.",
"cyan",
)
)
print(
colorText(
f" When complete, save the files to {working_dir}\\data\\Approved",
"cyan",
)
)
print(
colorText(" Choose this option when done to build your preflights", "cyan")
)
if selected_policies:
policy_id = selected_policies[0].name
reviewed_path = f"{working_dir}\\Approved\\{policy_id}_primary_Paths.csv"
preflight_paths = f"{working_dir}\\Preflight\\{policy_id}_approved_paths.csv"
preflight_hashes = f"{working_dir}\\Preflight\\{policy_id}_approved_hashes.csv"
print(
colorText(
(
" [✓] Reviewed path list detected"
if os.path.exists(reviewed_path)
else " [✗] Path review list has not been detected"
),
"green" if os.path.exists(reviewed_path) else "red",
)
)
preflight_ready = os.path.exists(preflight_paths) and os.path.exists(
preflight_hashes
)
print(
colorText(
(
" [✓] Preflight Path Exclusion List has been generated"
if preflight_ready
else " [✗] Preflight Path Exclusion List has not been generated"
),
"green" if preflight_ready else "red",
)
)
else:
print(
colorText(
" [✗] No policies selected, cannot check preflight status", "red"
)
)
# Final Steps
print(
colorText(
"6. Test ------------------------------------------------------", "cyan"
)
)
print(
colorText(
" Prints to console the changes that would be made, must be done to proceed. ",
"cyan",
)
)
print(
colorText(
"7. Liftoff ------------------------------------------------------", "cyan"
)
)
print(
colorText(
" Apply path exclusions and approved publishers to selected policy",
"cyan",
)
)
print(colorText(" Apply approved hashes to allowlist", "cyan"))
# Utility Options
print(colorText("F. Open Working Directory", "cyan"))
print(colorText("B. Back", "cyan"))
+261
View File
@@ -0,0 +1,261 @@
# 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 <https://www.gnu.org/licenses/>.
import datetime
import logging
from bson import ObjectId
from textual.app import ComposeResult
from textual.containers import Container, Vertical
from textual.widgets import Button, DataTable, Input, Static
from services.API import AirlockAPIWrapper
logger = logging.getLogger(__name__)
def skipback(days):
"""
Generate a MongoDB ObjectId for a given number of days ago from today.
"""
adjusted_days = days
date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(
days=adjusted_days
)
timestamp = int(date_days_ago.timestamp())
hex_timestamp = format(timestamp, "08x")
objectid_hex = hex_timestamp + "0000000000000000"
return ObjectId(objectid_hex)
class ServerLogWidget(Vertical):
"""Widget for displaying server activity logs in a DataTable."""
DEFAULT_CSS = """
ServerLogWidget {
width: 100%;
height: 100%;
}
ServerLogWidget #status_bar {
width: 100%;
height: auto;
background: $surface;
padding: 1;
margin-bottom: 1;
}
ServerLogWidget DataTable {
height: 1fr;
border: solid $primary;
}
ServerLogWidget #button_container {
width: 100%;
height: auto;
layout: horizontal;
padding: 1;
align: left middle;
}
ServerLogWidget .filter_label {
width: auto;
height: 3;
content-align: left middle;
padding-right: 1;
}
ServerLogWidget #filter_input {
width: 40;
margin-right: 1;
}
ServerLogWidget Button {
margin-right: 1;
}
"""
def __init__(self, api: AirlockAPIWrapper):
super().__init__()
self.api = api
self.all_logs = [] # Store all logs for filtering
self.columns = [] # Store column names
def compose(self) -> ComposeResult:
yield Static("Loading server logs (last 72 hours)...", id="status_bar")
yield DataTable(id="server_log_table")
with Container(id="button_container"):
yield Static("Filter:", classes="filter_label")
yield Input(placeholder="Filter (use * and ? wildcards)", id="filter_input")
yield Button("Refresh", id="refresh_button", variant="primary")
def on_mount(self) -> None:
"""Initialize the DataTable and load server logs."""
self.load_logs()
def load_logs(self) -> None:
"""Load server logs from the API and populate the DataTable."""
table = self.query_one("#server_log_table", DataTable)
status = self.query_one("#status_bar", Static)
try:
status.update("⏳ Loading server logs (last 72 hours)...")
# Create a fake checkpoint for 3 days ago (72 hours)
checkpoint = str(skipback(3))
# Get server logs from API
logs = self.api.server_logs(checkpoint=checkpoint)
if not logs:
status.update("ℹï¸ No server logs found in the last 72 hours.")
table.clear(columns=True)
self.all_logs = []
self.columns = []
return
# Store all logs for filtering
self.all_logs = logs
# Clear existing data
table.clear(columns=True)
# Add columns based on the first log entry
if logs:
first_log = logs[0]
self.columns = [col for col in first_log.keys() if col != "checkpoint"]
for col in self.columns:
table.add_column(col, key=col)
# Add rows in reverse order so newest entries are at the top
for log_entry in reversed(logs):
row_data = []
for col in self.columns:
value = log_entry.get(col, "")
# Format datetime column to be more readable
if col == "datetime" and value:
try:
# Parse ISO format and convert to readable format
dt = datetime.datetime.fromisoformat(
str(value).replace("Z", "+00:00")
)
value = dt.strftime("%Y-%m-%d %H:%M:%S")
except Exception:
# If parsing fails, just use the original value
pass
row_data.append(str(value))
table.add_row(*row_data)
status.update(
f"✅ Loaded {len(logs)} log entries from the last 72 hours"
)
logger.info(f"Loaded {len(logs)} server log entries")
else:
status.update("ℹï¸ No log entries found.")
self.all_logs = []
self.columns = []
except Exception as exc:
error_msg = f"❌ Error loading server logs: {exc}"
status.update(error_msg)
logger.error(f"Failed to load server logs: {exc}", exc_info=True)
table.clear(columns=True)
self.all_logs = []
self.columns = []
def filter_logs(self, filter_text: str) -> None:
"""Filter the logs based on the filter text with wildcard support."""
import fnmatch
table = self.query_one("#server_log_table", DataTable)
status = self.query_one("#status_bar", Static)
if not self.all_logs:
return
# Clear existing data
table.clear(columns=True)
# Re-add columns
for col in self.columns:
table.add_column(col, key=col)
# Filter logs
filtered_logs = []
if filter_text.strip():
filter_pattern = filter_text.strip().lower()
for log_entry in self.all_logs:
# Check if any field matches the filter pattern
match = False
for col in self.columns:
value = str(log_entry.get(col, "")).lower()
if fnmatch.fnmatch(value, filter_pattern):
match = True
break
if match:
filtered_logs.append(log_entry)
else:
# No filter, show all logs
filtered_logs = self.all_logs
# Add filtered rows in reverse order
for log_entry in reversed(filtered_logs):
row_data = []
for col in self.columns:
value = log_entry.get(col, "")
# Format datetime column to be more readable
if col == "datetime" and value:
try:
dt = datetime.datetime.fromisoformat(
str(value).replace("Z", "+00:00")
)
value = dt.strftime("%Y-%m-%d %H:%M:%S")
except Exception:
pass
row_data.append(str(value))
table.add_row(*row_data)
if filter_text.strip():
status.update(
f"✅ Showing {len(filtered_logs)} of {len(self.all_logs)} log entries (filtered)"
)
else:
status.update(
f"✅ Loaded {len(self.all_logs)} log entries from the last 72 hours"
)
logger.info(
f"Filtered to {len(filtered_logs)} entries with pattern: {filter_text}"
)
def on_input_changed(self, event: Input.Changed) -> None:
"""Handle filter input changes."""
if event.input.id == "filter_input":
self.filter_logs(event.value)
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
button_id = event.button.id
if button_id == "refresh_button":
self.load_logs()
# Clear the filter input when refreshing
try:
filter_input = self.query_one("#filter_input", Input)
filter_input.value = ""
except Exception:
pass
event.stop()
+567
View File
@@ -0,0 +1,567 @@
# 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 <https://www.gnu.org/licenses/>.
"""
Settings widget combining theme selection and update checking.
"""
import logging
import webbrowser
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.message import Message
from textual.widget import Widget
from textual.widgets import Button, Rule, Static
from utils.versionchecker import (
RELEASES_PAGE_URL,
UpdateCheckResult,
check_for_updates,
get_current_version,
get_version_checker,
)
logger = logging.getLogger(__name__)
class SettingsWidget(Widget):
"""Widget for application settings including themes and updates."""
DEFAULT_CSS = """
SettingsWidget {
height: 1fr;
}
/* Update section buttons - add margin between them */
#update_buttons Button {
margin-right: 1;
}
/* Theme buttons - consistent width within columns, slightly smaller */
.theme_btn {
width: 100%;
margin-bottom: 1;
}
/* Column headers */
.theme_column_header {
text-align: center;
text-style: bold;
margin-bottom: 1;
}
/* Section titles */
.settings_section_title {
text-style: bold;
margin-bottom: 1;
}
/* Theme columns - reduce overall width */
#theme_columns {
width: 80%;
}
/* Theme columns spacing */
#dark_themes_col1, #dark_themes_col2 {
margin-right: 1;
}
#light_themes_col {
margin-left: 1;
}
"""
class ThemeSelected(Message):
"""Message posted when a theme is selected."""
def __init__(self, theme_name: str):
super().__init__()
self.theme_name = theme_name
# Dark themes - Column 1
DARK_THEMES_COL1 = [
("Textual Dark", "textual-dark"),
("Nord", "nord"),
("Gruvbox", "gruvbox"),
("Dracula", "dracula"),
]
# Dark themes - Column 2
DARK_THEMES_COL2 = [
("Catppuccin Mocha", "catppuccin-mocha"),
("Tokyo Night", "tokyo-night"),
("Monokai", "monokai"),
]
# Light themes (third column)
LIGHT_THEMES = [
("Textual Light", "textual-light"),
("Flexoki", "flexoki"),
("Catppuccin Latte", "catppuccin-latte"),
("Solarized Light", "solarized-light"),
]
# Combined for backward compatibility
DARK_THEMES = DARK_THEMES_COL1 + DARK_THEMES_COL2
AVAILABLE_THEMES = DARK_THEMES + LIGHT_THEMES
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._update_result: UpdateCheckResult | None = None
self._checking = False
def compose(self):
# Wrap everything in a scrollable container with auto height children
with VerticalScroll(id="settings_scroll"):
# Version & Updates Section
with Vertical(id="updates_section") as updates:
updates.styles.height = "auto"
yield Static(
"📦 Version & Updates",
id="updates_title",
classes="settings_section_title",
)
version_text = f"Current Version: v{get_current_version()}"
yield Static(version_text, id="current_version")
with Horizontal(id="update_buttons") as btn_row:
btn_row.styles.height = "auto"
yield Button("🔍 Check for Updates", id="check_updates_btn")
yield Button("📥 View Releases", id="view_releases_btn")
yield Static("", id="update_status")
yield Rule()
# Theme Section - Three columns: Dark 1, Dark 2, Light
with Vertical(id="themes_section") as themes:
themes.styles.height = "auto"
yield Static(
"🎨 Theme Options",
id="theme_title",
classes="settings_section_title",
)
with Horizontal(id="theme_columns") as cols:
cols.styles.height = "auto"
# Dark themes section (2 columns under one header)
with Vertical(id="dark_themes_section") as dark_section:
dark_section.styles.width = "2fr"
dark_section.styles.height = "auto"
yield Static(
"🌙 Dark Themes",
classes="theme_column_header",
id="dark_header",
)
with Horizontal(id="dark_columns") as dark_cols:
dark_cols.styles.height = "auto"
# Dark themes column 1
with Vertical(id="dark_themes_col1") as dark_col1:
dark_col1.styles.width = "1fr"
dark_col1.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL1:
yield Button(
label,
id=f"set_theme_{btn_id}",
classes="theme_btn",
)
# Dark themes column 2
with Vertical(id="dark_themes_col2") as dark_col2:
dark_col2.styles.width = "1fr"
dark_col2.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL2:
yield Button(
label,
id=f"set_theme_{btn_id}",
classes="theme_btn",
)
# Light themes column
with Vertical(id="light_themes_col") as light_col:
light_col.styles.width = "1fr"
light_col.styles.height = "auto"
yield Static("☀️ Light Themes", classes="theme_column_header")
for label, btn_id in self.LIGHT_THEMES:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
def on_mount(self) -> None:
"""Check for cached update result on mount."""
checker = get_version_checker()
cached_result = checker.get_last_result()
if cached_result and cached_result.update_available:
self._update_result = cached_result
self._show_update_available(cached_result)
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id == "check_updates_btn":
self._check_for_updates()
event.stop()
elif button_id == "view_releases_btn":
self._open_releases_page()
event.stop()
elif button_id == "download_update_btn":
self._download_update()
event.stop()
elif button_id == "dismiss_update_btn":
self._dismiss_update()
event.stop()
elif button_id and button_id.startswith("set_theme_"):
theme_name = button_id.replace("set_theme_", "")
self.post_message(self.ThemeSelected(theme_name))
event.stop()
def _check_for_updates(self) -> None:
"""Check for updates and update UI."""
if self._checking:
return
self._checking = True
status = self.query_one("#update_status", Static)
check_btn = self.query_one("#check_updates_btn", Button)
# Show checking status
check_btn.disabled = True
check_btn.label = "⏳ Checking..."
status.update("🔄 Checking for updates...")
# Run check in worker to avoid blocking UI
self.run_worker(self._do_update_check, exclusive=True)
async def _do_update_check(self) -> None:
"""Worker to perform update check."""
try:
result = check_for_updates()
self._update_result = result
# Since we're in an async worker (not a thread), we can call directly
self._update_check_complete(result)
except Exception as e:
logger.error(f"Update check failed: {e}")
self._update_check_failed(str(e))
finally:
self._checking = False
def _update_check_complete(self, result: UpdateCheckResult) -> None:
"""Handle completed update check."""
check_btn = self.query_one("#check_updates_btn", Button)
check_btn.disabled = False
check_btn.label = "🔍 Check for Updates"
if result.error:
self._update_check_failed(result.error)
return
if result.update_available:
self._show_update_available(result)
self.app.notify(
f"🆕 Update available: {result.latest_version}",
title="Update Available",
severity="information",
timeout=8,
)
else:
status = self.query_one("#update_status", Static)
status.update(f"✅ Loxide is up to date (v{result.current_version})")
self.app.notify(
"✅ Loxide is up to date!",
severity="information",
timeout=5,
)
def _update_check_failed(self, error: str) -> None:
"""Handle failed update check."""
check_btn = self.query_one("#check_updates_btn", Button)
check_btn.disabled = False
check_btn.label = "🔍 Check for Updates"
status = self.query_one("#update_status", Static)
status.update(f"⚠️ Could not check for updates: {error}")
def _show_update_available(self, result: UpdateCheckResult) -> None:
"""Show update available UI with release notes."""
status = self.query_one("#update_status", Static)
msg = f"🆕 New version available: {result.latest_version}\n"
msg += f" Current: v{result.current_version}"
if result.release_info and result.release_info.body:
# Show release notes (truncate if very long)
notes = result.release_info.body.strip()
# Limit to ~500 chars to avoid overwhelming the UI
if len(notes) > 500:
notes = notes[:500] + "\n..."
msg += f"\n\n📋 Release Notes:\n{notes}"
status.update(msg)
# Add download/dismiss buttons if not already there
try:
self.query_one("#download_update_btn")
except Exception:
# Buttons don't exist, add them
button_container = self.query_one("#update_buttons", Horizontal)
download_btn = Button(
"📥 Download Update", id="download_update_btn", variant="success"
)
dismiss_btn = Button(
"✖ Dismiss", id="dismiss_update_btn", variant="default"
)
button_container.mount(download_btn)
button_container.mount(dismiss_btn)
def _open_releases_page(self) -> None:
"""Open the releases page in browser."""
try:
webbrowser.open(RELEASES_PAGE_URL)
self.app.notify("📂 Opened releases page in browser", timeout=3)
except Exception as e:
logger.error(f"Could not open browser: {e}")
self.app.notify(f"⚠️ Could not open browser: {e}", severity="warning")
def _download_update(self) -> None:
"""Download the update exe file."""
import os
from pathlib import Path
if not self._update_result or not self._update_result.release_info:
self.app.notify("⚠️ No update information available", severity="warning")
return
download_url = self._update_result.release_info.download_url
if not download_url:
# Fall back to opening the release page
url = self._update_result.release_info.html_url
try:
webbrowser.open(url)
self.app.notify(
"📥 Opened download page in browser (no direct download available)",
timeout=5,
)
except Exception as e:
logger.error(f"Could not open browser: {e}")
self.app.notify(f"⚠️ Could not open browser: {e}", severity="warning")
return
# Determine destination path
if os.name == "nt": # Windows
downloads_dir = Path.home() / "Downloads"
else:
downloads_dir = Path.home() / "Downloads"
if not downloads_dir.exists():
downloads_dir = Path.home()
# Extract filename from URL
filename = download_url.split("/")[-1]
if not filename.endswith(".exe"):
filename = f"Loxide_{self._update_result.latest_version}.exe"
dest_path = downloads_dir / filename
# Disable the button while downloading
try:
btn = self.query_one("#download_update_btn", Button)
btn.disabled = True
btn.label = "⏳ Downloading..."
except Exception:
pass
self.app.notify(
f"📥 Downloading to:\n{dest_path}", title="Download Starting", timeout=5
)
# Small delay so user sees the "downloading to" toast before download completes
self.set_timer(
0.5,
lambda: self.run_worker(
self._do_download(download_url, dest_path), exclusive=True
),
)
async def _do_download(self, download_url: str, dest_path) -> None:
"""Worker to download the update file."""
try:
# Download the file
import requests
response = requests.get(download_url, stream=True, timeout=60)
response.raise_for_status()
with open(dest_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
# Success
self.app.notify(
f"✅ Downloaded to:\n{dest_path}",
title="Download Complete",
severity="information",
timeout=10,
)
logger.info(f"Update downloaded to {dest_path}")
# Re-enable button
try:
btn = self.query_one("#download_update_btn", Button)
btn.disabled = False
btn.label = "📥 Download Again"
except Exception:
pass
except Exception as e:
logger.error(f"Download failed: {e}")
self.app.notify(f"❌ Download failed: {e}", severity="error", timeout=10)
# Re-enable button
try:
btn = self.query_one("#download_update_btn", Button)
btn.disabled = False
btn.label = "📥 Download Update"
except Exception:
pass
def _dismiss_update(self) -> None:
"""Dismiss the current update notification."""
if self._update_result and self._update_result.latest_version:
checker = get_version_checker()
checker.dismiss_update(self._update_result.latest_version)
# Remove the extra buttons
try:
self.query_one("#download_update_btn").remove()
self.query_one("#dismiss_update_btn").remove()
except Exception:
pass
status = self.query_one("#update_status", Static)
status.update(f"✓ Dismissed update {self._update_result.latest_version}")
self._update_result = None
# Keep ThemeSelector as a standalone for backward compatibility
class ThemeSelector(Widget):
"""Widget for selecting and applying Textual themes.
DEPRECATED: Use SettingsWidget instead for combined settings UI.
"""
DEFAULT_CSS = """
ThemeSelector {
height: 1fr;
}
/* Theme buttons - consistent width within columns */
.theme_btn {
width: 100%;
margin-bottom: 1;
}
/* Column headers */
.theme_column_header {
text-align: center;
text-style: bold;
margin-bottom: 1;
}
/* Theme columns - reduce overall width */
#theme_columns {
width: 80%;
}
/* Theme columns spacing */
#dark_themes_col1, #dark_themes_col2 {
margin-right: 1;
}
#light_themes_col {
margin-left: 1;
}
"""
class ThemeSelected(Message):
"""Message posted when a theme is selected."""
def __init__(self, theme_name: str):
super().__init__()
self.theme_name = theme_name
DARK_THEMES_COL1 = SettingsWidget.DARK_THEMES_COL1
DARK_THEMES_COL2 = SettingsWidget.DARK_THEMES_COL2
DARK_THEMES = SettingsWidget.DARK_THEMES
LIGHT_THEMES = SettingsWidget.LIGHT_THEMES
AVAILABLE_THEMES = SettingsWidget.AVAILABLE_THEMES
def compose(self):
with VerticalScroll(id="theme_scroll"):
yield Static("Theme Options", id="theme_title")
with Horizontal(id="theme_columns") as cols:
cols.styles.height = "auto"
# Dark themes section (2 columns under one header)
with Vertical(id="dark_themes_section") as dark_section:
dark_section.styles.width = "2fr"
dark_section.styles.height = "auto"
yield Static(
"🌙 Dark Themes",
classes="theme_column_header",
id="dark_header",
)
with Horizontal(id="dark_columns") as dark_cols:
dark_cols.styles.height = "auto"
# Dark themes column 1
with Vertical(id="dark_themes_col1") as dark_col1:
dark_col1.styles.width = "1fr"
dark_col1.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL1:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
# Dark themes column 2
with Vertical(id="dark_themes_col2") as dark_col2:
dark_col2.styles.width = "1fr"
dark_col2.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL2:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
# Light themes column
with Vertical(id="light_themes_col") as light_col:
light_col.styles.width = "1fr"
light_col.styles.height = "auto"
yield Static("☀️ Light Themes", classes="theme_column_header")
for label, btn_id in self.LIGHT_THEMES:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id and button_id.startswith("set_theme_"):
theme_name = button_id.replace("set_theme_", "")
self.post_message(self.ThemeSelected(theme_name))
+353 -587
View File
File diff suppressed because it is too large Load Diff
+12 -12
View File
@@ -1,31 +1,31 @@
[package] [package]
name = "airlock_libs" name = "airlock_libs"
version = "5.1.1" version = "7.2.0"
edition = "2024" edition = "2024"
[lib]
crate-type = ["cdylib"]
[dependencies] [dependencies]
chrono = "0.4.42" chrono = "0.4.42"
indicatif = "0.18.2" indicatif = "0.18.2"
mongodb = "3.3.0" mongodb = "3.3.0"
opentelemetry = { version = "0.18.0", features = ["rt-tokio", "metrics", "trace"] } opentelemetry = { version = "0.27.0", features = ["logs", "metrics", "trace"] }
opentelemetry-otlp = { version = "0.11.0", features = ["trace", "metrics"] } opentelemetry-otlp = { version = "0.27.0", features = ["trace", "metrics", "grpc-tonic", "http-proto", "tls", "reqwest-client", "reqwest-rustls"] }
opentelemetry-semantic-conventions = { version = "0.10.0" } opentelemetry-semantic-conventions = { version = "0.27.0" }
opentelemetry-proto = { version = "0.1.0"} opentelemetry-proto = { version = "0.27.0"}
pyo3 = { version = "0.27.0", features = ["extension-module", "generate-import-lib"] } pyo3 = { version = "0.27.0", features = ["extension-module", "generate-import-lib"] }
reqwest = { version = "0.12.24", features = ["json", "native-tls"] } reqwest = { version = "0.12.24", features = ["json", "native-tls", "rustls-tls"] }
serde = "1.0.228" serde = "1.0.228"
serde-pyobject = "0.8.0" serde-pyobject = "0.8.0"
serde_json = "1.0.145" serde_json = "1.0.145"
tokio = { version = "1.48.0", features = ["full"] } tokio = { version = "1.48.0", features = ["full"] }
tonic = { version = "0.8.2", features = ["tls-roots"] } tonic = { version = "0.12.3", features = ["tls-roots"] }
tracing = "0.1.41" tracing = "0.1.41"
tracing-subscriber = "0.3.20" tracing-subscriber = "0.3.20"
tracing-opentelemetry = "0.32.0" tracing-opentelemetry = "0.32.0"
pyo3-async-runtimes = { version = "0.27.0", features = ["async-std", "tokio"] }
crossbeam = "0.8.4" crossbeam = "0.8.4"
log = "0.4.29"
flexi_logger = "0.31.7"
opentelemetry-appender-log = "0.27.0"
opentelemetry_sdk = { version = "0.27.0", features = ["rt-tokio", "testing", "trace"] }
[package.metadata.maturin] [package.metadata.maturin]
generate-abi-stubs = true generate-abi-stubs = true
@@ -38,4 +38,4 @@ codegen-units = 1
panic = 'abort' panic = 'abort'
strip = true strip = true
debug-assertions = false debug-assertions = false
overflow-checks = false overflow-checks = true
+58 -23
View File
@@ -1,40 +1,47 @@
from typing import Dict, List from typing import Any, Dict, List, Optional
def pull_policy_exec_histories( def pull_policy_exec_histories(
self, type: List[str], checkpoint: str, policy: List[str] api,
type: str,
days: int,
policy_name: Optional[str] = None,
) -> str: ) -> str:
"""Retrieve execution history logs."""
def api(AirlockAPIWrapper):
""" """
An implementation of the python AirlockAPIWrapper class to pass Python data into Rust Pull execution history for policies.
Parameters Parameters
---------- ----------
base_url : str api : AirlockAPIWrapper
(Required) Base URL of the Airlock API, this should be in your .env file. The API wrapper instance
api_key : str policy_name : Optional[str]
(Required) API Key for your profile in airlock, this should be in your credential manager. Name of the policy to query. If None, returns history for all policies.
headers : {"X-APIKey": self.api_key} type : str
JSON-style string list of execution types, e.g., "[1,2,3]"
days : int
Number of days to look back
```def __init__(self, base_url: str, api_key: str): Returns
self.base_url = base_ur.rstrip("/") -------
self.api_key = api_key str
self.headers = {"X-APIKey": self.api_key} JSON string containing execution history response with structure:
``` {"response": {"exechistories": [...]}}
""" """
...
def history_logging( def history_logging(
api, api,
exec_types: str, exec_types: str,
checkpoint_number: str, checkpoint_number: str,
policy_names: str, policy_names: Optional[str] = None,
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
""" """
Query execution history logs from the Airlock API. Query execution history logs from the Airlock API.
Parameters Parameters
---------- ----------
api : AirlockAPIWrapper
The API wrapper instance
exec_types : str exec_types : str
A JSON-style string list of execution types to retrieve. A JSON-style string list of execution types to retrieve.
Example: "[3,5,8]" Example: "[3,5,8]"
@@ -42,17 +49,30 @@ def history_logging(
- 1 = Blocked Execution - 1 = Blocked Execution
- 2 = Untrusted Execution [Audit] - 2 = Untrusted Execution [Audit]
- 3 = Untrusted Execution [OTP] - 3 = Untrusted Execution [OTP]
- 4 = Trusted Path Execution
- 5 = Trusted Publisher Execution - 5 = Trusted Publisher Execution
- 6 = Blocklist Execution
- 7 = Blocklist Execution [Audit]
- 8 = Trusted Process Execution - 8 = Trusted Process Execution
(etc.) - 9 = Constrained Execution
- 10 = Trusted Metadata Execution
- 11 = Trusted Browser Execution
- 12 = Blocked Browser Execution
- 13 = Untrusted Browser Execution [Audit]
- 14 = Untrusted Browser Execution [OTP]
- 15 = Blocklist Browser Execution [Audit]
- 16 = Blocklist Browser Execution
- 17 = Trusted Installer Execution
- 18 = Trusted Browser Metadata Execution
checkpoint_number : str checkpoint_number : str
The checkpoint ID. Used to fetch results after a certain event. The checkpoint ID. Used to fetch results after a certain event.
Example: "601d275487bacb01e3470713" Example: "601d275487bacb01e3470713"
policy_names : str policy_names : Optional[str], default None
A comma-separated or JSON-style list of policy group names. A comma-separated or JSON-style list of policy group names.
Example: "Apple Mac" or "["Apple Mac", "Servers London"]" If None or not provided, retrieves execution history for ALL policies.
Example: "Apple Mac" or '["Apple Mac", "Servers London"]'
Returns Returns
------- -------
@@ -76,14 +96,29 @@ def history_logging(
- datetime: str - datetime: str
- ip: str - ip: str
- localip: str - localip: str
Raises Raises
------ ------
RuntimeError RuntimeError
If the request fails or the response cannot be parsed. If the request fails or the response cannot be parsed.
Example Examples
------- --------
>>> histories = await airlock_libs.history_logging("[3,5,8]", "601d275487bacb01e3470713", "Apple Mac") >>> # Get execution history for ALL policies
>>> histories = airlock_libs.history_logging(
... api,
... "[1,2,3]",
... "601d275487bacb01e3470713",
... None # or omit this parameter
... )
>>>
>>> # Get execution history for a specific policy
>>> histories = airlock_libs.history_logging(
... api,
... "[3,5,8]",
... "601d275487bacb01e3470713",
... "Apple Mac"
... )
>>> print(histories[0]["filename"]) >>> print(histories[0]["filename"])
'chrome.exe' 'chrome.exe'
""" """
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "maturin"
[project] [project]
name = "airlock_libs" name = "airlock_libs"
version = "5.1.1" version = "7.2.0"
description = "Airlock Digital API Wrapper" description = "Airlock Digital API Wrapper"
readme = "README.md" readme = "README.md"
license = { text = "AGPL-3.0-only" } license = { text = "AGPL-3.0-only" }
+46 -27
View File
@@ -8,7 +8,32 @@ pub struct TelemetryConfig {
} }
impl TelemetryConfig { impl TelemetryConfig {
pub fn load() -> Self { pub fn init_tracer() -> opentelemetry_sdk::trace::TracerProvider {
let cfg: TelemetryConfig = TelemetryConfig::load();
if !cfg.TELEMETRY {
return TracerProvider::builder().build();
}
let endpoint = cfg.TELEM_URL.unwrap_or_default();
let channel = Channel::from_shared(endpoint.clone())
.unwrap()
.tls_config(ClientTlsConfig::new().with_native_roots())
.unwrap()
.connect_lazy();
let exporter = opentelemetry_otlp::SpanExporter::builder()
.with_tonic()
.with_endpoint(endpoint.clone())
.with_channel(channel)
.build()
.expect("Failed to build exporter");
opentelemetry_sdk::trace::TracerProvider::builder()
.with_simple_exporter(exporter)
.with_resource(Resource::new(vec![KeyValue::new(
"service.name",
"LoxideLibs",
)]))
.build()
}
fn load() -> Self {
let cfg_path = get_base_directory().join("config\\user_config.json"); let cfg_path = get_base_directory().join("config\\user_config.json");
if !cfg_path.exists() { if !cfg_path.exists() {
return Self { return Self {
@@ -64,36 +89,30 @@ pub struct Group {
pub(crate) localip: String, pub(crate) localip: String,
} }
pub enum ExtractedValues { pub struct PyData {
Headers(reqwest::header::HeaderMap), pub headers: reqwest::header::HeaderMap,
BaseUrl(String), pub base_url: String,
} }
pub trait Converter { impl PyData {
fn convert(py: Python<'_>, py_self: &Py<PyAny>, extract_headers: bool) -> ExtractedValues; pub fn extract_data(py: Python<'_>, obj: &Py<PyAny>) -> Self {
} let headers_raw = obj.getattr(py, "headers").unwrap().to_string();
let headers_json = headers_raw.replace('\'', "\"");
pub struct PyData; let parsed: Value = serde_json::from_str(&headers_json).unwrap();
let mut header_map = HeaderMap::new();
impl Converter for PyData { if let Some(obj) = parsed.as_object() {
fn convert(py: Python<'_>, py_self: &Py<PyAny>, extract_headers: bool) -> ExtractedValues { for (key, val) in obj {
if extract_headers { if let Some(v) = val.as_str() {
let headers = py_self.getattr(py, "headers").unwrap().to_string(); let header_name = HeaderName::from_str(key).unwrap();
let headers_replace = headers.replace('\'', "\""); let header_value: HeaderValue = HeaderValue::from_str(v).unwrap();
let parsed: Value = serde_json::from_str(headers_replace.as_str()).unwrap(); header_map.insert(header_name, header_value);
let mut header_map = HeaderMap::new();
if let Some(obj) = parsed.as_object() {
for (_key, value) in obj {
if let Some(v) = value.as_str() {
let val = HeaderValue::from_str(v).unwrap();
header_map.insert(HeaderName::from_str("X-APIKey").unwrap(), val);
}
} }
} }
ExtractedValues::Headers(header_map) }
} else { let base_url = obj.getattr(py, "base_url").unwrap().to_string();
let base_url = py_self.getattr(py, "base_url").unwrap().to_string(); Self {
ExtractedValues::BaseUrl(base_url) headers: header_map,
base_url,
} }
} }
} }
View File
+11 -6
View File
@@ -1,21 +1,25 @@
pub use chrono::{Duration, Local, NaiveDate}; pub use chrono::{Duration, Local, NaiveDate};
pub use crossbeam::channel::unbounded;
pub use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle}; pub use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle};
pub use mongodb::bson::oid::ObjectId; pub use mongodb::bson::oid::ObjectId;
pub use opentelemetry::global::shutdown_tracer_provider; pub use opentelemetry::global::GlobalTracerProvider;
pub use opentelemetry::sdk::Resource;
pub use opentelemetry::trace::noop::NoopTracerProvider; pub use opentelemetry::trace::noop::NoopTracerProvider;
pub use opentelemetry::trace::{Status, TraceContextExt, TraceError}; pub use opentelemetry::trace::{Status, TraceContextExt, Tracer};
pub use opentelemetry::{Context, KeyValue, sdk::trace as sdktrace, trace::Tracer}; pub use opentelemetry::*;
pub use opentelemetry::{Key, global}; pub use opentelemetry_otlp::ExportConfig;
pub use opentelemetry_otlp::WithExportConfig; pub use opentelemetry_otlp::WithExportConfig;
pub use opentelemetry_otlp::WithTonicConfig;
pub use opentelemetry_sdk::Resource;
pub use opentelemetry_sdk::trace::{Config, TracerProvider};
pub use pyo3::{prelude::*, types::PyString}; pub use pyo3::{prelude::*, types::PyString};
pub use pyo3_async_runtimes::async_std;
pub use reqwest::{ pub use reqwest::{
Client, Client,
header::{HeaderMap, HeaderName, HeaderValue}, header::{HeaderMap, HeaderName, HeaderValue},
}; };
pub use serde::{Deserialize, Serialize}; pub use serde::{Deserialize, Serialize};
pub use serde_json::Value; pub use serde_json::Value;
pub use std::sync::{Arc, Mutex};
pub use std::thread;
pub use std::{ pub use std::{
collections::HashMap, collections::HashMap,
env, env,
@@ -25,3 +29,4 @@ pub use std::{
path::PathBuf, path::PathBuf,
str::FromStr, str::FromStr,
}; };
pub use tonic::transport::{Channel, ClientTlsConfig};
+82 -81
View File
@@ -1,38 +1,31 @@
use std::thread;
use crossbeam::channel::unbounded;
use crate::modules::datatypes::*; use crate::modules::datatypes::*;
use crate::prelude::*; use crate::prelude::*;
#[pyfunction] #[pyfunction]
pub fn pull_policy_exec_histories( pub fn pull_policy_exec_histories(
py: Python<'_>, py: Python<'_>,
py_self: Py<PyAny>, py_self: Py<PyAny>,
policy_names: String, policy_names: Option<String>,
exec_types: String, exec_types: String,
days: i64, days: i64,
) -> Py<PyString> { ) -> Py<PyString> {
let headers: HeaderMap = match PyData::convert(py, &py_self, true) { println!();
ExtractedValues::Headers(h) => h, let data: PyData = PyData::extract_data(py, &py_self);
ExtractedValues::BaseUrl(_) => std::process::abort(), let headers: HeaderMap = data.headers;
}; let base_url: String = data.base_url;
let base_url = match PyData::convert(py, &py_self, false) { let handle: thread::JoinHandle<String> = std::thread::spawn(move || {
ExtractedValues::Headers(_) => std::process::abort(), let rt: tokio::runtime::Runtime = match tokio::runtime::Runtime::new() {
ExtractedValues::BaseUrl(b) => b,
};
let handle = std::thread::spawn(move || {
let rt = match tokio::runtime::Runtime::new() {
Ok(rt) => rt, Ok(rt) => rt,
Err(e) => { Err(e) => {
println!("Failed to build Tokio Runtime: {:?}", e); println!("Failed to build Tokio Runtime: {:?}", e);
std::process::abort(); std::process::abort();
} }
}; };
rt.block_on(async { let tracer_provider = rt.block_on(async { TelemetryConfig::init_tracer() });
let _ = init_tracer(); global::set_tracer_provider(tracer_provider.clone());
}); let tracer: global::BoxedTracer = global::tracer("tracer");
let tracer = global::tracer("global_tracer"); let _cx: Context = Context::new();
let _cx = Context::new();
let file_path: PathBuf = format!( let file_path: PathBuf = format!(
"{}\\cache\\chunkinator.json", "{}\\cache\\chunkinator.json",
get_base_directory().display() get_base_directory().display()
@@ -58,14 +51,14 @@ pub fn pull_policy_exec_histories(
} }
} }
} }
let data = ApiResponse { let data: ApiResponse = ApiResponse {
error: "Success".to_string(), error: "Success".to_string(),
response: ExecHistories { response: ExecHistories {
exechistories: vec![], exechistories: vec![],
}, },
}; };
let writeable_filepath = file_path.clone(); let writeable_filepath: PathBuf = file_path.clone();
let data_write = serde_json::to_string_pretty(&data).expect("Failed to serialize"); let data_write: String = serde_json::to_string_pretty(&data).expect("Failed to serialize");
match fs::write(writeable_filepath.clone(), data_write) { match fs::write(writeable_filepath.clone(), data_write) {
Ok(_) => {} Ok(_) => {}
Err(e) => { Err(e) => {
@@ -74,17 +67,18 @@ pub fn pull_policy_exec_histories(
} }
} }
let mut checkpoint_number: String = SkipBack::find_checkpoint(days).to_string(); let mut checkpoint_number: String = SkipBack::find_checkpoint(days).to_string();
let multi_progress = MultiProgress::new(); let progress_bar = Arc::new(Mutex::new(ProgressBar::new(100)));
multi_progress.set_draw_target(ProgressDrawTarget::stderr()); progress_bar
let progress_bar = multi_progress.add(ProgressBar::new(100)); .lock()
progress_bar.set_style( .unwrap()
.set_draw_target(ProgressDrawTarget::stderr());
progress_bar.lock().unwrap().set_style(
ProgressStyle::default_bar() ProgressStyle::default_bar()
.template("Total Completion: {spinner:.green} [{elapsed_precise}] [{bar:40.green/blue}] {pos}/{len} {message}") .template("Total - Policy Name: {msg}: {spinner:.green} [{elapsed_precise}] [{bar:40.green/blue}] {pos}/{len}")
.unwrap(), .unwrap().progress_chars("⣿⣦⣀")
); );
progress_bar.enable_steady_tick(std::time::Duration::from_millis(100)); let client: Client = tracer.in_span("Building HTTP Client", |cx| {
let client = tracer.in_span("Building HTTP Client", |cx| { let client_result: Result<Client, reqwest::Error> = build_client(headers);
let client_result = build_client(headers);
match client_result { match client_result {
Ok(client_result) => { Ok(client_result) => {
cx.span().add_event( cx.span().add_event(
@@ -111,12 +105,14 @@ pub fn pull_policy_exec_histories(
} }
} }
}); });
let cutoff = Local::now().naive_local() - Duration::days(days); let cutoff: chrono::NaiveDateTime =
Local::now().naive_local() - chrono::Duration::days(days);
let (tx, rx) = unbounded::<Vec<Group>>(); let (tx, rx) = unbounded::<Vec<Group>>();
let pb_clone = progress_bar.clone();
thread::spawn(move || { thread::spawn(move || {
let mut seen: HashMap<(String, String, String), Group> = if writeable_filepath.exists() let mut seen: HashMap<(String, String, String), Group> = if writeable_filepath.exists()
{ {
let contents = fs::read_to_string(&writeable_filepath).unwrap_or_default(); let contents: String = fs::read_to_string(&writeable_filepath).unwrap_or_default();
let existing: ApiResponse = let existing: ApiResponse =
serde_json::from_str(&contents).unwrap_or(ApiResponse { serde_json::from_str(&contents).unwrap_or(ApiResponse {
error: "Success".to_string(), error: "Success".to_string(),
@@ -128,7 +124,7 @@ pub fn pull_policy_exec_histories(
.response .response
.exechistories .exechistories
.into_iter() .into_iter()
.map(|entry| { .map(|entry: Group| {
( (
( (
entry.sha256.clone(), entry.sha256.clone(),
@@ -147,7 +143,7 @@ pub fn pull_policy_exec_histories(
if executions.checkpoint.is_empty() || executions.datetime.is_empty() { if executions.checkpoint.is_empty() || executions.datetime.is_empty() {
continue; continue;
} }
let history_date = match NaiveDate::parse_from_str( let history_date: NaiveDate = match NaiveDate::parse_from_str(
&executions.datetime.replace(" +0000 UTC", ""), &executions.datetime.replace(" +0000 UTC", ""),
"%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%SZ",
) { ) {
@@ -155,7 +151,7 @@ pub fn pull_policy_exec_histories(
Err(_) => continue, Err(_) => continue,
}; };
if history_date >= cutoff.into() { if history_date >= cutoff.into() {
let key = ( let key: (String, String, String) = (
executions.sha256.clone(), executions.sha256.clone(),
executions.filename.clone(), executions.filename.clone(),
executions.hostname.clone(), executions.hostname.clone(),
@@ -163,13 +159,13 @@ pub fn pull_policy_exec_histories(
seen.entry(key).or_insert(executions.clone()); seen.entry(key).or_insert(executions.clone());
} }
} }
let final_response = ApiResponse { let final_response: ApiResponse = ApiResponse {
error: "Success".to_string(), error: "Success".to_string(),
response: ExecHistories { response: ExecHistories {
exechistories: seen.values().cloned().collect(), exechistories: seen.values().cloned().collect(),
}, },
}; };
let data_write = serde_json::to_string_pretty(&final_response).unwrap(); let data_write: String = serde_json::to_string_pretty(&final_response).unwrap();
match fs::write(&writeable_filepath, data_write) { match fs::write(&writeable_filepath, data_write) {
Ok(_) => {} Ok(_) => {}
Err(e) => { Err(e) => {
@@ -180,9 +176,22 @@ pub fn pull_policy_exec_histories(
}); });
let mut first_date: Option<NaiveDate> = None; let mut first_date: Option<NaiveDate> = None;
tracer.in_span("Airlock Data Retreival", |cx| { tracer.in_span("Airlock Data Retreival", |cx| {
let span = cx.span(); pb_clone
span.set_attribute(Key::new("Days").string(days.to_string())); .lock()
span.set_attribute(KeyValue::new("Policy Name", policy_names.clone())); .unwrap()
.enable_steady_tick(std::time::Duration::from_millis(100));
pb_clone
.lock()
.unwrap()
.set_message(policy_names.clone().unwrap_or("Statistics".to_string()));
let span: opentelemetry::trace::SpanRef<'_> = cx.span();
span.set_attribute(KeyValue::new("Days", days.to_string()));
span.set_attribute(KeyValue::new(
"Policy Name",
policy_names
.clone()
.unwrap_or("Statistics Monitoring".to_string()),
));
loop { loop {
let execution_histories = tracer.in_span(checkpoint_number.to_string(), |cx| { let execution_histories = tracer.in_span(checkpoint_number.to_string(), |cx| {
let results: ApiResponse = history_logging( let results: ApiResponse = history_logging(
@@ -198,7 +207,7 @@ pub fn pull_policy_exec_histories(
)); ));
results results
}); });
let parsed_responses = execution_histories.response.exechistories; let parsed_responses: Vec<Group> = execution_histories.response.exechistories;
if parsed_responses.is_empty() { if parsed_responses.is_empty() {
break; break;
} }
@@ -214,30 +223,37 @@ pub fn pull_policy_exec_histories(
first_date = Some(last_date); first_date = Some(last_date);
} }
if let Some(base_date) = first_date { if let Some(base_date) = first_date {
let date_diff = last_date - base_date; let date_diff: chrono::TimeDelta = last_date - base_date;
let total_span = (Local::now().naive_local().date() - base_date).num_days(); let total_span: i64 =
let percentage = ((date_diff.num_days() as f64 / total_span as f64) * 100.0) (Local::now().naive_local().date() - base_date).num_days();
let percentage: u64 = ((date_diff.num_days() as f64 / total_span as f64)
* 100.0)
.clamp(0.0, 100.0) .clamp(0.0, 100.0)
.round() as u64; .round() as u64;
progress_bar.set_position(percentage); pb_clone.lock().unwrap().set_position(percentage);
} }
} }
} }
}); });
progress_bar.finish_with_message("All Checkpoints Complete"); progress_bar
let return_data = match fs::read_to_string(file_path.clone()) { .lock()
.unwrap()
.finish_with_message("All Checkpoints Complete");
let return_data: String = match fs::read_to_string(file_path.clone()) {
Ok(return_data) => return_data, Ok(return_data) => return_data,
Err(e) => { Err(e) => {
println!("Failed to read data from: {:?}: {}", &file_path, e); println!("Failed to read data from: {:?}: {}", &file_path, e);
std::process::abort(); std::process::abort();
} }
}; };
tracer_provider
.shutdown()
.expect("Failed to Shutdown Tracer Provdier");
drop(tx); drop(tx);
shutdown_tracer_provider();
return_data.to_string() return_data.to_string()
}); });
let gil_value = handle.join().unwrap(); let gil_value: String = handle.join().unwrap();
Python::attach(|py| PyString::new(py, &gil_value).into()) Python::attach(|py: Python<'_>| PyString::new(py, &gil_value).into())
} }
fn build_client(headers: HeaderMap) -> Result<reqwest::Client, reqwest::Error> { fn build_client(headers: HeaderMap) -> Result<reqwest::Client, reqwest::Error> {
@@ -253,20 +269,24 @@ async fn history_logging(
base_url: &String, base_url: &String,
exec_types: &String, exec_types: &String,
checkpoint_number: &String, checkpoint_number: &String,
policy_names: &String, policy_names: &Option<String>,
client: &Client, client: &Client,
) -> ApiResponse { ) -> ApiResponse {
let policy_json = match policy_names {
Some(name) => format!(r#"[ "{}" ]"#, name), // JSON array with one element
None => "[]".to_string(), // Empty JSON array
};
let payload = format!( let payload = format!(
r#"{{ r#"{{
"type": {}, "type": {},
"checkpoint": "{}", "checkpoint": "{}",
"policy": ["{}"] "policy": {}
}}"#, }}"#,
exec_types, checkpoint_number, policy_names exec_types, checkpoint_number, policy_json
); );
let res = client let res: Result<reqwest::Response, reqwest::Error> = client
.post(format!("{}/v1/logging/exechistories", base_url)) .post(format!("{}/v1/logging/exechistories", base_url))
.body(payload) .json(&payload)
.send() .send()
.await; .await;
match res { match res {
@@ -300,29 +320,10 @@ pub fn get_base_directory() -> PathBuf {
.unwrap_or_else(|| home.join("AppData").join("Roaming")); .unwrap_or_else(|| home.join("AppData").join("Roaming"));
appdata.join("Loxide") appdata.join("Loxide")
} }
_ => home.join(".local").join("share").join("Loxide"), "linux" => home.join(".local").join("share").join("Loxide"),
_ => {
println!("{} is currently not compatible with LoxideLibs", os);
std::process::abort();
}
} }
} }
fn init_tracer() -> Result<Option<sdktrace::Tracer>, TraceError> {
let cfg = TelemetryConfig::load();
if !cfg.TELEMETRY {
global::set_tracer_provider(NoopTracerProvider::new());
return Ok(None);
}
let endpoint = cfg.TELEM_URL.unwrap_or_default();
let tracer =
opentelemetry_otlp::new_pipeline()
.tracing()
.with_exporter(
opentelemetry_otlp::new_exporter()
.tonic()
.with_endpoint(endpoint),
)
.with_trace_config(sdktrace::config().with_resource(Resource::new(vec![
KeyValue::new("service.name", "LoxideLibs"),
])))
.install_simple()
.unwrap();
Ok(Some(tracer))
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+476
View File
@@ -0,0 +1,476 @@
# Loxide
## API Reference Document
**Version 1.0 | December 2025**
---
## 1. Overview
This document provides a complete reference for the AirlockAPIWrapper class and all Airlock API endpoints used by Loxide.
### 1.1 Authentication
All API requests require an API key passed via the X-APIKey header:
```
Headers: { "X-APIKey": "your-api-key-here" }
```
### 1.2 Base URL
The base URL is configured in `system_config.json` and typically follows the pattern:
```
https://airlock.example.com/api
```
---
## 2. Agent Management
### 2.1 agent_find_all()
Retrieve all registered agents.
**Endpoint:** `POST /v1/agent/find`
**Payload:** `{}`
**Returns:** DataFrame with agent records
| Field | Type | Description |
|-------|------|-------------|
| `hostname` | str | Device hostname |
| `agentid` | str | Unique identifier |
| `clientversion` | str | Agent version |
| `groupid` | str | Policy group ID |
| `status` | int | 0=Offline, 1=Online, 2=Hidden, 3=Safemode |
| `lastcheckin` | str | Last check-in timestamp |
| `ip` | str | External IP address |
| `localip` | str | Internal IP address |
| `domain` | str | Network domain |
| `os` | str | Operating system |
| `username` | str | Logged-in user |
| `freespace` | int | Available disk space |
| `policyversion` | str | Active policy version |
### 2.2 agent_find_by_hostname(hostname: str)
Find agents matching a hostname pattern.
**Endpoint:** `POST /v1/agent/find`
**Payload:** `{ "hostname": "<pattern>" }`
**Returns:** DataFrame with matching agents
### 2.3 agent_find_by_id(agentid: str)
Find agent by unique ID.
**Endpoint:** `POST /v1/agent/find`
**Payload:** `{ "agentid": "<id>" }`
**Returns:** DataFrame with agent record
### 2.4 agent_find_by_status(status: int)
Find agents by status code.
**Endpoint:** `POST /v1/agent/find`
**Payload:** `{ "status": <code> }`
**Status Codes:**
- `0` - Offline
- `1` - Online
- `2` - Hidden
- `3` - Safemode
### 2.5 agent_move(agentid: str, groupid: str)
Move an agent to a different policy group.
**Endpoint:** `POST /v1/agent/move`
**Payload:** `{ "agentid": "<id>", "groupid": "<target_group>" }`
**Returns:** dict with operation result
### 2.6 agents_find_by_group(groupid: str)
Find all agents in a policy group.
**Endpoint:** `POST /v1/agent/find`
**Payload:** `{ "groupid": "<id>" }`
**Returns:** DataFrame with agents
---
## 3. Policy Management
### 3.1 policy_find_all()
Retrieve all policy groups.
**Endpoint:** `POST /v1/group`
**Payload:** `{}`
**Returns:** DataFrame with policy records
| Field | Type | Description |
|-------|------|-------------|
| `name` | str | Policy display name |
| `groupid` | int | Unique group identifier |
| `hidden` | bool | Visibility flag |
| `parent` | str | Parent policy name (if child) |
### 3.2 policy_set_auditmode(groupid: str, auditmode: str)
Toggle policy between audit and enforcement modes.
**Endpoint:** `POST /v1/group/settings/auditmode`
**Payload:** `{ "groupid": "<id>", "auditmode": "<mode>" }`
**Mode Values:**
- `"1"` - Audit mode (log only)
- `"0"` - Enforcement mode (block)
### 3.3 policy_list_agents(groupid: str)
List all agents assigned to a policy group.
**Endpoint:** `POST /v1/group/agents`
**Payload:** `{ "groupid": "<id>" }`
**Returns:** DataFrame with agents
### 3.4 policy_list_allowlists(groupid: str)
List allowlists assigned to a policy group.
**Endpoint:** `POST /v1/group/policies`
**Payload:** `{ "groupid": "<id>" }`
**Returns:** DataFrame with applications
### 3.5 policy_clone(source_groupid: str, target_groupid: str)
Clone a policy from one group to another.
**Endpoint:** `POST /v1/group/assign`
**Payload:** `{ "groupid": "<source>", "targetgroupid": "<target>" }`
### 3.6 policy_add_path_exclusions(groupid: str, paths: List[str])
Add path exclusions to a policy group.
**Endpoint:** `POST /v1/group/path/add`
**Payload:** `{ "groupid": "<id>", "path": ["<path1>", "<path2>"] }`
### 3.7 policy_add_publishers(groupid: str, publishers: List[str])
Add trusted publishers to a policy group.
**Endpoint:** `POST /v1/group/publisher/add`
**Payload:** `{ "groupid": "<id>", "publisher": ["<pub1>", "<pub2>"] }`
---
## 4. OTP Management
### 4.1 otp_generate(agentid: str, duration: int, purpose: str)
Generate a new One-Time Password for an agent.
**Endpoint:** `POST /v1/otp/retrieve`
**Payload:**
```json
{
"agentid": "<id>",
"duration": "<minutes>",
"purpose": "<description>"
}
```
**Returns:** str - The generated OTP code
### 4.2 otp_find_active()
Retrieve all active OTP sessions.
**Endpoint:** `POST /v1/otp/usage`
**Payload:** `{ "status": "1" }`
**Returns:** DataFrame with OTP records
| Field | Type | Description |
|-------|------|-------------|
| `otpid` | str | OTP session identifier |
| `agentid` | str | Associated agent ID |
| `hostname` | str | Agent hostname |
| `purpose` | str | OTP purpose description |
| `granted` | str | Grant timestamp |
| `expires` | str | Expiration timestamp |
### 4.3 otp_find_awaiting()
Retrieve OTPs awaiting activation.
**Endpoint:** `POST /v1/otp/usage`
**Payload:** `{ "status": "0" }`
### 4.4 otp_find_enforced()
Retrieve enforced OTPs.
**Endpoint:** `POST /v1/otp/usage`
**Payload:** `{ "status": "2" }`
### 4.5 otp_find_revoked()
Retrieve revoked OTPs.
**Endpoint:** `POST /v1/otp/usage`
**Payload:** `{ "status": "3" }`
### 4.6 otp_find_by_agent(agentid: str)
Retrieve OTPs for a specific agent.
**Endpoint:** `POST /v1/otp/usage`
**Payload:** `{ "agentid": "<id>" }`
### 4.7 otp_revoke(otpid: str)
Revoke an active OTP session.
**Endpoint:** `POST /v1/otp/revoke`
**Payload:** `{ "otpid": "<id>" }`
**Returns:** dict with operation result
### 4.8 otp_validate(otpcode: str)
Validate an OTP code.
**Endpoint:** `POST /v1/otp/validate`
**Payload:** `{ "otpcode": "<code>" }`
**Returns:** dict indicating validity
### 4.9 otp_get_activities(otpid: str)
Retrieve activity log for a specific OTP.
**Endpoint:** `POST /v1/otp/activities`
**Payload:** `{ "otpid": "<id>" }`
**Returns:** DataFrame with OTP activities
---
## 5. Execution History
### 5.1 history_execution(today: str, date_selected: str, agent_name: str)
Retrieve execution history for a specific agent.
**Endpoint:** `POST /v1/getexechistory`
**Payload:**
```json
{
"datefrom": "<YYYY-MM-DD>",
"dateto": "<YYYY-MM-DD>",
"hostname": "<name>"
}
```
**Returns:** List[Dict] with execution records
| Field | Type | Description |
|-------|------|-------------|
| `type` | int | Execution type code |
| `hostname` | str | Device hostname |
| `username` | str | User who executed |
| `filename` | str | Executed filename |
| `sha256` | str | File hash |
| `publisher` | str | Code signer |
| `datetime` | str | Execution timestamp |
| `policyname` | str | Active policy |
| `policyver` | str | Policy version |
| `commandline` | str | Full command line |
| `pprocess` | str | Parent process |
### 5.2 Execution Type Codes
| Code | Description |
|------|-------------|
| 0 | Trusted Execution |
| 1 | Blocked Execution |
| 2 | Untrusted Execution [Audit] |
| 3 | Untrusted Execution [OTP] |
| 4 | Trusted Path Execution |
| 5 | Trusted Publisher Execution |
| 6 | Blocklist Execution |
| 7 | Blocklist Execution [Audit] |
| 8 | Trusted Process Execution |
| 9 | Constrained Execution |
| 10 | Trusted Metadata Execution |
| 11 | Trusted Browser Execution |
| 12 | Blocked Browser Execution |
| 13 | Untrusted Browser Execution [Audit] |
| 14 | Untrusted Browser Execution [OTP] |
| 15 | Blocklist Browser Execution [Audit] |
| 16 | Blocklist Browser Execution |
| 17 | Trusted Installer Execution |
| 18 | Trusted Browser Metadata Execution |
### 5.3 history_logging(type: List[str], checkpoint: str, policy: Optional[List[str]])
Retrieve execution history logs with pagination.
**Endpoint:** `POST /v1/logging/exechistories`
**Payload:**
```json
{
"type": ["1", "2", "3"],
"checkpoint": "<checkpoint_id>",
"policy": ["<policy_name>"]
}
```
**Returns:** str with execution histories
---
## 6. Server Logs
### 6.1 server_logs(checkpoint: Optional[str])
Retrieve server activity logs.
**Endpoint:** `POST /v1/logging/svractivities`
**Payload:** `{}` or `{ "checkpoint": "<last_checkpoint>" }`
**Returns:** str with server activities
The checkpoint parameter enables pagination for large result sets. Pass the last checkpoint from a previous call to get subsequent records.
---
## 7. Allowlist and Blocklist Management
### 7.1 allowlist_find_all()
Retrieve all allowlist applications.
**Endpoint:** `POST /v1/application`
**Payload:** `{}`
**Returns:** DataFrame with application records
### 7.2 allowlist_export(applicationid: str)
Export allowlist as XML.
**Endpoint:** `POST /v1/application/export`
**Payload:** `{ "applicationid": "<id>" }`
**Returns:** bytes - XML content
### 7.3 baseline_find_all()
Retrieve all baselines.
**Endpoint:** `POST /v1/baseline`
**Payload:** `{}`
**Returns:** DataFrame with baseline records
### 7.4 baseline_export(baselineid: str)
Export baseline as XML.
**Endpoint:** `POST /v1/baseline/export`
**Payload:** `{ "baselineid": "<id>" }`
**Returns:** bytes - XML content
### 7.5 blocklist_find_all()
Retrieve all blocklists.
**Endpoint:** `POST /v1/blocklist`
**Payload:** `{}`
**Returns:** DataFrame with blocklist records
### 7.6 blocklist_export(blocklistid: str)
Export blocklist as XML.
**Endpoint:** `POST /v1/blocklist/export`
**Payload:** `{ "blocklistid": "<id>" }`
**Returns:** bytes - XML content
### 7.7 hash_add_to_allowlist(applicationid: str, hashes: List[str])
Add hashes to an allowlist.
**Endpoint:** `POST /v1/hash/application/add`
**Payload:** `{ "applicationid": "<id>", "hashes": ["<sha256>", ...] }`
**Returns:** dict with operation result
### 7.8 hash_query(hashes: List[str])
Query information about specific hashes.
**Endpoint:** `POST /v1/hash/query`
**Payload:** `{ "hashes": ["<sha256>", ...] }`
**Returns:** DataFrame with hash records
| Field | Type | Description |
|-------|------|-------------|
| `sha256` | str | Hash value |
| `filename` | str | Associated filename |
| `publisher` | str | Code signer (or "Not Signed") |
| `reputation` | dict | VirusTotal scan results |
| `applications` | str | Associated allowlists |
| `baselines` | str | Associated baselines |
| `blocklists` | str | Associated blocklists |
---
## 8. Rust Backend (airlock_libs)
The `airlock_libs` package provides Rust-accelerated functions for performance-critical operations.
### 8.1 pull_policy_exec_histories(api, type, days, policy_name)
Pull execution history for policies with optimized performance.
| Parameter | Type | Description |
|-----------|------|-------------|
| `api` | AirlockAPIWrapper | API wrapper instance |
| `type` | str | JSON list of exec types, e.g., "[1,2,3]" |
| `days` | int | Days to look back |
| `policy_name` | Optional[str] | Specific policy or None for all |
**Returns:** JSON string with execution history
### 8.2 history_logging(api, exec_types, checkpoint_number, policy_names)
Query execution logs with pagination support.
| Parameter | Type | Description |
|-----------|------|-------------|
| `api` | AirlockAPIWrapper | API wrapper instance |
| `exec_types` | str | JSON list of exec types, e.g., "[3,5,8]" |
| `checkpoint_number` | str | Checkpoint ID for pagination |
| `policy_names` | Optional[str] | Comma-separated policy names or None |
**Returns:** List[Dict] - Execution history records
---
## 9. Error Handling
All API methods may raise:
- `requests.exceptions.RequestException` - Network or HTTP errors
- `ValueError` - Invalid response format
- `KeyError` - Missing expected fields in response
Recommended pattern:
```python
try:
result = api.agent_find_all()
except requests.exceptions.RequestException as e:
logger.error(f"API request failed: {e}")
# Handle error appropriately
```
---
## 10. License
Copyright (C) 2025 James Brotosky, Brandon Wickline
GNU Affero General Public License v3.0
+510
View File
@@ -0,0 +1,510 @@
# Loxide
## User Stories and Use Cases
**Version 1.0 | December 2025**
---
## Epic 1: Multi-Agent Operations
**Epic Statement:** As a administrator, I need to perform bulk operations on multiple endpoints efficiently.
---
### US-1.1: Agent Selection
**User Story:** As a administrator, I want to select multiple agents using various methods so that I can perform bulk operations efficiently.
**Acceptance Criteria:**
- ✅ Can paste a list of device names (700+ lines)
- ✅ Can use wildcards (* and ?) for pattern matching
- ✅ Can import device names from a file
- ✅ Can toggle between exact and fuzzy matching
- ✅ Unmatched entries are clearly displayed
- ✅ Can select/deselect all matched agents
**Priority:** High
**Story Points:** 8
---
### US-1.2: Agent Policy Move
**User Story:** As a administrator, I want to move selected agents to a different policy group so that I can organize endpoints by security requirements.
**Acceptance Criteria:**
- ✅ Can select destination policy from list
- ✅ Move operation provides progress feedback
- ✅ Success/failure results are color-coded
- ✅ Can export results to CSV
**Priority:** High
**Story Points:** 5
---
### US-1.3: Toggle Enforcement Mode
**User Story:** As a administrator, I want to toggle agents between audit and enforcement mode so that I can gradually roll out policy enforcement.
**Acceptance Criteria:**
- ✅ Clear indication of current mode
- ✅ Confirmation before mode change
- ✅ Results displayed after operation
**Priority:** High
**Story Points:** 3
---
### US-1.4: View Agent Execution History
**User Story:** As a administrator, I want to view execution history for selected agents so that I can understand what applications are running.
**Acceptance Criteria:**
- ✅ Date range selector (1-365 days)
- ✅ Results displayed in DataTable
- ✅ Export to CSV functionality
- ✅ Sortable columns
**Priority:** Medium
**Story Points:** 5
---
## Epic 2: Policy Preparation
**Epic Statement:** As a administrator, I need to prepare policies for enforcement by analyzing execution history.
---
### US-2.1: Policy Selection
**User Story:** As a administrator, I want to select source policies for analysis so that I can review their execution history.
**Acceptance Criteria:**
- ✅ Multi-select with checkboxes
- ✅ Filter/search capability
- ✅ Policy hierarchy visible
**Priority:** High
**Story Points:** 5
---
### US-2.2: Configure Analysis Parameters
**User Story:** As a administrator, I want to configure history days and allowlists so that I can customize the analysis scope.
**Acceptance Criteria:**
- ✅ History days configurable from 1 to 365
- ✅ Default value clearly indicated
- ✅ Allowlist selection available
- ✅ Input validation with helpful error messages
**Priority:** High
**Story Points:** 3
---
### US-2.3: Review Analysis Results
**User Story:** As a administrator, I want to review categorized execution history so that I can make informed enforcement decisions.
**Acceptance Criteria:**
- ✅ Results categorized: Approved, Unapproved, Needs Review
- ✅ Hash reputation data displayed
- ✅ Publisher information shown
- ✅ Export to CSV available
- ✅ Color-coded categories
**Priority:** High
**Story Points:** 8
---
### US-2.4: Add Hashes to Allowlist
**User Story:** As a administrator, I want to add approved hashes to an allowlist so that they won't be blocked after enforcement.
**Acceptance Criteria:**
- ✅ Select hashes from analysis results
- ✅ Choose target allowlist
- ✅ Confirmation before adding
- ✅ Success/failure feedback
**Priority:** Medium
**Story Points:** 5
---
## Epic 3: Quiet Agent Detection
**Epic Statement:** As a administrator, I need to identify inactive endpoints ready for enforcement.
---
### US-3.1: Configure Quiet Threshold
**User Story:** As a administrator, I want to set quiet day thresholds so that I can define what constitutes an inactive agent.
**Acceptance Criteria:**
- ✅ Configurable quiet days parameter
- ✅ History days parameter
- ✅ Clear explanation of thresholds
- ✅ Input validation
**Priority:** High
**Story Points:** 3
---
### US-3.2: Run Quiet Agent Analysis
**User Story:** As a administrator, I want the system to analyze agent activity using high-performance Rust code so that I can quickly identify quiet agents.
**Acceptance Criteria:**
- ✅ Progress bar during analysis
- ✅ Console output for Rust progress
- ✅ Non-blocking UI during analysis
- ✅ Clear completion notification
**Priority:** High
**Story Points:** 8
---
### US-3.3: View Quiet Agents
**User Story:** As a administrator, I want to see a list of quiet agents so that I can move them to enforcement.
**Acceptance Criteria:**
- ✅ Results displayed in sortable table
- ✅ Can select agents for bulk move
- ✅ Last activity date shown
- ✅ Export to CSV
**Priority:** High
**Story Points:** 5
---
## Epic 4: OTP Management
**Epic Statement:** As an support technician, I need to manage temporary policy bypasses for end users.
---
### US-4.1: Generate OTP
**User Story:** As an support technician, I want to generate an OTP for an agent so that a user can temporarily bypass policy restrictions.
**Acceptance Criteria:**
- ✅ Select agent by hostname
- ✅ Specify duration in minutes
- ✅ Enter purpose/ticket number
- ✅ OTP code displayed clearly
- ✅ Copy to clipboard functionality
**Priority:** High
**Story Points:** 5
---
### US-4.2: View Active OTPs
**User Story:** As a administrator, I want to view all active OTP sessions so that I can monitor temporary policy bypasses.
**Acceptance Criteria:**
- ✅ List all active OTPs
- ✅ Show hostname, purpose, expiration
- ✅ Filter by status
- ✅ Refresh capability (r key)
- ✅ Sortable columns
**Priority:** High
**Story Points:** 5
---
### US-4.3: Revoke OTP
**User Story:** As a administrator, I want to revoke an active OTP so that I can end a temporary bypass immediately.
**Acceptance Criteria:**
- ✅ Select OTPs for revocation (checkbox)
- ✅ Bulk revoke capability
- ✅ Confirmation before revocation
- ✅ Results displayed after operation
- ✅ Auto-refresh list after revocation
**Priority:** High
**Story Points:** 5
---
### US-4.4: View OTP Activities
**User Story:** As a administrator, I want to see what applications were executed during an OTP session so that I can audit temporary bypasses.
**Acceptance Criteria:**
- ✅ Select OTP to view
- ✅ Display execution list
- ✅ Show file, hash, timestamp
- ✅ Export capability
**Priority:** Medium
**Story Points:** 5
---
## Epic 5: Execution History
**Epic Statement:** As a analyst, I need to investigate execution events on endpoints.
---
### US-5.1: Query Execution History
**User Story:** As a analyst, I want to query execution history for specific agents so that I can investigate security events.
**Acceptance Criteria:**
- ✅ Select agent by hostname
- ✅ Configure date range (start/end)
- ✅ Filter by execution type
- ✅ Results in sortable DataTable
- ✅ Pagination for large results
**Priority:** High
**Story Points:** 8
---
### US-5.2: Export History
**User Story:** As a analyst, I want to export execution history to CSV so that I can perform offline analysis.
**Acceptance Criteria:**
- ✅ Export button available (e key)
- ✅ All visible columns included
- ✅ Proper CSV formatting
- ✅ Timestamp in filename
- ✅ Notification on success
**Priority:** Medium
**Story Points:** 3
---
### US-5.3: Hash Reputation Lookup
**User Story:** As a analyst, I want to see reputation data for executed files so that I can assess risk.
**Acceptance Criteria:**
- ✅ VirusTotal score displayed
- ✅ Publisher information shown
- ✅ Known allowlist membership indicated
- ✅ Risk categorization (approved/unapproved/needs_review)
**Priority:** Medium
**Story Points:** 5
---
## Epic 6: Server Monitoring
**Epic Statement:** As a system administrator, I need to monitor Airlock server activity.
---
### US-6.1: View Server Logs
**User Story:** As a system administrator, I want to view recent server activity so that I can monitor system health.
**Acceptance Criteria:**
- ✅ Default view of last 72 hours
- ✅ Datetime properly formatted (YYYY-MM-DD HH:MM:SS)
- ✅ Refresh via keyboard shortcut (r)
- ✅ Escape to return to main menu
- ✅ Auto-scroll to latest entries
**Priority:** Medium
**Story Points:** 5
---
## Epic 7: Application Configuration
**Epic Statement:** As a power user, I need to customize the application to my preferences.
---
### US-7.1: Theme Selection
**User Story:** As a power user, I want to change the UI theme so that I can work comfortably in different lighting conditions.
**Acceptance Criteria:**
- ✅ Multiple themes available (textual-dark, gruvbox, retro-terminal, amber-terminal)
- ✅ Theme persists across sessions
- ✅ Preview before applying
- ✅ Saved to user config
**Priority:** Low
**Story Points:** 3
---
### US-7.2: Working Directory Access
**User Story:** As a power user, I want to access my working directory from within the application so that I can manage exported files.
**Acceptance Criteria:**
- ✅ Directory tree visible in main menu
- ✅ Keyboard shortcut to open in file manager (f)
- ✅ Configurable working directory
- ✅ Auto-create directory structure
**Priority:** Low
**Story Points:** 3
---
### US-7.3: Credential Management
**User Story:** As a user, I want my API credentials stored securely so that I don't have to enter them every time.
**Acceptance Criteria:**
- ✅ First-time setup prompts for API key
- ✅ Master password protects credentials
- ✅ Password complexity requirements enforced
- ✅ 3 retry attempts on wrong password
- ✅ Platform-native keyring used
**Priority:** High
**Story Points:** 8
---
## Epic 8: Statistics and Reporting
**Epic Statement:** As a administrator, I need visibility into my environment's security posture.
---
### US-8.1: View System Statistics
**User Story:** As a administrator, I want to see an overview of agents and policies so that I can understand my environment.
**Acceptance Criteria:**
- ✅ Total agent count
- ✅ Total policy count
- ✅ Agent status breakdown (Online/Offline/Hidden/Safemode)
- ✅ Visual charts using plotext
- ✅ Configurable time range (1/7/30 days)
**Priority:** Medium
**Story Points:** 5
---
### US-8.2: View Execution Statistics
**User Story:** As a administrator, I want to see execution statistics so that I can identify trends.
**Acceptance Criteria:**
- ✅ Execution counts by type
- ✅ Top executed files
- ✅ Top blocked files
- ✅ Visual bar charts
- ✅ Refresh capability
**Priority:** Medium
**Story Points:** 5
---
## Phase 2: LEMON Integration
**Project:** LEMON (Loxide Execution MONitoring)
**Previous Codename:** Overlock
**Status:** Phase 2 - Separate Project
LEMON user stories are documented separately in `LEMON_03_User_Stories.md`.
**Summary of Phase 2 Loxide Stories:**
- LEMON Sessions Screen - View/create/cancel sessions
- LEMON Hash Review Screen - Approve/reject pending hashes
- Certificate Setup - Configure mTLS authentication
- Audit Chain Verification - Verify log integrity
See LEMON documentation for complete user stories.
---
## Acceptance Test Scenarios
### Scenario: Bulk Agent Move
```gherkin
Given I am on the Multi-Agent Operations screen
When I paste 100 device names into the selector
And I click Search
Then I should see matched agents in the selection list
When I select 50 agents using checkboxes
And I choose "Production Policy" as destination
And I click "Move Selected"
Then I should see a progress indicator
And I should see 50 success results with green indicators
And I should be able to export results to CSV
```
### Scenario: OTP Generation and Revocation
```gherkin
Given I am on the OTP Management screen
When I search for agent "DESKTOP-001"
And I set duration to 60 minutes
And I enter purpose "Ticket #12345 - Software installation"
And I click Generate OTP
Then I should see an 8-character OTP code
And I should be able to copy it to clipboard
Given the OTP is active
When I navigate to OTP Revoke screen
And I select the OTP for "DESKTOP-001"
And I click "Revoke Selected"
Then I should see confirmation dialog
When I confirm revocation
Then I should see success message
And the OTP should no longer appear in active list
```
### Scenario: Policy Preparation Workflow
```gherkin
Given I am on the Policy Prep screen
When I select "Audit Policy A" and "Audit Policy B"
And I set history days to 30
And I click Next
Then I should see execution history being fetched
When the fetch completes
Then I should see categorized results:
| Category | Count |
| Approved | 150 |
| Unapproved | 25 |
| Needs Review | 10 |
When I click "Export to CSV"
Then I should see a file saved notification
And the CSV should contain all execution records
```
---
## License
Copyright (C) 2025 James Brotosky, Brandon Wickline
GNU Affero General Public License v3.0
+356 -35
View File
@@ -1,59 +1,380 @@
# Loxide
# 🛡️ Loxide A Python/Rust toolkit for secure, auditable, and automated Airlock agent and policy management. Designed for enterprise environments, Loxide provides advanced policy workflows, device tracking, and a modern terminal-based interface for security administrators who need bulk operations and streamlined workflows not available in the standard Airlock console.
Python/Rust/Oxide toolkit for secure, auditable, and automated airlock agent and policy management. Designed for enterprise environments, it supports advanced policy workflows, device tracking, and terminal-based interaction. ![Python](https://img.shields.io/badge/python-3.10+-blue.svg)
![Rust](https://img.shields.io/badge/rust-backend-orange.svg)
![License](https://img.shields.io/badge/license-AGPL--3.0-green.svg)
![Textual](https://img.shields.io/badge/textual-6.5.0-purple.svg)
--- ---
## Table of Contents
## 🚀 Features - [Features](#features)
- 🔍 **Fuzzy Device Search** - [Installation](#installation)
Quickly locate devices using partial or approximate matches. - [Configuration](#configuration)
- [Usage](#usage)
- 📦 **Batch Move Devices** - [Workflows](#workflows)
Move multiple devices between groups or policies easily. - [Architecture](#architecture)
- [Keyboard Reference](#keyboard-reference)
- 🔄 **Toggle Enforcement/Audit Policies** - [Roadmap](#roadmap)
Seamlessly switch devices between enforcement and audit modes. - [Authors](#authors)
- [License](#license)
- 🕵️‍♂️ **Device History Search**
Track agent executions.
- 🧰 **Prepare Policies for Enforcement**
Validate and stage policies before pushing them to enforcement.
- 💤 **Find Quiet Hosts**
Identify devices ready for enforcement.
- 🎛️ **TUI**
Navigate with arrow keys and F-key shortcuts using a custom ANSI-colored terminal UI.
--- ---
## 🧭 Roadmap ## Features
- ⚙️ **Rust-based Async API Calls** ### Multi-Agent Operations
Improve performance and concurrency with a Rust-powered backend.
- ✅ **Carbon Black-style Local Approval** Select and operate on multiple endpoints simultaneously using hostname, wildcard patterns (`*`, `?`), or file import. Supports lists of 700+ devices with fuzzy matching.
Enable local user approvals for policy exceptions and enforcement actions.
- 📊 **Audit Logging & Export** - Move agents between policy groups in bulk
Add detailed logging and export capabilities for compliance and analysis. - Toggle between audit and enforcement modes
- Generate OTPs for multiple agents at once
- View execution history across selected devices
- Export operation results to CSV
### Policy Preparation Workflow
A guided multi-step wizard for safely transitioning policies from audit to enforcement:
1. Select source policies for analysis (multi-select with checkboxes)
2. Choose destination policy and allowlist
3. Configure analysis parameters (1-365 day history window)
4. Fetch and categorize execution history data
5. Review unapproved applications by category
6. Add approved hashes to allowlists
7. Export analysis results for documentation
### Quiet Agent Detection
Identify inactive endpoints that are ready for enforcement using high-performance Rust-powered analysis:
- Configure quiet day threshold and history parameters
- Analyze agent activity patterns efficiently
- Review quiet agent list with policy information
- Bulk move quiet agents to enforcement policies
### OTP Management
Complete One-Time Password lifecycle management for handling application blocking issues:
- Generate OTPs with configurable duration and purpose
- Monitor active OTP sessions in real-time
- View applications executed during OTP windows
- Revoke active OTPs when needed
- Analyze historical OTP activity
### Execution History Analysis
Detailed analysis of application execution events across your environment:
- Date range selection (configurable 1-365 days)
- Filter by agent, policy, filename, publisher, or hash
- Categorize executions by approval status (18 execution types)
- VirusTotal reputation data via Airlock's integration
- Export results to CSV for external analysis
### Server Activity Monitoring
Real-time visibility into Airlock server operations:
- DataTable display of server activity logs
- Configurable time window (default 72 hours)
- Auto-refresh capability
- Search and filter functionality
### Modern Terminal Interface
Built with Textual 6.5.0 for a responsive, keyboard-driven experience:
- Full keyboard navigation with discoverable shortcuts
- Tab-based workflow organization
- Toast notifications for operations
- Color-coded status indicators
- Multiple theme options (Gruvbox, Amber Terminal, Retro Terminal)
--- ---
## 🧑‍💻 Requirements ## Installation
[airlock_libs](https://git.racooncity.org/brotoskyj/-/packages/pypi/airlock-libs/) ### Prerequisites
- Python 3.10 or higher
- Access to Airlock API server
- Network access to private PyPI server (for airlock_libs)
- Modern terminal with Unicode support (80x24 minimum, 120x40 recommended)
### Quick Start
```bash
# Clone the repository
git clone https://git.racooncity.org/brotoskyj/loxide.git
cd loxide
# Install dependencies
pip install -r requirements.txt
# Run the application
python Loxide.py
```
### Dependencies
```
# Core TUI Framework
textual==6.5.0
# API and Data Handling
requests==2.32.5
pandas==2.3.3
numpy==2.3.4
# Database
pymongo==4.15.3
# Security
cryptography==46.0.3
keyring==25.6.0
# Utilities
python-dotenv==1.2.1
tqdm==4.67.1
urllib3==2.5.0
pyperclip==1.11.0
# Private Package (Rust integration)
--extra-index-url https://git.racooncity.org/api/packages/brotoskyj/pypi/simple/
airlock_libs==5.2.1
```
See also: [airlock_libs documentation](https://git.racooncity.org/brotoskyj/-/packages/pypi/airlock-libs/)
--- ---
## 📜 License ## Configuration
### First Run
On first launch, Loxide will prompt for your Airlock API credentials. These are securely stored using your platform's native keyring (Windows Credential Manager, macOS Keychain, or Linux Secret Service) with an additional encryption layer.
### Configuration Files
Loxide uses two configuration files:
| File | Location | Purpose |
|------|----------|---------|
| `system_config.json` | Application directory | System defaults, bundled with application |
| `user_config.json` | User home directory | User preferences, created on first run |
### Environment Variables
| Variable | Description |
|----------|-------------|
| `LOXIDE_LOG_LEVEL` | Logging verbosity (DEBUG, INFO, WARNING, ERROR) |
| `HTTPS_PROXY` | Proxy server for Airlock API connections |
---
## Usage
```bash
python Loxide.py
```
### Main Interface
The application opens to a tabbed interface with the following sections:
| Tab | Description |
|-----|-------------|
| Dashboard | Summary statistics and environment overview |
| Agent Operations | Multi-agent selection and bulk operations |
| Policy Prep | Policy enforcement preparation wizard |
| Quiet Agents | Inactive agent detection workflow |
| OTP Management | OTP generation, monitoring, and revocation |
### User Classes
Loxide supports three primary user roles:
- **Administrator** — Full access to all functionality, policy management, bulk operations
- **Analyst** — Monitoring, investigation, and reporting (execution history, OTP activity, logs)
- **Support** — Targeted operations like OTP generation for end-user issues
---
## Workflows
### Moving Agents to a New Policy
1. Navigate to **Agent Operations** tab
2. Enter device names (paste list, use wildcards, or load from file)
3. Click **Search** to find matching agents
4. Select agents using checkboxes (or Select All)
5. Click **Move to Policy**
6. Select destination policy from the tree
7. Confirm the operation
### Preparing a Policy for Enforcement
1. Navigate to **Policy Prep** tab
2. Select source policies to analyze
3. Choose destination enforcement policy
4. Set history days (how far back to analyze)
5. Select target allowlist for approved hashes
6. Click **Fetch Data** to retrieve execution history
7. Review unapproved applications by category
8. Add approved hashes to the allowlist
9. Export results for documentation
### Generating an OTP
1. Navigate to **Agent Operations** tab
2. Search for and select the target agent
3. Click **Generate OTP**
4. Set duration (in minutes) and purpose
5. Copy the generated OTP code
---
## Architecture
```
Loxide/
├── Loxide.py # Main application entry point
├── API.py # AirlockAPIWrapper - all API calls
├── setup.py # Initialization, logging, config loading
├── configmanager.py # System/user configuration management
├── models/ # Data models
│ ├── agent.py # Agent data model
│ ├── policy.py # Policy data model
│ └── execution.py # Execution event model
├── services/ # Business logic
│ └── security.py # Credential management
├── utils/ # Helper functions
│ └── utils.py # Utility functions
└── TUI/
├── Screens/ # Full-page workflow screens
│ ├── policyprepworkflowscreen.py
│ ├── quietagentworkflowscreen.py
│ ├── moveagentworkflowscreen.py
│ ├── executionhistoryscreen.py
│ ├── otpworkflowscreen.py
│ ├── otpactivityscreen.py
│ ├── otprevokescreen.py
│ └── allowlistselectionscreen.py
├── Widgets/ # Reusable UI components
│ ├── policyselector.py
│ ├── policytreewidget.py
│ ├── multiagentselector.py
│ ├── agentmoveoperations.py
│ ├── serverlogwidget.py
│ ├── resultsdisplay.py
│ └── themeselector.py
└── Themes/ # Color themes
├── theme_amber_terminal.py
└── theme_retro_terminal.py
```
### Design Principles
- **Layered Architecture** — Clear separation between presentation, business logic, data access, and infrastructure
- **Screen vs Widget Pattern** — Screens have headers/footers and are pushed; widgets are embedded components
- **Secure by Default** — API keys encrypted at rest, secure credential storage
- **Cross-Platform** — Works on Windows 10/11 and Linux (Ubuntu 24+)
---
## Keyboard Reference
### Global
| Key | Action |
|-----|--------|
| `q` | Quit application |
| `Tab` | Next widget |
| `Shift+Tab` | Previous widget |
| `←` `→` | Navigate tabs |
| `?` | Show help |
### Screens
| Key | Action |
|-----|--------|
| `Escape` | Go back / Cancel |
| `r` | Refresh data |
| `e` | Export to CSV |
### Data Tables
| Key | Action |
|-----|--------|
| `↑` `↓` | Move cursor |
| `Page Up/Down` | Scroll page |
| `Home` / `End` | First / last row |
| `Space` | Toggle selection |
| `Enter` | Select / activate |
| `a` | Select all |
| `n` | Deselect all |
---
## Roadmap
### Planned Features
- **Carbon Black-style Local Approval** — Enable local user approvals for policy exceptions and enforcement actions
- **Audit Logging & Export** — Detailed logging and export capabilities for compliance and analysis
- **LEMON Integration** — Phase 2 integration with LEMON for enhanced analytics
- **Offline Mode** — Cached data review when Airlock server is unreachable
---
## Performance
| Metric | Target |
|--------|--------|
| Application startup | < 5 seconds |
| Agent list load (1000 agents) | < 10 seconds |
| Execution history (30 days, single agent) | < 15 seconds |
| UI response time | < 100 milliseconds |
| CSV export (10,000 rows) | < 30 seconds |
| Memory usage (idle) | < 200 MB |
---
## Authors
- **Brandon Wickline** — Lead Python Developer
- **James Brotosky** — Lead Rust Developer
---
## License
**Loxide** is licensed under the **GNU Affero General Public License v3.0**. **Loxide** is licensed under the **GNU Affero General Public License v3.0**.
You may copy, distribute, and modify the software under the terms of the AGPL-3.0 license. You may copy, distribute, and modify the software under the terms of the AGPL-3.0 license.
See the [LICENSE](LICENSE.md) file for full details, or visit ```
[https://www.gnu.org/license/agpl-3.0.html](https://www.gnu.org/license/agpl-3.0.html) 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.
```
See the [LICENSE](LICENSE.md) file for full details, or visit [gnu.org/licenses/agpl-3.0.html](https://www.gnu.org/licenses/agpl-3.0.html)
-252
View File
@@ -1,252 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
import logging
import os
import time
from typing import List, Optional
from models.agent import Agent
from services.agenthandler import moveAgentToRelatedPolicy, selectAgents
from services.API import AirlockAPIWrapper
from utils.configmanager import get_system_json
from utils.utils import colorText, get_sanitized_input
logger = logging.getLogger(__name__)
class LocalApprovalRequestor:
"""Handles creation of local approval requests in Loxide."""
def __init__(self, api: AirlockAPIWrapper, username: str = None):
"""
Initialize the local approval requestor.
Args:
api: AirlockAPIWrapper instance
username: Username creating the approvals (for tracking)
"""
self.api = api
self.policy_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
self.username = (
username or os.getenv("USERNAME") or os.getenv("USER") or "unknown"
)
def create_local_approval(
self, agent_id: str, duration_minutes: int, batch_id: Optional[int] = None
) -> bool:
"""
Create a single local approval request.
Args:
agent_id: Agent ID to create approval for
duration_minutes: Duration of approval in minutes
batch_id: Optional batch identifier (defaults to timestamp)
Returns:
True if successful, False otherwise
"""
if batch_id is None:
batch_id = int(time.time())
purpose = (
f" Local Approval - {duration_minutes} mins - "
f"batch:{batch_id} Client:{agent_id} User:{self.username}"
)
try:
self.api.otp_generate(agent_id, duration_minutes, purpose)
logger.info(
f"Generated local approval for {agent_id}, batch {batch_id}, by {self.username}"
)
return True
except Exception as e:
logger.error(f"Failed to generate local approval for {agent_id}: {e}")
return False
def move_agent_to_audit(self, agent: Agent) -> bool:
"""
Move an agent to its corresponding audit policy.
Args:
agent: Agent object to move
Returns:
True if successful, False otherwise
"""
try:
moveAgentToRelatedPolicy(self.api, agent, "audit")
logger.info(f"Moved {agent.hostname} to audit policy")
return True
except Exception as e:
logger.error(f"Failed to move {agent.hostname} to audit: {e}")
return False
def create_local_approval_batch(
self,
agents: List[Agent],
duration_minutes: int,
) -> tuple[int, int, int]:
"""
Create local approvals for multiple agents and move them to audit.
Args:
agents: List of Agent objects
duration_minutes: Duration of approval in minutes
db_path: Optional path to database for history tracking
Returns:
Tuple of (batch_id, success_count, failure_count)
"""
batch_id = int(time.time())
success_count = 0
failure_count = 0
print(colorText(f"\n Processing batch {batch_id}...", "cyan"))
print(colorText(f" Requested by: {self.username}", "cyan"))
print(colorText(f" Moving {len(agents)} agent(s) to local approval\n", "cyan"))
for agent in agents:
try:
# Create local approval
approval_success = self.create_local_approval(
agent.agentid, duration_minutes, batch_id
)
if not approval_success:
raise Exception("Failed to create local approval")
# Move to audit policy
move_success = self.move_agent_to_audit(agent)
if not move_success:
raise Exception("Failed to move to audit policy")
print(colorText(f" {agent.hostname}", "green"))
success_count += 1
except Exception as e:
print(colorText(f" {agent.hostname}: {e}", "red"))
logger.error(f"Error processing agent {agent.hostname}: {e}")
failure_count += 1
return batch_id, success_count, failure_count
def interactive_local_approval(self):
"""
Interactive workflow to create local approvals for selected agents.
This prompts the user to select a duration and agents, then creates
the local approvals and moves agents to audit policies.
"""
# Duration options in minutes
duration_options = [
(15, "15 minutes"),
(60, "1 hour"),
(360, "6 hours"),
(1440, "1 day"),
(10080, "1 week"),
]
# Display duration options
print(colorText("\n Select Local Approval Duration:", "white"))
print(colorText("=" * 50, "white"))
for i, (minutes, label) in enumerate(duration_options, start=1):
print(f" {i}. {label} ({minutes} minutes)")
print(colorText("=" * 50, "white"))
# Get user selection
try:
choice = int(get_sanitized_input("\nEnter the number of your choice: "))
if 1 <= choice <= len(duration_options):
duration_minutes, duration_label = duration_options[choice - 1]
print(colorText(f" Selected: {duration_label}", "green"))
logger.info(f"User selected duration: {duration_minutes} minutes")
else:
print(colorText("❌ Invalid choice.", "red"))
logger.warning("Invalid duration choice")
return
except ValueError:
print(colorText("❌ Invalid input. Please enter a number.", "red"))
logger.warning("Invalid input for duration selection")
return
# Select agents
print(colorText("\nSelect Agents for Local Approval:", "white"))
agents = selectAgents(self.api)
if not agents:
print(colorText("❌ No agents found or error retrieving agents.", "red"))
logger.warning("No agents selected or error retrieving agents")
return
# Confirm with user
print(colorText("\nSummary:", "cyan"))
print(colorText(f" Duration: {duration_label}", "white"))
print(colorText(f" Agents: {len(agents)}", "white"))
confirm = get_sanitized_input("\nProceed? (y/n): ").lower()
if confirm != "y":
print(colorText("❌ Operation cancelled.", "yellow"))
return
# Process the batch
batch_id, success_count, failure_count = self.create_local_approval_batch(
agents, duration_minutes
)
# Display summary
self._display_summary(batch_id, duration_label, success_count, failure_count)
def _display_summary(
self, batch_id: int, duration_label: str, success_count: int, failure_count: int
):
"""
Display operation summary.
Args:
batch_id: Batch identifier
duration_label: Human-readable duration
success_count: Number of successful operations
failure_count: Number of failed operations
"""
print(colorText(f"\n{'=' * 60}", "white"))
print(colorText(" Local Approval Summary", "cyan"))
print(colorText("=" * 60, "white"))
print(colorText(f" Successfully processed: {success_count}", "green"))
if failure_count > 0:
print(colorText(f" Failed: {failure_count}", "red"))
print(colorText(f"\n Batch ID: {batch_id}", "cyan"))
print(colorText(f" Duration: {duration_label}", "cyan"))
print(colorText("=" * 60, "white"))
print(colorText("\n Next Steps:", "yellow"))
print(colorText(" ✅ Agents have been moved to audit policies", "white"))
print(colorText(" ✅ Local approvals are active", "white"))
print(
colorText(
f" ✅ Agents will return to enforcement after {duration_label}",
"white",
)
)
print(colorText("=" * 60 + "\n", "white"))
-69
View File
@@ -1,69 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
import logging
import pandas as pd
from services.agenthandler import selectAgents
from services.API import AirlockAPIWrapper
from utils.selector import Selector
from utils.utils import get_sanitized_input
logger = logging.getLogger(__name__)
def otp_revoke(api: AirlockAPIWrapper):
activeagents = api.otp_find_active()
awaitingagents = api.otp_find_awaiting()
activeagents["status"] = "active"
awaitingagents["status"] = "awaiting"
combined_agents = pd.concat([activeagents, awaitingagents], ignore_index=True)
combined_agents = combined_agents.sort_values(by="otpid", ascending=False)
# Combine all into one DataFrame
combined_agents = pd.concat([activeagents, awaitingagents], ignore_index=True)
combined_agents = combined_agents.sort_values(by="otpid", ascending=False)
# Optionally, select specific hosts
user_input = (
get_sanitized_input("\nWould you like to search for a specific device? (y/n): ")
.strip()
.lower()
)
if user_input == "y":
agentnames = []
agents = selectAgents(api)
for agent in agents:
agentnames.append(agent.hostname)
combined_agents = combined_agents[combined_agents["hostname"].isin(agentnames)]
# Present and select rows
selected_rows = Selector.select_dataframe_with_mode(
combined_agents,
columns=["otpid", "hostname", "status", "purpose", "granted"],
header="OTP Sessions",
)
for row in selected_rows:
otpid = row["otpid"]
hostname = row["hostname"]
result = api.otp_revoke(otpid)
logger.info(f"{hostname} (otpid: {otpid}):\n{result}")
-870
View File
@@ -1,870 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
import logging
import os
import os.path
import re
from typing import List
import dotenv
import pandas as pd
from models.execution import ExecutionHistoryRecord
from models.policy import Allowlist, Policy
from services.API import AirlockAPIWrapper
from utils.configmanager import get_system_list, get_system_value, load_env
from utils.selector import Selector
from utils.utils import (
areYouSure,
clear_screen,
colorText,
formatHTML,
get_sanitized_input,
locked,
open_directory,
print_x_wide,
regulator,
)
logger = logging.getLogger(__name__)
dotenv.load_dotenv()
def selectPolicies(api: AirlockAPIWrapper, allow_multiple=True) -> List[Policy]:
policies = [Policy(**row.to_dict()) for _, row in api.policy_find_all().iterrows()]
logger.debug("Prompting for Policies")
print(colorText("Please select policy/policies", "white"))
selected = Selector.select_objects(policies, allow_multiple, prompt_each=True)
if selected is None:
return []
# Normalize to always return a list
logger.debug("Returning {selected.dict}")
return selected if isinstance(selected, list) else [selected]
def selectAllowlists(
api: AirlockAPIWrapper, policy=all, allow_multiple=True
) -> List[Allowlist]:
if policy == "all":
allowlists = [
Allowlist(**row.to_dict()) for _, row in api.allowlist_find_all().iterrows()
]
else:
allowlists = [
Allowlist(**row.to_dict())
for _, row in api.policy_list_allowlists(policy[0].groupid).iterrows()
]
logger.debug("Prompting for Allowlist(s)")
print(colorText("Please select allowlist(s)", "white"))
selected = Selector.select_objects(allowlists, allow_multiple, prompt_each=True)
if selected is None:
return []
# Normalize to always return a list
logger.debug(f"Returning {selected}")
return selected if isinstance(selected, list) else [selected]
def sortHashes(
api: AirlockAPIWrapper, selected_policies: List[Policy], type=[1, 2, 6, 7]
):
working_dir = load_env("WORKING_DIR")
history_days = Selector.select_value(
prompt="Enter how many days of history to pull (1-150): ",
value_type=int,
valid_range=(1, 150),
)
logger.debug(f"{history_days} day selected for history")
if history_days is None:
logging.warning("No history range selected. Aborting.")
return
policy_executions = ExecutionHistoryRecord.from_policies(
api, selected_policies, type_=type, history_days=history_days
)
logger.debug(f"Executions contains {policy_executions}")
enriched_executions = ExecutionHistoryRecord.enrich_with_hashes(
api, policy_executions
)
categorized_executions = (
ExecutionHistoryRecord.categorize_executions_by_hash_decision(
enriched_executions
)
)
approved, unapproved, needs_review, unknown = (
ExecutionHistoryRecord.sort_by_hash_decision(categorized_executions)
)
categories = {
"needs_review": needs_review,
"approved": approved,
"unapproved": unapproved,
"leftover": unknown,
}
for label, records in categories.items():
if not records:
continue # Skip empty or falsy categories
csv_path = f"{working_dir}\\Needs_Review\\Review_First\\{selected_policies[0].name}_{label}_executions.csv"
html_path = f"{working_dir}\\Needs_Review\\HTML\\{selected_policies[0].name}_{label}.html"
# Convert ExecutionHistoryRecord objects to dictionaries
df = pd.DataFrame([r.__dict__ for r in records])
# Optional: flatten hash_obj if needed
if not df.empty and "hash_obj" in df.columns:
hash_df = df["hash_obj"].apply(lambda h: h.to_dict() if h else {})
df = pd.concat([df.drop(columns=["hash_obj"]), hash_df], axis=1)
# Save to CSV
df.to_csv(csv_path, index=False)
logger.info(f"Saved {label} executions to {csv_path}")
# Generate HTML
formatHTML(df, html_path)
logger.info(f"Generated HTML report at {html_path}")
def buildPathsandPublishers(selected_policies: List[Policy], split):
working_dir = load_env("WORKING_DIR")
df1 = pd.DataFrame()
df2 = pd.DataFrame()
all_approved_hashes = pd.DataFrame()
path1 = (
f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv"
)
path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_needs_review_executions.csv"
path_exclusion_constant = get_system_value("PATH_EXCLUSION_CONST", cast_type=int)
if os.path.exists(path1):
df1 = pd.read_csv(path1)
else:
logger.warning(f"File not found: {path1}")
if os.path.exists(path2):
df2 = pd.read_csv(path2)
else:
logger.warning(f"File not found: {path2}")
if df1.empty and df2.empty:
logger.warning("Both DataFrames are empty. Skipping sort.")
all_approved_hashes = pd.DataFrame()
logger.debug(all_approved_hashes.head)
else:
all_approved_hashes = pd.concat([df1, df2], ignore_index=True)
if "filename" in all_approved_hashes.columns:
all_approved_hashes = all_approved_hashes.sort_values(by="filename")
else:
logger.warning(
"Warning: 'filename' column not found in concatenated DataFrame."
)
if not all_approved_hashes.empty and path_exclusion_constant:
primary_path_exclusions = calculatePath(
all_approved_hashes,
path_exclusion_constant,
split,
)
remaining_hashes = all_approved_hashes[
~all_approved_hashes["sha256"].isin(primary_path_exclusions["sha256"])
]
secondary_path_exclusions = calculatePath(
remaining_hashes, (path_exclusion_constant - 1), split
)
remaining_hashes = remaining_hashes[
~remaining_hashes["sha256"].isin(secondary_path_exclusions["sha256"])
]
dataframes = {
"all_approved_hashes": all_approved_hashes,
"primary_Paths": primary_path_exclusions,
"secondary_Paths": secondary_path_exclusions,
"hashes_not_approvable_by_path": remaining_hashes,
}
logger.debug("Preparing to sort dataframes")
for name, df in dataframes.items():
logger.debug(f" DataFrame headers: {list(df.columns)}")
if "hashes" in name:
df.sort_values(by="filename", inplace=True)
else:
df.sort_values(by="longestcfp", inplace=True)
df.to_csv(
f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_{name}.csv",
index=False,
)
formatHTML(
df,
f"{working_dir}\\Needs_Review\\HTML\\{selected_policies[0].name}_{name}.html",
)
if not all_approved_hashes.empty:
# Drop all not signed, only keep unique values
publist = all_approved_hashes[
all_approved_hashes["publisher"] != "Not Signed"
].drop_duplicates(subset=["publisher"])
# Remove Bad publisher if somehow they made it this far
pattern = regulator(get_system_list("BAD_PUBLISHERS"))
publist = publist[~publist["publisher"].str.contains(pattern, na=False)]
publist = publist[["publisher"]]
publist.sort_values(by="publisher", inplace=True)
publist.to_csv(
f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_publishers.csv",
index=False,
)
else:
logger.debug("Approved Hashes list appears empty")
def buildPreflights(selected_policies: List[Policy]):
working_dir = load_env("WORKING_DIR")
df1 = pd.DataFrame()
df2 = pd.DataFrame()
approved_hashes = pd.DataFrame()
approved_publishers = pd.DataFrame()
hash = f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_all_approved_hashes.csv"
path1 = f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv"
path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_secondary_Paths.csv"
publishers = f"{working_dir}\\Approved\\{selected_policies[0].name}_publishers.csv"
# Read in and combine the two path generations
if os.path.exists(path1):
df1 = pd.read_csv(path1)
else:
logger.warning(f"File not found: {path1}")
if os.path.exists(path2):
df2 = pd.read_csv(path2)
else:
logger.warning(f"File not found: {path2}")
if df1.empty and df2.empty:
logger.warning("Both DataFrames are empty. Skipping sort.")
approved_paths = pd.DataFrame()
else:
approved_paths = pd.concat([df1, df2], ignore_index=True)
approved_paths = approved_paths.drop_duplicates(subset="longestcfp", keep="first")
# We create a list of hashes that are left over if we exclude the ones that are covered by the path exclusions.
if os.path.exists(hash):
hashes = pd.read_csv(hash)
approved_hashes = hashes[~hashes["filename"].isin(approved_paths["longestcfp"])]
approved_hashes = approved_hashes.drop_duplicates(subset="sha256", keep="first")
else:
logger.warning(f"File not found: {hash}")
if os.path.exists(publishers):
approved_publishers = pd.read_csv(publishers)
else:
logger.warning(f"File not found: {publishers}")
dataframes = {
"approved_paths": approved_paths,
"approved_hashes": approved_hashes,
"approved_publishers": approved_publishers,
}
for name, df in dataframes.items():
logger.debug(f" DataFrame headers: {list(df.columns)}")
if name == "approved_paths":
df.sort_values(by="longestcfp", inplace=True)
elif name == "approved_hashes":
df.sort_values(by="filename", inplace=True)
elif name == "approved_publishers":
df.sort_values(by="publisher", inplace=True)
df.to_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_{name}.csv",
index=False,
)
formatHTML(
df,
f"{working_dir}\\Preflight\\HTML\\{selected_policies[0].name}_{name}.html",
)
def splitFilepathsGrouped(df, path_exclusion_constant, col="filename"):
min_files_for_path = get_system_value("MIN_FILES_FOR_PATH", cast_type=int)
def clean_split(path):
if not isinstance(path, (str, bytes, os.PathLike)):
return []
parts = str(os.path.normpath(path)).split(os.sep)
parts = [p for p in parts if p] # Remove empty strings
return parts
# Diagnostic: log any non-string entries
non_string_entries = df[
~df[col].apply(lambda x: isinstance(x, (str, bytes, os.PathLike)))
]
if not non_string_entries.empty:
print(f"[WARNING] Non-string entries found in column '{col}':")
print(non_string_entries)
df = df.copy()
split_paths = df[col].apply(clean_split)
if min_files_for_path is not None:
df = df[
split_paths.apply(lambda parts: len(parts) >= min_files_for_path)
].copy()
split_paths = split_paths[df.index]
df["group_key"] = split_paths.apply(
lambda parts: os.sep.join(parts[:path_exclusion_constant])
)
grouped = df.groupby("group_key")
new_rows = []
for _, group_df in grouped:
paths = group_df[col].tolist()
split_parts = [clean_split(p) for p in paths]
def longest_common_prefix(paths):
if not paths:
return []
prefix = paths[0]
for path in paths[1:]:
prefix = [a for a, b in zip(prefix, path) if a == b]
if not prefix:
break
return prefix
common_prefix = longest_common_prefix(split_parts)
prefix_str = os.sep.join(common_prefix)
for i, parts in enumerate(split_parts):
filename = parts[-1]
middle = (
os.sep.join(parts[len(common_prefix) : -1])
if len(parts) > len(common_prefix) + 1
else ""
)
row = group_df.iloc[i].copy()
row["longestcfp"] = prefix_str
row["middle"] = middle
row["filename_only"] = filename
row["file_extension"] = os.path.splitext(filename)[1].lower()
new_rows.append(row)
return pd.DataFrame(new_rows).drop(columns=["group_key"])
def calculatePath(approved_hashes, path_exclusion_constant, split):
if split:
dfs_by_policy = [group for _, group in approved_hashes.groupby("policy")]
else:
dfs_by_policy = [approved_hashes]
badpathparts = get_system_list("BAD_PATH_PARTS")
min_files_for_path = get_system_value("MIN_FILES_FOR_PATH", cast_type=int)
processed_dfs = []
for df in dfs_by_policy:
haslcp = splitFilepathsGrouped(df, path_exclusion_constant, "filename")
haslcp = haslcp.drop_duplicates()
forbidden = regulator(badpathparts, True)
forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False)
logger.debug("Removing forbidden filepaths for path exceptions")
print(colorText("Removing forbidden filepaths for path exceptions", "green"))
lcp_not_forbidden = haslcp[~forbidden_lcfp].copy()
lcp_not_forbidden_review = lcp_not_forbidden[
[
"policyname",
"longestcfp",
"middle",
"filename_only",
"file_extension",
"sha256",
]
]
unique_sha_counts = (
lcp_not_forbidden_review.groupby("longestcfp")["sha256"]
.nunique()
.reset_index()
)
unique_sha_counts.columns = ["longestcfp", "unique_sha256_count"]
lcp_not_forbidden_review = lcp_not_forbidden_review.merge(
unique_sha_counts, on="longestcfp", how="left"
)
lcp_not_forbidden_review = lcp_not_forbidden_review[
lcp_not_forbidden_review["unique_sha256_count"] >= min_files_for_path
]
processed_dfs.append(lcp_not_forbidden_review)
pathExclusions = pd.concat(processed_dfs, ignore_index=True)
return pathExclusions
def testChange(selected_policies, destination_policy, destination_allowlist):
working_dir = load_env("WORKING_DIR")
logger.info("These path exclusions would be added to:")
logger.info(destination_policy)
pathexclusions = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"
)
hashes = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv"
)
unique_combinations = pathexclusions[
["longestcfp", "file_extension"]
].drop_duplicates()
drive_letter_pattern = re.compile(r"^[a-zA-Z]:\\")
processed_paths = [
(path if drive_letter_pattern.match(path) else f"\\\\{path}") + f"\\**{ext}"
for path, ext in unique_combinations.itertuples(index=False, name=None)
]
for path in processed_paths:
logger.info(path)
print(colorText("These publishers would added", "yellow"))
processed_publishers = []
if os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv"
):
publishers = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv"
)
if publishers.empty:
print(colorText("The publishers list is empty.", "red"))
else:
processed_publishers = (
publishers[publishers["publisher"] != "Not Signed"]["publisher"]
.drop_duplicates()
.tolist()
)
for publisher in processed_publishers:
print(publisher)
print(colorText("These hashes would be added to:", "yellow"))
print(destination_allowlist)
processed_hashes = hashes["sha256"].unique().tolist()
print_x_wide(processed_hashes, 3)
return processed_paths, processed_hashes, processed_publishers
def menu_policy_enforce(
api: AirlockAPIWrapper,
): # TODO Need to clean up 6 and 7 into functions
selected_policies = []
destination_policy = []
destination_allowlist = []
processed_paths = []
processed_hashes = []
processed_publishers = []
working_dir = load_env("WORKING_DIR")
while True:
printEnforceChecklist(
selected_policies, destination_policy, destination_allowlist
)
choice = get_sanitized_input("\nEnter your choice: ")
if choice == "1":
clear_screen()
selected_policies = selectPolicies(api, True)
elif choice == "2":
clear_screen()
print(
colorText(
"Please choose destination_name Policy for Path Exclusions", "white"
)
)
destination_policy = selectPolicies(api, False)
print(colorText("Please choose Allowlist for Hashes", "white"))
destination_allowlist = selectAllowlists(api, destination_policy, False)
elif choice == "3":
clear_screen()
sortHashes(
api,
selected_policies,
type=[1, 2, 6, 7],
)
elif choice == "4":
clear_screen()
if os.path.exists(
f"{working_dir}\\Needs_Review\\Review_First\\{selected_policies[0].name}_approved_executions.csv"
):
buildPathsandPublishers(selected_policies, False)
else:
print(
"File not found. Please make sure it's saved correctly and try again."
)
elif choice == "5":
clear_screen()
if os.path.exists(
f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv"
) and os.path.exists(
f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv"
):
buildPreflights(selected_policies)
else:
print(
"File not found. Please make sure it's saved correctly and try again."
)
elif choice == "6":
clear_screen()
if (
os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"
)
and os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv"
)
and destination_policy
and destination_allowlist
):
processed_paths, processed_hashes, processed_publishers = testChange(
selected_policies, destination_policy, destination_allowlist
)
else:
# Log which condition(s) failed
missing_items = []
if not os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"
):
missing_items.append("approved_paths.csv not found")
if not os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv"
):
missing_items.append("approved_hashes.csv not found")
if not destination_policy:
missing_items.append("destination_policy is empty or None")
if not destination_allowlist:
missing_items.append("destination_allowlist is empty or None")
logger.error("Preflight check failed due to the following:")
for item in missing_items:
logger.error(f" - {item}")
elif choice == "7":
clear_screen()
areYouSure()
confirmation = get_sanitized_input("Type 'I AGREE' to continue: ")
if (
processed_paths
and processed_hashes
and processed_publishers
and destination_policy
and destination_allowlist
and confirmation.strip() == "I AGREE"
):
print(colorText("Proceeding with the code...", "yellow"))
api.hash_add_to_allowlist(
destination_allowlist[0].applicationid, processed_hashes
)
api.policy_add_path_exclusions(
destination_policy[0].groupid, processed_paths
)
if processed_publishers:
api.policy_add_publishers(
destination_policy[0].groupid, processed_publishers
)
locked()
else:
logger.error("Confirmation block failed. Reasons:")
if not processed_publishers or processed_hashes or processed_paths:
logger.error(" - Test not performed.")
if not destination_policy:
logger.error(" - `destination_policy` is missing or invalid.")
if not destination_allowlist:
logger.error(" - `destination_allowlist` is missing or invalid.")
if confirmation.strip() != "I AGREE":
logger.error(
" - User did not confirm with 'I AGREE'. Received: '%s'",
confirmation.strip(),
)
elif choice.upper() == "F":
open_directory(working_dir)
elif choice.upper() == "B":
break
else:
print(colorText("Invalid choice. Please try again.", "red"))
def section_header(title):
print(
colorText(
"\n --------------------------------------------------------------------",
"cyan",
)
)
print(colorText(f" ------------- {title} -------------", "cyan"))
print(
colorText(
" --------------------------------------------------------------------",
"cyan",
)
)
def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist):
working_dir = load_env("WORKING_DIR")
section_header("Prepare to Enforce Policy")
print(
colorText(
"\nSequentially follow these steps to prepare a policy for enforcement:",
"white",
)
)
# Step 1: Originating Policies
print(
colorText(
"\n1. Choose which policy or policies to gather execution info from", "cyan"
)
)
if not selected_policies:
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"))
# Step 2: Destination Policy and Allowlist
print(
colorText("2. Choose the destination policy and associated allowlist", "cyan")
)
if destination_policy:
print(
colorText(
f" [✅] {destination_policy[0].name} has been selected as the destination policy",
"green",
)
)
else:
print(colorText(" [❌] No destination policy has been chosen", "red"))
if destination_allowlist:
print(
colorText(
f" [✅] {destination_allowlist[0].name} has been selected as allowlist",
"green",
)
)
else:
print(colorText(" [❌] No allowlist has been chosen", "red"))
# Step 3: Data Preparation
print(
colorText(
f"3. Select to begin pulling execution history. The executions will be sorted and placed in {working_dir}\\data\\Needs_Review",
"cyan",
)
)
if selected_policies:
policy_id = selected_policies[0].name
review_path = f"{working_dir}\\Needs_Review\\Review_First\\{policy_id}_approved_executions.csv"
print(
colorText(
(
" [✅] Data has been fetched"
if os.path.exists(review_path)
else " [❌] Data has not been fetched"
),
"green" if os.path.exists(review_path) else "red",
)
)
else:
print(
colorText(
" [❌] No policies selected, cannot check data fetch status", "red"
)
)
# Step 4: Manual Review
print(colorText("4. Manually review the files:", "cyan"))
print(
colorText(
" Remove the rows containing hashes you do not approve of", "cyan"
)
)
print(
colorText(
f" When complete, save both csv files to {working_dir}\\data\\Approved and choose this option.",
"cyan",
)
)
print(
colorText(
" This will start the process to generate possible filepath approvals",
"cyan",
)
)
if selected_policies:
policy_id = selected_policies[0].name
approved_path = f"{working_dir}\\Approved\\{policy_id}_approved_executions.csv"
second_review_path = (
f"{working_dir}\\Needs_Review\\Review_Second\\{policy_id}_primary_Paths.csv"
)
print(
colorText(
(
" [✅] Reviewed hashes have been loaded"
if os.path.exists(approved_path)
else " [❌] Reviewed hashes have not been loaded"
),
"green" if os.path.exists(approved_path) else "red",
)
)
print(
colorText(
(
" [✅] Path review list created"
if os.path.exists(second_review_path)
else " [❌] Path review list has not been created"
),
"green" if os.path.exists(second_review_path) else "red",
)
)
else:
print(
colorText(
" [❌] No policies selected, cannot check reviewed hashes or path list",
"red",
)
)
# Step 5: Path Review
print(
colorText(
f"5. Manually review the files in {working_dir}\\Needs_Review\\Review_Second\\",
"cyan",
)
)
print(
colorText(
" Remove the rows containing path exclusions or publishers you do not approve of.",
"cyan",
)
)
print(
colorText(
f" When complete, save the files to {working_dir}\\data\\Approved",
"cyan",
)
)
print(
colorText(" Choose this option when done to build your preflights", "cyan")
)
if selected_policies:
policy_id = selected_policies[0].name
reviewed_path = f"{working_dir}\\Approved\\{policy_id}_primary_Paths.csv"
preflight_paths = f"{working_dir}\\Preflight\\{policy_id}_approved_paths.csv"
preflight_hashes = f"{working_dir}\\Preflight\\{policy_id}_approved_hashes.csv"
print(
colorText(
(
" [✅] Reviewed path list detected"
if os.path.exists(reviewed_path)
else " [❌] Path review list has not been detected"
),
"green" if os.path.exists(reviewed_path) else "red",
)
)
preflight_ready = os.path.exists(preflight_paths) and os.path.exists(
preflight_hashes
)
print(
colorText(
(
" [✅] Preflight Path Exclusion List has been generated"
if preflight_ready
else " [❌] Preflight Path Exclusion List has not been generated"
),
"green" if preflight_ready else "red",
)
)
else:
print(
colorText(
" [❌] No policies selected, cannot check preflight status", "red"
)
)
# Final Steps
print(
colorText(
"6. Test ------------------------------------------------------", "cyan"
)
)
print(
colorText(
" Prints to console the changes that would be made, must be done to proceed. ",
"cyan",
)
)
print(
colorText(
"7. Liftoff ------------------------------------------------------", "cyan"
)
)
print(
colorText(
" Apply path exclusions and approved publishers to selected policy",
"cyan",
)
)
print(colorText(" Apply approved hashes to allowlist", "cyan"))
# Utility Options
print(colorText("F. Open Working Directory", "cyan"))
print(colorText("B. Back", "cyan"))
BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 118 KiB

+1 -7
View File
@@ -29,7 +29,7 @@ import pandas as pd
import airlock_libs import airlock_libs
from services.API import AirlockAPIWrapper from services.API import AirlockAPIWrapper
from utils.configmanager import get_system_list, get_system_value from utils.configmanager import get_system_list, get_system_value
from utils.utils import colorText, regulator from utils.utils import regulator
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -293,12 +293,6 @@ class ExecutionHistoryRecord:
logger.debug( logger.debug(
f"Staging of Execution history for policy: {policy.name} is complete" f"Staging of Execution history for policy: {policy.name} is complete"
) )
print(
colorText(
f"Staging of Execution history for policy: {policy.name} is complete",
"green",
)
)
return executions return executions
+19 -7
View File
@@ -1,14 +1,26 @@
# Core TUI dependencies
textual==6.5.0
# API and data handling
Requests==2.32.5
pandas==2.3.3
numpy==2.3.4
# Database
pymongo==4.15.3
# Security and encryption
cryptography==46.0.3 cryptography==46.0.3
keyring==25.6.0 keyring==25.6.0
numpy==2.3.4
pandas==2.3.3 # Environment management
pymongo==4.15.3
python-dotenv==1.2.1 python-dotenv==1.2.1
Requests==2.32.5
textual==6.5.0 # Utilities
tqdm==4.67.1
urllib3==2.5.0 urllib3==2.5.0
plotext==5.3.2
pyperclip==1.11.0 pyperclip==1.11.0
# Custom/Private packages
--extra-index-url https://git.racooncity.org/api/packages/brotoskyj/pypi/simple/ --extra-index-url https://git.racooncity.org/api/packages/brotoskyj/pypi/simple/
airlock_libs==5.1.1 airlock_libs==7.2.0
+12 -2
View File
@@ -315,10 +315,12 @@ class AirlockAPIWrapper:
# Execution History # Execution History
def history_logging( def history_logging(
self, type: List[str], checkpoint: str, policy: List[str] self, type: List[str], checkpoint: str, policy: Optional[List[str]] = None
) -> str: ) -> str:
"""Retrieve execution history logs.""" """Retrieve execution history logs."""
payload = {"type": type, "checkpoint": checkpoint, "policy": policy} payload = {"type": type, "checkpoint": checkpoint}
if policy is not None:
payload["policy"] = policy
result = self._post("/v1/logging/exechistories", payload) result = self._post("/v1/logging/exechistories", payload)
return result["response"]["exechistories"] return result["response"]["exechistories"]
@@ -351,6 +353,14 @@ class AirlockAPIWrapper:
result = self._post("/v1/getexechistory", payload) result = self._post("/v1/getexechistory", payload)
return result["response"]["exechistory"] return result["response"]["exechistory"]
def server_logs(self, checkpoint: str | None = None) -> str:
"""Retrieves Server Activity History Logs."""
payload = {}
if checkpoint is not None:
payload["checkpoint"] = checkpoint
result = self._post("/v1/logging/svractivities?checkpoint", payload)
return result["response"]["svractivities"]
""" """
from services.API import AirlockAPIWrapper from services.API import AirlockAPIWrapper
-388
View File
@@ -1,388 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
from dataclasses import asdict
from datetime import datetime, timedelta
import json
import logging
import os
import re
from typing import List
import pandas as pd
from flows.prepPolicy import selectPolicies
from models.agent import Agent
from models.policy import Policy
from services.API import AirlockAPIWrapper
from utils.configmanager import get_system_json, load_env
from utils.selector import Selector
from utils.utils import colorText, get_sanitized_input
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): ",
value_type=int,
valid_range=(1, 150),
)
if not agents or not history_days:
print(colorText("No agents selected or invalid history range.", "red"))
return
historical_date = (datetime.now() - timedelta(days=history_days)).strftime(
"%Y-%m-%d"
)
today = datetime.now().strftime("%Y-%m-%d")
all_history = []
for agent in agents:
try:
exechistory = api.history_execution(today, historical_date, agent.hostname)
except Exception as e:
print(
colorText(
f"❌ Error retrieving history for {agent.hostname}: {e}", "red"
)
)
continue
if isinstance(exechistory, list):
for block in exechistory:
record = {
"Command": block.get("commandline", "N/A"),
"Date": block.get("datetime", "N/A"),
"Filename": block.get("filename", "N/A"),
"Policy Name": block.get("policyname", "N/A"),
"Hostname": block.get("hostname", "N/A"),
"Hash": block.get("sha256", "N/A"),
}
all_history.append(record)
if not outputjson:
for key, value in record.items():
print(colorText(f"{key}: {value}", "green"))
print("\n")
else:
print(
colorText(f"No execution history found for {agent.hostname}.", "yellow")
)
if outputjson:
print(json.dumps(all_history, indent=2))
def findAllAgents(api):
# Step 1: Load data from API
policies = [Policy(**row["data"]) for _, row in api.policy_find_all().iterrows()]
agents = [Agent(**row["data"]) for _, row in api.agent_find_all().iterrows()]
for agent in agents:
agent.enrich_with_policies(policies)
return agents
def findAgents(api, return_dataframe):
agents = selectAgents(api)
working_dir = load_env("WORKING_DIR")
if not agents:
logging.warning("No agents or policies found.")
print("No agents matched the criteria.")
return
# Convert enriched agents to DataFrame
agent_dicts = [asdict(agent) for agent in agents]
agent_df = pd.DataFrame(agent_dicts)
if return_dataframe:
logging.debug("Returning DataFrame to caller.")
return agent_df
# Otherwise, print and optionally export
print(agent_df)
logging.debug("Displayed DataFrame to console.")
user_input = (
get_sanitized_input(
"\nWould you like to export the results to a CSV file? (y/n): "
)
.strip()
.lower()
)
if user_input == "y":
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
filename = f"agentsearch_{timestamp}.csv"
file_path = os.path.join(str(working_dir), filename)
agent_df.to_csv(file_path, index=False)
logging.info(f"Exported DataFrame to {file_path}")
print(
colorText(
f"\n✓ Matched devices exported to: {working_dir}\\{filename}",
"green",
)
)
else:
logging.debug("User declined to export the DataFrame.")
def collect_device_names() -> List[str]:
print(colorText("🖥Â Device Search", "cyan"))
print(
colorText(
"Enter the device hostnames you'd like to search for, one per line.", "cyan"
)
)
print(
colorText(
"When you're done, press Enter twice (Three times if you have a single device).\n",
"cyan",
)
)
print(colorText("Example:", "cyan"))
print(colorText("H00000\nUTN00000\ni-hSuperSecretServer\nu-hVenderBroke\n", "cyan"))
print(colorText("Paste or type your device names below:", "white"))
device_input_lines = []
empty_line_count = 0
valid_line_pattern = re.compile(r"^[a-zA-Z0-9_\- ]+$")
while True:
line = get_sanitized_input("")
stripped_line = line.strip()
if stripped_line == "":
empty_line_count += 1
if empty_line_count == 2:
break
continue
else:
empty_line_count = 0
if valid_line_pattern.match(stripped_line):
device_input_lines.append(stripped_line)
else:
print(
colorText(
f"⚠️ Invalid input: '{stripped_line}' — only letters, numbers, underscores, spaces, and hyphens are allowed.",
"yellow",
)
)
return [name for name in device_input_lines if name]
def choose_match_type() -> bool:
print(colorText("Use exact match? (Y for exact, N for fuzzy):", "white"))
return get_sanitized_input("").strip().lower() in ["y", "yes"]
def match_agents(
device_names: List[str], agents: List["Agent"], use_exact: bool
) -> List["Agent"]:
if use_exact:
return [
agent
for agent in agents
if agent.hostname.lower() in [name.lower() for name in device_names]
]
else:
pattern = "|".join(map(re.escape, device_names))
regex = re.compile(pattern, re.IGNORECASE)
return [agent for agent in agents if regex.search(agent.hostname)]
def show_unmatched(
device_names: List[str], matched_agents: List["Agent"], use_exact: bool
):
if use_exact:
unmatched = [
name
for name in device_names
if not any(
agent.hostname.lower() == name.lower() for agent in matched_agents
)
]
else:
unmatched = [
name
for name in device_names
if not any(
re.search(re.escape(name), agent.hostname, re.IGNORECASE)
for agent in matched_agents
)
]
if unmatched:
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"]):
for agent in agents:
agent.enrich_with_policies(policies)
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"))
return []
use_exact = choose_match_type()
policies = [Policy(**row.to_dict()) for _, row in api.policy_find_all().iterrows()]
agents = [Agent(**row.to_dict()) for _, row in api.agent_find_all().iterrows()]
matched_agents = match_agents(device_names, agents, use_exact)
matched_agents.sort(key=lambda agent: agent.hostname.lower())
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"))
return []
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):
line = ""
for col in range(3):
idx = row + col * rows
if idx < len(matched_agents):
line += f"{matched_agents[idx].hostname:<30}"
logger.info(line)
matched_agents = Selector.select_with_mode(
matched_agents,
label_func=lambda agent: agent.hostname,
header="Matched Devices:",
)
if not matched_agents:
logger.debug("❌ No matching devices remain after refinement.")
print(colorText("❌ No matching devices remain after refinement.", "red"))
return []
enrich_agents(matched_agents, policies)
return matched_agents
def moveAgentToRelatedPolicy(
api: AirlockAPIWrapper,
agent: Agent,
mode: str = "audit",
):
"""
Moves an agent between audit and enforcement policies based on the mode.
Args:
api: AirlockAPIWrapper instance.
agent: Agent object.
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", "{}")
if mode == "audit":
if agent.groupid in policy_relationship_map:
target_policy = policy_relationship_map[agent.groupid]
elif agent.groupid in policy_relationship_map.values():
logger.debug(
f"Agent {agent.hostname} is already in an audit group. No action needed."
)
print(
f"Agent {agent.hostname} is already in an audit group. No action needed."
)
return
else:
logger.warning(
f"Error: No corresponding audit policy found for groupid: {agent.groupid}."
)
return
elif mode == "enforcement":
inverse_map = {v: k for k, v in policy_relationship_map.items()}
if agent.groupid in inverse_map:
target_policy = inverse_map[agent.groupid]
elif agent.groupid in inverse_map.values():
logger.info(
f"Agent {agent.hostname} is already in an enforcement group. No action needed."
)
return
else:
logger.warning(
f"Error: No corresponding enforcement policy found for groupid: {agent.groupid}."
)
return
else:
logger.error(f"Unknown mode '{mode}'. Use 'audit' or 'enforcement'.")
return
result = api.agent_move(agent.agentid, target_policy)
return result
def toggleEnforcement(api: AirlockAPIWrapper):
choices = ["Audit", "Enforcement", "Exit"]
print(colorText("Move devices to which state?:", "yellow"))
direction = Selector.select_string(choices, False, False)
if direction == "Exit":
pass
else:
devices = selectAgents(api)
for device in devices:
print(device.hostname)
confirm = Selector.confirm(
"Would you like to continue with these devices? Y/N: "
)
if direction and devices and confirm:
for device in devices:
result = moveAgentToRelatedPolicy(api, device, str(direction).lower())
logger.info(f"{device.hostname}: result: {result}")
get_sanitized_input("Press enter to continue")
def moveAgents(api: AirlockAPIWrapper):
devices = selectAgents(api)
for device in devices:
print(device.hostname)
confirm_devices = Selector.confirm(
"Would you like to continue with these devices? Y/N: "
)
if devices and confirm_devices:
policies = selectPolicies(api, False)
confirm_move = Selector.confirm(
f"Would you like to move these devices to {policies[0].name}?"
)
if confirm_move:
for device in devices:
result = api.agent_move(device.agentid, policies[0].groupid)
logger.info(f"{device.hostname}: result: {result}")
else:
logger.info("Exiting without change")
get_sanitized_input("Press enter to continue")
-234
View File
@@ -1,234 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
import datetime
import gc
import json
import logging
import os
import sys
from bson import ObjectId
import pandas as pd
import tqdm
from models.policy import Policy
from services.API import AirlockAPIWrapper
from utils.setup import get_base_directory
from utils.utils import colorText
logger = logging.getLogger(__name__)
def pullPolicyExechistories(
api: AirlockAPIWrapper,
policy: Policy,
type: list,
days,
outputjson: bool,
):
file_path = f"{get_base_directory()}\\cache\\chunkinator.json"
# Ensure the file exists
if not os.path.exists(file_path):
with open(file_path, "w") as file:
json.dump({"error": "Success", "response": {"exechistories": []}}, file)
logger.debug(f"File '{file_path}' has been created.")
else:
logger.debug(f"File '{file_path}' already exists.")
checkpoint = str(skipback(days))
json_output = {"error": "Success", "response": {"exechistories": []}}
with tqdm.tqdm(
file=sys.stdout,
leave=True,
total=10000,
desc=f"Checkpoint Progress: {checkpoint}",
colour="blue",
initial=1,
) as filebar:
with tqdm.tqdm(
file=sys.stdout,
leave=True,
total=100,
desc=f"Total of {policy} Complete: ",
) as pbar:
while True:
histories = api.history_logging(
type=type, checkpoint=checkpoint, policy=[policy.name]
)
# Ensure histories is a list of dictionaries
if not isinstance(histories, list) or not all(
isinstance(h, dict) for h in histories
):
logger.error(
"Unexpected response format from API. Expected list of dictionaries."
)
break
filebar.total = len(histories)
if not histories:
break
for index, history_item in enumerate(histories):
if (
"checkpoint" not in history_item
or "datetime" not in history_item
):
continue # Skip malformed entries
# Update checkpoint on last item
if index == len(histories) - 1:
checkpoint = history_item[
"checkpoint"
] # pyright: ignore[reportArgumentType]
filebar.desc = f"Checkpoint Progress: {checkpoint}"
break
try:
history_date = datetime.datetime.strptime(
history_item["datetime"].replace(
" +0000 UTC", ""
), # pyright: ignore[reportArgumentType]
"%Y-%m-%dT%H:%M:%SZ",
).date()
except ValueError:
continue # Skip if date format is invalid
if (
datetime.date.today() - datetime.timedelta(days=days)
) <= history_date:
json_output["response"]["exechistories"].append(history_item)
filebar.update(1)
filebar.refresh()
# Deduplicate entries
seen = {}
if os.path.exists(file_path):
with open(file_path, "r") as file:
existing_data = json.load(file)
combined = (
existing_data["response"]["exechistories"]
+ json_output["response"]["exechistories"]
)
else:
combined = json_output["response"]["exechistories"]
for entry in combined:
key = (
entry.get("sha256"),
entry.get("filename"),
entry.get("hostname"),
)
seen[key] = entry
deduplicated = list(seen.values())
with open(file_path, "w") as file:
json.dump(
{
"error": "Success",
"response": {"exechistories": deduplicated},
},
file,
)
json_output["response"]["exechistories"].clear()
# Update progress bar based on last valid item
try:
last_date = datetime.datetime.strptime(
history_item["datetime"].replace(" +0000 UTC", ""), # type: ignore
"%Y-%m-%dT%H:%M:%SZ",
).date()
date_diff = datetime.date.today() - last_date
percentage_diff = (
((days + 10) - date_diff.days) / (days + 10)
) * 100
pbar.n = round(percentage_diff)
pbar.set_description_str(f"Total of {policy} Complete: ")
pbar.refresh()
except Exception:
pass
filebar.n = 1
# Final output
with open(file_path, "r") as file:
final_output = json.load(file)
os.remove(file_path)
return json.dumps(final_output) if outputjson else None
def getPolicyInfo(api: AirlockAPIWrapper, policy, type, days):
import airlock_libs
executionhist_policy = pd.DataFrame()
exehist = airlock_libs.pull_policy_exec_histories(api, policy.name, str(type), days)
if exehist is not None:
data = json.loads(exehist)
executionhist_policy = pd.DataFrame(data["response"]["exechistories"])
if not executionhist_policy.empty:
executionhist_policy = executionhist_policy[
[
"datetime",
"sha256",
"publisher",
"filename",
"hostname",
"username",
"pprocess",
"gprocess",
"commandline",
]
]
executionhist_policy["policy"] = policy # Add policy column here
executionhist_policy = executionhist_policy.drop_duplicates(
subset=["sha256", "filename", "hostname"]
)
executionhist_policy = executionhist_policy.sort_values(
by=["sha256", "filename"]
)
logger.debug(f"Staging of Execution history for policy: {policy} is complete")
print(
colorText(
f"Staging of Execution history for policy: {policy} is complete",
"green",
)
)
del data
del exehist
gc.collect()
return executionhist_policy
def skipback(days):
"""
Generate a MongoDB ObjectId for a given number of days ago from today.
"""
adjusted_days = days
date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(
days=adjusted_days
)
timestamp = int(date_days_ago.timestamp())
hex_timestamp = format(timestamp, "08x")
objectid_hex = hex_timestamp + "0000000000000000"
return ObjectId(objectid_hex)
-14
View File
@@ -169,17 +169,3 @@ def getAPI(USERNAME, SERVICE_NAME):
logging.warning( logging.warning(
"Password does not meet complexity requirements. Try again." "Password does not meet complexity requirements. Try again."
) )
class APIKeyManager:
_api_key = None
@classmethod
def load(cls, service: str, username: str, password: str):
cls._api_key = retrieve_api_key(service, username, password)
@classmethod
def get(cls) -> str:
if cls._api_key is None:
raise ValueError("API key not loaded. Call APIKeyManager.load() first.")
return cls._api_key
-7
View File
@@ -336,10 +336,3 @@ def load_env_json(key: str, default: str = "[]") -> Any:
except Exception as e: except Exception as e:
logger.error(f"Failed to parse {key}: {e}") logger.error(f"Failed to parse {key}: {e}")
return json.loads(default) return json.loads(default)
# Backwards compatibility aliases (deprecated - use get_system_value instead)
get_protected_value = get_system_value
get_protected_json = get_system_json
load_protected_config = load_system_config
PROTECTED_KEYS = SYSTEM_CONFIG_KEYS # For backwards compatibility
+78
View File
@@ -0,0 +1,78 @@
# 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 <https://www.gnu.org/licenses/>.
import datetime
import gc
import json
import logging
from bson import ObjectId
import pandas as pd
from services.API import AirlockAPIWrapper
logger = logging.getLogger(__name__)
def getExecutions(api: AirlockAPIWrapper, policy, type, days):
import airlock_libs
executionhist_policy = pd.DataFrame()
exehist = airlock_libs.pull_policy_exec_histories(api, policy.name, str(type), days)
if exehist is not None:
data = json.loads(exehist)
executionhist_policy = pd.DataFrame(data["response"]["exechistories"])
if not executionhist_policy.empty:
executionhist_policy = executionhist_policy[
[
"datetime",
"sha256",
"publisher",
"filename",
"hostname",
"username",
"pprocess",
"gprocess",
"commandline",
]
]
executionhist_policy["policy"] = policy # Add policy column here
executionhist_policy = executionhist_policy.drop_duplicates(
subset=["sha256", "filename", "hostname"]
)
executionhist_policy = executionhist_policy.sort_values(
by=["sha256", "filename"]
)
logger.debug(f"Staging of Execution history for policy: {policy} is complete")
del data
del exehist
gc.collect()
return executionhist_policy
def skipback(days):
"""
Generate a MongoDB ObjectId for a given number of days ago from today.
"""
adjusted_days = days
date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(
days=adjusted_days
)
timestamp = int(date_days_ago.timestamp())
hex_timestamp = format(timestamp, "08x")
objectid_hex = hex_timestamp + "0000000000000000"
return ObjectId(objectid_hex)
-357
View File
@@ -1,357 +0,0 @@
# 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 <https://www.gnu.org/licenses/>.
import logging
from typing import Any, Callable, List, Optional, Union
import pandas as pd
from utils.utils import colorText, get_sanitized_input
logger = logging.getLogger(__name__)
class Selector:
@staticmethod
def _get_sorted_items(
items: List[Any], label_func: Callable[[Any], str]
) -> List[Any]:
return sorted(items, key=lambda item: label_func(item).lower())
@staticmethod
def _display_choices(
items: List[Any],
label_func: Callable[[Any], str],
num_columns: int = 3,
header: str = "Available Choices:",
) -> None:
# Force single column if items are DataFrame rows
if items and isinstance(items[0], (pd.Series, dict)):
num_columns = 1
rows = (len(items) + num_columns - 1) // num_columns
print(f"\n{header}")
for row in range(rows):
line = ""
for col in range(num_columns):
idx = row + col * rows
if idx < len(items):
label = label_func(items[idx])
line += f"{idx + 1}: {label:<30}"
print(line)
@staticmethod
def _display_selected_items(
selected: List[Any], label_func: Callable[[Any], str], num_columns: int = 3
) -> None:
print(colorText("\nCurrent selections:", "cyan"))
if not selected:
print(" (none)")
return
sorted_selected = sorted(selected, key=lambda item: label_func(item).lower())
rows = (len(sorted_selected) + num_columns - 1) // num_columns
for row in range(rows):
line = ""
for col in range(num_columns):
idx = row + col * rows
if idx < len(sorted_selected):
label = label_func(sorted_selected[idx])
line += f"{label:<30}"
print(line)
@staticmethod
def _parse_selection_input(input_str: str, max_index: int) -> List[int]:
selections = []
for part in input_str.split(","):
part = part.strip()
if "-" in part:
try:
start, end = map(int, part.split("-"))
selections.extend(range(start, end + 1))
except ValueError:
continue
elif part.isdigit():
selections.append(int(part))
return [i for i in selections if 1 <= i <= max_index]
@staticmethod
def _select_from_list(
items: List[Any],
label_func: Callable[[Any], str],
allow_multiple: bool = False,
prompt_each: bool = False,
header: str = "Available Choices:",
num_columns: int = 3,
) -> Union[Optional[Any], List[Any]]:
if not items:
logger.warning("No items available for selection.")
return None
full_sorted_items = Selector._get_sorted_items(items, label_func)
remaining_items = full_sorted_items.copy()
selected = []
if allow_multiple:
while True:
Selector._display_choices(
remaining_items, label_func, num_columns=num_columns, header=header
)
Selector._display_selected_items(
selected, label_func, num_columns=num_columns
)
choice = (
get_sanitized_input(
"Select item(s) by number (e.g. 1,3-5), R to reset, Q to finish: "
)
.strip()
.lower()
)
if choice == "q":
break
elif choice == "r":
selected.clear()
remaining_items = full_sorted_items.copy()
print(colorText("🔄 Selections reset.", "yellow"))
continue
indices = Selector._parse_selection_input(choice, len(remaining_items))
newly_selected = []
for index in indices:
item = remaining_items[index - 1]
if item not in selected:
selected.append(item)
newly_selected.append(item)
if prompt_each:
logger.info(f"Selected: {label_func(item)}")
else:
logger.warning("Item already selected.")
remaining_items = [
item for item in remaining_items if item not in newly_selected
]
return selected if selected else None
else:
Selector._display_choices(
full_sorted_items, label_func, num_columns=num_columns, header=header
)
try:
choice = int(get_sanitized_input("Select one item by number: "))
if 1 <= choice <= len(full_sorted_items):
selected_item = full_sorted_items[choice - 1]
logger.info(f"Selected: {label_func(selected_item)}")
return selected_item
else:
logger.warning("Selection out of range.")
except ValueError:
logger.warning("Invalid input.")
return None
@staticmethod
def select_with_mode(
items: List[Any],
label_func: Callable[[Any], str],
header: str = "Available Choices:",
) -> List[Any]:
print(
colorText(
"Choose selection mode: [I]nclude only selected, [E]xclude selected, [A]ll (skip):",
"white",
)
)
mode = get_sanitized_input("").strip().lower()
if mode == "a":
return items
selected = Selector._select_from_list(
items,
label_func=label_func,
allow_multiple=True,
prompt_each=False,
header=header,
)
if not selected:
return items
if mode == "i":
print(colorText(f"✅ Included {len(selected)} item(s).", "green"))
return selected
elif mode == "e":
print(colorText(f"👫 Excluded {len(selected)} item(s).", "yellow"))
return [item for item in items if item not in selected]
else:
print(colorText("⚠️ Invalid mode. Returning all items.", "yellow"))
return items
@staticmethod
def select_objects(
objects: List[Any], allow_multiple: bool = False, prompt_each: bool = False
) -> Union[Optional[Any], List[Any]]:
return Selector._select_from_list(
objects,
label_func=lambda obj: getattr(obj, "name", str(obj)),
allow_multiple=allow_multiple,
prompt_each=prompt_each,
header="Available Objects:",
)
@staticmethod
def select_string(
options: List[str], allow_multiple: bool = False, prompt_each: bool = False
) -> Union[Optional[str], List[str]]:
return Selector._select_from_list(
options,
label_func=str,
allow_multiple=allow_multiple,
prompt_each=prompt_each,
header="Available Options:",
)
@staticmethod
def select_int(
options: List[int], allow_multiple: bool = False, prompt_each: bool = False
) -> Union[Optional[int], List[int]]:
return Selector._select_from_list(
options,
label_func=lambda x: str(x),
allow_multiple=allow_multiple,
prompt_each=prompt_each,
header="Available Integers:",
)
@staticmethod
def select_value(
prompt: str,
value_type: type = int,
valid_range: Optional[tuple] = None,
allow_quit: bool = False,
) -> Optional[Any]:
while True:
user_input = get_sanitized_input(prompt).strip().lower()
if allow_quit and user_input == "q":
logger.info("User opted to quit value selection.")
return None
try:
value = value_type(user_input)
if valid_range:
min_val, max_val = valid_range
if not (min_val <= value <= max_val):
logger.warning(f"Value out of range ({min_val}{max_val}).")
continue
logger.info(f"User selected value: {value}")
return value
except ValueError:
logger.warning(f"Invalid input. Expected a {value_type.__name__}.")
@staticmethod
def confirm(prompt: str = "Are you sure? (Y/N): ") -> bool:
while True:
response = get_sanitized_input(prompt).strip().lower()
if response in ["y", "yes"]:
logger.info("User confirmed action.")
return True
elif response in ["n", "no"]:
logger.info("User declined action.")
return False
else:
logger.warning("Invalid confirmation input. Expected 'Y' or 'N'.")
@staticmethod
def select_dataframe_rows(
df: pd.DataFrame,
columns: Optional[List[str]] = None,
allow_multiple: bool = False,
prompt_each: bool = False,
header: str = "Available Rows:",
) -> List[pd.Series]:
if df.empty:
print("DataFrame is empty.")
return []
if columns:
df = df[columns]
items = [row for _, row in df.iterrows()]
def label_func(row):
return str(row.to_dict())
result = Selector._select_from_list(
items,
label_func=label_func,
allow_multiple=allow_multiple,
prompt_each=prompt_each,
header=header,
)
if isinstance(result, pd.Series):
return [result]
elif isinstance(result, list):
return result
else:
return []
@staticmethod
def select_dataframe_with_mode(
df: pd.DataFrame,
columns: Optional[List[str]] = None,
header: str = "Available Rows:",
) -> List[pd.Series]:
if df.empty:
print("⚠️ DataFrame is empty.")
return []
# Filter columns if specified
if columns:
df = df[columns]
items = df.to_dict("records")
def label_func(row):
return " | ".join(str(row[col]) for col in df.columns)
# Show rows first
print(colorText(header, "cyan"))
for i, row in enumerate(items):
print(f"{i}: {label_func(row)}")
# Prompt for mode once
print(
colorText(
"\nChoose selection mode: [I]nclude only selected, [E]xclude selected, [A]ll (skip):",
"white",
)
)
mode = get_sanitized_input("").strip().lower()
if mode == "a":
return [pd.Series(row) for row in items]
# Prompt for selection only once
selected = Selector._select_from_list(
items,
label_func=label_func,
allow_multiple=True,
prompt_each=False,
header=header,
)
if not selected:
return [pd.Series(row) for row in items]
if mode == "i":
print(colorText(f"✅ Included {len(selected)} row(s).", "green"))
return [pd.Series(row) for row in selected]
elif mode == "e":
print(colorText(f"👫 Excluded {len(selected)} row(s).", "yellow"))
return [pd.Series(row) for row in items if row not in selected]
else:
print(colorText("⚠️ Invalid mode. Returning no rows.", "yellow"))
return []
+99 -23
View File
@@ -29,6 +29,49 @@ from utils.configmanager import (
) )
class TextualNotificationHandler(logging.Handler):
"""
Custom logging handler that sends ERROR, WARNING, and CRITICAL logs
to Textual toast notifications.
"""
def __init__(self, app):
super().__init__()
self.app = app
def emit(self, record):
try:
# Only handle ERROR, WARNING, and CRITICAL
if record.levelno >= logging.WARNING:
# Format the message
msg = self.format(record)
# Map log levels to Textual severity
severity_map = {
logging.WARNING: "warning",
logging.ERROR: "error",
logging.CRITICAL: "error",
}
severity = severity_map.get(record.levelno, "information")
# Send to Textual notification
# Use call_from_thread if logging from non-main thread
try:
self.app.notify(msg, severity=severity, timeout=5)
except Exception:
# If we're not on the main thread, schedule it
try:
self.app.call_from_thread(
self.app.notify, msg, severity=severity, timeout=5
)
except Exception:
# Silently fail to avoid breaking the logging system
pass
except Exception:
# Silently fail to avoid breaking the logging system
pass
def get_base_directory() -> Path: def get_base_directory() -> Path:
system = platform.system() system = platform.system()
home = Path.home() home = Path.home()
@@ -40,42 +83,46 @@ def get_base_directory() -> Path:
return home / ".local" / "share" / "Loxide" return home / ".local" / "share" / "Loxide"
def configure_logging(log_dir: Path, log_level: str = "INFO"): def configure_logging(log_dir: Path, cache_dir: Path, log_level: str = "INFO"):
"""
Configure logging and return function to attach notification handler.
Args:
log_dir: Directory for log files
cache_dir: Directory for cache files (used by version checker)
log_level: Logging level string
Returns:
Function to attach notification handler to Textual app
"""
log_file = log_dir / "Loxide.log" log_file = log_dir / "Loxide.log"
config = { config = {
"version": 1, # Required key for dictConfig format version "version": 1,
"disable_existing_loggers": False, # Keeps existing loggers active "disable_existing_loggers": False,
"formatters": { "formatters": {
"detailed": { "detailed": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s" "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
# Includes timestamp, logger name, level, and message
},
"simple": {
"format": "%(levelname)s - %(message)s"
# Minimal format for console output
}, },
"simple": {"format": "%(levelname)s - %(message)s"},
"toast": {"format": "%(name)s: %(message)s"}, # Simpler format for toasts
}, },
"handlers": { "handlers": {
"file": { "file": {
"class": "logging.handlers.TimedRotatingFileHandler", "class": "logging.handlers.TimedRotatingFileHandler",
"filename": str(log_file), "filename": str(log_file),
"when": "midnight", # Rotate logs at midnight "when": "midnight",
"interval": 1, # Every 1 day "interval": 1,
"backupCount": 7, # Keep 7 days of logs "backupCount": 7,
"encoding": "utf-8", # Ensure UTF-8 encoding "encoding": "utf-8",
"level": "DEBUG", # Always log DEBUG and above to file "level": "DEBUG",
"formatter": "detailed", # Use detailed format "formatter": "detailed",
},
"console": {
"class": "logging.StreamHandler",
"level": log_level.upper(), # System-configured level for console
"formatter": "simple", # Use simple format
}, },
# REMOVED console handler - it interferes with Textual TUI
}, },
"root": { "root": {
"level": "DEBUG", # Root logger level "level": "DEBUG",
"handlers": ["file", "console"], # Attach both handlers "handlers": ["file"], # Only use file handler, not console
}, },
} }
@@ -96,6 +143,30 @@ def configure_logging(log_dir: Path, log_level: str = "INFO"):
logging.config.dictConfig(config) logging.config.dictConfig(config)
logging.getLogger().debug("✅ Logging configured.") logging.getLogger().debug("✅ Logging configured.")
# Return a function to attach the notification handler once the app is created
def attach_notification_handler(app):
"""Attach the Textual notification handler and version checker to the app."""
# Attach logging handler
handler = TextualNotificationHandler(app)
handler.setLevel(logging.WARNING) # Only WARNING and above
formatter = logging.Formatter("%(name)s: %(message)s")
handler.setFormatter(formatter)
logging.getLogger().addHandler(handler)
logging.getLogger().debug("✅ Textual notification handler attached.")
# Attach version checker (checks in background, notifies if update available)
try:
from utils.versionchecker import create_update_notifier
create_update_notifier(app, cache_dir=cache_dir)
logging.getLogger().debug("✅ Version checker attached.")
except ImportError as e:
logging.getLogger().debug(f"Version checker not available: {e}")
except Exception as e:
logging.getLogger().warning(f"Could not initialize version checker: {e}")
return attach_notification_handler
def setup(): def setup():
""" """
@@ -105,6 +176,9 @@ def setup():
3. Load user config (mutable, from user_config.json) 3. Load user config (mutable, from user_config.json)
4. Configure logging 4. Configure logging
5. Set up .env with WORKING_DIR only 5. Set up .env with WORKING_DIR only
Returns:
attach_notification_handler: Function to attach notification handler to TUI app
""" """
base_dir = get_base_directory() base_dir = get_base_directory()
dirs = { dirs = {
@@ -122,7 +196,7 @@ def setup():
# Configure logging with system-defined log level # Configure logging with system-defined log level
log_level = get_system_value("LOG_LEVEL", str, "INFO") log_level = get_system_value("LOG_LEVEL", str, "INFO")
configure_logging(dirs["logs"], log_level) attach_handler = configure_logging(dirs["logs"], dirs["cache"], log_level)
# Load user config (mutable) # Load user config (mutable)
load_user_config(dirs["config"]) load_user_config(dirs["config"])
@@ -145,7 +219,6 @@ def setup():
"Approved": [], "Approved": [],
"Needs_Review": ["Review_First", "Review_Second", "HTML"], "Needs_Review": ["Review_First", "Review_Second", "HTML"],
"Preflight": ["HTML"], "Preflight": ["HTML"],
"Archived": [],
} }
for folder_name, subfolders in folders_structure.items(): for folder_name, subfolders in folders_structure.items():
@@ -158,3 +231,6 @@ def setup():
logging.debug(f"'{subfolder}' subfolder created at: {subfolder_path}") logging.debug(f"'{subfolder}' subfolder created at: {subfolder_path}")
logging.info("✅ Setup complete") logging.info("✅ Setup complete")
# Return the attach handler function
return attach_handler
-437
View File
@@ -19,81 +19,10 @@ import os
import platform import platform
import re import re
import subprocess import subprocess
import tempfile
import tkinter as tk
from tkinter import filedialog
import pandas as pd
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def import_to_dataframe(file_path: str) -> pd.DataFrame:
df = pd.DataFrame()
try:
if not os.path.exists(file_path):
print(colorText(f"Error: File '{file_path}' does not exist.", "red"))
return df
ext = os.path.splitext(file_path)[1].lower()
if ext == ".csv":
df = pd.read_csv(file_path)
elif ext == ".parquet":
df = pd.read_parquet(file_path)
else:
print(colorText(f"Error: Unsupported file extension '{ext}'.", "red"))
return df
if df.empty:
print(colorText("Error: File has headers but no data rows.", "red"))
else:
print(colorText(f"Data loaded successfully from {file_path}", "green"))
return df
except pd.errors.EmptyDataError:
print(
colorText(
"Notice: CSV file is completely empty, falling back to empty frame",
"white",
)
)
return pd.DataFrame()
except Exception as e:
print(colorText(f"Error reading file: {e}", "red"))
return pd.DataFrame()
def choose_directory():
root = tk.Tk()
root.withdraw() # Hide the main window
directory = filedialog.askdirectory(title="Select a Directory")
print("Selected directory:", directory)
return directory
def choose_file(initial_directory=None, required_substring=None):
"""Open a file dialog and ensure the selected file contains a required substring."""
while True:
root = tk.Tk()
root.withdraw() # Hide the main window
file_path = filedialog.askopenfilename(initialdir=initial_directory)
if not file_path:
print("No file selected.")
return None
if required_substring and required_substring not in file_path:
print(
f"The selected file must contain '{required_substring}' in its path or name. Please try again."
)
else:
return file_path
def get_sanitized_input(prompt: str) -> str: def get_sanitized_input(prompt: str) -> str:
while True: while True:
user_input = input(prompt) user_input = input(prompt)
@@ -151,235 +80,6 @@ def irtang():
) )
def section_header(title):
print(
colorText(
"\n --------------------------------------------------------------------",
"cyan",
)
)
print(colorText(f" ------------- {title} -------------", "cyan"))
print(
colorText(
" --------------------------------------------------------------------",
"cyan",
)
)
def areYouSure():
print(
colorText(
"🛑****************************************************************************************************************************************🛑",
"red",
)
)
print(
colorText(
"⚠️=========================================================================================================================================⚠️",
"yellow",
)
)
print(
colorText(
"🛑========================================================================================================================================🛑",
"red",
)
)
print(
colorText(
"⚠️-------------This program will now begin to make changes to the Airlock Console. Do you understand and agree to proceed? ----------------⚠️",
"yellow",
)
)
print(
colorText(
"🛑========================================================================================================================================🛑",
"red",
)
)
print(
colorText(
"⚠️=========================================================================================================================================⚠️",
"yellow",
)
)
print(
colorText(
"🛑****************************************************************************************************************************************🛑",
"red",
)
)
def locked():
print(
colorText(
r"""
""",
"yellow",
)
)
def printDeviceEnforceChecklist():
print(
colorText(
"\n --------------------------------------------------------------------",
"cyan",
)
)
print(
colorText(
" ------------- 🛠️ 🔒 Prepare to Enforce Policy 🛠️ 🔒 ------------------",
"cyan",
)
)
print(
colorText(
" --------------------------------------------------------------------",
"cyan",
)
)
print(
colorText(
"\nSequentually follow these steps to prepare a policy for enforcement:",
"white",
)
)
print(
colorText(
"\n1. Choose which originating policy or policies to move to enforcement",
"cyan",
)
)
print(
colorText(
"2. Pull and stage event history, combine the histories, add hash info, then categorize the hashes",
"cyan",
)
)
print(colorText("3. Manually review the files:", "cyan"))
print(
colorText(
" 'needs_approved\\good_{first_policy}_{second_policy}.csv' and 'needs_approved\\unknown_{first_policy}_{second_policy}.csv'",
"cyan",
)
)
print(
colorText(
" Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.",
"cyan",
)
)
print(
colorText(
" If metarules need to be created, please make note of them, and remove the row from the csv.",
"cyan",
)
)
print(
colorText(
" When complete, save both csv files to the directory 'approved' and choose this option.",
"cyan",
)
)
print(
colorText(
" This will combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed",
"cyan",
)
)
print(
colorText(
"4. Manually review the file 'needs_approved\\paths_needing_review.csv'",
"cyan",
)
)
print(
colorText(
" Remove the rows containing path exclusions you do not approve of",
"cyan",
)
)
print(
colorText(
" When complete, save the csv file to the directory 'approved'", "cyan"
)
)
print(
colorText(
" Do the same process with the list of publishers forthe same directories",
"cyan",
)
)
print(colorText(" Preflight Lists will be generated", "cyan"))
print(
colorText(
"5. Choose the destination policy and parent and child allow list", "cyan"
)
)
print(
colorText(
"6. Test ------------------------------------------------------", "cyan"
)
)
print(colorText(" Print rather than apply selected data.", "cyan"))
print(
colorText(
"7. Liftoff ------------------------------------------------------", "cyan"
)
)
print(
colorText(
" Apply path exclusions according to allowed and approved paths",
"cyan",
)
)
print(
colorText(" Apply signed or attested hashes to Parent Allow List", "cyan")
)
print(
colorText(
" Apply approved, but unsigned hashes to the Child Allow List", "cyan"
)
)
print(
colorText(
"R. Remove/Reset Generated data - will prompt to allow keeping execution history",
"cyan",
)
)
print(colorText("B. Back", "cyan"))
def colorText(text, color): def colorText(text, color):
colors = { colors = {
"red": "\033[91m", "red": "\033[91m",
@@ -394,133 +94,6 @@ def colorText(text, color):
return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}" return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}"
def formatHTML(df, output_html_path=None, overwrite=True):
from datetime import datetime
# Get current date and filename for subtitle
today = datetime.now().strftime("%d %B %Y") # Changed to "Day Month Year"
filename = output_html_path.replace(".html", "") if output_html_path else "Report"
dark_css = """
<style>
body {
background-color: #000000;
margin: 0;
padding: 0;
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
color: #f8f8f2;
}
.header {
text-align: center;
margin: 20px auto;
padding: 10px;
border-bottom: 2px solid #ffd700;
max-width: 95%;
}
.header h1 {
color: #ffd700;
margin: 0;
font-size: 32px;
}
.header p {
color: #00bfff;
margin: 5px 0 0 0;
font-size: 18px;
}
.table-container {
overflow-y: scroll;
margin: 0 auto;
width: 95%;
max-height: calc(80vh - 100px);
display: block;
border: 1px solid #3a3a4d;
margin-bottom: 0;
}
table {
border-collapse: collapse;
font-size: 14px;
background-color: #1e1e2f;
color: #f8f8f2;
width: max-content;
}
th, td {
border: 1px solid #3a3a4d;
text-align: left;
padding: 10px;
max-width: 300px;
word-wrap: break-word;
overflow-wrap: break-word;
}
/* First column: no wrap */
td:nth-child(1), th:nth-child(1) {
white-space: nowrap;
max-width: none !important;
word-wrap: normal !important;
}
th {
background-color: #2e2e40;
color: #ffd700;
position: sticky;
top: 0;
z-index: 10;
}
tr:nth-child(even) {
background-color: #262638;
}
tr:hover {
background-color: #33334d;
color: #00bfff;
}
/* Custom scrollbar styling */
.table-container::-webkit-scrollbar {
width: 12px;
}
.table-container::-webkit-scrollbar-track {
background: #1e1e2f;
}
.table-container::-webkit-scrollbar-thumb {
background-color: #3a3a4d;
border-radius: 6px;
}
</style>
"""
header = f"""
<div class="header">
<h1>Airlock Tools</h1>
<p>{filename} - {today}</p>
</div>
"""
html_table = df.to_html(index=False, escape=False)
styled_html = (
f"<html>\n"
f"<head><title>Airlock Tools Report</title></head>\n"
f"<body>\n"
f"{dark_css}\n"
f"{header}\n"
f"<div class='table-container'>\n"
f" {html_table}\n"
f"</div>\n"
f"</body>\n"
f"</html>"
)
if output_html_path:
with open(output_html_path, "w", encoding="utf-8") as f:
f.write(styled_html)
print(f"✅ Styled table saved to '{output_html_path}'")
elif overwrite:
with tempfile.NamedTemporaryFile(
suffix=".html", delete=False, mode="w", encoding="utf-8"
) as f:
f.write(styled_html)
temp_path = f.name
print(f"✅ Styled table saved to temporary file: {temp_path}")
else:
return styled_html
def open_directory(path): def open_directory(path):
system = platform.system() system = platform.system()
@@ -530,13 +103,3 @@ def open_directory(path):
subprocess.run(["xdg-open", path]) subprocess.run(["xdg-open", path])
else: else:
raise OSError(f"Unsupported operating system: {system}") raise OSError(f"Unsupported operating system: {system}")
def print_x_wide(items: list, width: int):
for i in range(0, len(items), width):
row = items[i : i + width]
print(" | ".join(row))
def clear_screen():
os.system("cls" if os.name == "nt" else "clear")
+560
View File
@@ -0,0 +1,560 @@
# 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 <https://www.gnu.org/licenses/>.
"""
Version checking and update notification system for Loxide.
Checks against Gitea releases at:
https://git.racooncity.org/brotoskyj/AirlockTools/releases
"""
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
import json
import logging
from pathlib import Path
import re
import threading
from typing import Callable, Optional
import requests
logger = logging.getLogger(__name__)
# Current application version - UPDATE THIS ON EACH RELEASE
__version__ = "1.0.0"
# Gitea release API configuration
GITEA_API_BASE = "https://git.racooncity.org/api/v1"
REPO_OWNER = "brotoskyj"
REPO_NAME = "AirlockTools"
RELEASES_URL = f"{GITEA_API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/releases"
RELEASES_PAGE_URL = f"https://git.racooncity.org/{REPO_OWNER}/{REPO_NAME}/releases"
# How often to check for updates (in hours)
CHECK_INTERVAL_HOURS = 24
@dataclass
class ReleaseInfo:
"""Information about a release."""
tag_name: str
version: tuple # Parsed semantic version (major, minor, patch)
name: str
body: str # Release notes
published_at: datetime
html_url: str
download_url: Optional[str] = None # URL to download the release asset
is_prerelease: bool = False
@dataclass
class UpdateCheckResult:
"""Result of an update check."""
current_version: str
latest_version: Optional[str]
update_available: bool
release_info: Optional[ReleaseInfo]
error: Optional[str] = None
def parse_version(version_str: str) -> Optional[tuple]:
"""
Parse a version string into a comparable tuple.
Supports formats: v1.2.3, 1.2.3, v1.2, 1.2
Returns (major, minor, patch) tuple or None if parsing fails.
"""
if not version_str:
return None
# Strip 'v' prefix if present
clean = version_str.lstrip("vV").strip()
# Match semantic version pattern
match = re.match(r"^(\d+)(?:\.(\d+))?(?:\.(\d+))?", clean)
if not match:
return None
major = int(match.group(1))
minor = int(match.group(2)) if match.group(2) else 0
patch = int(match.group(3)) if match.group(3) else 0
return (major, minor, patch)
def compare_versions(v1: tuple, v2: tuple) -> int:
"""
Compare two version tuples.
Returns:
-1 if v1 < v2
0 if v1 == v2
1 if v1 > v2
"""
for a, b in zip(v1, v2):
if a < b:
return -1
if a > b:
return 1
return 0
def get_current_version() -> str:
"""Get the current application version."""
return __version__
def _parse_release_response(release_data: dict) -> Optional[ReleaseInfo]:
"""Parse a release from Gitea API response."""
try:
tag_name = release_data.get("tag_name", "")
version = parse_version(tag_name)
if not version:
logger.debug(f"Could not parse version from tag: {tag_name}")
return None
# Parse published date
published_str = release_data.get("published_at", "")
try:
published_at = datetime.fromisoformat(published_str.replace("Z", "+00:00"))
except (ValueError, AttributeError):
published_at = datetime.now(UTC)
# Get download URL from assets if available
download_url = None
assets = release_data.get("assets", [])
for asset in assets:
# Prefer .exe or .zip files
name = asset.get("name", "").lower()
if name.endswith((".exe", ".zip", ".msi")):
download_url = asset.get("browser_download_url")
break
return ReleaseInfo(
tag_name=tag_name,
version=version,
name=release_data.get("name", tag_name),
body=release_data.get("body", ""),
published_at=published_at,
html_url=release_data.get("html_url", RELEASES_PAGE_URL),
download_url=download_url,
is_prerelease=release_data.get("prerelease", False),
)
except Exception as e:
logger.warning(f"Failed to parse release data: {e}")
return None
def fetch_latest_release(
include_prerelease: bool = False, timeout: int = 10
) -> Optional[ReleaseInfo]:
"""
Fetch the latest release from Gitea.
Args:
include_prerelease: Whether to include pre-release versions
timeout: Request timeout in seconds
Returns:
ReleaseInfo for the latest release, or None if fetch fails
"""
try:
response = requests.get(
RELEASES_URL,
params={"limit": 10}, # Get last 10 releases to find latest stable
timeout=timeout,
headers={"Accept": "application/json"},
)
response.raise_for_status()
releases = response.json()
if not releases:
logger.debug("No releases found")
return None
# Find the latest release (first non-prerelease if we're excluding them)
for release_data in releases:
release_info = _parse_release_response(release_data)
if release_info is None:
continue
if include_prerelease or not release_info.is_prerelease:
return release_info
# If all are prereleases and we're excluding them, return the first one anyway
# but log a warning
if releases:
logger.debug("All releases are pre-releases")
return _parse_release_response(releases[0])
return None
except requests.exceptions.Timeout:
logger.warning("Timeout fetching releases from Gitea")
return None
except requests.exceptions.RequestException as e:
logger.warning(f"Failed to fetch releases: {e}")
return None
except (json.JSONDecodeError, KeyError) as e:
logger.warning(f"Failed to parse release response: {e}")
return None
def check_for_updates(include_prerelease: bool = False) -> UpdateCheckResult:
"""
Check if a newer version is available.
Args:
include_prerelease: Whether to consider pre-release versions
Returns:
UpdateCheckResult with the check results
"""
current = get_current_version()
current_parsed = parse_version(current)
if not current_parsed:
return UpdateCheckResult(
current_version=current,
latest_version=None,
update_available=False,
release_info=None,
error="Could not parse current version",
)
release_info = fetch_latest_release(include_prerelease=include_prerelease)
if release_info is None:
return UpdateCheckResult(
current_version=current,
latest_version=None,
update_available=False,
release_info=None,
error="Could not fetch release information",
)
is_newer = compare_versions(release_info.version, current_parsed) > 0
return UpdateCheckResult(
current_version=current,
latest_version=release_info.tag_name,
update_available=is_newer,
release_info=release_info,
)
class VersionChecker:
"""
Background version checker that periodically checks for updates
and can notify the application when updates are available.
"""
def __init__(
self,
cache_dir: Optional[Path] = None,
check_interval_hours: int = CHECK_INTERVAL_HOURS,
on_update_available: Optional[Callable[[UpdateCheckResult], None]] = None,
):
"""
Initialize the version checker.
Args:
cache_dir: Directory to store last check timestamp
check_interval_hours: Hours between automatic checks
on_update_available: Callback when update is available
"""
self.cache_dir = cache_dir
self.check_interval = timedelta(hours=check_interval_hours)
self.on_update_available = on_update_available
self._last_check: Optional[datetime] = None
self._last_result: Optional[UpdateCheckResult] = None
self._check_thread: Optional[threading.Thread] = None
self._dismissed_version: Optional[str] = None
# Load cached state
self._load_cache()
@property
def cache_file(self) -> Optional[Path]:
if self.cache_dir:
return self.cache_dir / "version_check_cache.json"
return None
def _load_cache(self) -> None:
"""Load cached check state."""
if not self.cache_file or not self.cache_file.exists():
return
try:
with open(self.cache_file, "r") as f:
data = json.load(f)
if "last_check" in data:
self._last_check = datetime.fromisoformat(data["last_check"])
# Don't load dismissed_version - dismiss is session-only
except (json.JSONDecodeError, ValueError, OSError) as e:
logger.debug(f"Could not load version check cache: {e}")
def _save_cache(self) -> None:
"""Save check state to cache."""
if not self.cache_file:
return
try:
self.cache_file.parent.mkdir(parents=True, exist_ok=True)
data = {}
if self._last_check:
data["last_check"] = self._last_check.isoformat()
# Don't save dismissed_version - dismiss is session-only
with open(self.cache_file, "w") as f:
json.dump(data, f)
except OSError as e:
logger.debug(f"Could not save version check cache: {e}")
def should_check(self) -> bool:
"""Determine if enough time has passed to check again."""
if self._last_check is None:
return True
elapsed = datetime.now(UTC) - self._last_check
return elapsed >= self.check_interval
def check_now(
self, force: bool = False, include_prerelease: bool = False
) -> UpdateCheckResult:
"""
Check for updates immediately.
Args:
force: Check even if recently checked
include_prerelease: Include pre-release versions
Returns:
UpdateCheckResult
"""
if not force and not self.should_check() and self._last_result:
return self._last_result
result = check_for_updates(include_prerelease=include_prerelease)
self._last_check = datetime.now(UTC)
self._last_result = result
self._save_cache()
# Notify if update available and not dismissed
if (
result.update_available
and self.on_update_available
and result.latest_version != self._dismissed_version
):
self.on_update_available(result)
return result
def check_async(
self, force: bool = False, include_prerelease: bool = False
) -> None:
"""
Check for updates in background thread.
Args:
force: Check even if recently checked
include_prerelease: Include pre-release versions
"""
if self._check_thread and self._check_thread.is_alive():
return # Already checking
if not force and not self.should_check():
return # Too soon to check again
def _check():
try:
self.check_now(force=True, include_prerelease=include_prerelease)
except Exception as e:
logger.debug(f"Background version check failed: {e}")
self._check_thread = threading.Thread(target=_check, daemon=True)
self._check_thread.start()
def dismiss_update(self, version: str) -> None:
"""
Dismiss update notification for a specific version.
Only lasts for the current session - will nag again on next startup.
Args:
version: Version to dismiss (e.g., "v1.2.3")
"""
# Session-only dismiss - don't save to cache
self._dismissed_version = version
def clear_dismissed(self) -> None:
"""Clear the dismissed version so user will be nagged again."""
self._dismissed_version = None
def get_last_result(self) -> Optional[UpdateCheckResult]:
"""Get the result of the last check."""
return self._last_result
# Global instance for easy access
_global_checker: Optional[VersionChecker] = None
def get_version_checker(
cache_dir: Optional[Path] = None,
on_update_available: Optional[Callable[[UpdateCheckResult], None]] = None,
) -> VersionChecker:
"""
Get or create the global version checker instance.
Args:
cache_dir: Directory for caching (only used on first call)
on_update_available: Callback for updates (only used on first call)
Returns:
The global VersionChecker instance
"""
global _global_checker
if _global_checker is None:
_global_checker = VersionChecker(
cache_dir=cache_dir,
on_update_available=on_update_available,
)
return _global_checker
def format_update_message(result: UpdateCheckResult, short: bool = False) -> str:
"""
Format a human-readable update message.
Args:
result: The update check result
short: Whether to use a short format
Returns:
Formatted message string
"""
if not result.update_available:
return f"✅ Loxide is up to date (v{result.current_version})"
if short:
return f"🆕 Update available: {result.latest_version}"
msg = f"🆕 Loxide {result.latest_version} is available! (current: v{result.current_version})"
if result.release_info:
msg += f"\n📥 Download: {result.release_info.html_url}"
# Include release notes preview if available
if result.release_info.body:
notes = result.release_info.body.strip()
# Truncate if too long
if len(notes) > 200:
notes = notes[:200] + "..."
msg += f"\n\n📋 Release Notes:\n{notes}"
return msg
# ---------------------------------------------------------------------------
# Textual TUI Integration
# ---------------------------------------------------------------------------
def create_update_notifier(
app, cache_dir: Optional[Path] = None, nag_on_startup: bool = True
):
"""
Create a version checker that notifies via Textual toast notifications.
This should be called after the Textual app is created.
Args:
app: The Textual App instance
cache_dir: Directory for caching check state
nag_on_startup: Always show notification on startup if update available
Returns:
The VersionChecker instance
"""
def on_update_available(result: UpdateCheckResult):
"""Callback when update is available - show toast notification."""
try:
msg = f"🆕 Update available: {result.latest_version}\nGo to Settings to download"
try:
app.notify(
msg, title="Loxide Update Available", severity="warning", timeout=15
)
except RuntimeError:
app.call_from_thread(
app.notify,
msg,
title="Loxide Update Available",
severity="warning",
timeout=15,
)
except Exception as e:
logger.debug(f"Could not show update notification: {e}")
checker = get_version_checker(
cache_dir=cache_dir,
on_update_available=on_update_available,
)
# Store checker on app so Loxide.on_mount can use it
if nag_on_startup:
app._version_checker = checker
app._version_nag_shown = False
return checker
def check_for_updates_startup(
cache_dir: Optional[Path] = None,
) -> Optional[UpdateCheckResult]:
"""
Check for updates during application startup.
This performs a synchronous check but respects the cache interval,
so it will only actually query the network once per CHECK_INTERVAL_HOURS.
Returns the result if an update is available, None otherwise.
Example usage:
result = check_for_updates_startup(cache_dir)
if result and result.update_available:
print(format_update_message(result))
"""
checker = get_version_checker(cache_dir=cache_dir)
# Only check if enough time has passed (uses cache)
if not checker.should_check():
result = checker.get_last_result()
if result and result.update_available:
return result
return None
result = checker.check_now(force=False)
if result.update_available:
return result
return None