URL: https://linuxfr.org/users/jul/journaux/construire-un-c-c-basique-en-python-forth-au-dessus-de-mqtt Title: Construire un C&C basique en python/FORTH au dessus de MQTT Authors: Jul Date: 2026年02月02日T14:56:58+01:00 License: CC By-SA Tags: python et forth Score: 8 # Introduction Qui n'a pas rêvé de faire sa console de pirates qui contrôle ses agents au doigt et à l'œil traditionnellement sur IRC ? C'est le principe d'un [Control & Command](https://en.wikipedia.org/wiki/Botnet#Command_and_control) parfois appelé C&C pour les botnets. Mais ici, on en fait un éducationnel. Il s'agit de piloter des agents à distance en leur envoyant des commandes sur un BUS qui résulte dans des actions prédéfinies comme : - arrêtes toi, - reprends, - dis si tu es présent et en vie ... Ci-suit un petit exemple en python de l'implémentation d'une telle logique en moins de 100 lignes de codes # Ingrédients Cette recette nécessite : python, et en dépendances : *paho-mqtt*, *confined*, ainsi qu'un serveur [MQTT](https://mosquitto.org/) (mosquitto avec ses utilitaires en ligne de commande) correctement configurés. # La pièce de résistance Pour tout process que l'on veut piloter on va écrire du code python comme : ```python import paho.mqtt.client as mqtt from time import time, sleep from confined import parse, Value, pop from subprocess import Popen,PIPE import pathlib import os import socket stack = [] client_id = socket.gethostname() show_must_go = False def on_connect(client, userdata, flags, reason_code, properties): print(f"Connected with result code {reason_code}") client.subscribe(f"BUS/{client_id}") client.subscribe(f"BUS") def on_lun(*a, **kw): kw["ctx"]["state"]="RAZ" def on_set_time_slice(*a, **kw): kw["ctx"]["time_slice"]=stack.pop().float def on_ping(*a, **kw): global client_id client = kw["ctx"]["client"] client.publish("RES", f"'{client_id}':PONG") def on_sel(stack, **kw): global client_id, show_must_go if stack.pop().str == client_id: show_must_go = True def on_unsel(stack, **kw): global client_id, show_must_go if stack.pop().str == client_id: show_must_go = False def on_test(stack, **kw): print("Yo") ctx = dict( cap=["www", "forth" ], time_slice=10, dispatch=dict( lun=on_lun, ping=on_ping, sel=on_sel, unsel=on_unsel, tsset=on_set_time_slice, _TEST=on_test, ), ) def on_message(client, userdata, msg): global stack, ctx ctx["client"] = client client.publish( "RES", str(parse( ctx, msg.payload.decode(), data=stack, ) ) ) mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id) mqttc.on_connect = on_connect mqttc.on_message = on_message mqttc.username = "pub" mqttc.password= "pub" mqttc.tls_set( keyfile="./cfg/pub.key", certfile="./cfg/pub.crt", ca_certs="./cfg/RootCA.crt", tls_version=2 ) mqttc.connect("badass.home", 8883, 60) mqttc.loop_start() os.chdir(os.path.dirname(__file__)) plugins = pathlib.Path("../plugin") while True: start = time() if show_must_go: for p in plugins.glob("*_enabled"): with Popen([ p, ], stdout=PIPE, stdin=PIPE, stderr=PIPE, bufsize=0,) as writer: while res := writer.stdout.read(): writer.stdout.flush() for msg in res.split(): mqttc.publish(f"DATA/{client_id}", msg.decode()) mqttc.publish("DATA/", f"{client_id}:core.processing_time:{time()-start}:GAUGE") if time() -start < ctx["time_slice"]: sleep(ctx["time_slice"] - (time() - start)) ``` Le client est en mode parano : TLS activé avec login/pass enforcé coté serveur par MQTT. On va définir 2 commandes : pub.sh ```bash #!/usr/bin/env bash HERE=$( dirname 0ドル ) pushd $HERE mosquitto_pub --cert ../cfg/4711.crt --key ../cfg/4711.key --cafile ../cfg/RootCA.crt -h badass.home -t BUS -u 4711 -P 4711 -m "1ドル" popd ``` qui publie en MQTT sur un canal BUS et listen.sh ```bash #!/usr/bin/env bash HERE=$( dirname 0ドル ) pushd $HERE mosquitto_sub -h badass.home -t RES/# & mosquitto_sub -h badass.home -t DATA/# & popd ``` # La recette Pour pinger votre agent il vous suffit de faire *pub.sh PING* ce qui répond *'badass':PONG* Pour tester la fonction de test locale : *pub.sh _TEST* Normalement sur sortie standard du process python vous pouvez lire : 'Yo'. *pub.sh "'badass': SEL"* La sortie de LISTEN devrait montrer des lignes comme suit: ```erlang badass:cpu.load:1.39:GAUGE badass:stat.procs_running:1:GAUGE badass:stat.procs_blocked:1:GAUGE badass:stat.intr:235643164:DERIVE badass:stat.ctxt:474453646:DERIVE badass:stat.processes:237036:DERIVE badass:ps.processes:320:GAUGE badass:ps.uninterruptible:1:GAUGE badass:ps.runnable:1:GAUGE badass:ps.sleeping:256:GAUGE badass:ps.idle:62:GAUGE badass:ps.stopped:0:GAUGE badass:ps.paging:0:GAUGE badass:ps.dead:0:GAUGE badass:ps.zombie:0:GAUGE badass:files.opened:17184:GAUGE badass:core.processing_time:0.039983272552490234:GAUGE ``` Qui sont le résultat de l'enclenchement de la mesure qui se fait toutes les 10 secondes par défaut *pub.sh "20: TSSET"* Passe les mesures à 20 secondes d'écart. *pub.sh "'badass': UNSEL"* Arrête les mesures. # Diverses Questions Réponses ## Un forth dans la boucle était il nécessaire ? Non. Mais il se trouve que [j'en ai un en stock](https://github.com/jul/confined) et que je rêvais de l'utiliser pour faire des tâches innocentes (principalement du templating). Il est pas utilisé, mais disons que si vous faisiez : *pub.sh "2: 2: ADD"* en sortie vous auriez autant de *4:* qu'il y a d'agents connectés sur le BUS. Vous avez de base une calculette 4 opérations distribuées en prime. ## Un vrai command & control préférerait un protocole plus passe partout comme *https* pour être moins visible [Ça tombe bien paho-mqtt et mosquito permettent de passer en websocket](http://www.steves-internet-guide.com/mqtt-websockets/) :) ## Et si je voulais sécuriser ? [MQTT le permet](https://infosys.beckhoff.com/english.php?content=../content/1033/tc3_grundlagen/17769201803.html&id=) avec des ACL par sujets. ## Où est l'arborescente complète du projet il manque les plugins de mesure ? [Les plugins sont ici](https://github.com/jul/FAIM/tree/main/plugin) ## Mais c'est bête en fait... C'est pas faux, c'est pour ça que je sais pas quoi [faire avec ce « projet »](https://linuxfr.org/users/jul/journaux/j-ai-des-noeuds-au-cerveau-je-ne-sais-pas-comment-continuer) Trop petit pour être un projet, ultra dur à tester, mais assez rigolo pour être utile ## Des idées de futurs ? Un projet « Bus Of Things » (BOT) qui standardiserait les commandes envoyées et leurs API pour faire comme une sorte d'ansible. Une logique d'IPC/messaging générique pour des systèmes distribués (inclurais la gestion de process & co). Un FORTH qui verrait toute fonction/agent comme reliée à un BUS MQTT et pour lequel les messages serait du FORTH qui agirait sur la fonction que s'appelerio objective FORTH. (Ça implique de sacrément développé la partie langage). Pleins d'idées, trop d'idées ... ## Les sorties actuellement ressemblent au format d'entrée ... C'est suspect non ? Oui, j'ai envie de tester de laisser l'orchestrateur accepter des injections de code depuis les agents. Ex légitime, quand une sonde de mesure est en OVERRUN (trop de temps passé à mesurer comparé à une cadence attendue) qu'elle puisse changer la « clock » avec TSSET de l'orchestrateur. J'ai envie d'expérimenter des systèmes scheduler less où chaque agent peut devenir le contrôleur et prendre la main et/ou modifier l'orchestrateur qui envoie les commandes.

AltStyle によって変換されたページ (->オリジナル) /