Kernkomponenten¶

LXMFBot¶

Die Hauptklasse des Bots. Sie übernimmt das Routing von Nachrichten, die Befehlsverarbeitung und die Verwaltung des Bot-Lebenszyklus.

from lxmfy import LXMFBot

bot = LXMFBot(
    name="MyBot",
    command_prefix="/",
    admins=set(),
    config_path=None,                 # Standard "config" im Arbeitsverzeichnis
    reticulum_config_dir=None,        # oder LXMFY_RETICULUM_CONFIG_DIR / "~/.reticulum"
    test_mode=False,                  # RNS-Start überspringen, für Tests
    log_level="INFO",                 # lxmfy-Logger-Level, None lässt Logging unberührt
    loglevel=None,                    # RNS-Log-Level 0-7, None nutzt die Reticulum-Konfig

    # Announces
    announce=600,
    announce_immediately=True,
    announce_enabled=True,
    announce_display_name_file=None,  # Dateiname unter config_path, der den announced
                                      # Anzeigenamen überschreibt (Standarddatei:
                                      # bot_display_name.txt)

    # Spam-Schutz
    rate_limit=5,
    cooldown=60,
    max_warnings=3,
    warning_timeout=300,

    # Cogs
    cogs_dir="cogs",
    cogs_enabled=True,
    dynamic_cogs_enabled=True,
    external_cogs_enabled=True,
    external_cogs_sandbox_enabled=True,
    external_cogs_sandbox_type="auto",  # "auto", "landlock", "bwrap", "firejail", "none"
    external_cogs_timeout=30,
    hot_reloading=False,

    # Speicher, Ereignisse, Berechtigungen
    storage_type="json",              # "json", "sqlite", "msgpack" oder "memory"
    storage_path="data",
    permissions_enabled=False,
    first_message_enabled=True,
    event_logging_enabled=True,
    max_logged_events=1000,
    event_middleware_enabled=True,

    # Sicherheit
    signature_verification_enabled=False,
    require_message_signatures=False,
    require_stamps=False,             # Nachrichten mit ungültigen Stamps ablehnen
    request_unknown_identities=False, # Absender-Identitäten aus dem Netz anfordern
    stamp_cost=None,                  # eingehende Stamp-Kosten, None deaktiviert
    include_tickets=True,             # Reply-Tickets an Ausgangsnachrichten anhängen
    identity_pinning_enabled=False,
    landlock_enabled=True,

    # Optionale Funktionen
    nlp_enabled=False,
    nlp_threshold=0.5,
    link_support_enabled=False,
    lxmf_commands_enabled=True,

    # Zustellung
    message_persistence_enabled=True,
    message_queue_size=50,
    opportunistic_sending=True,
    direct_delivery_retries=3,
    propagation_fallback_enabled=True,
    propagation_node=None,            # Hash des ausgehenden Propagationsknotens
    autopeer_propagation=False,       # Propagationsknoten aus Announces erkennen
    autopeer_maxdepth=4,              # max. Hop-Tiefe für Autopeering, None = unbegrenzt
    enable_propagation_node=False,    # diesen Bot als Propagationsknoten betreiben
    message_storage_limit_mb=500,     # Speicherlimit des Knotens, nur Knotenmodus

    # Zurückgestellte Sends
    pending_sends_enabled=True,       # Sends für unbekannte Ziele zurückstellen
    pending_sends_max=200,
    pending_sends_ttl=604800,         # 7 Tage
    pending_sends_retry=300,          # Sekunden zwischen Wiederholungsläufen

    # RRC
    rrc_enabled=False,
    rrc_hubs=[],
    rrc_rooms=[],
    rrc_nick=None,
    rrc_dest_name="rrc.hub",
    rrc_auto_reconnect=True,
    rrc_persist_sessions=True,
)

Alle diese Optionen sind Felder von BotConfig. LXMFBot(**kwargs) reicht jedes Schlüsselwortargument durch, sodass bot.config die aufgelösten Werte enthält.

Wichtige Methoden¶

• run(delay=10): Hauptschleife des Bots starten
• cleanup(): Warteschlangen persistieren, Konversationen abbrechen, Scheduler, Router und RNS herunterfahren. Wird beim Verlassen von run() automatisch aufgerufen.
• send(destination, message, title="Reply", lxmf_fields=None, stamp_cost=None, opportunistic=None, method=None, include_ticket=None, defer=None, reply_to=None, quote=None, thread=None): Nachricht an ein Ziel senden. stamp_cost überschreibt die Ausgangskosten für diese Nachricht, opportunistic überschreibt opportunistic_sending, include_ticket überschreibt include_tickets und defer überschreibt pending_sends_enabled. reply_to, quote und thread setzen die Reply-Threading-Felder.
• send_with_attachment(destination, message, attachment, title="Reply", stamp_cost=None, opportunistic=None): Nachricht mit Anhang senden
• command(name, description="No description provided", admin_only=False, permissions=None, usage=None, examples=None, category=None, aliases=None, threaded=False, rate_limit=None): Dekorator zur Registrierung von Befehlen. permissions überschreibt die DefaultPerms-Hürde (ALL bei admin_only, sonst USE_COMMANDS), threaded führt den Callback in einem Worker-Thread aus, rate_limit begrenzt Aufrufe pro Absender und Cooldown-Fenster, und usage, examples, category, aliases speisen das Hilfesystem. Befehle unterstützen typannotierte Argumente für automatische Konvertierung.
• intent(name, examples): Dekorator zur Registrierung von NLP-Intent-Handlern.
• nlp.export_model(): Trainierte NLP-Modelldaten exportieren.
• nlp.import_model(model_data): Zuvor exportierte NLP-Modelldaten importieren.
• request_link(destination_hash, callback=None, app_name="lxmf", *aspects): RNS-Link zu einem Ziel anfordern. Erlaubt eigenes app_name und aspects (Standard "lxmf" und "delivery").
• on_link(callback): Handler für eingehende RNS-Links registrieren.
• load_extension(name): Cog-Erweiterungsmodul anhand des Namens laden (z. B. "cogs.utility").
• reload_extension(name): Cog-Erweiterungsmodul neu laden.
• add_cog(cog_instance): Cog-Klasseninstanz zum Bot hinzufügen.
• remove_cog(cog_name): Cog anhand seines Klassennamens entfernen.
• on_first_message(): Dekorator zur Verarbeitung erster Nachrichten von Benutzern
• on_message(): Dekorator zur Verarbeitung aller Nachrichten (wird vor der Befehlsverarbeitung aufgerufen)
• received(function): Callback registrieren, der mit dem Nachrichtenkontext für jede eingehende Nachricht aufgerufen wird, die die Pipeline durchlief, ohne von einem Befehl oder Intent konsumiert zu werden
• on_reaction(): Dekorator zur Verarbeitung eingehender Reaktionen. Handler erhalten (sender, reaction), wobei reaction die Schlüssel reaction_to, reaction_emoji und reaction_sender enthält
• react(destination, message_hash, reaction): Reaktion auf eine Nachricht über das LXMF-Feld FIELD_REACTION senden
• validate(): Validierungsprüfungen der Bot-Konfiguration ausführen
• get_landlock_status(): Verfügbarkeit und Aktivierungsstatus der Landlock-LSM-Sandbox für den Bot-Prozess zurückgeben
• diagnose_destination(destination, request_path=False, wait=0.0): Identitäts- und Pfadstatus eines Ziel-Hashes prüfen
• diagnose_connectivity(destination=None, request_path=False, wait=0.0): Vollständigen Doctor-Report ausführen und als Dict zurückgeben
• get_debugger(): Einen an diesen Bot gebundenen Debugger zurückgeben
• set_propagation_node(node_hash): Ausgehenden Propagationsknoten festlegen
• get_propagation_node_status(): Konfigurierte, erkannte und aktuell genutzte ausgehende Propagationsknoten
• set_message_storage_limit(megabytes): Speicherlimit im Betrieb als Propagationsknoten
• get_propagation_storage_stats(): Speichernutzung des Knotens oder ein Dict mit dem Grund der Nichtverfügbarkeit
• connect_rrc(hub_hash, rooms=None, nick=None, dest_name=None, auto_reconnect=None): Als Client mit einem RRC-Hub verbinden
• disconnect_rrc(hub_hash=None): Eine oder alle RRC-Hub-Sitzungen trennen
• on_rrc(callback=None): Dekorator oder Handler-Registrierung für RRC-Ereignisse (handler(event, client, payload))
• on_delivery_event(callback=None): Den ausgehenden Zustell-Ereignisstrom abonnieren, als Dekorator oder direkter Aufruf

