네트워크 엔지니어링 & 보안 완전 정복: TCP/IP부터 서비스 메시, AI 서빙 네트워크까지

HTTP/1.1:  [Req1] → [Res1] → [Req2] → [Res2]  (순차)
HTTP/2:    [Req1, Req2, Req3] → [Res1, Res2, Res3]  (멀티플렉싱, 단일 TCP)
HTTP/3:    [Req1, Req2, Req3] → [Res1, Res2, Res3]  (멀티플렉싱, 독립 QUIC 스트림)

Client                          Server
  |--- ClientHello (키 교환) ---->|
  |<-- ServerHello + Certificate-|
  |<-- + EncryptedExtensions -----|
  |--- Finished ----------------->|
  |<-> 암호화된 데이터 교환 <------>|

import asyncio

async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    addr = writer.get_extra_info('peername')
    print(f"연결됨: {addr}")

    try:
        while True:
            data = await reader.read(1024)
            if not data:
                break
            message = data.decode('utf-8').strip()
            print(f"수신: {message} from {addr}")

            # 에코 응답
            response = f"Echo: {message}\n"
            writer.write(response.encode('utf-8'))
            await writer.drain()
    except asyncio.IncompleteReadError:
        pass
    finally:
        print(f"연결 종료: {addr}")
        writer.close()
        await writer.wait_closed()

async def main():
    server = await asyncio.start_server(
        handle_client, '0.0.0.0', 8888
    )
    addr = server.sockets[0].getsockname()
    print(f"서버 시작: {addr}")

    async with server:
        await server.serve_forever()

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
import aiohttp
from typing import List

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
        return {
            'url': url,
            'status': response.status,
            'body': await response.text()
        }

async def fetch_all(urls: List[str]) -> List[dict]:
    connector = aiohttp.TCPConnector(
        limit=100,          # 최대 동시 연결 수
        limit_per_host=10,  # 호스트당 최대 연결
        keepalive_timeout=30
    )
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 실행
urls = [f"https://httpbin.org/get?id={i}" for i in range(20)]
results = asyncio.run(fetch_all(urls))

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: ml-inference-vs
  namespace: production
spec:
  hosts:
    - ml-inference-svc
  http:
    - match:
        - headers:
            x-model-version:
              exact: 'v2'
      route:
        - destination:
            host: ml-inference-svc
            subset: v2
          weight: 100
    - route:
        - destination:
            host: ml-inference-svc
            subset: v1
          weight: 90
        - destination:
            host: ml-inference-svc
            subset: v2
          weight: 10

Root CA (최상위 인증기관, 브라우저/OS에 내장)
  └── Intermediate CA (중간 인증기관)
        └── Leaf Certificate (실제 서버 인증서)

일반 TLS:  Client → [서버 인증서 검증] → Server
mTLS:      Client ↔ [양방향 인증서 검증] ↔ Server

