""" Network Intrusion Detection System (NIDS) Machine Learning Implementation using Random Forest Project focus: - Dataset: CICIDS2017 or any labelled network-flow CSV dataset - Task: Classify network traffic as BENIGN / ATTACK or multiclass attack labels - Model: Random Forest Classifier - Metrics: Accuracy, Precision, Recall, F1-score, Confusion Matrix, False Positive Rate Required libraries: pip install pandas numpy scikit-learn matplotlib seaborn joblib Google Colab pip install pandas numpy scikit-learn matplotlib seaborn (library run) https://www.unb.ca/cic/datasets/ids-2017.html (data source) """ import os import warnings from pathlib import Path import joblib import matplotlib.pyplot as plt import numpy as np import pandas as pd import seaborn as sns from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import ( accuracy_score, classification_report, confusion_matrix, f1_score, precision_score, recall_score, ) from sklearn.model_selection import train_test_split from sklearn.preprocessing import LabelEncoder, StandardScaler warnings.filterwarnings("ignore") # ========================= # CONFIGURATION # ========================= DATASET_PATH = "combinenew.csv" MODEL_OUTPUT_PATH = "nids_random_forest_model.pkl" SCALER_OUTPUT_PATH = "nids_scaler.pkl" TEST_SIZE = 0.20 RANDOM_STATE = 42 # ========================= # DATA LOADING # ========================= def load_dataset(path: str) -> pd.DataFrame: """Load dataset from a CSV file.""" if not os.path.exists(path): raise FileNotFoundError( f"Dataset file not found: {path}\n" "Please place the CSV file in this folder or update DATASET_PATH." ) df = pd.read_csv(path) print(f"[INFO] Dataset loaded successfully: {df.shape[0]} rows, {df.shape[1]} columns") return df # ========================= # PREPROCESSING # ========================= def clean_dataset(df: pd.DataFrame) -> pd.DataFrame: """Clean missing values, infinite values, duplicates, and column names.""" df = df.copy() # Remove spaces from column names df.columns = df.columns.str.strip() # Replace infinite values with NaN df.replace([np.inf, -np.inf], np.nan, inplace=True) # Drop rows with missing values df.dropna(inplace=True) # Drop duplicate records df.drop_duplicates(inplace=True) print(f"[INFO] Dataset after cleaning: {df.shape[0]} rows, {df.shape[1]} columns") return df def find_label_column(df: pd.DataFrame) -> str: """Automatically identify the label column.""" possible_labels = ["Label", "label", "Class", "class", "Attack", "attack"] for col in possible_labels: if col in df.columns: return col raise ValueError( "Could not find label column. Rename your target column to 'Label' " "or update find_label_column()." ) def prepare_features_and_labels(df: pd.DataFrame): """Separate features and labels, encode categorical fields, and scale features.""" label_col = find_label_column(df) X = df.drop(columns=[label_col]) y = df[label_col] # Convert categorical feature columns into numeric values for col in X.select_dtypes(include=["object"]).columns: encoder = LabelEncoder() X[col] = encoder.fit_transform(X[col].astype(str)) # Encode labels label_encoder = LabelEncoder() y_encoded = label_encoder.fit_transform(y.astype(str)) # Scale numeric features scaler = StandardScaler() X_scaled = scaler.fit_transform(X) print(f"[INFO] Label column detected: {label_col}") print(f"[INFO] Classes found: {list(label_encoder.classes_)}") return X_scaled, y_encoded, label_encoder, scaler, X.columns # ========================= # MODEL TRAINING # ========================= def train_model(X_train, y_train) -> RandomForestClassifier: """Train Random Forest classifier.""" model = RandomForestClassifier( n_estimators=100, random_state=RANDOM_STATE, n_jobs=-1, class_weight="balanced" ) model.fit(X_train, y_train) print("[INFO] Random Forest model trained successfully") return model # ========================= # EVALUATION # ========================= def calculate_false_positive_rate(cm: np.ndarray) -> float: """ Calculate false positive rate for binary classification. For multiclass classification, returns macro-average FPR. """ if cm.shape == (2, 2): tn, fp, fn, tp = cm.ravel() return fp / (fp + tn) if (fp + tn) > 0 else 0.0 fpr_values = [] for i in range(len(cm)): fp = cm[:, i].sum() - cm[i, i] tn = cm.sum() - (cm[i, :].sum() + cm[:, i].sum() - cm[i, i]) fpr_values.append(fp / (fp + tn) if (fp + tn) > 0 else 0.0) return float(np.mean(fpr_values)) def evaluate_model(model, X_test, y_test, label_encoder): """Evaluate trained model using key NIDS performance metrics.""" y_pred = model.predict(X_test) accuracy = accuracy_score(y_test, y_pred) precision = precision_score(y_test, y_pred, average="weighted", zero_division=0) recall = recall_score(y_test, y_pred, average="weighted", zero_division=0) f1 = f1_score(y_test, y_pred, average="weighted", zero_division=0) cm = confusion_matrix(y_test, y_pred) fpr = calculate_false_positive_rate(cm) print("\n========== MODEL EVALUATION RESULTS ==========") print(f"Accuracy: {accuracy:.4f}") print(f"Precision: {precision:.4f}") print(f"Recall: {recall:.4f}") print(f"F1-Score: {f1:.4f}") print(f"False Positive Rate: {fpr:.4f}") print("\n========== CLASSIFICATION REPORT ==========") print(classification_report( y_test, y_pred, target_names=label_encoder.classes_, zero_division=0 )) plot_confusion_matrix(cm, label_encoder.classes_) def plot_confusion_matrix(cm, class_names): """Plot and save confusion matrix.""" plt.figure(figsize=(10, 7)) sns.heatmap( cm, annot=False, cmap="Blues", xticklabels=class_names, yticklabels=class_names ) plt.title("NIDS Confusion Matrix") plt.xlabel("Predicted Label") plt.ylabel("Actual Label") plt.tight_layout() plt.savefig("nids_confusion_matrix.png") plt.show() print("[INFO] Confusion matrix saved as nids_confusion_matrix.png") # ========================= # FEATURE IMPORTANCE # ========================= def save_feature_importance(model, feature_names): """Save feature importance ranking to CSV.""" importance_df = pd.DataFrame({ "Feature": feature_names, "Importance": model.feature_importances_ }).sort_values(by="Importance", ascending=False) importance_df.to_csv("nids_feature_importance.csv", index=False) print("[INFO] Feature importance saved as nids_feature_importance.csv") # ========================= # SINGLE PREDICTION # ========================= def predict_network_traffic(sample_data, model, scaler, label_encoder): """ Predict whether a single network traffic sample is normal or malicious. sample_data must be a 2D list or NumPy array with the same feature order used during training. """ sample_scaled = scaler.transform(sample_data) prediction = model.predict(sample_scaled) predicted_label = label_encoder.inverse_transform(prediction) return predicted_label[0] # ========================= # MAIN PROGRAM # ========================= def main(): print("========== INTELLIGENT NETWORK INTRUSION DETECTION SYSTEM ==========") # Step 1: Load dataset df = load_dataset(DATASET_PATH) # Step 2: Clean dataset df = clean_dataset(df) # Step 3: Prepare features and labels X, y, label_encoder, scaler, feature_names = prepare_features_and_labels(df) # Step 4: Split dataset into training and testing sets X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=TEST_SIZE, random_state=RANDOM_STATE, stratify=y ) print(f"[INFO] Training set size: {X_train.shape[0]}") print(f"[INFO] Testing set size: {X_test.shape[0]}") # Step 5: Train model model = train_model(X_train, y_train) # Step 6: Evaluate model evaluate_model(model, X_test, y_test, label_encoder) # Step 7: Save model and scaler joblib.dump(model, MODEL_OUTPUT_PATH) joblib.dump(scaler, SCALER_OUTPUT_PATH) print(f"[INFO] Model saved as {MODEL_OUTPUT_PATH}") print(f"[INFO] Scaler saved as {SCALER_OUTPUT_PATH}") # Step 8: Save feature importance save_feature_importance(model, feature_names) print("\n[INFO] NIDS implementation completed successfully") if __name__ == "__main__": main()