# Fichier: python_cheats/cheatsheets/socket.txt
# Cheatsheet Sockets Python - Guide Complet Programmation Réseau


## [LISTE] Contenu principal :

1. **Bases des sockets** - Types, création, concepts fondamentaux
2. **Serveur/Client TCP** - Exemples simples et avancés avec threading
3. **Serveur/Client UDP** - Datagrammes et communication sans connexion
4. **Options socket** - `setsockopt`, `getsockopt`, configuration avancée
5. **Méthodes principales** - `bind`, `listen`, `accept`, `connect`, `send`, `recv`
6. **Socket non-bloquant** - `select`, multiplexing I/O
7. **Asyncio** - Programmation asynchrone moderne
8. **Fonctions utilitaires** - DNS, résolution, conversions
9. **Transfert de fichiers** - Envoi/réception avec progression
10. **HTTP basique** - Client/serveur HTTP sans bibliothèques
11. **SSL/TLS** - Connexions sécurisées, certificats
12. **Broadcast/Multicast** - Communication groupe
13. **Unix Domain Sockets** - IPC local
14. **Raw Sockets** - Accès bas niveau, sniffer de paquets
15. **Gestion d'erreurs** - Toutes les exceptions possibles
16. **Context managers** - Utilisation avec `with`
17. **Patterns avancés** - Pool de connexions, protocoles, heartbeat
18. **Debugging/Monitoring** - Logging, statistiques
19. **Tests** - Unittest pour sockets
20. **Performance** - Optimisation, benchmarks
21. **Sécurité** - Validation, rate limiting, timeouts
22. **Exemples complets** - Chat, proxy, tunnel, scanner de ports, speed test, Wake-on-LAN
23. **Commandes système** - netcat, tcpdump, netstat
24. **Configuration** - systemd, Docker, Nginx
25. **Troubleshooting** - Solutions aux problèmes courants


[OK] BASES DES SOCKETS

# Importer le module
import socket

# Types de sockets
socket.AF_INET          # IPv4
socket.AF_INET6         # IPv6
socket.AF_UNIX          # Unix Domain Sockets (IPC local)

# Protocoles
socket.SOCK_STREAM      # TCP (connexion, fiable, ordonné)
socket.SOCK_DGRAM       # UDP (sans connexion, rapide, non fiable)
socket.SOCK_RAW         # Raw sockets (accès direct IP)

# Créer un socket basique
# TCP IPv4
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# UDP IPv4
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# TCP IPv6
sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)

# Unix socket
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)


[OK] SERVEUR TCP (SOCKET STREAM)

# === Serveur TCP basique ===
import socket

# Créer socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Adresse et port
HOST = '0.0.0.0'        # Écoute sur toutes les interfaces
PORT = 8080

# Bind (lier) socket à l'adresse
server_socket.bind((HOST, PORT))

# Écouter les connexions (backlog = 5)
server_socket.listen(5)
print(f"Serveur écoute sur {HOST}:{PORT}")

# Accepter connexion client
while True:
    client_socket, client_address = server_socket.accept()
    print(f"Connexion de {client_address}")
    
    # Recevoir données
    data = client_socket.recv(1024)  # Buffer 1024 bytes
    print(f"Reçu: {data.decode()}")
    
    # Envoyer réponse
    response = "Message reçu!"
    client_socket.send(response.encode())
    
    # Fermer connexion client
    client_socket.close()

# Fermer serveur
server_socket.close()


# === Serveur TCP avec gestion d'erreurs ===
import socket
import sys

def start_server(host='127.0.0.1', port=8080):
    try:
        # Créer socket
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
        # Options socket (réutiliser adresse immédiatement)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        
        # Bind et listen
        server.bind((host, port))
        server.listen(5)
        print(f"Serveur démarré sur {host}:{port}")
        
        while True:
            try:
                client, addr = server.accept()
                print(f"Connexion: {addr}")
                
                # Recevoir avec timeout
                client.settimeout(5.0)
                data = client.recv(4096)
                
                if data:
                    print(f"Reçu: {data.decode('utf-8', errors='ignore')}")
                    client.sendall(b"OK\n")
                
                client.close()
                
            except socket.timeout:
                print("Timeout client")
            except Exception as e:
                print(f"Erreur client: {e}")
                
    except KeyboardInterrupt:
        print("\nArrêt serveur...")
    except Exception as e:
        print(f"Erreur serveur: {e}")
    finally:
        server.close()

if __name__ == '__main__':
    start_server()


# === Serveur TCP multi-client (threading) ===
import socket
import threading

def handle_client(client_socket, address):
    """Gère un client dans un thread séparé"""
    print(f"[+] Nouveau client: {address}")
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            
            print(f"[{address}] {data.decode()}")
            response = f"Echo: {data.decode()}"
            client_socket.send(response.encode())
            
    except Exception as e:
        print(f"[!] Erreur {address}: {e}")
    finally:
        print(f"[-] Déconnexion: {address}")
        client_socket.close()

def start_threaded_server(host='0.0.0.0', port=8080):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(10)
    print(f"[*] Serveur multi-thread sur {host}:{port}")
    
    try:
        while True:
            client, addr = server.accept()
            
            # Créer thread pour ce client
            client_thread = threading.Thread(
                target=handle_client,
                args=(client, addr)
            )
            client_thread.daemon = True
            client_thread.start()
            
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur")
    finally:
        server.close()


[OK] CLIENT TCP

# === Client TCP basique ===
import socket

# Créer socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Connexion au serveur
HOST = '127.0.0.1'
PORT = 8080
client_socket.connect((HOST, PORT))

# Envoyer données
message = "Hello Server!"
client_socket.send(message.encode())

# Recevoir réponse
response = client_socket.recv(1024)
print(f"Réponse: {response.decode()}")

# Fermer connexion
client_socket.close()


# === Client TCP avec gestion d'erreurs ===
import socket

def tcp_client(host='127.0.0.1', port=8080, message='Hello'):
    try:
        # Créer et connecter
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.settimeout(5.0)  # Timeout 5 secondes
        
        print(f"Connexion à {host}:{port}...")
        client.connect((host, port))
        print("Connecté!")
        
        # Envoyer
        client.sendall(message.encode())
        print(f"Envoyé: {message}")
        
        # Recevoir
        response = client.recv(4096)
        print(f"Reçu: {response.decode()}")
        
        return response
        
    except socket.timeout:
        print("Timeout de connexion")
    except ConnectionRefusedError:
        print("Connexion refusée")
    except Exception as e:
        print(f"Erreur: {e}")
    finally:
        client.close()


# === Client TCP interactif ===
import socket

def interactive_client(host='127.0.0.1', port=8080):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        client.connect((host, port))
        print(f"Connecté à {host}:{port}")
        print("Tapez 'quit' pour quitter")
        
        while True:
            message = input("> ")
            
            if message.lower() == 'quit':
                break
            
            client.send(message.encode())
            response = client.recv(4096)
            print(f"Serveur: {response.decode()}")
            
    except Exception as e:
        print(f"Erreur: {e}")
    finally:
        client.close()


[OK] SERVEUR UDP (SOCKET DATAGRAM)

# === Serveur UDP basique ===
import socket

# Créer socket UDP
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# Bind
HOST = '0.0.0.0'
PORT = 9090
udp_socket.bind((HOST, PORT))

print(f"Serveur UDP sur {HOST}:{PORT}")

while True:
    # Recevoir données et adresse client
    data, client_address = udp_socket.recvfrom(1024)
    print(f"Reçu de {client_address}: {data.decode()}")
    
    # Envoyer réponse
    response = "ACK"
    udp_socket.sendto(response.encode(), client_address)


# === Serveur UDP avancé ===
import socket
import time

def udp_server(host='0.0.0.0', port=9090):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind((host, port))
    
    print(f"[*] Serveur UDP: {host}:{port}")
    
    try:
        while True:
            data, addr = sock.recvfrom(4096)
            timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
            print(f"[{timestamp}] {addr}: {data.decode()}")
            
            # Echo avec timestamp
            response = f"{timestamp} - {data.decode()}"
            sock.sendto(response.encode(), addr)
            
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur UDP")
    finally:
        sock.close()


[OK] CLIENT UDP

# === Client UDP basique ===
import socket

# Créer socket UDP
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# Adresse serveur
SERVER = '127.0.0.1'
PORT = 9090

# Envoyer données
message = "Hello UDP Server!"
udp_socket.sendto(message.encode(), (SERVER, PORT))

# Recevoir réponse
response, server_address = udp_socket.recvfrom(1024)
print(f"Réponse: {response.decode()}")