server {
    listen 443 ssl http2;
    server_name api.example.com;

    ssl_certificate     /etc/ssl/certs/server.crt;
    ssl_certificate_key /etc/ssl/private/server.key;

    # TLS 1.2 이상만 허용
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256;
    ssl_prefer_server_ciphers off;

    # HSTS (1년)
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;

    # OCSP Stapling
    ssl_stapling on;
    ssl_stapling_verify on;

    location /api/ {
        proxy_pass http://backend_pool;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

export default {
  async fetch(request, env) {
    const url = new URL(request.url)

    // 엣지에서 AI 추론 라우팅
    if (url.pathname.startsWith('/inference/')) {
      const modelId = url.searchParams.get('model') || 'default'

      // 가장 가까운 GPU 클러스터로 라우팅
      const region = request.cf.region
      const backendUrl = selectBackend(region, modelId)

      return fetch(backendUrl, {
        method: request.method,
        headers: request.headers,
        body: request.body,
      })
    }

    // 정적 자산 캐싱
    const cache = caches.default
    const cachedResponse = await cache.match(request)
    if (cachedResponse) return cachedResponse

    const response = await fetch(request)
    // 성공 응답만 캐싱
    if (response.status === 200) {
      const responseToCache = response.clone()
      await cache.put(request, responseToCache)
    }
    return response
  },
}

syntax = "proto3";
package inference;

service InferenceService {
  rpc Predict(PredictRequest) returns (PredictResponse);
  rpc StreamPredict(PredictRequest) returns (stream PredictResponse);
}

message PredictRequest {
  string model_name = 1;
  repeated float input_data = 2;
  map<string, string> metadata = 3;
}

message PredictResponse {
  repeated float output_data = 1;
  float confidence = 2;
  int64 latency_ms = 3;
}

import grpc
from concurrent import futures
import inference_pb2
import inference_pb2_grpc
import numpy as np
import time

class InferenceServicer(inference_pb2_grpc.InferenceServiceServicer):
    def Predict(self, request, context):
        start = time.time()
        input_array = np.array(request.input_data)

        # 실제 모델 추론 (예시)
        output = input_array * 2.0

        latency = int((time.time() - start) * 1000)
        return inference_pb2.PredictResponse(
            output_data=output.tolist(),
            confidence=0.95,
            latency_ms=latency
        )

    def StreamPredict(self, request, context):
        # 스트리밍 응답 (LLM 토큰 생성 등에 활용)
        tokens = ["안녕", "하세요", "!", " gRPC", " 스트리밍", "입니다."]
        for token in tokens:
            yield inference_pb2.PredictResponse(
                output_data=[float(ord(c)) for c in token],
                confidence=0.9,
                latency_ms=10
            )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    inference_pb2_grpc.add_InferenceServiceServicer_to_server(
        InferenceServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    print("gRPC 서버 시작: 포트 50051")
    server.wait_for_termination()

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json

app = FastAPI()

async def llm_token_generator(prompt: str):
    """LLM 토큰을 SSE 형식으로 스트리밍"""
    # 실제 환경에서는 LLM 라이브러리 사용
    tokens = prompt.split() + ["[완료]"]
    for i, token in enumerate(tokens):
        data = json.dumps({"token": token, "index": i})
        yield f"data: {data}\n\n"
        await asyncio.sleep(0.05)  # 토큰 생성 시뮬레이션
    yield "data: [DONE]\n\n"

@app.get("/stream")
async def stream_llm(prompt: str = "안녕하세요"):
    return StreamingResponse(
        llm_token_generator(prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginx 버퍼링 비활성화
        }
    )

# 특정 인터페이스에서 HTTP 트래픽 캡처
sudo tcpdump -i eth0 -w capture.pcap port 80 or port 443

# TCP 핸드셰이크만 필터링 (SYN 패킷)
sudo tcpdump -i any 'tcp[tcpflags] & tcp-syn != 0'

# 특정 호스트와의 통신
sudo tcpdump -i eth0 host 10.0.0.1 -n

# DNS 쿼리 모니터링
sudo tcpdump -i any udp port 53 -v

# gRPC (HTTP/2) 트래픽
sudo tcpdump -i eth0 port 50051 -w grpc_trace.pcap

# TLS 인증서 정보 확인
curl -vI https://api.example.com 2>&1 | grep -A 20 "SSL connection"

# HTTP/2 사용 여부 확인
curl --http2 -I https://api.example.com

# 응답 시간 측정
curl -o /dev/null -s -w \
  "DNS: %{time_namelookup}s\nTCP: %{time_connect}s\nTLS: %{time_appconnect}s\nTotal: %{time_total}s\n" \
  https://api.example.com

# gRPC 엔드포인트 테스트 (grpcurl)
grpcurl -plaintext localhost:50051 list
grpcurl -plaintext -d '{"model_name": "bert", "input_data": [1.0, 2.0]}' \
  localhost:50051 inference.InferenceService/Predict

# 네트워크 대역폭 측정
iperf3 -s  # 서버
iperf3 -c server_ip -t 30 -P 4  # 클라이언트 (4개 병렬 스트림)