(Статья) Сказ о том, как я ботнет делал

import requests
import socket
import threading
import time
import json
import sys
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib3.exceptions import InsecureRequestWarning


requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

class colors:
    RED = '\033[91m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    MAGENTA = '\033[95m'
    CYAN = '\033[96m'
    WHITE = '\033[97m'
    RESET = '\033[0m'
    BOLD = '\033[1m'

BANNER = f"""
{colors.BOLD}{colors.CYAN}
╔══════════════════════════════════════════════════════════════╗
║               D-Link & Zyxel Router Scanner                  ║
╚══════════════════════════════════════════════════════════════╝
{colors.RESET}
"""

def print_banner():
    print(BANNER)
    print(f"{colors.BOLD}Scan started at: {time.strftime('%Y-%m-%d %H:%M:%S')}{colors.RESET}")
    print()

def is_port_open(host, port, timeout=2):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((host, port))
        sock.close()
        return result == 0
    except:
        return False

def check_http_service(host, port, timeout=5):
    protocols = ['http', 'https']
    
    for proto in protocols:
        url = f"{proto}://{host}:{port}"
        try:
            response = requests.get(
                url,
                timeout=timeout,
                verify=False,
                allow_redirects=True,
                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'
                }
            )
            if response.status_code < 500:
                return proto, port, response
        except requests.exceptions.RequestException:
            continue
        except Exception as e:
            continue
    
    return None, None, None

def detect_router_type(response):
    if not response:
        return None
    
    text = response.text.lower()
    headers = str(response.headers).lower()

    dlink_indicators = [
        'd-link', 'dlink', 'realm="dlink"', 'realm="d-link"',
        'd-link systems', 'dlinkrouter', 'dir-', 'dsl-'
    ]
    
    for indicator in dlink_indicators:
        if indicator in text or indicator in headers:
            return 'D-Link'

    zyxel_indicators = [
        'zyxel', 'realm="zyxel"', 'zywall', 'p-', 'vmg', 'emg',
        'zyxel communications', 'zyxel router'
    ]
    
    for indicator in zyxel_indicators:
        if indicator in text or indicator in headers:
            return 'Zyxel'

    server_header = response.headers.get('Server', '').lower()
    if any(indicator in server_header for indicator in dlink_indicators):
        return 'D-Link'
    if any(indicator in server_header for indicator in zyxel_indicators):
        return 'Zyxel'
    
    if '<title>' in text and '</title>' in text:
        title_start = text.find('<title>') + 7
        title_end = text.find('</title>')
        title = text[title_start:title_end].lower()
        
        if any(indicator in title for indicator in dlink_indicators):
            return 'D-Link'
        if any(indicator in title for indicator in zyxel_indicators):
            return 'Zyxel'
    
    return None

def check_common_ports(host):
    common_ports = [80, 443, 8080, 8081, 8888, 23, 22, 21]
    open_ports = []
    
    for port in common_ports:
        if is_port_open(host, port, 1):
            open_ports.append(port)
    
    return open_ports

def scan_single_ip(ip, timeout=5):
    results = {
        'ip': ip,
        'router_type': None,
        'open_ports': [],
        'http_ports': [],
        'services': [],
        'details': {}
    }
    
    try:
        open_ports = check_common_ports(ip)
        results['open_ports'] = open_ports
        
        if not open_ports:
            return results
        
        http_ports = [port for port in open_ports if port in [80, 443, 8080, 8081, 8888]]
        results['http_ports'] = http_ports
        
        for port in http_ports:
            proto, http_port, response = check_http_service(ip, port, timeout)
            if proto and response:
                results['services'].append({
                    'protocol': proto,
                    'port': port,
                    'status_code': response.status_code
                })

                router_type = detect_router_type(response)
                if router_type:
                    results['router_type'] = router_type
                    results['details'] = {
                        'server': response.headers.get('Server', 'Unknown'),
                        'title': extract_title(response.text),
                        'auth_type': detect_auth_type(response)
                    }
                    break
        
        for port in open_ports:
            if port == 23:
                results['services'].append({'protocol': 'telnet', 'port': 23})
            elif port == 22:
                results['services'].append({'protocol': 'ssh', 'port': 22})
                
    except Exception as e:
        results['error'] = str(e)
    
    return results

def extract_title(html):
    try:
        text = html.lower()
        if '<title>' in text and '</title>' in text:
            title_start = text.find('<title>') + 7
            title_end = text.find('</title>')
            return html[title_start:title_end].strip()
    except:
        pass
    return "Unknown"

def detect_auth_type(response):
    headers = str(response.headers).lower()
    
    if 'www-authenticate' in headers:
        if 'basic' in headers:
            return 'HTTP Basic Auth'
        elif 'digest' in headers:
            return 'HTTP Digest Auth'
    
    if 'login' in response.text.lower() or 'password' in response.text.lower():
        return 'Web Form Auth'
    
    return 'No Auth'

def load_ips_from_file(filename):
    ips = []
    
    try:
        with open(filename, 'r') as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith('#'):
                    continue

                if '/' in line:
                    base_ip = line.split('/')[0]
                    ips.append(base_ip)
                elif '-' in line:
                    base_ip = line.split('-')[0]
                    ips.append(base_ip)
                else:
                    ips.append(line)
                    
    except FileNotFoundError:
        print(f"{colors.RED}[!] File not found: {filename}{colors.RESET}")
        sys.exit(1)
    
    return list(set(ips))

