身份验证与授权最佳实践

// 1. 随机生成代码验证器
function generateCodeVerifier() {
  const array = new Uint8Array(32);
  crypto.getRandomValues(array);
  return btoa(String.fromCharCode(...array))
    .replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
}

// 2. 用 SHA-256 生成代码挑战
async function generateCodeChallenge(verifier) {
  const encoder = new TextEncoder();
  const data = encoder.encode(verifier);
  const digest = await crypto.subtle.digest('SHA-256', data);
  return btoa(String.fromCharCode(...new Uint8Array(digest)))
    .replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
}

// 3. 授权请求
const verifier = generateCodeVerifier();
const challenge = await generateCodeChallenge(verifier);

const authUrl = `https://auth.example.com/authorize?` +
  `response_type=code&` +
  `client_id=${clientId}&` +
  `redirect_uri=${redirectUri}&` +
  `code_challenge=${challenge}&` +
  `code_challenge_method=S256&` +
  `scope=read write`;

const jwt = require('jsonwebtoken');

// --- 代币发行 ---
function issueTokens(user) {
  const accessToken = jwt.sign(
    { sub: user.id, role: user.role },
    process.env.JWT_SECRET,
    {
      algorithm: 'HS256',
      expiresIn: '15m',
      issuer: 'api.example.com',
      audience: 'app.example.com',
    }
  );

  const refreshToken = jwt.sign(
    { sub: user.id, type: 'refresh' },
    process.env.REFRESH_SECRET,
    { algorithm: 'HS256', expiresIn: '7d' }
  );

  return { accessToken, refreshToken };
}

// --- 令牌验证 ---
function verifyAccessToken(token) {
  return jwt.verify(token, process.env.JWT_SECRET, {
    algorithms: ['HS256'],   // ← 始终指定！
    issuer: 'api.example.com',
    audience: 'app.example.com',
  });
}

const crypto = require('crypto');

// 使用 SHA-256 对应用程序接口密钥进行散列并比较
async function validateApiKey(req, res, next) {
  const apiKey = req.headers['x-api-key'];
  if (!apiKey) {
    return res.status(401).json({ error: 'API key required' });
  }

  // 哈希值并在数据库中查找（由于我们比较哈希值，因此不需要 timingSafeEqual）
  const hashedKey = crypto
    .createHash('sha256')
    .update(apiKey)
    .digest('hex');

  const keyRecord = await db.findApiKey(hashedKey);

  if (!keyRecord || keyRecord.revokedAt) {
    return res.status(403).json({ error: 'Invalid API key' });
  }

  // 检查有效期
  if (keyRecord.expiresAt && new Date() > keyRecord.expiresAt) {
    return res.status(403).json({ error: 'API key expired' });
  }

  req.apiClient = keyRecord;
  next();
}

import jwt
import time
from datetime import datetime, timedelta

class AgentTokenIssuer:
    """Issues scoped, short-lived tokens for AI agents."""

    ALLOWED_SCOPES = {
        "reader": ["data:read"],
        "writer": ["data:read", "data:write"],
        "executor": ["data:read", "data:write", "exec:sandbox"],
    }

    def __init__(self, secret_key: str):
        self.secret_key = secret_key

    def issue_agent_token(
        self,
        agent_id: str,
        role: str,
        tenant_id: str,
        token_budget: int = 10000,
        ttl_minutes: int = 15,
    ) -> str:
        # Validate role and resolve scopes
        if role not in self.ALLOWED_SCOPES:
            raise ValueError(f"Invalid role: {role}")

        payload = {
            "sub": agent_id,
            "tenant": tenant_id,
            "scopes": self.ALLOWED_SCOPES[role],
            "token_budget": token_budget,
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(minutes=ttl_minutes),
            "type": "agent",
        }
        return jwt.encode(payload, self.secret_key, algorithm="HS256")

    def verify_tool_access(self, token: str, required_scope: str) -> dict:
        # Decode and verify agent token
        payload = jwt.decode(
            token, self.secret_key, algorithms=["HS256"]
        )
        if required_scope not in payload["scopes"]:
            raise PermissionError(
                f"Agent {payload['sub']} lacks scope: {required_scope}"
            )
        return payload