Add-On SDK

Übersicht

YADS-Module sind einfache Python-Klassen, die von BaseScannerModule erben. Du implementierst genau eine Methode — run_scan() — die einen Domain-Namen entgegennimmt und ein Python-Dict zurückgibt. YADS übernimmt alles andere: Scheduling, Persistenz, Change Detection, Hashing, UI-Darstellung, Findings-Aggregation und Scan-Historie.

Ein Modul wird als ZIP-Datei bereitgestellt, die deine .py-Datei und eine module_manifest.json-Beschreibungsdatei enthält. Das ZIP wird einmalig über den Plugin Manager (System → Plugin Manager) hochgeladen — danach läuft das Modul automatisch bei jedem geplanten Scan jedes Targets im Tenant.

5-Minuten-Quickstart

Ein minimales Modul, das prüft ob ein Domain-Redirect auf /redirect offen ist:

1. Python-Datei erstellen — open_redirect_check.py
2. Manifest erstellen — module_manifest.json
3. Beide Dateien in ein ZIP packen
4. Upload via System → Plugin Manager und bestätigen

# open_redirect_check.py
import logging
import requests
from typing import Any, Dict, Optional
from yads.core.base import BaseScannerModule

logger = logging.getLogger(__name__)


class OpenRedirectCheck(BaseScannerModule):

    @property
    def module_name(self) -> str:
        return "open_redirect_check"

    def run_scan(self, target: str, target_id: Optional[int] = None) -> Dict[str, Any]:
        logger.info(f"[OpenRedirectCheck] Scan für {target}")
        url = f"https://{target}/redirect?url=https://evil.example.com"
        findings = []

        try:
            resp = requests.get(url, timeout=6, allow_redirects=False, verify=False)
            location = resp.headers.get("Location", "")
            if resp.status_code in (301, 302, 303, 307, 308) and "evil.example.com" in location:
                findings.append({
                    "title":       "Open Redirect gefunden",
                    "description": f"Endpunkt /redirect leitet auf externe Domain weiter: {location}",
                    "severity":    "medium",
                    "url":         url,
                })
        except requests.RequestException as e:
            logger.warning(f"[OpenRedirectCheck] Request fehlgeschlagen: {e}")

        return {
            "target":   target,
            "findings": findings,
            "summary":  {"findings_count": len(findings)},
        }

{
  "module_name": "open_redirect_check",
  "label":       "Open Redirect Check",
  "label_de":    "Open-Redirect-Prüfung",
  "version":     "1.0.0",
  "author":      "Dein Name",
  "description": "Prüft den /redirect-Endpunkt auf Open-Redirect-Schwachstellen.",
  "module_file": "open_redirect_check.py",
  "class_name":  "OpenRedirectCheck",
  "category":    "web",
  "passive":     false,
  "finding_module": true,
  "requires_https": true
}

zip open_redirect_check.zip open_redirect_check.py module_manifest.json

Danach: System → Plugin Manager in YADS öffnen, ZIP hineinziehen, Manifest-Vorschau prüfen, Installieren klicken.

So funktioniert es

In jedem Scan-Zyklus iteriert der YADS-Worker über alle aktiven Module eines Targets. Für jedes Modul wird process(target_id, domain) aufgerufen, das intern folgende Schritte ausführt:

1. Ruft dein run_scan() auf
2. Bereinigt Null-Bytes aus dem zurückgegebenen Dict (PostgreSQL-JSONB-Anforderung)
3. Berechnet einen SHA-256-Hash der kanonischen JSON-Darstellung
4. Schlägt den gespeicherten Hash in ModuleState für dieses Target/Modul-Paar nach
5. Bei geändertem Hash (oder erstem Scan): speichert einen neuen ScanResult-Eintrag und erzeugt ein ChangeEvent
6. Bei unverändertem Hash: aktualisiert nur den last_scanned_at-Zeitstempel

Du musst also nur korrekte Daten produzieren — die gesamte Persistenz, Deduplizierung, Change Detection und UI-Infrastruktur ist für dich erledigt.

BaseScannerModule

