diff --git a/README.txt b/README.txt
@@ -11,7 +11,7 @@ FEATURES
- Custom DNS Server Support: Specify internal DNS servers
- Offline-Only Modules: All techniques work without internet
- - 10 Enumeration Modules:
+ - 12 Enumeration Modules:
* DNS Brute Force
* Zone Transfer (AXFR)
* DNSSEC Walking (NSEC/NSEC3)
@@ -22,6 +22,9 @@ FEATURES
* Recursive Sub-subdomain Enumeration
* Virtual Host Discovery
* TLS Certificate SAN Extraction
+ * Subdomain Takeover Detection (NEW)
+ * Web Technology Detection with Wappalyzer (NEW)
+ - Pipeline Modes: Output formats for tool chaining (subs, web, ips, json)
- Multiple Output Formats: JSON, CSV, TXT
- Bundled Wordlists: Small (~500), Medium (~3500), Large (~18000)
- Progress Tracking: Real-time progress with Rich terminal UI
@@ -34,11 +37,14 @@ From PyPI (Recommended):
pip install usfx
+ # With Wappalyzer support for web technology detection
+ pip install usfx[webtech]
+
From GitHub Releases:
# Download wheel from:
# https://github.com/devastator-x/usfx/releases/latest
- pip install usfx-1.1.0-py3-none-any.whl
+ pip install usfx-1.2.0-py3-none-any.whl
From Source (Development):
@@ -60,7 +66,7 @@ Wordlist Options:
usfx corp.local -d 10.0.0.1 -s small # ~500 words
usfx corp.local -d 10.0.0.1 -s medium # ~3500 words (default)
- usfx corp.local -d 10.0.0.1 -s large # ~10000 words
+ usfx corp.local -d 10.0.0.1 -s large # ~18000 words
usfx corp.local -d 10.0.0.1 -w /path/to/custom.txt
Output Options:
@@ -84,6 +90,33 @@ Module Selection:
recursive - Recursive sub-subdomain enumeration
vhost - Virtual host discovery
tls - TLS certificate analysis
+ takeover - Subdomain takeover detection
+ webtech - Web technology detection
+
+Extended Scanning (NEW):
+
+ # Subdomain takeover vulnerability detection
+ usfx corp.local -d 10.0.0.1 --takeover
+
+ # Web technology detection (requires python-Wappalyzer)
+ usfx corp.local -d 10.0.0.1 --web-tech
+
+ # Custom web ports for tech detection
+ usfx corp.local -d 10.0.0.1 --web-tech --web-ports 80,443,8080,8443
+
+Pipeline Modes (NEW):
+
+ # Output only subdomains (one per line) - pipe to other tools
+ usfx corp.local -d 10.0.0.1 --pipe-subs | httpx
+
+ # Output only web URLs
+ usfx corp.local -d 10.0.0.1 --web-tech --pipe-web
+
+ # Output only IP addresses
+ usfx corp.local -d 10.0.0.1 --pipe-ips | nmap -iL -
+
+ # JSON output to stdout (for jq processing)
+ usfx corp.local -d 10.0.0.1 --pipe-json | jq '.subdomains'
Advanced Options:
@@ -110,6 +143,13 @@ Options:
-m, --modules TEXT Comma-separated module list
--reverse-range TEXT CIDR range for reverse DNS
--vhost-ip TEXT IP for vhost scanning
+ --takeover Enable subdomain takeover detection
+ --web-tech Enable web technology detection
+ --web-ports TEXT Ports for web tech scanning (default: 80,443,8080,8443)
+ --pipe-subs Pipeline: output subdomains only
+ --pipe-web Pipeline: output web URLs only
+ --pipe-ips Pipeline: output IPs only
+ --pipe-json Pipeline: JSON to stdout
-v, --verbose Verbose output
-q, --quiet Suppress non-essential output
--no-color Disable colored output
@@ -128,7 +168,9 @@ PYTHON API
dns_servers=['10.0.0.1', '10.0.0.2'],
wordlist_size=WordlistSize.MEDIUM,
threads=50,
- timeout=3.0
+ timeout=3.0,
+ takeover=True, # Enable takeover detection
+ web_tech=True, # Enable web tech detection
)
engine = SubdomainEngine()
@@ -138,6 +180,14 @@ PYTHON API
for sub in result.subdomains:
print(f" {sub.subdomain} -> {sub.ip}")
+ # Takeover vulnerabilities
+ for vuln in result.takeover_results:
+ print(f" VULN: {vuln.subdomain} -> {vuln.service}")
+
+ # Web technologies
+ for web in result.web_tech_results:
+ print(f" WEB: {web.url} - {', '.join(web.technologies)}")
+
MODULE DESCRIPTIONS
-------------------
@@ -152,6 +202,8 @@ MODULE DESCRIPTIONS
recursive_enum Sub-subdomain discovery Medium
vhost_scanner Host header brute force Slow
tls_analyzer TLS certificate SAN extraction Medium
+ takeover Subdomain takeover detection Fast
+ web_tech Web technology detection Medium
REQUIREMENTS
@@ -164,12 +216,15 @@ REQUIREMENTS
- cryptography >= 41.0.0
- rich >= 13.0.0
+Optional:
+ - python-Wappalyzer >= 0.3.1 (for web technology detection)
+
USE CASES
---------
Internal Network Penetration Testing:
- usfx corp.internal -d 10.0.0.53 -s large -o findings.json
+ usfx corp.internal -d 10.0.0.53 -s large --takeover -o findings.json
Active Directory Reconnaissance:
usfx ad.corp.local -d 192.168.1.1 -m records,reverse,zone
@@ -177,6 +232,12 @@ Active Directory Reconnaissance:
IT Asset Discovery:
usfx internal.company -d 172.16.0.1 --reverse-range 172.16.0.0/16
+Web Application Mapping:
+ usfx corp.local -d 10.0.0.1 --web-tech --pipe-web | httpx -silent
+
+Tool Integration:
+ usfx corp.local --pipe-subs | nuclei -t takeovers/
+
LICENSE
-------
diff --git a/pyproject.toml b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "usfx"
-version = "1.1.0"
+version = "1.2.0"
description = "USFX - Ultimate Subdomain Finder X - Discover subdomains via custom DNS servers"
readme = "README.txt"
license = "MIT"
@@ -42,6 +42,9 @@ dev = [
"black>=23.0.0",
"ruff>=0.1.0",
]
+webtech = [
+ "python-Wappalyzer>=0.3.1",
+]
[project.scripts]
usfx = "usfx.cli:main"
diff --git a/src/usfx/__init__.py b/src/usfx/__init__.py
@@ -6,7 +6,7 @@ using custom DNS servers. Designed for air-gapped environments without
internet connectivity.
"""
-__version__ = "1.1.0"
+__version__ = "1.2.0"
__author__ = "LACRYMARIA Team"
from .config import ScanConfig
diff --git a/src/usfx/cli.py b/src/usfx/cli.py
@@ -20,7 +20,7 @@ _current_engine = None
_interrupted = False
from . import __version__
-from .config import OutputFormat, ScanConfig, WordlistSize
+from .config import OutputFormat, PipelineMode, ScanConfig, WordlistSize
from .engine import SubdomainEngine
from .utils.output import ConsoleOutput, OutputFormatter
@@ -60,7 +60,9 @@ def parse_modules(modules_str: Optional[str]) -> Optional[set]:
'permutation',
'recursive', 'recursive_enum',
'vhost', 'vhost_scanner',
- 'tls', 'tls_analyzer'
+ 'tls', 'tls_analyzer',
+ 'takeover',
+ 'webtech', 'web_tech',
}
# Normalize module names
@@ -74,6 +76,7 @@ def parse_modules(modules_str: Optional[str]) -> Optional[set]:
'recursive': 'recursive_enum',
'vhost': 'vhost_scanner',
'tls': 'tls_analyzer',
+ 'webtech': 'web_tech',
}
modules = set()
@@ -152,6 +155,43 @@ def parse_modules(modules_str: Optional[str]) -> Optional[set]:
multiple=True,
help='IP address for virtual host scanning'
)
+@click.option(
+ '--takeover',
+ is_flag=True,
+ help='Enable subdomain takeover vulnerability detection'
+)
+@click.option(
+ '--web-tech',
+ 'web_tech',
+ is_flag=True,
+ help='Enable web technology detection (Wappalyzer)'
+)
+@click.option(
+ '--web-ports',
+ 'web_ports',
+ default='80,443,8080,8443',
+ help='Comma-separated ports for web tech scanning (default: 80,443,8080,8443)'
+)
+@click.option(
+ '--pipe-subs',
+ is_flag=True,
+ help='Pipeline mode: output only subdomains (one per line)'
+)
+@click.option(
+ '--pipe-web',
+ is_flag=True,
+ help='Pipeline mode: output only web URLs (one per line)'
+)
+@click.option(
+ '--pipe-ips',
+ is_flag=True,
+ help='Pipeline mode: output only IP addresses (one per line)'
+)
+@click.option(
+ '--pipe-json',
+ is_flag=True,
+ help='Pipeline mode: JSON output to stdout (no banner/progress)'
+)
@click.option(
'-v', '--verbose',
is_flag=True,
@@ -180,6 +220,13 @@ def main(
modules_str: Optional[str],
reverse_ranges: Tuple[str, ...],
vhost_ips: Tuple[str, ...],
+ takeover: bool,
+ web_tech: bool,
+ web_ports: str,
+ pipe_subs: bool,
+ pipe_web: bool,
+ pipe_ips: bool,
+ pipe_json: bool,
verbose: bool,
quiet: bool,
no_color: bool
@@ -220,7 +267,41 @@ def main(
# Scan with vhost discovery on specific IP
usfx corp.local -d 10.0.0.1 --vhost-ip 192.168.1.100
+
+ # Enable subdomain takeover detection
+ usfx corp.local -d 10.0.0.1 --takeover
+
+ # Enable web technology detection
+ usfx corp.local -d 10.0.0.1 --web-tech
+
+ # Pipeline mode: subdomains only (for piping to other tools)
+ usfx corp.local -d 10.0.0.1 --pipe-subs | httpx
+
+ # Pipeline mode: JSON to stdout
+ usfx corp.local -d 10.0.0.1 --pipe-json | jq '.subdomains'
"""
+ # Determine pipeline mode
+ pipeline_mode = PipelineMode.NONE
+ if pipe_subs:
+ pipeline_mode = PipelineMode.SUBS
+ elif pipe_web:
+ pipeline_mode = PipelineMode.WEB
+ elif pipe_ips:
+ pipeline_mode = PipelineMode.IPS
+ elif pipe_json:
+ pipeline_mode = PipelineMode.JSON
+
+ # Pipeline modes force quiet mode (no banner/progress)
+ if pipeline_mode != PipelineMode.NONE:
+ quiet = True
+
+ # Parse web ports
+ try:
+ parsed_web_ports = [int(p.strip()) for p in web_ports.split(',')]
+ except ValueError:
+ click.echo("Error: Invalid web ports format", err=True)
+ sys.exit(1)
+
# Setup logging
setup_logging(verbose, quiet)
@@ -249,8 +330,12 @@ def main(
modules=modules,
reverse_dns_ranges=list(reverse_ranges),
vhost_ips=list(vhost_ips),
+ takeover=takeover,
+ web_tech=web_tech,
+ web_ports=parsed_web_ports,
output_file=Path(output_path) if output_path else None,
output_format=OutputFormat(output_format),
+ pipeline_mode=pipeline_mode,
verbose=verbose,
quiet=quiet
)
@@ -332,9 +417,22 @@ def main(
signal.signal(signal.SIGINT, original_handler)
_current_engine = None
- # Print results
- console.print_result_table(result)
- console.print_summary(result)
+ # Handle pipeline mode output
+ if pipeline_mode != PipelineMode.NONE:
+ output = OutputFormatter.to_pipeline(result, pipeline_mode)
+ click.echo(output)
+ else:
+ # Print results
+ console.print_result_table(result)
+ console.print_summary(result)
+
+ # Print takeover results if available
+ if result.takeover_results:
+ console.print_takeover_results(result.takeover_results)
+
+ # Print web tech results if available
+ if result.web_tech_results:
+ console.print_web_tech_results(result.web_tech_results)
# Write output file if requested
if config.output_file:
diff --git a/src/usfx/config.py b/src/usfx/config.py
@@ -24,6 +24,15 @@ class OutputFormat(Enum):
TXT = "txt"
+class PipelineMode(Enum):
+ """Pipeline output modes for integration with other tools"""
+ NONE = "none" # Normal output
+ SUBS = "subs" # Only subdomains (one per line)
+ WEB = "web" # Only web servers (URLs)
+ IPS = "ips" # Only IP addresses
+ JSON = "json" # JSON to stdout (no banner/progress)
+
+
@dataclass
class ScanConfig:
"""Configuration for a subdomain scan"""
@@ -49,9 +58,15 @@ class ScanConfig:
reverse_dns_ranges: List[str] = field(default_factory=list) # CIDR ranges
vhost_ips: List[str] = field(default_factory=list) # IPs for vhost scanning
+ # Extended scan options
+ takeover: bool = False # Enable subdomain takeover detection
+ web_tech: bool = False # Enable web technology detection
+ web_ports: List[int] = field(default_factory=lambda: [80, 443, 8080, 8443])
+
# Output configuration
output_file: Optional[Path] = None
output_format: OutputFormat = OutputFormat.JSON
+ pipeline_mode: PipelineMode = field(default_factory=lambda: PipelineMode.NONE)
# Verbosity
verbose: bool = False
@@ -92,10 +107,18 @@ class ScanConfig:
"recursive_enum",
"vhost_scanner",
"tls_analyzer",
+ "takeover",
+ "web_tech",
}
if self.modules is None:
- return all_modules
+ # Default: exclude takeover and web_tech unless explicitly enabled
+ enabled = all_modules - {"takeover", "web_tech"}
+ if self.takeover:
+ enabled.add("takeover")
+ if self.web_tech:
+ enabled.add("web_tech")
+ return enabled
return self.modules & all_modules
@@ -120,6 +143,28 @@ class ModuleResult:
error: Optional[str] = None
+@dataclass
+class TakeoverResult:
+ """Takeover vulnerability result"""
+ subdomain: str
+ cname: str
+ service: str
+ status: str # "vulnerable" or "potential"
+ reason: str
+
+
+@dataclass
+class WebTechResult:
+ """Web technology detection result"""
+ subdomain: str
+ url: str
+ port: int
+ status: int
+ title: Optional[str]
+ server: Optional[str]
+ technologies: List[str]
+
+
@dataclass
class ScanResult:
"""Complete scan result"""
@@ -130,3 +175,7 @@ class ScanResult:
duration_seconds: float
dns_servers_used: List[str]
config: ScanConfig
+
+ # Extended results
+ takeover_results: List[TakeoverResult] = field(default_factory=list)
+ web_tech_results: List[WebTechResult] = field(default_factory=list)
diff --git a/src/usfx/engine.py b/src/usfx/engine.py
@@ -15,7 +15,7 @@ import logging
import time
from typing import List, Optional
-from .config import ModuleResult, ScanConfig, ScanResult, SubdomainResult
+from .config import ModuleResult, ScanConfig, ScanResult, SubdomainResult, TakeoverResult, WebTechResult
from .orchestrator import SubdomainOrchestrator
logger = logging.getLogger(__name__)
@@ -105,6 +105,30 @@ class SubdomainEngine:
error=module.error
))
+ # Build takeover results
+ takeover_results = []
+ for t in state.takeover_results:
+ takeover_results.append(TakeoverResult(
+ subdomain=t['subdomain'],
+ cname=t['cname'],
+ service=t['service'],
+ status=t['status'],
+ reason=t['reason']
+ ))
+
+ # Build web tech results
+ web_tech_results = []
+ for w in state.web_tech_results:
+ web_tech_results.append(WebTechResult(
+ subdomain=w['subdomain'],
+ url=w['url'],
+ port=w['port'],
+ status=w['status'],
+ title=w.get('title'),
+ server=w.get('server'),
+ technologies=w.get('technologies', [])
+ ))
+
return ScanResult(
domain=config.domain,
total_found=state.total_found,
@@ -112,7 +136,9 @@ class SubdomainEngine:
module_results=module_results,
duration_seconds=duration,
dns_servers_used=resolver.dns_servers,
- config=config
+ config=config,
+ takeover_results=takeover_results,
+ web_tech_results=web_tech_results
)
except Exception as e:
diff --git a/src/usfx/modules/__init__.py b/src/usfx/modules/__init__.py
@@ -12,6 +12,8 @@ Offline-capable modules for subdomain enumeration:
- recursive_enum: Sub-subdomain discovery
- vhost_scanner: Host header brute force
- tls_analyzer: TLS certificate SAN extraction
+- takeover: Subdomain takeover vulnerability detection
+- web_tech: Web technology detection (Wappalyzer)
"""
from .base import BaseModule
@@ -25,6 +27,8 @@ from .permutation import PermutationEngine
from .recursive_enum import RecursiveEnumerator
from .vhost_scanner import VHostScanner
from .tls_analyzer import TLSAnalyzer
+from .takeover import TakeoverDetector
+from .web_tech import WebTechDetector
__all__ = [
"BaseModule",
@@ -38,4 +42,6 @@ __all__ = [
"RecursiveEnumerator",
"VHostScanner",
"TLSAnalyzer",
+ "TakeoverDetector",
+ "WebTechDetector",
]
diff --git a/src/usfx/modules/takeover.py b/src/usfx/modules/takeover.py
@@ -0,0 +1,249 @@
+"""
+Subdomain Takeover Detection Module
+
+Detects potential subdomain takeover vulnerabilities by checking
+CNAME records pointing to unclaimed cloud services.
+
+Works offline - only requires DNS access to target network.
+"""
+
+import logging
+from typing import Dict, List, Optional, Set, Tuple, TYPE_CHECKING
+
+from .base import BaseModule
+
+if TYPE_CHECKING:
+ from ..utils.dns_resolver import InternalDNSResolver
+
+logger = logging.getLogger(__name__)
+
+
+# Fingerprints for subdomain takeover detection
+# Format: (cname_pattern, service_name, is_vulnerable_check)
+TAKEOVER_FINGERPRINTS = [
+ # AWS
+ (".s3.amazonaws.com", "AWS S3", ["NoSuchBucket"]),
+ (".s3-website", "AWS S3 Website", ["NoSuchBucket"]),
+ (".cloudfront.net", "AWS CloudFront", ["Bad request", "ERROR: The request could not be satisfied"]),
+ (".elasticbeanstalk.com", "AWS Elastic Beanstalk", ["NXDOMAIN"]),
+
+ # Azure
+ (".azurewebsites.net", "Azure App Service", ["NXDOMAIN"]),
+ (".cloudapp.azure.com", "Azure Cloud App", ["NXDOMAIN"]),
+ (".azure-api.net", "Azure API Management", ["NXDOMAIN"]),
+ (".azureedge.net", "Azure CDN", ["NXDOMAIN"]),
+ (".blob.core.windows.net", "Azure Blob Storage", ["BlobNotFound", "NXDOMAIN"]),
+ (".trafficmanager.net", "Azure Traffic Manager", ["NXDOMAIN"]),
+
+ # Google Cloud
+ (".storage.googleapis.com", "Google Cloud Storage", ["NoSuchBucket"]),
+ (".appspot.com", "Google App Engine", ["NXDOMAIN"]),
+
+ # GitHub
+ (".github.io", "GitHub Pages", ["There isn't a GitHub Pages site here", "NXDOMAIN"]),
+ (".githubusercontent.com", "GitHub User Content", ["NXDOMAIN"]),
+
+ # Heroku
+ (".herokuapp.com", "Heroku", ["No such app", "NXDOMAIN"]),
+ (".herokudns.com", "Heroku DNS", ["NXDOMAIN"]),
+
+ # Shopify
+ (".myshopify.com", "Shopify", ["Sorry, this shop is currently unavailable", "NXDOMAIN"]),
+
+ # Tumblr
+ (".tumblr.com", "Tumblr", ["There's nothing here", "NXDOMAIN"]),
+
+ # WordPress
+ (".wordpress.com", "WordPress.com", ["Do you want to register", "NXDOMAIN"]),
+
+ # Zendesk
+ (".zendesk.com", "Zendesk", ["Help Center Closed", "NXDOMAIN"]),
+
+ # Fastly
+ (".fastly.net", "Fastly", ["Fastly error: unknown domain", "NXDOMAIN"]),
+
+ # Pantheon
+ (".pantheonsite.io", "Pantheon", ["The gods are wise", "NXDOMAIN"]),
+
+ # Ghost
+ (".ghost.io", "Ghost", ["NXDOMAIN"]),
+
+ # Surge
+ (".surge.sh", "Surge", ["project not found", "NXDOMAIN"]),
+
+ # Bitbucket
+ (".bitbucket.io", "Bitbucket", ["Repository not found", "NXDOMAIN"]),
+
+ # Netlify
+ (".netlify.app", "Netlify", ["Not Found", "NXDOMAIN"]),
+ (".netlify.com", "Netlify", ["Not Found", "NXDOMAIN"]),
+
+ # Vercel
+ (".vercel.app", "Vercel", ["NXDOMAIN"]),
+ (".now.sh", "Vercel (now.sh)", ["NXDOMAIN"]),
+
+ # Firebase
+ (".firebaseapp.com", "Firebase", ["NXDOMAIN"]),
+ (".web.app", "Firebase Hosting", ["NXDOMAIN"]),
+
+ # Fly.io
+ (".fly.dev", "Fly.io", ["NXDOMAIN"]),
+
+ # Render
+ (".onrender.com", "Render", ["NXDOMAIN"]),
+
+ # Railway
+ (".up.railway.app", "Railway", ["NXDOMAIN"]),
+
+ # Cargo Collective
+ (".cargocollective.com", "Cargo", ["NXDOMAIN"]),
+
+ # Feedpress
+ (".redirect.feedpress.me", "Feedpress", ["NXDOMAIN"]),
+
+ # Unbounce
+ (".unbouncepages.com", "Unbounce", ["NXDOMAIN"]),
+
+ # Help Scout
+ (".helpscoutdocs.com", "Help Scout", ["NXDOMAIN"]),
+
+ # Tilda
+ (".tilda.ws", "Tilda", ["NXDOMAIN"]),
+
+ # Webflow
+ (".webflow.io", "Webflow", ["NXDOMAIN"]),
+
+ # Kinsta
+ (".kinsta.cloud", "Kinsta", ["NXDOMAIN"]),
+]
+
+
+class TakeoverDetector(BaseModule):
+ """
+ Subdomain Takeover Detection Module
+
+ Checks discovered subdomains for potential takeover vulnerabilities
+ by analyzing CNAME records pointing to cloud services.
+ """
+
+ MODULE_NAME = "takeover"
+
+ def __init__(
+ self,
+ domain: str,
+ resolver: Optional['InternalDNSResolver'] = None,
+ threads: int = 10,
+ **kwargs
+ ):
+ super().__init__(domain, resolver, **kwargs)
+ self.threads = threads
+ self.vulnerable: List[Dict] = []
+ self.potential: List[Dict] = []
+
+ def _check_cname(self, subdomain: str) -> Optional[str]:
+ """Get CNAME record for subdomain"""
+ try:
+ return self.resolver.resolve_cname(subdomain)
+ except:
+ return None
+
+ def _check_nxdomain(self, hostname: str) -> bool:
+ """Check if hostname returns NXDOMAIN"""
+ try:
+ result = self.resolver.resolve_a(hostname)
+ return result is None
+ except:
+ return True
+
+ def _analyze_cname(self, subdomain: str, cname: str) -> Optional[Dict]:
+ """
+ Analyze CNAME for potential takeover
+
+ Returns dict with vulnerability info or None
+ """
+ cname_lower = cname.lower()
+
+ for pattern, service, indicators in TAKEOVER_FINGERPRINTS:
+ if pattern in cname_lower:
+ # Check if the CNAME target is NXDOMAIN (unclaimed)
+ is_nxdomain = "NXDOMAIN" in indicators and self._check_nxdomain(cname)
+
+ if is_nxdomain:
+ return {
+ "subdomain": subdomain,
+ "cname": cname,
+ "service": service,
+ "status": "vulnerable",
+ "reason": f"CNAME points to unclaimed {service} resource"
+ }
+ else:
+ return {
+ "subdomain": subdomain,
+ "cname": cname,
+ "service": service,
+ "status": "potential",
+ "reason": f"CNAME points to {service} - verify manually"
+ }
+
+ return None
+
+ def enumerate(self, known_subs: Dict[str, Optional[str]] = None, **kwargs) -> Dict[str, Optional[str]]:
+ """
+ Check subdomains for takeover vulnerabilities
+
+ Args:
+ known_subs: Dict of known subdomains to check
+
+ Returns:
+ Dict of vulnerable subdomains (for compatibility)
+ """
+ if not known_subs:
+ logger.info("No subdomains provided for takeover check")
+ return {}
+
+ logger.info(f"Checking {len(known_subs)} subdomains for takeover vulnerabilities")
+
+ from concurrent.futures import ThreadPoolExecutor, as_completed
+
+ def check_subdomain(subdomain: str) -> Optional[Dict]:
+ if self.is_cancelled():
+ return None
+
+ cname = self._check_cname(subdomain)
+ if cname:
+ return self._analyze_cname(subdomain, cname)
+ return None
+
+ with ThreadPoolExecutor(max_workers=self.threads) as executor:
+ futures = {
+ executor.submit(check_subdomain, sub): sub
+ for sub in known_subs.keys()
+ }
+
+ for future in as_completed(futures):
+ if self.is_cancelled():
+ break
+
+ result = future.result()
+ if result:
+ if result["status"] == "vulnerable":
+ self.vulnerable.append(result)
+ self._add_discovered(result["subdomain"], None)
+ else:
+ self.potential.append(result)
+
+ logger.info(f"Takeover check complete: {len(self.vulnerable)} vulnerable, {len(self.potential)} potential")
+
+ return self.discovered
+
+ def get_vulnerable(self) -> List[Dict]:
+ """Get list of vulnerable subdomains"""
+ return self.vulnerable
+
+ def get_potential(self) -> List[Dict]:
+ """Get list of potentially vulnerable subdomains"""
+ return self.potential
+
+ def get_all_findings(self) -> List[Dict]:
+ """Get all findings (vulnerable + potential)"""
+ return self.vulnerable + self.potential
diff --git a/src/usfx/modules/web_tech.py b/src/usfx/modules/web_tech.py
@@ -0,0 +1,324 @@
+"""
+Web Technology Detection Module
+
+Detects web technologies, frameworks, and servers on discovered subdomains
+using Wappalyzer fingerprints. Works completely offline - only requires
+HTTP access to target servers.
+"""
+
+import logging
+import socket
+import ssl
+import urllib.request
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Dict, List, Optional, Set, TYPE_CHECKING
+from urllib.error import HTTPError, URLError
+
+from .base import BaseModule
+
+if TYPE_CHECKING:
+ from ..utils.dns_resolver import InternalDNSResolver
+
+logger = logging.getLogger(__name__)
+
+
+# Try to import Wappalyzer
+try:
+ from Wappalyzer import Wappalyzer, WebPage
+ WAPPALYZER_AVAILABLE = True
+except ImportError:
+ WAPPALYZER_AVAILABLE = False
+ logger.warning("python-Wappalyzer not installed. Web tech detection will be limited.")
+
+
+class WebTechDetector(BaseModule):
+ """
+ Web Technology Detection Module
+
+ Scans discovered subdomains for web services and identifies
+ technologies using Wappalyzer fingerprints.
+
+ Works offline - only needs HTTP access to target servers.
+ """
+
+ MODULE_NAME = "web_tech"
+
+ def __init__(
+ self,
+ domain: str,
+ resolver: Optional['InternalDNSResolver'] = None,
+ threads: int = 10,
+ timeout: float = 5.0,
+ ports: List[int] = None,
+ **kwargs
+ ):
+ super().__init__(domain, resolver, timeout=timeout, **kwargs)
+ self.threads = threads
+ self.ports = ports or [80, 443, 8080, 8443]
+ self.web_servers: Dict[str, Dict] = {} # subdomain -> tech info
+
+ # Initialize Wappalyzer if available
+ self.wappalyzer = None
+ if WAPPALYZER_AVAILABLE:
+ try:
+ self.wappalyzer = Wappalyzer.latest()
+ except Exception as e:
+ logger.warning(f"Failed to initialize Wappalyzer: {e}")
+
+ def _check_http(self, url: str) -> Optional[Dict]:
+ """
+ Check if URL responds and get basic info
+
+ Returns dict with status, headers, title or None
+ """
+ try:
+ # Create SSL context that doesn't verify certificates
+ ctx = ssl.create_default_context()
+ ctx.check_hostname = False
+ ctx.verify_mode = ssl.CERT_NONE
+
+ req = urllib.request.Request(
+ url,
+ headers={
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
+ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
+ }
+ )
+
+ with urllib.request.urlopen(req, timeout=self.timeout, context=ctx) as response:
+ status = response.status
+ headers = dict(response.headers)
+ content = response.read(50000).decode('utf-8', errors='ignore')
+
+ # Extract title
+ title = None
+ if '<title>' in content.lower():
+ start = content.lower().find('<title>') + 7
+ end = content.lower().find('</title>')
+ if end > start:
+ title = content[start:end].strip()[:100]
+
+ return {
+ 'status': status,
+ 'headers': headers,
+ 'title': title,
+ 'content': content[:10000] # Limit content for analysis
+ }
+
+ except HTTPError as e:
+ # Still return info for HTTP errors (403, 404, etc.)
+ return {
+ 'status': e.code,
+ 'headers': dict(e.headers) if e.headers else {},
+ 'title': None,
+ 'content': ''
+ }
+ except (URLError, socket.timeout, ssl.SSLError, ConnectionError, OSError):
+ return None
+ except Exception as e:
+ logger.debug(f"Error checking {url}: {e}")
+ return None
+
+ def _detect_tech_wappalyzer(self, url: str) -> Set[str]:
+ """Detect technologies using Wappalyzer"""
+ if not self.wappalyzer:
+ return set()
+
+ try:
+ webpage = WebPage.new_from_url(url, verify=False, timeout=self.timeout)
+ results = self.wappalyzer.analyze(webpage)
+ return results
+ except Exception as e:
+ logger.debug(f"Wappalyzer error for {url}: {e}")
+ return set()
+
+ def _detect_tech_headers(self, headers: Dict) -> Set[str]:
+ """Detect technologies from HTTP headers"""
+ technologies = set()
+
+ # Server header
+ server = headers.get('Server', headers.get('server', ''))
+ if server:
+ server_lower = server.lower()
+ if 'nginx' in server_lower:
+ technologies.add('Nginx')
+ if 'apache' in server_lower:
+ technologies.add('Apache')
+ if 'iis' in server_lower:
+ technologies.add('IIS')
+ if 'cloudflare' in server_lower:
+ technologies.add('Cloudflare')
+ if 'gunicorn' in server_lower:
+ technologies.add('Gunicorn')
+ if 'uvicorn' in server_lower:
+ technologies.add('Uvicorn')
+
+ # X-Powered-By
+ powered_by = headers.get('X-Powered-By', headers.get('x-powered-by', ''))
+ if powered_by:
+ powered_lower = powered_by.lower()
+ if 'php' in powered_lower:
+ technologies.add('PHP')
+ if 'asp.net' in powered_lower:
+ technologies.add('ASP.NET')
+ if 'express' in powered_lower:
+ technologies.add('Express')
+ if 'next.js' in powered_lower:
+ technologies.add('Next.js')
+
+ # Other headers
+ if 'X-Drupal' in headers or 'x-drupal' in headers:
+ technologies.add('Drupal')
+ if 'X-Generator' in headers:
+ gen = headers['X-Generator'].lower()
+ if 'wordpress' in gen:
+ technologies.add('WordPress')
+ if 'drupal' in gen:
+ technologies.add('Drupal')
+
+ return technologies
+
+ def _detect_tech_content(self, content: str) -> Set[str]:
+ """Detect technologies from HTML content"""
+ technologies = set()
+ content_lower = content.lower()
+
+ # CMS/Frameworks
+ patterns = {
+ 'wp-content': 'WordPress',
+ 'wp-includes': 'WordPress',
+ '/joomla': 'Joomla',
+ 'drupal.js': 'Drupal',
+ 'sites/default/files': 'Drupal',
+ 'magento': 'Magento',
+ 'shopify': 'Shopify',
+ 'react': 'React',
+ 'vue.js': 'Vue.js',
+ 'angular': 'Angular',
+ 'jquery': 'jQuery',
+ 'bootstrap': 'Bootstrap',
+ 'tailwind': 'Tailwind CSS',
+ 'laravel': 'Laravel',
+ 'django': 'Django',
+ 'flask': 'Flask',
+ 'spring': 'Spring',
+ 'rails': 'Ruby on Rails',
+ 'next.js': 'Next.js',
+ 'nuxt': 'Nuxt.js',
+ 'gatsby': 'Gatsby',
+ }
+
+ for pattern, tech in patterns.items():
+ if pattern in content_lower:
+ technologies.add(tech)
+
+ return technologies
+
+ def _scan_subdomain(self, subdomain: str, ip: Optional[str]) -> Optional[Dict]:
+ """Scan a single subdomain for web technologies"""
+ if self.is_cancelled():
+ return None
+
+ result = {
+ 'subdomain': subdomain,
+ 'ip': ip,
+ 'web_servers': [],
+ 'technologies': set(),
+ }
+
+ # Try different ports and protocols
+ for port in self.ports:
+ if self.is_cancelled():
+ break
+
+ protocol = 'https' if port in [443, 8443] else 'http'
+ if port in [80, 443]:
+ url = f"{protocol}://{subdomain}"
+ else:
+ url = f"{protocol}://{subdomain}:{port}"
+
+ http_result = self._check_http(url)
+ if http_result:
+ server_info = {
+ 'url': url,
+ 'port': port,
+ 'status': http_result['status'],
+ 'title': http_result.get('title'),
+ 'server': http_result['headers'].get('Server', http_result['headers'].get('server')),
+ }
+ result['web_servers'].append(server_info)
+
+ # Detect technologies
+ result['technologies'].update(
+ self._detect_tech_headers(http_result['headers'])
+ )
+ if http_result.get('content'):
+ result['technologies'].update(
+ self._detect_tech_content(http_result['content'])
+ )
+
+ # Try Wappalyzer for more accurate detection
+ if self.wappalyzer:
+ result['technologies'].update(
+ self._detect_tech_wappalyzer(url)
+ )
+
+ if result['web_servers']:
+ return result
+ return None
+
+ def enumerate(self, known_subs: Dict[str, Optional[str]] = None, **kwargs) -> Dict[str, Optional[str]]:
+ """
+ Scan subdomains for web technologies
+
+ Args:
+ known_subs: Dict of known subdomains to scan
+
+ Returns:
+ Dict of subdomains with web services
+ """
+ if not known_subs:
+ logger.info("No subdomains provided for web tech detection")
+ return {}
+
+ logger.info(f"Scanning {len(known_subs)} subdomains for web technologies")
+
+ with ThreadPoolExecutor(max_workers=self.threads) as executor:
+ futures = {
+ executor.submit(self._scan_subdomain, sub, ip): sub
+ for sub, ip in known_subs.items()
+ }
+
+ for future in as_completed(futures):
+ if self.is_cancelled():
+ break
+
+ result = future.result()
+ if result:
+ self.web_servers[result['subdomain']] = result
+ self._add_discovered(result['subdomain'], result.get('ip'))
+
+ logger.info(f"Web tech scan complete: {len(self.web_servers)} web servers found")
+
+ return self.discovered
+
+ def get_web_servers(self) -> Dict[str, Dict]:
+ """Get detailed web server information"""
+ return self.web_servers
+
+ def get_technologies_summary(self) -> Dict[str, int]:
+ """Get summary of detected technologies"""
+ tech_count = {}
+ for info in self.web_servers.values():
+ for tech in info.get('technologies', []):
+ tech_count[tech] = tech_count.get(tech, 0) + 1
+ return dict(sorted(tech_count.items(), key=lambda x: x[1], reverse=True))
+
+ def format_results(self) -> str:
+ """Format results for display"""
+ lines = []
+ for subdomain, info in sorted(self.web_servers.items()):
+ techs = ', '.join(sorted(info.get('technologies', []))) or 'Unknown'
+ for server in info.get('web_servers', []):
+ lines.append(f"{server['url']} [{server['status']}] - {techs}")
+ return '\n'.join(lines)
diff --git a/src/usfx/orchestrator.py b/src/usfx/orchestrator.py
@@ -34,6 +34,7 @@ class Phase(Enum):
DNS_ENUMERATION = "dns_enumeration"
EXTENSION = "extension"
VALIDATION = "validation"
+ EXTENDED = "extended" # Takeover detection, web tech analysis
@dataclass
@@ -67,6 +68,9 @@ class OrchestrationState:
phase_results: List[PhaseResult] = field(default_factory=list)
started_at: float = field(default_factory=time.time)
is_cancelled: bool = False
+ # Extended results
+ takeover_results: List[dict] = field(default_factory=list)
+ web_tech_results: List[dict] = field(default_factory=list)
class SubdomainOrchestrator:
@@ -75,9 +79,10 @@ class SubdomainOrchestrator:
# Phase weight for overall progress calculation
PHASE_WEIGHTS = {
Phase.QUICK_DISCOVERY: (0, 15), # 0-15%
- Phase.DNS_ENUMERATION: (15, 60), # 15-60%
- Phase.EXTENSION: (60, 90), # 60-90%
- Phase.VALIDATION: (90, 100), # 90-100%
+ Phase.DNS_ENUMERATION: (15, 55), # 15-55%
+ Phase.EXTENSION: (55, 80), # 55-80%
+ Phase.VALIDATION: (80, 90), # 80-90%
+ Phase.EXTENDED: (90, 100), # 90-100%
}
def __init__(
@@ -293,12 +298,24 @@ class SubdomainOrchestrator:
def _get_enabled_modules(self) -> Set[str]:
"""Get set of enabled module names"""
if self.config and self.config.modules:
- return self.config.modules
- return {
- "zone_transfer", "dns_records", "dns_bruteforce", "dnssec_walker",
- "reverse_dns", "cname_chaser", "permutation", "recursive_enum",
- "vhost_scanner", "tls_analyzer"
- }
+ # Start with user-specified modules
+ modules = set(self.config.modules)
+ else:
+ # Default modules
+ modules = {
+ "zone_transfer", "dns_records", "dns_bruteforce", "dnssec_walker",
+ "reverse_dns", "cname_chaser", "permutation", "recursive_enum",
+ "vhost_scanner", "tls_analyzer"
+ }
+
+ # Add extended modules if enabled via config flags (regardless of -m)
+ if self.config:
+ if self.config.takeover:
+ modules.add("takeover")
+ if self.config.web_tech:
+ modules.add("web_tech")
+
+ return modules
def run_phase1_quick_discovery(self) -> PhaseResult:
"""Phase 1: Quick Discovery - Zone Transfer and DNS Records"""
@@ -532,6 +549,116 @@ class SubdomainOrchestrator:
duration_seconds=time.time() - phase_start
)
+ def run_phase5_extended(self) -> PhaseResult:
+ """Phase 5: Extended - Takeover detection and Web technology analysis"""
+ from .modules.takeover import TakeoverDetector
+ from .modules.web_tech import WebTechDetector
+
+ phase_start = time.time()
+ results = []
+ enabled = self._get_enabled_modules()
+
+ threads = self.config.threads if self.config else 10
+ timeout = self.config.timeout if self.config else 5.0
+ web_ports = self.config.web_ports if self.config else [80, 443, 8080, 8443]
+
+ # Get current known subdomains
+ with self._results_lock:
+ known_subs = {k: v.get('ip') for k, v in self.state.discovered.items()}
+
+ if not known_subs:
+ logger.info("No subdomains found for extended analysis")
+ self._update_progress(Phase.EXTENDED, "complete", 100)
+ return PhaseResult(
+ phase=Phase.EXTENDED,
+ modules=[],
+ total_found=0,
+ duration_seconds=time.time() - phase_start
+ )
+
+ # Run takeover detection
+ if "takeover" in enabled:
+ self._update_progress(Phase.EXTENDED, "takeover", 0)
+ try:
+ takeover_module = TakeoverDetector(
+ self.domain, self.resolver, threads=threads
+ )
+ takeover_module.enumerate(known_subs=known_subs)
+
+ # Store results
+ self.state.takeover_results = takeover_module.get_all_findings()
+
+ logger.info(f"Takeover check: {len(takeover_module.get_vulnerable())} vulnerable, "
+ f"{len(takeover_module.get_potential())} potential")
+
+ results.append(ModuleResult(
+ module_name="takeover",
+ found_count=len(self.state.takeover_results),
+ subdomains={},
+ duration_seconds=0
+ ))
+ except Exception as e:
+ logger.error(f"Takeover detection failed: {e}")
+ results.append(ModuleResult(
+ module_name="takeover",
+ found_count=0,
+ subdomains={},
+ duration_seconds=0,
+ error=str(e)
+ ))
+
+ # Run web technology detection
+ if "web_tech" in enabled:
+ self._update_progress(Phase.EXTENDED, "web_tech", 50)
+ try:
+ web_tech_module = WebTechDetector(
+ self.domain, self.resolver,
+ threads=threads, timeout=timeout, ports=web_ports
+ )
+ web_tech_module.enumerate(known_subs=known_subs)
+
+ # Store results - convert to list of dicts
+ web_servers = web_tech_module.get_web_servers()
+ self.state.web_tech_results = []
+ for subdomain, info in web_servers.items():
+ for server in info.get('web_servers', []):
+ self.state.web_tech_results.append({
+ 'subdomain': subdomain,
+ 'url': server['url'],
+ 'port': server['port'],
+ 'status': server['status'],
+ 'title': server.get('title'),
+ 'server': server.get('server'),
+ 'technologies': list(info.get('technologies', [])),
+ })
+
+ logger.info(f"Web tech detection: {len(self.state.web_tech_results)} web servers found")
+
+ results.append(ModuleResult(
+ module_name="web_tech",
+ found_count=len(self.state.web_tech_results),
+ subdomains={},
+ duration_seconds=0
+ ))
+ except Exception as e:
+ logger.error(f"Web tech detection failed: {e}")
+ results.append(ModuleResult(
+ module_name="web_tech",
+ found_count=0,
+ subdomains={},
+ duration_seconds=0,
+ error=str(e)
+ ))
+
+ self._update_progress(Phase.EXTENDED, "complete", 100)
+
+ return PhaseResult(
+ phase=Phase.EXTENDED,
+ modules=results,
+ total_found=len(self.state.takeover_results) + len(self.state.web_tech_results),
+ duration_seconds=time.time() - phase_start
+ )
+
def cancel(self):
"""Cancel the orchestration"""
self.state.is_cancelled = True
@@ -561,6 +688,14 @@ class SubdomainOrchestrator:
# Phase 4: Validation
result = self.run_phase4_validation()
self.state.phase_results.append(result)
+ if self.state.is_cancelled:
+ return self.state
+
+ # Phase 5: Extended (Takeover, Web Tech) - only if enabled
+ enabled = self._get_enabled_modules()
+ if "takeover" in enabled or "web_tech" in enabled:
+ result = self.run_phase5_extended()
+ self.state.phase_results.append(result)
logger.info(f"Enumeration complete: {self.state.total_found} subdomains found")
return self.state
diff --git a/src/usfx/utils/output.py b/src/usfx/utils/output.py
@@ -2,7 +2,7 @@
Output Formatter
Handles formatting and writing scan results to various formats:
-JSON, CSV, and TXT.
+JSON, CSV, TXT, and pipeline modes.
"""
import csv
@@ -13,7 +13,7 @@ from pathlib import Path
from typing import List, Optional, TYPE_CHECKING
if TYPE_CHECKING:
- from ..config import ScanResult, SubdomainResult
+ from ..config import PipelineMode, ScanResult, SubdomainResult, TakeoverResult, WebTechResult
logger = logging.getLogger(__name__)
@@ -59,6 +59,34 @@ class OutputFormatter:
]
}
+ # Add takeover results if present
+ if result.takeover_results:
+ data['takeover_vulnerabilities'] = [
+ {
+ 'subdomain': t.subdomain,
+ 'cname': t.cname,
+ 'service': t.service,
+ 'status': t.status,
+ 'reason': t.reason,
+ }
+ for t in result.takeover_results
+ ]
+
+ # Add web tech results if present
+ if result.web_tech_results:
+ data['web_technologies'] = [
+ {
+ 'subdomain': w.subdomain,
+ 'url': w.url,
+ 'port': w.port,
+ 'status': w.status,
+ 'title': w.title,
+ 'server': w.server,
+ 'technologies': w.technologies,
+ }
+ for w in result.web_tech_results
+ ]
+
if pretty:
return json.dumps(data, indent=2, ensure_ascii=False)
return json.dumps(data, ensure_ascii=False)
@@ -114,6 +142,50 @@ class OutputFormatter:
return '\n'.join(lines)
+ @staticmethod
+ def to_pipeline(result: 'ScanResult', mode: 'PipelineMode') -> str:
+ """
+ Convert scan result to pipeline output format.
+
+ Args:
+ result: ScanResult object
+ mode: Pipeline mode (SUBS, WEB, IPS, JSON)
+
+ Returns:
+ Pipeline output string
+ """
+ from ..config import PipelineMode
+
+ if mode == PipelineMode.SUBS:
+ # One subdomain per line
+ return '\n'.join(s.subdomain for s in result.subdomains)
+
+ elif mode == PipelineMode.WEB:
+ # Web URLs from web tech results
+ urls = set()
+ for w in result.web_tech_results:
+ urls.add(w.url)
+ # If no web tech results, generate URLs from active subdomains
+ if not urls:
+ for s in result.subdomains:
+ if s.is_active:
+ urls.add(f"https://{s.subdomain}")
+ return '\n'.join(sorted(urls))
+
+ elif mode == PipelineMode.IPS:
+ # Unique IP addresses
+ ips = set()
+ for s in result.subdomains:
+ if s.ip:
+ ips.add(s.ip)
+ return '\n'.join(sorted(ips))
+
+ elif mode == PipelineMode.JSON:
+ # JSON to stdout (no pretty print for piping)
+ return OutputFormatter.to_json(result, pretty=False)
+
+ return ''
+
@staticmethod
def write_file(
result: 'ScanResult',
@@ -383,6 +455,104 @@ class ConsoleOutput:
else:
print(f"Warning: {message}")
+ def print_takeover_results(self, takeover_results: List['TakeoverResult']) -> None:
+ """Print takeover vulnerability results"""
+ if self.quiet or not takeover_results:
+ return
+
+ vulnerable = [t for t in takeover_results if t.status == 'vulnerable']
+ potential = [t for t in takeover_results if t.status == 'potential']
+
+ if self.use_rich:
+ from rich.table import Table
+
+ if vulnerable:
+ self.console.print(f"\n[bold red]⚠ Takeover Vulnerabilities Found: {len(vulnerable)}[/bold red]")
+ table = Table()
+ table.add_column("Subdomain", style="red")
+ table.add_column("CNAME", style="yellow")
+ table.add_column("Service", style="cyan")
+ table.add_column("Reason", style="dim")
+
+ for t in vulnerable:
+ table.add_row(t.subdomain, t.cname, t.service, t.reason)
+
+ self.console.print(table)
+
+ if potential:
+ self.console.print(f"\n[bold yellow]Potential Takeover Issues: {len(potential)}[/bold yellow]")
+ for t in potential[:10]:
+ self.console.print(f" [dim]•[/dim] {t.subdomain} → {t.cname} ({t.service})")
+ if len(potential) > 10:
+ self.console.print(f" [dim]... and {len(potential) - 10} more[/dim]")
+ else:
+ if vulnerable:
+ print(f"\n⚠ Takeover Vulnerabilities Found: {len(vulnerable)}")
+ print("-" * 60)
+ for t in vulnerable:
+ print(f" {t.subdomain} -> {t.cname} ({t.service})")
+ print(f" Reason: {t.reason}")
+
+ if potential:
+ print(f"\nPotential Takeover Issues: {len(potential)}")
+ for t in potential[:10]:
+ print(f" • {t.subdomain} → {t.cname} ({t.service})")
+ if len(potential) > 10:
+ print(f" ... and {len(potential) - 10} more")
+
+ def print_web_tech_results(self, web_tech_results: List['WebTechResult']) -> None:
+ """Print web technology detection results"""
+ if self.quiet or not web_tech_results:
+ return
+
+ if self.use_rich:
+ from rich.table import Table
+
+ self.console.print(f"\n[bold blue]Web Technologies Detected: {len(web_tech_results)} servers[/bold blue]")
+
+ table = Table()
+ table.add_column("URL", style="cyan")
+ table.add_column("Status", style="green")
+ table.add_column("Title", style="white", max_width=30)
+ table.add_column("Technologies", style="yellow")
+
+ for w in web_tech_results[:30]:
+ techs = ', '.join(w.technologies[:5]) if w.technologies else '-'
+ if len(w.technologies) > 5:
+ techs += f' (+{len(w.technologies) - 5})'
+ title = (w.title[:27] + '...') if w.title and len(w.title) > 30 else (w.title or '-')
+ table.add_row(w.url, str(w.status), title, techs)
+
+ if len(web_tech_results) > 30:
+ table.add_row("...", "", f"({len(web_tech_results) - 30} more)", "")
+
+ self.console.print(table)
+
+ # Technology summary
+ tech_count = {}
+ for w in web_tech_results:
+ for tech in w.technologies:
+ tech_count[tech] = tech_count.get(tech, 0) + 1
+
+ if tech_count:
+ sorted_techs = sorted(tech_count.items(), key=lambda x: x[1], reverse=True)[:10]
+ self.console.print("\n[bold]Top Technologies:[/bold]")
+ for tech, count in sorted_techs:
+ self.console.print(f" [dim]•[/dim] {tech}: {count}")
+ else:
+ print(f"\nWeb Technologies Detected: {len(web_tech_results)} servers")
+ print("-" * 60)
+
+ for w in web_tech_results[:30]:
+ techs = ', '.join(w.technologies[:5]) if w.technologies else '-'
+ print(f" {w.url} [{w.status}]")
+ if w.title:
+ print(f" Title: {w.title[:50]}")
+ print(f" Tech: {techs}")
+
+ if len(web_tech_results) > 30:
+ print(f" ... ({len(web_tech_results) - 30} more)")
+
def create_progress(self):
"""Create a progress context manager"""
if self.use_rich and not self.quiet:
|