ネットワークエンジニアリング＆セキュリティ完全ガイド: 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

ルートCA（ブラウザ/OSに組み込み済み）
  └── 中間CA
        └── リーフ証明書（実際のサーバー証明書）

通常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ステープリング
    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並列ストリーム）