# Fermer
udp_socket.close()


# === Client UDP avec timeout ===
import socket

def udp_client(host='127.0.0.1', port=9090, message='Hello'):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(3.0)  # Timeout 3 secondes
    
    try:
        # Envoyer
        sock.sendto(message.encode(), (host, port))
        print(f"Envoyé à {host}:{port}")
        
        # Recevoir avec timeout
        data, addr = sock.recvfrom(4096)
        print(f"Réponse de {addr}: {data.decode()}")
        
    except socket.timeout:
        print("Timeout - pas de réponse")
    except Exception as e:
        print(f"Erreur: {e}")
    finally:
        sock.close()


[OK] OPTIONS SOCKET (SETSOCKOPT)

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# === Options niveau SOL_SOCKET ===

# Réutiliser adresse immédiatement (évite "Address already in use")
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# Réutiliser port (BSD seulement)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)

# Keep-alive (maintenir connexion)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

# Buffer de réception
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)

# Buffer d'envoi
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)

# Timeout réception (secondes, microsecondes)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVTIMEO, 
                socket.pack('ll', 5, 0))  # 5 secondes

# Timeout envoi
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDTIMEO,
                socket.pack('ll', 5, 0))

# Mode broadcast (UDP)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)


# === Options niveau IPPROTO_TCP ===

# Désactiver algorithme Nagle (envoi immédiat, latence réduite)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

# Keep-alive timeout
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)

# Keep-alive interval
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)

# Keep-alive probes
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)


# === Obtenir options (getsockopt) ===

# Obtenir valeur option
buffer_size = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
print(f"Buffer réception: {buffer_size} bytes")

# Vérifier si keep-alive actif
keepalive = sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE)
print(f"Keep-alive: {bool(keepalive)}")


[OK] MÉTHODES SOCKET PRINCIPALES

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# === Serveur ===

# Lier à adresse
sock.bind(('0.0.0.0', 8080))

# Écouter (backlog = nombre connexions en attente max)
sock.listen(5)

# Accepter connexion (bloquant)
client, address = sock.accept()

# Accepter avec timeout
sock.settimeout(5.0)
try:
    client, address = sock.accept()
except socket.timeout:
    print("Timeout")


# === Client ===

# Connecter (bloquant)
sock.connect(('127.0.0.1', 8080))

# Connecter avec timeout
sock.settimeout(3.0)
try:
    sock.connect(('127.0.0.1', 8080))
except socket.timeout:
    print("Connexion timeout")

# Connecter (non-bloquant) - retourne erreur si pas prêt
sock.setblocking(False)
try:
    sock.connect(('127.0.0.1', 8080))
except BlockingIOError:
    pass  # Normal pour non-bloquant


# === Envoi/Réception TCP ===

# Envoyer données (peut envoyer partiellement)
bytes_sent = sock.send(b"Hello")

# Envoyer toutes les données (boucle jusqu'à envoi complet)
sock.sendall(b"Hello World")

# Recevoir données (max 1024 bytes)
data = sock.recv(1024)

# Recevoir exactement N bytes
def recv_exactly(sock, n):
    data = b''
    while len(data) < n:
        chunk = sock.recv(n - len(data))
        if not chunk:
            raise ConnectionError("Connexion fermée")
        data += chunk
    return data


# === Envoi/Réception UDP ===

# Envoyer à adresse
sock.sendto(b"Hello", ('127.0.0.1', 9090))

# Recevoir avec adresse source
data, address = sock.recvfrom(1024)


# === Gestion connexion ===

# Fermer envoi (half-close)
sock.shutdown(socket.SHUT_WR)   # Plus d'envoi
sock.shutdown(socket.SHUT_RD)   # Plus de réception
sock.shutdown(socket.SHUT_RDWR) # Les deux

# Fermer socket complètement
sock.close()

# Obtenir nom local (adresse liée)
local = sock.getsockname()
print(f"Local: {local}")

# Obtenir nom distant (adresse connectée)
remote = sock.getpeername()
print(f"Remote: {remote}")


# === Modes bloquant/non-bloquant ===

# Mode bloquant (défaut)
sock.setblocking(True)

# Mode non-bloquant
sock.setblocking(False)

# Timeout (bloquant avec limite)
sock.settimeout(5.0)    # 5 secondes
sock.settimeout(None)   # Bloquant infini

# Obtenir timeout
timeout = sock.gettimeout()


[OK] SOCKET NON-BLOQUANT (SELECT)

import socket
import select

# === Serveur avec select (multiplexing I/O) ===

def select_server(host='0.0.0.0', port=8080):
    # Socket serveur
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    server.setblocking(False)
    
    print(f"Serveur select sur {host}:{port}")
    
    # Liste des sockets à surveiller
    sockets_list = [server]
    clients = {}
    
    try:
        while True:
            # Attendre activité (timeout 1 seconde)
            read_sockets, _, exception_sockets = select.select(
                sockets_list, [], sockets_list, 1.0
            )
            
            for notified_socket in read_sockets:
                # Nouvelle connexion
                if notified_socket == server:
                    client, addr = server.accept()
                    sockets_list.append(client)
                    clients[client] = addr
                    print(f"[+] Connexion: {addr}")
                
                # Données d'un client
                else:
                    try:
                        data = notified_socket.recv(1024)
                        
                        if data:
                            addr = clients[notified_socket]
                            print(f"[{addr}] {data.decode()}")
                            notified_socket.send(b"ACK\n")
                        else:
                            # Connexion fermée
                            addr = clients[notified_socket]
                            print(f"[-] Déconnexion: {addr}")
                            sockets_list.remove(notified_socket)
                            del clients[notified_socket]
                            notified_socket.close()
                    
                    except Exception as e:
                        print(f"Erreur: {e}")
                        sockets_list.remove(notified_socket)
                        del clients[notified_socket]
                        notified_socket.close()
            
            # Gérer exceptions
            for notified_socket in exception_sockets:
                sockets_list.remove(notified_socket)
                if notified_socket in clients:
                    del clients[notified_socket]
                notified_socket.close()
    
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur")
    finally:
        for sock in sockets_list:
            sock.close()


# === Client non-bloquant avec select ===

def nonblocking_client(host='127.0.0.1', port=8080):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setblocking(False)
    
    # Connexion non-bloquante
    try:
        sock.connect((host, port))
    except BlockingIOError:
        pass
    
    # Attendre que socket soit prêt
    _, write_sockets, _ = select.select([], [sock], [], 5.0)
    
    if sock in write_sockets:
        print("Connecté!")
        sock.send(b"Hello\n")
        
        # Attendre réponse
        read_sockets, _, _ = select.select([sock], [], [], 5.0)
        if sock in read_sockets:
            data = sock.recv(1024)
            print(f"Reçu: {data.decode()}")
    else:
        print("Timeout connexion")
    
    sock.close()


[OK] SOCKET ASYNCHRONE (ASYNCIO)

import asyncio

# === Serveur TCP asyncio ===

async def handle_client(reader, writer):
    """Gère un client de manière asynchrone"""
    addr = writer.get_extra_info('peername')
    print(f"[+] Connexion: {addr}")
    
    try:
        while True:
            data = await reader.read(1024)
            
            if not data:
                break
            
            message = data.decode()
            print(f"[{addr}] {message}")
            
            response = f"Echo: {message}"
            writer.write(response.encode())
            await writer.drain()
    
    except Exception as e:
        print(f"Erreur {addr}: {e}")
    finally:
        print(f"[-] Déconnexion: {addr}")
        writer.close()
        await writer.wait_closed()


async def async_server(host='127.0.0.1', port=8080):
    server = await asyncio.start_server(
        handle_client, host, port
    )
    
    addr = server.sockets[0].getsockname()
    print(f"[*] Serveur asyncio sur {addr}")
    
    async with server:
        await server.serve_forever()


# Démarrer serveur
if __name__ == '__main__':
    asyncio.run(async_server())


# === Client TCP asyncio ===

async def async_client(host='127.0.0.1', port=8080):
    try:
        reader, writer = await asyncio.open_connection(host, port)
        print(f"Connecté à {host}:{port}")
        
        # Envoyer
        message = "Hello async server!"
        writer.write(message.encode())
        await writer.drain()
        
        # Recevoir
        data = await reader.read(1024)
        print(f"Reçu: {data.decode()}")
        
        writer.close()
        await writer.wait_closed()
        
    except Exception as e:
        print(f"Erreur: {e}")


asyncio.run(async_client())


# === Serveur UDP asyncio ===

