SDU(Service Data Unit, 서비스 데이터 단위)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

상황: 웹 브라우저가 HTTP 요청을 전송

L7 (Application):
  HTTP Request (100 bytes) 
  → L7-SDU (100 bytes) for Transport Layer

L4 (Transport):
  받은 L4-SDU (100 bytes) + TCP Header (20 bytes)
  → L4-PDU (TCP Segment, 120 bytes)
  → L4-PDU는 동시에 L3-SDU가 됨

L3 (Network):
  받은 L3-SDU (120 bytes) + IP Header (20 bytes)
  → L3-PDU (IP Packet, 140 bytes)

1
2
3
4
5
6
7
8
9

L7: HTTP Request (200 bytes)
    ↓ (L7-SDU for L4)
L4: [TCP Header 20B] + [L4-SDU 200B] = TCP Segment (220 bytes)
    ↓ (L4-PDU = L3-SDU)
L3: [IP Header 20B] + [L3-SDU 220B] = IP Packet (240 bytes)
    ↓ (L3-PDU = L2-SDU)
L2: [Eth Header 14B] + [L2-SDU 240B] + [FCS 4B] = Ethernet Frame (258 bytes)
    ↓ (L2-PDU = L1-SDU)
L1: 258 bytes → 2064 bits (Physical transmission)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

# 송신 측: SDU 분할
def segment_sdu(sdu_data, mss):
    """
    SDU를 MSS 크기에 맞게 분할
    
    Args:
        sdu_data: 원본 SDU 바이트 배열
        mss: Maximum Segment Size
    
    Returns:
        세그먼트 리스트 (각 세그먼트는 dict로 seq_num과 data 포함)
    """
    segments = []
    seq_num = 0
    
    for i in range(0, len(sdu_data), mss):
        segment_data = sdu_data[i:i+mss]
        segments.append({
            'seq_num': seq_num,
            'data': segment_data,
            'length': len(segment_data)
        })
        seq_num += len(segment_data)
    
    return segments

# 수신 측: PDU 재조립
def reassemble_segments(segments):
    """
    수신한 세그먼트를 원본 SDU로 재조립
    
    Args:
        segments: 수신한 세그먼트 리스트
    
    Returns:
        재조립된 원본 SDU
    """
    # 시퀀스 번호로 정렬
    sorted_segments = sorted(segments, key=lambda x: x['seq_num'])
    
    # 연속성 검증
    expected_seq = 0
    for seg in sorted_segments:
        if seg['seq_num'] != expected_seq:
            raise Exception(f"Missing segment at seq {expected_seq}")
        expected_seq += seg['length']
    
    # 데이터 결합
    reassembled_sdu = b''.join(seg['data'] for seg in sorted_segments)
    return reassembled_sdu

# 사용 예시
original_sdu = b'A' * 10000  # 10KB 데이터
mss = 1460

# 송신
segments = segment_sdu(original_sdu, mss)
print(f"분할된 세그먼트 수: {len(segments)}")  # 출력: 7

# 수신 (역순으로 도착 가정)
import random
random.shuffle(segments)
recovered_sdu = reassemble_segments(segments)

assert original_sdu == recovered_sdu, "재조립 실패"
print("재조립 성공")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

클라이언트 → 웹 서버 HTTP 요청

L4 (Transport):
  Source TSAP: 클라이언트 Port 54321
  Dest TSAP: 서버 Port 80 (HTTP)

L3 (Network):
  Source NSAP: 클라이언트 IP 192.168.1.100
  Dest NSAP: 서버 IP 203.0.113.10

L2 (Data Link):
  Source LSAP: 클라이언트 MAC aa:bb:cc:dd:ee:ff
  Dest LSAP: 게이트웨이 MAC 00:11:22:33:44:55

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

