70 lines
2.4 KiB
Python
70 lines
2.4 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/>.
|
|
|
|
|
|
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}")
|