class UDPServerProtocol(asyncio.DatagramProtocol):
    def connection_made(self, transport):
        self.transport = transport
        print("[*] Serveur UDP asyncio démarré")
    
    def datagram_received(self, data, addr):
        message = data.decode()
        print(f"[{addr}] {message}")
        
        response = f"Echo: {message}"
        self.transport.sendto(response.encode(), addr)


async def udp_server_async(host='127.0.0.1', port=9090):
    loop = asyncio.get_running_loop()
    
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: UDPServerProtocol(),
        local_addr=(host, port)
    )
    
    try:
        await asyncio.sleep(3600)  # Garder actif 1h
    finally:
        transport.close()


[OK] FONCTIONS UTILITAIRES RÉSEAU

import socket
import struct

# === Résolution DNS ===

# Résoudre hostname en IP
ip = socket.gethostbyname('www.google.com')
print(ip)  # '142.250.185.36'

# Résoudre avec infos complètes
hostname, aliaslist, ipaddrlist = socket.gethostbyname_ex('www.google.com')
print(hostname)      # 'www.google.com'
print(ipaddrlist)    # ['142.250.185.36', ...]

# Résolution inverse (IP -> hostname)
hostname = socket.gethostbyaddr('8.8.8.8')
print(hostname)  # ('dns.google', [], ['8.8.8.8'])

# Obtenir hostname local
local_hostname = socket.gethostname()
print(local_hostname)

# Obtenir FQDN (Fully Qualified Domain Name)
fqdn = socket.getfqdn()
print(fqdn)

# Obtenir IP locale
local_ip = socket.gethostbyname(socket.gethostname())


# === getaddrinfo (moderne, IPv4/IPv6) ===

# Résoudre avec protocole spécifique
results = socket.getaddrinfo(
    'www.google.com',   # hostname
    80,                 # port
    socket.AF_UNSPEC,   # AF_INET, AF_INET6, AF_UNSPEC
    socket.SOCK_STREAM  # SOCK_STREAM ou SOCK_DGRAM
)

for result in results:
    family, socktype, proto, canonname, sockaddr = result
    print(f"{sockaddr}")

# Exemple connexion avec getaddrinfo
def connect_to(host, port):
    for res in socket.getaddrinfo(host, port, socket.AF_UNSPEC, 
                                   socket.SOCK_STREAM):
        af, socktype, proto, canonname, sa = res
        try:
            sock = socket.socket(af, socktype, proto)
            sock.connect(sa)
            return sock
        except OSError:
            sock.close()
            continue
    raise OSError("Impossible de se connecter")


# === Conversion adresses ===

# IP string -> bytes (network byte order)
ip_packed = socket.inet_aton('192.168.1.1')
print(ip_packed)  # b'\xc0\xa8\x01\x01'

# Bytes -> IP string
ip_string = socket.inet_ntoa(ip_packed)
print(ip_string)  # '192.168.1.1'

# IPv4/IPv6 (moderne)
# String -> bytes
ip_packed = socket.inet_pton(socket.AF_INET, '192.168.1.1')
ipv6_packed = socket.inet_pton(socket.AF_INET6, '::1')

# Bytes -> string
ip_string = socket.inet_ntop(socket.AF_INET, ip_packed)
ipv6_string = socket.inet_ntop(socket.AF_INET6, ipv6_packed)


# === Conversion byte order ===

# Host to network (long)
net_long = socket.htonl(12345)

# Network to host (long)
host_long = socket.ntohl(net_long)

# Host to network (short)
net_short = socket.htons(80)

# Network to host (short)
host_short = socket.ntohs(net_short)


# === Informations protocoles ===

# Nom protocole -> numéro
proto_num = socket.getprotobyname('tcp')  # 6
proto_num = socket.getprotobyname('udp')  # 17

# Nom service -> port
port = socket.getservbyname('http')       # 80
port = socket.getservbyname('https')      # 443

# Port -> nom service
service = socket.getservbyport(80)        # 'http'
service = socket.getservbyport(443)       # 'https'


[OK] TRANSFERT FICHIERS

# === Envoyer fichier (TCP) ===

def send_file(sock, filepath):
    """Envoie un fichier via socket TCP"""
    import os
    
    # Taille fichier
    filesize = os.path.getsize(filepath)
    
    # Envoyer nom fichier et taille
    filename = os.path.basename(filepath)
    header = f"{filename}|{filesize}".encode()
    sock.send(header + b'\n')
    
    # Attendre ACK
    ack = sock.recv(3)
    
    # Envoyer fichier
    with open(filepath, 'rb') as f:
        sent = 0
        while sent < filesize:
            chunk = f.read(4096)
            sock.sendall(chunk)
            sent += len(chunk)
            progress = (sent / filesize) * 100
            print(f"\rEnvoi: {progress:.1f}%", end='')
    
    print("\nFichier envoyé!")


# === Recevoir fichier (TCP) ===

def receive_file(sock, save_dir='.'):
    """Reçoit un fichier via socket TCP"""
    import os
    
    # Recevoir header
    header = b''
    while b'\n' not in header:
        header += sock.recv(1)
    
    filename, filesize = header.decode().strip().split('|')
    filesize = int(filesize)
    
    # Envoyer ACK
    sock.send(b'ACK')
    
    # Recevoir fichier
    filepath = os.path.join(save_dir, filename)
    with open(filepath, 'wb') as f:
        received = 0
        while received < filesize:
            chunk = sock.recv(min(4096, filesize - received))
            f.write(chunk)
            received += len(chunk)
            progress = (received / filesize) * 100
            print(f"\rRéception: {progress:.1f}%", end='')
    
    print(f"\nFichier reçu: {filepath}")
    return filepath


# === Exemple serveur transfert fichier ===

def file_server(host='0.0.0.0', port=8080):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(1)
    
    print(f"Serveur fichiers sur {host}:{port}")
    
    while True:
        client, addr = server.accept()
        print(f"Connexion: {addr}")
        
        try:
            receive_file(client, save_dir='./uploads')
        except Exception as e:
            print(f"Erreur: {e}")
        finally:
            client.close()


[OK] HTTP CLIENT BASIQUE (SANS REQUESTS)

# === Requête HTTP GET simple ===

def http_get(host, path='/', port=80):
    """Effectue une requête HTTP GET"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        sock.connect((host, port))
        
        # Construire requête HTTP
        request = f"GET {path} HTTP/1.1\r\n"
        request += f"Host: {host}\r\n"
        request += "User-Agent: Python-Socket/1.0\r\n"
        request += "Connection: close\r\n"
        request += "\r\n"
        
        sock.sendall(request.encode())
        
        # Recevoir réponse
        response = b''
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
        
        return response.decode('utf-8', errors='ignore')
        
    finally:
        sock.close()


# Utilisation
response = http_get('www.example.com', '/')
print(response)


# === Serveur HTTP minimaliste ===

def mini_http_server(host='0.0.0.0', port=8000):
    """Serveur HTTP simple servant du HTML"""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    
    print(f"Serveur HTTP sur http://{host}:{port}")
    
    html_content = """
    <!DOCTYPE html>
    <html>
    <head><title>Python Socket Server</title></head>
    <body>
        <h1>Hello from Python Socket!</h1>
        <p>Serveur HTTP minimaliste</p>
    </body>
    </html>
    """
    
    try:
        while True:
            client, addr = server.accept()
            
            try:
                # Recevoir requête
                request = client.recv(4096).decode()
                print(f"[{addr}] {request.split()[0:2]}")
                
                # Réponse HTTP
                response = "HTTP/1.1 200 OK\r\n"
                response += "Content-Type: text/html; charset=utf-8\r\n"
                response += f"Content-Length: {len(html_content)}\r\n"
                response += "Connection: close\r\n"
                response += "\r\n"
                response += html_content
                
                client.sendall(response.encode())
                
            except Exception as e:
                print(f"Erreur: {e}")
            finally:
                client.close()
                
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur")
    finally:
        server.close()


[OK] SSL/TLS (CONNEXIONS SÉCURISÉES)

import socket
import ssl

# === Client HTTPS ===

def https_get(host, path='/', port=443):
    """Requête HTTPS avec SSL"""
    # Socket normal
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # Créer contexte SSL
    context = ssl.create_default_context()
    
    # Wrapper SSL
    secure_sock = context.wrap_socket(sock, server_hostname=host)
    
    try:
        secure_sock.connect((host, port))
        
        # Vérifier certificat
        cert = secure_sock.getpeercert()
        print(f"Certificat: {cert['subject']}")
        
        # Requête HTTPS
        request = f"GET {path} HTTP/1.1\r\n"
        request += f"Host: {host}\r\n"
        request += "Connection: close\r\n\r\n"
        
        secure_sock.sendall(request.encode())
        
        # Recevoir réponse
        response = b''
        while True:
            chunk = secure_sock.recv(4096)
            if not chunk:
                break
            response += chunk
        
        return response.decode('utf-8', errors='ignore')
        
    finally:
        secure_sock.close()


# Utilisation
response = https_get('www.google.com', '/')


# === Serveur SSL/TLS ===

def ssl_server(host='0.0.0.0', port=8443, 
               certfile='server.crt', keyfile='server.key'):
    """Serveur TCP avec SSL/TLS"""
    
    # Socket serveur
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    
    # Contexte SSL
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain(certfile=certfile, keyfile=keyfile)
    
    print(f"Serveur SSL/TLS sur {host}:{port}")
    
    try:
        while True:
            client, addr = server.accept()
            
            # Wrapper SSL
            secure_client = context.wrap_socket(client, server_side=True)
            
            print(f"[+] Connexion SSL: {addr}")
            
            try:
                data = secure_client.recv(1024)
                print(f"Reçu: {data.decode()}")
                
                secure_client.send(b"OK\n")
                
            finally:
                secure_client.close()
                
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur SSL")
    finally:
        server.close()


# === Générer certificat auto-signé (pour tests) ===
# Terminal:
# openssl req -x509 -newkey rsa:4096 -nodes \
#   -keyout server.key -out server.crt -days 365 \
#   -subj "/CN=localhost"


# === Options SSL avancées ===

# Contexte avec options personnalisées
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain('server.crt', 'server.key')

# Vérifier certificats clients
context.verify_mode = ssl.CERT_REQUIRED
context.load_verify_locations('ca.crt')

# Protocoles et ciphers
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM')


[OK] BROADCAST ET MULTICAST

# === UDP Broadcast ===

def udp_broadcast_sender(port=9090, message="Hello"):
    """Envoie message en broadcast"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    
    broadcast_address = ('255.255.255.255', port)
    sock.sendto(message.encode(), broadcast_address)
    print(f"Broadcast envoyé: {message}")
    
    sock.close()


