Add advanced statistical and ML plugins to PatternLab

Introduces new plugins for advanced statistical tests (Dieharder, TestU01, DFT spectral, entropy), cryptographic structure detection (ECB, frequency, known constants), and machine learning-based anomaly/label detection (autoencoder, LSTM/GRU, classifier). Updates CLI to support these plugins and handle numpy/bytes in JSON output. Engine now gracefully skips missing/unregistered tests. Adds corresponding tests for new plugins and removes obsolete report/test_results JSON files.

Author: EdgeTypE

patternlab/cli.py

@@ -5,7 +5,24 @@
 import os
 from .engine import Engine
 from .plugin_api import serialize_testresult
-
+ 
+def _json_default(obj):
+    """Fallback serializer to convert numpy scalars/arrays and bytes into JSON-friendly types."""
+    try:
+        import numpy as _np  # type: ignore
+        if isinstance(obj, _np.generic):
+            return obj.item()
+        if isinstance(obj, _np.ndarray):
+            return obj.tolist()
+    except Exception:
+        pass
+    if isinstance(obj, (bytes, bytearray)):
+        try:
+            return bytes(obj).hex()
+        except Exception:
+            return str(obj)
+    return str(obj)
+ 
 try:
     import yaml  # optional dependency for YAML config files
 except Exception:
@@ -83,6 +100,21 @@ def analyze(input_file, output_file, xor_value, discover, config_path, profile,
             "fft_spectral",
             "autocorrelation",
             "linear_complexity",
+            # Advanced tests added by integration
+            "diehard_birthday_spacings",
+            "diehard_overlapping_sums",
+            "diehard_3d_spheres",
+            "testu01_smallcrush",
+            "dft_spectral_advanced",
+            "hurst_exponent",
+            # Crypto-specific analysis plugins (ECB/frequency/known-constants)
+            "ecb_detector",
+            "frequency_pattern",
+            "known_constants_search",
+            # Machine-learning based anomaly detection plugins
+            "lstm_gru_anomaly",
+            "autoencoder_anomaly",
+            "classifier_labeler",
         ]
 
         file_conf = {}
