Merge pull request #274 from shredzwho/feature/hudson-rock-scan

feat: integrate Hudson Rock Infostealer Intelligence

Author: kaifcodec

docs/FLAGS.md

@@ -8,6 +8,7 @@
 | `-ef, --email-file FILE`    | Scan multiple emails from file (one per line)               |
 | `--only-found`              | Only show sites where the username/email was found          |
 | `--allow-loud`              | Enable scanning sites that may send emails/notifications    |
+| `--hudson, --hudson-scan`   | Check for infostealer intelligence using Hudson Rock's API  |
 | `-c, --category CATEGORY`   | Scan all platforms in a specific category                   |
 | `-lu, --list-user`          | List all available modules for username scanning            |
 | `-le, --list-email`         | List all available modules for email scanning               |

tests/test_helpers.py

@@ -1,5 +1,6 @@
 import sys
 import threading
+import json
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
@@ -8,6 +9,7 @@
 from user_scanner.__main__ import main
 from user_scanner.core import helpers
 from user_scanner.core.result import Result
+from user_scanner.core.helpers import _get_config_path, load_config, save_config_value, CONFIG_PATH
 
 
 def test_get_site_name():
@@ -336,3 +338,72 @@ def worker():
         t.join()
 
     assert len(results) == 50
+
+
+def test_get_config_path_default(monkeypatch):
+    monkeypatch.delenv("USER_SCANNER_CONFIG", raising=False)
+    path = _get_config_path()
+    assert path == CONFIG_PATH
+
+def test_get_config_path_env_override(monkeypatch, tmp_path):
+    custom_path = tmp_path / "custom_config.json"
+    monkeypatch.setenv("USER_SCANNER_CONFIG", str(custom_path))
+    path = _get_config_path()
+    assert path == custom_path
+
+def test_load_config_creates_default(tmp_path):
+    config_file = tmp_path / "new_config.json"
+    data = load_config(path=config_file)
+
+    assert config_file.exists()
+    assert data["auto_update_status"] is True
+    assert data["auto_hudson_prompt"] is True
+
+def test_load_config_reads_existing(tmp_path):
+    config_file = tmp_path / "existing.json"
+    existing_data = {
+        "auto_update_status": False,
+        "auto_hudson_prompt": False
+    }
+    config_file.write_text(json.dumps(existing_data))
+
+    data = load_config(path=config_file)
+    assert data == existing_data
+
+def test_save_config_value_updates_keys(tmp_path):
+    config_file = tmp_path / "test_save.json"
+
+    save_config_value("auto_update_status", False, path=config_file)
+    data = load_config(path=config_file)
+    assert data["auto_update_status"] is False
+    assert data["auto_hudson_prompt"] is True
+
+    save_config_value("auto_hudson_prompt", False, path=config_file)
+    data = load_config(path=config_file)
+    assert data["auto_hudson_prompt"] is False
+    assert data["auto_update_status"] is False
+
+def test_load_config_handles_corrupt_json(tmp_path):
+    config_file = tmp_path / "corrupt.json"
+    config_file.write_text("{ 'broken': true, }")
+
+    data = load_config(path=config_file)
+    assert data["auto_update_status"] is True
+    assert data["auto_hudson_prompt"] is True
+
+def test_save_config_value_creates_directory(tmp_path):
+    nested_path = tmp_path / "subdir" / "nested_config.json"
+
+    save_config_value("auto_update_status", True, path=nested_path)
+
+    assert nested_path.exists()
+    assert nested_path.parent.is_dir()
+
+def test_hudson_config_value():
+    from user_scanner.core.helpers import _get_config_path, load_config
+
+    actual_path = _get_config_path()
+    data = load_config(path=actual_path)
+    status = data.get("auto_hudson_prompt")
+
+    assert status is True, f"FAIL: Actual config at {actual_path} has auto_hudson_prompt set to {status} (Expected: True)"

tests/test_update.py

@@ -1,15 +1,17 @@
-from user_scanner.utils import updater_logic as ul
+from user_scanner.core import helpers as hl
+
 
 def test_default_config():
-    configs = ul.load_config()
+    configs = hl.load_config()
     assert "auto_update_status" in configs
     # Make sure config.json has "auto_update_status" set to true
     assert configs["auto_update_status"]
 
+
 def test_config_json(tmp_path, monkeypatch):
     cfg = tmp_path / "config.json"
     monkeypatch.setenv("USER_SCANNER_CONFIG", str(cfg))
-    configs = ul.load_config()
+    configs = hl.load_config()
     assert "auto_update_status" in configs
     # Should be default True
     assert configs["auto_update_status"] is True
@@ -20,10 +22,10 @@ def test_config_set(tmp_path, monkeypatch):
     monkeypatch.setenv("USER_SCANNER_CONFIG", str(cfg))
 
     def get_status():
