SSL/TLS 安全通訊完全實戰指南

#!/bin/bash

# OpenSSL 憑證生成完整流程示範
# 展示從密鑰生成到自簽憑證的完整過程

echo "=== SSL/TLS 憑證生成實戰範例 ==="

# 步驟 1: 生成私鑰
# 使用 RSA 2048 位元金鑰（生產環境建議 4096 位元）
echo -e "\n[步驟 1] 生成 RSA 私鑰..."
openssl genrsa -out server.key 2048

# 驗證私鑰
echo "[驗證] 檢查私鑰格式..."
openssl rsa -in server.key -text -noout | head -n 5

# 步驟 2: 生成憑證簽發請求 (CSR)
# 包含伺服器的識別資訊
echo -e "\n[步驟 2] 生成憑證簽發請求..."
openssl req -new -key server.key -out server.csr \
  -subj "/C=TW/ST=Taiwan/L=Taipei/O=Example Corp/OU=IT/CN=example.com"

# 查看 CSR 內容
echo "[驗證] 檢查 CSR 內容..."
openssl req -in server.csr -text -noout | grep -A 5 "Subject:"

# 步驟 3: 生成自簽憑證（測試環境使用）
# 生產環境應提交 CSR 給受信任的 CA
echo -e "\n[步驟 3] 生成自簽憑證 (365 天有效期)..."
openssl x509 -req -days 365 \
  -in server.csr \
  -signkey server.key \
  -out server.crt

# 查看憑證詳細資訊
echo -e "\n[驗證] 憑證詳細資訊:"
openssl x509 -in server.crt -text -noout | head -n 20

# 步驟 4: 驗證憑證與私鑰配對
echo -e "\n[驗證] 確認憑證與私鑰配對..."
cert_modulus=$(openssl x509 -noout -modulus -in server.crt | openssl md5)
key_modulus=$(openssl rsa -noout -modulus -in server.key | openssl md5)

if [ "$cert_modulus" = "$key_modulus" ]; then
    echo "✓ 憑證與私鑰配對正確"
else
    echo "✗ 憑證與私鑰不配對"
fi

# 步驟 5: 生成包含 SAN 的憑證（支援多網域）
echo -e "\n[進階] 生成包含 SAN 擴展的憑證..."
cat > san.cnf <<EOF
[req]
distinguished_name = req_distinguished_name
req_extensions = v3_req

[req_distinguished_name]
countryName = Country Name (2 letter code)
stateOrProvinceName = State or Province Name
localityName = Locality Name
organizationName = Organization Name
commonName = Common Name

[v3_req]
subjectAltName = @alt_names

[alt_names]
DNS.1 = example.com
DNS.2 = www.example.com
DNS.3 = api.example.com
IP.1 = 192.168.1.100
EOF

# 使用配置檔生成憑證
openssl req -new -x509 -days 365 \
  -key server.key \
  -out server-san.crt \
  -config san.cnf \
  -extensions v3_req \
  -subj "/C=TW/ST=Taiwan/L=Taipei/O=Example Corp/CN=example.com"

echo "[驗證] SAN 憑證內容:"
openssl x509 -in server-san.crt -text -noout | grep -A 5 "Subject Alternative Name"

# 步驟 6: 生成 PKCS#12 格式（用於 Windows/Java）
echo -e "\n[格式轉換] 生成 PKCS#12 格式..."
openssl pkcs12 -export \
  -out server.p12 \
  -inkey server.key \
  -in server.crt \
  -password pass:changeit

echo -e "\n=== 憑證生成完成 ==="
echo "生成的檔案:"
ls -lh server.* san.cnf

# 安全建議
echo -e "\n=== 安全建議 ==="
echo "1. 私鑰檔案 (server.key) 應設定嚴格權限: chmod 600 server.key"
echo "2. 定期更新憑證，建議有效期不超過 90 天"
echo "3. 生產環境使用受信任 CA 簽發的憑證"
echo "4. 啟用 OCSP Stapling 提升效能與隱私"
echo "5. 使用 4096 位元或更高的 RSA 金鑰長度"

