پیاده‌سازی الگوریتم Echidna در پایتون برای خوشه‌بندی ترافیک شبکه

1. مقدمه

انتقال از فرمول‌های ریاضی و منطق درختی الگوریتم اکیدنا (Echidna) به یک سامانه نرم‌افزاری پویا، نیازمند درک دقیق نحوه بازنمایی ویژگی‌های ترکیبی (عددی، دسته‌ای و سلسله‌مراتبی) در حافظه رایانه است. در بخش نظری، فاصله ترکیبی را به عنوان پیوندی از سه سناریوی محاسباتی تعریف کردیم. در این بخش، با استفاده از زبان برنامه‌نویسی پایتون، ساختار کلاسی طراحی می‌کنیم که گره‌های درخت ویژگی خوشه‌ای (CF-Tree) را مدیریت کرده، محاسبات فاصله چندگانه را به طور کارا انجام دهد و داده‌های ترافیک شبکه را به صورت تک‌گذره و پویا دسته‌بندی نماید.

.

2. پیش‌نیازها و ساختار داده ورودی

2.1 کتابخانه‌های مورد نیاز

برای پیاده‌سازی و ارزیابی الگوریتم Echidna به کتابخانه‌های استاندارد زیر نیاز داریم:

• numpy: جهت انجام سریع محاسبات برداری، محاسبه فواصل عددی و ذخیره بردارهای ویژگی.
• pandas: جهت بارگذاری، ساختاردهی و پیش‌پردازش داده‌های ورودی ناهمگون (پروتکل، آدرس‌های IP و حجم بسته‌ها).
• scikit-learn: صرفاً جهت محاسبه معیارهای ارزیابی خوشه‌بندی بدون ناظر (مانند ضریب نیم‌رخ یا Silhouette Score) و نرمال‌سازی داده‌ها.

2.2 ویژگی‌های داده ورودی

داده‌های ورودی برای پردازش توسط Echidna باید ساختاری جدول‌بندی‌شده داشته باشند که در آن هر ستون نماینده یکی از سه نوع ویژگی زیر باشد:

1. ویژگی‌های عددی (Numerical): مقادیری مانند حجم کل بایت‌های انتقال‌یافته یا تعداد بسته‌ها. این ویژگی‌ها باید پیش از ورود به محاسبات فاصله، مقیاس‌دهی (مانند Min-Max Scaling) شوند تا دامنه‌ای بین [0,1] داشته باشند.
2. ویژگی‌های دسته‌ای (Categorical): مقادیری نظیر نوع پروتکل (TCP، UDP، ICMP). این مقادیر به صورت رشته وارد شده و توزیع فراوانی آن‌ها در هر خوشه ذخیره می‌شود.
3. ویژگی‌های سلسله‌مراتبی (Hierarchical): آدرس‌های IP نسخه ۴ (IPv4). این داده‌ها باید به فرمت رشته‌ای استاندارد (مانند 192.168.1.1) ارائه شوند تا الگوریتم بتواند با تفکیک بایت‌ها (Octets)، طول پیشوند مشترک (LCP) را محاسبه کند.

.

3. پیاده‌سازی در Python

در ادامه، پیاده‌سازی شیءگرا و تمیز الگوریتم Echidna ارائه شده است. این کد شامل ساختار درخت، محاسبه فاصله ترکیبی و مدیریت خوشه‌ها است.

import numpy as np
import pandas as pd
from sklearn.metrics import silhouette_score