Attribute¶

• config: Die aufgelöste BotConfig
• commands, cogs: Registrierungen für Befehle und Cogs
• storage: Das aktive Speicher-Backend
• scheduler: TaskScheduler für Cron-artige Aufgaben
• events: EventManager für Ereignis-Handler und Dispatch
• middleware: MiddlewareManager für Befehls-Middleware
• permissions: PermissionManager für Rollen und Flags
• spam_protection: SpamProtection für Rate-Limits, Verwarnungen, Bans
• signature_manager: Signatur-Richtlinienschicht
• nlp: Der Intent-Klassifikator (matched nur bei nlp_enabled)
• delivery: DeliveryTracker, der ausgehende Ereignisstrom
• conversations: ConversationManager für msg.ask-Fragen
• rrc: RRCManager für Multi-Hub-Sitzungen, None bis RRC aktiviert ist oder connect_rrc() läuft
• local: Die RNS.Destination des Bots (die LXMF-Adresse ist bot.local.hash)

Angezeigter Announce-Name¶

Der Name, den Peers sehen, kommt aus name, aber zwei Überschreibungen für Announces existieren. Wenn announce_display_name_file gesetzt ist und die Datei unter config_path existiert, gewinnt ihr Inhalt. Andernfalls wird bot_display_name.txt unter config_path gelesen, falls vorhanden. In beiden Fällen synchronisiert bot.name = "Neuer Name" den angezeigten Announce-Namen zur Laufzeit neu.

So können Betreiber einen Bot umbenennen, ohne Code anzufassen, und der announced Name kann vom internen Konfigurationsnamen abweichen.

Strukturierte Befehle über LXMF-Felder¶

Bots können Befehle empfangen, die über das LXMF-Feld FIELD_COMMANDS (0x09) gesendet werden, und antworten automatisch mit FIELD_RESULTS (0x0A). Damit sind strukturierte Anfrage/Antwort-Workflows neben normalen Textbefehlen möglich.

Eingehende FIELD_COMMANDS werden geparst und über dieselbe Befehlsregistrierung wie Textbefehle geroutet, mit gemeinsamen Berechtigungsprüfungen, typannotiertem Argument-Parsing, Threading und Middleware.

from lxmfy import LXMFBot, FIELD_COMMANDS, FIELD_RESULTS, pack_result, unpack_commands

bot = LXMFBot(name="FieldBot")

@bot.command(name="status", description="Return bot status")
def status_cmd(ctx):
    # ctx.fields enthält das rohe LXMF-Felder-Dict
    # ctx.request_id wird automatisch gesetzt, wenn der Befehl eine enthielt
    ctx.reply("Bot is online")

# Senden eines strukturierten Befehls von einem anderen LXMF-Client:
# lxm.fields[FIELD_COMMANDS] = {"command": "status", "args": [], "request_id": "abc123"}
# router.handle_outbound(lxm)

# Die Bot-Antwort enthält automatisch FIELD_RESULTS mit Antwort und request_id.

Um die Verarbeitung von Feldbefehlen zu deaktivieren, lxmf_commands_enabled=False in BotConfig setzen.

Reaktionen¶

Reaktionen werden als LXMF-Feld FIELD_REACTION (0x40) in einer ansonsten leeren Nachricht übertragen. pack_reaction und unpack_reaction erzeugen und parsen dieses Feld.

from lxmfy import pack_reaction, unpack_reaction

# Eine Reaktion auf eine Nachricht senden
bot.react(destination_hash, message_hash_hex, "thumbs up emoji")

# Reaktionen empfangen
@bot.on_reaction()
def on_reaction(sender, reaction):
    # reaction["reaction_to"]  - Hex-Hash der Zielnachricht
    # reaction["reaction_emoji"] - Reaktionstext (bis zu 16 Zeichen)
    # reaction["reaction_sender"] - Absender
    print(f"{sender} reacted {reaction['reaction_emoji']} to {reaction['reaction_to']}")
    return True

Der Reaktionstext ist auf 16 druckbare Zeichen begrenzt. Das Rohfeld bleibt in ctx.fields und msg.fields verfügbar (Kompatibilität).

Antwort-Threading¶

Antworten können die LXMF-Felder FIELD_REPLY_TO (0x30), FIELD_REPLY_QUOTE (0x31) und FIELD_THREAD (0x08) tragen. Clients mit Threading, wie MeshChatX und Sideband, zeigen sie als echte Zitat-Antworten statt als flache Nachrichten.

msg.reply() threaded automatisch: es setzt FIELD_REPLY_TO auf den Hash der eingehenden Nachricht und FIELD_THREAD auf die Wurzel der Konversation.

@bot.command("status")
def status(msg):
    msg.reply("all systems nominal")          # Threaded-Antwort
    msg.reply("flat", reply_to=None)          # Threading abwählen
    msg.reply("noted", quote=True)            # Eingangstext zitieren

Für Sends, die keine Antworten sind, die Felder explizit übergeben:

bot.send(dest, "see above", reply_to=msg_hash_hex, quote="earlier text")

Eingehende Antworten werden auf dem Nachrichtenkontext geparst:

@bot.command("ctx")
def ctx_cmd(msg):
    msg.reply_to      # Hex-Hash, auf den diese Nachricht antwortet, oder None
    msg.reply_quote   # zitierter Text der Antwort, oder None
    msg.thread        # Hex-Hash der Thread-Wurzel, oder None

pack_reply(message_hash, quote=..., thread=...) und unpack_reply(fields) sind für die manuelle Feldbehandlung exportiert.

Konversationen¶

Befehle können dem Absender eine Frage stellen und seine nächste Nachricht als Antwort behandeln, statt sie als Befehl zu dispatchen:

@bot.command("report")
def report(msg):
    title = msg.ask("Report title?", timeout=300)
    if title is None:
        msg.reply("Timed out.")
        return
    body = msg.ask("Describe the issue.", timeout=600)
    if body is None:
        msg.reply("Timed out.")
        return
    msg.reply(f"Filed: {title.content}")

msg.ask(prompt, timeout=..., validator=...) blockiert den Handler, bis die Antwort ankommt, der Timeout abläuft oder die Konversation abgebrochen wird. Es gibt ein Answer mit content, fields, hash, sender und einem reply(text)-Shortcut zurück.

Ein Validator lehnt schlechte Antworten ab und fragt erneut:

num = msg.ask(
    "Pick a number",
    validator=lambda a: None if a.content.isdigit() else "Digits only",
)

In Async-Befehlshandlern await msg.ask_async(...) verwenden. Für lange Wartezeiten oder viele offene Konversationen den Callback-Stil nutzen, damit kein Thread geparkt bleibt:

msg.ask(
    "Send the log file",
    on_answer=lambda ans: ans.reply("received"),
    on_timeout=lambda sender: bot.send(sender, "Too slow."),
    timeout=3600,
)

Hinweise:

• Ein registrierter Befehl während einer offenen Frage bricht die Frage ab und führt den Befehl aus. Benutzer haben immer einen Ausweg.
• bot.conversations.pending_count() und bot.conversations.cancel(sender) geben die Registry für Diagnose und Admin-Werkzeuge frei.
• Die Registry ist auf 1024 offene Fragen begrenzt. ask gibt None zurück, wenn sie voll ist.
• Blockierendes ask parkt den Zustell-Thread dieser Nachricht. Das ist bei Direktzustellungen sicher, aber Bots, die große Batches von einem Propagationsknoten synchronisieren, sollten on_answer-Callbacks bevorzugen.

Speicher¶

Das Framework bietet drei Speicher-Backends:

JSONStorage¶

from lxmfy import JSONStorage

storage = JSONStorage("data")

SQLiteStorage¶

from lxmfy import SQLiteStorage

storage = SQLiteStorage("data/bot.db")

MsgPackStorage¶

from lxmfy.storage import MsgPackStorage

