Pydantic v2 data model, storage, config refactor

eagleosint/cli.py

@@ -6,7 +6,7 @@
 
 import click
 
-from eagleosint.config import CONFIGS, logger, save_config
+from eagleosint.config import settings, logger, save_config
 from eagleosint.display import (
     RED, YELLOW, BLUE, WHITE, DARK_GRAY, LIGHT_RED,
     BG_WHITE, BG_RED,
@@ -108,15 +108,14 @@ def settings():
 
     setting_counter = 0
     config_options = {}
-    for setting_key_name, setting_value_item in CONFIGS.items():
-        if setting_key_name != "headers":
-            setting_counter += 1
-            config_options[str(setting_counter)] = setting_key_name
-            print(
-                f"         {WHITE}{RED}  0{setting_counter} {setting_key_name}"
-                + " " * (20 - len(setting_key_name))
-                + f'{LIGHT_RED}:  "{setting_value_item}" '
-            )
+    for setting_key_name, setting_value_item in settings.display_items().items():
+        setting_counter += 1
+        config_options[str(setting_counter)] = setting_key_name
+        print(
+            f"         {WHITE}{RED}  0{setting_counter} {setting_key_name}"
+            + " " * (20 - len(setting_key_name))
+            + f'{LIGHT_RED}:  "{setting_value_item}" '
+        )
     exit_option_key = "exit".upper()
     print(
         f"         {WHITE}{RED}  00{RED} {exit_option_key}"
@@ -136,7 +135,7 @@ def settings():
         f"{SPACE_PREFIX}{LIGHT_RED}>{RED} Insert the new value of "
         f"{config_options[chosen_option]} :{LIGHT_RED} "
     )
-    CONFIGS[config_options[chosen_option]] = new_setting_value
+    settings.set_key(config_options[chosen_option], new_setting_value)
     save_config()
 
 
@@ -161,6 +160,20 @@ def _run_update() -> None:
         print(f"{RED}Update timed out. Check your internet connection and retry.{WHITE}")
 
 
+def _write_output(
+    results: list,
+    fmt: str,
+    filepath: str | None,
+) -> None:
+    """Write structured output to file or stdout."""
+    from eagleosint.output import write_results
+    if filepath:
+        with open(filepath, "w", encoding="utf-8") as fh:
+            write_results(results, fmt, fh)
+        print(f"{BLUE}>{WHITE} output written to {filepath}")
+    else:
+        write_results(results, fmt)
+
 @click.group(
     invoke_without_command=True,
     context_settings=dict(help_option_names=["-h", "--help"]),
@@ -252,27 +265,98 @@ def cmd_riplookup() -> None:
     print(LOGO)
     infoga("reverseiplookup")
 
+@main.group("investigation")
+def cmd_investigation() -> None:
+    """Commands related to investigations."""
+
+@cmd_investigation.command("new")
+@click.argument("name")
+@click.option("--tags", "-t", default="", help="Comma-separated tags.")
+@click.option("--notes", "-n", default="", help="Initial notes.")
+def cmd_investigation_new(name: str, tags: str, notes: str) -> None:
+    """Create a new investigation session."""
+    from eagleosint.storage import create_investigation
+    tag_list = [t.strip() for t in tags.split(",") if t.strip()]
+    row = create_investigation(name, tags=tag_list, notes=notes)
+    print(f"{BLUE}>{WHITE} investigation created")
+    print(f"{BLUE}  id    :{WHITE} {row.id}")
+    print(f"{BLUE}  name  :{WHITE} {row.name}")
+
+@cmd_investigation.command("list")
+def cmd_investigation_list() -> None:
+    """List all investigation sessions."""
+    from eagleosint.storage import list_investigations
+    import json
+    rows = list_investigations()
+    if not rows:
+        print(f"{YELLOW}no investigations found.{WHITE}")
+        return
+    for row in rows:
+        tags = ", ".join(json.loads(row.tags)) or "-"
+        print(
+            f"{BLUE}  {row.id}{WHITE}\n"
+            f"   name  : {row.name}\n"
+            f"   tags  : {tags}\n"
+            f"   updated : {row.updated_at[:19]}\n"
+        )
+
 
 @main.command("iplocation")
-def cmd_iplocation() -> None:
+@click.option("--output", "-o", type=click.Choice(["json", "csv"]), default=None,
+              help="Emit structured output.")
+@click.option("--output-file", "-f", "output_file", type=click.Path(), default=None,
+              help="Write output to this file path.")
+@click.option("--save-to", "save_to", default=None,
+              help="Save result to this investigation ID.")
+def cmd_iplocation(output: str | None, output_file: str | None, save_to: str | None) -> None:
     """IP address geolocation."""
     print(LOGO)
-    iplocation()
+    result = iplocation()
+    if output and result:
+        _write_output([result], output, output_file)
+    if save_to and result:
+        from eagleosint.storage import save_result
+        save_result(save_to, result)
+        print(f"{BLUE}>{WHITE} result saved to investigation {save_to}")
 
 
 @main.command("bitly")
-def cmd_bitly() -> None:
+@click.option("--output", "-o", type=click.Choice(["json", "csv"]), default=None,
+              help="Emit structured output.")
+@click.option("--output-file", "-f", "output_file", type=click.Path(), default=None,
+              help="Write output to this file path.")
+@click.option("--save-to", "save_to", default=None,
+              help="Save result to this investigation ID.")
+def cmd_bitly(output: str | None, output_file: str | None, save_to: str | None) -> None:
     """Resolve and bypass Bitly short URLs."""
     print(LOGO)
-    bypass_bitly()
+    result = bypass_bitly()
+    if output and result:
+        _write_output([result], output, output_file)
+    if save_to and result:
+        from eagleosint.storage import save_result
+        save_result(save_to, result)
+        print(f"{BLUE}>{WHITE} result saved to investigation {save_to}")
+
 
 
 @main.command("github")
-def cmd_github() -> None:
+@click.option("--output", "-o", type=click.Choice(["json", "csv"]), default=None,
+              help="Emit structured output instead of (or in addition to) terminal display.")
+@click.option("--output-file", "-f", "output_file", type=click.Path(), default=None,
+              help="Write output to this file path.")
+@click.option("--save-to", "save_to", default=None,
+              help="Save result to this investigation ID.")
+def cmd_github(output: str | None, output_file: str | None, save_to: str | None) -> None:
     """GitHub user profile lookup."""
     print(LOGO)
-    github_lookup()
-
+    result = github_lookup()
+    if output and result:
+        _write_output([result], output, output_file)
+    if save_to and result:
+        from eagleosint.storage import save_result
+        save_result(save_to, result)
+        print(f"{BLUE}>{WHITE} result saved to investigation {save_to}")
 
 @main.command("tempmail")
 def cmd_tempmail() -> None:

eagleosint/config.py

@@ -3,12 +3,13 @@
 import logging
 import os
 from logging.handlers import RotatingFileHandler
+from platformdirs import user_config_dir
 
-CONFIG_DIR  = os.path.join(os.path.expanduser("~"), ".config", "E4GL30S1NT")
+CONFIG_DIR = user_config_dir("E4GL30S1NT", appauthor=False)
 CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json")
 LOG_PATH    = os.path.join(CONFIG_DIR, "eagleosint.log")
 COOKIE_FILE = os.path.join(os.path.expanduser("~"), ".cookies")
-REALEMAIL_API_CONFIG_KEY = "real-email-api-key"
+PINGUTIL_API_CONFIG_KEY = "pingutil-api-key"
 VERIPHONE_API_CONFIG_KEY = "veriphone-api-key"
 
 os.makedirs(CONFIG_DIR, exist_ok=True)
@@ -20,7 +21,7 @@
     CONFIGS: dict = json.load(_cfg)
 
 _ENV_KEY_MAP = {
-    REALEMAIL_API_CONFIG_KEY: "E4GL30S1NT_REALEMAIL_KEY",
+    PINGUTIL_API_CONFIG_KEY:  "E4GL30S1NT_PINGUTIL_KEY",
     VERIPHONE_API_CONFIG_KEY: "E4GL30S1NT_VERIPHONE_KEY",
 }
 for _k, _v in _ENV_KEY_MAP.items():
@@ -29,6 +30,93 @@
         CONFIGS[_k] = _val
 
 
+# --------------------------------------------------------------
+# Settings model
+# --------------------------------------------------------------
+
+from pydantic import BaseModel, ConfigDict, SecretStr
+
+class Settings(BaseModel):
+    """
+    Typed, validated, secrets-masked application configuration.
+
+    Secrets are stored as SecretStr - they appear as '*********' in
+    logs and repr, never as plaintext.
+    Adding a new provider key = one new field here.
+    """
+    model_config = ConfigDict(extra="ignore")
+
+    pingutil_api_key: SecretStr | None = None
+    veriphone_api_key: SecretStr | None = None
+
+    def get_key(self, name: str) -> str | None:
+        """Return a secret value by its dash-separated key name."""
+        val = getattr(self, name.replace("-", "_"), None)
+        return val.get_secret_value() if isinstance(val, SecretStr) else None
+
+    def set_key(self, name: str, value: str) -> None:
+        """Update a key in-place, Raises KeyError for unknown keys."""
+        attr = name.replace("-", "_")
+        if attr not in type(self).model_fields:
+            raise KeyError(f"unknown config key: {name!r}")
+        setattr(self, attr, SecretStr(value) if value else None)
+
+    def to_file_dict(self) -> dict[str, str]:
+        """Non-null keys as plaintext dict - for JSON persistence only"""
+        out = {}
+        for field_name in type(self).model_fields:
+            val = getattr(self, field_name)
+            if isinstance(val, SecretStr):
+                out[field_name.replace("_", "-")] = val.get_secret_value()
+
+        return out
+
+    def display_items(self) -> dict[str, str]:
+       """All keys with partially masked values - for the settings UI"""
+       out = {}
+       for field_name in type(self).model_fields:
+           val = getattr(self, field_name)
+           dash_name = field_name.replace("_", "-")
+           if isinstance(val, SecretStr):
+               raw = val.get_secret_value()
+               out[dash_name] = f"{raw[:4]}{'*' * 8}" if raw else "(empty)"
+           else:
+               out[dash_name] = "(not set)"
+       return out
+
+# --------------------------------------------------------------
+# Load settings: JSON file -> env var overrides
+# --------------------------------------------------------------
+
+def _to_secret(val: str | None) -> SecretStr | None:
+    return SecretStr(val) if val else None
+
+def _load_settings() -> Settings:
+    data: dict[str, str] = {}
+
+    if os.path.exists(CONFIG_PATH):
+        try:
+            with open(CONFIG_PATH, "r", encoding="utf-8") as f:
+                data = json.load(f)
+        except (json.JSONDecodeError, OSError):
+            pass
+
+    _env_map = {
+        "pingutil-api-key": "E4GL30S1NT_PINGUTIL_KEY",
+        "veriphone-api-key": "E4GL30S1NT_VERIPHONE_KEY",
+    }
+    for key_name, env_var in _env_map.items():
+        val = os.getenv(env_var)
+        if val:
+            data[key_name] = val
+
+    return Settings(
+        pingutil_api_key=_to_secret(data.get("pingutil-api-key")),
+        veriphone_api_key=_to_secret(data.get("veriphone-api-key")),
+    )
+
+settings = _load_settings()
+
 def _setup_logger() -> logging.Logger:
     _logger = logging.getLogger("eagleosint")
     if _logger.handlers:
@@ -54,9 +142,9 @@ def _setup_logger() -> logging.Logger:
 
 
 def save_config() -> None:
-    """Write CONFIGS atomically — write-to-temp then os.replace()."""
+    """Persist settings to JSON file atomically."""
     tmp = CONFIG_PATH + ".tmp"
     with open(tmp, "w", encoding="utf-8") as f:
-        json.dump(CONFIGS, f, indent=2)
+        json.dump(settings.to_file_dict(), f, indent=2)
     os.replace(tmp, CONFIG_PATH)
     logger.debug("Config saved to %s", CONFIG_PATH)