class EchidnaCluster:
    def __init__(self, numeric_center, categorical_freq, ip_prefix):
        """
        نمایش خلاصه آماری (CF) یک خوشه
        """
        self.n = 1
        self.numeric_center = np.array(numeric_center, dtype=float)
        # categorical_freq: dict {feature_idx: {category_value: count}}
        self.categorical_freq = {k: {v: 1} for k, v in categorical_freq.items()}
        # ip_prefix: dict {feature_idx: "ip_string"}
        self.ip_prefix = ip_prefix

    def update(self, numeric_val, categorical_val, ip_val):
        """
        به‌روزرسانی خلاصه آماری خوشه با ورود نمونه جدید
        """
        self.n += 1
        # به‌روزرسانی میانگین عددی
        self.numeric_center = self.numeric_center + (np.array(numeric_val) - self.numeric_center) / self.n
        
        # به‌روزرسانی فراوانی دسته‌ای
        for idx, val in categorical_val.items():
            if val in self.categorical_freq[idx]:
                self.categorical_freq[idx][val] += 1
            else:
                self.categorical_freq[idx][val] = 1
                
        # به‌روزرسانی پیشوند مشترک IP (سلسله‌مراتبی)
        for idx, ip_str in ip_val.items():
            self.ip_prefix[idx] = self._find_lcp(self.ip_prefix[idx], ip_str)

    def _find_lcp(self, ip1, ip2):
        """
        یافتن پیشوند مشترک بین دو آدرس IP بر اساس بخش‌های هشت‌بیتی (Octets)
        """
        parts1 = ip1.split('.')
        parts2 = ip2.split('.')
        common = []
        for p1, p2 in zip(parts1, parts2):
            if p1 == p2:
                common.append(p1)
            else:
                break
        return ".".join(common) if common else ""


class EchidnaEngine:
    def __init__(self, threshold=0.4, w_num=0.4, w_cat=0.3, w_hier=0.3):
        """
        موتور اصلی خوشه‌بندی اکیدنا
        """
        self.threshold = threshold
        self.w_num = w_num
        self.w_cat = w_cat
        self.w_hier = w_hier
        self.clusters = []

    def _calculate_distance(self, cluster, num_val, cat_val, ip_val):
        # 1. فاصله عددی (اقلیدسی نرمال شده)
        d_num = np.linalg.norm(np.array(num_val) - cluster.numeric_center)
        
        # 2. فاصله دسته‌ای (بر اساس عدم تطابق فراوانی غالب)
        d_cat_list = []
        for idx, val in cat_val.items():
            freq_dict = cluster.categorical_freq[idx]
            total = sum(freq_dict.values())
            val_freq = freq_dict.get(val, 0)
            # فاصله نسبت عکس با فراوانی مقدار در خوشه دارد
            d_cat_list.append(1.0 - (val_freq / total))
        d_cat = np.mean(d_cat_list) if d_cat_list else 0.0

        # 3. فاصله سلسله‌مراتبی IP
        d_hier_list = []
        for idx, val in ip_val.items():
            lcp = cluster._find_lcp(cluster.ip_prefix[idx], val)
            lcp_len = len(lcp.split('.')) if lcp else 0
            d_hier_list.append(1.0 - (lcp_len / 4.0)) # نرمال شده با تعداد هشت‌بیتی‌ها
        d_hier = np.mean(d_hier_list) if d_hier_list else 0.0

        # ترکیب وزنی فواصل
        return (self.w_num * d_num) + (self.w_cat * d_cat) + (self.w_hier * d_hier)

    def fit_predict(self, df, num_cols, cat_cols, ip_cols):
        """
        خوشه‌بندی تک‌گذره روی دیتافریم ورودی
        """
        labels = []
        for idx, row in df.iterrows():
            num_val = row[num_cols].values.astype(float)
            cat_val = {c: row[c] for c in cat_cols}
            ip_val = {c: row[c] for c in ip_cols}

            if not self.clusters:
                # ایجاد اولین خوشه
                new_cluster = EchidnaCluster(num_val, cat_val, ip_val)
                self.clusters.append(new_cluster)
                labels.append(0)
                continue

            # یافتن نزدیک‌ترین خوشه
            min_dist = float('inf')
            closest_idx = -1
            for c_idx, cluster in enumerate(self.clusters):
                dist = self._calculate_distance(cluster, num_val, cat_val, ip_val)
                if dist < min_dist:
                    min_dist = dist
                    closest_idx = c_idx

            # بررسی حد آستانه برای پذیرش در خوشه یا ایجاد خوشه جدید
            if min_dist <= self.threshold:
                self.clusters[closest_idx].update(num_val, cat_val, ip_val)
                labels.append(closest_idx)
            else:
                new_cluster = EchidnaCluster(num_val, cat_val, ip_val)
                self.clusters.append(new_cluster)
                labels.append(len(self.clusters) - 1)

        return np.array(labels)

