Scala Web 安全与鉴权实战：JWT、OAuth2 与安全加固

- 认证：验证身份凭证，产出 Principal（主体）这个不可变值
- 授权：基于 Principal + 资源 + 动作，判定 Allow 或 Deny
- 会话 Session：服务端存状态，客户端只拿不透明随机 ID
- Token：状态编码进凭证本身，服务端无状态或弱状态
- 无状态不等于无撤销：Token 必须有吊销通道，否则登出形同虚设
- 认证失败与授权失败必须区分：401 未认证，403 已认证但无权限

final case class Principal(
  userId: String,
  tenantId: String,
  roles: Set[String],
  scopes: Set[String]
)

sealed trait Decision
object Decision {
  case object Allow extends Decision
  final case class Deny(reason: String) extends Decision
}

// 认证层只负责把凭证换成 Principal，其余一概不管
trait Authenticator[F[_]] {
  def authenticate(token: String): F[Either[String, Principal]]
}

- 标准 claims：iss 签发者、sub 主体、aud 受众、exp 过期、nbf 生效、iat 签发时间、jti 唯一 ID
- exp 必须校验，且服务端与客户端时钟要同步（NTP 偏差是「莫名 401」的常客）
- access token 短命（5~15 分钟），refresh token 长命且可轮换
- 撤销只能靠黑名单：把 jti 或 token 指纹写入 Redis，TTL 设为 token 剩余寿命
- 常见漏洞：alg=none 降级、RS256 公钥当 HS256 密钥的密钥混淆、kid 注入

import pdi.jwt._
import java.time.Instant

val claim = JwtClaim(
  issuer = Some("https://auth.example.com"),
  subject = Some(userId),
  audience = Some(Set("api")),
  expiration = Some(Instant.now.plusSeconds(900).getEpochSecond),
  issuedAt = Some(Instant.now.getEpochSecond),
  jwtId = Some(java.util.UUID.randomUUID().toString)
)

// HS256：对称密钥，务必 32 字节以上随机值
val hsToken = Jwt.encode(claim, hmacSecret, JwtAlgorithm.HS256)

// RS256：私钥签名，服务只拿公钥验签，杜绝密钥扩散
val rsToken = Jwt.encode(claim, privateKey, JwtAlgorithm.RS256)
val verified = Jwt.decode(rsToken, publicKey, Seq(JwtAlgorithm.RS256))

// 撤销：登出时把 jti 写入黑名单，TTL 与 exp 对齐
def revoke(jti: String, ttlSeconds: Long): Unit =
  redis.setex(s"jwt:blacklist:$jti", ttlSeconds, "1")

- 授权码 + PKCE：前端生成 code_verifier，传 SHA-256 摘要作 code_challenge
- 回调时必须带原始 verifier，授权服务器比对，防授权码拦截重放
- state 参数防 CSRF，nonce 参数防 id_token 重放，二者都要随机且一次性
- scope 最小化：只申请用得到的，如 openid profile email
- 资源服务器校验：本地 JWT 验签 或 RFC 7662 token introspection
- Keycloak 自托管、Auth0 与 Okta 托管，三者都兼容标准发现端点

import java.security.{MessageDigest, SecureRandom}
import java.util.Base64

val rnd = new SecureRandom()
val bytes = new Array[Byte](32)
rnd.nextBytes(bytes)

// PKCE：verifier 随机且只存本地，challenge 走授权请求
val verifier = Base64.getUrlEncoder.withoutPadding.encodeToString(bytes)
val challenge = Base64.getUrlEncoder.withoutPadding.encodeToString(
  MessageDigest.getInstance("SHA-256").digest(verifier.getBytes("UTF-8"))
)

val authorizeUrl =
  s"https://idp.example.com/auth?response_type=code" +
    s"&client_id=$clientId&redirect_uri=$redirectUri" +
    s"&scope=openid%20profile&state=$state&nonce=$nonce" +
    s"&code_challenge=$challenge&code_challenge_method=S256"

import org.mindrot.jbcrypt.BCrypt
import java.security.MessageDigest

// bcrypt：cost 12 约 250ms，硬件升级后提高 cost 并渐进重算
def hashPassword(raw: String, pepper: String): String =
  BCrypt.hashpw(raw + pepper, BCrypt.gensalt(12))

def verify(raw: String, pepper: String, stored: String): Boolean =
  BCrypt.checkpw(raw + pepper, stored)

// 若必须自研比较（如校验 HMAC 指纹），一定要 constant-time
def safeEquals(a: Array[Byte], b: Array[Byte]): Boolean =
  MessageDigest.isEqual(a, b)

- HttpOnly：JS 拿不到，XSS 无法直接窃取会话
- Secure：仅 HTTPS 传输，防中间人嗅探
- SameSite=Lax 默认够用，Strict 更严但会断外链跳转后的登录态
- 登录成功后必须轮换 session id，防会话固定攻击
- CSRF：SameSite 之外再加同步令牌或双重提交 Cookie
- 双重提交：Cookie 存随机值，请求头带同值，服务端比对

import org.http4s.ResponseCookie
import org.http4s.SameSite

val sessionCookie = ResponseCookie(
  name = "sid",
  content = newSessionId,
  httpOnly = true,
  secure = true,
  sameSite = Some(SameSite.Lax),
  path = Some("/"),
  maxAge = Some(3600)
)

