Journal Construire un C&C basique en python/FORTH au dessus de MQTT

Sommaire

• Ingrédients
• La pièce de résistance
• La recette
• Diverses Questions Réponses
  • Un forth dans la boucle était il nécessaire ?
  • Un vrai command & control préférerait un protocole plus passe partout comme https pour être moins visible
  • Et si je voulais sécuriser ?
  • Où est l'arborescente complète du projet il manque les plugins de mesure ?
  • Mais c'est bête en fait…
  • Des idées de futurs ?
  • Les sorties actuellement ressemblent au format d'entrée … C'est suspect non ?

Ingrédients

Cette recette nécessite : python, et en dépendances : paho-mqtt, confined, ainsi qu'un serveur MQTT (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 :

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

#!/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

#!/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:

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 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 :)

Et si je voulais sécuriser ?

MQTT le permet avec des ACL par sujets.

Où est l'arborescente complète du projet il manque les plugins de mesure ?

Les plugins sont ici

Mais c'est bête en fait…

C'est pas faux, c'est pour ça que je sais pas quoi faire avec ce « projet »

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.