app.py

"""Hugging Face Spaces app for NIDS deployment.

This app downloads pre-trained models from Hugging Face Hub and serves
a Gradio interface for real-time network intrusion detection.
"""

import os
import sys
import json
import numpy as np
import joblib
import gradio as gr

MODELS_REPO = "Alaudeen/nids-models"
MODELS_DIR = "outputs/models"
os.makedirs(MODELS_DIR, exist_ok=True)

# Download models from HF Hub if not present locally
models = {}


def download_models():
    """Download models from HF Hub if they don't exist locally."""
    try:
        from huggingface_hub import hf_hub_download
    except ImportError:
        print("huggingface_hub not installed, models must exist locally")
        return

    model_files = [
        "XGBoost.joblib",
        "RandomForest.joblib",
        "IsolationForest_Unsupervised.joblib",
        "MLP.pt",
        "LSTM.pt",
        "Transformer.pt",
        "Autoencoder.pt"
    ]

    for fname in model_files:
        local_path = os.path.join(MODELS_DIR, fname)
        if not os.path.exists(local_path):
            try:
                print(f"Downloading {fname} from {MODELS_REPO}...")
                hf_hub_download(
                    repo_id=MODELS_REPO,
                    filename=fname,
                    repo_type="model",
                    local_dir=MODELS_DIR,
                    local_dir_use_symlinks=False
                )
                print(f"  Downloaded: {fname}")
            except Exception as e:
                print(f"  Failed to download {fname}: {e}")


def load_models():
    """Load all available models."""
    global models
    models.clear()
    for fname in sorted(os.listdir(MODELS_DIR)):
        path = os.path.join(MODELS_DIR, fname)
        if fname.endswith(".joblib"):
            name = fname.replace(".joblib", "")
            try:
                models[name] = joblib.load(path)
                print(f"Loaded: {name}")
            except Exception as e:
                print(f"Failed to load {name}: {e}")
        elif fname.endswith(".pt"):
            name = fname.replace(".pt", "")
            models[name] = path  # Store path, load on-demand
            print(f"Found: {name}")


# Download and load
print("Initializing NIDS Space...")
download_models()
load_models()
print(f"Models available: {list(models.keys())}")

# Sample flows
SAMPLE_NORMAL = [0, 1, 45, 0, 491, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
                 1, 1, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 2, 2, 0.0, 0.0,
                 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.17, 0.03]

SAMPLE_ANOMALY = [0, 1, 44, 4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
                  1, 1, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 123, 6, 1.0, 1.0,
                  0.0, 0.0, 0.05, 0.07, 0.0, 0.0, 0.1, 0.05]

# Pad to 41
SAMPLE_NORMAL = SAMPLE_NORMAL[:41]
SAMPLE_ANOMALY = SAMPLE_ANOMALY[:41]


def detect_single(features_text: str, model_name: str) -> str:
    """Detect intrusion on a single flow."""
    try:
        features = [float(x.strip()) for x in features_text.split(",")]
    except Exception as e:
        return f"❌ Error parsing features: {e}"

    if len(features) != 41:
        return f"❌ Expected 41 features, got {len(features)}"

    if model_name not in models:
        return f"❌ Model '{model_name}' not available. Loaded: {list(models.keys())}"

    # ML models (joblib)
    if model_name in models and hasattr(models[model_name], 'predict'):
        model = models[model_name]
        X = np.array(features).reshape(1, -1)
        pred = int(model.predict(X)[0])
        proba = model.predict_proba(X)[0] if hasattr(model, "predict_proba") else [0.5, 0.5]
        confidence = float(proba[pred])
    else:
        return f"⚠️ Model {model_name} not loaded (path only)"

    if pred == 0:
        level = "✅ SAFE — Normal Traffic"
    elif confidence > 0.9:
        level = f"🔴 CRITICAL THREAT — Anomaly Detected (Confidence: {confidence:.1%})"
    elif confidence > 0.75:
        level = f"🟠 HIGH THREAT — Anomaly Detected (Confidence: {confidence:.1%})"
    else:
        level = f"🟡 MEDIUM THREAT — Suspicious Activity (Confidence: {confidence:.1%})"

    return level


def detect_batch(batch_text: str, model_name: str) -> str:
    """Batch detection on multiple flows."""
    lines = [l.strip() for l in batch_text.strip().split("\n") if l.strip()]
    flows = []
    for line in lines:
        try:
            vals = [float(x.strip()) for x in line.split(",")]
            if len(vals) == 41:
                flows.append(vals)
        except:
            continue

    if not flows:
        return "❌ No valid 41-feature flows found."

    if model_name not in models or not hasattr(models[model_name], 'predict'):
        return f"❌ Model '{model_name}' not available."

    model = models[model_name]
    X = np.array(flows)
    preds = model.predict(X)

    normals = int(sum(preds == 0))
    anomalies = int(sum(preds == 1))

    return (
        f"**Batch Detection Results**\n\n"
        f"- Total Flows: {len(flows)}\n"
        f"- ✅ Normal: {normals} ({normals/len(flows)*100:.1f}%)\n"
        f"- 🚨 Anomalies: {anomalies} ({anomalies/len(flows)*100:.1f}%)\n"
        f"- Model: {model_name}"
    )