// 双重提交校验：Header 与 Cookie 必须一致且为 constant-time 比较
def checkCsrf(header: String, cookie: String): Boolean =
  header.nonEmpty && MessageDigest.isEqual(
    header.getBytes("UTF-8"),
    cookie.getBytes("UTF-8")
  )

- RBAC：User 到 Role 到 Permission 三层，权限用 resource:action 命名
- ABAC：策略即函数，输入 Principal 与 Resource，输出 Decision
- 行级权限：租户隔离必须落到 SQL where 条件，不能只在应用层过滤
- tagless-final：把鉴权抽象成 Authz[F] 代数，实现可替换、可测试
- http4s：用 Kleisli 中间件在路由前统一拦截
- ZIO：用 ZLayer 注入 Authz，在 effect 里短路失败

import cats.Applicative
import cats.syntax.all._

trait Authz[F[_]] {
  def authorize(p: Principal, action: String, resource: Resource): F[Decision]
}

object Authz {
  def rbac[F[_]: Applicative]: Authz[F] = new Authz[F] {
    def authorize(p: Principal, action: String, resource: Resource): F[Decision] = {
      val required = s"${resource.kind}:$action"
      if (p.roles.contains(required)) Applicative[F].pure(Decision.Allow)
      else Applicative[F].pure(Decision.Deny(s"missing $required"))
    }
  }

  // ABAC 补充：租户与资源归属必须同时成立
  def tenantScoped[F[_]: Applicative](delegate: Authz[F]): Authz[F] = new Authz[F] {
    def authorize(p: Principal, action: String, resource: Resource): F[Decision] =
      if (resource.tenantId != p.tenantId)
        Applicative[F].pure(Decision.Deny("cross-tenant access"))
      else delegate.authorize(p, action, resource)
  }
}

- SQL 注入：一律参数化查询，doobie 的 sql 插值天然安全
- XSS：输出编码而非输入过滤，模板默认转义，富文本用白名单清洗
- SSRF：出站请求走域名与协议白名单，禁跳转到内网与云元数据地址
- 反序列化：禁止 Java 原生序列化，JSON 用 circe 显式 derive
- CORS：不要 Access-Control-Allow-Origin 加通配符再带凭证
- 安全响应头：HSTS、CSP、X-Content-Type-Options、Referrer-Policy

// 参数化查询：占位符由驱动处理，注入无从谈起
val q = sql"SELECT id, email FROM users WHERE email = $email AND tenant_id = $tid"
q.query[User].option

// SSRF 防护：协议与主机双白名单，且禁止跟随重定向到内网
val allowedHosts = Set("api.partner.com", "cdn.example.com")
def isAllowed(uri: java.net.URI): Boolean =
  uri.getScheme == "https" && allowedHosts.contains(uri.getHost)

// 安全响应头：CSP 用 nonce 而非 unsafe-inline
val csp = "default-src 'self'; script-src 'self' 'nonce-abc123'; object-src 'none'"

- 密钥绝不入库：代码、镜像、CI 日志三处都不许出现
- 轮换：签名密钥保留 kid，新旧并行一段时间再下线旧钥
- KMS 与 Vault：动态凭证、租约到期自动失效，减少长期密钥
- 环境隔离：dev、staging、prod 的密钥与数据库完全独立
- 日志脱敏：Authorization 头、password、token、身份证一律打码
- 配置项缺失要启动即失败，不要回退到硬编码默认值

// 启动即校验，缺失就崩，避免带病运行
val jwtSecret: String =
  sys.env.getOrElse("JWT_SECRET", sys.error("JWT_SECRET is required"))

// 日志脱敏：只留前 6 位与后 4 位，便于排查又不可还原
def mask(token: String): String =
  if (token.length <= 12) "***"
  else s"${token.take(6)}...${token.takeRight(4)}"

// 密钥轮换：验签时按 kid 查表，支持新旧公钥并存
val publicKeys: Map[String, java.security.PublicKey] = loadKeysFromVault()
def keyFor(kid: String): Option[java.security.PublicKey] = publicKeys.get(kid)

- 依赖扫描：sbt-dependency-check 或 OWASP DC，CVSS 7 以上阻断合并
- 渗透测试：至少覆盖认证绕过、越权、注入、SSRF 四类
- 防爆破：按账号与 IP 双维度限流，失败次数指数退避
- 审计日志：谁在何时对哪个资源做了什么，成功失败都记
- 合规：密码强度、日志留存周期、数据出境，按行业要求落地
- 应急：密钥泄露有明确吊销与全量重置预案

// 登录限流：同一账号 5 分钟内失败 5 次即锁定，Redis 计数
def tooManyFailures(userId: String): Boolean =
  redis.incr(s"login:fail:$userId") > 5 && {
    redis.expire(s"login:fail:$userId", 300)
    true
  }

// 审计日志：结构化输出，绝不记录凭证原文
final case class AuditEvent(
  actor: String, action: String, resource: String,
  result: String, ip: String, at: java.time.Instant
)
def audit(e: AuditEvent): Unit =
  logger.info(s"audit actor=${e.actor} action=${e.action} " +
    s"resource=${e.resource} result=${e.result} ip=${e.ip}")