Слабая криптография и форензик разбор трафика. Аудит TLS, детект слабых cipher suites, анализ зашифрованного C2

#!/usr/bin/env python3
"""
tls_config_auditor.py
TLS Configuration Security Audit Tool
Cryptography Assessment

что проверяем:
  1. поддерживаемые версии TLS
  2. cipher suites — слабые и небезопасные
  3. сертификат — алгоритм подписи, размер ключа, срок
  4. Forward Secrecy
  5. HSTS, HPKP заголовки
  6. известные уязвимости (CRIME, BEAST, SWEET32)

зависимости: pip install cryptography pyopenssl requests
"""

import ssl
import socket
import json
import sys
from datetime import datetime, timezone
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa, ec, dsa


# слабые cipher suites (substring matching по OpenSSL именам)
WEAK_CIPHERS = {
    "RC4":        "CRITICAL — статистически сломан",
    "DES":        "CRITICAL — 56-bit ключ, trivial brute force",
    "3DES":       "HIGH — SWEET32 атака, 64-bit блок",
    "EXPORT":     "CRITICAL — намеренно ослабленное крипто (FREAK)",
    "NULL":       "CRITICAL — нет шифрования",
    "anon":       "CRITICAL — нет аутентификации",
    "MD5":        "HIGH — collision attacks",
    "SHA":        "MEDIUM — только SHA1, рекомендуется SHA256+",
    "CBC":        "LOW — уязвимо к BEAST/Lucky13 в TLS 1.0/1.1",
    "ADH":        "CRITICAL — анонимный DH, нет аутентификации",
    "AECDH":      "CRITICAL — анонимный ECDH",
}

# cipher suites без Forward Secrecy
NON_FS_PATTERNS = ["RSA", "kRSA"]  # RSA key exchange без FS

# хорошие cipher suites (должны присутствовать)
RECOMMENDED = [
    "TLS_AES_128_GCM_SHA256",
    "TLS_AES_256_GCM_SHA384",
    "TLS_CHACHA20_POLY1305_SHA256",
    "ECDHE-RSA-AES256-GCM-SHA384",
    "ECDHE-ECDSA-AES256-GCM-SHA384",
]