def udp_broadcast_receiver(port=9090):
    """Reçoit messages broadcast"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(('', port))
    
    print(f"Écoute broadcast sur port {port}")
    
    try:
        while True:
            data, addr = sock.recvfrom(1024)
            print(f"[{addr}] {data.decode()}")
    except KeyboardInterrupt:
        print("\n[*] Arrêt")
    finally:
        sock.close()


# === Multicast ===

import struct

def multicast_sender(group='224.1.1.1', port=5007, message="Hello Multicast"):
    """Envoie message multicast"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
    
    sock.sendto(message.encode(), (group, port))
    print(f"Multicast envoyé à {group}:{port}")
    
    sock.close()


def multicast_receiver(group='224.1.1.1', port=5007):
    """Reçoit messages multicast"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # Bind au port
    sock.bind(('', port))
    
    # Joindre groupe multicast
    mreq = struct.pack('4sl', socket.inet_aton(group), socket.INADDR_ANY)
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
    
    print(f"Écoute multicast {group}:{port}")
    
    try:
        while True:
            data, addr = sock.recvfrom(1024)
            print(f"[{addr}] {data.decode()}")
    except KeyboardInterrupt:
        print("\n[*] Arrêt")
    finally:
        sock.close()


[OK] UNIX DOMAIN SOCKETS (IPC LOCAL)

import socket
import os

# === Serveur Unix Socket ===

def unix_socket_server(socket_path='/tmp/my.sock'):
    """Serveur utilisant Unix domain socket"""
    
    # Supprimer socket existant
    try:
        os.unlink(socket_path)
    except OSError:
        if os.path.exists(socket_path):
            raise
    
    # Créer socket Unix
    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server.bind(socket_path)
    server.listen(5)
    
    print(f"Serveur Unix socket: {socket_path}")
    
    try:
        while True:
            client, _ = server.accept()
            
            try:
                data = client.recv(1024)
                print(f"Reçu: {data.decode()}")
                
                client.send(b"ACK\n")
            finally:
                client.close()
                
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur")
    finally:
        server.close()
        os.unlink(socket_path)


# === Client Unix Socket ===

def unix_socket_client(socket_path='/tmp/my.sock', message='Hello'):
    """Client Unix domain socket"""
    client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    
    try:
        client.connect(socket_path)
        
        client.send(message.encode())
        response = client.recv(1024)
        print(f"Réponse: {response.decode()}")
        
    finally:
        client.close()


[OK] RAW SOCKETS (NIVEAU IP)

import socket
import struct

# Nécessite privilèges root/admin !

# === Créer raw socket ===

# IPv4 raw socket
raw_sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)

# Capturer tous les paquets IP
sniff_sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP)


# === Sniffer de paquets basique ===

def packet_sniffer(interface='eth0'):
    """Capture paquets réseau (nécessite root)"""
    
    # Linux: capturer tous protocoles
    sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(3))
    
    # Bind à interface
    # sock.bind((interface, 0))
    
    print(f"Capture paquets sur {interface}")
    
    try:
        while True:
            packet = sock.recvfrom(65565)
            
            # packet[0] = données
            # packet[1] = informations
            
            ethernet_data = packet[0]
            
            # Header Ethernet (14 bytes)
            eth_header = struct.unpack('!6s6sH', ethernet_data[:14])
            eth_protocol = socket.ntohs(eth_header[2])
            
            print(f"Protocole Ethernet: {eth_protocol}")
            
            # Si IP (0x0800)
            if eth_protocol == 8:
                ip_header = ethernet_data[14:34]
                iph = struct.unpack('!BBHHHBBH4s4s', ip_header)
                
                version_ihl = iph[0]
                version = version_ihl >> 4
                ihl = version_ihl & 0xF
                
                protocol = iph[6]
                src_addr = socket.inet_ntoa(iph[8])
                dst_addr = socket.inet_ntoa(iph[9])
                
                print(f"  IP: {src_addr} -> {dst_addr} (proto: {protocol})")
                
    except KeyboardInterrupt:
        print("\n[*] Arrêt capture")
    finally:
        sock.close()


# === Envoyer paquet IP custom ===

def send_raw_packet(src_ip, dst_ip, data):
    """Envoie paquet IP brut (nécessite root)"""
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
    
    # Construction header IP
    version = 4
    ihl = 5
    tos = 0
    tot_len = 20 + len(data)
    id = 54321
    frag_off = 0
    ttl = 64
    protocol = socket.IPPROTO_TCP
    check = 0
    saddr = socket.inet_aton(src_ip)
    daddr = socket.inet_aton(dst_ip)
    
    ihl_version = (version << 4) + ihl
    
    # Header IP (sans checksum)
    ip_header = struct.pack('!BBHHHBBH4s4s',
        ihl_version, tos, tot_len, id, frag_off,
        ttl, protocol, check, saddr, daddr
    )
    
    # Paquet complet
    packet = ip_header + data.encode()
    
    # Envoyer
    sock.sendto(packet, (dst_ip, 0))
    sock.close()


[OK] GESTION D'ERREURS SOCKET

import socket
import errno

# === Erreurs courantes ===

try:
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(('127.0.0.1', 8080))
    
except socket.timeout:
    print("Timeout de connexion")

except ConnectionRefusedError:
    print("Connexion refusée (serveur pas démarré?)")

except ConnectionResetError:
    print("Connexion réinitialisée par pair")

except ConnectionAbortedError:
    print("Connexion interrompue")

except BrokenPipeError:
    print("Broken pipe (écriture sur socket fermé)")

except OSError as e:
    if e.errno == errno.EADDRINUSE:
        print("Adresse déjà utilisée")
    elif e.errno == errno.ECONNREFUSED:
        print("Connexion refusée")
    elif e.errno == errno.ETIMEDOUT:
        print("Timeout")
    elif e.errno == errno.ENETUNREACH:
        print("Réseau inaccessible")
    elif e.errno == errno.EHOSTUNREACH:
        print("Hôte inaccessible")
    else:
        print(f"Erreur OS: {e}")

except socket.gaierror as e:
    print(f"Erreur résolution DNS: {e}")

except socket.herror as e:
    print(f"Erreur hôte: {e}")

except Exception as e:
    print(f"Erreur inattendue: {e}")

finally:
    sock.close()


# === Gestion robuste ===

def robust_connect(host, port, timeout=5, retries=3):
    """Connexion avec retry"""
    for attempt in range(retries):
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            sock.connect((host, port))
            return sock
            
        except (socket.timeout, ConnectionRefusedError) as e:
            print(f"Tentative {attempt + 1}/{retries} échouée: {e}")
            if sock:
                sock.close()
            if attempt < retries - 1:
                import time
                time.sleep(1)
        except Exception as e:
            print(f"Erreur: {e}")
            if sock:
                sock.close()
            raise
    
    raise ConnectionError(f"Impossible de se connecter après {retries} tentatives")


[OK] CONTEXT MANAGER (WITH)

import socket

# === Socket avec context manager ===

# Fermeture automatique
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    sock.connect(('127.0.0.1', 8080))
    sock.send(b"Hello\n")
    data = sock.recv(1024)
    print(data.decode())
# Socket fermé automatiquement


# === Custom context manager ===

class SocketConnection:
    """Context manager pour socket avec auto-reconnexion"""
    
    def __init__(self, host, port, timeout=5):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.sock = None
    
    def __enter__(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.settimeout(self.timeout)
        self.sock.connect((self.host, self.port))
        return self.sock
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.sock:
            self.sock.close()
        return False


# Utilisation
with SocketConnection('127.0.0.1', 8080) as sock:
    sock.send(b"Hello\n")
    response = sock.recv(1024)


[OK] PATTERNS AVANCÉS

# === Pool de connexions ===

import queue
import threading

class SocketPool:
    """Pool de connexions socket réutilisables"""
    
    def __init__(self, host, port, pool_size=5):
        self.host = host
        self.port = port
        self.pool = queue.Queue(maxsize=pool_size)
        
        # Créer connexions initiales
        for _ in range(pool_size):
            sock = self._create_connection()
            self.pool.put(sock)
    
    def _create_connection(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((self.host, self.port))
        return sock
    
    def get_connection(self, timeout=5):
        """Obtenir connexion du pool"""
        try:
            return self.pool.get(timeout=timeout)
        except queue.Empty:
            # Créer nouvelle connexion si pool vide
            return self._create_connection()
    
    def return_connection(self, sock):
        """Retourner connexion au pool"""
        try:
            self.pool.put_nowait(sock)
        except queue.Full:
            sock.close()
    
    def close_all(self):
        """Fermer toutes les connexions"""
        while not self.pool.empty():
            sock = self.pool.get()
            sock.close()


# Utilisation
pool = SocketPool('127.0.0.1', 8080, pool_size=10)

sock = pool.get_connection()
sock.send(b"Hello\n")
response = sock.recv(1024)
pool.return_connection(sock)


# === Protocol handler pattern ===

class Protocol:
    """Classe de base pour protocole"""
    
    def __init__(self, socket):
        self.socket = socket
    
    def send_message(self, message):
        raise NotImplementedError
    
    def receive_message(self):
        raise NotImplementedError


class LengthPrefixedProtocol(Protocol):
    """Protocole avec longueur en préfixe"""
    
    def send_message(self, message):
        data = message.encode()
        length = len(data)
        # Envoyer longueur (4 bytes) puis données
        self.socket.sendall(struct.pack('!I', length) + data)
    
    def receive_message(self):
        # Recevoir longueur
        length_data = self._recv_exactly(4)
        length = struct.unpack('!I', length_data)[0]
        
        # Recevoir message
        data = self._recv_exactly(length)
        return data.decode()
    
    def _recv_exactly(self, n):
        data = b''
        while len(data) < n:
            chunk = self.socket.recv(n - len(data))
            if not chunk:
                raise ConnectionError("Connexion fermée")
            data += chunk
        return data


# Utilisation
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('127.0.0.1', 8080))

protocol = LengthPrefixedProtocol(sock)
protocol.send_message("Hello, this is a test message!")
response = protocol.receive_message()


# === Heartbeat pattern ===

import time
import threading

class HeartbeatSocket:
    """Socket avec heartbeat automatique"""
    
    def __init__(self, sock, interval=30, heartbeat_msg=b'PING'):
        self.sock = sock
        self.interval = interval
        self.heartbeat_msg = heartbeat_msg
        self.alive = True
        self.last_activity = time.time()
        
        # Thread heartbeat
        self.heartbeat_thread = threading.Thread(target=self._heartbeat_loop)
        self.heartbeat_thread.daemon = True
        self.heartbeat_thread.start()
    
    def _heartbeat_loop(self):
        while self.alive:
            time.sleep(self.interval)
            if time.time() - self.last_activity > self.interval:
                try:
                    self.sock.send(self.heartbeat_msg)
                    self.last_activity = time.time()
                except:
                    self.alive = False
    
    def send(self, data):
        self.sock.send(data)
        self.last_activity = time.time()
    
    def recv(self, bufsize):
        data = self.sock.recv(bufsize)
        self.last_activity = time.time()
        return data
    
    def close(self):
        self.alive = False
        self.sock.close()


[OK] DEBUGGING ET MONITORING

import socket
import logging

# === Logging socket ===

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class LoggingSocket:
    """Wrapper socket avec logging"""
    
    def __init__(self, sock):
        self.sock = sock
        self.logger = logging.getLogger('socket')
    
    def connect(self, address):
        self.logger.info(f"Connexion à {address}")
        try:
            self.sock.connect(address)
            self.logger.info(f"Connecté à {address}")
        except Exception as e:
            self.logger.error(f"Échec connexion: {e}")
            raise
    
    def send(self, data):
        self.logger.debug(f"Envoi {len(data)} bytes: {data[:50]}")
        return self.sock.send(data)
    
    def recv(self, bufsize):
        data = self.sock.recv(bufsize)
        self.logger.debug(f"Reçu {len(data)} bytes: {data[:50]}")
        return data
    
    def close(self):
        self.logger.info("Fermeture socket")
        self.sock.close()


# === Statistiques socket ===

class SocketStats:
    """Collecte statistiques socket"""
    
    def __init__(self, sock):
        self.sock = sock
        self.bytes_sent = 0
        self.bytes_received = 0
        self.messages_sent = 0
        self.messages_received = 0
        self.errors = 0
        self.start_time = time.time()
    
    def send(self, data):
        try:
            sent = self.sock.send(data)
            self.bytes_sent += sent
            self.messages_sent += 1
            return sent
        except Exception as e:
            self.errors += 1
            raise
    
    def recv(self, bufsize):
        try:
            data = self.sock.recv(bufsize)
            self.bytes_received += len(data)
            if data:
                self.messages_received += 1
            return data
        except Exception as e:
            self.errors += 1
            raise
    
    def get_stats(self):
        elapsed = time.time() - self.start_time
        return {
            'bytes_sent': self.bytes_sent,
            'bytes_received': self.bytes_received,
            'messages_sent': self.messages_sent,
            'messages_received': self.messages_received,
            'errors': self.errors,
            'uptime': elapsed,
            'send_rate': self.bytes_sent / elapsed if elapsed > 0 else 0,
            'recv_rate': self.bytes_received / elapsed if elapsed > 0 else 0
        }


[OK] TESTS SOCKET

import socket
import unittest
import threading
import time

# === Test serveur/client ===

class TestSocketServer(unittest.TestCase):
    
    @classmethod
    def setUpClass(cls):
        """Démarrer serveur de test"""
        cls.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        cls.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        cls.server.bind(('127.0.0.1', 0))  # Port aléatoire
        cls.port = cls.server.getsockname()[1]
        cls.server.listen(1)
        
        # Thread serveur
        cls.server_thread = threading.Thread(target=cls.run_server)
        cls.server_thread.daemon = True
        cls.server_thread.start()
        
        time.sleep(0.1)  # Attendre démarrage
    
    @classmethod
    def run_server(cls):
        while True:
            try:
                client, addr = cls.server.accept()
                data = client.recv(1024)
                client.send(b"ECHO:" + data)
                client.close()
            except:
                break
    
    def test_connection(self):
        """Test connexion basique"""
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.connect(('127.0.0.1', self.port))
        client.close()
    
    def test_echo(self):
        """Test echo"""
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.connect(('127.0.0.1', self.port))
        
        client.send(b"TEST")
        response = client.recv(1024)
        
        self.assertEqual(response, b"ECHO:TEST")
        client.close()
    
    @classmethod
    def tearDownClass(cls):
        """Arrêter serveur"""
        cls.server.close()


if __name__ == '__main__':
    unittest.main()


[OK] PERFORMANCE ET OPTIMISATION

# === Buffer sizing ===

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Augmenter buffers pour haute performance
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 2**20)  # 1MB
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2**20)  # 1MB

# Vérifier tailles effectives
rcvbuf = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
sndbuf = sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF)
print(f"RCV: {rcvbuf}, SND: {sndbuf}")


# === Zero-copy avec sendfile ===

import os

def send_file_optimized(sock, filepath):
    """Envoi fichier optimisé (Linux uniquement)"""
    with open(filepath, 'rb') as f:
        # sendfile: transfert zero-copy kernel-space
        offset = 0
        filesize = os.path.getsize(filepath)
        
        while offset < filesize:
            sent = os.sendfile(sock.fileno(), f.fileno(), offset, 
                              filesize - offset)
            offset += sent


# === Benchmark socket ===

import time

def benchmark_socket(host='127.0.0.1', port=8080, 
                     data_size=1024, iterations=1000):
    """Benchmark performance socket"""
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))
    
    data = b'X' * data_size
    
    start = time.time()
    
    for _ in range(iterations):
        sock.sendall(data)
        response = sock.recv(data_size)
    
    elapsed = time.time() - start
    
    throughput = (data_size * iterations * 2) / elapsed / 1024 / 1024
    latency = (elapsed / iterations) * 1000
    
    print(f"Throughput: {throughput:.2f} MB/s")
    print(f"Latency moyenne: {latency:.2f} ms")
    
    sock.close()


[OK] SÉCURITÉ

# === Validation entrées ===

def safe_recv(sock, max_size=4096):
    """Réception sécurisée avec limite taille"""
    data = sock.recv(min(max_size, 4096))
    
    if len(data) > max_size:
        raise ValueError("Données trop volumineuses")
    
    return data


# === Rate limiting ===

import time
from collections import deque

class RateLimiter:
    """Limite taux de requêtes"""
    
    def __init__(self, max_requests=100, window=60):
        self.max_requests = max_requests
        self.window = window
        self.requests = {}
    
    def allow_request(self, client_addr):
        now = time.time()
        
        if client_addr not in self.requests:
            self.requests[client_addr] = deque()
        
        # Nettoyer anciennes requêtes
        while (self.requests[client_addr] and 
               now - self.requests[client_addr][0] > self.window):
            self.requests[client_addr].popleft()
        
        # Vérifier limite
        if len(self.requests[client_addr]) >= self.max_requests:
            return False
        
        self.requests[client_addr].append(now)
        return True


# === Timeout protection ===

def safe_server(host='0.0.0.0', port=8080):
    """Serveur avec protections"""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    
    rate_limiter = RateLimiter(max_requests=10, window=60)
    
    while True:
        client, addr = server.accept()
        
        # Rate limiting
        if not rate_limiter.allow_request(addr):
            client.send(b"429 Too Many Requests\n")
            client.close()
            continue
        
        # Timeout client
        client.settimeout(5.0)
        
        try:
            # Limite taille données
            data = safe_recv(client, max_size=1024)
            
            # Traitement...
            client.send(b"OK\n")
            
        except socket.timeout:
            client.send(b"408 Request Timeout\n")
        except Exception as e:
            print(f"Erreur: {e}")
        finally:
            client.close()


[OK] BONNES PRATIQUES

# 1. TOUJOURS fermer les sockets
try:
    sock = socket.socket()
    # Utiliser socket...
finally:
    sock.close()

# Ou utiliser context manager
with socket.socket() as sock:
    pass  # Auto-fermé


# 2. Gérer tous les cas d'erreur
try:
    sock.connect((host, port))
except ConnectionRefusedError:
    # Serveur pas démarré
    pass
except socket.timeout:
    # Timeout
    pass
except Exception as e:
    # Autre erreur
    pass


# 3. Utiliser SO_REUSEADDR pour serveurs
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)


# 4. Définir timeouts
sock.settimeout(5.0)  # Évite blocage infini


# 5. Recevoir en boucle pour données complètes
def recv_all(sock, size):
    data = b''
    while len(data) < size:
        chunk = sock.recv(size - len(data))
        if not chunk:
            raise ConnectionError("Connexion fermée")
        data += chunk
    return data


# 6. Utiliser sendall() au lieu de send()
sock.sendall(data)  # Envoie tout


# 7. Encoder/décoder explicitement
sock.send(message.encode('utf-8'))
data = sock.recv(1024).decode('utf-8')


# 8. Gérer SIGPIPE (Unix)
import signal
signal.signal(signal.SIGPIPE, signal.SIG_IGN)


# 9. Nettoyer ressources
def cleanup(server, clients):
    for client in clients:
        client.close()
    server.close()


# 10. Logger les erreurs
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

try:
    sock.connect((host, port))
except Exception as e:
    logger.error(f"Erreur connexion: {e}")


[OK] EXEMPLES COMPLETS

# === Chat serveur/client ===

# chat_server.py
import socket
import threading

clients = []
clients_lock = threading.Lock()

def broadcast(message, sender=None):
    """Envoie message à tous les clients"""
    with clients_lock:
        for client in clients:
            if client != sender:
                try:
                    client.send(message)
                except:
                    clients.remove(client)

def handle_client(client, addr):
    """Gère un client chat"""
    print(f"[+] {addr} connecté")
    
    # Demander pseudo
    client.send(b"Pseudo: ")
    username = client.recv(1024).decode().strip()
    
    with clients_lock:
        clients.append(client)
    
    broadcast(f"[{username}] a rejoint le chat\n".encode())
    
    try:
        while True:
            message = client.recv(1024)
            if not message:
                break
            
            formatted = f"[{username}] {message.decode()}"
            print(formatted)
            broadcast(formatted.encode(), sender=client)
    
    except Exception as e:
        print(f"Erreur {addr}: {e}")
    finally:
        with clients_lock:
            if client in clients:
                clients.remove(client)
        broadcast(f"[{username}] a quitté le chat\n".encode())
        client.close()
        print(f"[-] {addr} déconnecté")

def start_chat_server(host='0.0.0.0', port=9999):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen()
    
    print(f"[*] Chat serveur sur {host}:{port}")
    
    try:
        while True:
            client, addr = server.accept()
            thread = threading.Thread(target=handle_client, 
                                     args=(client, addr))
            thread.daemon = True
            thread.start()
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur")
    finally:
        server.close()


# chat_client.py
import socket
import threading

def receive_messages(sock):
    """Reçoit messages du serveur"""
    while True:
        try:
            message = sock.recv(1024).decode()
            if message:
                print(message, end='')
            else:
                break
        except:
            break

def start_chat_client(host='127.0.0.1', port=9999):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        client.connect((host, port))
        
        # Thread réception
        recv_thread = threading.Thread(target=receive_messages, 
                                      args=(client,))
        recv_thread.daemon = True
        recv_thread.start()
        
        # Boucle envoi
        while True:
            message = input()
            if message.lower() == '/quit':
                break
            client.send(f"{message}\n".encode())
    
    except Exception as e:
        print(f"Erreur: {e}")
    finally:
        client.close()


[OK] RESSOURCES

# Documentation officielle:
# https://docs.python.org/3/library/socket.html
# https://docs.python.org/3/library/ssl.html
# https://docs.python.org/3/library/select.html
# https://docs.python.org/3/library/asyncio.html

# Guides:
# - Real Python - Socket Programming Guide
# - Beej's Guide to Network Programming
# - Python Network Programming Cookbook

# RFCs importantes:
# - RFC 793: TCP
# - RFC 768: UDP
# - RFC 791: IP
# - RFC 5246: TLS 1.2
# - RFC 8446: TLS 1.3

# Outils debugging:
# - netcat (nc): Test connexions
# - telnet: Test TCP
# - wireshark: Capture paquets
# - tcpdump: Capture ligne de commande
# - nmap: Scanner réseau
# - socat: Multiplexeur socket avancé


[OK] EXEMPLES PRATIQUES SUPPLÉMENTAIRES

# === Port scanner ===

def port_scanner(host, start_port=1, end_port=1024, timeout=1):
    """Scanner de ports simple"""
    open_ports = []
    
    print(f"Scan de {host} ports {start_port}-{end_port}")
    
    for port in range(start_port, end_port + 1):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        
        result = sock.connect_ex((host, port))
        
        if result == 0:
            try:
                service = socket.getservbyport(port)
            except:
                service = "unknown"
            
            print(f"[+] Port {port} ouvert ({service})")
            open_ports.append(port)
        
        sock.close()
    
    print(f"\nPorts ouverts: {len(open_ports)}")
    return open_ports


# Utilisation
# port_scanner('127.0.0.1', 1, 100)


# === Reverse shell (pour tests uniquement!) ===

def reverse_shell_server(host='0.0.0.0', port=4444):
    """Serveur recevant reverse shell (tests seulement!)"""
    import subprocess
    
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(1)
    
    print(f"[*] Écoute sur {host}:{port}")
    
    client, addr = server.accept()
    print(f"[+] Connexion de {addr}")
    
    while True:
        command = input("shell> ")
        
        if command.lower() in ['exit', 'quit']:
            break
        
        client.send(command.encode() + b'\n')
        response = client.recv(4096)
        print(response.decode())
    
    client.close()
    server.close()


# === Proxy TCP simple ===

def tcp_proxy(local_host='127.0.0.1', local_port=8080,
              remote_host='example.com', remote_port=80):
    """Proxy TCP simple"""
    
    def forward_data(source, destination):
        """Transfère données entre sockets"""
        try:
            while True:
                data = source.recv(4096)
                if not data:
                    break
                destination.sendall(data)
        except:
            pass
    
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((local_host, local_port))
    server.listen(5)
    
    print(f"[*] Proxy {local_host}:{local_port} -> {remote_host}:{remote_port}")
    
    try:
        while True:
            client, addr = server.accept()
            print(f"[+] Connexion de {addr}")
            
            # Connexion au serveur distant
            remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                remote.connect((remote_host, remote_port))
                
                # Threads pour forward bidirectionnel
                client_to_remote = threading.Thread(
                    target=forward_data, args=(client, remote)
                )
                remote_to_client = threading.Thread(
                    target=forward_data, args=(remote, client)
                )
                
                client_to_remote.daemon = True
                remote_to_client.daemon = True
                
                client_to_remote.start()
                remote_to_client.start()
                
            except Exception as e:
                print(f"[!] Erreur connexion distante: {e}")
                client.close()
                remote.close()
                
    except KeyboardInterrupt:
        print("\n[*] Arrêt proxy")
    finally:
        server.close()


# === TCP tunnel (port forwarding) ===

def tcp_tunnel(local_port, remote_host, remote_port):
    """Tunnel TCP - forward port local vers distant"""
    
    def handle_tunnel(local_sock):
        remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
        try:
            remote_sock.connect((remote_host, remote_port))
            
            # Bidirectional forwarding
            def forward(src, dst):
                while True:
                    try:
                        data = src.recv(4096)
                        if not data:
                            break
                        dst.sendall(data)
                    except:
                        break
            
            t1 = threading.Thread(target=forward, args=(local_sock, remote_sock))
            t2 = threading.Thread(target=forward, args=(remote_sock, local_sock))
            
            t1.daemon = True
            t2.daemon = True
            
            t1.start()
            t2.start()
            
            t1.join()
            t2.join()
            
        finally:
            local_sock.close()
            remote_sock.close()
    
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(('127.0.0.1', local_port))
    server.listen(5)
    
    print(f"[*] Tunnel: localhost:{local_port} -> {remote_host}:{remote_port}")
    
    try:
        while True:
            client, addr = server.accept()
            print(f"[+] Nouvelle connexion tunnel: {addr}")
            
            thread = threading.Thread(target=handle_tunnel, args=(client,))
            thread.daemon = True
            thread.start()
            
    except KeyboardInterrupt:
        print("\n[*] Arrêt tunnel")
    finally:
        server.close()


# === DNS resolver custom ===

def dns_lookup(hostname, dns_server='8.8.8.8', port=53):
    """Requête DNS custom (A record)"""
    import struct
    import random
    
    # Créer socket UDP
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(3)
    
    # Construction requête DNS
    transaction_id = random.randint(0, 65535)
    flags = 0x0100  # Standard query
    questions = 1
    
    # Header DNS
    header = struct.pack('!HHHHHH', transaction_id, flags, 
                         questions, 0, 0, 0)
    
    # Question
    question = b''
    for part in hostname.split('.'):
        question += struct.pack('B', len(part)) + part.encode()
    question += b'\x00'  # Fin du nom
    question += struct.pack('!HH', 1, 1)  # Type A, Class IN
    
    query = header + question
    
    try:
        # Envoyer requête
        sock.sendto(query, (dns_server, port))
        
        # Recevoir réponse
        response, _ = sock.recvfrom(512)
        
        # Parser réponse (simplifié)
        # Skip header et question
        offset = len(query)
        
        # Lire réponse
        # Format: name (variable), type (2), class (2), TTL (4), length (2), data (variable)
        
        # Skip name (compression pointer)
        if response[offset] & 0xC0:
            offset += 2
        
        # Type, class, TTL
        rtype, rclass, ttl = struct.unpack('!HHI', response[offset:offset+8])
        offset += 8
        
        # Data length
        data_len = struct.unpack('!H', response[offset:offset+2])[0]
        offset += 2
        
        # IP address (pour type A)
        if rtype == 1 and data_len == 4:
            ip = socket.inet_ntoa(response[offset:offset+4])
            return ip
        
    except socket.timeout:
        print("Timeout DNS")
    except Exception as e:
        print(f"Erreur DNS: {e}")
    finally:
        sock.close()
    
    return None


# Utilisation
# ip = dns_lookup('www.google.com')
# print(f"IP: {ip}")


# === Ping (ICMP) ===

def ping(host, timeout=3):
    """Ping ICMP (nécessite privilèges root)"""
    import struct
    import time
    import select
    
    # Créer raw socket ICMP
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, 
                            socket.IPPROTO_ICMP)
    except PermissionError:
        print("Erreur: nécessite privilèges root/admin")
        return None
    
    sock.settimeout(timeout)
    
    # Calculer checksum
    def checksum(data):
        s = 0
        for i in range(0, len(data), 2):
            if i + 1 < len(data):
                s += (data[i] << 8) + data[i+1]
            else:
                s += data[i]
        s = (s >> 16) + (s & 0xFFFF)
        s = ~s & 0xFFFF
        return s
    
    # Construire paquet ICMP Echo Request
    icmp_type = 8  # Echo Request
    icmp_code = 0
    icmp_checksum = 0
    icmp_id = os.getpid() & 0xFFFF
    icmp_seq = 1
    
    # Header sans checksum
    header = struct.pack('!BBHHH', icmp_type, icmp_code, icmp_checksum,
                        icmp_id, icmp_seq)
    
    # Data
    data = b'Python Ping!'
    
    # Calculer checksum
    icmp_checksum = checksum(header + data)
    
    # Reconstruire avec checksum
    header = struct.pack('!BBHHH', icmp_type, icmp_code, icmp_checksum,
                        icmp_id, icmp_seq)
    
    packet = header + data
    
    # Résoudre hostname
    try:
        dest_addr = socket.gethostbyname(host)
    except socket.gaierror:
        print(f"Impossible de résoudre {host}")
        return None
    
    # Envoyer
    send_time = time.time()
    sock.sendto(packet, (dest_addr, 0))
    
    # Recevoir
    try:
        ready = select.select([sock], [], [], timeout)
        if not ready[0]:
            print(f"Timeout de {host}")
            return None
        
        recv_packet, addr = sock.recvfrom(1024)
        recv_time = time.time()
        
        # Extraire réponse ICMP (skip IP header)
        icmp_header = recv_packet[20:28]
        recv_type, recv_code, recv_checksum, recv_id, recv_seq = \
            struct.unpack('!BBHHH', icmp_header)
        
        if recv_type == 0 and recv_id == icmp_id:  # Echo Reply
            rtt = (recv_time - send_time) * 1000  # ms
            print(f"Reply from {dest_addr}: time={rtt:.2f}ms")
            return rtt
        
    except socket.timeout:
        print(f"Timeout de {host}")
    except Exception as e:
        print(f"Erreur: {e}")
    finally:
        sock.close()
    
    return None


# === Speed test simple ===

def speed_test_server(host='0.0.0.0', port=5001):
    """Serveur pour test de vitesse"""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(1)
    
    print(f"[*] Serveur speed test sur {host}:{port}")
    
    while True:
        client, addr = server.accept()
        print(f"[+] Test depuis {addr}")
        
        # Recevoir données
        total = 0
        start = time.time()
        
        while True:
            data = client.recv(65536)
            if not data:
                break
            total += len(data)
        
        elapsed = time.time() - start
        speed = (total / elapsed) / 1024 / 1024  # MB/s
        
        print(f"[*] Reçu {total / 1024 / 1024:.2f} MB en {elapsed:.2f}s")
        print(f"[*] Vitesse: {speed:.2f} MB/s")
        
        client.close()


def speed_test_client(host='127.0.0.1', port=5001, size_mb=100):
    """Client test de vitesse"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))
    
    data = b'X' * (1024 * 1024)  # 1 MB
    total = size_mb * 1024 * 1024
    sent = 0
    
    print(f"Envoi de {size_mb} MB à {host}:{port}")
    
    start = time.time()
    
    while sent < total:
        sock.send(data)
        sent += len(data)
    
    elapsed = time.time() - start
    speed = (sent / elapsed) / 1024 / 1024
    
    print(f"Envoyé {sent / 1024 / 1024:.2f} MB en {elapsed:.2f}s")
    print(f"Vitesse: {speed:.2f} MB/s")
    
    sock.close()