# 전송 계층 서비스 요청 (의사 코드)
def transport_send_request(sdu, ici):
    """
    상위 계층이 전송 계층에 데이터 전송 요청
    
    Args:
        sdu: L4-SDU (애플리케이션 데이터)
        ici: 인터페이스 제어 정보 (dict)
            - dest_address: 목적지 주소
            - dest_port: 목적지 포트
            - priority: 우선순위 (0-7)
            - qos: QoS 플래그 (LOW_LATENCY, HIGH_THROUGHPUT 등)
            - timeout: 타임아웃 (초)
    """
    # ICI 파라미터 추출
    dest_addr = ici['dest_address']
    dest_port = ici['dest_port']
    priority = ici.get('priority', 0)
    qos = ici.get('qos', 'BEST_EFFORT')
    
    # TCP 헤더 생성 (PCI)
    tcp_header = create_tcp_header(
        dest_port=dest_port,
        flags=get_flags_from_qos(qos)
    )
    
    # PDU 생성 (PCI + SDU)
    tcp_segment = tcp_header + sdu
    
    # 네트워크 계층에 전달 (새로운 ICI)
    network_send_request(
        sdu=tcp_segment,  # L3-SDU가 됨
        ici={
            'dest_address': dest_addr,
            'protocol': 'TCP',
            'tos': map_priority_to_tos(priority)  # ICI → L3 ICI 매핑
        }
    )

# 사용 예시
app_data = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"

