同時ログイン防止実装ガイド — IP・セッション・JWT・Nginxレイヤー別戦略と実践コード

1. ログイン → セッション作成 → Redis: user:{userId}:sessions = {sessionId}
2. 新規ログイン発生 → 既存セッションID照会 → 既存セッション無効化（削除）
3. 既存デバイスからリクエスト → セッションなし → 401 + "別の場所でログインされました"メッセージ

1. ログイン → access_token + refresh_token発行
2. refresh_tokenをサーバーDB/Redisに保存（user_id → token_family）
3. 新デバイスログイン → 新token_family発行 + 既存family全体廃棄
4. 既存デバイスがrefresh試行 → familyが廃棄済み → 401 + 再ログイン誘導

# Rate Limiting — 同一IPからの過度なリクエストを制限
limit_req_zone $binary_remote_addr zone=login:10m rate=5r/s;

server {
    location /api/auth/login {
        limit_req zone=login burst=10 nodelay;
        proxy_pass http://backend;
    }
}

# IPベースのアクセス制御
geo $blocked {
    default 0;
    10.0.0.0/8 0;      # 内部ネットワーク許可
    192.168.0.0/16 0;   # 内部ネットワーク許可
    # 特定の悪意あるIPをブロック
    203.0.113.50 1;
}

server {
    if ($blocked) {
        return 403;
    }
}

# 国別ブロック（GeoIP2モジュール）
geoip2 /etc/nginx/GeoLite2-Country.mmdb {
    $geoip2_country_code country iso_code;
}

map $geoip2_country_code $allowed_country {
    KR 1;
    US 1;
    JP 1;
    default 0;
}

-- OpenResty + Lua例：JWTからuser_idを抽出しRedisセッション検証
local jwt = require "resty.jwt"
local redis = require "resty.redis"

local token = ngx.var.http_authorization:sub(8) -- "Bearer "を除去
local jwt_obj = jwt:load_jwt(token)

if jwt_obj.valid then
    local red = redis:new()
    red:connect("127.0.0.1", 6379)
    local active_session = red:get("user:" .. jwt_obj.payload.sub .. ":session")
    if active_session ~= jwt_obj.payload.jti then
        ngx.status = 401
        ngx.say('{"error": "session_expired", "message": "別の場所でログインされました"}')
        return ngx.exit(401)
    end
end

@Configuration
@EnableWebSecurity
public class SecurityConfig {

    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
        http
            .sessionManagement(session -> session
                .maximumSessions(1)                    // 最大1セッション
                .maxSessionsPreventsLogin(false)        // false: 既存セッション期限切れ（デフォルト）
                                                       // true: 新規ログインブロック
                .expiredSessionStrategy(event -> {
                    HttpServletResponse response = event.getResponse();
                    response.setStatus(401);
                    response.setContentType("application/json;charset=UTF-8");
                    response.getWriter().write(
                        "{\"error\":\"session_expired\",\"message\":\"別の場所でログインされました\"}"
                    );
                })
            );
        return http.build();
    }

    // 分散環境ではSpring Session + Redisを使用
    @Bean
    public SessionRegistry sessionRegistry() {
        return new SpringSessionBackedSessionRegistry<>(this.sessionRepository);
    }

    @Autowired
    private FindByIndexNameSessionRepository<?> sessionRepository;
}

<dependency>
    <groupId>org.springframework.session</groupId>
    <artifactId>spring-session-data-redis</artifactId>
</dependency>

# middleware.py
import redis
from django.conf import settings
from django.contrib.auth import logout

r = redis.Redis(host=settings.REDIS_HOST, port=6379, db=0, decode_responses=True)

ACTIVE_SESSION_PREFIX = "user:active_session:"

class SingleSessionMiddleware:
    """1アカウントに1セッションのみ許可するミドルウェア"""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        if request.user.is_authenticated:
            key = f"{ACTIVE_SESSION_PREFIX}{request.user.id}"
            current_session = request.session.session_key
            active_session = r.get(key)

            if active_session and active_session != current_session:
                # 現在のセッションがアクティブセッションではない → 強制ログアウト
                logout(request)
                from django.http import JsonResponse
                return JsonResponse(
                    {"error": "session_expired",
                     "message": "別の場所でログインされました"},
                    status=401
                )

        return self.get_response(request)

# signals.py — ログイン時にアクティブセッションを登録
from django.contrib.auth.signals import user_logged_in
from django.dispatch import receiver

@receiver(user_logged_in)
def register_active_session(sender, request, user, **kwargs):
    key = f"user:active_session:{user.id}"
    # 既存セッションを削除（Djangoセッションストアから）
    old_session_key = r.get(key)
    if old_session_key:
        from django.contrib.sessions.models import Session
        Session.objects.filter(session_key=old_session_key).delete()

    # 新しいセッションを登録
    r.set(key, request.session.session_key, ex=settings.SESSION_COOKIE_AGE)

// Node.js / Express例
const redis = require('ioredis')
const jwt = require('jsonwebtoken')
const crypto = require('crypto')

const client = new redis()