class TLSAuditor:
    def __init__(self, host: str, port: int = 443):
        self.host     = host
        self.port     = port
        self.findings = []
        self.score    = 100  # начинаем с 100, вычитаем за проблемы

    def _add(self, severity: str, desc: str, detail: str = ""):
        penalties = {"CRITICAL": 30, "HIGH": 20, "MEDIUM": 10, "LOW": 5}
        self.score -= penalties.get(severity, 0)
        self.findings.append({
            "severity": severity,
            "desc":     desc,
            "detail":   detail
        })

    def check_protocol_versions(self) -> None:
        """проверяем какие версии TLS принимает сервер"""
        print(f"[*] проверка поддерживаемых протоколов...")

        legacy_protocols = [
            (ssl.PROTOCOL_TLSv1,   "TLS 1.0"),
            (ssl.PROTOCOL_TLSv1_1, "TLS 1.1") if hasattr(ssl, 'PROTOCOL_TLSv1_1') else None,
        ]

        for item in legacy_protocols:
            if item is None:
                continue
            proto_const, proto_name = item
            try:
                ctx = ssl.SSLContext(proto_const)
                ctx.check_hostname = False
                ctx.verify_mode    = ssl.CERT_NONE
                with socket.create_connection((self.host, self.port), timeout=5) as sock:
                    with ctx.wrap_socket(sock, server_hostname=self.host):
                        self._add("HIGH",
                            f"{proto_name} поддерживается сервером",
                            "устаревший протокол — PCI DSS требует минимум TLS 1.2")
                        print(f"  [FAIL] {proto_name}: поддерживается")
            except ssl.SSLError:
                print(f"  [OK]   {proto_name}: отклонён")
            except Exception:
                print(f"  [?]    {proto_name}: не удалось проверить")

    def check_certificate(self, cert_der: bytes) -> None:
        """анализируем сертификат"""
        try:
            cert = x509.load_der_x509_certificate(cert_der)
        except Exception as e:
            print(f"  [-] не удалось разобрать сертификат: {e}")
            return

        # алгоритм подписи
        sig_alg = cert.signature_algorithm_oid._name
        print(f"  Подпись: {sig_alg}")
        if "md5" in sig_alg.lower():
            self._add("CRITICAL", "сертификат подписан MD5", sig_alg)
        elif "sha1" in sig_alg.lower() and "sha1With" in sig_alg:
            self._add("HIGH", "сертификат подписан SHA1", sig_alg)

        # размер ключа
        pub_key = cert.public_key()
        if isinstance(pub_key, rsa.RSAPublicKey):
            key_size = pub_key.key_size
            print(f"  RSA ключ: {key_size} бит")
            if key_size < 2048:
                self._add("CRITICAL", f"RSA ключ слишком мал: {key_size} бит",
                          "минимум 2048 бит, рекомендуется 4096")
            elif key_size < 3072:
                self._add("LOW", f"RSA {key_size} бит — рассмотреть 4096 бит")
        elif isinstance(pub_key, ec.EllipticCurvePublicKey):
            curve = pub_key.curve.name
            print(f"  EC кривая: {curve}")
            if curve in ("secp192r1", "prime192v1"):
                self._add("HIGH", f"слабая EC кривая: {curve}", "минимум secp256r1")

        # срок действия
        now  = datetime.now(timezone.utc)
        exp  = cert.not_valid_after_utc
        diff = (exp - now).days
        print(f"  Истекает: {exp.date()} (через {diff} дней)")
        if diff < 0:
            self._add("CRITICAL", "сертификат просрочен", str(exp.date()))
        elif diff < 30:
            self._add("HIGH", f"сертификат истекает через {diff} дней")
        elif diff < 90:
            self._add("MEDIUM", f"сертификат истекает через {diff} дней")

        # self-signed
        if cert.issuer == cert.subject:
            self._add("HIGH", "self-signed сертификат",
                      "нет цепочки доверия — уязвимо к MITM")

    def check_cipher_suites(self) -> None:
        """получаем и анализируем согласованный cipher suite"""
        print(f"\n[*] проверка cipher suite...")
        try:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode    = ssl.CERT_NONE

            with socket.create_connection((self.host, self.port), timeout=10) as sock:
                with ctx.wrap_socket(sock, server_hostname=self.host) as ssock:
                    cipher = ssock.cipher()
                    proto  = ssock.version()
                    cert_der = ssock.getpeercert(binary_form=True)

                    print(f"  Протокол: {proto}")
                    print(f"  Cipher:   {cipher[0]}")
                    print(f"  Bits:     {cipher[2]}")

                    # проверяем cipher на слабость
                    for weak, reason in WEAK_CIPHERS.items():
                        if weak.upper() in cipher[0].upper():
                            self._add(
                                "CRITICAL" if "CRITICAL" in reason else "HIGH",
                                f"слабый cipher suite: {cipher[0]}",
                                reason
                            )

                    # Forward Secrecy
                    has_fs = any(kex in cipher[0].upper()
                                 for kex in ["ECDHE", "DHE", "TLS_AES", "TLS_CHACHA"])
                    if not has_fs:
                        self._add("HIGH",
                            "нет Forward Secrecy",
                            "захваченный трафик может быть расшифрован после компромисса ключа")

                    if cert_der:
                        print(f"\n[*] анализ сертификата...")
                        self.check_certificate(cert_der)

        except Exception as e:
            print(f"  [-] ошибка подключения: {e}")

    def check_security_headers(self) -> None:
        """проверяем HTTP security заголовки"""
        print(f"\n[*] проверка security headers...")
        try:
            import urllib.request
            req = urllib.request.Request(
                f"https://{self.host}:{self.port}/",
                headers={"User-Agent": "TLS-Auditor/1.0"}
            )
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode    = ssl.CERT_NONE

            with urllib.request.urlopen(req, context=ctx, timeout=10) as resp:
                headers = dict(resp.headers)

                if "strict-transport-security" not in {k.lower(): v for k, v in headers.items()}:
                    self._add("MEDIUM", "HSTS заголовок отсутствует",
                              "без HSTS возможен SSL stripping")
                else:
                    print(f"  [OK] HSTS присутствует")

                if "content-security-policy" not in {k.lower() for k in headers}:
                    self._add("LOW", "Content-Security-Policy отсутствует")

        except Exception as e:
            print(f"  [-] не удалось проверить HTTP заголовки: {e}")

    def audit(self) -> None:
        print(f"\n{'='*65}")
        print(f"TLS AUDIT: {self.host}:{self.port}")
        print(f"{'='*65}\n")

        self.check_protocol_versions()
        self.check_cipher_suites()
        self.check_security_headers()

        grade = ("A" if self.score >= 90 else
                 "B" if self.score >= 75 else
                 "C" if self.score >= 60 else
                 "D" if self.score >= 45 else "F")

        print(f"\n{'='*65}")
        print(f"ИТОГ: Grade={grade}  Score={max(0, self.score)}/100")
        print(f"{'='*65}\n")

        for f in sorted(self.findings,
                        key=lambda x: {"CRITICAL":0,"HIGH":1,"MEDIUM":2,"LOW":3}
                        .get(x["severity"], 4)):
            print(f"  [{f['severity']:8}] {f['desc']}")
            if f["detail"]:
                print(f"              {f['detail']}")


