Paillasse/portail/tests/Reels/parcours-lab.py

#!/usr/bin/env python3
"""Parcours réel de l'étape 11, joué comme deux navigateurs (prof et élève) :

1. l'élève rejoint la séance, le prof l'admet : chronomètre jusqu'à ce que
   l'invite de connexion de la VM s'affiche dans la console de l'élève ;
2. l'élève configure le lab 08 par les consoles série, sans activer le
   routage : « Vérifier mon travail » doit donner « échoué » avec un message ;
3. il active le routage : la vérification doit donner « réussi ».

Usage (sur le serveur, après preparer-parcours.php) :
  sudo -u www-data python3 tests/Reels/parcours-lab.py
"""
import json
import re
import threading
import time

import requests
import websocket

BASE = "http://127.0.0.1"
ids = json.load(open("/tmp/parcours-identifiants.json"))
seance = ids["seance_id"]


def jeton_csrf(session, chemin):
    page = session.get(BASE + chemin).text
    return re.search(r'name="_token" value="([^"]+)"', page).group(1)


def connecter(chemin_formulaire, champs):
    s = requests.Session()
    champs["_token"] = jeton_csrf(s, "/connexion")
    r = s.post(BASE + chemin_formulaire, data=champs, allow_redirects=False)
    assert r.status_code == 302, f"connexion refusée ({r.status_code})"
    return s


eleve = connecter("/connexion/eleve", {"identifiant": ids["eleve_identifiant"], "code": ids["eleve_code"]})
prof = connecter("/connexion/prof", {"email": ids["prof_email"], "password": ids["prof_mot_de_passe"]})

# L'élève rejoint : sa page d'attente interroge l'état chaque seconde.
eleve.post(f"{BASE}/seances/{seance}/rejoindre", data={"_token": jeton_csrf(eleve, "/eleve")})
page_prof = prof.get(f"{BASE}/seances/{seance}/attente").text
url_admettre = re.search(r'action="(http[^"]+/admettre/\d+)"', page_prof).group(1).replace("http://localhost", BASE)
url_admettre = re.sub(r"^https?://[^/]+", BASE, url_admettre)

etat_change = threading.Event()


def interroger():
    while not etat_change.is_set():
        if eleve.get(f"{BASE}/seances/{seance}/etat-eleve").json()["etat"] != "en_attente":
            etat_change.set()
            return
        time.sleep(1)


threading.Thread(target=interroger, daemon=True).start()
time.sleep(1.5)

# --- 1. Chronomètre : clic « Admettre » → invite de connexion visible ---
t0 = time.time()
prof.post(url_admettre, data={"_token": re.search(r'name="_token" value="([^"]+)"', page_prof).group(1)})
etat_change.wait(60)

page = eleve.get(f"{BASE}/seances/{seance}/attente-eleve").text
csrf = re.search(r"'X-CSRF-TOKEN': '([^']+)'", page).group(1)
consoles = {}
for url, nom in zip(re.findall(r"fetch\('(http[^']+/console)'", page), re.findall(r'data-onglet="\d+"[^>]*>([^<]+)<', page)):
    url = re.sub(r"^https?://[^/]+", BASE, url)
    jeton = eleve.post(url, headers={"X-CSRF-TOKEN": csrf}).json()["jeton"]
    consoles[nom] = websocket.create_connection(f"ws://127.0.0.1/console/{jeton}", timeout=10)


def lire_jusqua(ws, motif, delai=15):
    fin, sortie = time.time() + delai, b""
    while time.time() < fin:
        ws.settimeout(max(0.1, fin - time.time()))
        try:
            morceau = ws.recv()
        except websocket.WebSocketTimeoutException:
            break
        sortie += morceau if isinstance(morceau, bytes) else morceau.encode()
        if re.search(motif, sortie):
            return sortie
    raise TimeoutError(f"motif {motif!r} absent : {sortie[-200:]!r}")


premiere = next(iter(consoles.values()))
premiere.send("\r")  # comme l'option « reveil » de console-reelle.js
lire_jusqua(premiere, rb"login:")
duree = time.time() - t0
print(f"1. Admission → invite de connexion : {duree:.2f} s ({'OK' if duree < 5 else 'TROP LENT'}, critère < 5 s)")
print(f"   machines ouvertes : {', '.join(consoles)}")

# --- 2. L'élève configure son lab par les consoles série ---
def taper(ws, commande, attendre=rb"\$ $"):
    ws.send(commande + "\r")
    return lire_jusqua(ws, attendre)


for nom, ws in consoles.items():
    if ws is not premiere:
        ws.send("\r")
        lire_jusqua(ws, rb"login:")
    taper(ws, "eleve", rb"[Pp]assword:")
    taper(ws, "eleve")
    if b"password for" in taper(ws, "sudo -v", rb"(\$ $|password for eleve:)"):
        taper(ws, "eleve")

reglages = {
    "pc-a": ["sudo ip addr add 192.168.1.10/24 dev ens3", "sudo ip link set ens3 up", "sudo ip route add default via 192.168.1.1"],
    "routeur": ["sudo ip addr add 192.168.1.1/24 dev ens3", "sudo ip addr add 192.168.2.1/24 dev ens4", "sudo ip link set ens3 up", "sudo ip link set ens4 up"],
    "pc-b": ["sudo ip addr add 192.168.2.10/24 dev ens3", "sudo ip link set ens3 up", "sudo ip route add default via 192.168.2.1"],
}
for nom, commandes in reglages.items():
    for commande in commandes:
        taper(consoles[nom], commande)
time.sleep(2)


def verifier():
    page = eleve.get(f"{BASE}/seances/{seance}/attente-eleve").text
    jeton = re.search(r'name="_token" value="([^"]+)"', page).group(1)
    session_id = re.search(r"/lab-sessions/(\d+)/corriger", page).group(1)
    t = time.time()
    page = eleve.post(f"{BASE}/lab-sessions/{session_id}/corriger", data={"_token": jeton}).text
    verdict = re.search(r'text-lg font-extrabold">\s*([^<]+?)\s*<span class="font-mono">· (\d+/\d+)', page)
    messages = re.findall(r'text-\[#A63D2A\]">([^<]+)<', page)
    return verdict.group(1), verdict.group(2), [m.replace("&#039;", "'") for m in messages], time.time() - t


resultat, note, messages, d = verifier()
print(f"2. Sans routage activé : « {resultat} » {note} (vérification en {d:.1f} s)")
for m in messages:
    print(f"   - {m}")

# --- 3. Le routage activé, le lab doit être réussi ---
taper(consoles["routeur"], "sudo sysctl -w net.ipv4.ip_forward=1")
time.sleep(1)
resultat, note, messages, d = verifier()
print(f"3. Routage activé : « {resultat} » {note} (vérification en {d:.1f} s)")
for m in messages:
    print(f"   - {m}")

for ws in consoles.values():
    ws.close()