Set-OPS-Public/scripts/tests/test_sonde_tcp.py

75 lines
2.1 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
"""La sonde TCP distingue une destination qui PARLE, qui REFUSE en fermant, et le SILENCE.
POURQUOI (2026-09-29). Le serveur par defaut de l'edge ferme sans rien dire un nom inconnu
(`return 444`). La sonde de `frontiere-mesurer` y lisait « rien n'est livre » et accusait
l'hote de refuser ce que la frontiere autorise. Or une fermeture propre (FIN) apres la
requete est une REPONSE de la destination ; le silence, lui — ce que rend une frontiere
qui repond a la poignee sans relayer — doit rester AMBIGU. Confondre les deux dans l'autre
sens rouvrirait le piege du `synproxy` (voir l'en-tete de `sonde_tcp.py`).
"""
from __future__ import annotations
import socket
import sys
import threading
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import sonde_tcp # noqa: E402
sonde_tcp.DELAI_CONNEXION = 1.0
sonde_tcp.DELAI_LECTURE = 0.5
def serveur(comportement) -> int:
srv = socket.socket()
srv.bind(("127.0.0.1", 0))
srv.listen(4)
def boucle():
while True:
try:
c, _ = srv.accept()
except OSError:
return
threading.Thread(target=comportement, args=(c,), daemon=True).start()
threading.Thread(target=boucle, daemon=True).start()
return srv.getsockname()[1]
def parle(c):
c.sendall(b"SSH-2.0-test\r\n")
c.close()
def lit_puis_ferme(c):
c.settimeout(2)
try:
c.recv(256)
except OSError:
pass
c.close()
def muet(c):
threading.Event().wait(5) # accepte, ne dit rien, ne ferme pas
def main() -> int:
attendu = {"parle": "LIVRE", "lit_puis_ferme": "LIVRE", "muet": "AMBIGU"}
echecs = 0
for nom, f in (("parle", parle), ("lit_puis_ferme", lit_puis_ferme), ("muet", muet)):
v = sonde_tcp._sonder("127.0.0.1", serveur(f))
if v["verdict"] != attendu[nom]:
print(f"ECHEC {nom} : attendu {attendu[nom]}, obtenu {v}")
echecs += 1
if echecs:
return 1
print("OK — parle : LIVRE ; lit puis ferme : LIVRE ; muet : AMBIGU.")
return 0
if __name__ == "__main__":
sys.exit(main())