transport_send_request(
    sdu=app_data,
    ici={
        'dest_address': '203.0.113.10',
        'dest_port': 80,
        'priority': 5,
        'qos': 'LOW_LATENCY'
    }
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

L7 → L4: 
  SDU: HTTP 요청 데이터
  ICI: { priority: HIGH, latency: LOW }

L4 → L3:
  SDU: TCP 세그먼트 (L4-PDU)
  ICI: { tos: 0xE0 (high priority), protocol: TCP }
  → L4는 상위 ICI를 TCP 헤더(PCI)와 L3 ICI로 변환

L3 → L2:
  SDU: IP 패킷 (L3-PDU)
  ICI: { vlan_priority: 5 }
  → L3는 상위 ICI를 IP 헤더(PCI)와 L2 ICI로 변환

L2 → L1:
  SDU: 이더넷 프레임 (L2-PDU)
  ICI: { signal_strength: HIGH }
  → L2는 상위 ICI를 이더넷 헤더(PCI)와 물리 파라미터로 변환

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# 현재 MTU 확인
ip link show eth0

# MTU 변경 (임시, 재부팅 시 초기화)
sudo ip link set eth0 mtu 9000

# TCP MSS 확인
ss -i | grep mss

# iptables로 MSS Clamping (VPN/터널 환경)
sudo iptables -A FORWARD -p tcp --tcp-flags SYN,RST SYN \
  -j TCPMSS --clamp-mss-to-pmtu

# Path MTU Discovery 테스트 (ping)
# -M do: DF(Don't Fragment) 플래그 설정
# -s: 패킷 크기 (ICMP 헤더 8B 제외)
ping -M do -s 1472 example.com  # 성공하면 MTU ≥ 1500
ping -M do -s 1473 example.com  # 실패하면 MTU < 1501

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import socket

def create_optimized_socket(dest_ip, dest_port, custom_mss=None):
    """
    MSS가 최적화된 TCP 소켓 생성
    
    Args:
        dest_ip: 목적지 IP
        dest_port: 목적지 포트
        custom_mss: 사용자 정의 MSS (None이면 기본값)
    
    Returns:
        설정된 소켓 객체
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # TCP_MAXSEG 옵션으로 MSS 설정 (일부 OS에서만 지원)
    if custom_mss and hasattr(socket, 'TCP_MAXSEG'):
        try:
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG, custom_mss)
            print(f"MSS를 {custom_mss}로 설정")
        except OSError as e:
            print(f"MSS 설정 실패: {e}")
    
    # Don't Fragment 플래그 설정 (PMTUD 활성화)
    if hasattr(socket, 'IP_MTU_DISCOVER'):
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MTU_DISCOVER, 2)  # IP_PMTUDISC_DO
    
    sock.connect((dest_ip, dest_port))
    
    # 협상된 MSS 확인 (Linux)
    if hasattr(socket, 'TCP_MAXSEG'):
        try:
            negotiated_mss = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG)
            print(f"협상된 MSS: {negotiated_mss}")
        except OSError:
            pass
    
    return sock

# 사용 예시
sock = create_optimized_socket('203.0.113.10', 80, custom_mss=1400)
sock.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
response = sock.recv(4096)
sock.close()

1
2
3
4
5
6
7
8
9

시나리오: 1 byte 응용 데이터 전송 (예: 키 입력)

L7-SDU: 1 byte
L4-PDU: 1 + 20 (TCP) = 21 bytes
L3-PDU: 21 + 20 (IP) = 41 bytes
L2-PDU: 41 + 14 (Eth) + 4 (FCS) = 59 bytes

페이로드 효율성 = 1 / 59 = 1.7%
오버헤드 = 58 / 1 = 5800%

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# TCP 스트림 필터 (특정 연결의 SDU 추적)
tcp.stream eq 0

# 특정 포트의 SDU 크기 분석
tcp.len > 0 and tcp.port == 80

# 분할된 IP 패킷 찾기
ip.flags.mf == 1 or ip.frag_offset > 0

# TCP 재전송 식별
tcp.analysis.retransmission

# MSS 협상 확인 (SYN 패킷)
tcp.flags.syn == 1 and tcp.options.mss

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from scapy.all import sniff, TCP, IP

def analyze_sdu(packet):
    """
    패킷에서 SDU 정보 추출 및 분석
    
    Args:
        packet: Scapy 패킷 객체
    """
    if packet.haslayer(TCP) and packet.haslayer(IP):
        ip_layer = packet[IP]
        tcp_layer = packet[TCP]
        
        # L3-SDU 크기 (IP 페이로드 = TCP 세그먼트)
        l3_sdu_size = ip_layer.len - (ip_layer.ihl * 4)  # IP total length - IP header
        
        # L4-SDU 크기 (TCP 페이로드)
        tcp_header_size = tcp_layer.dataofs * 4
        l4_sdu_size = l3_sdu_size - tcp_header_size
        
        # 오버헤드 계산
        overhead = tcp_header_size + (ip_layer.ihl * 4) + 14 + 4  # TCP + IP + Eth Header/FCS
        efficiency = l4_sdu_size / (l4_sdu_size + overhead) * 100 if l4_sdu_size > 0 else 0
        
        if l4_sdu_size > 0:  # 페이로드가 있는 경우만
            print(f"[{ip_layer.src}:{tcp_layer.sport} → {ip_layer.dst}:{tcp_layer.dport}]")
            print(f"  L4-SDU: {l4_sdu_size} bytes")
            print(f"  Total: {ip_layer.len + 14 + 4} bytes")
            print(f"  Efficiency: {efficiency:f}%")
            print(f"  TCP Flags: {tcp_layer.flags}")
            
            # 비정상 패턴 감지
            if l4_sdu_size < 10 and efficiency < 10:
                print(f"  ⚠️ 매우 작은 SDU (Nagle 비활성화 가능성)")
            if ip_layer.flags.MF or ip_layer.frag_offset > 0:
                print(f"  ⚠️ IP 단편화 발생")
            if tcp_layer.flags & 0x08:  # PSH 플래그
                print(f"  📌 푸시 데이터 (즉시 전달)")

# 실시간 모니터링 시작
print("SDU 모니터링 시작 (Ctrl+C로 중지)…")
sniff(filter="tcp", prn=analyze_sdu, store=0)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

Protocol: MyProtocol (Hypothetical)
Layer: N
Service: 신뢰성 있는 메시지 전달

1. SDU 정의
   - N-SDU: 가변 길이 바이트 배열
   - 최소 크기: 1 byte
   - 최대 크기: 65,535 bytes
   - 인코딩: 8-bit 바이너리

2. PDU 구조
   N-PDU = N-PCI + N-SDU
   
   N-PCI 필드:
   - Version (4 bits): 프로토콜 버전
   - Length (16 bits): SDU 길이 (bytes)
   - Sequence (32 bits): 순서 번호
   - Flags (8 bits): SYN, ACK, FIN, RST
   - Checksum (16 bits): CRC-16

3. 서비스 프리미티브
   - N-DATA.req(dest_addr, sdu, priority)
   - N-DATA.ind(src_addr, sdu)
   - N-DATA-ACK.req(seq_num)
   - N-DATA-ACK.cnf(seq_num, status)

4. SDU 분할
   - 최대 전송 단위: (하위 계층 MTU - N-PCI 크기)
   - 분할 시 Sequence 번호 증분
   - 재조립 버퍼: 10 × MTU

5. 에러 처리
   - Checksum 실패: SDU 폐기, NACK 전송
   - 순서 오류: 재정렬 버퍼 사용, 타임아웃 5초
   - 중복: Sequence 번호로 감지 및 제거

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

class LayerNService:
    """
    N 계층 서비스 인터페이스 (추상 클래스)
    """
    
    def data_request(self, dest_addr, sdu, ici=None):
        """
        N-DATA.request 프리미티브
        
        Args:
            dest_addr: 목적지 SAP 주소
            sdu: N-SDU (bytes)
            ici: 인터페이스 제어 정보 (dict)
                - priority: 우선순위 (0-7)
                - qos: QoS 플래그
                - timeout: 타임아웃 (초)
        
        Returns:
            request_id: 요청 식별자
        
        Raises:
            ValueError: SDU 크기 초과
            ConnectionError: 연결 실패
        """
        # SDU 크기 검증
        if len(sdu) > self.MAX_SDU_SIZE:
            raise ValueError(f"SDU too large: {len(sdu)} > {self.MAX_SDU_SIZE}")
        
        # ICI 기본값 설정
        if ici is None:
            ici = {'priority': 0, 'qos': 'BEST_EFFORT'}
        
        # PDU 생성 (PCI + SDU)
        pdu = self._create_pdu(dest_addr, sdu, ici)
        
        # 하위 계층에 전달
        return self._send_to_lower_layer(pdu)
    
    def data_indication(self, src_addr, sdu):
        """
        N-DATA.indication 프리미티브
        
        Args:
            src_addr: 송신 SAP 주소
            sdu: N-SDU (bytes)
        """
        # 상위 계층에 전달
        self._deliver_to_upper_layer(src_addr, sdu)
    
    def _create_pdu(self, dest_addr, sdu, ici):
        """
        PDU 생성 (PCI + SDU)
        """
        pci = self._create_pci(dest_addr, len(sdu), ici)
        return pci + sdu
    
    def _create_pci(self, dest_addr, sdu_length, ici):
        """
        프로토콜 제어 정보 생성
        """
        # 실제 구현에서는 프로토콜 규격에 따라 헤더 생성
        pass
    
    def _send_to_lower_layer(self, pdu):
        """
        하위 계층 서비스 호출
        """
        # (N-1)-DATA.request(…) 호출
        pass
    
    def _deliver_to_upper_layer(self, src_addr, sdu):
        """
        상위 계층에 indication 전달
        """
        # 콜백 또는 이벤트로 상위 계층에 알림
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# TCP 버퍼 크기 증가 (BDP 기반)
# 예: 1 Gbps 네트워크, RTT 100ms → BDP = 12.5 MB
sudo sysctl -w net.core.rmem_max=134217728  # 128 MB
sudo sysctl -w net.core.wmem_max=134217728
sudo sysctl -w net.ipv4.tcp_rmem="4096 87380 134217728"
sudo sysctl -w net.ipv4.tcp_wmem="4096 65536 134217728"

# TCP Window Scaling 활성화
sudo sysctl -w net.ipv4.tcp_window_scaling=1

# Nagle 알고리즘 제어 (애플리케이션 레벨)
# Python에서: sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

# MTU 설정 (Jumbo Frame)
sudo ip link set eth0 mtu 9000

# QoS 마킹 (iptables)
sudo iptables -t mangle -A OUTPUT -p tcp --dport 22 \
  -j DSCP --set-dscp-class EF  # SSH 트래픽 우선순위

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

def calculate_optimal_buffer(bandwidth_mbps, rtt_ms):
    """
    최적 TCP 버퍼 크기 계산
    
    Args:
        bandwidth_mbps: 대역폭 (Mbps)
        rtt_ms: 왕복 지연 시간 (ms)
    
    Returns:
        최적 버퍼 크기 (bytes)
    """
    # BDP = Bandwidth × RTT
    bandwidth_bps = bandwidth_mbps * 1_000_000
    rtt_sec = rtt_ms / 1000
    bdp_bits = bandwidth_bps * rtt_sec
    bdp_bytes = bdp_bits / 8
    
    # 안전 마진 추가 (2x)
    optimal_buffer = int(bdp_bytes * 2)
    
    return optimal_buffer

# 사용 예시
print("최적 버퍼 크기 계산기")
print("-" * 40)

scenarios = [
    ("로컬 네트워크", 1000, 1),
    ("광역 네트워크", 100, 50),
    ("인터넷", 10, 100),
    ("위성 링크", 1, 600)
]

for name, bw, rtt in scenarios:
    buffer_size = calculate_optimal_buffer(bw, rtt)
    print(f"{name}: {bw} Mbps, RTT {rtt}ms")
    print(f"  권장 버퍼: {buffer_size:,} bytes ({buffer_size/1024/1024:f} MB)")
    print()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

로컬 네트워크: 1000 Mbps, RTT 1ms
  권장 버퍼: 250,000 bytes (0.2 MB)

광역 네트워크: 100 Mbps, RTT 50ms
  권장 버퍼: 1,250,000 bytes (1.2 MB)

인터넷: 10 Mbps, RTT 100ms
  권장 버퍼: 250,000 bytes (0.2 MB)

위성 링크: 1 Mbps, RTT 600ms
  권장 버퍼: 150,000 bytes (0.1 MB)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

Application (L7):
  평문 SDU (민감 데이터)
      ↓
TLS (L6/L7):
  암호화된 SDU
  [TLS Header] + Encrypted(평문 SDU) + MAC
      ↓
TCP (L4):
  TCP SDU = 암호화된 TLS 레코드
  [TCP Header] + TLS PDU
      ↓
IP (L3):
  IP SDU = TCP 세그먼트
  [IP Header] + TCP PDU

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

class SecureSDUHandler:
    """
    보안이 강화된 SDU 처리기
    """
    
    MIN_SDU_SIZE = 1
    MAX_SDU_SIZE = 65535
    
    def validate_sdu(self, sdu):
        """
        SDU 크기 및 내용 검증
        
        Args:
            sdu: SDU 바이트 배열
        
        Raises:
            ValueError: 크기 초과/미달
            SecurityError: 악의적 패턴 탐지
        """
        # 크기 검증
        if len(sdu) < self.MIN_SDU_SIZE:
            raise ValueError(f"SDU too small: {len(sdu)} < {self.MIN_SDU_SIZE}")
        if len(sdu) > self.MAX_SDU_SIZE:
            raise ValueError(f"SDU too large: {len(sdu)} > {self.MAX_SDU_SIZE}")
        
        # 악의적 패턴 검증 (예: 반복 바이트)
        if self._detect_suspicious_pattern(sdu):
            raise SecurityError("Suspicious SDU pattern detected")
        
        return True
    
    def _detect_suspicious_pattern(self, sdu):
        """
        악의적 패턴 탐지 (간단한 예시)
        """
        # 동일 바이트 90% 이상 반복 시 의심
        if len(sdu) > 100:
            most_common = max(set(sdu), key=sdu.count)
            count = sdu.count(most_common)
            if count / len(sdu) > 0.9:
                return True
        return False
    
    def encrypt_sdu(self, sdu, key):
        """
        SDU 암호화 (TLS/IPSec 대신 간단한 예시)
        
        Args:
            sdu: 평문 SDU
            key: 암호화 키
        
        Returns:
            암호화된 SDU
        """
        from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
        from cryptography.hazmat.backends import default_backend
        import os
        
        # IV (Initialization Vector) 생성
        iv = os.urandom(16)
        
        # AES-256-CBC 암호화
        cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
        encryptor = cipher.encryptor()
        
        # 패딩 (PKCS7)
        padded_sdu = self._pkcs7_pad(sdu, 16)
        
        # 암호화
        encrypted = encryptor.update(padded_sdu) + encryptor.finalize()
        
        # IV + 암호문 반환
        return iv + encrypted
    
    def _pkcs7_pad(self, data, block_size):
        """
        PKCS7 패딩
        """
        padding_length = block_size - (len(data) % block_size)
        padding = bytes([padding_length] * padding_length)
        return data + padding

# 사용 예시
handler = SecureSDUHandler()

# 정상 SDU
normal_sdu = b"Hello, World!"
handler.validate_sdu(normal_sdu)  # 통과

# 비정상 SDU (너무 큼)
try:
    huge_sdu = b"A" * 100000
    handler.validate_sdu(huge_sdu)
except ValueError as e:
    print(f"검증 실패: {e}")

# SDU 암호화
key = os.urandom(32)  # 256-bit 키
encrypted_sdu = handler.encrypt_sdu(normal_sdu, key)
print(f"암호화된 SDU 크기: {len(encrypted_sdu)} bytes")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

from scapy.all import rdpcap, TCP, Raw

def extract_tcp_stream_sdu(pcap_file, stream_id=0):
    """
    PCAP 파일에서 특정 TCP 스트림의 SDU를 추출
    
    Args:
        pcap_file: PCAP 파일 경로
        stream_id: 추출할 TCP 스트림 ID (0부터 시작)
    
    Returns:
        client_data, server_data: 클라이언트/서버 SDU (bytes)
    """
    packets = rdpcap(pcap_file)
    
    # TCP 스트림 식별 (src/dst IP:Port 쌍)
    streams = {}
    stream_counter = 0
    
    for packet in packets:
        if packet.haslayer(TCP):
            tcp = packet[TCP]
            ip = packet['IP']
            
            # 스트림 키 생성 (정규화: 작은 포트를 앞에)
            key = tuple(sorted([
                (ip.src, tcp.sport),
                (ip.dst, tcp.dport)
            ]))
            
            if key not in streams:
                streams[key] = {'id': stream_counter, 'packets': []}
                stream_counter += 1
            
            streams[key]['packets'].append(packet)
    
    # 요청한 스트림 찾기
    target_stream = None
    for key, stream in streams.items():
        if stream['id'] == stream_id:
            target_stream = stream
            break
    
    if not target_stream:
        raise ValueError(f"Stream {stream_id} not found")
    
    # SDU 추출 (TCP 페이로드)
    client_data = b''
    server_data = b''
    
    first_packet = target_stream['packets'][0]
    client_addr = (first_packet['IP'].src, first_packet[TCP].sport)
    
    for packet in target_stream['packets']:
        if packet.haslayer(Raw):
            payload = bytes(packet[Raw])
            
            # 방향 판별
            if (packet['IP'].src, packet[TCP].sport) == client_addr:
                client_data += payload
            else:
                server_data += payload
    
    return client_data, server_data

# 사용 예시
pcap_file = 'capture.pcap'

print("TCP 스트림 SDU 추출")
print("=" * 50)

client_sdu, server_sdu = extract_tcp_stream_sdu(pcap_file, stream_id=0)

print(f"클라이언트 SDU: {len(client_sdu)} bytes")
print(f"서버 SDU: {len(server_sdu)} bytes")
print()

# HTTP 요청 파싱 (간단한 예시)
if client_sdu.startswith(b'GET') or client_sdu.startswith(b'POST'):
    print("HTTP 요청:")
    print(client_sdu.decode('utf-8', errors='ignore')[:200])
    print()

# HTTP 응답 파싱
if server_sdu.startswith(b'HTTP/'):
    print("HTTP 응답:")
    print(server_sdu.decode('utf-8', errors='ignore')[:200])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

#!/bin/bash
# MTU 최적화 도구

TARGET_HOST="example.com"
INTERFACE="eth0"

echo "=== MTU 최적화 도구 ==="
echo ""

# 1. 현재 MTU 확인
echo "[1] 현재 MTU 확인"
current_mtu=$(ip link show $INTERFACE | grep -oP 'mtu \K[0-9]+')
echo "  인터페이스 $INTERFACE MTU: $current_mtu"
echo ""

# 2. Path MTU Discovery (바이너리 서치)
echo "[2] Path MTU Discovery (대상: $TARGET_HOST)"
min_mtu=68
max_mtu=1500
optimal_mtu=$min_mtu

while [ $min_mtu -le $max_mtu ]; do
    mid=$(( (min_mtu + max_mtu) / 2 ))
    # ICMP 헤더 8 bytes 제외
    payload_size=$(( mid - 28 ))
    
    echo -n "  테스트 MTU $mid … "
    
    # Don't Fragment 플래그로 ping
    if ping -M do -c 1 -W 1 -s $payload_size $TARGET_HOST > /dev/null 2>&1; then
        echo "성공"
        optimal_mtu=$mid
        min_mtu=$(( mid + 1 ))
    else
        echo "실패 (단편화 필요)"
        max_mtu=$(( mid - 1 ))
    fi
done

echo ""
echo "  최적 MTU: $optimal_mtu"
echo ""

# 3. MTU 설정 변경 제안
if [ $optimal_mtu -lt $current_mtu ]; then
    echo "[3] MTU 변경 권장"
    echo "  현재: $current_mtu → 권장: $optimal_mtu"
    echo ""
    echo "  변경 명령어:"
    echo "    sudo ip link set $INTERFACE mtu $optimal_mtu"
    echo ""
    
    # 사용자 확인
    read -p "  지금 변경하시겠습니까? (y/n): " confirm
    if [ "$confirm" = "y" ]; then
        sudo ip link set $INTERFACE mtu $optimal_mtu
        echo "  MTU 변경 완료"
    fi
else
    echo "[3] 현재 MTU가 최적값입니다."
fi

echo ""

# 4. MSS Clamping 확인
echo "[4] TCP MSS Clamping 확인"
optimal_mss=$(( optimal_mtu - 40 ))  # IP(20) + TCP(20)
echo "  최적 MSS: $optimal_mss"
echo ""
echo "  MSS Clamping 설정 (VPN/터널 환경 권장):"
echo "    sudo iptables -A FORWARD -p tcp --tcp-flags SYN,RST SYN \\"
echo "      -j TCPMSS --set-mss $optimal_mss"
echo ""

# 5. 성능 테스트 제안
echo "[5] 성능 테스트 명령어"
echo "  처리량 측정:"
echo "    iperf3 -c $TARGET_HOST"
echo ""
echo "  지연 측정:"
echo "    ping -c 100 $TARGET_HOST | tail -1"
echo ""

echo "=== 완료 ==="