Alle Module müssen von BaseScannerModule erben — zu finden in yads.core.base.

from yads.core.base import BaseScannerModule

class MeinScanner(BaseScannerModule):

    @property
    def module_name(self) -> str:
        return "mein_scanner"          # muss mit module_manifest.json übereinstimmen

    def run_scan(self, target: str, target_id: Optional[int] = None) -> Dict[str, Any]:
        # Deine Scan-Logik hier
        return {"findings": [], "summary": {}}

Konstruktor

BaseScannerModule.__init__(self, db_session) speichert die SQLAlchemy-Session als self.db. In der Regel musst du das nicht überschreiben. Falls doch, ruf immer super().__init__(db_session) auf.

Abstrakte Member, die du implementieren musst

Vererbte Hilfsmittel

run_scan()

def run_scan(self, target: str, target_id: Optional[int] = None) -> Dict[str, Any]:

Parameter

Vertragsregeln

• Muss immer ein dict zurückgeben — niemals None oder absichtlich eine Exception werfen.
• Unbehandelte Exceptions werden von process() still geschluckt und auf stderr geloggt. Fang deine eigenen Exceptions ab und gibt stattdessen ein Fehler-Dict zurück.
• Das Dict wird als JSONB in PostgreSQL gespeichert. Vermeide Null-Bytes (\u0000) in String-Werten — verwende ggf. sanitize_null_bytes().
• Halte die Struktur zwischen Läufen stabil. Wenn sich die Daten nicht geändert haben, soll das zurückgegebene Dict identisch sein — das verhindert falsche Change Events.

Rückgabeformat

Du kannst ein beliebiges JSON-serialisierbares Dict zurückgeben. YADS hat keine strengen Schema-Vorgaben. Die folgenden konventionellen Schlüssel werden jedoch von UI, Findings-Aggregator und Report-Builder erkannt:

Minimales gültiges Rückgabe-Dict

return {
    "findings": [],
    "summary": {"findings_count": 0},
}

Vollständiges Beispiel

return {
    "reachable": True,
    "score":     72,
    "findings": [
        {
            "title":          "X-Content-Type-Options fehlt",
            "description":    "Der Header X-Content-Type-Options ist nicht gesetzt.",
            "severity":       "medium",
            "url":            "https://example.com/",
            "recommendation": "Hinzufügen: X-Content-Type-Options: nosniff",
        }
    ],
    "summary": {
        "score":           72,
        "findings_count":  1,
        "checked_headers": 8,
    },
    "headers": {"content-type": "text/html", "server": "nginx"},
}

Findings-Format

Findings sind der wichtigste Teil deiner Modul-Ausgabe. Sie erscheinen in der Security Findings-Liste, der Attack Surface-Heatmap und bilden die Grundlage für die Compliance-Gap-Analyse. Jedes Finding muss mindestens einen title enthalten.

Beispiel-Findings-Liste

"findings": [
    {
        "title":          "Exponiertes .git-Verzeichnis",
        "severity":       "high",
        "description":    "Das .git-Verzeichnis ist öffentlich zugänglich und gibt Quellcode und Commit-Verlauf preis.",
        "recommendation": "Zugriff auf /.git/ im Webserver blockieren.",
        "url":            "https://example.com/.git/config",
        "evidence":       "HTTP 200, Body: [core]\\n\\trepositoryformatversion = 0",
        "cwe":            "CWE-538",
    },
    {
        "title":    "HSTS-Header fehlt",
        "severity": "medium",
        "url":      "https://example.com/",
    },
]

process() — intern

Du rufst process() nicht selbst auf und überschreibst es nicht — aber ein Verständnis hilft beim Design stabiler Module.

# Vereinfachte Darstellung von process():

raw_data = self.run_scan(domain, target_id=target_id)
raw_data = sanitize_null_bytes(raw_data)   # entfernt \u0000 aus allen Strings
new_hash = sha256(json.dumps(raw_data, sort_keys=True))

state = db.query(ModuleState).filter_by(target_id=target_id, module_name=self.module_name).first()

if state is None:
    # Erster Scan
    save_result(); emit_change_event("FIRST_SCAN")