storage = MsgPackStorage("data") # {key}.msgpack files

MemoryStorage¶

from lxmfy.storage import MemoryStorage

storage = MemoryStorage() # Vollständig im Arbeitsspeicher

Befehle¶

Befehlsregistrierung und -verarbeitung:

@bot.command(name="hello", description="Says hello")
def hello(ctx):
    ctx.reply(f"Hello {ctx.sender}!")

Typannotierte Argumente¶

Befehle parsen und konvertieren Argumente automatisch anhand der Typannotationen in der Callback-Funktion.

@bot.command(name="add", description="Adds two numbers")
def add(ctx, a: int, b: int):
    result = a + b
    ctx.reply(f"The result is {result}")

Hilfemetadaten und Zugriffskontrolle kommen aus zusätzlichen Dekorator-Kwargs:

from lxmfy import DefaultPerms

@bot.command(
    name="purge",
    description="Clear stored data",
    permissions=DefaultPerms.MANAGE_MESSAGES,
    usage="/purge <key>",
    examples=["/purge cache"],
    category="Admin",
    aliases=["clear"],
)
def purge(ctx, key: str):
    bot.storage.delete(key)
    ctx.reply(f"Deleted {key}")

permissions überschreibt die Standardhürde: USE_COMMANDS für normale Befehle, ALL für admin_only. category gruppiert den Befehl in der /help-Ausgabe. aliases ist nur Hilfe-Metadatum: Aliasnamen werden angezeigt, aber nicht für den Dispatch registriert, sodass /clear purge nicht ausführt, es sei denn, es wird als zweiter Befehl registriert.

Pro-Befehl-Rate-Limits¶

Begrenzt, wie oft ein einzelner Absender einen Befehl innerhalb des globalen cooldown-Fensters aufrufen darf. Beim Erreichen wird nur der Aufruf abgelehnt, es gibt keine Verwarnungen oder Bans.

@bot.command(name="report", rate_limit=3)
def report(ctx):
    # jeder Absender kann diesen Befehl 3-mal pro Cooldown-Fenster aufrufen
    ...

Erfordert permissions_enabled=True, wie das globale Rate-Limit. Benutzer mit der Admin-Rolle oder BYPASS_SPAM überspringen die Prüfung.

Spam-Schutz¶

bot.spam_protection setzt das globale Rate-Limit durch: ein Absender darf rate_limit Nachrichten pro cooldown-Fenster senden. Bei Überschreitung gibt es eine Verwarnung und die Nachricht wird abgelehnt. Bei max_warnings wird der Absender gebannt. Verwarnungen verfallen nach warning_timeout Sekunden ohne Fehlverhalten.

Spam-Prüfungen laufen im message_received-Ereignis und erfordern permissions_enabled=True.

bot = LXMFBot(
    name="GuardedBot",
    permissions_enabled=True,
    rate_limit=5,        # Nachrichten pro Cooldown-Fenster
    cooldown=60,         # Fensterlänge in Sekunden
    max_warnings=3,      # Verwarnungen vor einem Ban
    warning_timeout=300, # Sekunden bis zum Zurücksetzen der Verwarnungen
)

# Ban manuell aufheben
bot.spam_protection.unban(sender_hash)

Verwarnungen, Bans und Zähler persistieren im konfigurierten Speicher-Backend und überleben Neustarts. Absender mit BYPASS_SPAM werden nie ratenlimitiert oder gebannt. Das rate_limit pro Befehl bei @bot.command ist milder: es lehnt nur den Aufruf ab und warnt oder bannt nie.

Hilfesystem¶

Das Framework enthält einen interaktiven Hilfe-Generator, der kategorisierte Hilfemenüs auf Basis der Cog- und Command-Metadaten erzeugt.

# Der help-Befehl wird automatisch registriert.
# Benutzer können '/help' oder '/help <command>' verwenden

Befehle in separaten Threads¶

Für lang laufende oder blockierende Operationen, die nicht direkt mit dem Reticulum Network Stack interagieren, können Befehle in einem separaten Thread laufen, damit der Bot reaktionsfähig bleibt.

import time

@bot.command(name="long_task", description="Performs a long-running task in a separate thread", threaded=True)
def long_task_command(ctx):
    ctx.reply("Starting a long task... please wait.")
    time.sleep(10) # Läuft in einem separaten Thread
    ctx.reply("Long task completed!")

Ereignisse¶

Ereignissystem zur Verarbeitung verschiedener Bot-Ereignisse:

@bot.events.on("message_received", EventPriority.HIGHEST)
def handle_message(event):
    # event.data trägt die Nutzlast, z. B. sender und message
    event.cancel()  # spätere Handler und weitere Verarbeitung stoppen

Handler laufen in EventPriority-Reihenfolge: HIGHEST, HIGH, NORMAL, LOW. Die Spam-Prüfung selbst ist ein message_received-Handler mit HIGHEST, das Canceln dieses Ereignisses ist also der Mechanismus, mit dem der Rate-Limiter Nachrichten verwirft.

Eigene Ereignisse dispatchen:

from lxmfy import Event

bot.events.dispatch(Event("order_placed", data={"user": ctx.sender}))

event_logging_enabled, max_logged_events und event_middleware_enabled existieren in BotConfig, sind aber nicht verdrahtet: Ereignisse werden nicht in den Speicher geschrieben und bot.events.use() ist ein Stub. Als reserviert behandeln.

Testen¶

lxmfy.testing.TestBot ist ein für Tests vorkonfigurierter LXMFBot. Es startet keine Reticulum-Instanz. Eingehende Nachrichten laufen durch die echte Empfangspipeline (Middleware, Spam-Prüfungen, Berechtigungen, Dispatch) und ausgehende Sends werden für Assertions abgefangen.

from lxmfy import TestBot

def test_ping():
    with TestBot() as bot:
        @bot.command("ping")
        def ping(msg):
            msg.reply("pong")

        sent = bot.receive("/ping", sender="alice")
        assert sent[0].content == "pong"
        assert sent[0].destination == bot.sender_hex("alice")

• bot.receive(content, sender=..., fields=..., message_hash=...) injiziert eine Nachricht und gibt die erzeugten SentMessage-Objekte zurück. Absender sind benannt: "alice" mappt auf einen stabilen Fake-Hash, oder man übergibt direkt einen Hex-Ziel-Hash.
• bot.drain() holt die ausgehende Warteschlange ab. bot.outbox sammelt alles Gesendete. bot.last_sent(sender=...) liefert das neueste.
• bot.wait_sent(n, timeout=...) wartet auf Threaded-Befehle.
• bot.receive_later(content, sender=..., delay=...) beantwortet blockierende msg.ask-Aufrufe aus einem Daemon-Thread.
• fake_message(content, source_hash=..., ...) baut eine eingehende Nachricht, um bot._message_received direkt zu treiben.

SentMessage kapselt jede abgefangene Ausgangsnachricht: destination (Hex), content, title, fields, method, include_ticket, stamp_cost und raw für das zugrunde liegende Objekt.

Die Projekt-Tests umfassen außerdem Zuverlässigkeits- und Stresstest-Szenarien in der Testsuite des Repositorys. Führen Sie sie mit dem Test-Runner des Repositorys aus.

Erweiterte Zuverlässigkeits-Testsuite¶

Das Framework enthält eine umfangreiche Suite automatisierter Tests für raue Umgebungsbedingungen:

• Manifold Testing: Validiert die mathematische Topologie des NLP-Intent-Vektorraums.
• Chaos Engineering: Simuliert Bitrott, SD-Karten-Ausfall und Speicherkorruption.
• Temporal Drift: Prüft die Robustheit gegenüber Systemuhrsprüngen (±1 Jahr).
• Leak Detection: Langzeit-Überwachung von Speicher, Dateideskriptoren und Threads.

Berechtigungen¶

Berechtigungssystem zur Zugriffskontrolle auf Bot-Funktionen:

from lxmfy import DefaultPerms

@bot.command(name="admin", description="Admin command", admin_only=True)
def admin_command(ctx):
    if ctx.is_admin:
        ctx.reply("Admin command executed")

Mit permissions_enabled=True aktivieren. DefaultPerms-Flags:

