Module L22 · Partie I · Robotique et systèmes embarqués

Faire parler les composants : bus, protocoles, réseaux, sécurité.

Un robot est un réseau : des capteurs sur I2C et SPI, des moteurs sur CAN, un microcontrôleur qui parle à un ordinateur par USB, des nœuds ROS sur Ethernet, une télémétrie en WiFi ou LoRa, un serveur dans le nuage. Chaque lien a son protocole, ses erreurs, ses latences et ses attaques possibles. Ce module descend du bit sur le fil jusqu’au chiffrement, en construisant les mécanismes essentiels : trames, sommes de contrôle, acquittements, files de messages, sockets, MQTT, TLS.

Durée : 3 séances · Prérequis : L07, L08, L18, séances 25-26. Objectifs : couches (physique, liaison, réseau, transport, application), UART/SPI/I2C/CAN et leurs compromis, conception d’un protocole binaire (trames, CRC, séquence, acquittement), sockets TCP/UDP, MQTT et publish/subscribe, ROS 2 sur DDS, latence et bande passante, cryptographie appliquée (hachage, HMAC, chiffrement, TLS, mise à jour signée).

Fiche de cours · Définitions

Communication, réseaux, cryptographie : définitions

Fiche de cours · Formules

Formules à connaître

Fiche de cours · Théorèmes et démonstrations

Démonstrations à savoir refaire (1/2)

Fiche de cours · Théorèmes et démonstrations

Démonstrations à savoir refaire (2/2)

Fiche de cours · Méthodes

Méthodes et pièges

Pièges : inventer sa propre cryptographie ; utiliser MD5/SHA-1 pour la sécurité ; réutiliser un nonce ou un IV ; comparer des MAC avec == (fuite temporelle : comparaison en temps constant) ; CAN sans authentification exposé à l’extérieur ; supposer qu’UDP livre dans l’ordre ; oublier l’endianness dans les trames binaires ; CRC calculé sur les mauvais octets (inclure la longueur et le type).

Fiche de cours · Exercices corrigés

Exercices corrigés

01 / Bus embarqués

UART, SPI, I2C, CAN : quatre façons de mettre des bits sur un fil

01 / Bus embarqués

I2C et SPI vus du code : lire une IMU

// I2C sur ESP32 (Arduino) : MPU-6050 à l'adresse 0x68. Registre 0x3B = accéléro X (16 bits, big-endian)
#include <Wire.h>
void setup() {
  Wire.begin(21, 22, 400000);         // SDA, SCL, 400 kHz
  Wire.beginTransmission(0x68);
  Wire.write(0x6B); Wire.write(0);    // PWR_MGMT_1 = 0 : réveiller
  Wire.endTransmission();
}
void lire_accel(int16_t *ax, int16_t *ay, int16_t *az) {
  Wire.beginTransmission(0x68);
  Wire.write(0x3B);                   // pointer sur le premier registre
  Wire.endTransmission(false);        // START répété : garder le bus
  Wire.requestFrom(0x68, 6);          // 6 octets : X, Y, Z
  *ax = (Wire.read() << 8) | Wire.read();
  *ay = (Wire.read() << 8) | Wire.read();
  *az = (Wire.read() << 8) | Wire.read();
}
// Conversion : ±2 g pleine échelle sur 16 bits → 16384 LSB/g

// SPI (Pico, C SDK) : même capteur en SPI, 10× plus vite. Bit 7 de l'adresse = lecture.
spi_init(spi0, 10 * 1000 * 1000);
gpio_put(CS, 0);                              // sélectionner
uint8_t tx[7] = { 0x3B | 0x80 };              // lecture à partir de 0x3B, 6 octets suivent
uint8_t rx[7];
spi_write_read_blocking(spi0, tx, rx, 7);     // full-duplex : on envoie et reçoit en même temps
gpio_put(CS, 1);
int16_t ax = (rx[1] << 8) | rx[2];

Trois pièges universels : l’endianness (quel octet en premier ?), le signe (complément à deux), et l’échelle (LSB par unité physique, dans la fiche technique). struct en Python et les casts en C rendent tout cela explicite.

02 / Protocoles

Concevoir un protocole binaire : trame, longueur, CRC, séquence

