171 lines
6.3 KiB
Python
171 lines
6.3 KiB
Python
# 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
|
|
import logging
|
|
import os
|
|
|
|
import pandas as pd
|
|
|
|
from services.agenthandler import selectAgents
|
|
from services.API import AirlockAPIWrapper
|
|
from utils.configmanager import load_env
|
|
from utils.selector import Selector
|
|
from utils.utils import colorText, get_sanitized_input
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def otp_generate(api: AirlockAPIWrapper):
|
|
otp_dict = {}
|
|
agents = selectAgents(api)
|
|
print(colorText("Would you like to continue with these devices?","white"))
|
|
for agent in agents:
|
|
print(agent.hostname)
|
|
confirm = Selector.confirm()
|
|
if agents and confirm:
|
|
requester = get_sanitized_input("Who is requesting the OTP: ")
|
|
because = get_sanitized_input("Why/What work are they doing?: ")
|
|
|
|
purpose = f"Requester: {requester} - for : {because}"
|
|
possible_durations = [15, 60, 360, 1440, 10080]
|
|
|
|
print(colorText("Please select a duration in minutes: ", "white"))
|
|
print(colorText("15 mins, 60 mins, 360 mins(6 Hours), 1440 mins (24 Hours), 10080 mins (7 Days):", "white"))
|
|
duration_selected = Selector.select_int(possible_durations)
|
|
|
|
if isinstance(duration_selected, list):
|
|
duration_selected = duration_selected[0] if duration_selected else None
|
|
|
|
if duration_selected is not None:
|
|
for agent in agents:
|
|
logging.info(f"Querying API for {agent.hostname}")
|
|
otp_code = api.otp_generate(agent.agentid, duration_selected, purpose)
|
|
logger.debug(f"Generated OTP for {agent.hostname}: {otp_code}")
|
|
otp_dict[agent.hostname] = otp_code
|
|
|
|
print(colorText("Requested Codes:", "green"))
|
|
for key, value in otp_dict.items():
|
|
print(colorText(f"{key} | {value}","green"))
|
|
|
|
def otp_activities_by_agent(api: AirlockAPIWrapper):
|
|
activeagents = api.otp_find_active()
|
|
awaitingagents = api.otp_find_awaiting()
|
|
enforcedagents = api.otp_find_enforced()
|
|
revokedagents = api.otp_find_revoked()
|
|
|
|
|
|
# Add a 'status' column to each DataFrame
|
|
activeagents['status'] = 'active'
|
|
awaitingagents['status'] = 'awaiting'
|
|
enforcedagents['status'] = 'enforced'
|
|
revokedagents['status'] = 'revoked'
|
|
|
|
# Combine all into one DataFrame
|
|
combined_agents = pd.concat([activeagents, awaitingagents, enforcedagents, revokedagents], 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"
|
|
)
|
|
combined_df = pd.DataFrame()
|
|
|
|
for row in selected_rows:
|
|
otpid = row['otpid']
|
|
hostname = row['hostname']
|
|
result = api.otp_get_activities(otpid)
|
|
result['hostname'] = hostname
|
|
if not result.empty:
|
|
logger.info(f"Activities for {hostname} (otpid: {otpid}):\n{result}")
|
|
combined_df = pd.concat([combined_df, result], ignore_index=True)
|
|
else:
|
|
logger.info(f"No activities found for {hostname} (otpid: {otpid})")
|
|
|
|
user_input = get_sanitized_input("\nWould you like to export the results to a CSV file? (y/n): ").strip().lower()
|
|
if user_input == 'y':
|
|
working_dir = load_env("WORKING_DIR")
|
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
filename = f"otp_activities_{timestamp}.csv"
|
|
file_path = os.path.join(str(working_dir), filename)
|
|
|
|
combined_df.to_csv(file_path, index=False)
|
|
logging.info(f"Exported Data to {file_path}")
|
|
|
|
print(
|
|
colorText(
|
|
f"\n✅ OTP Activity exported to: {working_dir}\\{filename}",
|
|
"green",
|
|
)
|
|
)
|
|
else:
|
|
logging.debug("User declined to export the DataFrame.")
|
|
|
|
|
|
|
|
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}")
|
|
|
|
|