# === Wake-on-LAN ===

def wake_on_lan(mac_address):
    """Envoie magic packet Wake-on-LAN"""
    # Retirer séparateurs
    mac = mac_address.replace(':', '').replace('-', '')
    
    if len(mac) != 12:
        raise ValueError("Adresse MAC invalide")
    
    # Convertir en bytes
    mac_bytes = bytes.fromhex(mac)
    
    # Construire magic packet
    # 6 bytes FF + 16 répétitions de l'adresse MAC
    magic_packet = b'\xFF' * 6 + mac_bytes * 16
    
    # Envoyer en broadcast
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    
    sock.sendto(magic_packet, ('255.255.255.255', 9))
    sock.close()
    
    print(f"Magic packet envoyé à {mac_address}")


# Utilisation
# wake_on_lan('00:11:22:33:44:55')


# === HTTP/WebSocket upgrade ===

def websocket_handshake(client_socket):
    """Handshake WebSocket basique"""
    import base64
    import hashlib
    
    # Recevoir requête HTTP
    request = client_socket.recv(4096).decode()
    
    # Extraire Sec-WebSocket-Key
    key = None
    for line in request.split('\r\n'):
        if line.startswith('Sec-WebSocket-Key:'):
            key = line.split(':', 1)[1].strip()
            break
    
    if not key:
        return False
    
    # Calculer Sec-WebSocket-Accept
    magic = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
    accept_key = base64.b64encode(
        hashlib.sha1((key + magic).encode()).digest()
    ).decode()
    
    # Réponse handshake
    response = (
        'HTTP/1.1 101 Switching Protocols\r\n'
        'Upgrade: websocket\r\n'
        'Connection: Upgrade\r\n'
        f'Sec-WebSocket-Accept: {accept_key}\r\n'
        '\r\n'
    )
    
    client_socket.sendall(response.encode())
    return True