02 / Protocoles

Fiabilité sur canal avec pertes : acquittements, fenêtre, temporisateurs (l’idée de TCP)

TCP fait exactement cela : numéros de séquence, acquittements cumulatifs, fenêtre glissante, temporisateur estimé à partir du RTT mesuré (Jacobson), et en plus un contrôle de congestion (réduire la fenêtre quand le réseau sature). UDP ne fait rien de tout cela : c’est à l’application de décider si elle veut de la fiabilité (ROS 2 sur DDS choisit par topic : « fiable » pour les commandes, « best effort » pour les images).

03 / Réseaux

Les couches, et ce qu’un paquet traverse

$ ping robot.local                 # ICMP : le robot répond-il ? latence aller-retour (RTT)
$ ip addr / ifconfig               # mes adresses
$ ss -tulpn / netstat -an          # qui écoute sur quels ports
$ nc -l 5000                       # écouter en TCP sur 5000 (netcat) ;  nc robot.local 5000 pour se connecter
$ tcpdump -i wlan0 port 1883       # voir passer les paquets MQTT (Wireshark en graphique)
$ iperf3 -s / iperf3 -c robot      # mesurer la bande passante réelle du WiFi
$ mtr 8.8.8.8                      # la route et la latence de chaque saut

Ordres de grandeur : RTT sur un câble Ethernet local ≈ 0,2 ms ; WiFi ≈ 2-20 ms avec des pics à 200 ms ; 4G ≈ 50 ms ; à travers l’Atlantique ≈ 80 ms (la lumière ne va pas plus vite). Un contrôle de robot à 100 Hz ne se fait jamais à travers le WiFi : la boucle rapide reste sur le microcontrôleur (module L18), le réseau ne transporte que des objectifs.

03 / Réseaux

Sockets : TCP et UDP en Python (à exécuter sur PC), et une simulation dans la page

# serveur TCP : reçoit des commandes ligne par ligne
import socket
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("0.0.0.0", 5000)); srv.listen()
while True:
    conn, addr = srv.accept()                 # bloque jusqu'à un client
    with conn:
        tampon = b""
        while chunk := conn.recv(1024):        # TCP est un FLUX : recv peut rendre une demi-ligne ou trois
            tampon += chunk
            while b"\n" in tampon:
                ligne, tampon = tampon.split(b"\n", 1)
                conn.sendall(b"OK " + ligne + b"\n")

# client
c = socket.create_connection(("robot.local", 5000), timeout=2)
c.sendall(b"avancer 0.3\n"); print(c.recv(1024))

# UDP : télémétrie à 50 Hz, un datagramme par mesure, sans connexion
import socket, struct, time
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
while True:
    s.sendto(struct.pack("<Ifff", int(time.time() * 1000), x, y, th), ("192.168.1.10", 6000))
    time.sleep(0.02)
# récepteur
r = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); r.bind(("0.0.0.0", 6000)); r.settimeout(0.5)
try:
    data, addr = r.recvfrom(64); t, x, y, th = struct.unpack("<Ifff", data)
except socket.timeout:
    print("robot silencieux depuis 0,5 s")

03 / Réseaux

MQTT : le publish/subscribe de l’Internet des objets

# pip install paho-mqtt ; un broker : mosquitto (local) ou test.mosquitto.org (public, pour essayer)
import paho.mqtt.client as mqtt, json

def sur_message(client, userdata, msg):
    print(msg.topic, json.loads(msg.payload))

c = mqtt.Client(); c.on_message = sur_message
c.connect("localhost", 1883)
c.subscribe("robots/+/telemetrie")          # + : joker d'un niveau ; # : tout ce qui suit
c.loop_start()
c.publish("robots/r2/commande", json.dumps({"action": "avancer", "v": 0.3}), qos=1)

// ESP32 (PubSubClient) : publie la batterie toutes les 10 s, s'abonne aux commandes
client.setServer("192.168.1.10", 1883);
client.setCallback([](char *topic, byte *p, unsigned len) { /* parser JSON, agir */ });
client.connect("r2", "user", "motdepasse", "robots/r2/statut", 1, true, "hors-ligne");   // testament : publié si r2 disparaît
client.subscribe("robots/r2/commande");
client.publish("robots/r2/statut", "en-ligne", true);      // retenu : un nouvel abonné le reçoit immédiatement