elif state.last_result_hash != new_hash:
    # Daten seit letztem Scan geändert
    save_result(); emit_change_event("DATA_CHANGED")

else:
    # Keine Änderung — nur Timestamp aktualisieren
    state.last_scanned_at = now

Das sort_keys=True bei der Hash-Berechnung macht die Schlüsselreihenfolge im Dict irrelevant — nur Werte zählen. Die Reihenfolge von Listen spielt jedoch eine Rolle. Wenn dein Modul Listen zurückgibt, die in unterschiedlicher Reihenfolge auftreten können (z. B. DNS-Records, offene Ports), sortiere sie vor dem Zurückgeben.

# Gut: stabile Listen-Reihenfolge
return {
    "name_servers": sorted(name_servers),
    "open_ports":   sorted(open_ports),
    "findings":     sorted(findings, key=lambda f: f["title"]),
}

module_manifest.json

Jedes ZIP muss eine module_manifest.json im Stammverzeichnis enthalten. Der Plugin Manager liest sie beim Upload und speichert ihren Inhalt in der Datenbank.

Vollständiges Manifest-Beispiel

{
  "module_name":    "git_exposure_check",
  "label":          "Git Exposure Check",
  "label_de":       "Git-Expositions-Prüfung",
  "version":        "2.1.0",
  "author":         "ACME Security Team",
  "description":    "Prüft auf öffentlich zugängliche .git-Verzeichnisse und andere VCS-Artefakte.",
  "module_file":    "git_exposure_check.py",
  "class_name":     "GitExposureCheck",
  "category":       "exposure",
  "passive":        false,
  "finding_module": true,
  "requires_https": true,
  "default_on":     false
}

ZIP-Struktur

Das ZIP muss mindestens zwei Dateien auf Root-Ebene enthalten (kein Unterverzeichnis):

mein_modul.zip
├── mein_modul.py           # deine Scanner-Klasse
└── module_manifest.json    # Beschreibung

Zusätzliche Hilfsdateien sind erlaubt:

mein_modul.zip
├── mein_scanner.py
├── helpers.py              # gemeinsamer Utility-Code
├── signatures.json         # Datendateien des Scanners
└── module_manifest.json

Paketierungs-Befehle

# Unix / macOS
zip mein_modul.zip mein_scanner.py module_manifest.json

# Mit zusätzlichen Dateien
zip mein_modul.zip mein_scanner.py helpers.py signatures.json module_manifest.json

# Python One-Liner (plattformübergreifend)
python3 -c "import zipfile; z=zipfile.ZipFile('mein_modul.zip','w'); [z.write(f) for f in ['mein_scanner.py','module_manifest.json']]; z.close()"

Kategorien

Wähle die Kategorie, die am besten beschreibt, was dein Modul scannt. Sie bestimmt die Gruppe im Plugin Manager und im Scan-Typ-Selektor.

Beispiel: Passiver Scanner

Ein passiver Scanner sammelt Informationen, ohne Probes zu senden oder Alarme auszulösen. Dieses Beispiel prüft, ob die Domain eine gültige security.txt-Datei hat — ein einfaches HTTP GET.

# security_txt_check.py
import logging
import requests
import urllib3
from typing import Any, Dict, Optional
from yads.core.base import BaseScannerModule

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logger = logging.getLogger(__name__)

TIMEOUT = 8
REQUIRED_FIELDS = ["Contact", "Expires"]


