network‑monitor/
│
├─ .env                     # Global env vars (secrets are injected in CI)
├─ docker-compose.yml       # Orchestrates all services
│
├─ ingestion/
│   ├─ Dockerfile
│   └─ app.py
│
├─ preprocess/
│   ├─ Dockerfile
│   └─ preprocess.py
│
├─ anomaly/
│   ├─ Dockerfile
│   └─ model.py
│
├─ nlp/
│   ├─ Dockerfile
│   └─ risk_assess.py
│
├─ aggregator/
│   ├─ Dockerfile
│   └─ aggregator.py
│
├─ alert/
│   ├─ Dockerfile
│   └─ alert.py
│
├─ retrain/
│   ├─ Dockerfile
│   └─ retrain.py
│
└─ ci/
    ├─ github/
    │   └─ workflow.yml
    └─ jenkins/
        └─ Jenkinsfile

# ingestion/Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .
ENV PYTHONUNBUFFERED=1
CMD ["python", "app.py"]

# preprocess/Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY preprocess.py .
CMD ["python", "preprocess.py"]

# anomaly/Dockerfile
FROM python:3.11-slim

# Install system deps for torch (Ubuntu libs)
RUN apt-get update && apt-get install -y --no-install-recommends \
    libglib2.0-0 libgl1 && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY model.py .
CMD ["python", "model.py"]

# nlp/Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY risk_assess.py .
CMD ["python", "risk_assess.py"]

# aggregator/Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY aggregator.py .
CMD ["python", "aggregator.py"]

# alert/Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY alert.py .
CMD ["python", "alert.py"]

# retrain/Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY retrain.py .
CMD ["python", "retrain.py"]

# docker-compose.yml
version: "3.9"

services:
  # 1️⃣ Ingestion – pulls network telemetry from Kafka or syslog
  ingestion:
    build: ./ingestion
    container_name: ingestion
    env_file: .env
    volumes:
      - logs:/data/raw
    depends_on:
      - preprocess

  # 2️⃣ Preprocess – normalizes, filters, and enriches data
  preprocess:
    build: ./preprocess
    container_name: preprocess
    env_file: .env
    volumes:
      - logs:/data/raw
      - processed:/data/processed
    depends_on:
      - anomaly

  # 3️⃣ Anomaly – PyTorch model that outputs a score
  anomaly:
    build: ./anomaly
    container_name: anomaly
    env_file: .env
    volumes:
      - processed:/data/processed
      - anomalies:/data/anomalies
    depends_on:
      - nlp

  # 4️⃣ NLP – risk classification using a transformer
  nlp:
    build: ./nlp
    container_name: nlp
    env_file: .env
    volumes:
      - processed:/data/processed
      - nlp_out:/data/nlp
    depends_on:
      - aggregator

  # 5️⃣ Aggregator – merges scores, decides severity
  aggregator:
    build: ./aggregator
    container_name: aggregator
    env_file: .env
    volumes:
      - anomalies:/data/anomalies
      - nlp_out:/data/nlp
      - alerts:/data/alerts
    depends_on:
      - alert

  # 6️⃣ Alert – pushes alerts to OpenSearch & Slack
  alert:
    build: ./alert
    container_name: alert
    env_file: .env
    volumes:
      - alerts:/data/alerts
    depends_on:
      - opensearch

  # 7️⃣ Retraining – nightly job that pulls labeled data, retrains, and pushes new model
  retrain:
    build: ./retrain
    container_name: retrain
    env_file: .env
    volumes:
      - model:/model
    deploy:
      restart_policy:
        condition: none   # run‑once per schedule, see CI below

  # 8️⃣ OpenSearch – observability backbone
  opensearch:
    image: opensearchproject/opensearch:2.12.0
    container_name: opensearch
    environment:
      - discovery.type=single-node
      - OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m
      - plugins.security.disabled=true
    ulimits:
      memlock:
        soft: -1
        hard: -1
    ports:
      - "9200:9200"
      - "9600:9600"
    volumes:
      - opensearch-data:/usr/share/opensearch/data

volumes:
  logs:
  processed:
  anomalies:
  nlp_out:
  alerts:
  model:
  opensearch-data:

import os, time, json
import paho.mqtt.client as mqtt   # Example source, replace with Kafka if needed

BROKER = os.getenv("MQTT_BROKER", "mqtt://localhost")
TOPIC  = os.getenv("MQTT_TOPIC", "network/telemetry")
OUTDIR = "/data/raw"

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    ts = int(time.time())
    fname = f"{OUTDIR}/{ts}.json"
    with open(fname, "w") as f:
        json.dump({"topic": msg.topic, "payload": payload}, f)

client = mqtt.Client()
client.on_message = on_message
client.connect(BROKER)
client.subscribe(TOPIC)
client.loop_forever()

import os, json, glob
import pandas as pd

INDIR  = "/data/raw"
OUTDIR = "/data/processed"

def normalize(record):
    # Dummy normalization – real code would parse PCAP, extract fields, etc.
    data = json.loads(record)
    return {"ts": os.path.splitext(os.path.basename(record))[0],
            "source_ip": data.get("payload", "").split()[0],
            "bytes": len(data["payload"])}