async function login(userId, deviceId) {
  const tokenFamily = crypto.randomUUID()
  const jti = crypto.randomUUID()

  // 既存token familyを廃棄
  const oldFamily = await client.get(`user:${userId}:token_family`)
  if (oldFamily) {
    await client.del(`token_family:${oldFamily}`)
  }

  // 新token familyを登録
  await client.set(`user:${userId}:token_family`, tokenFamily, 'EX', 86400 * 7)
  await client.set(
    `token_family:${tokenFamily}`,
    JSON.stringify({
      userId,
      deviceId,
      jti,
      rotationCount: 0,
    }),
    'EX',
    86400 * 7
  )

  const accessToken = jwt.sign({ sub: userId, jti, device_id: deviceId }, process.env.JWT_SECRET, {
    expiresIn: '15m',
  })

  const refreshToken = jwt.sign(
    { sub: userId, family: tokenFamily, jti: crypto.randomUUID() },
    process.env.JWT_REFRESH_SECRET,
    { expiresIn: '7d' }
  )

  return { accessToken, refreshToken }
}

async function refresh(refreshToken) {
  const decoded = jwt.verify(refreshToken, process.env.JWT_REFRESH_SECRET)
  const familyData = await client.get(`token_family:${decoded.family}`)

  if (!familyData) {
    // Token familyが廃棄済み → 窃取疑い → 全セッション無効化
    await client.del(`user:${decoded.sub}:token_family`)
    throw new Error('TOKEN_FAMILY_REVOKED')
  }

  const family = JSON.parse(familyData)

  // Rotation：新トークン発行 + 既存refresh無効化
  const newJti = crypto.randomUUID()
  family.jti = newJti
  family.rotationCount += 1

  await client.set(`token_family:${decoded.family}`, JSON.stringify(family), 'EX', 86400 * 7)

  const newAccessToken = jwt.sign(
    { sub: decoded.sub, jti: newJti, device_id: family.deviceId },
    process.env.JWT_SECRET,
    { expiresIn: '15m' }
  )

  const newRefreshToken = jwt.sign(
    { sub: decoded.sub, family: decoded.family, jti: crypto.randomUUID() },
    process.env.JWT_REFRESH_SECRET,
    { expiresIn: '7d' }
  )

  return { accessToken: newAccessToken, refreshToken: newRefreshToken }
}

# /etc/nginx/conf.d/security.conf

# ログインエンドポイントRate Limit
limit_req_zone $binary_remote_addr zone=auth_limit:10m rate=5r/s;

# グローバル接続制限
limit_conn_zone $binary_remote_addr zone=conn_limit:10m;

# 疑わしいIPマップ
map $remote_addr $is_suspicious {
    default 0;
    ~^192\.168\.100\. 1;  # 例：内部テスト帯域をブロック
}

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

    # ログインエンドポイント
    location /api/auth/login {
        limit_req zone=auth_limit burst=10 nodelay;
        limit_conn conn_limit 10;

        if ($is_suspicious) {
            return 403;
        }

        proxy_pass http://backend;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }

    # 認証必要API — アプリでセッション/トークン検証
    location /api/ {
        proxy_pass http://backend;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

┌────────────────────────────────────────────────┐
│                  クライアント                      │
│  （ブラウザ/アプリ — device_id, fingerprint生成）   │
└─────────────────────┬──────────────────────────┘
                      │
┌─────────────────────▼──────────────────────────┐
│              Nginx / L7 Proxy                    │
│  ✅ Rate Limit（loginブルートフォース防御）        │
│  ✅ IPブロック（geo, deny）                       │
│  ✅ 基本ヘッダー検証                              │
│  ❌ セッション有効性判断（アプリの役割）            │
└─────────────────────┬──────────────────────────┘
                      │
┌─────────────────────▼──────────────────────────┐
│           認証サーバー / API Gateway               │
│  ✅ JWT検証（署名、期限切れ、jtiブラックリスト）     │
│  ✅ session_version比較                           │
│  ✅ Refresh Token Rotation管理                    │
│  ✅ 同時セッション数ポリシー判断                    │
└─────────────────────┬──────────────────────────┘
                      │
┌─────────────────────▼──────────────────────────┐
│              セッション/トークンストア（Redis）      │
│  - user:{id}:sessions → Set of sessionId         │
│  - user:{id}:token_family → familyId             │
│  - token_family:{id} → {userId, deviceId, jti}   │
│  - blacklist:jti:{jti} → TTL                     │
└──────────────────────────────────────────────────┘

原因：VPN接続時にIPがVPNゲートウェイIPに変更される
解決：IP変更だけではセッションを切断せず、refresh tokenが有効ならセッション維持
      + IP変更イベントを監査ログにのみ記録

{
  "concurrent_session_policy": {
    "max_total": 5,
    "per_device_type": {
      "web_browser": 2,
      "mobile_app": 2,
      "tablet_app": 1
    },
    "on_exceed": "expire_oldest",
    "admin_override": true
  }
}

# 例：デバイスタイプ別スロット管理
def check_session_limit(user_id, device_type, new_session_id):
    key = f"user:{user_id}:sessions:{device_type}"
    sessions = redis.lrange(key, 0, -1)
    max_slots = POLICY['per_device_type'].get(device_type, 1)

    if len(sessions) >= max_slots:
        if POLICY['on_exceed'] == 'expire_oldest':
            oldest = redis.lpop(key)
            invalidate_session(oldest)  # 最も古いセッションを期限切れに
        elif POLICY['on_exceed'] == 'deny_new':
            raise SessionLimitExceeded()

    redis.rpush(key, new_session_id)
    redis.expire(key, SESSION_TTL)