• USE_BOT, SEND_MESSAGES, USE_COMMANDS: Basiszugriff
• MANAGE_MESSAGES, MANAGE_COMMANDS, MANAGE_USERS: erhöht
• BYPASS_RATELIMIT, BYPASS_SPAM, VIEW_ADMIN_COMMANDS: speziell
• VIEW_EVENTS, MANAGE_EVENTS, BYPASS_EVENT_CHECKS: Ereignissystem
• NONE, ALL: Kurzformen

bot.permissions verwaltet Rollen und Zuweisungen:

bot.permissions.create_role("moderator", DefaultPerms.MANAGE_MESSAGES | DefaultPerms.BYPASS_SPAM)
bot.permissions.assign_role(user_hash, "moderator")
bot.permissions.remove_role(user_hash, "moderator")
bot.permissions.has_permission(user_hash, DefaultPerms.USE_COMMANDS)

Rollen und Zuweisungen persistieren im konfigurierten Speicher-Backend. Zwei eingebaute Rollen existieren und können nicht gelöscht werden: user (Standard) und admin, die automatisch jedem Hash in admins zugeteilt wird.

Middleware¶

Middleware-System zur Verarbeitung von Nachrichten und Ereignissen:

from lxmfy import MiddlewareType

@bot.middleware.register(MiddlewareType.PRE_COMMAND)
def pre_command_middleware(ctx):
    # ctx kapselt den Nachrichtenkontext, ctx.cancelled verwirft ihn
    if "spamword" in ctx.data.content:
        ctx.cancel()

Drei Punkte in der Pipeline führen Middleware aus:

• PRE_COMMAND: vor dem Befehls-Dispatch, nach den Spam-Prüfungen. Gibt die Kette None zurück, wird die Nachricht komplett abgebrochen.
• POST_COMMAND: nach dem Callback eines Befehls (auch Threaded-Befehle, die ihn im Worker-Thread auslösen).
• PRE_EVENT: vor dem Dispatch des message_received-Ereignisses.

POST_EVENT, REQUEST und RESPONSE existieren in MiddlewareType, aber noch nichts in der Pipeline führt sie aus.

Anhänge¶

Unterstützung für das Senden von Dateien, Bildern und Audio:

from lxmfy import Attachment, AttachmentType

attachment = Attachment(
    type=AttachmentType.IMAGE,
    name="image.jpg",
    data=image_data,
    format="jpg"
)
bot.send_with_attachment(destination, "Here's an image", attachment)

Icon-Darstellung (LXMF-Feld)¶

Sie können für Ihren Bot ein eigenes Icon setzen, das kompatible LXMF-Clients anzeigen können. Dabei wird LXMF.FIELD_ICON_APPEARANCE verwendet.

from lxmfy import IconAppearance, pack_icon_appearance_field
import LXMF # Erforderlich für LXMF.FIELD_ICON_APPEARANCE

# Icon-Darstellung definieren
icon_data = IconAppearance(
    icon_name="smart_toy",  # Name aus Material Symbols
    fg_color=b'\xFF\xFF\xFF',  # Weißer Vordergrund (3 Bytes)
    bg_color=b'\x4A\x90\xE2'   # Blauer Hintergrund (3 Bytes)
)

# In das LXMF-Feldformat packen
icon_lxmf_field = pack_icon_appearance_field(icon_data)

# Nachricht mit diesem Icon senden
bot.send(
    destination_hash_str,
    "Hello from your friendly bot!",
    title="Bot Message",
    lxmf_fields=icon_lxmf_field
)

# Kann auch mit anderen Feldern kombiniert werden, z. B. Anhängen:
# attachment_field = pack_attachment(some_attachment)
# combined_fields = {**icon_lxmf_field, **attachment_field}
# bot.send(destination, "Message with icon and attachment", lxmf_fields=combined_fields)

Scheduler¶

System zur Aufgabenplanung:

@bot.scheduler.schedule(name="daily_task", cron_expr="0 0 * * *")
def daily_task():
    # Läuft täglich um Mitternacht
    pass

Signaturen¶

LXMFy bietet Konfigurationsoptionen für die in LXMF eingebaute kryptografische Nachrichtensignierung und -prüfung:

from lxmfy import LXMFBot

bot = LXMFBot(
    name="SecureBot",
    signature_verification_enabled=True,  # Signaturprüfung aktivieren
    require_message_signatures=False      # Auf True setzen, um unsignierte Nachrichten abzulehnen
)

Die eigentlichen kryptografischen Operationen führt LXMF/RNS aus, nicht LXMFy.

Landlock-LSM-Sandbox¶

Auf Linux-Kerneln mit Landlock-Unterstützung (5.13+) kann LXMFy den Dateisystemzugriff für den Bot-Prozess und für externe Skript-Cogs einschränken.

Sandbox des Bot-Prozesses

Wenn landlock_enabled=True (Standard) und der Bot nicht im test_mode läuft, ruft der Bot apply_landlock_sandbox() während der Initialisierung auf. Systemverzeichnisse sind nur lesbar. Bot-Speicher, Konfiguration, Cogs, Reticulum-Konfiguration und Temp-Pfade bleiben schreibbar.

bot = LXMFBot(
    name="SecureBot",
    landlock_enabled=True,
)

status = bot.get_landlock_status()
# status-Schlüssel: landlock_kernel_supported, landlock_requested,
# landlock_auto_enabled, landlock_disabled_by_env, landlock_active

Überschreibung per Umgebungsvariable

• LXMFY_LANDLOCK=0: Landlock auch auf unterstützten Kerneln deaktivieren
• LXMFY_LANDLOCK=1: Landlock unter Linux unabhängig von der Auto-Erkennung versuchen
• nicht gesetzt: landlock_enabled und Kernel-Auto-Erkennung folgen

Sandbox für externe Cogs

Skript-Cogs verwenden external_cogs_sandbox_type. Im Modus auto wird Landlock bevorzugt, wenn verfügbar, da keine externen Werkzeuge nötig sind. Die vollständige Liste der Sandbox-Optionen steht in der Anleitung Bots erstellen.

Identitäts-Pinning¶

LXMFy unterstützt optionales Identitäts-Pinning, um Identitätsdiebstahl zu verhindern, falls eine Identität rotiert oder kompromittiert wird. Wenn aktiviert, "pinnt" der Bot eine LXMF-Adresse an den zuerst gesehenen öffentlichen Schlüssel.

bot = LXMFBot(
    identity_pinning_enabled=True
)

SignatureManager-Methoden¶

Der SignatureManager ist als bot.signature_manager verfügbar, wenn signature_verification_enabled=True:

• should_verify_message(sender): Feststellen, ob eine Nachricht des gegebenen Absenders geprüft werden soll
• handle_unsigned_message(sender, message_hash): Nachrichten ohne gültige Signatur gemäß Richtlinie behandeln

Wie LXMF-Signaturen funktionieren¶

LXMF signiert alle ausgehenden Nachrichten automatisch mit der RNS-Identität des Absenders während der pack()-Operation. Beim Empfang validiert LXMF die Signaturen und stellt bereit:

• message.signature_validated: Boolescher Wert, ob die Signatur gültig ist
• message.unverified_reason: Grundcode, falls die Validierung fehlschlug (z. B. SIGNATURE_INVALID, SOURCE_UNKNOWN)

LXMFy nutzt diese eingebauten LXMF-Eigenschaften, um die Signaturrichtlinie Ihres Bots durchzusetzen.

Nachrichtenzustellung¶

LXMFy bietet erweiterte Zustellfunktionen, darunter Propagationsknoten und automatische Wiederholungen:

Propagationsknoten¶

Nachrichten über bestimmte Propagationsknoten senden, um die Zustellzuverlässigkeit im Reticulum-Netzwerk zu erhöhen:

# Propagationsknoten einmalig auf Konfigurations-/Laufzeitebene setzen
bot.set_propagation_node("<propagation_node_hash>")

# Mit konfiguriertem Zustellverhalten senden
bot.send(
    destination_hash,
    "Message content"
)

# Der Propagationsknoten-Hash sollte ein gültiger LXMF-Propagationsknoten
# im Reticulum-Netzwerk sein

Der Knoten kann auch bei der Erstellung mit propagation_node="<hash>" gesetzt werden, oder der Bot erkennt Knoten selbst:

bot = LXMFBot(
    name="AutoBot",
    autopeer_propagation=True, # Knoten aus Announces lernen
    autopeer_maxdepth=4,       # Knoten tiefer als 4 Hops ignorieren
)

bot.get_propagation_node_status() meldet den manuellen Knoten, die erkannten Knoten und den aktuell genutzten ausgehenden Knoten.

Automatische Wiederholungen¶

Automatische Wiederholungsversuche für fehlgeschlagene Direktzustellungen konfigurieren:

bot = LXMFBot(
    name="ReliableBot",
    direct_delivery_retries=5,  # Direktzustellung bis zu 5-mal wiederholen
    propagation_fallback_enabled=True
)

bot.send(destination_hash, "Important message")

# direct_delivery_retries ist standardmäßig 3
# Die Wiederholungslogik verarbeitet Zustell-Callbacks automatisch

Das Wiederholungssystem verfolgt Zustellversuche pro Ziel und wiederholt fehlgeschlagene Zustellungen automatisch. Erfolgreiche Zustellungen setzen den Zähler für dieses Ziel zurück.

Zurückgestellte Sends¶

Ein Send an ein Ziel, dessen Identität der Knoten noch nicht gehört hat, schlägt normalerweise direkt fehl. Mit pending_sends_enabled (Standard) wird die Nachricht stattdessen im Speicher gehalten und automatisch geflusht, wenn das Ziel announcet oder beim periodischen Lauf.

bot = LXMFBot(
    pending_sends_enabled=True,
    pending_sends_max=200,     # älteste gehaltene Nachrichten fallen darüber hinaus
    pending_sends_ttl=604800,  # gehaltene Nachrichten laufen nach 7 Tagen ab
    pending_sends_retry=300,   # Sekunden zwischen den Läufen in run()
)

# Überschreibung pro Send
bot.send(dest, "hold this", defer=True)
bot.send(dest, "send or drop", defer=False)

Gehaltene Sends erscheinen als held (unknown peers) im Admin-Befehl /queue und erzeugen deferred-Ereignisse im Delivery-Tracker.

Nachrichten-Persistenz¶

Ausgehende Nachrichten können auf die Festplatte persistiert werden, damit sie auch nach einem Neustart des Bots zugestellt werden. Persistenz ist standardmäßig aktiviert. Die Ausgangswarteschlange im Speicher ist begrenzt (message_queue_size, Standard 50) und verwirft bei Volllast die älteste Nachricht. Ungültige Ziel-Hashes werden nicht wiederhergestellt.

bot = LXMFBot(
    message_persistence_enabled=True,
    message_queue_size=50,
)

Stamps und Tickets¶

Stamp-Kosten lassen Absender Proof-of-Work zahlen, bevor ihre Nachricht akzeptiert wird, was unerwünschten Verkehr drosselt. LXMFy legt beide Seiten des Mechanismus offen.

bot = LXMFBot(
    stamp_cost=16,          # diese eingehenden Stamp-Kosten verlangen
    require_stamps=True,    # Nachrichten mit ungültigen Stamps ablehnen
    include_tickets=True,   # Peers ohne Stamp-Aufwand antworten lassen
)

• stamp_cost ist die eingehende Anforderung. Die Ausgangskosten eines Sends kommen weiterhin aus dem Announce des Peers, außer stamp_cost= wird an bot.send() übergeben.
• include_tickets (Standard True) hängt ein Reply-Ticket an ausgehende Nachrichten, damit ein Peer mit Stamp-Pflicht antworten kann, ohne zu zahlen. Pro Send mit include_ticket= überschreiben.
• request_unknown_identities=True fordert vom Netz eine Absender-Identität an, wenn eine Nachricht aus unbekannter Quelle ankommt, damit Stamp- und Signaturprüfungen auflösen statt blind zu scheitern.

Laufzeitkontrolle liegt unter Router-Steuerung: set_inbound_stamp_cost, enforce_stamps, ignore_stamps, generate_ticket und die Ticket-Inspektionsmethoden.

Als Propagationsknoten laufen¶

Ein Bot kann zugleich als LXMF-Propagationsknoten dienen und Nachrichten für offline Peers speichern:

bot = LXMFBot(
    enable_propagation_node=True,
    message_storage_limit_mb=500,
)

bot.set_message_storage_limit(750)
stats = bot.get_propagation_storage_stats()

get_propagation_node_status() funktioniert für beide Rollen: es meldet den ausgehenden Knoten, den dieser Bot nutzt, und ob er selbst als Knoten dient. Verwandte Steuerungen: announce_propagation_node() bewirbt den Knoten, set_retain_on_node() behält zugestellte Nachrichten darauf, und allow_control_identity() / disallow_control_identity() verwalten, welche Identitäten den Steuerkanal des Knotens nutzen dürfen.

Zustell-Ereignisse¶

bot.delivery zeichnet einen begrenzten Strom ausgehender Lebenszyklus-Ereignisse auf, sodass der Nachrichtenfluss ohne Log-Lektüre beobachtbar ist. Stufen: queued, deferred, dispatched, delivered, failed, cancelled, dropped. Das jüngste Ende wird im Speicher persistiert und beim Start wiederhergestellt.

@bot.on_delivery_event()
def watch(event):
    print(event["stage"], event.get("destination"), event.get("reason"))

# Oder direkt inspizieren
recent = bot.delivery.recent(20)
failures = bot.delivery.recent(stage="failed")
to_peer = bot.delivery.recent(destination="aa11bb...")

Jedes Ereignis ist ein Dict mit ts, stage und optional destination, message_id, hash, method, attempts, reason, title.

Admins erhalten einen /delivery [limit]-Befehl, der dieselbe Zeitleiste im Chat rendert, und lxmfy debug zeigt eine Zustell-Zeitleisten-Zusammenfassung in den Send-Pipeline-Prüfungen.

Eingebaute Admin-Befehle¶

Diese Befehle werden automatisch registriert und verlangen, dass der Absender in admins steht, wenn Berechtigungen aktiviert sind:

Router-Steuerung¶

Dünne Wrapper über den zugrunde liegenden LXMRouter für Absendersteuerung, Tickets, Ausgangswarteschlangen-Verwaltung und Propagationsknoten-Sync. Alle nehmen Ziel-Hashes als Hex-Strings und geben False zurück, wenn der Router nicht läuft (etwa im test_mode).

Absendersteuerung (eingehend)

• ignore_destination(destination) / unignore_destination(destination) / is_ignored(destination): Eingehende Nachrichten eines Absenders verwerfen
• allow_destination(destination) / disallow_destination(destination): Whitelist-Verwaltung, wenn der Router im Allow-List-Modus läuft
• prioritise_destination(destination) / unprioritise_destination(destination): Liste priorisierter Absender
• set_inbound_stamp_cost(stamp_cost): Stamp-Kosten für eingehende Nachrichten verlangen (None löscht)
• enforce_stamps() / ignore_stamps(): Eingehende Stamp-Durchsetzung umschalten

Tickets

• generate_ticket(destination, expiry=None): Ein eingehendes Stamp-Ticket für einen Absender ausstellen
• get_inbound_tickets(destination): Für einen Absender gehaltene Tickets
• get_outbound_ticket(destination) / get_outbound_ticket_expiry(destination) / get_outbound_stamp_cost(destination): Aus dem Netz gelerntes ausgehendes Ticket

Ausgangswarteschlange

• outbound_queue(): Snapshot ausstehender ausgehender Nachrichten
• get_outbound_progress(lxm_hash): Zustellfortschritt für einen Nachrichten-Hash oder None
• cancel_outbound(message_id): Eine wartende Nachricht vor der Zustellung entfernen
• delivery_link_available(destination): Ob ein aktiver RNS-Link zum Ziel existiert

Eingangswarteschlange

• has_message(message_hash): Ob ein eingehender LXM-Hash bereits zugestellt wurde
• inbound_count(): Laufende aktive Eingangs-Ressourcenübertragungen
• inbound_transfers(): Snapshot jeder Übertragung mit Hash, Größe, Fortschritt und Status
• cancel_inbound(resource_hash): Eine aktive Eingangsübertragung abbrechen
• cancel_all_inbound(): Jede aktive Eingangsübertragung abbrechen, gibt die Anzahl zurück

Peer-Erkennung