@@ -160,7 +192,8 @@ def analyze(input_file, output_file, xor_value, discover, config_path, profile,
 
         # Write output directly as JSON following the new schema
         with open(output_file, 'w', encoding='utf-8') as f:
-            json.dump(output, f, indent=2)
+            # Use custom default serializer to handle numpy types and bytes produced by plugins
+            json.dump(output, f, indent=2, default=_json_default)
 
         click.echo(f"Analysis complete. Results written to {output_file}")
 

patternlab/engine.py

@@ -589,7 +589,11 @@ def _analyze_impl(self, input_bytes: bytes, config: Dict[str, Any]) -> Dict[str,
                                 raw_results.append({"test_name": rem.get('name'), "status": "skipped", "reason": "budget_exhausted"})
                         break
 
-                tp = self._tests[c['name']]
+                # Allow missing/unregistered tests to be gracefully skipped instead of raising KeyError.
+                tp = self._tests.get(c['name'])
+                if tp is None:
+                    raw_results.append({"test_name": c.get('name'), "status": "skipped", "reason": "test_not_registered"})
+                    continue
                 reqs = getattr(tp, 'requires', []) or []
                 missing_reqs: List[str] = []
                 # Lazy caches for data views that may be expensive

patternlab/plugins/autoencoder_anomaly.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+"""
+Autoencoder tabanlı anomalisi tespiti plugin.
+
+- TestPlugin'den türetilir.
+- BytesView kullanır.
+- Hem batch (run) hem streaming (update/finalize) API'sini destekler.
+- Varsayılan olarak hızlı stub modu kullanır (use_stub=True).
+- Gerçek model için params['model_path'] ile keras modeli lazy-load eder.
+"""
+
+from typing import Dict, Any
+import time
+try:
+    from ..plugin_api import TestPlugin, TestResult, BytesView
+except Exception:
+    from patternlab.plugin_api import TestPlugin, TestResult, BytesView  # type: ignore
+
+
+class AutoencoderAnomalyPlugin(TestPlugin):
+    """
+    Params:
+      - model_path: optional path to saved keras autoencoder
+      - window_size: int (default 1024)
+      - downsample: int
+      - reconstruction_threshold: float (anomaly threshold)
+      - use_stub: bool (default True)
+      - max_buffer_bytes: int for streaming buffer
+    """
+
+    requires = []
+
+    def __init__(self):
+        self._buffer = bytearray()
+        self._model = None
+        self._model_path = None
+
+    def describe(self) -> str:
+        return "Autoencoder-based reconstruction anomaly detection (batch + streaming)."
+
+    def _load_model(self, model_path: str):
+        if self._model is not None and self._model_path == model_path:
+            return self._model
+        try:
+            import tensorflow as tf  # type: ignore
+            model = tf.keras.models.load_model(model_path)
+            self._model = model
+            self._model_path = model_path
+            return model
+        except Exception as e:
+            raise RuntimeError(f"failed_to_load_model:{e}")
+
+    def _bytes_to_series(self, b: BytesView, window_size: int, downsample: int = 1):
+        mv = b.data
+        arr = mv.tobytes()
+        if downsample <= 1:
+            series = [float(x) / 255.0 for x in arr[:window_size]]
+        else:
+            series = []
+            step = downsample
+            for i in range(0, min(len(arr), window_size * step), step):
+                block = arr[i:i+step]
+                if not block:
+                    break
+                series.append(sum(block) / (len(block) * 255.0))
+        if len(series) < window_size:
+            series += [0.0] * (window_size - len(series))
+        else:
+            series = series[:window_size]
+        return series
+
+    def run(self, data: BytesView, params: Dict[str, Any]) -> TestResult:
+        start = time.time()
+        use_stub = params.get("use_stub", True)
+        window_size = int(params.get("window_size", min(len(data), 1024)))
+        downsample = int(params.get("downsample", 1))
+        threshold = float(params.get("reconstruction_threshold", 0.02))
+        inference_timeout_ms = int(params.get("inference_timeout_ms", 5000))
+
+        series = self._bytes_to_series(data, window_size, downsample)
+
+        if use_stub:
+            # Simple reconstruction via moving-average filter as stub
+            import statistics
+            smoothed = []
+            k = 5
+            for i in range(len(series)):
+                window = series[max(0, i-k+1):i+1]
+                smoothed.append(sum(window)/len(window))
+            mse = sum((a-b)**2 for a,b in zip(series, smoothed)) / (len(series) or 1)
+            passed = mse < threshold
+            tr = TestResult(
+                test_name="autoencoder_anomaly",
+                passed=bool(passed),
+                p_value=max(0.0, min(1.0, 1.0 - mse)),
+                category="ml_anomaly",
+                metrics={"reconstruction_mse": mse, "method": "stub"},
+                time_ms=(time.time()-start)*1000.0,
+                bytes_processed=len(data),
+            )
+            return tr
+
+        model_path = params.get("model_path")
+        if not model_path:
+            raise ValueError("model_path is required when use_stub=False")
+
+        model = self._load_model(model_path)
+
+        import numpy as np
+        x = np.array(series, dtype=np.float32).reshape((1, len(series), 1))
+        t0 = time.time()
+        preds = model.predict(x, verbose=0)
+        t1 = time.time()
+        elapsed_ms = (t1 - t0) * 1000.0
+        if elapsed_ms > inference_timeout_ms:
+            raise RuntimeError("inference_timeout")
+
+        # interpret as reconstruction
+        recon = preds.reshape(-1)[:len(series)]
+        mse = float(((recon - np.array(series))**2).mean())
+        passed = mse < threshold
+        tr = TestResult(
+            test_name="autoencoder_anomaly",
+            passed=bool(passed),
+            p_value=max(0.0, min(1.0, 1.0 - mse)),
+            category="ml_anomaly",
+            metrics={"reconstruction_mse": mse, "method": "model"},
+            time_ms=(time.time()-start)*1000.0,
+            bytes_processed=len(data),
+        )
+        return tr
+
+    def update(self, chunk: bytes, params: Dict[str, Any]) -> None:
+        max_buffer = int(params.get("max_buffer_bytes", 10 * 1024 * 1024))
+        if not isinstance(chunk, (bytes, bytearray)):
+            raise ValueError("chunk must be bytes")
+        self._buffer.extend(chunk)
+        if len(self._buffer) > max_buffer:
+            extra = len(self._buffer) - max_buffer
+            del self._buffer[:extra]
+
+    def finalize(self, params: Dict[str, Any]) -> TestResult:
+        bv = BytesView(bytes(self._buffer))
+        params = dict(params)
+        params.setdefault("use_stub", True)
+        tr = self.run(bv, params)
+        tr.metrics.setdefault("streaming", True)
+        return tr

patternlab/plugins/classifier_labeler.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+"""
+Classifier labeler plugin.
+
+- TestPlugin'den türetilir.
+- BytesView kullanır.
+- Streaming ve batch destekler.
+- Lazy-load scikit-learn modeller (joblib) veya stub heuristics.
+"""
+
+from typing import Dict, Any, Optional, List
+import time
+
+try:
+    from ..plugin_api import TestPlugin, TestResult, BytesView
+except Exception:
+    from patternlab.plugin_api import TestPlugin, TestResult, BytesView  # type: ignore
+
+class ClassifierLabelerPlugin(TestPlugin):
+    """
+    Params:
+      - model_path: optional path to sklearn joblib file
+      - labels: optional list of labels expected by the model (default common set)
+      - use_stub: bool (default True) -> quick heuristic
+      - inference_timeout_ms: int
+      - max_buffer_bytes: int for streaming
+    """
+
+    requires = []
+
+    def __init__(self):
+        self._buffer = bytearray()
+        self._model = None
+        self._model_path = None
+
+    def describe(self) -> str:
+        return "Classifier-based labeler (ransomware/encrypted/etc.) using pre-trained sklearn models."
+
+    def _load_model(self, model_path: str):
+        if self._model is not None and self._model_path == model_path:
+            return self._model
+        try:
+            import joblib  # type: ignore
+            model = joblib.load(model_path)
+            # Ensure model supports predict and predict_proba
+            if not hasattr(model, "predict"):
+                raise RuntimeError("model_missing_predict")
+            self._model = model
+            self._model_path = model_path
+            return model
+        except Exception as e:
+            raise RuntimeError(f"failed_to_load_model:{e}")
+
+    def _extract_features(self, b: BytesView, n_bins: int = 64):
+        # Simple, fast histogram of byte values normalized; suitable for sklearn classifiers
+        mv = b.data
+        arr = mv.tobytes()
+        bins = [0] * n_bins
+        if arr:
+            step = 256 // n_bins
+            for v in arr:
+                idx = min(n_bins - 1, v // step)
+                bins[idx] += 1
+            total = len(arr)
+            features = [c / total for c in bins]
+        else:
+            features = [0.0] * n_bins
+        return features
+
+    def run(self, data: BytesView, params: Dict[str, Any]) -> TestResult:
+        start = time.time()
+        use_stub = params.get("use_stub", True)
+        model_path = params.get("model_path")
+        labels: Optional[List[str]] = params.get("labels")
+        inference_timeout_ms = int(params.get("inference_timeout_ms", 2000))
+
+        features = self._extract_features(data, n_bins=int(params.get("n_bins", 64)))
+
+        if use_stub:
+            # Heuristic: high entropy -> likely encrypted; high repetition of small set -> maybe ransomware marker
+            import math, statistics
+            entropy = 0.0
+            for p in features:
+                if p > 0:
+                    entropy -= p * math.log2(p)
+            # simple thresholds (heuristic)
+            is_encrypted = entropy > 7.0
+            is_ransomware = features[0] > 0.2 and features[255 // max(1, (256//len(features)))] > 0.01 if len(features) > 1 else False
+            detected = "encrypted" if is_encrypted else ("ransomware" if is_ransomware else "unknown")
+            p_value = min(1.0, entropy / 8.0)
+            tr = TestResult(
+                test_name="classifier_labeler",
+                passed= not (detected in ("encrypted","ransomware")),
+                p_value=float(p_value),
+                category="ml_label",
+                metrics={"label": detected, "entropy": entropy, "method": "stub"},
+                time_ms=(time.time()-start)*1000.0,
+                bytes_processed=len(data),
+            )
+            return tr
+
+        if not model_path:
+            raise ValueError("model_path is required when use_stub=False")
+
+        model = self._load_model(model_path)
+
+        import numpy as np
+        x = np.array(features, dtype=np.float32).reshape((1, -1))
+
+        t0 = time.time()
+        # predict_proba may not be available for all models; fallback to decision_function or predict
+        probs = None
+        label = None
+        try:
+            if hasattr(model, "predict_proba"):
+                probs = model.predict_proba(x)[0].tolist()
+                classes = getattr(model, "classes_", None)
+                if classes is None:
+                    classes = list(range(len(probs)))
+                # choose top
+                top_idx = int(np.argmax(probs))
+                label = str(classes[top_idx])
+            else:
+                pred = model.predict(x)[0]
+                label = str(pred)
+                probs = None
+        except Exception as e:
+            raise RuntimeError(f"inference_failed:{e}")
+        t1 = time.time()
+        elapsed_ms = (t1 - t0) * 1000.0
+        if elapsed_ms > inference_timeout_ms:
+            raise RuntimeError("inference_timeout")
+
+        metrics = {"label": label, "method": "model"}
+        if probs is not None:
+            metrics["probabilities"] = probs
+        tr = TestResult(
+            test_name="classifier_labeler",
+            passed=(label not in ("ransomware","encrypted")),
+            p_value=(1.0 - max(probs)) if probs is not None else None,
+            category="ml_label",
+            metrics=metrics,
+            time_ms=(time.time()-start)*1000.0,
+            bytes_processed=len(data),
+        )
+        return tr
+
+    def update(self, chunk: bytes, params: Dict[str, Any]) -> None:
+        max_buffer = int(params.get("max_buffer_bytes", 10 * 1024 * 1024))
+        if not isinstance(chunk, (bytes, bytearray)):
+            raise ValueError("chunk must be bytes")
+        self._buffer.extend(chunk)
+        if len(self._buffer) > max_buffer:
+            extra = len(self._buffer) - max_buffer
+            del self._buffer[:extra]
+
+    def finalize(self, params: Dict[str, Any]) -> TestResult:
+        bv = BytesView(bytes(self._buffer))
+        params = dict(params)
+        params.setdefault("use_stub", True)
+        tr = self.run(bv, params)
+        tr.metrics.setdefault("streaming", True)
+        return tr