if __name__ == "__main__":
    targets = sys.argv[1:] if len(sys.argv) > 1 else ["example.com"]
    for target in targets:
        host = target.split(":")[0]
        port = int(target.split(":")[1]) if ":" in target else 443
        TLSAuditor(host, port).audit()

TLS Client Hello содержит открытым текстом:
  - TLS версию
  - список cipher suites (в порядке предпочтения)
  - список extensions
  - длины и порядок всего вышеперечисленного

JA3 = MD5(SSLVersion,Ciphers,Extensions,EllipticCurves,EllipticCurvePointFormats)

каждая реализация TLS имеет уникальный JA3:
  Chrome 120:        772,4865-4866-4867-...,0-23-65281-10-11-...
  Firefox 121:       другой набор
  Cobalt Strike:     e7d705a3286e19ea42f587b344ee6865  ← известный
  Metasploit:        6734f37431670b3ab4292b8f60f29984
  Sliver C2:         a0e9f5d64349fb13191bc781f81f42e1
  Python requests:   d9f955de00eddd3c2000dadf1cc16daf

если JA3 совпадает с известным C2 фреймворком — ALERT
если JA3 неизвестный и процесс не браузер — ALERT

#!/usr/bin/env python3
"""
ja3_c2_analyzer.py
JA3 Fingerprint Analysis для детектирования C2 трафика
Network Forensics

что делаем:
  1. парсим pcap файл (или используем tshark для live capture)
  2. вычисляем JA3 для каждого TLS Client Hello
  3. сравниваем с базой известных C2 фреймворков
  4. детектируем аномальные паттерны без расшифровки

зависимости: pip install scapy pyshark dpkt
tshark должен быть установлен для live анализа
"""

import subprocess
import json
import hashlib
import re
import sys
from collections import defaultdict
from datetime import datetime