# === Serveur echo avec protocole personnalisé ===

class CustomProtocol:
    """Protocole custom avec header length + type"""
    
    TYPE_MESSAGE = 0x01
    TYPE_FILE = 0x02
    TYPE_COMMAND = 0x03
    
    @staticmethod
    def pack(msg_type, data):
        """Empaqueter message"""
        if isinstance(data, str):
            data = data.encode()
        
        length = len(data)
        # Format: type (1 byte) + length (4 bytes) + data
        return struct.pack('!BI', msg_type, length) + data
    
    @staticmethod
    def unpack(sock):
        """Dépaqueter message"""
        # Lire header (5 bytes)
        header = CustomProtocol._recv_exactly(sock, 5)
        msg_type, length = struct.unpack('!BI', header)
        
        # Lire data
        data = CustomProtocol._recv_exactly(sock, length)
        
        return msg_type, data
    
    @staticmethod
    def _recv_exactly(sock, n):
        data = b''
        while len(data) < n:
            chunk = sock.recv(n - len(data))
            if not chunk:
                raise ConnectionError("Connexion fermée")
            data += chunk
        return data


def protocol_server(host='0.0.0.0', port=8080):
    """Serveur utilisant protocole custom"""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    
    print(f"[*] Serveur protocole custom sur {host}:{port}")
    
    try:
        while True:
            client, addr = server.accept()
            print(f"[+] Connexion: {addr}")
            
            try:
                while True:
                    msg_type, data = CustomProtocol.unpack(client)
                    
                    if msg_type == CustomProtocol.TYPE_MESSAGE:
                        print(f"Message: {data.decode()}")
                        response = CustomProtocol.pack(
                            CustomProtocol.TYPE_MESSAGE,
                            f"Echo: {data.decode()}"
                        )
                        client.sendall(response)
                    
                    elif msg_type == CustomProtocol.TYPE_COMMAND:
                        cmd = data.decode()
                        print(f"Commande: {cmd}")
                        
                        if cmd == 'QUIT':
                            break
                    
            except Exception as e:
                print(f"Erreur: {e}")
            finally:
                client.close()
                
    except KeyboardInterrupt:
        print("\n[*] Arrêt serveur")
    finally:
        server.close()


