scanner.py

import socket
import argparse
import threading
import time
import random
import json
from queue import Queue

queue = Queue()
results = []
lock = threading.Lock()

COMMON_SERVICES = {
    21: "FTP",
    22: "SSH",
    23: "Telnet",
    25: "SMTP",
    53: "DNS",
    80: "HTTP",
    110: "POP3",
    143: "IMAP",
    443: "HTTPS",
    3306: "MySQL",
    6379: "Redis",
}

def banner_grab(target, port, timeout):
    try:
        s = socket.socket()
        s.settimeout(timeout)
        s.connect((target, port))
        s.send(b"HEAD / HTTP/1.0\r\n\r\n")
        banner = s.recv(1024).decode(errors="ignore").strip()
        s.close()
        return banner[:100]
    except:
        return None

def scan_tcp(target, port, timeout, stealth):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        res = s.connect_ex((target, port))
        if res == 0:
            service = COMMON_SERVICES.get(port, "Unknown")
            banner = banner_grab(target, port, timeout)
            with lock:
                results.append({
                    "port": port,
                    "protocol": "TCP",
                    "service": service,
                    "banner": banner
                })
                print(f"[+] TCP {port} OPEN ({service})")
        s.close()
        if stealth:
            time.sleep(random.uniform(0.2, 0.8))
    except:
        pass

def scan_udp(target, port, timeout):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.settimeout(timeout)
        s.sendto(b"", (target, port))
        s.recvfrom(1024)
        with lock:
            results.append({
                "port": port,
                "protocol": "UDP",
                "service": "Unknown",
                "banner": None
            })
            print(f"[+] UDP {port} OPEN")
        s.close()
    except:
        pass

def worker(args):
    while not queue.empty():
        port = queue.get()
        if args.udp:
            scan_udp(args.target, port, args.timeout)
        else:
            scan_tcp(args.target, port, args.timeout, args.stealth)
        queue.task_done()

def main():
    parser = argparse.ArgumentParser(description="🔴 Python Port Scanner (Red Team)")
    parser.add_argument("target", help="Target IP or domain")
    parser.add_argument("-p", "--ports", default="1-1024", help="Port range (ex: 1-1000)")
    parser.add_argument("-t", "--threads", type=int, default=100, help="Threads")
    parser.add_argument("--timeout", type=float, default=1, help="Socket timeout")
    parser.add_argument("--stealth", action="store_true", help="Stealth mode")
    parser.add_argument("--udp", action="store_true", help="UDP scan")
    parser.add_argument("--json", help="Export results to JSON file")
    parser.add_argument("--txt", help="Export results to TXT file")
    args = parser.parse_args()

    start, end = map(int, args.ports.split("-"))
    for port in range(start, end + 1):
        queue.put(port)

    print(f"\n🔴 Scanning {args.target} [{args.ports}]")

    threads = []
    for _ in range(args.threads):
        t = threading.Thread(target=worker, args=(args,))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

    if args.json:
        with open(args.json, "w") as f:
            json.dump(results, f, indent=4)
        print(f"[+] JSON export: {args.json}")

    if args.txt:
        with open(args.txt, "w") as f:
            for r in results:
                f.write(f"{r['protocol']} {r['port']} {r['service']}\n")
        print(f"[+] TXT export: {args.txt}")

    print("\n✅ Scan terminé")

if __name__ == "__main__":
    main()