# известные JA3 хэши C2 фреймворков и малвари
# источник: публичные базы (salesforce JA3, abuse.ch)
KNOWN_MALICIOUS_JA3 = {
    # C2 фреймворки
    "e7d705a3286e19ea42f587b344ee6865": ("Cobalt Strike", "CRITICAL"),
    "6734f37431670b3ab4292b8f60f29984": ("Metasploit", "CRITICAL"),
    "a0e9f5d64349fb13191bc781f81f42e1": ("Sliver C2", "CRITICAL"),
    "36f7277af969b5c813cc24f60e4d4a5b": ("Havoc C2", "CRITICAL"),
    "f1c55573e3adb06bb1e8e1bb6a7fe5b3": ("Brute Ratel", "CRITICAL"),
    # малварь
    "b742b407d1f7a7e7b5e13a61f52e9f7e": ("TrickBot", "CRITICAL"),
    "d9f955de00eddd3c2000dadf1cc16daf": ("Python requests (suspicious)", "HIGH"),
    "6dcbc7531a0f8c65c01419d4de71fe91": ("Tor Browser", "HIGH"),
    # инструменты сканирования
    "13303773a27f969b5c813cc24f60e4d4": ("nmap ssl-scan", "MEDIUM"),
    "3b5074b1b5d032e5620f69f9159c1432": ("Masscan", "MEDIUM"),
}

# легитимные известные JA3 (whitelist)
KNOWN_LEGITIMATE_JA3 = {
    "66918128f1b9b03303d77c6f2eefd128": "Chrome 120",
    "579ccef312d18482fc42e2b822ca2430": "Chrome 119",
    "b32309a26951912be7dba376398abc3b": "Firefox 121",
    "a4e18002e0b19e2fb10f1f4800a4d3f3": "Safari 17",
    "b1b561f2f40f7fe5b6a0a4ba0bf26a23": "Edge Chromium",
    "3b5074b1b5d032e5620f69f9159c1432": "curl",
}


def compute_ja3(tls_record: dict) -> str:
    """
    вычисляем JA3 хэш из полей TLS Client Hello
    tls_record: словарь с полями из парсинга пакета
    """
    version    = str(tls_record.get("version", 771))
    ciphers    = "-".join(str(c) for c in tls_record.get("ciphers", [])
                          if c not in (0x00FF, 0x5600))  # исключаем GREASE
    extensions = "-".join(str(e) for e in tls_record.get("extensions", [])
                          if e not in range(0x0A0A, 0xFAFA, 0x1010))
    curves     = "-".join(str(c) for c in tls_record.get("curves", [])
                          if c not in range(0x0A0A, 0xFAFA, 0x1010))
    points     = "-".join(str(p) for p in tls_record.get("ec_points", []))

    ja3_string = f"{version},{ciphers},{extensions},{curves},{points}"
    return hashlib.md5(ja3_string.encode()).hexdigest(), ja3_string


def analyze_pcap_with_tshark(pcap_path: str) -> list:
    """
    используем tshark для извлечения TLS Client Hello данных
    это наиболее надёжный метод без написания полного TLS парсера
    """
    findings = []

    # tshark команда для извлечения JA3 и метаданных
    cmd = [
        "tshark", "-r", pcap_path,
        "-Y", "tls.handshake.type == 1",  # Client Hello only
        "-T", "json",
        "-e", "ip.src",
        "-e", "ip.dst",
        "-e", "tcp.dstport",
        "-e", "tls.handshake.version",
        "-e", "tls.handshake.ciphersuite",
        "-e", "tls.handshake.extension.type",
        "-e", "tls.handshake.extensions_server_name",
        "-e", "frame.time",
        "-e", "ja3.hash",  # если установлен ja3 плагин
    ]

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        packets = json.loads(result.stdout)
    except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError) as e:
        print(f"  [-] tshark ошибка: {e}")
        print(f"  [*] установи tshark и ja3 плагин: pip install pyshark")
        return []

    # группируем по src IP для анализа поведения
    src_ja3_map = defaultdict(set)

    for pkt in packets:
        layers = pkt.get("_source", {}).get("layers", {})

        src_ip = layers.get("ip.src", [""])[0]
        dst_ip = layers.get("ip.dst", [""])[0]
        dst_port = layers.get("tcp.dstport", ["443"])[0]
        sni    = layers.get("tls.handshake.extensions_server_name", [""])[0]
        ja3    = layers.get("ja3.hash", [""])[0]

        if not ja3:
            continue

        src_ja3_map[src_ip].add(ja3)

        # проверяем против известных вредоносных JA3
        if ja3 in KNOWN_MALICIOUS_JA3:
            name, severity = KNOWN_MALICIOUS_JA3[ja3]
            findings.append({
                "severity":  severity,
                "type":      "known_malicious_ja3",
                "src_ip":    src_ip,
                "dst_ip":    dst_ip,
                "dst_port":  dst_port,
                "sni":       sni or "(no SNI — подозрительно)",
                "ja3":       ja3,
                "matched":   name,
            })

        # подозрительные признаки без расшифровки
        # нет SNI в TLS 1.2 — аномалия для легитимных приложений
        if not sni and dst_port == "443":
            findings.append({
                "severity": "MEDIUM",
                "type":     "no_sni_in_tls",
                "src_ip":   src_ip,
                "dst_ip":   dst_ip,
                "ja3":      ja3,
                "reason":   "легитимные браузеры всегда передают SNI"
            })

    # детект: один хост использует много разных JA3 (инструмент сканирования)
    for src_ip, ja3_set in src_ja3_map.items():
        if len(ja3_set) > 5:
            findings.append({
                "severity": "MEDIUM",
                "type":     "multiple_ja3_from_single_host",
                "src_ip":   src_ip,
                "ja3_count": len(ja3_set),
                "reason":   "много разных TLS fingerprints с одного IP = инструмент"
            })

    return findings