def show_results() -> str:
    """Show model performance table."""
    return """
## Model Performance (NSL-KDD Dataset)

| Model | Accuracy | Macro F1 | AUC-ROC | Type |
|-------|----------|----------|---------|------|
| **XGBoost** | 76.18% | 76.04% | 95.75% | Supervised |
| RandomForest | 73.10% | 73.05% | 95.34% | Supervised |
| MLP | 73.28% | 73.21% | 89.33% | Supervised |
| Autoencoder | 71.84% | 71.34% | 73.60% | Unsupervised |
| LSTM | 70.65% | 70.58% | 87.80% | Unsupervised |
| Transformer | 57.94% | 57.26% | 80.29% | Supervised |
| IsolationForest | 56.55% | 55.96% | 65.24% | Unsupervised |

**Key Insight:** XGBoost achieves the best performance (95.75% AUC-ROC) and runs at ~1ms latency per flow, making it ideal for real-time deployment.
"""


# Build Gradio interface
with gr.Blocks(title="🛡️ Network Intrusion Detection System") as demo:
    gr.Markdown("""
    # 🛡️ Network Intrusion Detection System (NIDS)

    Detect network intrusions in real-time using ML models trained on the **NSL-KDD** dataset.
    Enter 41 comma-separated network flow features to classify as **Normal** or **Anomaly**.

    **Models from:** [Alaudeen/nids-models](https://huggingface.co/Alaudeen/nids-models)
    """)

    with gr.Tab("🔍 Single Flow Detection"):
        with gr.Row():
            with gr.Column(scale=2):
                feature_input = gr.Textbox(
                    label="Flow Features (41 comma-separated values)",
                    value=",".join(map(str, SAMPLE_ANOMALY)),
                    lines=2,
                    placeholder="Enter 41 NSL-KDD features..."
                )
                model_choice = gr.Dropdown(
                    choices=list(models.keys()) if models else ["XGBoost", "RandomForest"],
                    value="XGBoost",
                    label="Detection Model",
                    info="XGBoost is recommended (best accuracy + speed)"
                )
                detect_btn = gr.Button("🔍 Detect Intrusion", variant="primary", size="lg")

            with gr.Column(scale=1):
                result = gr.Textbox(
                    label="Detection Result",
                    lines=4,
                    interactive=False
                )
                gr.Markdown("""
                **Alert Levels:**
                - 🟢 **Safe** — Normal traffic
                - 🟡 **Medium** — Suspicious activity
                - 🟠 **High** — Likely intrusion
                - 🔴 **Critical** — Confirmed attack
                """)

        detect_btn.click(
            detect_single,
            inputs=[feature_input, model_choice],
            outputs=result
        )

        with gr.Row():
            gr.Button("📋 Load Normal Sample").click(
                lambda: ",".join(map(str, SAMPLE_NORMAL)),
                outputs=feature_input
            )
            gr.Button("⚠️ Load Anomaly Sample").click(
                lambda: ",".join(map(str, SAMPLE_ANOMALY)),
                outputs=feature_input
            )

    with gr.Tab("📊 Batch Detection"):
        with gr.Row():
            with gr.Column(scale=2):
                batch_input = gr.Textbox(
                    label="Batch Flows (one per line, 41 values each)",
                    value=",".join(map(str, SAMPLE_NORMAL)) + "\n" +
                          ",".join(map(str, SAMPLE_ANOMALY)),
                    lines=8
                )
                batch_model = gr.Dropdown(
                    choices=list(models.keys()) if models else ["XGBoost", "RandomForest"],
                    value="XGBoost",
                    label="Model"
                )
                batch_btn = gr.Button("📊 Batch Detect", variant="primary")
            with gr.Column(scale=1):
                batch_result = gr.Markdown(label="Results")

        batch_btn.click(detect_batch, inputs=[batch_input, batch_model], outputs=batch_result)

    with gr.Tab("📈 Model Performance"):
        gr.Markdown(show_results())

    with gr.Tab("📖 API Documentation"):
        gr.Markdown("""
        ## REST API Usage

        Deploy the FastAPI server locally:
        ```bash
        pip install fastapi uvicorn
        uvicorn api:app --host 0.0.0.0 --port 8000
        ```

        ### Endpoints

        | Endpoint | Method | Description |
        |----------|--------|-------------|
        | `/health` | GET | Health check |
        | `/models` | GET | List available models |
        | `/predict` | POST | Single flow detection |
        | `/predict/batch` | POST | Batch detection |
        | `/stats` | GET | Usage statistics |
        | `/sample` | GET | Sample flows |

        ### Example Request
        ```bash
        curl -X POST http://localhost:8000/predict \\
          -H "Content-Type: application/json" \\
          -d '{
            "features": [0,1,45,0,491,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1,1,0,0,0,0,1,0,0,0,2,2,0,0,0,0,1,0,0,0,0.17],
            "model": "XGBoost"
          }'
        ```

        ### Example Response
        ```json
        {
          "flow_id": "flow_1",
          "prediction": 1,
          "confidence": 0.9634,
          "model": "XGBoost",
          "latency_ms": 2.77,
          "alert_level": "critical",
          "timestamp": 1778206436.82
        }
        ```
        """)

    gr.Markdown("""
    ---
    **Project:** [github.com/Alaudeen/nids](https://huggingface.co/Alaudeen/nids-models) |
    **Dataset:** [Mireu-Lab/NSL-KDD](https://huggingface.co/datasets/Mireu-Lab/NSL-KDD) |
    **License:** MIT
    """)

if __name__ == "__main__":
    demo.launch()