import requests import ssl import socket import sys import re import json from datetime import datetime from urllib.parse import urlparse from concurrent.futures import ThreadPoolExecutor, as_completed import queue class HighPerformanceWebScanner: def __init__(self, url, threads=10): self.url = url.rstrip('/') self.parsed_url = urlparse(self.url) if self.parsed_url.scheme: self.scheme = self.parsed_url.scheme self.domain = self.parsed_url.netloc else: self.scheme = 'https' self.domain = self.url.replace('http://', '').replace('https://', '') self.base_url = f"{self.scheme}://{self.domain}" self.threads = threads self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8' } self.timeout = 5 # Common subdomains for enumeration self.subdomain_wordlist = [ 'www', 'api', 'dev', 'staging', 'test', 'mail', 'ftp', 'admin', 'blog', 'shop', 'app', 'm', 'mobile', 'static', 'cdn', 'assets', 'db', 'mysql', 'postgres', 'backup', 'docs', 'wiki', 'vpn', 'ssh' ] # Common files for enumeration self.common_files = [ '/robots.txt', '/sitemap.xml', '/.git/config', '/.env', '/wp-config.php.bak', '/admin/', '/backup.sql', '/config.json', '/phpinfo.php', '/server-status', '/.htaccess' ] def fetch(self, path='', method='GET', params=None): """Helper to make HTTP requests.""" full_url = f"{self.base_url}{path}" if path.startswith('/') else f"{self.base_url}/{path}" try: resp = requests.get(full_url, headers=self.headers, timeout=self.timeout, params=params) return resp except requests.exceptions.RequestException: return None def check_security_headers(self): """Check for important HTTP security headers.""" print("\n[+] Security Headers Check:") response = self.fetch('/') if not response: print(" [!] Could not fetch root URL.") return important_headers = { 'X-Content-Type-Options': 'nosniff', 'X-Frame-Options': ['DENY', 'SAMEORIGIN'], 'Strict-Transport-Security': 'max-age=31536000', 'Content-Security-Policy': None, } for header, expected in important_headers.items(): if header in response.headers: value = response.headers[header] if isinstance(expected, list): if value in expected: print(f" [✓] {header}: {value} (Correct)") else: print(f" [!] {header}: {value} (Suboptimal)") else: print(f" [✓] {header}: Present") else: print(f" [✗] {header}: MISSING") def check_ssl_certificate(self): """Check SSL/TLS certificate validity and details.""" print("\n[+] SSL/TLS Certificate Check:") try: context = ssl.create_default_context() with socket.create_connection((self.domain, 443), timeout=self.timeout) as sock: with context.wrap_socket(sock, server_hostname=self.domain) as ssock: cert_dict = ssock.getpeercert() subject = dict(x[0] for x in cert_dict['subject']) issuer = dict(x[0] for x in cert_dict['issuer']) not_before_str = cert_dict['notBefore'] not_after_str = cert_dict['notAfter'] not_before = datetime.strptime(not_before_str, '%b %d %H:%M:%S %Y %Z') not_after = datetime.strptime(not_after_str, '%b %d %H:%M:%S %Y %Z') now = datetime.utcnow() print(f" [✓] Issued To: {subject.get('commonName', 'Unknown')}") print(f" [✓] Issued By: {issuer.get('commonName', 'Unknown')}") print(f" [✓] Valid From: {not_before.strftime('%Y-%m-%d')}") print(f" [✓] Valid Until: {not_after.strftime('%Y-%m-%d')}") days_left = (not_after - now).days if now > not_after: print(f" [✗] WARNING: Certificate is EXPIRED!") elif days_left < 30: print(f" [!] WARNING: Certificate expires in {days_left} days.") else: print(f" [✓] Certificate is Valid ({days_left} days left)") except ssl.SSLError as e: print(f" [✗] SSL Error: {e}") except socket.timeout: print(f" [!] Connection timed out on port 443.") except Exception as e: print(f" [!] Could not connect to check SSL: {e}") def _check_subdomain(self, sub): """Check a single subdomain.""" url = f"{self.scheme}://{sub}.{self.domain}" try: resp = requests.get(url, headers=self.headers, timeout=3) if resp.status_code == 200: return (url, 200) elif resp.status_code == 404: # Check if it's a custom 404 or just missing pass except requests.exceptions.ConnectionError: pass except Exception: pass return None def check_subdomains(self): """Enumerate subdomains using multiple threads.""" print("\n[+] Subdomain Enumeration (Multi-threaded):") found_subdomains = [] with ThreadPoolExecutor(max_workers=self.threads) as executor: futures = {executor.submit(self._check_subdomain, sub): sub for sub in self.subdomain_wordlist} for future in as_completed(futures): result = future.result() if result: url, status = result print(f" [!] Subdomain Found: {url} (Status: {status})") found_subdomains.append(url) if not found_subdomains: print(" [-] No common subdomains found.") def _check_file(self, file_path): """Check a single file.""" resp = self.fetch(file_path) if resp: status = resp.status_code if status in [200, 403, 401]: return (file_path, status) return None def check_common_files(self): """Enumerate common files and directories using multiple threads.""" print("\n[+] Common File/Dir Enumeration:") with ThreadPoolExecutor(max_workers=self.threads) as executor: futures = {executor.submit(self._check_file, file_path): file_path for file_path in self.common_files} for future in as_completed(futures): result = future.result() if result: file_path, status = result print(f" [!] {file_path} -> Status: {status}") def detect_vulnerabilities(self): """Basic detection for SQLi and XSS in URL parameters.""" print("\n[+] Basic Vulnerability Detection (SQLi/XSS):") test_payloads = { 'sql': "' OR 1=1--", 'xss': "" } param_names = ['id', 'page', 'search', 'q', 'user'] for param in param_names: # Test SQL Injection sqli_resp = self.fetch(path='/', params={param: test_payloads['sql']}) if sqli_resp and sqli_resp.status_code == 200: print(f" [?] Potential SQLi in param '{param}': Response OK (Check manually)") # Test XSS xss_resp = self.fetch(path='/', params={param: test_payloads['xss']}) if xss_resp and xss_resp.status_code == 200: print(f" [?] Potential XSS in param '{param}': Response OK (Check manually)") def run_scan(self): print(f"[*] Starting High-Performance Scan on: {self.base_url}") # Run checks sequentially where order matters or data is needed, # but parallelize independent tasks like subdomain and file checking. self.check_security_headers() self.check_ssl_certificate() # Parallel execution for heavy tasks print("\n[+] Running Parallel Checks...") with ThreadPoolExecutor(max_workers=self.threads) as executor: future_subdomains = executor.submit(self.check_subdomains) future_files = executor.submit(self.check_common_files) # Wait for both to complete future_subdomains.result() future_files.result() self.detect_vulnerabilities() print("\n[*] Scan Complete.") def main(): if len(sys.argv) != 2: print("Usage: python high_performance_scanner.py ") sys.exit(1) target = sys.argv[1] if not target.startswith(('http://', 'https://')): target = 'https://' + target scanner = HighPerformanceWebScanner(target, threads=10) scanner.run_scan() if __name__ == "__main__": main()