def analyze_tls_metadata_patterns(pcap_path: str) -> list:
    """
    анализируем поведенческие паттерны зашифрованного трафика:
      - beacon timing (регулярные интервалы = C2 heartbeat)
      - Certificate transparency аномалии
      - аномальные размеры пакетов
    """
    findings = []

    # beacon timing через tshark — ищем регулярные подключения к одному IP
    cmd = [
        "tshark", "-r", pcap_path,
        "-Y", "tcp.flags.syn == 1 and tcp.flags.ack == 0",
        "-T", "fields",
        "-e", "frame.time_epoch",
        "-e", "ip.dst",
        "-e", "tcp.dstport",
    ]

    try:
        result  = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        lines   = result.stdout.strip().splitlines()
    except Exception:
        return []

    # группируем по dst IP:port
    connections = defaultdict(list)
    for line in lines:
        parts = line.strip().split('\t')
        if len(parts) >= 3:
            try:
                ts   = float(parts[0])
                dst  = f"{parts[1]}:{parts[2]}"
                connections[dst].append(ts)
            except ValueError:
                pass

    # ищем регулярные интервалы — признак C2 beacon
    for dst, timestamps in connections.items():
        if len(timestamps) < 5:
            continue

        timestamps.sort()
        intervals = [timestamps[i+1] - timestamps[i]
                     for i in range(len(timestamps)-1)]

        if not intervals:
            continue

        avg = sum(intervals) / len(intervals)
        deviation = max(abs(i - avg) for i in intervals) if intervals else 0

        # если отклонение < 15% от среднего — это beacon
        if avg > 10 and deviation / avg < 0.15 and len(intervals) >= 4:
            findings.append({
                "severity": "HIGH",
                "type":     "c2_beacon_timing_pattern",
                "dst":      dst,
                "avg_interval_sec": f"{avg:.1f}",
                "deviation_pct":    f"{deviation/avg*100:.1f}%",
                "count":    len(timestamps),
                "reason":   "регулярные соединения с одинаковым интервалом = C2 heartbeat"
            })

    return findings