class SecurityTxtChecker(BaseScannerModule):

    @property
    def module_name(self) -> str:
        return "security_txt_check"

    def run_scan(self, target: str, target_id: Optional[int] = None) -> Dict[str, Any]:
        logger.info(f"[SecurityTxtCheck] Scanne {target}")

        content, url_used = self._fetch(target)
        if content is None:
            return {
                "found": False,
                "url": url_used,
                "findings": [{
                    "title":          "security.txt nicht gefunden",
                    "severity":       "low",
                    "description":    "Keine security.txt unter /.well-known/security.txt oder /security.txt.",
                    "recommendation": "security.txt gemäß RFC 9116 erstellen, damit Sicherheitsforscher Schwachstellen melden können.",
                }],
                "summary": {"findings_count": 1},
            }

        fields = self._parse_fields(content)
        findings = self._validate(fields)
        score = max(0, 100 - len(findings) * 15)

        return {
            "found":    True,
            "url":     url_used,
            "fields":  fields,
            "findings": findings,
            "score":   score,
            "summary": {"score": score, "findings_count": len(findings)},
        }

    def _fetch(self, target: str):
        for path in ("/.well-known/security.txt", "/security.txt"):
            url = f"https://{target}{path}"
            try:
                r = requests.get(url, timeout=TIMEOUT, verify=False)
                if r.status_code == 200 and "Contact" in r.text:
                    return r.text, url
            except requests.RequestException:
                continue
        return None, f"https://{target}/.well-known/security.txt"

    def _parse_fields(self, content: str) -> Dict[str, str]:
        fields = {}
        for line in content.splitlines():
            if ":" in line and not line.startswith("#"):
                key, _, val = line.partition(":")
                fields[key.strip()] = val.strip()
        return fields

    def _validate(self, fields: Dict[str, str]):
        findings = []
        for f in REQUIRED_FIELDS:
            if f not in fields:
                findings.append({
                    "title":          f"security.txt: Pflichtfeld fehlt: {f}",
                    "severity":       "medium",
                    "recommendation": f"Feld '{f}:' zur security.txt hinzufügen (RFC 9116)",
                })
        return findings

Beispiel: Aktiver Scanner mit Findings

Ein aktiver Scanner sendet gezielte Probes. Dieses Beispiel prüft mehrere bekannte Pfade auf exponierte Admin-Panels, Backup-Dateien und Debug-Endpoints.

# sensitive_path_scanner.py
import logging, requests, urllib3
from typing import Any, Dict, List, Optional
from yads.core.base import BaseScannerModule

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logger = logging.getLogger(__name__)
TIMEOUT = 6

# (Pfad, Beschreibung, Schweregrad)
SENSITIVE_PATHS = [
    ("/.env",           "Environment-Datei",          "critical"),
    ("/phpinfo.php",    "PHP-Info-Seite",              "high"),
    ("/server-status", "Apache Server-Status",        "high"),
    ("/actuator/env",  "Spring Boot Env-Endpoint",    "critical"),
    ("/backup.zip",    "Backup-Archiv",               "critical"),
    ("/debug",         "Debug-Endpoint",              "high"),
    ("/admin/",        "Admin-Panel",                 "medium"),
]

class SensitivePathScanner(BaseScannerModule):

    @property
    def module_name(self) -> str:
        return "sensitive_path_scanner"

    def run_scan(self, target: str, target_id: Optional[int] = None) -> Dict[str, Any]:
        logger.info(f"[SensitivePaths] Starte für {target} — {len(SENSITIVE_PATHS)} Pfade")
        findings, checked = [], []
        session = requests.Session()
        session.headers["User-Agent"] = "YADS-SecurityAudit/1.0"

        for path, desc, severity in SENSITIVE_PATHS:
            url = f"https://{target}{path}"
            try:
                resp = session.get(url, timeout=TIMEOUT, verify=False, allow_redirects=False)
                status = resp.status_code
                checked.append({"path": path, "status": status})
                if status in (200, 403):
                    eff_sev = severity if status == 200 else "low"
                    findings.append({
                        "title":          f"{desc} — {'Zugänglich' if status == 200 else 'Existiert (403)'}",
                        "severity":       eff_sev,
                        "description":    f"HTTP {status} auf {url}",
                        "url":            url,
                        "recommendation": f"Zugriff auf {path} einschränken oder entfernen.",
                    })
            except requests.RequestException:
                checked.append({"path": path, "status": None})

        findings.sort(key=lambda f: (f["severity"], f["url"]))
        checked.sort(key=lambda c: c["path"])
        weights = {"critical": 25, "high": 15, "medium": 8, "low": 3}
        score = max(0, 100 - sum(weights.get(f["severity"], 0) for f in findings))

        return {
            "checked":  checked,
            "findings": findings,
            "score":    score,
            "summary":  {"score": score, "findings_count": len(findings), "paths_checked": len(checked)},
        }