-        return ul.load_config()["auto_update_status"]
+        return hl.load_config()["auto_update_status"]
 
-    ul.save_config_change(False)
+    hl.save_config_value("auto_update_status", False)
     assert get_status() is False
 
-    ul.save_config_change(True)
+    hl.save_config_value("auto_update_status", True)
     assert get_status() is True

user_scanner/__main__.py

@@ -32,6 +32,7 @@
 )
 from user_scanner.core.result import Status
 from user_scanner.core.version import load_local_version
+from user_scanner.core.hudson import run_hudson_scan
 from user_scanner.utils.update import update_self
 from user_scanner.utils.updater_logic import check_for_updates
 
@@ -134,6 +135,14 @@ def main():
 
     parser.add_argument("-U", "--update", action="store_true", help="Update the tool")
 
+    parser.add_argument(
+        "--hudson", "--hudson-scan",
+        action="store_true",
+        dest="hudson_scan",
+        help="Check for infostealer intelligence using Hudson Rock's API",
+    )
+
+
     parser.add_argument("--version", action="store_true", help="Print version")
 
     args = parser.parse_args()
@@ -209,6 +218,7 @@ def main():
     check_for_updates()
     print_banner()
 
+
     # Handle bulk email file
     if args.email_file:
         try:
@@ -310,6 +320,17 @@ def main():
         else:
             print(f"\n{Fore.CYAN} Checking username: {target}{Style.RESET_ALL}")
 
+
+        if args.hudson_scan:
+            if args.category or args.module:
+                print(f"{R}[✘] Error: --hudson cannot be used with -m or -c {X}")
+                print(f"{Y}[i] Use it independently{X}")
+                sys.exit(1)
+
+            run_hudson_scan(target, is_email)
+            continue
+
+
         if args.module:
             modules = find_module(args.module.replace(".", "_"), is_email)
             fn = run_email_module_batch if is_email else run_user_module
@@ -345,6 +366,10 @@ def main():
             fn = run_email_full_batch if is_email else run_user_full
             results.extend(fn(target, config))
 
+
+    if args.hudson_scan:
+        sys.exit(0)
+
     if args.output:
         content = (
             formatter.into_csv(results)
@@ -386,7 +411,7 @@ def main():
         print(G + f"\n[+] Results saved to {args.output}" + Style.RESET_ALL)
 
     total_found = len([r for r in results if r.is_found()])
-    total_skipped = len([r for r in results if r == Status.SKIPPED])
+    total_skipped = len([r for r in results if r.status == Status.SKIPPED])
 
     if args.only_found and total_found == 0:
         print(f"\n{R}[✘] No results found for the given target(s).{X}")

user_scanner/config.json

@@ -1,3 +1,4 @@
 {
-  "auto_update_status": true
+  "auto_update_status": true,
+  "auto_hudson_prompt": true
 }

user_scanner/core/helpers.py

@@ -4,6 +4,8 @@
 from pathlib import Path
 from typing import Dict, List, Optional
 import inspect
+import json
+import os
 import random
 import threading
 import functools
@@ -27,6 +29,9 @@
     ],
 }
 
+CONFIG_PATH = Path(__file__).parent.parent / "config.json"
+
+
 @dataclass(frozen=True)
 class ScanConfig:
     allow_loud: bool = False
@@ -232,3 +237,47 @@ def get_random_user_agent():
     """return random"""
     random_agent = random.choice(agents)
     return random_agent
+
+
+def _get_config_path(path: str | Path | None = None) -> Path:
+    """
+    Determine the config path in this order:
+      1. explicit path argument (if provided)
+      2. environment variable USER_SCANNER_CONFIG (if set)
+      3. default CONFIG_PATH
+    """
+    if path:
+        return Path(path)
+    env = os.environ.get("USER_SCANNER_CONFIG")
+    if env:
+        return Path(env)
+    return CONFIG_PATH
+
+
+def load_config(path: str | Path | None = None) -> dict:
+    cp = _get_config_path(path)
+    if cp.exists():
+        try:
+            return json.loads(cp.read_text())
+        except json.JSONDecodeError:
+            # This prevents the crash on corrupted JSON
+            pass
+
+    default = {
+        "auto_update_status": True,
+        "auto_hudson_prompt": True
+    }
+    cp.parent.mkdir(parents=True, exist_ok=True)
+    cp.write_text(json.dumps(default, indent=2))
+    return default
+
+
+
+def save_config_value(key: str, value: Any, path: str | Path | None = None):
+    """Generic helper to update any specific key in the config."""
+    cp = _get_config_path(path)
+    content = load_config(path)
+    content[key] = value
+    cp.parent.mkdir(parents=True, exist_ok=True)
+    cp.write_text(json.dumps(content, indent=2))
+