04 / Sécurité

Cryptographie appliquée : ce que garantit chaque brique

Cours

Cours 1 — Signal, bruit, débit : ce que la physique impose à toute communication

Cours

Cours 2 — Exemple travaillé : le CRC, pourquoi il détecte ce qu’il détecte

Principe. Le message est vu comme un polynôme M(x) sur F₂ (coefficients 0/1, + = XOR). On choisit un polynôme générateur G(x) de degré r (CRC-16-CCITT : x¹⁶ + x¹² + x⁵ + 1 = 0x1021). Le CRC est le reste R(x) de M(x)·x^r divisé par G(x) ; on transmet M·x^r + R, qui est divisible par G. Le récepteur divise : reste nul ⇔ pas d’erreur détectée.

Une erreur E(x) est détectée ssi G ne divise pas E. D’où les propriétés : (1) toute erreur d’1 bit (E = x^k) est détectée car G a au moins deux termes ; (2) toute erreur de 2 bits (E = x^k(x^j + 1)) est détectée si G ne divise aucun x^j + 1 pour j < longueur — vrai pour un G bien choisi (période de G grande) ; (3) tout nombre impair d’erreurs est détecté si (x + 1) divise G ; (4) toute rafale de longueur ≤ r est détectée (E = x^k·B(x) avec deg B < r : G ne peut pas diviser B). Une rafale plus longue échappe avec probabilité 2^−r.

Cours

Cours 3 — Cryptographie : les primitives, leurs garanties, leurs usages corrects

TP guidé

TP — Un lien robot ↔ PC complet : bus, protocole, réseau, sécurité (carte + PC, 4 h)

• Analyseur logique. Branchez un analyseur (Saleae, ou clone 8 canaux + PulseView) sur l’UART et l’I2C de votre carte. Capturez une transaction I2C avec l’IMU : identifiez START, adresse + R/W, ACK, données, STOP ; mesurez la fréquence d’horloge. Sur l’UART : vérifiez 8N1 et la vitesse. Capture d’écran annotée.
• Protocole binaire en C et Python. Implémentez côté carte l’encodeur/décodeur (sync, longueur, séquence, type, CRC-16 par table) et côté PC en Python (struct, pyserial). Messages : télémétrie 50 Hz, commande, ping/pong. Testez : débranchement/rebranchement (resynchronisation), corruption (court-circuiter brièvement RX : CRC rejetés comptés), latence aller-retour (histogramme sur 1000 pings).
• Fiabilité. Ajoutez acquittement + retransmission avec numéro de séquence pour les commandes seulement (pas pour la télémétrie) ; simulez 20 % de pertes côté PC (ignorer aléatoirement) et vérifiez qu’aucune commande n’est perdue ni dupliquée.
• Réseau. Le PC (ou un Raspberry Pi sur le robot) expose la télémétrie : serveur UDP (diffusion) + serveur TCP de commandes (L08) + MQTT vers un broker mosquitto. Wireshark : capturez les trois flux, comparez les en-têtes et les tailles totales pour 1000 messages.
• Sécurité. Authentifiez les commandes par HMAC-SHA256 avec compteur anti-rejeu (sur la carte : implémentation SHA-256 légère, ou mbedTLS sur ESP32) ; mesurez le coût CPU. Passez MQTT en MQTTS (TLS, certificat auto-signé) ; vérifiez dans Wireshark que le contenu est illisible. Tentez un rejeu d’une commande capturée : elle doit être refusée.
• Livrable. Dépôt : firmware, bibliothèque Python du protocole (avec tests par propriétés sur l’encodeur/décodeur : décoder(encoder(m)) = m, resynchronisation après bruit aléatoire), captures d’analyseur et Wireshark, histogramme de latence, mesures de coût du HMAC.

Exercices

Exercices auto-corrigés — bits et trames

def cobs_encoder(data):
    out = bytearray(); bloc = bytearray()
    def flush(): out.append(len(bloc) + 1); out.extend(bloc); bloc.clear()
    for b in data:
        if b == 0: flush()
        else:
            bloc.append(b)
            if len(bloc) == 254: flush()
    flush(); return bytes(out)