Beispiel: Externe API-Integration

Viele Module benötigen externe API-Keys (z. B. Shodan, VirusTotal, HIBP). Die empfohlene Methode ist, Keys als Umgebungsvariablen zu übergeben:

# virustotal_domain_check.py
import logging, os, requests
from typing import Any, Dict, Optional
from yads.core.base import BaseScannerModule

logger = logging.getLogger(__name__)
VT_API_BASE = "https://www.virustotal.com/api/v3"

class VirusTotalDomainCheck(BaseScannerModule):

    @property
    def module_name(self) -> str:
        return "virustotal_domain_check"

    def run_scan(self, target: str, target_id: Optional[int] = None) -> Dict[str, Any]:
        api_key = os.environ.get("VIRUSTOTAL_API_KEY")
        if not api_key:
            logger.warning("[VT] VIRUSTOTAL_API_KEY nicht gesetzt — übersprungen")
            return {"error": "VIRUSTOTAL_API_KEY nicht konfiguriert", "findings": []}

        logger.info(f"[VT] Abfrage für {target}")
        try:
            resp = requests.get(
                f"{VT_API_BASE}/domains/{target}",
                headers={"x-apikey": api_key},
                timeout=10,
            )
            resp.raise_for_status()
            data = resp.json()
        except requests.RequestException as e:
            return {"error": str(e), "findings": []}

        attrs = data.get("data", {}).get("attributes", {})
        malicious = attrs.get("last_analysis_stats", {}).get("malicious", 0)
        findings = []
        if malicious > 0:
            findings.append({
                "title":    f"Domain von {malicious} VirusTotal-Engine(s) markiert",
                "severity": "critical" if malicious > 3 else "high",
                "url":      f"https://www.virustotal.com/gui/domain/{target}",
            })
        return {"malicious": malicious, "findings": findings,
                "summary": {"findings_count": len(findings), "malicious": malicious}}

services:
  yads-worker:           # ← Worker läuft die Scans aus!
    environment:
      VIRUSTOTAL_API_KEY: "dein-key-hier"
  yads-api:
    environment:
      VIRUSTOTAL_API_KEY: "dein-key-hier"

Vollständiges Produktionsbeispiel

Ein komplettes, produktionsreifes Modul, das auf S3-Bucket-Fehlkonfigurationen prüft, indem es gängige Bucket-Namensmuster aus dem Target-Domain ableitet. Demonstriert: Logging, Fehlerbehandlung, stabile Output-Sortierung, Score-Berechnung, Null-Byte-Sicherheit.

# s3_bucket_probe.py
import logging, re, requests, urllib3
from typing import Any, Dict, List, Optional
from yads.core.base import BaseScannerModule, sanitize_null_bytes

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logger = logging.getLogger(__name__)
TIMEOUT = 8

def _derive_bucket_names(domain: str) -> List[str]:
    stem = re.split(r'[.-]', domain)[0]
    candidates = [
        stem, f"{stem}-assets", f"{stem}-backup", f"{stem}-prod",
        f"{stem}-dev", f"{stem}-staging", f"{stem}-data", f"{stem}-logs",
        domain.replace(".", "-"),
    ]
    return list(dict.fromkeys(candidates))

