レート制限・入力検証・CORS

const rateLimit = require('express-rate-limit');
const RedisStore = require('rate-limit-redis');

// グローバル制限: 全APIに適用
const globalLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15分
  max: 100,
  standardHeaders: true,    // RateLimit-* ヘッダー返却
  legacyHeaders: false,
  message: { error: 'Too many requests, please try again later.' },
});

// 認証エンドポイント用: より厳しい制限
const authLimiter = rateLimit({
  windowMs: 15 * 60 * 1000,
  max: 5,  // ログイン試行は15分で5回まで
  skipSuccessfulRequests: true, // 成功時はカウントしない
});

// 分散環境向け: Redis バックエンド
const distributedLimiter = rateLimit({
  store: new RedisStore({ sendCommand: (...args) => redisClient.sendCommand(args) }),
  windowMs: 60 * 1000,
  max: 30,
});

app.use(globalLimiter);
app.use('/api/auth', authLimiter);

RateLimit-Limit: 100
RateLimit-Remaining: 42
RateLimit-Reset: 1672531200
Retry-After: 30

const Ajv = require('ajv');
const addFormats = require('ajv-formats');

const ajv = new Ajv({ allErrors: true, removeAdditional: true });
addFormats(ajv);

// ユーザー作成APIのスキーマ
const createUserSchema = {
  type: 'object',
  required: ['name', 'email'],
  additionalProperties: false,
  properties: {
    name: {
      type: 'string',
      minLength: 1,
      maxLength: 100,
      pattern: '^[a-zA-Z0-9\\s\\-]+$', // 許可文字を限定
    },
    email: {
      type: 'string',
      format: 'email',
      maxLength: 254,
    },
    age: {
      type: 'integer',
      minimum: 0,
      maximum: 150,
    },
  },
};

// バリデーションミドルウェア
function validateBody(schema) {
  const validate = ajv.compile(schema);
  return (req, res, next) => {
    if (!validate(req.body)) {
      return res.status(400).json({
        error: 'Validation failed',
        details: validate.errors,
      });
    }
    next();
  };
}

app.post('/api/users', validateBody(createUserSchema), createUser);

const cors = require('cors');

// 許可するオリジンを明示的に指定
const allowedOrigins = [
  'https://app.example.com',
  'https://admin.example.com',
];

app.use(cors({
  origin(origin, callback) {
    // サーバー間通信（originなし）は許可
    if (!origin || allowedOrigins.includes(origin)) {
      callback(null, true);
    } else {
      callback(new Error('Not allowed by CORS'));
    }
  },
  methods: ['GET', 'POST', 'PUT', 'DELETE'],
  allowedHeaders: ['Content-Type', 'Authorization', 'X-API-Key'],
  credentials: true,
  maxAge: 86400,  // プリフライトキャッシュ: 24時間
}));

import tiktoken
import time
from collections import defaultdict

class TokenRateLimiter:
    """Rate limiter that counts tokens, not just requests."""

    def __init__(self, tokens_per_minute=100_000, requests_per_minute=60):
        self.tpm_limit = tokens_per_minute
        self.rpm_limit = requests_per_minute
        self.usage = defaultdict(lambda: {"tokens": [], "requests": []})
        self.encoder = tiktoken.encoding_for_model("gpt-4")

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def check_limit(self, user_id: str, prompt: str) -> dict:
        now = time.time()
        window = now - 60  # 1分間のスライディングウィンドウ
        user = self.usage[user_id]

        # 期限切れエントリのクリーンアップ
        user["tokens"] = [(t, c) for t, c in user["tokens"] if t > window]
        user["requests"] = [t for t in user["requests"] if t > window]

        # RPMのチェック
        if len(user["requests"]) >= self.rpm_limit:
            return {"allowed": False, "reason": "RPM limit exceeded"}

        # TPMのチェック
        token_count = self.count_tokens(prompt)
        used_tokens = sum(c for _, c in user["tokens"])
        if used_tokens + token_count > self.tpm_limit:
            return {"allowed": False, "reason": "TPM limit exceeded"}

        # 使用量を記録
        user["tokens"].append((now, token_count))
        user["requests"].append(now)
        return {"allowed": True, "tokens_used": token_count}

from pydantic import BaseModel, Field, field_validator
import re

class LLMRequest(BaseModel):
    """Validated LLM request with prompt injection defenses."""
    user_message: str = Field(..., max_length=4000)
    max_tokens: int = Field(default=1000, ge=1, le=4096)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

    @field_validator("user_message")
    @classmethod
    def sanitize_prompt(cls, v: str) -> str:
        # 既知のインジェクションパターンをブロック
        patterns = [
            r"(?i)ignore\s+(previous|above|all)\s+(instructions?|prompts?)",
            r"(?i)you\s+are\s+now\s+",
            r"(?i)system\s*:\s*",
            r"(?i)\[INST\]|\[\/INST\]|<\|im_start\|>",
        ]
        for pattern in patterns:
            if re.search(pattern, v):
                raise ValueError("Input contains disallowed patterns")
        return v

# FastAPIでの使用例
@app.post("/api/chat")
async def chat(request: LLMRequest):
    # システム命令とユーザー入力を分離
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"<user_input>{request.user_message}</user_input>"},
    ]
    return await call_llm(messages, request.max_tokens, request.temperature)

# コンテンツタイプの自動判別を無効化
X-Content-Type-Options: nosniff

# iframe埋め込みを禁止
X-Frame-Options: DENY

# HTTPS強制
Strict-Transport-Security: max-age=31536000; includeSubDomains

# CSP: APIはJSONのみ返すためscript実行を全て禁止
Content-Security-Policy: default-src 'none'; frame-ancestors 'none'

# Referrer情報の制限
Referrer-Policy: no-referrer

# ブラウザの機能制限
Permissions-Policy: geolocation=(), camera=(), microphone=()