def cobs_decoder(data):
    out = bytearray(); i = 0
    while i < len(data):
        code = data[i]; out.extend(data[i + 1:i + code]); i += code
        if code < 255 and i < len(data): out.append(0)
    return bytes(out)

def test_proprietes(n=300, p_corruption=0.01):
    charges = [bytes(random.randrange(256) for _ in range(random.randint(0, 40))) for _ in range(n)]
    flux = bytearray()
    for i, c in enumerate(charges):
        flux += encoder(i, 1, c)
        if random.random() < 0.3: flux += bytes(random.randrange(256) for _ in range(random.randint(1, 6)))
    for i in range(len(flux)):
        if random.random() < p_corruption: flux[i] ^= random.randrange(1, 256)
    d = Decodeur()
    for b in flux: d.alimenter(b)
    emises = set(charges)
    for seq, t, charge in d.trames: assert charge in emises, "trame acceptée jamais émise"
    return len(d.trames), d.erreurs

Exercices

Exercices auto-corrigés — réseau et cryptographie

class Estimateur:
    def __init__(self): self.srtt = None; self.var = 0.0
    def mesure(self, r):
        if self.srtt is None: self.srtt, self.var = r, r / 2
        else: self.var = 0.75 * self.var + 0.25 * abs(self.srtt - r); self.srtt = 0.875 * self.srtt + 0.125 * r
    def rto(self): return self.srtt + 4 * self.var

def dh_cle_publique(secret): return pow(g, secret, p)
def dh_secret_partage(mon_secret, sa_cle_publique): return pow(sa_cle_publique, mon_secret, p)
def signer(cle_auth, cle_publique): return hmac.new(cle_auth, str(cle_publique).encode(), hashlib.sha256).digest()
def verifier(cle_auth, cle_publique, tag): return hmac.compare_digest(signer(cle_auth, cle_publique), tag)

05 / Défis

Défi ★ — Un protocole complet PC ↔ carte

static uint16_t table[256];
void crc_init(void) { for (int i = 0; i < 256; i++) { uint16_t c = i << 8; for (int b = 0; b < 8; b++) c = (c & 0x8000) ? (c << 1) ^ 0x1021 : c << 1; table[i] = c; } }
uint16_t crc16(const uint8_t *d, size_t n) { uint16_t c = 0xFFFF; while (n--) c = (c << 8) ^ table[(c >> 8) ^ *d++]; return c; }

05 / Défis

Défi ★★ — Flotte de robots par MQTT et tableau de bord

05 / Défis

Défi ★★★ — Esprit prépa : codes correcteurs et échange de clés

G = np.array([[1,0,0,0,1,1,0],[0,1,0,0,1,0,1],[0,0,1,0,0,1,1],[0,0,0,1,1,1,1]])     # génératrice (systématique)
Hm = np.array([[1,1,0,1,1,0,0],[1,0,1,1,0,1,0],[0,1,1,1,0,0,1]])                  # de contrôle : H Gᵀ = 0 mod 2
def encoder(d): return d @ G % 2
def decoder(r):
    s = Hm @ r % 2                                            # syndrome : colonne de H où l'erreur est
    if s.any():
        col = [tuple(Hm[:, j]) for j in range(7)].index(tuple(s)); r = r.copy(); r[col] ^= 1
    return r[:4]
d = np.array([1, 0, 1, 1]); c = encoder(d); c[5] ^= 1; print(decoder(c), d)

Référence

Les mots à retenir

Pour continuer

Fin de la partie I : votre robot est un système complet

Partie J : vers la recherche. Module suivant : l’informatique théorique — calculabilité, machines de Turing, complexité P/NP, automates et langages : ce que les ordinateurs ne peuvent pas faire, et pourquoi.

À faire chez soi

• Installer Wireshark et observer une conversation MQTT, puis HTTPS (que voit-on encore ?).
• Lire les chapitres 1-3 de Kurose & Ross.
• Faire les exercices « Cryptopals » sets 1-2 (cryptographie par la pratique).

← L21SommaireL23 : Informatique théorique →