Ktor中访客、JWT、自定义Token认证与差异化限流最佳方案

需求回顾 #

• 访客访问：无Authorization请求头，对应最低限流阈值
• 自定义Token访问：携带Authorization: Token {token}请求头，需自定义逻辑将Token映射至AuthUser，无请求限制
• JWT Token访问：携带Authorization: Bearer {jwt}请求头，对应更高限流阈值

现有问题分析 #

你之前的代码将两种Token认证逻辑塞进同一个jwt认证块，导致认证逻辑混乱，且未正确处理访客请求的匿名场景，进而出现call.principal()返回null的异常情况。

解决方案步骤 #

1. 拆分多类型认证策略 #

将匿名访客、JWT、自定义Token拆分为独立的认证配置，逻辑更清晰且避免冲突：

import io.ktor.server.auth.*
import io.ktor.server.auth.jwt.*
import com.auth0.jwt.JWT
import com.auth0.jwt.algorithms.Algorithm

// 假设你的AuthUser实体定义如下
data class AuthUser(
    val role: Role,
    val id: String? = null
) {
    enum class Role {
        VISITOR, JWT_USER, CUSTOM_TOKEN_USER
    }

    fun isVisitor() = role == Role.VISITOR
    fun isJwtUser() = role == Role.JWT_USER
    fun isCustomTokenUser() = role == Role.CUSTOM_TOKEN_USER
}

fun Application.configureAuthentication() {
    install(Authentication) {
        // 匿名认证：处理无Authorization头的访客请求
        anonymous("visitor") {
            validate {
                AuthUser(role = AuthUser.Role.VISITOR)
            }
        }

        // JWT认证：处理Bearer格式的Token
        jwt("jwt") {
            val jwtSecret = environment.config.property("jwt.secret").getString()
            val issuer = environment.config.property("jwt.issuer").getString()
            
            verifier(
                JWT.require(Algorithm.HMAC256(jwtSecret))
                    .withIssuer(issuer)
                    .build()
            )
            
            validate { credential ->
                val userId = credential.payload.getClaim("userId").asString()
                if (userId.isNotEmpty()) {
                    AuthUser(role = AuthUser.Role.JWT_USER, id = userId)
                } else {
                    null
                }
            }
            
            challenge { _, _ ->
                call.respond(DataVo(401, "无效的JWT Token", null))
            }
        }

        // 自定义Token认证：处理Token格式的自定义凭证
        apiKey("custom-token") {
            authHeader { call ->
                val header = call.request.headers["Authorization"]
                if (header?.startsWith("Token ") == true) {
                    HttpAuthHeader.Single("Token", header.removePrefix("Token "))
                } else {
                    null
                }
            }
            
            validate { apiKeyCredential ->
                // 替换为你的自定义Token验证逻辑：比如从数据库/缓存查询有效性
                val token = apiKeyCredential.key
                if (isValidCustomToken(token)) {
                    AuthUser(role = AuthUser.Role.CUSTOM_TOKEN_USER, id = getUserIdByToken(token))
                } else {
                    null
                }
            }
            
            challenge { _, _ ->
                call.respond(DataVo(401, "无效的自定义Token", null))
            }
        }
    }
}

// 自定义Token验证示例方法
fun isValidCustomToken(token: String): Boolean {
    // 实现你的Token有效性校验逻辑
    return token.isNotEmpty()
}

// 根据自定义Token获取用户ID的示例方法
fun getUserIdByToken(token: String): String {
    // 实现从Token映射到用户ID的逻辑
    return "custom_user_123"
}

2. 全局应用多认证策略 #

在路由层全局配置认证，让所有请求依次尝试三种认证方式，确保访客也能获得合法的AuthUser实例：

fun Application.configureRouting() {
    routing {
        // 全局认证：按顺序尝试访客、JWT、自定义Token认证
        authenticate("visitor", "jwt", "custom-token") {
            // 所有业务端点都将受此认证规则覆盖
            get("/api/resource") {
                val user = call.principal<AuthUser>() ?: AuthUser(AuthUser.Role.VISITOR)
                call.respond(DataVo(200, "请求成功", mapOf("userRole" to user.role.name)))
            }

            // 其他业务端点...
        }
    }
}

3. 配置差异化限流规则 #

根据AuthUser的角色设置请求权重，实现不同级别的限流控制：

import io.ktor.server.plugins.ratelimit.*
import kotlin.time.Duration.Companion.minutes

fun Application.configureRateLimiting() {
    install(RateLimit) {
        register {
            // 请求标识：登录用户用ID区分，访客用IP区分
            requestKey { call ->
                val user = call.principal<AuthUser>()
                user?.id ?: run {
                    val headers = call.request.headers
                    headers["X-Real-IP"] ?: headers["X-Forwarded-For"] ?: call.request.origin.remoteHost
                }
            }

            // 基础限流池配置：每分钟补充6000个请求额度
            rateLimiter(limit = 6000, refillPeriod = 1.minutes)

            // 根据用户角色设置请求权重：权重越低，相同时间内可发起的请求越多
            requestWeight { call, _ ->
                val user = call.principal<AuthUser>() ?: AuthUser(AuthUser.Role.VISITOR)
                when (user.role) {
                    // 访客：权重20 → 每分钟最多可发起6000/20=300次请求
                    AuthUser.Role.VISITOR -> 20
                    // JWT用户：权重8 → 每分钟最多可发起6000/8=750次请求
                    AuthUser.Role.JWT_USER -> 8
                    // 自定义Token用户：权重0 → 不消耗限流额度，实现无限制
                    AuthUser.Role.CUSTOM_TOKEN_USER -> 0
                }
            }
        }
    }
}

核心逻辑说明 #

• 拆分认证策略：将三种认证场景独立配置，避免逻辑耦合，便于维护和扩展
• 全局认证覆盖：通过多策略认证确保所有请求都能匹配到对应的AuthUser，不会出现principal为null的情况
• 权重式限流：利用请求权重实现差异化限流，权重为0的请求不消耗限流额度，直接实现无限制访问
• 精准错误反馈：每个认证块独立配置challenge逻辑，返回更具体的错误信息

内容的提问来源于stack exchange，提问作者fgxfhb ggvkf