class S3BucketProbe(BaseScannerModule):

    @property
    def module_name(self) -> str:
        return "s3_bucket_probe"

    def run_scan(self, target: str, target_id: Optional[int] = None) -> Dict[str, Any]:
        candidates = _derive_bucket_names(target)
        logger.info(f"[S3Probe] {target} — {len(candidates)} Bucket-Kandidaten")
        findings, results = [], []

        for name in candidates:
            r = self._probe(name)
            results.append(r)
            if r["public_list"]:
                findings.append({
                    "title":          f"Öffentlicher S3-Bucket mit Directory-Listing: s3://{name}",
                    "severity":       "critical",
                    "description":    f"S3-Bucket '{name}' ist öffentlich zugänglich und erlaubt Directory-Listing.",
                    "url":            r["url"],
                    "recommendation": "Bucket-ACL auf Private setzen und Block Public Access aktivieren.",
                    "evidence":       r.get("snippet", ""),
                })
            elif r["exists"]:
                findings.append({
                    "title":    f"S3-Bucket existiert (Listing eingeschränkt): s3://{name}",
                    "severity": "low",
                    "url":      r["url"],
                })

        findings.sort(key=lambda f: (f["severity"], f["title"]))
        results.sort(key=lambda r: r["bucket"])
        weights = {"critical": 30, "high": 20, "medium": 10, "low": 5}
        score = max(0, 100 - sum(weights.get(f["severity"], 0) for f in findings))

        return sanitize_null_bytes({
            "target": target, "candidates": candidates, "results": results,
            "findings": findings, "score": score,
            "summary": {"score": score, "findings_count": len(findings),
                        "buckets_checked": len(candidates),
                        "public_buckets": sum(1 for r in results if r["public_list"])},
        })

    def _probe(self, name: str) -> Dict:
        url = f"https://{name}.s3.amazonaws.com/"
        try:
            resp = requests.get(url, timeout=TIMEOUT)
            pub = resp.status_code == 200 and "ListBucketResult" in resp.text
            return {"bucket": name, "url": url, "status": resp.status_code,
                    "exists": resp.status_code in (200, 403), "public_list": pub,
                    "snippet": resp.text[:300] if pub else ""}
        except requests.RequestException:
            return {"bucket": name, "url": url, "status": None, "exists": False, "public_list": False, "snippet": ""}

Logging

YADS erfasst Python-Logging-Ausgaben während Scans und streamt sie über Redis in die Scan-Logs-UI. Immer das Standard-logging-Modul verwenden — niemals print().

import logging
logger = logging.getLogger(__name__)

logger.info(f"[MeinModul] Scan für {target} gestartet")
logger.warning("[MeinModul] API-Rate-Limit erreicht, Ergebnisse möglicherweise unvollständig")
logger.error(f"[MeinModul] Unerwarteter Fehler: {e}")
logger.debug(f"[MeinModul] Response-Body: {resp.text[:200]}")

Log-Level

Null-Bytes & Encoding

PostgreSQL JSONB unterstützt keine Null-Bytes (\u0000 / \x00) in Text-Werten. Diese können in Banner-Grabs, binären Protokoll-Antworten oder schlecht codierten API-Payloads auftreten. YADS ruft sanitize_null_bytes() automatisch in process() auf. Für Defence-in-depth kann man es auch selbst aufrufen:

from yads.core.base import BaseScannerModule, sanitize_null_bytes

# Einzelnen String bereinigen:
clean = sanitize_null_bytes("banner: \x00\x00nginx")  # → "banner: nginx"

# Gesamtes Rückgabe-Dict bereinigen (rekursiv):
return sanitize_null_bytes({"data": raw_data})

# Binär / ungültiges UTF-8 behandeln:
safe_banner = raw_bytes.decode("utf-8", errors="replace")[:500]

Weitere Encoding-Tipps:

• Übermäßig lange String-Werte vor dem Speichern kürzen (z. B. Response-Bodies). Nur Zusammenfassungen oder die ersten N Zeichen speichern.
• Niemals rohe Binärdaten direkt im Result-Dict speichern. Base64-kodieren oder nur die relevanten Felder extrahieren.
• Alle Strings müssen gültiges UTF-8 sein. Ungültige Bytes vor dem Zurückgeben ersetzen oder ignorieren.

Passiv vs. aktiv

"passive": true im Manifest setzen für Module, die nur öffentlich verfügbare Daten lesen, ohne nicht-standardmäßige Requests zu senden. Das ist relevant für Tenants, die YADS in compliance-beschränkten Umgebungen betreiben, wo aktives Probing eine Change-Request erfordert.

Modul-Signierung (optional)