def main():
    files = glob.glob(f"{INDIR}/*.json")
    rows = [normalize(f) for f in files]
    df = pd.DataFrame(rows)
    df.to_parquet(f"{OUTDIR}/batch.parquet", index=False)

if __name__ == "__main__":
    main()

import os, torch, pandas as pd
from torch import nn

MODEL_PATH = "/model/anomaly.pt"
INFILE = "/data/processed/batch.parquet"
OUTDIR = "/data/anomalies"

class SimpleAE(nn.Module):
    def __init__(self, dim):
        super().__init__()
        self.encoder = nn.Linear(dim, dim//2)
        self.decoder = nn.Linear(dim//2, dim)

    def forward(self, x):
        z = torch.relu(self.encoder(x))
        return torch.relu(self.decoder(z))

def load_model():
    dim = 10   # placeholder
    model = SimpleAE(dim)
    if os.path.exists(MODEL_PATH):
        model.load_state_dict(torch.load(MODEL_PATH))
    model.eval()
    return model

def infer():
    df = pd.read_parquet(INFILE)
    model = load_model()
    tensor = torch.tensor(df.values, dtype=torch.float32)
    recon = model(tensor)
    error = torch.mean((tensor - recon) ** 2, dim=1)
    df["anomaly_score"] = error.detach().numpy()
    df.to_csv(f"{OUTDIR}/scores.csv", index=False)

if __name__ == "__main__":
    infer()

import os, json, pandas as pd
from transformers import pipeline

MODEL_NAME = os.getenv("HF_MODEL", "distilbert-base-uncased-finetuned-sst-2-english")
INFILE = "/data/processed/batch.parquet"
OUTDIR = "/data/nlp"

sentiment = pipeline("sentiment-analysis", model=MODEL_NAME)

def classify(text):
    result = sentiment(text[:512])[0]   # truncate for speed
    return result["label"], result["score"]

def main():
    df = pd.read_parquet(INFILE)
    # Pretend each row has a 'payload' column with raw log text
    df["risk_label"], df["risk_score"] = zip(*df["payload"].apply(classify))
    df.to_csv(f"{OUTDIR}/risk.csv", index=False)

if __name__ == "__main__":
    main()

import os, pandas as pd

ANOM = "/data/anomalies/scores.csv"
NLP  = "/data/nlp/risk.csv"
OUT  = "/data/alerts/merged.csv"

def main():
    a = pd.read_csv(ANOM)
    n = pd.read_csv(NLP)
    merged = pd.merge(a, n, left_on="ts", right_on="ts")
    # Simple rule: if anomaly_score > 0.5 and risk_score > 0.7 → high severity
    merged["severity"] = merged.apply(
        lambda r: "high" if r.anomaly_score > 0.5 and r.risk_score > 0.7 else "medium",
        axis=1
    )
    merged.to_csv(OUT, index=False)

if __name__ == "__main__":
    main()

import os, json, pandas as pd, requests

OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://opensearch:9200")
INDEX = "network-anomalies"
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK")

def push_to_opensearch(df):
    bulk = ""
    for _, row in df.iterrows():
        action = {"index": {"_index": INDEX}}
        bulk += json.dumps(action) + "\n"
        bulk += json.dumps(row.to_dict()) + "\n"
    headers = {"Content-Type": "application/x-ndjson"}
    resp = requests.post(f"{OPENSEARCH_URL}/_bulk", data=bulk, headers=headers)
    resp.raise_for_status()

def slack_notify(df):
    high = df[df.severity == "high"]
    if high.empty:
        return
    text = f"*⚠️ High‑severity anomalies detected:*\\n"
    for _, r in high.iterrows():
        text += f"`{r.ts}` – score:{r.anomaly_score:.2f}, risk:{r.risk_label}\\n"
    payload = {"text": text}
    requests.post(SLACK_WEBHOOK, json=payload)

def main():
    df = pd.read_csv("/data/alerts/merged.csv")
    push_to_opensearch(df)
    slack_notify(df)

if __name__ == "__main__":
    main()

import os, torch, pandas as pd
from torch import nn

DATASET = "/data/labeled/training.parquet"
MODEL_PATH = "/model/anomaly.pt"

class SimpleAE(nn.Module):
    def __init__(self, dim):
        super().__init__()
        self.encoder = nn.Linear(dim, dim//2)
        self.decoder = nn.Linear(dim//2, dim)

    def forward(self, x):
        z = torch.relu(self.encoder(x))
        return torch.relu(self.decoder(z))

def train():
    df = pd.read_parquet(DATASET)
    X = torch.tensor(df.values, dtype=torch.float32)
    model = SimpleAE(X.shape[1])
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    loss_fn = nn.MSELoss()

    for epoch in range(5):
        optimizer.zero_grad()
        recon = model(X)
        loss = loss_fn(recon, X)
        loss.backward()
        optimizer.step()
        print(f"Epoch {epoch+1} – loss: {loss.item():.4f}")

    torch.save(model.state_dict(), MODEL_PATH)
    print("✅ New model saved")

if __name__ == "__main__":
    train()