[OK] COMMANDES UTILES SYSTÈME

# === Commandes shell utiles ===

# Vérifier port ouvert (Linux)
# netstat -tuln | grep :8080
# ss -tuln | grep :8080
# lsof -i :8080

# Tuer processus sur port
# Linux: fuser -k 8080/tcp
# MacOS: lsof -ti:8080 | xargs kill

# Tester connexion TCP
# nc -zv 127.0.0.1 8080
# telnet 127.0.0.1 8080

# Tester UDP
# nc -zvu 127.0.0.1 9090

# Écouter sur port (netcat)
# nc -l 8080              # Linux
# nc -l -p 8080           # BSD/Mac

# Envoyer données
# echo "Hello" | nc 127.0.0.1 8080

# Proxy TCP avec netcat
# mkfifo backpipe
# nc -l 8080 < backpipe | nc example.com 80 | tee backpipe

# Capture trafic
# tcpdump -i eth0 port 8080
# tcpdump -i any -A port 8080  # Afficher ASCII

# Voir connexions actives
# netstat -an | grep ESTABLISHED
# ss -t

# Vérifier firewall (Linux)
# iptables -L -n
# ufw status


[OK] SNIPPETS CONFIGURATION

# === systemd service (Linux) ===

# /etc/systemd/system/myserver.service
"""
[Unit]
Description=My Socket Server
After=network.target

[Service]
Type=simple
User=myuser
WorkingDirectory=/opt/myserver
ExecStart=/usr/bin/python3 /opt/myserver/server.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
"""