def save_targets(targets, filename):
    try:
        with open(filename, 'w') as f:
            for target in targets:
                if target['router_type']:
                    for port in target['http_ports']:
                        f.write(f"{target['ip']}:{port}\n")

                    if 23 in target['open_ports']:
                        f.write(f"{target['ip']}:23\n")
                    
                    if 22 in target['open_ports']:
                        f.write(f"{target['ip']}:22\n")
        
        print(f"{colors.GREEN}[+] Saved {len(targets)} targets to {filename}{colors.RESET}")
        
    except Exception as e:
        print(f"{colors.RED}[!] Error saving targets: {e}{colors.RESET}")

def save_detailed_results(results, filename):
    try:
        with open(filename, 'w') as f:
            json.dump(results, f, indent=2, ensure_ascii=False)
        print(f"{colors.GREEN}[+] Detailed results saved to {filename}{colors.RESET}")
    except Exception as e:
        print(f"{colors.RED}[!] Error saving detailed results: {e}{colors.RESET}")

def print_results_summary(results):
    router_count = sum(1 for r in results if r['router_type'])
    dlink_count = sum(1 for r in results if r['router_type'] == 'D-Link')
    zyxel_count = sum(1 for r in results if r['router_type'] == 'Zyxel')
    
    print(f"\n{colors.BOLD}=== SCAN SUMMARY ==={colors.RESET}")
    print(f"Total IPs scanned: {len(results)}")
    print(f"Routers found: {router_count}")
    print(f"D-Link routers: {dlink_count}")
    print(f"Zyxel routers: {zyxel_count}")
    print(f"Targets saved to: targets.txt")

def print_router_details(results):
    routers = [r for r in results if r['router_type']]
    
    if not routers:
        print(f"{colors.YELLOW}[!] No routers found{colors.RESET}")
        return
    
    print(f"\n{colors.BOLD}=== FOUND ROUTERS ==={colors.RESET}")
    
    for router in routers:
        print(f"\n{colors.GREEN}[+] {router['router_type']} at {router['ip']}{colors.RESET}")
        print(f"   Open ports: {', '.join(map(str, router['open_ports']))}")
        print(f"   HTTP ports: {', '.join(map(str, router['http_ports']))}")
        
        if router['details']:
            if router['details'].get('title'):
                print(f"   Title: {router['details']['title']}")
            if router['details'].get('server'):
                print(f"   Server: {router['details']['server']}")
            if router['details'].get('auth_type'):
                print(f"   Auth: {router['details']['auth_type']}")

def main():
    parser = argparse.ArgumentParser(description='D-Link & Zyxel Router Scanner')
    parser.add_argument('-i', '--input', default='ip.txt', help='Input file with IP addresses (default: ip.txt)')
    parser.add_argument('-o', '--output', default='targets.txt', help='Output file for targets (default: targets.txt)')
    parser.add_argument('-j', '--json', default='scan_results.json', help='JSON output file (default: scan_results.json)')
    parser.add_argument('-t', '--threads', type=int, default=20, help='Number of threads (default: 20)')
    parser.add_argument('-to', '--timeout', type=int, default=3, help='Timeout in seconds (default: 3)')
    parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
    
    args = parser.parse_args()
    
    print_banner()
    
    print(f"{colors.BLUE}[*] Loading IP addresses from {args.input}{colors.RESET}")
    ips = load_ips_from_file(args.input)
    
    if not ips:
        print(f"{colors.RED}[!] No IP addresses found in {args.input}{colors.RESET}")
        sys.exit(1)
    
    print(f"{colors.BLUE}[*] Loaded {len(ips)} unique IP addresses{colors.RESET}")
    print(f"{colors.BLUE}[*] Starting scan with {args.threads} threads...{colors.RESET}")
    
    all_results = []
    found_routers = []
    
    with ThreadPoolExecutor(max_workers=args.threads) as executor:
        future_to_ip = {
            executor.submit(scan_single_ip, ip, args.timeout): ip
            for ip in ips
        }
        
        for i, future in enumerate(as_completed(future_to_ip), 1):
            ip = future_to_ip[future]
            try:
                result = future.result()
                all_results.append(result)
                
                if args.verbose:
                    if result['router_type']:
                        print(f"{colors.GREEN}[+] Found {result['router_type']} at {ip}{colors.RESET}")
                    else:
                        print(f"{colors.BLUE}[*] Scanned {ip} ({i}/{len(ips)}){colors.RESET}")
                else:
                    if result['router_type']:
                        print(f"{colors.GREEN}[+] Found {result['router_type']} at {ip}{colors.RESET}")
                
                if result['router_type']:
                    found_routers.append(result)
                    
            except Exception as e:
                print(f"{colors.RED}[!] Error scanning {ip}: {e}{colors.RESET}")
                all_results.append({'ip': ip, 'error': str(e)})
    

    save_targets(found_routers, args.output)
    save_detailed_results(all_results, args.json)
    print_results_summary(all_results)
    print_router_details(all_results)
    
    print(f"\n{colors.GREEN}[+] Scan completed!{colors.RESET}")

if __name__ == "__main__":
    main()