Announce-Metadaten für Ziele, die dieser Knoten gehört hat:

• get_peer_app_data(destination): Rohe announced app_data-Bytes
• get_peer_lxmf_data(destination): Dekodierte LXMF-Announce-Metadaten (display_name, stamp_cost, capabilities) oder None, wenn der Peer keine gültigen LXMF-Daten announced hat
• get_peer_announce(destination): Vollständiger Announce-Datensatz mit Hops, received_at, Interface und app_data
• list_peer_announces(limit=100): Alle gehörten Announces, neueste zuerst

Propagation

• sync_propagation_node(max_messages=None): Nachrichten vom konfigurierten Propagationsknoten abrufen
• cancel_propagation_sync(): Einen laufenden Sync stoppen
• get_propagation_stats(): Übertragungsstatus und Limits des Knotens oder None
• set_retain_on_node(retain): Zugestellte Nachrichten auf dem Knoten behalten
• announce_propagation_node(): Diesen Knoten als Propagationsknoten announcen
• allow_control_identity(destination) / disallow_control_identity(destination): Whitelist des Propagations-Steuerkanals

Ingest

• ingest_lxm_uri(uri): Eine lxm://-URI-Nachricht in die Eingangswarteschlange importieren

Diagnose¶

Wenn Nachrichten nicht fließen, prüft der Debugger den ganzen Pfad statt zu raten: Reticulum-Konfiguration, Shared-Instance-Status, Interfaces, Identität, Announce-Verhalten, Zustellkonfiguration und Send-Pipeline.

lxmfy debug                          # vollständiger Doctor-Report, in Datei gespeichert
lxmfy debug probe <hash> --request-path --wait 30
lxmfy debug send <hash>              # einen Testsend nachverfolgen
lxmfy debug receive                  # Empfangsbereitschaft prüfen
lxmfy debug compare <hash_a> <hash_b>
lxmfy debug tips                     # Korrekturen für häufige Fehler

Reports sind standardmäßig datenschutzreduziert: Home-Pfade und Hashes werden gekürzt. --json erzeugt maschinenlesbare Ausgabe, -o DATEI schreibt den Report, --no-save überspringt die Datei, --no-privacy behält volle Werte für lokale Nutzung, und --no-color oder NO_COLOR deaktiviert ANSI-Ausgabe.

Dieselben Prüfungen sind aus Code aufrufbar:

report = bot.diagnose_connectivity()            # Doctor-Report als Dict
probe = bot.diagnose_destination(               # Identitäts- und Pfadprobe
    "<peer_hash>", request_path=True, wait=30,
)

debugger = bot.get_debugger()                   # volle API
checks = debugger.check_send_pipeline()
verdict = debugger.run_doctor(destination)

lxmfy/debugger.py exportiert außerdem einen eigenständigen diagnose_destination(hash, ...)-Helper plus die Report-Typen CheckResult, DestinationProbe, DoctorReport und MessageDebugger für eigenes Tooling.

Nachrichten-Handler¶

LXMFy stellt Dekoratoren für verschiedene Arten eingehender Nachrichten bereit:

Handler für erste Nachrichten¶

Die erste Nachricht jedes Benutzers verarbeiten:

@bot.on_first_message()
def welcome_user(sender, message):
    content = message.content.decode("utf-8")
    bot.send(sender, f"Welcome! You said: {content}")
    return True  # True zurückgeben, um die weitere Verarbeitung zu stoppen

Allgemeiner Nachrichten-Handler¶

Alle eingehenden Nachrichten vor der Befehlsverarbeitung verarbeiten:

@bot.on_message()
def handle_all_messages(sender, message):
    content = message.content.decode("utf-8").strip()

    # Eigene Logik hier
    if content.startswith("echo:"):
        bot.send(sender, content[5:])
        return True  # Weitere Verarbeitung stoppen

    return False  # Weiter zur Befehlsverarbeitung

Nachrichten-Handler werden in dieser Reihenfolge aufgerufen: 1. Handler für erste Nachrichten (wenn dies die erste Nachricht dieses Absenders ist) 2. Allgemeine Nachrichten-Handler (registriert mit on_message()) 3. Befehlsverarbeitung (wenn die Nachricht mit dem Befehlspräfix beginnt)

Fallback-Callback¶

bot.received(fn) registriert einen Callback, der am Ende der Pipeline für Nachrichten läuft, die nichts anderes konsumiert hat: kein First-Message-Handler, kein on_message-Handler mit True-Rückgabe, kein passender Befehl oder NLP-Intent. Der Callback erhält denselben Nachrichtenkontext wie Befehle, mit msg.sender, msg.content, msg.reply() und mehr.

@bot.received
def fallback(msg):
    msg.reply("Sorry, I did not understand that.")

Als Catch-all für Freitext-Eingaben verwenden.

Reticulum Relay Chat (RRC)¶

Bots können RRC-Hubs über RNS-Links mit CBOR-Envelopes beitreten. Paket: lxmfy.rrc.

BotConfig-Optionen¶

• rrc_enabled (bool, Standard False): Konfigurierte Hubs beim Start verbinden
• rrc_hubs (Liste von Hex-Hashes): Ziel-Hashes der Hubs
• rrc_rooms (Liste von Strings): Räume, die nach WELCOME automatisch betreten werden
• rrc_nick (String oder None): Nickname in HELLO und Raumnachrichten
• rrc_dest_name (String, Standard "rrc.hub"): Zielname zum Aufbau der Hub-Destination
• rrc_auto_reconnect (bool, Standard True): Nach Link-Verlust neu verbinden
• rrc_persist_sessions (bool, Standard True): Hubs und Räume über Neustarts hinweg speichern
• reticulum_config_dir (String oder None): Reticulum-Konfigurationsverzeichnis. Kann auch über LXMFY_RETICULUM_CONFIG_DIR gesetzt werden. Dieselbe Konfiguration wie MeshChatX verwenden (oft ~/.reticulum), damit Hub-Announces sichtbar sind.

Beispiel¶

from lxmfy import LXMFBot, RRCMessage

bot = LXMFBot(
    name="RoomBot",
    reticulum_config_dir="~/.reticulum",
    rrc_enabled=True,
    rrc_hubs=["664fc0e8d2e448658e37bb3f34e6c88f"],
    rrc_rooms=["general"],
    rrc_nick="RoomBot",
)

@bot.on_rrc
def on_rrc(event, client, payload):
    if event == "msg" and isinstance(payload, RRCMessage) and payload.mention:
        client.send_message(payload.room, f"Hi {payload.nick}")

# Laufzeit-API
# bot.connect_rrc(hub_hash, rooms=["general"])
# bot.rrc.send_message("general", "hello")
# bot.rrc.send_notice("general", "notice")
# bot.rrc.send_action("general", "waves")
# bot.rrc.join("ops")
# bot.rrc.part("ops")
# bot.rrc.status()
# bot.disconnect_rrc()

Exportierte Typen¶

• RRCClient: Sitzung mit einem Hub
• RRCManager: Manager für mehrere Hubs (bot.rrc)
• RRCMessage: Payload eines Raumereignisses (kind, room, text, nick, src, mention, ...)
• RRC_VERSION: Konstante der Wire-Protokollversion
• DEFAULT_DEST_NAME: Standard-Hub-Zielname ("rrc.hub")
• make_envelope, encode_envelope, decode_envelope, validate_envelope, normalize_room: Wire-Format-Helper für Tooling, das direkt mit Hubs spricht

Übliche Ereignisse, die an @bot.on_rrc-Handler übergeben werden: status, welcome, joined, parted, msg, notice, action, motd, error und rtt.

Vorlagen¶

Das Framework enthält mehrere einsatzbereite Bot-Vorlagen:

EchoBot¶

Einfacher Echo-Bot, der Nachrichten wiederholt:

from lxmfy.templates import EchoBot

bot = EchoBot()
bot.run()

NoteBot¶

Notizen-Bot mit JSON-Speicher:

from lxmfy.templates import NoteBot

bot = NoteBot()
bot.run()

ReminderBot¶

Erinnerungs-Bot mit SQLite-Speicher:

from lxmfy.templates import ReminderBot

bot = ReminderBot()
bot.run()

RRCBot¶