# Commandes:
# sudo systemctl daemon-reload
# sudo systemctl start myserver
# sudo systemctl enable myserver
# sudo systemctl status myserver


# === Docker ===

# Dockerfile
"""
FROM python:3.11-slim

WORKDIR /app
COPY server.py .

EXPOSE 8080

CMD ["python", "server.py"]
"""

# docker build -t myserver .
# docker run -p 8080:8080 myserver


# === Nginx reverse proxy ===

"""
server {
    listen 80;
    server_name example.com;

    location / {
        proxy_pass http://127.0.0.1:8080;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        
        # WebSocket support
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}
"""


[OK] MÉMO RAPIDE

# Créer socket TCP
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Serveur: bind + listen + accept
sock.bind(('0.0.0.0', 8080))
sock.listen(5)
client, addr = sock.accept()

# Client: connect
sock.connect(('127.0.0.1', 8080))

# Envoyer/recevoir
sock.send(data)           # Peut envoyer partiellement
sock.sendall(data)        # Envoie tout
data = sock.recv(4096)    # Recevoir max 4096 bytes

# UDP
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.sendto(data, (host, port))
data, addr = sock.recvfrom(4096)

# Options
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(5.0)
sock.setblocking(False)

# Fermer
sock.close()

# Context manager
with socket.socket() as sock:
    # Usage...
    pass  # Auto-fermé


[OK] TROUBLESHOOTING

# Problème: "Address already in use"
# Solution: SO_REUSEADDR
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# Problème: "Connection refused"
# - Vérifier que serveur est démarré
# - Vérifier firewall
# - Vérifier bon port/IP

# Problème: "Connection reset by peer"
# - Pair a fermé connexion brutalement
# - Gérer exception et reconnecter

# Problème: Blocage infini
# - Ajouter timeout
sock.settimeout(5.0)

# Problème: Données partielles
# - Utiliser sendall() au lieu de send()
# - Boucler recv() jusqu'à avoir tout

# Problème: "Permission denied" (port < 1024)
# - Utiliser sudo (Linux)
# - Utiliser port > 1024

# Problème: "Network unreachable"
# - Vérifier connectivité réseau
# - Vérifier configuration IP/route

# Problème: "Broken pipe"
# - Socket fermé côté distant
# - Ignorer signal SIGPIPE (Unix)

# Problème: Fuite mémoire
# - Toujours fermer sockets
# - Utiliser context managers

# Problème: Performance faible
# - Augmenter taille buffers
# - Désactiver Nagle (TCP_NODELAY)
# - Utiliser threading/asyncio