4. مثال آموزشی

4.1 تعریف داده نمونه

فرض کنید ۵ رکورد شبکه شامل ویژگی عددی (حجم بایت‌ها به صورت نرمال‌شده بین ۰ و ۱)، ویژگی دسته‌ای (پروتکل ارتباطی) و ویژگی سلسله‌مراتبی (آدرس IP مبدأ) به شرح زیر در اختیار داریم:

شناسه	حجم بایت (عددی)	پروتکل (دسته‌ای)	آدرس IP (سلسله‌مراتبی)
1	0.10	TCP	192.168.1.10
2	0.12	TCP	192.168.1.15
3	0.85	UDP	10.0.0.5
4	0.88	UDP	10.0.0.9
5	0.15	TCP	192.168.2.1

4.2.کد پایتون جهت اجرا

# ساخت دیتافریم نمونه
data = {
    'bytes': [0.10, 0.12, 0.85, 0.88, 0.15],
    'protocol': ['TCP', 'TCP', 'UDP', 'UDP', 'TCP'],
    'src_ip': ['192.168.1.10', '192.168.1.15', '10.0.0.5', '10.0.0.9', '192.168.2.1']
}
df_mini = pd.DataFrame(data)

# مقداردهی به مدل با آستانه مناسب برای خوشه‌بندی
engine = EchidnaEngine(threshold=0.35, w_num=0.4, w_cat=0.3, w_hier=0.3)
labels = engine.fit_predict(df_mini, num_cols=['bytes'], cat_cols=['protocol'], ip_cols=['src_ip'])

print("Predicted labels:", labels)
for i, clus in enumerate(engine.clusters):
    print(f"Cluster {i} -> Size: {clus.n}, Numeric Center: {clus.numeric_center}, IP Prefix: {clus.ip_prefix}")

4.3 چرا این الگوریتم برای مثال مناسب است؟

در این نمونه، تفاوت‌های آشکاری بین رفتارهای ترافیکی وجود دارد. رکوردهای اول و دوم مربوط به یک زیرشبکه کلاس C با حجم دانلود کم و پروتکل یکسان هستند. رکوردهای سوم و چهارم متعلق به یک شبکه محلی دیگر با پهنای باند بالا و پروتکل UDP می‌باشند. رکورد پنجم نیز با اینکه پروتکل و حجم یکسانی با دسته اول دارد، اما در زیرشبکه متفاوتی (192.168.2.0/24) قرار گرفته است. الگوریتم Echidna به خوبی می‌تواند بر اساس فاصله ترکیبی، مرزبندی‌های سلسله‌مراتبی و محتوایی را تشخیص دهد.

4.4 خروجی مورد انتظار و نتایج محاسبات

پس از اجرای کد فوق، خروجی به شکل زیر خواهد بود:

Predicted labels: [0 0 1 1 2]
Cluster 0 -> Size: 2, Numeric Center: [0.11], IP Prefix: {'src_ip': '192.168.1'}
Cluster 1 -> Size: 2, Numeric Center: [0.865], IP Prefix: {'src_ip': '10.0.0'}
Cluster 2 -> Size: 1, Numeric Center: [0.15], IP Prefix: {'src_ip': '192.168.2.1'}

تفسیر محاسبات:

• نمونه‌های ۱ و ۲ به دلیل شباهت بالا در تمام بخش‌ها در خوشه 0 قرار گرفتند و پیشوند آدرس IP آن‌ها به شبکه مشترک 192.168.1 خلاصه شد.
• نمونه‌های ۳ و ۴ به دلیل حجم متمایز، پروتکل UDP و آدرس متناظر در خوشه 1 جای گرفتند.
• نمونه ۵ با وجود شباهت با خوشه ۰، به علت تفاوت در بخش سوم آدرس IP (192.168.2.1 در مقابل 192.168.1.x) حد آستانه فاصله را رد کرد و خوشه 2 را پدید آورد.

.

5. مطالعه موردی 1: پایش و تفکیک جریان‌های ترافیکی در بخش لبه شبکه دانشگاهی

5.1 معرفی مطالعه موردی

در یک شبکه دانشگاهی، تفکیک ترافیک عادی (وب‌گردی، دانلود فایل‌های آموزشی) از ترافیک‌های متفرقه یا مشکوک در لبه شبکه از اهمیت ویژه‌ای برخوردار است. الگوریتم Echidna به دلیل قابلیت پردازش جریان‌های داده به صورت تک‌گذره و بدون نیاز به ذخیره‌سازی داده‌های حساس و پرحجم، برای خلاصه‌سازی ترافیک لبه شبکه انتخاب شده است.

5.2 تشریح داده‌ها

داده‌های ورودی شامل رکوردهایی با ساختار زیر هستند:

• ویژگی عددی: bytes_scaled (حجم مبادلات نرمال‌سازی شده)
• ویژگی دسته‌ای: service_type (انواع دسته‌ها نظیر HTTP، DNS، SSH)
• ویژگی سلسله‌مراتبی: client_ip (آدرس IP کلاینت‌های داخلی دانشگاه)

5.3 آماده‌سازی داده و سناریوی پیشرفته

برای شبیه‌سازی سناریوی دانشگاهی، کدی جهت تولید داده‌های سنتتیک اما واقع‌گرایانه در حجم بزرگ‌تر پیاده‌سازی می‌کنیم.

# تولید داده‌های سنتتیک برای مطالعه موردی 1
np.random.seed(42)
n_samples = 200