user_scanner/core/hudson.py

@@ -0,0 +1,105 @@
+import httpx
+import json
+from colorama import Fore, Style
+from user_scanner.core.helpers import load_config, _get_config_path
+
+# Color configs
+R = Fore.RED
+G = Fore.GREEN
+C = Fore.CYAN
+Y = Fore.YELLOW
+M = Fore.MAGENTA
+X = Style.RESET_ALL
+
+def update_hudson_preference(show_prompt: bool):
+    """Updates the config file using your existing helper logic."""
+    cp = _get_config_path()
+    content = load_config()
+    content["auto_hudson_prompt"] = show_prompt
+    cp.write_text(json.dumps(content, indent=2))
+
+def check_hudson_permission(target: str) -> bool:
+    """Disclaimers and prompts based on user config."""
+    config = load_config()
+
+    # If the user already set this to false, skip the prompt and run
+    if not config.get("auto_hudson_prompt", True):
+        return True
+
+    print(f"\n{Y}[!] PRIVACY NOTICE & DISCLAIMER:{X}")
+    print("    Hudson Rock is a third-party intelligence service.")
+    print(f"    By proceeding, the identifier '{C}{target}{Y}' will be sent to their API.")
+    print(f"    They may log queries. We are not responsible for their data handling.{X}")
+
+    while True:
+        choice = input(f"\n{C}Query Hudson Rock? (y/n/d for don't ask again): {X}").lower().strip()
+        if choice == 'y':
+            return True
+        elif choice == 'n':
+            print(f"{Y}[i] Hudson Rock scan skipped.{X}")
+            return False
+        elif choice == 'd':
+            update_hudson_preference(False)
+            print(f"{G}[+] Preference saved. You won't be prompted again.{X}")
+            return True
+        else:
+            print(f"{R}[!] Please enter 'y', 'n', or 'd'.{X}")
+
+def run_hudson_scan(target: str, is_email: bool = False):
+    """
+    Fetch and display intelligence from Hudson Rock's OSINT API.
+    """
+    # Permission check using the new logic
+    if not check_hudson_permission(target):
+        return
+
+    base_url = "https://cavalier.hudsonrock.com/api/json/v2/osint-tools/"
+    endpoint = "search-by-email" if is_email else "search-by-username"
+    param = "email" if is_email else "username"
+
+    url = f"{base_url}{endpoint}?{param}={target}"
+
+    print(f"\n{M}== HUDSON ROCK INFOSTEALER INTELLIGENCE =={X}")
+    print(f"{C}[i] Attribution: Data provided by Hudson Rock (https://www.hudsonrock.com){X}")
+    print(f"{C}[*] Querying Hudson Rock for {target}...{X}")
+
+    try:
+        with httpx.Client(timeout=10.0) as client:
+            response = client.get(url)
+
+            if response.status_code == 200:
+                data = response.json()
+                stealers = data.get("stealers", [])
+
+                if not stealers:
+                    print(f"{G}[✔] No infostealer infections found for this {param}.{X}")
+                    return
+
+                total_stealers = len(stealers)
+                print(f"{R}[!] FOUND {total_stealers} INFOSTEALER INFECTION(S) ASSOCIATED WITH THIS {param.upper()}!{X}")
+
+                for i, stealer in enumerate(stealers, 1):
+                    print(f"\n  {Y}Infection #{i}:{X}")
+                    print(f"    - Stealer Family: {stealer.get('stealer_family', 'Unknown')}")
+                    print(f"    - Date Compromised: {stealer.get('date_compromised', 'Unknown')}")
+                    print(f"    - Operating System: {stealer.get('operating_system', 'Unknown')}")
+                    print(f"    - Computer Name: {stealer.get('computer_name', 'Unknown')}")
+
+                    antiviruses = stealer.get('antiviruses', [])
+                    if antiviruses:
+                        print(f"    - Antiviruses: {', '.join(antiviruses)}")
+
+                    top_logins = stealer.get('top_logins', [])
+                    if top_logins:
+                        print(f"    - Sample Logins Found: {', '.join(top_logins[:3])}...")
+
+                print(f"\n{Y}[!] Recommendation: All credentials saved on infected computers are at risk.{X}")
+                print(f"{Y}[!] Visit https://www.hudsonrock.com/free-tools for more info.{X}")
+
+            elif response.status_code == 404:
+                print(f"{G}[✔] No data found for this {param} in Hudson Rock database.{X}")
+            else:
+                print(f"{R}[✘] Hudson Rock API error: HTTP {response.status_code}{X}")
+
+    except Exception as e:
+        print(f"{R}[✘] Error connecting to Hudson Rock: {e}{X}")