feat(cli): automated cross-scanning and metadata pivoting

user_scanner/__main__.py

@@ -164,6 +164,18 @@ def main():
         help="Check for infostealer intelligence using Hudson Rock's API",
     )
 
+    parser.add_argument(
+        "--cross", "--cross-scan",
+        action="store_true",
+        dest="cross_scan",
+        help="Automatically extract and prompt to cross-scan found targets (Username <-> Email)",
+    )
+
+    parser.add_argument(
+        "--cross-auto",
+        action="store_true",
+        help="Automatically run cross-scans on all extracted targets without prompting",
+    )
 
     parser.add_argument("--version", action="store_true", help="Print version")
 
@@ -436,6 +448,82 @@ def main():
             results.extend(fn(target, config))
 
 
+    if args.cross_scan:
+        from user_scanner.core.cross_scanner import extract_emails, extract_usernames, prompt_target_selection
+
+        cross_targets = []
+        target_type = ""
+        cross_is_email = False
+
+        if is_email:
+            cross_is_email = False
+            target_type = "username"
+            cross_targets = extract_usernames(results, targets[0] if targets else "")
+        else:
+            cross_is_email = True
+            target_type = "email"
+            cross_targets = extract_emails(results)
+
+        selected_targets = prompt_target_selection(cross_targets, target_type, args.cross_auto)
+
+        cross_results = []
+        if selected_targets:
+            cross_fn = run_email_full_batch if cross_is_email else run_user_full
+
+            for cross_target in selected_targets:
+                if cross_is_email:
+                    print(f"\n{Fore.CYAN} Cross-checking email: {cross_target}{Style.RESET_ALL}")
+                else:
+                    print(f"\n{Fore.CYAN} Cross-checking username: {cross_target}{Style.RESET_ALL}")
+
+                cross_results.extend(cross_fn(cross_target, config))
+
+            if args.output:
+                base, ext = os.path.splitext(args.output)
+                # Clean targets for filename
+                clean_targets = "_".join("".join(c if c.isalnum() else "_" for c in t) for t in selected_targets)
+                # Truncate if too long
+                if len(clean_targets) > 50:
+                    clean_targets = clean_targets[:47] + "..."
+
+                cross_output = f"{base}_{clean_targets}_{target_type}_cross{ext}"
+
+                # Check for PDF output support
+                if ext.lower() == ".pdf":
+                    try:
+                        from user_scanner.core.formatter import into_pdf
+                        cross_content = into_pdf(
+                            results=cross_results,
+                            target=", ".join(selected_targets),
+                            scan_type="Email" if cross_is_email else "Username",
+                            total_modules=len(load_modules(load_categories(cross_is_email, args.no_nsfw))), # approximate module count
+                            include_media=True,
+                            version=load_local_version()[0]
+                        )
+                        with open(cross_output, "wb") as f:
+                            f.write(cross_content)
+                        print(f"{Fore.GREEN}\n[+] Cross-scan results saved to {cross_output}{Style.RESET_ALL}")
+                    except ImportError:
+                        print(f"{Fore.YELLOW}[i] PDF export requires reportlab. Skipping cross-scan PDF export.{Style.RESET_ALL}")
+                else:
+                    cross_content = (
+                        formatter.into_csv(cross_results)
+                        if args.format == "csv"
+                        else formatter.into_json(cross_results)
+                    )
+
+                    if args.format == "json":
+                        cross_items = formatter.get_json_data(cross_results)
+                        with open(cross_output, "w", encoding="utf-8") as f:
+                            json.dump(cross_items, f, indent=2, ensure_ascii=False)
+                    elif args.format == "csv":
+                        with open(cross_output, "w", encoding="utf-8") as f:
+                            f.write(cross_content)
+
+                    print(f"{Fore.GREEN}\n[+] Cross-scan results saved to {cross_output}{Style.RESET_ALL}")
+
+            results.extend(cross_results)
+
     if args.hudson_scan:
         sys.exit(0)
 

user_scanner/core/cross_scanner.py