RRC-Raum-Bot, der konfigurierten Hubs beitritt und auf @mentions antwortet. Standardmäßig Hub 664fc0e8d2e448658e37bb3f34e6c88f, Raum #general und ~/.reticulum, falls verfügbar.

from lxmfy.templates import RRCBot

bot = RRCBot(
    hubs=["664fc0e8d2e448658e37bb3f34e6c88f"],
    rooms=["general"],
    nick="RRCBot",
    reticulum_config_dir="~/.reticulum",
)
bot.run()

CLI-Werkzeuge¶

Das Framework stellt Kommandozeilen-Werkzeuge für die Bot-Verwaltung bereit. lxmfy ohne Argumente öffnet ein interaktives Menü.

# Ein volles Projekt interaktiv scaffolden
lxmfy init mybot                 # Projektverzeichnis, bot.py, cogs/, README
lxmfy init --here --yes          # aktuelles Verzeichnis, alle Defaults annehmen

# Eine einzelne Bot-Datei erstellen
lxmfy create mybot
lxmfy create --template echo mybot
lxmfy create --template rrc my_rrc_bot
lxmfy create mybot --no-cogs     # das Cogs-Paket überspringen
lxmfy create --output dir/bot.py --name MyBot

# Einen Vorlagen-Bot ausführen
lxmfy run echo
lxmfy run rrc
lxmfy run reminder --name "MyReminder"

# Konnektivität diagnostizieren (siehe Abschnitt Diagnose)
lxmfy debug
lxmfy debug probe <hash> --request-path --wait 30

# Signaturprüfung mit einer Nachricht testen
lxmfy signatures test

# Signaturprüfung aktivieren
lxmfy signatures enable

# Signaturprüfung deaktivieren
lxmfy signatures disable

lxmfy init fragt nach Projektname, Vorlage, Speicher-Backend, Befehlspräfix und Admin-Hashes. Jede Frage akzeptiert stattdessen ein Flag: --dir, --bot-name, --template, --storage, --prefix, --admins, --no-cogs, --force, --yes. Auf nicht-TTY-stdin werden die Defaults genommen.

Vorlagen für create und run: basic, echo, reminder, note, cogtest, rrc.

Fehlerbehandlung¶

Shutdown- und Laufzeitfehler rund um bot.run() abfangen:

try:
    bot.run()
except KeyboardInterrupt:
    bot.cleanup()
except Exception as e:
    logger.error(f"Error running bot: {str(e)}")

Modulreferenz¶

Wird aus den Docstrings des Quellcodes generiert.

 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531

