diff --git a/pyproject.toml b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "usfx"
-version = "1.0.0"
+version = "1.1.0"
description = "USFX - Ultimate Subdomain Finder X - Discover subdomains via custom DNS servers"
readme = "README.txt"
license = "MIT"
diff --git a/src/usfx/__init__.py b/src/usfx/__init__.py
@@ -6,8 +6,8 @@ using custom DNS servers. Designed for air-gapped environments without
internet connectivity.
"""
-__version__ = "1.0.0"
-__author__ = "NEXTVT Team"
+__version__ = "1.1.0"
+__author__ = "LACRYMARIA Team"
from .config import ScanConfig
from .engine import SubdomainEngine
diff --git a/src/usfx/cli.py b/src/usfx/cli.py
@@ -6,12 +6,19 @@ Supports custom DNS servers for internal network enumeration.
"""
import logging
+import os
+import signal
import sys
+import threading
from pathlib import Path
from typing import List, Optional, Tuple
import click
+# Global state for signal handling
+_current_engine = None
+_interrupted = False
+
from . import __version__
from .config import OutputFormat, ScanConfig, WordlistSize
from .engine import SubdomainEngine
@@ -260,29 +267,61 @@ def main(
)
# Create engine and run scan
+ global _current_engine, _interrupted
engine = SubdomainEngine()
+ _current_engine = engine
+ _interrupted = False
+
+ # Signal handler for Ctrl+C
+ def signal_handler(signum, frame):
+ global _interrupted
+ if _interrupted:
+ # Second Ctrl+C - force exit immediately
+ if not quiet:
+ click.echo("\nForce exit.", err=True)
+ os._exit(130)
+
+ _interrupted = True
+ if not quiet:
+ click.echo("\n\nInterrupted. Stopping scan...", err=True)
+
+ if _current_engine:
+ _current_engine.cancel_scan()
+
+ # Register signal handler
+ original_handler = signal.signal(signal.SIGINT, signal_handler)
# Progress tracking
last_phase = None
def progress_callback(state):
nonlocal last_phase
+ if _interrupted:
+ return
if state.current_phase and state.current_phase != last_phase:
console.print_phase(state.current_phase.value)
last_phase = state.current_phase
try:
result = engine.scan(config, progress_callback=progress_callback if not quiet else None)
+
+ if _interrupted:
+ sys.exit(130)
+
except KeyboardInterrupt:
- console.print_error("Scan cancelled by user")
- engine.cancel_scan()
+ if not quiet:
+ click.echo("\nScan cancelled.", err=True)
sys.exit(130)
except Exception as e:
- console.print_error(f"Scan failed: {e}")
- if verbose:
- import traceback
- traceback.print_exc()
+ if not _interrupted:
+ console.print_error(f"Scan failed: {e}")
+ if verbose:
+ import traceback
+ traceback.print_exc()
sys.exit(1)
+ finally:
+ signal.signal(signal.SIGINT, original_handler)
+ _current_engine = None
# Print results
console.print_result_table(result)
diff --git a/src/usfx/modules/base.py b/src/usfx/modules/base.py
@@ -6,8 +6,9 @@ Provides common functionality and interface contract.
"""
import logging
+import threading
from abc import ABC, abstractmethod
-from typing import Dict, Optional, TYPE_CHECKING
+from typing import Callable, Dict, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from ..utils.dns_resolver import InternalDNSResolver
@@ -35,7 +36,8 @@ class BaseModule(ABC):
self,
domain: str,
resolver: Optional['InternalDNSResolver'] = None,
- timeout: float = 3.0
+ timeout: float = 3.0,
+ cancel_check: Optional[Callable[[], bool]] = None
):
"""
Initialize the module.
@@ -44,10 +46,13 @@ class BaseModule(ABC):
domain: Target domain (e.g., 'corp.local')
resolver: Optional custom DNS resolver
timeout: Default timeout for operations
+ cancel_check: Optional function that returns True if cancelled
"""
self.domain = domain.lower().strip('.')
self.timeout = timeout
self.discovered: Dict[str, Optional[str]] = {}
+ self._cancel_check = cancel_check
+ self._cancelled = False
# Use provided resolver or create a new one
if resolver:
@@ -56,6 +61,19 @@ class BaseModule(ABC):
from ..utils.dns_resolver import InternalDNSResolver
self.resolver = InternalDNSResolver(timeout=timeout)
+ def is_cancelled(self) -> bool:
+ """Check if the module should stop"""
+ if self._cancelled:
+ return True
+ if self._cancel_check and self._cancel_check():
+ self._cancelled = True
+ return True
+ return False
+
+ def cancel(self):
+ """Cancel the module"""
+ self._cancelled = True
+
@abstractmethod
def enumerate(self, **kwargs) -> Dict[str, Optional[str]]:
"""
diff --git a/src/usfx/modules/dns_bruteforce.py b/src/usfx/modules/dns_bruteforce.py
@@ -31,7 +31,8 @@ class DNSBruteforcer(BaseModule):
wordlist_path: Optional[str] = None,
wordlist_size: str = 'medium',
threads: int = 30,
- timeout: float = 2.0
+ timeout: float = 2.0,
+ cancel_check: Optional[Callable[[], bool]] = None
):
"""
Initialize DNS brute forcer.
@@ -43,8 +44,9 @@ class DNSBruteforcer(BaseModule):
wordlist_size: Size of bundled wordlist ('small', 'medium', 'large')
threads: Number of parallel threads
timeout: DNS query timeout
+ cancel_check: Optional function that returns True if cancelled
"""
- super().__init__(domain, resolver, timeout)
+ super().__init__(domain, resolver, timeout, cancel_check)
self.threads = threads
self.wordlist_path = wordlist_path
@@ -84,6 +86,9 @@ class DNSBruteforcer(BaseModule):
Returns:
Dict of {subdomain: ip_address}
"""
+ if self.is_cancelled():
+ return self.discovered
+
logger.info(f"Starting DNS brute force for {self.domain}")
# Detect wildcards first
@@ -91,6 +96,9 @@ class DNSBruteforcer(BaseModule):
if has_wildcard:
logger.warning(f"Wildcard DNS detected for {self.domain}: {wildcard_ips}")
+ if self.is_cancelled():
+ return self.discovered
+
# Load wordlist
wordlist = self.wordlist_manager.get_wordlist(
size=self.wordlist_size,
@@ -112,8 +120,17 @@ class DNSBruteforcer(BaseModule):
}
for future in as_completed(futures):
+ # Check for cancellation
+ if self.is_cancelled():
+ # Cancel remaining futures and exit
+ for f in futures:
+ f.cancel()
+ executor.shutdown(wait=False, cancel_futures=True)
+ logger.info("DNS brute force cancelled")
+ return self.discovered
+
try:
- fqdn, ip, all_ips = future.result()
+ fqdn, ip, all_ips = future.result(timeout=self.timeout + 1)
completed += 1
if ip:
diff --git a/src/usfx/orchestrator.py b/src/usfx/orchestrator.py
@@ -161,6 +161,16 @@ class SubdomainOrchestrator:
"""Run a single module as a task"""
start_time = time.time()
+ # Check cancellation before starting
+ if self.state.is_cancelled:
+ return ModuleResult(
+ module_name=module_name,
+ found_count=0,
+ subdomains={},
+ duration_seconds=0,
+ error="cancelled"
+ )
+
try:
module = module_factory()
@@ -173,6 +183,8 @@ class SubdomainOrchestrator:
results = module.enumerate(known_hosts=known_subs)
elif module_name == "permutation":
def resolve(subdomain):
+ if self.state.is_cancelled:
+ return None
return self.resolver.resolve_a(subdomain)
results = module.enumerate(known_subs, resolver_func=resolve)
else:
@@ -180,6 +192,16 @@ class SubdomainOrchestrator:
else:
results = module.enumerate()
+ # Check cancellation after module completion
+ if self.state.is_cancelled:
+ return ModuleResult(
+ module_name=module_name,
+ found_count=0,
+ subdomains={},
+ duration_seconds=time.time() - start_time,
+ error="cancelled"
+ )
+
duration = time.time() - start_time
self._add_results(results, module_name)
@@ -193,6 +215,14 @@ class SubdomainOrchestrator:
)
except Exception as e:
+ if self.state.is_cancelled:
+ return ModuleResult(
+ module_name=module_name,
+ found_count=0,
+ subdomains={},
+ duration_seconds=time.time() - start_time,
+ error="cancelled"
+ )
logger.error(f"Error in module {module_name}: {e}")
return ModuleResult(
module_name=module_name,
@@ -213,12 +243,17 @@ class SubdomainOrchestrator:
completed_count = 0
total_modules = len(module_tasks)
+ if self.state.is_cancelled:
+ return results
+
module_names = [name for _, name, _ in module_tasks]
self._update_progress(phase, f"parallel: {', '.join(module_names)}", 0)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures: Dict[Future, str] = {}
for module_factory, module_name, kwargs in module_tasks:
+ if self.state.is_cancelled:
+ break
future = executor.submit(
self._run_module_task,
module_factory,
@@ -228,6 +263,13 @@ class SubdomainOrchestrator:
futures[future] = module_name
for future in as_completed(futures):
+ if self.state.is_cancelled:
+ # Cancel remaining futures
+ for f in futures:
+ f.cancel()
+ executor.shutdown(wait=False, cancel_futures=True)
+ break
+
module_name = futures[future]
try:
result = future.result(timeout=300)
@@ -246,11 +288,6 @@ class SubdomainOrchestrator:
progress = int(completed_count * 100 / total_modules)
self._update_progress(phase, module_name, progress)
- if self.state.is_cancelled:
- for f in futures:
- f.cancel()
- break
-
return results
def _get_enabled_modules(self) -> Set[str]:
@@ -316,6 +353,9 @@ class SubdomainOrchestrator:
wordlist_size = self.config.wordlist_size.value if self.config else 'medium'
wordlist_path = str(self.config.custom_wordlist) if self.config and self.config.custom_wordlist else None
+ # Cancellation check function for modules
+ cancel_check = lambda: self.state.is_cancelled
+
module_tasks = []
if "dns_bruteforce" in enabled:
@@ -324,7 +364,8 @@ class SubdomainOrchestrator:
self.domain, self.resolver,
wordlist_path=wordlist_path,
wordlist_size=wordlist_size,
- threads=threads
+ threads=threads,
+ cancel_check=cancel_check
),
"dns_bruteforce",
{}
diff --git a/src/usfx/utils/output.py b/src/usfx/utils/output.py
@@ -241,24 +241,52 @@ class ConsoleOutput:
if self.quiet:
return
- ascii_art = r"""
- __ __ _____ _______ __
- / / / // ___// ____/ |/ /
- / / / / \__ \/ /_ | /
-/ /_/ / ___/ / __/ / |
-\____/ /____/_/ /_/|_|
-"""
- subtitle = f"Ultimate Subdomain Finder X v{version}"
- tagline = "Internal Network Subdomain Discovery Tool"
-
if self.use_rich:
- self.console.print(ascii_art, style="bold cyan", highlight=False)
- self.console.print(f" {subtitle}", style="bold white")
- self.console.print(f" {tagline}\n", style="dim")
+ # Big ASCII art with gradient
+ art_lines = [
+ " ██╗ ██╗███████╗███████╗██╗ ██╗",
+ " ██║ ██║██╔════╝██╔════╝╚██╗██╔╝",
+ " ██║ ██║███████╗█████╗ ╚███╔╝ ",
+ " ██║ ██║╚════██║██╔══╝ ██╔██╗ ",
+ " ╚██████╔╝███████║██║ ██╔╝ ██╗",
+ " ╚═════╝ ╚══════╝╚═╝ ╚═╝ ╚═╝",
+ ]
+
+ colors = [
+ "bright_cyan",
+ "cyan",
+ "bright_blue",
+ "blue",
+ "bright_magenta",
+ "magenta",
+ ]
+
+ self.console.print()
+ for i, line in enumerate(art_lines):
+ color = colors[i % len(colors)]
+ self.console.print(f" [{color}]{line}[/{color}]", highlight=False)
+
+ self.console.print()
+ self.console.print(f" [bold bright_white]Ultimate Subdomain Finder X[/bold bright_white] [bright_yellow]v{version}[/bright_yellow]")
+ self.console.print(f" [dim italic]Internal Network Subdomain Discovery Tool[/dim italic]")
+ self.console.print()
+ self.console.print(f" [bright_green]◆[/bright_green] [dim]Offline Mode[/dim] [bright_blue]◆[/bright_blue] [dim]Custom DNS[/dim] [bright_magenta]◆[/bright_magenta] [dim]10 Modules[/dim]")
+ self.console.print()
else:
+ ascii_art = r"""
+ ██╗ ██╗███████╗███████╗██╗ ██╗
+ ██║ ██║██╔════╝██╔════╝╚██╗██╔╝
+ ██║ ██║███████╗█████╗ ╚███╔╝
+ ██║ ██║╚════██║██╔══╝ ██╔██╗
+ ╚██████╔╝███████║██║ ██╔╝ ██╗
+ ╚═════╝ ╚══════╝╚═╝ ╚═╝ ╚═╝
+"""
print(ascii_art)
- print(f" {subtitle}")
- print(f" {tagline}\n")
+ print(f" Ultimate Subdomain Finder X v{version}")
+ print(f" Internal Network Subdomain Discovery Tool")
+ print()
+ print(f" * Offline Mode * Custom DNS * 10 Modules")
+ print()
def print_config(self, domain: str, dns_servers: List[str], modules: List[str]) -> None:
"""Print scan configuration"""
@@ -266,20 +294,18 @@ class ConsoleOutput:
return
if self.use_rich:
- self.console.print(f"\n[bold]Target:[/bold] {domain}")
+ self.console.print(f" [bold]Target:[/bold] [bright_white]{domain}[/bright_white]")
if dns_servers:
- self.console.print(f"[bold]DNS Servers:[/bold] {', '.join(dns_servers)}")
+ self.console.print(f" [bold]DNS:[/bold] [dim]{', '.join(dns_servers)}[/dim]")
else:
- self.console.print("[bold]DNS Servers:[/bold] system default")
- self.console.print(f"[bold]Modules:[/bold] {', '.join(modules)}")
+ self.console.print(f" [bold]DNS:[/bold] [dim]system default[/dim]")
self.console.print()
else:
- print(f"\nTarget: {domain}")
+ print(f" Target: {domain}")
if dns_servers:
- print(f"DNS Servers: {', '.join(dns_servers)}")
+ print(f" DNS: {', '.join(dns_servers)}")
else:
- print("DNS Servers: system default")
- print(f"Modules: {', '.join(modules)}")
+ print(f" DNS: system default")
print()
def print_phase(self, phase: str) -> None:
|