Für Produktionsumgebungen, in denen sichergestellt werden soll, dass nur genehmigte Module installiert werden können, unterstützt YADS Ed25519-Signatur-Verifizierung. Wenn MODULE_SIGNING_PUBLIC_KEY gesetzt ist, muss jedes hochgeladene ZIP eine gültige Signatur enthalten.

Schlüsselpaar generieren

python3 -c "
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, PrivateFormat, NoEncryption
import base64

key = Ed25519PrivateKey.generate()
priv = base64.b64encode(key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())).decode()
pub  = base64.b64encode(key.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo)).decode()
print('PRIVAT:', priv)
print('PUBLIC: ', pub)
"

ZIP signieren

# sign_module.py
import base64, hashlib, sys
from cryptography.hazmat.primitives.serialization import load_pem_private_key

zip_path = sys.argv[1]
key_b64  = sys.argv[2]

with open(zip_path, "rb") as f:
    content = f.read()

priv_key = load_pem_private_key(base64.b64decode(key_b64), password=None)
sig = base64.urlsafe_b64encode(priv_key.sign(hashlib.sha256(content).digest())).decode()
print(f"Signatur: {sig}")

Signierung in YADS erzwingen

services:
  yads-api:
    environment:
      MODULE_SIGNING_PUBLIC_KEY: "<dein base64-Public-Key>"

Lokales Testen

Module können außerhalb von YADS mit einem minimalen Stub getestet werden, der die Datenbank-Session ersetzt. Keine YADS-Installation erforderlich — nur Python und die Abhängigkeiten deines Moduls.

# test_mein_modul.py  — Aufruf:  python3 test_mein_modul.py example.com
import sys, json, types
from unittest.mock import MagicMock

# YADS-Basisklasse stubben
yads_core = types.ModuleType("yads.core.base")

class _FakeBase:
    def __init__(self, db): self.db = db

yads_core.BaseScannerModule = _FakeBase
yads_core.sanitize_null_bytes = lambda x: x
sys.modules["yads"]           = types.ModuleType("yads")
sys.modules["yads.core"]      = types.ModuleType("yads.core")
sys.modules["yads.core.base"] = yads_core

# Jetzt dein Modul importieren
from sensitive_path_scanner import SensitivePathScanner

if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else "example.com"
    scanner = SensitivePathScanner(db_session=MagicMock())
    result = scanner.run_scan(target)
    print(json.dumps(result, indent=2, default=str))

Installation via Plugin Manager

1. System → Plugin Manager in YADS öffnen. Dieser Menüpunkt ist nur für Platform-Administratoren sichtbar.
2. Sicherstellen, dass kein Tenant ausgewählt ist im Kontext-Switcher oben. Der Install-Button ist nur im Platform-Admin-Modus aktiv (kein Tenant-Kontext).
3. "Modul installieren" klicken und ZIP in den Upload-Bereich ziehen oder per Dateiauswahl hochladen. Das System validiert das ZIP und liest das Manifest.
4. Manifest-Vorschau prüfen. YADS zeigt Modulname, Version, Autor, Kategorie, Passiv/Aktiv-Badge und eventuelle Warnungen (z. B. wenn bereits ein Modul mit gleichem Namen existiert).
5. "Installieren" klicken zur Bestätigung. Die Moduldatei wird nach yads/modules/custom/ kopiert und in der Datenbank registriert.
6. Das Modul ist sofort aktiv. Es läuft beim nächsten geplanten Scan jedes Targets. Alternativ kann auch ein manueller Scan von der Target-Detailseite aus gestartet werden.

Modul aktualisieren

Ein neues ZIP mit dem gleichen module_name hochladen. Der Plugin Manager erkennt den Konflikt und zeigt eine Warnung in der Vorschau. Bei Bestätigung wird die bestehende Datei überschrieben und der Datenbankeintrag mit den neuen Versions-Metadaten aktualisiert.

Modul entfernen

Modul im Plugin Manager per Toggle-Switch deaktivieren. Zur dauerhaften Entfernung die .py-Datei aus yads/modules/custom/ löschen — YADS erkennt die fehlende Datei beim nächsten Start und markiert den Datenbankeintrag als inaktiv.