@@ -0,0 +1,147 @@
+import re
+from colorama import Fore, Style
+from user_scanner.core.result import Result
+from user_scanner.core.helpers import is_valid_email
+
+R = Fore.RED
+G = Fore.GREEN
+C = Fore.CYAN
+Y = Fore.YELLOW
+X = Style.RESET_ALL
+
+def extract_emails(results: list[Result]) -> list[str]:
+    """Extract emails from the extra dictionary of username scan results."""
+    emails = set()
+    for result in results:
+        if not result.is_found():
+            continue
+        for key, value in result.extra.items():
+            if not isinstance(value, str):
+                continue
+
+            # Check if key implies an email
+            if "email" in key.lower() and is_valid_email(value.strip()):
+                emails.add(value.strip().lower())
+
+            # Sometimes the value itself is a valid email (often buried in bios/descriptions)
+            # A simple regex check helps uncover emails in larger text blocks
+            matches = re.findall(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', value)
+            for match in matches:
+                if is_valid_email(match):
+                    emails.add(match.lower())
+
+    return list(emails)
+
+
+def extract_usernames(results: list[Result], original_email: str) -> list[str]:
+    """Extract usernames from email scan results and the email prefix."""
+    usernames = set()
+
+    # The part before the @ symbol is often the primary username
+    if "@" in original_email:
+        prefix = original_email.split("@")[0]
+        if prefix:
+            usernames.add(prefix)
+
+    # Check OSINT module extra fields for exposed usernames
+    for result in results:
+        if not result.is_found():
+            continue
+        for key, value in result.extra.items():
+            if not isinstance(value, str):
+                continue
+
+            key_lower = key.lower()
+            if "username" in key_lower or "handle" in key_lower:
+                # Some modules return raw usernames in these fields
+                # We should ignore URLs if they somehow ended up here
+                val_strip = value.strip()
+                if val_strip and not val_strip.startswith("http"):
+                    usernames.add(val_strip)
+
+            # Look for linked social accounts in verified fields
+            # For instance, Gravatar might return connected GitHub or Twitter profiles
+            if "verified" in key_lower or "accounts" in key_lower or "wallet" in key_lower:
+                for regex in [
+                    r'github\.com/([^/\s\(\)]+)',
+                    r'twitter\.com/([^/\s\(\)]+)',
+                    r'paypal\.me/([^/\s\(\)]+)',
+                    r'patreon\.com/([^/\s\(\)]+)',
+                    r'venmo\.com/(?:u/)?([^/\s\(\)]+)'
+                ]:
+                    matches = re.findall(regex, value, re.IGNORECASE)
+                    for match in matches:
+                        # Exclude common false positives like "paypal.me/crypto" or API paths if needed
+                        # But for now, grab the clean username
+                        usernames.add(match.strip())
+
+            # Some platforms return URLs as values directly. 
+            # We can run the same URL regexes on the raw value string to catch them.
+            for regex in [
+                r'paypal\.me/([^/\s\(\)]+)',
+                r'patreon\.com/([^/\s\(\)]+)',
+                r'venmo\.com/(?:u/)?([^/\s\(\)]+)'
+            ]:
+                matches = re.findall(regex, value, re.IGNORECASE)
+                for match in matches:
+                    usernames.add(match.strip())
+
+    return list(usernames)
+
+
+def prompt_target_selection(targets: list[str], target_type: str, auto_select: bool = False) -> list[str]:
+    """
+    Presents a list of extracted targets to the user and asks them to select which ones to scan.
+    If auto_select is True, returns all targets without prompting.
+    """
+    if not targets:
+        return []
+
+    if auto_select:
+        print(f"\n{G}[+] Auto-selected {len(targets)} extracted {target_type}(s) for cross-scan.{X}")
+        return targets
+
+    print(f"\n{C}=== EXTRACTED {target_type.upper()}S FOR CROSS-SCAN ==={X}")
+    print(f"{Y}[i] Found {len(targets)} potential {target_type}(s). Select which to scan:{X}")
+
+    for i, target in enumerate(targets, 1):
+        print(f"  {G}[{i}]{X} {target}")
+
+    print(f"  {G}[A]{X} All")
+    print(f"  {Y}[S]{X} Skip (Cancel cross-scan)")
+
+    while True:
+        try:
+            choice = input(f"\n{C}Select numbers (e.g. 1,3), 'A' for all, or 'S' to skip: {X}").strip().lower()
+        except (EOFError, KeyboardInterrupt):
+            print(f"\n{Y}[i] Cross-scan skipped.{X}")
+            return []
+
+        if not choice or choice == 's' or choice == 'skip':
+            print(f"{Y}[i] Cross-scan skipped.{X}")
+            return []
+
+        if choice == 'a' or choice == 'all':
+            return targets
+
+        selected = []
+        parts = choice.replace(" ", "").split(",")
+        valid = True
+
+        for part in parts:
+            if not part.isdigit():
+                valid = False
+                break
+            idx = int(part)
+            if 1 <= idx <= len(targets):
+                selected.append(targets[idx - 1])
+            else:
+                valid = False
+                break
+
+        if not valid:
+            print(f"{R}[!] Invalid selection. Please enter comma-separated numbers (e.g. 1,3), 'A' or 'S'.{X}")
+            continue
+
+        # Deduplicate selections in case user entered "1,1"
+        return list(dict.fromkeys(selected))