Automatisation Réseau et SOC

Une API REST expose des ressources via des URL, manipulées avec les verbes HTTP : GET (lire), POST (créer), PUT/PATCH (modifier), DELETE (supprimer). Le serveur répond avec un code de statut : 2xx succès, 4xx erreur client, 5xx erreur serveur :

# Codes a connaitre par coeur
# 200 OK | 201 Created | 204 No Content
# 400 Bad Request | 401 Unauthorized | 403 Forbidden | 404 Not Found
# 429 Too Many Requests | 500 Internal Server Error

Les API réseau (NetBox pour l'inventaire, contrôleurs SDN, plateformes SOC) renvoient du JSON et exigent un token d'authentification passé dans l'en-tête Authorization. Les volumes importants se parcourent par pagination (paramètres limit / offset ou page suivante).

Exercice 9.1

La bibliothèque requests (pip install requests) simplifie le HTTP. requests.get() envoie la requête, .status_code donne le code, .json() convertit la réponse JSON en dictionnaire :

import requests

url = "https://api.example.com/devices"
headers = {"Authorization": "Token MON_TOKEN_SECRET"}
try:
    r = requests.get(url, headers=headers, timeout=10)
    r.raise_for_status()          # leve une erreur si code 4xx/5xx
    for eq in r.json()["results"]:
        print(eq["hostname"], "-", eq["ip"])
except requests.RequestException as e:
    print(f"Erreur API : {e}")

Imposez toujours un timeout (sinon le script peut attendre indéfiniment) et appelez raise_for_status() pour transformer les erreurs HTTP en exceptions traitables. Les paramètres d'URL passent par params={"limit": 50}.

Exercice 9.2

requests.post() envoie des données avec le paramètre json= (sérialisation automatique + bon Content-Type). On vérifie le code 201 et on exploite la réponse pour récupérer l'objet créé :

import requests

payload = {"hostname": "SW3", "ip": "10.0.0.5", "role": "access"}
r = requests.post("https://api.example.com/devices",
                  json=payload, headers={"Authorization": "Token X"},
                  timeout=10)
if r.status_code == 201:
    print("Cree :", r.json()["hostname"])
elif r.status_code == 401:
    print("Token invalide : verifiez vos identifiants")
else:
    print(f"Echec ({r.status_code}) : {r.text}")

Ne stockez jamais un token en dur dans le code : lisez-le depuis une variable d'environnement (os.environ["API_TOKEN"]) ou un fichier de configuration exclu du versionnement.

Exercice 9.3

Netmiko (pip install netmiko) pilote les CLI réseau en SSH. On décrit l'équipement par un dictionnaire (type, hôte, identifiants) et on utilise ConnectHandler en gestionnaire de contexte pour fermer proprement la session :

from netmiko import ConnectHandler

r1 = {
    "device_type": "cisco_ios",
    "host": "10.0.0.1",
    "username": "admin",
    "password": "mot_de_passe_lab",
}
with ConnectHandler(**r1) as ssh:
    sortie = ssh.send_command("show ip interface brief")
    print(sortie)

Types courants : cisco_ios, cisco_nxos, arista_eos, juniper_junos, linux. Stockez les identifiants hors du code (variables d'environnement ou coffre) et testez d'abord en lecture seule (show) avant toute écriture.

Exercice 10.1

send_config_set() envoie une liste de commandes en mode configuration. En bouclant sur un inventaire CSV, on déploie la même politique (VLAN, banner, NTP) sur tout le parc en quelques secondes :

import csv
from netmiko import ConnectHandler

config_vlan = ["vlan 30", "name INVITES", "exit",
               "interface range Fa0/5-8",
               "switchport mode access",
               "switchport access vlan 30"]

with open("parc.csv", encoding="utf-8", newline="") as f:
    for eq in csv.DictReader(f):
        dev = {"device_type": "cisco_ios", "host": eq["ip"],
               "username": "admin", "password": "mot_de_passe_lab"}
        with ConnectHandler(**dev) as ssh:
            print(ssh.send_config_set(config_vlan))
            ssh.save_config()   # = write memory

Bonnes pratiques : déployez d'abord sur un équipement pilote, journalisez chaque résultat avec logging, et prévoyez un plan de retour arrière (sauvegarde préalable, leçon 10.3).

Exercice 10.2

Avant tout changement, sauvegardez la running-config avec un nom horodaté. Ensuite, parsez les sorties show avec des regex pour en extraire des données structurées (interfaces down, VLAN, routes) :

import re
from datetime import datetime
from netmiko import ConnectHandler

with ConnectHandler(**r1) as ssh:
    conf = ssh.send_command("show running-config")
    nom = f"sauvegardes/R1_{datetime.now():%Y%m%d_%H%M}.cfg"
    with open(nom, "w", encoding="utf-8") as f:
        f.write(conf)

    brief = ssh.send_command("show ip interface brief")
    down = re.findall(r"^(\S+)\s+\S+\s+\S+\s+\S+\s+down",
                      brief, re.MULTILINE)
    print("Interfaces down :", down)

Planifiez ce script (tâche cron ou planificateur Windows) pour une sauvegarde quotidienne, et versionnez les sauvegardes avec Git pour tracer chaque changement de configuration.

Exercice 10.3

Les logs (pare-feu, proxy, IDS) sont des lignes semi-structurées. La recette SOC : lire ligne par ligne (jamais tout en mémoire pour les gros fichiers), extraire les champs par regex, filtrer les événements intéressants :

import re

motif = re.compile(r"SRC=(?P<src>\d+\.\d+\.\d+\.\d+).*DPT=(?P<port>\d+)")
evenements = []
with open("firewall.log", encoding="utf-8") as f:
    for ligne in f:
        if "DROP" in ligne:
            m = motif.search(ligne)
            if m:
                evenements.append(m.groupdict())
print(f"{len(evenements)} paquets bloques analyses")

m.groupdict() retourne directement un dictionnaire des groupes nommés — parfait pour alimenter un CSV ou une API. Filtrez tôt (ici « DROP ») pour ne parser que l'utile : c'est 10 fois plus rapide sur des millions de lignes.

Exercice 11.1

collections.Counter compte les occurrences en une ligne : top talkers, ports les plus attaqués, codes HTTP anormaux. On croise ensuite avec une liste d'IOC (indicateurs de compromission : IP, domaines, hash malveillants) via des sets :

from collections import Counter

sources = [e["src"] for e in evenements]
top = Counter(sources).most_common(5)
print("Top sources bloquees :", top)

ioc = {"5.6.7.8", "9.9.9.9"}   # issus du flux de threat intel
detectees = set(sources) & ioc
for ip in detectees:
    print(f"CRITIQUE : IOC connu detecte : {ip}")

Ajoutez un seuil (ex. plus de 10 tentatives) pour distinguer le bruit de fond des vraies attaques, et journalisez chaque détection avec horodatage pour la traçabilité de l'incident.

Exercice 11.2

Un rapport SOC horodaté, chiffré et actionnable : période couverte, volumes, top 5, IOC détectés, recommandations. On le génère en texte et en CSV pour l'outil de ticketing :

import csv
from datetime import datetime

def generer_rapport(evenements, detectees, seuil=10):
    lignes = [f"Rapport SOC - {datetime.now():%Y-%m-%d %H:%M}",
              f"Evenements analyses : {len(evenements)}",
              f"IOC detectes : {len(detectees)}"]
    for ip in sorted(detectees):
        lignes.append(f"  - {ip} : blocage immediat recommande")
    texte = "\n".join(lignes)
    with open("rapport_soc.txt", "w", encoding="utf-8") as f:
        f.write(texte)
    return texte

Automatisez l'envoi (e-mail, webhook, API du SIEM) et archivez chaque rapport : un SOC doit pouvoir prouver ce qui a été détecté, quand, et quelle action a été recommandée.

Exercice 11.3

Le module socket ouvre des connexions TCP avec un timeout court : si la connexion réussit, le port est ouvert. On boucle sur une liste de ports courants et on structure le résultat :

import socket

def scanner(hote, ports, timeout=1.0):
    """Retourne le dict {port: True/False} des ports ouverts."""
    resultats = {}
    for p in ports:
        s = socket.socket()
        s.settimeout(timeout)
        try:
            s.connect((hote, p))
            resultats[p] = True
        except OSError:
            resultats[p] = False
        finally:
            s.close()
    return resultats

print(scanner("127.0.0.1", [22, 80, 443]))

Limites à connaître : scannez uniquement vos machines de lab, utilisez des timeouts courts, parallélisez avec concurrent.futures pour les gros périmètres, et journalisez chaque résultat.

Exercice 12.1

Le projet final combine les modules 5, 8 et 11 : lire un CSV d'événements, filtrer par regex, compter avec Counter, croiser des IOC, écrire un rapport texte + CSV. Chaque étape est une fonction testable :

import csv
from collections import Counter

def charger(csv_path):
    with open(csv_path, encoding="utf-8", newline="") as f:
        return list(csv.DictReader(f))

def top_sources(lignes, n=5):
    return Counter(l["src"] for l in lignes).most_common(n)

def exporter_csv(top, out="top_sources.csv"):
    with open(out, "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["ip", "tentatives"])
        w.writerows(top)

lignes = charger("evenements.csv")
top = top_sources(lignes)
exporter_csv(top)
print("Top :", top)

Présentez ce projet avec un README (objectif, installation, exemple d'exécution) : c'est exactement ce qu'un recruteur ou un jury veut voir.

Exercice 12.2

• Nommage : fonctions en verbes (extraire_ip), constantes en MAJUSCULES (SEUIL = 10).
• Fonctions courtes à tâche unique, avec docstring d'une ligne.
• Gestion d'erreurs : try/except ciblés + messages de log explicites.
• Secrets : jamais en dur — variables d'environnement ou fichiers exclus du Git.
• Tests : vérifiez chaque fonction avec 2-3 exemples avant d'assembler le script.
• Style : python -m pyflakes ou flake8 pour traquer les erreurs et le style.

Checklist avant mise en production : testé en lab, sauvegardes prévues, logs horodatés, plan de retour arrière, revue par un pair.

Exercice 12.3 (projet final)