# تولید آی‌پی‌ها
ips = (['192.168.10.' + str(i) for i in np.random.randint(1, 50, n_samples // 2)] +  # دانشکده فنی
       ['192.168.20.' + str(i) for i in np.random.randint(1, 50, n_samples // 2)])   # کتابخانه

# تولید سرویس‌ها و حجم بایت متناسب با رفتار کاربران
services = ['HTTP'] * 120 + ['DNS'] * 60 + ['SSH'] * 20
np.random.shuffle(services)

bytes_data = []
for s in services:
    if s == 'HTTP':
        bytes_data.append(np.random.uniform(0.1, 0.4))
    elif s == 'DNS':
        bytes_data.append(np.random.uniform(0.01, 0.08))
    else: # SSH
        bytes_data.append(np.random.uniform(0.6, 0.9))

df_case1 = pd.DataFrame({
    'bytes_scaled': bytes_data,
    'service_type': services,
    'client_ip': ips
})

5.4 پیاده‌سازی پایتون برای مطالعه موردی 1

# مقداردهی مدل
model_case1 = EchidnaEngine(threshold=0.3, w_num=0.5, w_cat=0.2, w_hier=0.3)

# اجرای الگوریتم
labels_case1 = model_case1.fit_predict(
    df_case1, 
    num_cols=['bytes_scaled'], 
    cat_cols=['service_type'], 
    ip_cols=['client_ip']
)
df_case1['cluster'] = labels_case1

5.5 اجرای الگوریتم و ارزیابی

به دلیل ناهمگن بودن ویژگی‌ها، برای ارزیابی کیفی، توزیع خوشه‌ها و ویژگی‌های میانگین آن‌ها را استخراج می‌کنیم.

print(f"تعداد خوشه‌های شناسایی شده: {len(model_case1.clusters)}")
print(df_case1['cluster'].value_counts())

5.6 تحلیل و تفسیر خروجی

خروجی این شبیه‌سازی خوشه‌های مجزایی را بر اساس نوع فعالیت دانشکده‌ها نشان می‌دهد:

• یک خوشه بزرگ شامل کاربران دانشکده فنی (192.168.10.x) که از سرویس HTTP استفاده می‌کنند و حجم دانلود متوسط دارند.
• خوشه‌ای متمایز مربوط به رفتارهای اداری یا کتابخانه با حجم کم (پروتکل DNS).
• خوشه‌های کوچک یا تک‌عضوی که نشان‌دهنده اتصالات پرحجم SSH خارج از محدوده عادی زیرشبکه کلاینت‌ها هستند که می‌توانند به عنوان کاندیدای ناهنجاری یا دسترسی غیرمجاز بررسی شوند.

5.7 عیب‌یابی (Troubleshooting) برای مطالعه موردی 1

• حساسیت به ترتیب ورود داده‌ها (Order Dependency): از آنجا که الگوریتم افزایشی و یک‌گذره است، تغییر در ترتیب ورود نمونه‌ها می‌تواند خوشه‌های متفاوتی با مرزهای متغیر ایجاد کند.
•  راهکار: پیش از شروع پردازش جریانی، داده‌ها را به صورت تصادفی مخلوط (Shuffle) کنید یا از یک پنجره پیش‌پردازش اولیه برای تثبیت مراکز خوشه‌های پایه استفاده نمایید.
• تعداد بسیار زیاد خوشه‌های تک‌عضوی (Cluster Explosion): اگر پارامتر آستانه T را بیش از حد کوچک انتخاب کنیم، هر نمونه جدید که با نمونه‌های قبلی تفاوت اندکی دارد، یک خوشه جدید می‌سازد.
•  راهکار: مقدار T را به تدریج افزایش دهید یا فواصل عددی را با استفاده از فیلترهای استوارتر مقیاس‌دهی کنید.

.

6. مطالعه موردی 2: تشخیص ناهنجاری و اسکن پورت‌ها در لایه توزیع مرکز داده

6.1 معرفی مطالعه موردی

ترافیک ناشی از حملات اسکن پورت (Port Scanning) معمولاً با ارسال بسته‌های کوچک با تعداد بالا از یک یا چند IP مشخص به پورت‌های مختلف سرورها مشخص می‌شود. در این مطالعه موردی، از Echidna استفاده می‌کنیم تا الگوهای ارتباطی ماشین‌های شبکه را خوشه‌بندی کرده و ماشین‌هایی را که از الگوهای مرسوم پیروی نکرده و خوشه‌های پرت با ویژگی‌های رفتاری خاص ایجاد می‌کنند، شناسایی کنیم.

6.2 تشریح داده‌ها

• ویژگی عددی: conn_duration (مدت زمان اتصال به صورت نرمال‌شده)
• ویژگی دسته‌ای: dest_port (پورت مقصد به عنوان رشته دسته‌ای مانند ‘80’، ‘22’، ‘443’)
• ویژگی سلسله‌مراتبی: dest_ip (آدرس IP سرورهای هدف)

6.3 آماده‌سازی داده و سناریوی پیشرفته

کد شبیه‌سازی سناریوی اسکن پورت:

# ۲۰۰ اتصال عادی و ۱۰ اتصال اسکن پورت
n_normal = 200
n_scan = 10

# داده‌های عادی: اتصال به پورت ۸۰ یا ۴۴۳ سرورهای وب داخلی (زیرشبکه 172.16.1.x)
normal_ips = ['172.16.1.' + str(i) for i in np.random.randint(10, 20, n_normal)]
normal_ports = np.random.choice(['80', '443'], n_normal)
normal_duration = np.random.uniform(0.1, 0.3, n_normal)

# داده‌های اسکن پورت: یک سیستم با آی‌پی هدف متغیر در رنج وسیع و مدت زمان اتصال بسیار کوتاه
scan_ips = ['172.16.2.' + str(i) for i in range(1, n_scan + 1)]
scan_ports = [str(p) for p in np.random.randint(1000, 5000, n_scan)]
scan_duration = np.random.uniform(0.001, 0.005, n_scan) # اتصالات بسیار سریع

df_case2 = pd.DataFrame({
    'conn_duration': np.concatenate([normal_duration, scan_duration]),
    'dest_port': np.concatenate([normal_ports, scan_ports]),
    'dest_ip': normal_ips + scan_ips
})

6.4 پیاده‌سازی پایتون برای مطالعه موردی 2

model_case2 = EchidnaEngine(threshold=0.5, w_num=0.3, w_cat=0.4, w_hier=0.3)

labels_case2 = model_case2.fit_predict(
    df_case2,
    num_cols=['conn_duration'],
    cat_cols=['dest_port'],
    ip_cols=['dest_ip']
)
df_case2['cluster'] = labels_case2

6.5 اجرای الگوریتم و ارزیابی

بررسی تعداد اعضای هر خوشه برای پیدا کردن خوشه‌های پرت (زیر ۲ یا ۳ عضو):

cluster_counts = df_case2['cluster'].value_counts()
print("Cluster size distribution:\n", cluster_counts)
anomaly_clusters = cluster_counts[cluster_counts <= 2].index.tolist()
print("خوشه‌های کاندید ناهنجاری (اسکن پورت):", anomaly_clusters)

 6.6 تحلیل و تفسیر خروجی

به دلیل اینکه فعالیت‌های اسکن پورت پورت‌های کاملاً متفاوتی (تصادفی در رنج ۱۰۰۰ تا ۵۰۰۰) و رنج IP متفاوتی (172.16.2.x) را هدف قرار می‌دهند، فاصله ترکیبی آن‌ها با خوشه‌های سنگین سرورهای وب بسیار زیاد می‌شود. در نتیجه، الگوریتم Echidna این اتصالات را در خوشه‌های بسیار کوچک و تک‌عضوی رها می‌کند. این امر فرآیند تشخیص ناهنجاری در لایه توزیع مرکز داده را بدون نیاز به تحلیل‌های پیچیده و پسینی تسهیل می‌کند.

6.7 عیب‌یابی (Troubleshooting) برای مطالعه موردی 2

• مشکل حساسیت به نویز و بروز خطای کم‌برازش (Underfitting): در صورتی که وزن ویژگی عددی یا دسته‌ای خیلی پایین تنظیم شود، الگوریتم تفاوت‌های مهم در رفتار اتصالات کوتاه را نادیده گرفته و همه را در یک خوشه بسیار بزرگ ادغام می‌کند. راهکار: وزن‌ها (www) باید با توجه به ماهیت حمله تنظیم شوند. برای تشخیص اسکن پورت، پورت‌های مقصد و آدرس‌های IP اهمیت بیشتری دارند، لذا باید ضریب  w_cat و  w_hier  نسبت به  w_num ​ بیشتر در نظر گرفته شود.

.

7. جمع‌بندی

در این پیوست عملیاتی، نحوه پیاده‌سازی کد الگوریتم Echidna به صورت شیءگرا در زبان پایتون نمایش داده شد. نشان دادیم که چگونه با ترکیب سه معیار فاصله (اقلیدسی، تطابق فراوانی دسته‌ای و پیشوند مشترک IP) می‌توان داده‌های ترافیک شبکه را به صورت بلادرنگ و افزایشی تحلیل کرد. از طریق دو مطالعه موردی مجزا، کارایی این روش در تفکیک الگوهای عادی کلاینت‌های دانشگاهی و همچنین کشف حملات اسکن پورت در لایه توزیع مراکز داده به تصویر کشیده شد. تنظیم دقیق پارامترهای آستانه (T) و وزن‌دهی به ویژگی‌ها، کلید اصلی موفقیت در پیاده‌سازی‌های صنعتی این الگوریتم محسوب می‌شوند.