#!/usr/bin/env python3
"""
SSL/TLS 安全配置檢測工具

使用 Python 的 ssl 模組檢測目標伺服器的 SSL/TLS 配置
分析協定版本、密碼套件與憑證資訊
"""

import ssl
import socket
import sys
from datetime import datetime
from typing import Dict, List, Tuple

class SSLConfigAnalyzer:
    """SSL/TLS 配置分析器"""
    
    # 已知不安全的密碼套件關鍵字
    INSECURE_CIPHERS = [
        'NULL', 'EXPORT', 'DES', 'RC4', 'MD5',
        'PSK', 'SRP', 'CAMELLIA', 'IDEA', 'SEED'
    ]
    
    # 推薦的密碼套件（按優先級排序）
    RECOMMENDED_CIPHERS = [
        'TLS_AES_256_GCM_SHA384',
        'TLS_CHACHA20_POLY1305_SHA256',
        'TLS_AES_128_GCM_SHA256',
        'ECDHE-RSA-AES256-GCM-SHA384',
        'ECDHE-RSA-AES128-GCM-SHA256'
    ]
    
    def __init__(self, hostname: str, port: int = 443):
        self.hostname = hostname
        self.port = port
        self.results = {}
    
    def check_protocol_versions(self) -> Dict[str, bool]:
        """
        檢測支援的 SSL/TLS 協定版本
        
        Returns:
            包含各協定版本支援狀態的字典
        """
        protocols = {
            'SSLv2': ssl.PROTOCOL_SSLv23,  # 已棄用
            'SSLv3': ssl.PROTOCOL_SSLv23,  # 已棄用
            'TLSv1.0': ssl.PROTOCOL_TLSv1,
            'TLSv1.1': ssl.PROTOCOL_TLSv1_1,
            'TLSv1.2': ssl.PROTOCOL_TLSv1_2,
        }
        
        # TLS 1.3 在較新的 Python 版本中支援
        if hasattr(ssl, 'PROTOCOL_TLS'):
            protocols['TLSv1.3'] = ssl.PROTOCOL_TLS
        
        results = {}
        
        for name, protocol in protocols.items():
            try:
                context = ssl.SSLContext(protocol)
                context.check_hostname = False
                context.verify_mode = ssl.CERT_NONE
                
                with socket.create_connection((self.hostname, self.port), timeout=5) as sock:
                    with context.wrap_socket(sock, server_hostname=self.hostname) as ssock:
                        results[name] = True
                        
            except Exception as e:
                results[name] = False
        
        return results
    
    def get_cipher_suites(self) -> List[str]:
        """
        獲取伺服器支援的密碼套件
        
        Returns:
            密碼套件列表
        """
        try:
            context = ssl.create_default_context()
            
            with socket.create_connection((self.hostname, self.port), timeout=5) as sock:
                with context.wrap_socket(sock, server_hostname=self.hostname) as ssock:
                    return [cipher[0] for cipher in ssock.shared_ciphers() or []]
        
        except Exception as e:
            print(f"獲取密碼套件失敗: {e}")
            return []
    
    def analyze_cipher_security(self, ciphers: List[str]) -> Dict[str, List[str]]:
        """
        分析密碼套件的安全性
        
        Args:
            ciphers: 密碼套件列表
            
        Returns:
            分類後的密碼套件字典
        """
        secure = []
        weak = []
        insecure = []
        
        for cipher in ciphers:
            # 檢查是否為不安全的套件
            if any(keyword in cipher.upper() for keyword in self.INSECURE_CIPHERS):
                insecure.append(cipher)
            # 檢查是否為推薦的套件
            elif any(rec in cipher for rec in self.RECOMMENDED_CIPHERS):
                secure.append(cipher)
            else:
                weak.append(cipher)
        
        return {
            'secure': secure,
            'weak': weak,
            'insecure': insecure
        }
    
    def get_certificate_info(self) -> Dict:
        """
        獲取伺服器憑證資訊
        
        Returns:
            憑證資訊字典
        """
        try:
            context = ssl.create_default_context()
            
            with socket.create_connection((self.hostname, self.port), timeout=5) as sock:
                with context.wrap_socket(sock, server_hostname=self.hostname) as ssock:
                    cert = ssock.getpeercert()
                    
                    # 解析憑證資訊
                    info = {
                        'subject': dict(x[0] for x in cert['subject']),
                        'issuer': dict(x[0] for x in cert['issuer']),
                        'version': cert['version'],
                        'serial_number': cert['serialNumber'],
                        'not_before': cert['notBefore'],
                        'not_after': cert['notAfter'],
                        'san': cert.get('subjectAltName', [])
                    }
                    
                    # 檢查憑證有效期
                    not_after = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
                    days_to_expiry = (not_after - datetime.now()).days
                    info['days_to_expiry'] = days_to_expiry
                    
                    return info
        
        except Exception as e:
            print(f"獲取憑證資訊失敗: {e}")
            return {}
    
    def run_full_analysis(self) -> None:
        """執行完整的 SSL/TLS 配置分析"""
        print(f"\n{'='*60}")
        print(f"SSL/TLS 安全配置分析報告")
        print(f"目標: {self.hostname}:{self.port}")
        print(f"{'='*60}")
        
        # 檢測協定版本
        print("\n[協定版本支援]")
        protocols = self.check_protocol_versions()
        for name, supported in protocols.items():
            status = "✓ 支援" if supported else "✗ 不支援"
            security = ""
            
            if name in ['SSLv2', 'SSLv3', 'TLSv1.0', 'TLSv1.1']:
                security = " [不安全，應禁用]"
            elif name == 'TLSv1.3':
                security = " [推薦]"
            
            print(f"  {name}: {status}{security}")
        
        # 分析密碼套件
        print("\n[密碼套件分析]")
        ciphers = self.get_cipher_suites()
        
        if ciphers:
            analysis = self.analyze_cipher_security(ciphers)
            
            print(f"\n  安全套件 ({len(analysis['secure'])} 個):")
            for cipher in analysis['secure'][:5]:
                print(f"    ✓ {cipher}")
            
            if analysis['weak']:
                print(f"\n  弱安全套件 ({len(analysis['weak'])} 個):")
                for cipher in analysis['weak'][:3]:
                    print(f"    ⚠ {cipher}")
            
            if analysis['insecure']:
                print(f"\n  不安全套件 ({len(analysis['insecure'])} 個):")
                for cipher in analysis['insecure']:
                    print(f"    ✗ {cipher}")
        
        # 憑證資訊
        print("\n[憑證資訊]")
        cert_info = self.get_certificate_info()
        
        if cert_info:
            print(f"  主體: {cert_info['subject'].get('commonName', 'N/A')}")
            print(f"  簽發者: {cert_info['issuer'].get('commonName', 'N/A')}")
            print(f"  有效期: {cert_info['not_before']} 至 {cert_info['not_after']}")
            
            days = cert_info['days_to_expiry']
            if days < 0:
                print(f"  到期狀態: ✗ 已過期 {abs(days)} 天")
            elif days < 30:
                print(f"  到期狀態: ⚠ 即將到期 ({days} 天)")
            else:
                print(f"  到期狀態: ✓ 有效 (剩餘 {days} 天)")
            
            if cert_info['san']:
                print(f"  SAN: {', '.join(x[1] for x in cert_info['san'][:3])}")
        
        # 安全建議
        print("\n[安全建議]")
        if protocols.get('SSLv2') or protocols.get('SSLv3'):
            print("  ✗ 禁用 SSLv2/SSLv3 協定")
        if protocols.get('TLSv1.0') or protocols.get('TLSv1.1'):
            print("  ⚠ 考慮禁用 TLSv1.0/1.1")
        if not protocols.get('TLSv1.3'):
            print("  ⚠ 啟用 TLSv1.3 以獲得最佳安全性")
        
        if ciphers:
            analysis = self.analyze_cipher_security(ciphers)
            if analysis['insecure']:
                print("  ✗ 移除不安全的密碼套件")
            if not analysis['secure']:
                print("  ⚠ 配置安全的密碼套件")
        
        print(f"\n{'='*60}\n")