class LXMFBot(
    AnnounceMixin,
    DispatchMixin,
    CogMixin,
    InboundMixin,
    OutboundMixin,
    LinkMixin,
    RRCMixin,
    PropagationMixin,
):
    """Main bot class for handling LXMF messages and commands.

    This class manages the bot's lifecycle, including:
    - Message routing and delivery
    - Command registration and execution
    - Cog (extension) loading and management
    - Spam protection
    - Admin privileges
    """

    def __init__(self, name: str | None = None, **kwargs: Any):
        """Initialize a new LXMFBot instance.

        Args:
            name: Optional bot name, same as the name config key.
            **kwargs: Override default configuration settings

        """
        if name is not None:
            kwargs["name"] = name
        self.config = BotConfig(**kwargs)
        self.commands = {}
        self.cogs = {}
        self.first_message_handlers = []
        self.message_handlers = []
        self.reaction_handlers = []
        self.delivery_callbacks = []
        self.receipts = []
        self._receive_lock = threading.Lock()
        self._delivery_lock = threading.Lock()
        self._last_pending_flush = 0.0
        queue_size = max(1, int(getattr(self.config, "message_queue_size", 50) or 50))
        self.queue = Queue(maxsize=queue_size)
        self.announce_time = 600
        self.router = None
        self.local = None
        self.logger = logging.getLogger("lxmfy")
        self._configure_logging()
        self.thread_pool = ThreadPoolExecutor(
            max_workers=5,
        )  # For offloading CPU-bound or blocking I/O tasks
        self.scheduler = TaskScheduler(self)  # Initialize the scheduler

        if self.config.config_path:
            self.config_path = self.config.config_path
        else:
            self.config_path = os.path.join(os.getcwd(), "config")

        os.makedirs(self.config_path, exist_ok=True)
        if self.config.test_mode and not self.config.reticulum_config_dir:
            self.reticulum_config_dir = os.path.abspath(self.config_path)
        else:
            self.reticulum_config_dir = resolve_reticulum_config_dir(
                self.config.reticulum_config_dir,
                self.config_path,
            )
        os.makedirs(self.reticulum_config_dir, exist_ok=True)
        if not self.config.test_mode and is_isolated_reticulum_dir(
            self.reticulum_config_dir,
            self.config_path,
        ):
            ensure_isolated_share_instance_disabled(self.reticulum_config_dir)

        if self.config.storage_type == "json":
            self.storage = Storage(JSONStorage(self.config.storage_path))
        elif self.config.storage_type == "sqlite":
            self.storage = Storage(SQLiteStorage(self.config.storage_path))
        elif self.config.storage_type == "msgpack":
            self.storage = Storage(MsgPackStorage(self.config.storage_path))
        elif self.config.storage_type == "memory":
            self.storage = Storage(MemoryStorage())
        else:
            raise ValueError(
                f"Unknown storage_type {self.config.storage_type!r}; "
                "expected 'json', 'sqlite', 'msgpack', or 'memory'",
            )

        self.admins = set(self.config.admins or [])
        self.permissions = PermissionManager(
            storage=self.storage,
            enabled=self.config.permissions_enabled,
            admins=self.admins,
        )

        self.events = EventManager(self.storage)

        self._register_builtin_events()

        self.middleware = MiddlewareManager()

        self.cogs_dir = os.path.join(self.config_path, self.config.cogs_dir)
        if self.config.cogs_enabled:
            os.makedirs(self.cogs_dir, exist_ok=True)
            init_file = os.path.join(self.cogs_dir, "__init__.py")
            if not os.path.exists(init_file):
                open(init_file, "w", encoding="utf-8").close()

        self.transport = Transport(self, self.storage)
        self.delivery = DeliveryTracker(storage=self.storage)
        self.delivery.load_persisted()
        self.spam_protection = SpamProtection(
            storage=self.storage,
            bot=self,
            rate_limit=self.config.rate_limit,
            cooldown=self.config.cooldown,
            max_warnings=self.config.max_warnings,
            warning_timeout=self.config.warning_timeout,
        )

        self.delivery_attempts = {}
        self._load_delivery_attempts()
        self.conversations = ConversationManager(self)

        self.landlock_active = False
        if not self.config.test_mode:
            storage_dir = os.path.abspath(
                os.path.expanduser(self.config.storage_path),
            )
            self.landlock_active = apply_landlock_sandbox(
                storage_dir=storage_dir,
                reticulum_config_dir=self.reticulum_config_dir,
                config_dir=self.config_path,
                cogs_dir=self.cogs_dir,
                config_enabled=self.config.landlock_enabled,
            )

        identity_file = os.path.join(self.config_path, "identity")

        self._owns_reticulum = False
        if not self.config.test_mode:
            # Initialize Reticulum (will raise exception if already running)
            if RNS.Reticulum.get_instance() is None:
                try:
                    RNS.Reticulum(
                        configdir=self.reticulum_config_dir,
                        loglevel=self.config.loglevel,
                    )
                    self._owns_reticulum = True
                except OSError:
                    if RNS.Reticulum.get_instance() is None:
                        raise

            if not os.path.isfile(identity_file):
                RNS.log("No Primary Identity file found, creating new...", RNS.LOG_INFO)
                identity = RNS.Identity(True)
                identity.to_file(identity_file)
            identity = RNS.Identity.from_file(identity_file)
            if identity is None:
                raise RuntimeError(f"Failed to load identity from {identity_file}")
            self.identity = identity
            RNS.log("Loaded identity from file", RNS.LOG_INFO)

            self.router = LXMRouter(
                identity=self.identity,
                storagepath=self.config_path,
                autopeer=self.config.autopeer_propagation,
                autopeer_maxdepth=self.config.autopeer_maxdepth,
                enforce_stamps=self.config.require_stamps,
            )
            self.local = self.router.register_delivery_identity(
                self.identity,
                display_name=self.config.name,
                stamp_cost=self.config.stamp_cost,
            )
            if self.local is None:
                raise RuntimeError("Failed to register delivery identity")
            self._sync_delivery_display_name()
            self.router.register_delivery_callback(self._message_received)
            self.local.set_link_established_callback(
                self._delivery_link_established,
            )

            if self.config.pending_sends_enabled:
                RNS.Transport.register_announce_handler(
                    PendingSendAnnounceHandler(self),
                )

        self._configure_propagation()

        if self.local:
            RNS.log(
                f"LXMF Router ready to receive on: {RNS.prettyhexrep(self.local.hash)}",
                RNS.LOG_INFO,
            )
        else:
            # Test mode - create mock components
            if os.path.isfile(identity_file):
                identity = RNS.Identity.from_file(identity_file)
                if identity is None:
                    raise RuntimeError(
                        f"Failed to load identity from {identity_file}",
                    )
                self.identity = identity
            else:
                self.identity = RNS.Identity()  # Create a basic identity for testing
                if self.config.config_path:
                    self.identity.to_file(identity_file)

            self.router = None
            self.local = None

        self.announce_enabled = self.config.announce_enabled
        self.announce_time = self.config.announce

        if self.announce_enabled and not self.config.test_mode:
            if self.announce_time > 0:
                # Schedule the announce task. announce_now throttles on the
                # announce interval file, so a sub-minute cron step still
                # respects announce_time.
                minutes = max(1, self.announce_time // 60)
                self.scheduler.add_task(
                    "announce_task",
                    self.announce_now,
                    f"*/{minutes} * * * *",  # Convert seconds to minutes for cron
                )
            if self.config.announce_immediately:
                self.announce_now(force=True)
                RNS.log("Initial announce sent", RNS.LOG_INFO)

        self.hot_reloading = self.config.hot_reloading
        self.command_prefix = self.config.command_prefix

        self.help_system = HelpSystem(self)
        register_admin_commands(self)

        self.nlp = IntentClassifier(threshold=self.config.nlp_threshold)
        self.intents = {}  # {intent_name: callback}

        self.link_handlers = []
        self.links = {}  # {dest_hash: Link}

        self.rrc = None
        self.rrc_handlers = []
        self._init_rrc()

        self.signature_manager = SignatureManager(
            self,
            verification_enabled=self.config.signature_verification_enabled,
            require_signatures=self.config.require_message_signatures,
            request_unknown_identities=self.config.request_unknown_identities,
        )

        self._load_persisted_queue()

        if self.config.cogs_enabled:
            load_cogs_from_directory(self)

    def _configure_logging(self) -> None:
        level = self.config.log_level
        if level is None:
            return
        if isinstance(level, str):
            level = logging.getLevelNamesMapping().get(level.upper())
            if level is None:
                raise ValueError(
                    f"Unknown log_level {self.config.log_level!r}",
                )
        logger = logging.getLogger("lxmfy")
        if not logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(
                logging.Formatter(
                    "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
                    "%H:%M:%S",
                ),
            )
            logger.addHandler(handler)
        logger.setLevel(level)

    def get_debugger(self):
        """Return a Debugger bound to this bot."""
        from .debugger import Debugger

        return Debugger(bot=self)

    def diagnose_destination(
        self,
        destination: str,
        *,
        request_path: bool = False,
        wait: float = 0.0,
    ) -> dict:
        """Probe path and identity for a destination hash.

        Args:
            destination: Hex destination hash.
            request_path: Whether to request a path if missing.
            wait: Seconds to wait for a path after requesting.

        Returns:
            Dict describing identity/path status and hints.

        """
        return (
            self.get_debugger()
            .probe_destination(
                destination,
                request_path=request_path,
                wait=wait,
            )
            .to_dict()
        )

    def diagnose_connectivity(
        self,
        destination: str | None = None,
        *,
        request_path: bool = False,
        wait: float = 0.0,
    ) -> dict:
        """Run a connectivity doctor report for this bot.

        Args:
            destination: Optional peer hash to include in send diagnosis.
            request_path: Request a path when probing destination.
            wait: Seconds to wait for path discovery.

        Returns:
            Structured doctor report dict.

        """
        return (
            self.get_debugger()
            .run_doctor(
                destination,
                request_path=request_path,
                wait=wait,
            )
            .to_dict()
        )

    def on_delivery_event(self, callback=None):
        """Subscribe to the outbound delivery event stream.

        Usable as bot.on_delivery_event(fn) or as a decorator
        @bot.on_delivery_event(). Subscribers receive event dicts with
        ts, stage (queued, deferred, dispatched, delivered, failed,
        cancelled, dropped), and destination/message_id/hash/reason
        fields when available.
        """
        if callback is None:
            return self.delivery.subscribe
        return self.delivery.subscribe(callback)

    def get_landlock_status(self) -> dict[str, bool]:
        """Return Landlock LSM sandbox availability and activation state."""
        return landlock_status_dict(
            active=self.landlock_active,
            config_enabled=self.config.landlock_enabled,
        )

    def run(self, delay=10):
        """Run the bot"""
        self.scheduler.start()  # Start the scheduler
        try:
            while True:
                # Process outgoing queue with a timeout to prevent hanging
                while not self.queue.empty():
                    try:
                        lxm = self.queue.get(block=False)
                    except Exception:
                        break
                    try:
                        if self.router:
                            self.router.handle_outbound(lxm)
                            self.delivery.record(
                                "dispatched",
                                destination=_delivery_dest_hex(lxm),
                                message_id=_delivery_id_hex(lxm),
                                hash=_delivery_hex(lxm),
                            )
                        self._persist_queue()
                    except Exception as e:
                        self.logger.exception(
                            "Outbound send failed, requeueing",
                        )
                        self.delivery.record(
                            "failed",
                            destination=_delivery_dest_hex(lxm),
                            reason=str(e)[:120],
                        )
                        if not self._enqueue_outbound(lxm):
                            self.logger.exception(
                                "Failed to requeue after outbound error"
                            )
                        break

                retry_interval = max(
                    0,
                    int(getattr(self.config, "pending_sends_retry", 300) or 0),
                )
                if retry_interval and (
                    time.time() - self._last_pending_flush >= retry_interval
                ):
                    self._last_pending_flush = time.time()
                    self._flush_pending_sends()

                time.sleep(delay)

        except KeyboardInterrupt:
            pass
        finally:
            self.cleanup()

    def request_page(
        self,
        destination_hash: str,
        page_path: str,
        field_data: dict | None = None,
    ) -> dict:
        """Request a page from a destination.

        Args:
            destination_hash: The destination hash.
            page_path: The path to the page.
            field_data: Optional field data to send with the request.

        Returns:
            The response from the destination.

        """
        try:
            dest_hash_bytes = bytes.fromhex(destination_hash)
            return self.transport.request_page(dest_hash_bytes, page_path, field_data)
        except Exception:
            self.logger.exception("Error requesting page")
            raise

    def cleanup(self):
        """Clean up resources."""
        RNS.log("Cleaning up LXMFBot...", RNS.LOG_DEBUG)
        try:
            self._persist_queue()
        except Exception:
            self.logger.exception("Failed to persist queue during cleanup")
        self.conversations.cancel_all()
        if hasattr(self, "rrc") and self.rrc:
            try:
                self.rrc.shutdown()
            except Exception:
                self.logger.exception("RRC shutdown failed")
        self.transport.cleanup()
        self.thread_pool.shutdown(wait=False)
        self.scheduler.stop()
        if hasattr(self, "router") and self.router:
            try:
                self.router.exit_handler()
            except Exception as e:
                self.logger.debug("Router exit handler failed: %s", e)

        # Ensure Reticulum exits cleanly, but only if this bot started it.
        # A shared instance may still be serving other bots or tests.
        if not self.config.test_mode and self._owns_reticulum:
            try:
                RNS.Reticulum.exit_handler()
            except Exception as e:
                self.logger.debug("Reticulum exit handler failed: %s", e)
        RNS.log("LXMFBot cleanup complete", RNS.LOG_DEBUG)

    def validate(self) -> str:
        """Run validation checks and return formatted results."""
        results = validate_bot(self)
        return format_validation_results(results)