Fixed issue with generating multiple OTP - now prints a nice list for Copy/Paste. Began splitting client / server functionality. Demoted selector and setup to util from service. Fixed some typos.

utils/menus.py

@@ -30,6 +30,13 @@ from flows.prepPolicy import (
     sortHashes,
 )
 from flows.quietAgent import findQuietAgents
+
+from flows.otp import (
+    generate,
+    otp_activities_by_agent,
+    revoke
+)
+
 from services.agenthandler import findAgents
 from services.API import AirlockAPIWrapper
 from utils.utils import (
@@ -41,6 +48,8 @@ from utils.utils import (
     printEnforceChecklist,
 )
 
+
+
 logger = logging.getLogger(__name__)
 
 dotenv.load_dotenv()
@@ -237,24 +246,33 @@ def menu_policy_enforce(api: AirlockAPIWrapper):
 
 
 def menu_otp(api: AirlockAPIWrapper):
+    working_dir = load_env("WORKING_DIR")
     while True:
+
         print(colorText("\n--- 🎫 OTP Submenu 🎫 ---", "cyan"))
-        print(colorText("1. Generate OTP", "cyan"))
-        # print(colorText("2. Sub-option B","cyan"))
-        print(colorText("Q. Return to Main Menu", "cyan"))
+        print(colorText("1. 🔐  -Generate OTPs", "cyan"))
+        print(colorText("2. 📊  -OTP Activities By Agent", "cyan"))
+        print(colorText("3. ❌  -Revoke OTPs", "cyan"))
+        print(colorText("F. 📂  - Open Working Directory", "yellow"))
+        print(colorText("S. 🛠️  - Settings", "yellow"))
+        print(colorText("Q. 🔚  - Quit", "yellow"))
+
         choice = input("Enter your choice: ")
 
         if choice == "1":
-            # TODO generateOTP(api,findAgents()
-            break
-
+            otp_list = generate(api)
+            print(colorText(otp_list,"green"))
         elif choice == "2":
-            print("You selected Sub-option B")
+            otp_activities_by_agent(api)
+        elif choice == "3":
+            revoke(api)
+        elif choice == "F":
+            open_directory(working_dir)
+        elif choice == "S":
+            menu_settings()
         elif choice == "Q":
-            print("Returning to Main Menu...")
             break
-        else:
-            print("Invalid choice. Please try again.")
+   
 
 
 def menu_settings():

utils/selector.py

@@ -0,0 +1,174 @@
+#    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, List, Optional, Union
+
+logger = logging.getLogger(__name__)
+
+
+from typing import List, Optional, Union, Any, Callable
+import logging
+
+logger = logging.getLogger(__name__)
+
+class Selector:
+    @staticmethod
+    def _display_choices(
+        items: List[Any],
+        label_func: Callable[[Any], str],
+        num_columns: int = 4,
+        header: str = "Available Choices:"
+    ) -> None:
+        sorted_items = sorted(items, key=lambda item: label_func(item).lower())
+        rows = (len(sorted_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(sorted_items):
+                    label = label_func(sorted_items[idx])
+                    line += f"{idx + 1}: {label:<30}"
+            print(line)
+
+    @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:"
+    ) -> Union[Optional[Any], List[Any]]:
+        if not items:
+            logger.warning("No items available for selection.")
+            return None
+
+        Selector._display_choices(items, label_func, header=header)
+        sorted_items = sorted(items, key=lambda item: label_func(item).lower())
+        selected = []
+
+        if allow_multiple:
+            while True:
+                choice = input("Select an item by number (or Q to finish): ").strip().lower()
+                if choice == "q":
+                    break
+                try:
+                    index = int(choice)
+                    if 1 <= index <= len(sorted_items):
+                        item = sorted_items[index - 1]
+                        if item not in selected:
+                            selected.append(item)
+                            if prompt_each:
+                                logger.info(f"Selected: {label_func(item)}")
+                        else:
+                            logger.warning("Item already selected.")
+                    else:
+                        logger.warning("Selection out of range. Try again.")
+                except ValueError:
+                    logger.warning("Invalid input. Enter a number or 'Q' to quit.")
+            return selected if selected else None
+        else:
+            try:
+                choice = int(input("Select one item by number: "))
+                if 1 <= choice <= len(sorted_items):
+                    selected_item = 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_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 = 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 = 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'.")

utils/setup.py

@@ -0,0 +1,182 @@
+#    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 json
+import logging
+import logging.handlers
+import os
+import platform
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv, set_key
+
+PROTECTED_KEYS = [
+    "APPNAME",
+    "PATH_EXCLUSION_CONST",
+    "MIN_FILES_FOR_PATH",
+    "VT_THREAT_TOLERANCE",
+    "POLICY_MAP_ENF_AUD"
+]
+
+def get_base_directory() -> Path:
+    system = platform.system()
+    home = Path.home()
+    if system == 'Windows':
+        return Path(os.getenv('APPDATA', home / 'AppData' / 'Roaming')) / "AirlockTools"
+    elif system == 'Darwin':
+        return home / 'Library' / 'Application Support' / "AirlockTools"
+    else:
+        return home / '.local' / 'share' / "AirlockTools"
+
+def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
+    log_file = log_dir / "airlocktools.log"
+    logger = logging.getLogger()
+    logger.setLevel(getattr(logging, log_level.upper(), logging.DEBUG))
+
+    # 🔧 Clear existing handlers
+    for handler in logger.handlers[:]:
+        logger.removeHandler(handler)
+
+    file_handler = logging.handlers.RotatingFileHandler(
+        log_file, maxBytes=5_000_000, backupCount=5, encoding='utf-8'
+    )
+    file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
+    logger.addHandler(file_handler)
+
+    console_handler = logging.StreamHandler()
+    console_handler.setFormatter(logging.Formatter('%(levelname)s - %(message)s'))
+    logger.addHandler(console_handler)
+
+    if platform.system() == "Windows":
+        try:
+            event_handler = logging.handlers.NTEventLogHandler("AirlockTools")
+            event_handler.setLevel(logging.CRITICAL)
+            event_handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
+            logger.addHandler(event_handler)
+        except Exception as e:
+            logger.warning(f"Could not attach Windows Event Log handler: {e}")
+
+    logger.debug("✅ Logging configured.")
+
+
+def get_system_config_path() -> Path:
+    base_path = Path(getattr(sys, '_MEIPASS', os.path.dirname(os.path.abspath(__file__))))
+    return base_path.parent / "system_config.json"
+
+def load_system_config() -> dict:
+    try:
+        config_path = get_system_config_path()
+        with open(config_path, "r") as f:
+            return json.load(f)
+    except FileNotFoundError:
+        logging.warning("⚠️ system_config.json not found. Using built-in defaults.")
+        return {
+            "APPNAME": "AirlockTools",
+            "LOG_LEVEL": "DEBUG",
+            "PATH_EXCLUSION_CONST": 4,
+            "MIN_FILES_FOR_PATH": 4,
+            "VT_THREAT_TOLERANCE": 4,
+            "POLICY_MAP_ENF_AUD": {
+                "enforced_id": "audit_id"
+            }
+        }
+
+def load_user_config(config_dir: Path) -> dict:
+    user_config_path = config_dir / "user_config.json"
+    if not user_config_path.exists():
+        default_user_config = {
+            "URL": "",
+            "LOG_LEVEL": "INFO"
+        }
+        with open(user_config_path, "w") as f:
+            json.dump(default_user_config, f, indent=4)
+        logging.debug(f"Created user config at {user_config_path}")
+    with open(user_config_path, "r") as f:
+        return json.load(f)
+
+def write_config_to_env(config: dict, env_path: Path):
+    for key, value in config.items():
+        try:
+            serialized = json.dumps(value) if isinstance(value, (list, dict)) else str(value)
+            set_key(env_path, key, serialized)
+        except Exception as e:
+            logging.warning(f"Failed to write {key} to .env: {e}")
+
+def setup() -> Path:
+    base_dir = get_base_directory()
+    dirs = {
+        'config': base_dir / 'config',
+        'cache': base_dir / 'cache',
+        'logs': base_dir / 'logs',
+    }
+
+    for name, path in dirs.items():
+        path.mkdir(parents=True, exist_ok=True)
+        logging.debug(f"{name.capitalize()} directory ensured at: {path}")
+
+    system_config = load_system_config()
+    configure_logging(dirs['logs'], system_config.get("LOG_LEVEL", "DEBUG"))
+
+    env_path = base_dir / ".env"
+    if not env_path.exists():
+        env_path.touch()
+    load_dotenv(dotenv_path=env_path, override=True)
+
+    working_dir = Path(os.getenv("WORKING_DIR") or (base_dir / "data"))
+    working_dir.mkdir(parents=True, exist_ok=True)
+    set_key(env_path, "WORKING_DIR", str(working_dir))
+    os.environ["WORKING_DIR"] = str(working_dir)
+    logging.debug(f"Working directory set to: {working_dir}")
+
+    folders_structure = {
+        "Approved": [],
+        "Needs_Review": ["Review_First", "Review_Second", "HTML"],
+        "Preflight": ["HTML"],
+        "Archived": []
+    }
+
+    for folder_name, subfolders in folders_structure.items():
+        folder_path = working_dir / folder_name
+        folder_path.mkdir(parents=True, exist_ok=True)
+        logging.debug(f"'{folder_name}' folder ensured at: {folder_path}")
+        for subfolder in subfolders:
+            subfolder_path = folder_path / subfolder
+            subfolder_path.mkdir(parents=True, exist_ok=True)
+            logging.debug(f"  └─ '{subfolder}' subfolder created at: {subfolder_path}")
+
+    user_config = load_user_config(dirs['config'])
+    merged_config = {**system_config, **user_config}
+
+    for key in PROTECTED_KEYS:
+        merged_config[key] = system_config.get(key, "")
+
+    # ✅ URL resolution order: system_config → .env → user prompt
+    url = system_config.get("URL")
+    if not url:
+        url = os.getenv("URL")
+    if not url:
+        url = input("🌐 Enter the service URL (e.g., https://example.com/api): ").strip()
+
+    merged_config["URL"] = url
+    set_key(env_path, "URL", url)
+    os.environ["URL"] = url
+    logging.debug(f"Service URL set to: {url}")
+
+    write_config_to_env(merged_config, env_path)
+
+    return working_dir