# 使用範例
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"使用方式: {sys.argv[0]} <hostname> [port]")
        sys.exit(1)
    
    hostname = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 443
    
    analyzer = SSLConfigAnalyzer(hostname, port)
    analyzer.run_full_analysis()

#!/usr/bin/env python3
"""
mitmproxy 基礎 SSL/TLS 流量分析工具

展示如何使用 mitmproxy 攔截與分析 HTTPS 流量
包含請求/回應記錄、標頭分析與內容檢查
"""

from mitmproxy import http, ctx
from typing import Optional
import json
import re

class HTTPSTrafficAnalyzer:
    """HTTPS 流量分析器"""
    
    def __init__(self):
        self.request_count = 0
        self.sensitive_patterns = {
            'email': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
            'credit_card': r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
            'phone': r'\b09\d{8}\b',
            'id_number': r'\b[A-Z]\d{9}\b'
        }
    
    def request(self, flow: http.HTTPFlow) -> None:
        """
        處理 HTTP/HTTPS 請求
        
        Args:
            flow: HTTP 流物件
        """
        self.request_count += 1
        request = flow.request
        
        # 記錄基本請求資訊
        ctx.log.info(f"\n{'='*60}")
        ctx.log.info(f"請求 #{self.request_count}")
        ctx.log.info(f"{'='*60}")
        ctx.log.info(f"方法: {request.method}")
        ctx.log.info(f"URL: {request.pretty_url}")
        ctx.log.info(f"Host: {request.host}")
        
        # 分析請求標頭
        ctx.log.info("\n[請求標頭]")
        for name, value in request.headers.items():
            # 標記敏感標頭
            is_sensitive = name.lower() in [
                'authorization', 'cookie', 'x-csrf-token'
            ]
            marker = "🔒" if is_sensitive else "  "
            ctx.log.info(f"{marker} {name}: {value}")
        
        # 檢查 Cookie
        if 'Cookie' in request.headers:
            self._analyze_cookies(request.headers['Cookie'])
        
        # 分析請求內容
        if request.content:
            self._analyze_request_content(request)
    
    def response(self, flow: http.HTTPFlow) -> None:
        """
        處理 HTTP/HTTPS 回應
        
        Args:
            flow: HTTP 流物件
        """
        response = flow.response
        
        ctx.log.info(f"\n[回應資訊]")
        ctx.log.info(f"狀態碼: {response.status_code}")
        ctx.log.info(f"內容類型: {response.headers.get('Content-Type', 'N/A')}")
        ctx.log.info(f"內容長度: {len(response.content)} bytes")
        
        # 檢查安全標頭
        self._check_security_headers(response)
        
        # 分析回應內容
        if response.content:
            self._analyze_response_content(response)
        
        ctx.log.info(f"{'='*60}\n")
    
    def _analyze_cookies(self, cookie_header: str) -> None:
        """
        分析 Cookie 標頭
        
        Args:
            cookie_header: Cookie 標頭值
        """
        ctx.log.info("\n[Cookie 分析]")
        cookies = cookie_header.split('; ')
        
        for cookie in cookies:
            if '=' in cookie:
                name, value = cookie.split('=', 1)
                # 檢查 Cookie 屬性
                is_secure = 'Secure' in cookie
                is_httponly = 'HttpOnly' in cookie
                is_samesite = 'SameSite' in cookie
                
                security_flags = []
                if is_secure:
                    security_flags.append("Secure")
                if is_httponly:
                    security_flags.append("HttpOnly")
                if is_samesite:
                    security_flags.append("SameSite")
                
                flags_str = f" [{', '.join(security_flags)}]" if security_flags else " [無安全屬性]"
                ctx.log.info(f"  {name}: {value[:20]}...{flags_str}")
    
    def _analyze_request_content(self, request: http.Request) -> None:
        """
        分析請求內容
        
        Args:
            request: HTTP 請求物件
        """
        content_type = request.headers.get('Content-Type', '')
        
        if 'application/json' in content_type:
            try:
                data = json.loads(request.content)
                ctx.log.info("\n[JSON 請求資料]")
                ctx.log.info(json.dumps(data, indent=2, ensure_ascii=False))
                
                # 檢查敏感資料
                self._detect_sensitive_data(json.dumps(data))
                
            except json.JSONDecodeError:
                pass
        
        elif 'application/x-www-form-urlencoded' in content_type:
            ctx.log.info("\n[表單資料]")
            try:
                body = request.content.decode('utf-8')
                params = body.split('&')
                for param in params:
                    if '=' in param:
                        key, value = param.split('=', 1)
                        ctx.log.info(f"  {key}: {value}")
            except UnicodeDecodeError:
                pass
    
    def _analyze_response_content(self, response: http.Response) -> None:
        """
        分析回應內容
        
        Args:
            response: HTTP 回應物件
        """
        content_type = response.headers.get('Content-Type', '')
        
        if 'application/json' in content_type:
            try:
                data = json.loads(response.content)
                ctx.log.info("\n[JSON 回應資料]")
                
                # 只顯示前幾個欄位以避免輸出過長
                if isinstance(data, dict):
                    preview = {k: v for k, v in list(data.items())[:3]}
                    ctx.log.info(json.dumps(preview, indent=2, ensure_ascii=False))
                    if len(data) > 3:
                        ctx.log.info(f"  ... (還有 {len(data) - 3} 個欄位)")
                
            except json.JSONDecodeError:
                pass
    
    def _check_security_headers(self, response: http.Response) -> None:
        """
        檢查安全相關的 HTTP 標頭
        
        Args:
            response: HTTP 回應物件
        """
        security_headers = {
            'Strict-Transport-Security': 'HSTS',
            'Content-Security-Policy': 'CSP',
            'X-Frame-Options': 'Clickjacking 防護',
            'X-Content-Type-Options': 'MIME 嗅探防護',
            'X-XSS-Protection': 'XSS 防護',
            'Referrer-Policy': 'Referrer 政策'
        }
        
        ctx.log.info("\n[安全標頭檢查]")
        for header, description in security_headers.items():
            if header in response.headers:
                ctx.log.info(f"  ✓ {description}: {response.headers[header]}")
            else:
                ctx.log.info(f"  ✗ 缺少 {description}")
    
    def _detect_sensitive_data(self, content: str) -> None:
        """
        檢測敏感資料
        
        Args:
            content: 要檢測的內容
        """
        ctx.log.info("\n[敏感資料檢測]")
        found_any = False
        
        for data_type, pattern in self.sensitive_patterns.items():
            matches = re.findall(pattern, content)
            if matches:
                found_any = True
                ctx.log.warn(f"  ⚠ 偵測到 {data_type}: {len(matches)} 處")
                for match in matches[:2]:  # 只顯示前兩個
                    masked = match[:3] + '*' * (len(match) - 6) + match[-3:] if len(match) > 6 else '***'
                    ctx.log.warn(f"    - {masked}")
        
        if not found_any:
            ctx.log.info("  ✓ 未偵測到已知敏感資料模式")

# mitmproxy 插件註冊
addons = [
    HTTPSTrafficAnalyzer()
]