def main():
    if len(sys.argv) < 2:
        print("Usage: ja3_c2_analyzer.py <capture.pcap>")
        print("\nдля захвата live трафика:")
        print("  tshark -i eth0 -w /tmp/capture.pcap -a duration:60")
        sys.exit(1)

    pcap_path = sys.argv[1]
    print(f"=== JA3 C2 Traffic Analyzer ===")
    print(f"файл: {pcap_path}\n")

    print("[*] анализ JA3 fingerprints...")
    ja3_findings = analyze_pcap_with_tshark(pcap_path)

    print("[*] анализ timing паттернов (beacon detection)...")
    timing_findings = analyze_tls_metadata_patterns(pcap_path)

    all_findings = ja3_findings + timing_findings

    print(f"\n{'='*65}")
    if not all_findings:
        print("[OK] подозрительных C2 паттернов в трафике не обнаружено")
    else:
        crit = [f for f in all_findings if f["severity"] == "CRITICAL"]
        high = [f for f in all_findings if f["severity"] == "HIGH"]
        print(f"[!!!] КРИТИЧЕСКИХ: {len(crit)}, ВЫСОКИХ: {len(high)}\n")

        for f in sorted(all_findings,
                        key=lambda x: {"CRITICAL":0,"HIGH":1,"MEDIUM":2}.get(x["severity"],3)):
            print(f"  [{f['severity']:8}] {f['type']}")
            for k, v in f.items():
                if k not in ("severity", "type") and v:
                    print(f"    {k}: {v}")
            print()


if __name__ == "__main__":
    main()

// детект устаревших TLS версий в корпоративном трафике
// источник: Azure Firewall / NSG Flow Logs / Defender for Endpoint

DeviceNetworkEvents
| where Timestamp > ago(7d)
| extend TlsVersion = tostring(parse_json(AdditionalFields).TlsVersion)
| where TlsVersion in ("TLS 1.0", "TLS 1.1", "SSL 3.0", "SSL 2.0")
| where not (RemoteIP startswith "10." or RemoteIP startswith "192.168.")
// внутренние legacy системы — отдельный тикет
| project Timestamp, DeviceName, AccountName,
          RemoteIP, RemotePort, TlsVersion,
          InitiatingProcessFileName,
          InitiatingProcessCommandLine
| order by Timestamp desc

// детект C2 beacon паттерна через регулярные соединения
// ищем процессы с ритмичными outbound соединениями

DeviceNetworkEvents
| where Timestamp > ago(1h)
| where ActionType == "ConnectionSuccess"
| where not (RemoteIP startswith "10." or
             RemoteIP startswith "192.168." or
             RemoteIP startswith "172.")
| summarize
    ConnectionCount = count(),
    TimeSpan = max(Timestamp) - min(Timestamp)
    by RemoteIP, RemotePort, InitiatingProcessFileName,
       DeviceName, bin(Timestamp, 5m)
| where ConnectionCount > 3
// вычисляем приблизительный интервал
| extend AvgIntervalSec = toreal(TimeSpan) / toreal(ConnectionCount)
// подозрительно: 4-30 соединений в час с почти равным интервалом
| where ConnectionCount between (4 .. 30)
| where AvgIntervalSec between (60.0 .. 3600.0)  // от 1 до 60 минут
| project Timestamp, DeviceName, RemoteIP, RemotePort,
          InitiatingProcessFileName, ConnectionCount,
          AvgIntervalSec
| order by ConnectionCount desc

// JA3 аномалии через Defender for Endpoint
// известные вредоносные JA3 хэши

let MaliciousJA3 = datatable(JA3Hash: string, Name: string) [
    "e7d705a3286e19ea42f587b344ee6865", "Cobalt Strike",
    "6734f37431670b3ab4292b8f60f29984", "Metasploit",
    "a0e9f5d64349fb13191bc781f81f42e1", "Sliver C2",
    "36f7277af969b5c813cc24f60e4d4a5b", "Havoc C2",
    "f1c55573e3adb06bb1e8e1bb6a7fe5b3", "Brute Ratel",
];

DeviceNetworkEvents
| where Timestamp > ago(7d)
| extend JA3 = tostring(parse_json(AdditionalFields).JA3)
| where isnotempty(JA3)
| join kind=inner MaliciousJA3 on $left.JA3 == $right.JA3Hash
| project Timestamp, DeviceName, AccountName,
          RemoteIP, RemotePort, JA3,
          Name, InitiatingProcessFileName,
          InitiatingProcessCommandLine
| order by Timestamp desc