1
server {
2
    listen 443 ssl http2;
3
    ssl_protocols TLSv1.2 TLSv1.3;
4
    ssl_prefer_server_ciphers on;
5
    ssl_ciphers TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256:TLS_AES_128_GCM_SHA256;
6
    ssl_certificate /etc/ssl/certs/server.crt;
7
    ssl_certificate_key /etc/ssl/private/server.key;
8
    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;
9
    ssl_stapling on;
10
    ssl_stapling_verify on;
11
}

1
# 安装 certbot
2
sudo apt install certbot python3-certbot-nginx
3

4
# 首次获取证书（standalone 模式）
5
sudo certbot certonly --standalone \
6
  -d auth.example.com -d api.example.com \
7
  --email admin@example.com --agree-tos --non-interactive
8

9
# Nginx 插件模式（无需停服）
10
sudo certbot --nginx -d auth.example.com -d api.example.com
11

12
# 自动续期
13
sudo systemctl enable certbot.timer
14
sudo certbot renew --dry-run  # 测试续期流程

1
# 授权码 + PKCE 流程
2
import hashlib, base64, os
3

4
# 1. 生成 PKCE 参数
5
code_verifier = base64.urlsafe_b64encode(os.urandom(32)).rstrip(b'=').decode()
6
code_challenge = base64.urlsafe_b64encode(
7
    hashlib.sha256(code_verifier.encode()).digest()
8
).rstrip(b'=').decode()
9

10
# 2. 授权请求
11
auth_url = f"https://auth.example.com/authorize?response_type=code&client_id={CLIENT_ID}&redirect_uri={REDIRECT_URI}&scope=openid+profile+email&code_challenge={code_challenge}&code_challenge_method=S256"
12

13
# 3. 用 code 换 token
14
token_response = requests.post("https://auth.example.com/token", data={
15
    "grant_type": "authorization_code",
16
    "code": auth_code,
17
    "redirect_uri": REDIRECT_URI,
18
    "client_id": CLIENT_ID,
19
    "code_verifier": code_verifier
20
})

1
import jwt
2
from datetime import datetime, timedelta
3

4
# 签发 JWT
5
def create_access_token(user_id: str, roles: list) -> str:
6
    payload = {
7
        "sub": user_id, "roles": roles,
8
        "iss": "https://auth.example.com",
9
        "aud": "https://api.example.com",
10
        "exp": datetime.utcnow() + timedelta(minutes=15),
11
        "iat": datetime.utcnow(),
12
        "jti": str(uuid.uuid4())  # 唯一 ID，防重放
13
    }
14
    return jwt.encode(payload, PRIVATE_KEY, algorithm="ES256", headers={"kid": KEY_ID})
15

16
# 验证 JWT
17
def verify_access_token(token: str) -> dict:
18
    return jwt.decode(
19
        token, PUBLIC_KEY, algorithms=["ES256"],  # 白名单算法
20
        audience="https://api.example.com",
21
        issuer="https://auth.example.com",
22
        options={"require": ["exp", "sub", "iss", "aud"]}
23
    )

1
# Refresh Token 轮换策略
2
import secrets
3
from datetime import datetime, timedelta
4

5
class RefreshTokenManager:
6
    def __init__(self, db):
7
        self.db = db
8

9
    def create_refresh_token(self, user_id: str) -> dict:
10
        token = secrets.token_urlsafe(48)
11
        token_hash = hashlib.sha256(token.encode()).hexdigest()
12
        self.db.insert("refresh_tokens", {
13
            "token_hash": token_hash, "user_id": user_id,
14
            "expires_at": datetime.utcnow() + timedelta(days=30),
15
            "is_revoked": False, "family_id": str(uuid.uuid4()),
16
        })
17
        return {"refresh_token": token, "expires_in": 2592000}
18

19
    def rotate_refresh_token(self, old_token: str) -> dict:
20
        """轮换 Refresh Token — 每次使用后旧 Token 失效"""
21
        old_hash = hashlib.sha256(old_token.encode()).hexdigest()
22
        record = self.db.find_one("refresh_tokens", {"token_hash": old_hash})
23
        if not record or record["is_revoked"]:
24
            # 检测到重放攻击！吊销整个 Token 族
25
            self.db.update("refresh_tokens",
26
                {"family_id": record["family_id"]}, {"is_revoked": True})
27
            raise SecurityException("Refresh token reuse detected — family revoked")
28
        self.db.update("refresh_tokens", {"token_hash": old_hash}, {"is_revoked": True})
29
        new_data = self.create_refresh_token(record["user_id"])
30
        self.db.update("refresh_tokens",
31
            {"token_hash": hashlib.sha256(new_data["refresh_token"].encode()).hexdigest()},
32
            {"family_id": record["family_id"]})
33
        return new_data

1
# Istio: 强制 mTLS
2
apiVersion: security.istio.io/v1beta1
3
kind: PeerAuthentication
4
metadata:
5
  name: default
6
  namespace: production
7
spec:
8
  mtls:
9
    mode: STRICT

1
# Cert-Manager: 自动签发服务证书
2
apiVersion: cert-manager.io/v1
3
kind: Certificate
4
metadata:
5
  name: api-server-cert
6
spec:
7
  secretName: api-server-tls
8
  duration: 24h        # 24 小时有效期
9
  renewBefore: 8h     # 提前 8 小时续期
10
  issuerRef:
11
    name: cluster-ca
12
    kind: ClusterIssuer
13
  dnsNames:
14
    - api.production.svc.cluster.local

1
# 使用 KMS 信封加密进行字段级加密
2
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
3
import boto3
4

5
kms = boto3.client('kms')
6

7
def encrypt_field(plaintext: str, kms_key_id: str) -> dict:
8
    # 1. 获取数据密钥
9
    response = kms.generate_data_key(KeyId=kms_key_id, KeySpec='AES_256')
10
    data_key, encrypted_key = response['Plaintext'], response['CiphertextBlob']
11
    # 2. 加密字段
12
    nonce = os.urandom(12)
13
    ciphertext = AESGCM(data_key).encrypt(nonce, plaintext.encode(), None)
14
    # 3. 返回加密数据
15
    return {
16
        'ciphertext': base64.b64encode(ciphertext).decode(),
17
        'encrypted_key': base64.b64encode(encrypted_key).decode(),
18
        'nonce': base64.b64encode(nonce).decode()
19
    }

1
# JWT 签名密钥轮换管理器
2
from datetime import datetime, timedelta
3
from cryptography.hazmat.primitives.asymmetric import ec
4
from cryptography.hazmat.primitives import serialization
5

6
class KeyRotationManager:
7
    def __init__(self, kms_client, db):
8
        self.kms = kms_client
9
        self.db = db
10
        self.rotation_interval = timedelta(days=30)
11
        self.overlap_period = timedelta(hours=24)
12

13
    def rotate_signing_key(self) -> dict:
14
        """执行密钥轮换"""
15
        # 1. 生成新密钥对
16
        private_key = ec.generate_private_key(ec.SECP256R1())
17
        kid = f"key-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
18
        # 2. 用 KMS 加密存储私钥
19
        private_pem = private_key.private_bytes(
20
            serialization.Encoding.PEM,
21
            serialization.PrivateFormat.PKCS8,
22
            serialization.NoEncryption())
23
        encrypted_key = self.kms.encrypt(KeyId=self.kms_key_id, Plaintext=private_pem)
24
        # 3. 存储密钥记录
25
        key_record = {
26
            "kid": kid,
27
            "public_key": private_key.public_key().public_bytes(
28
                serialization.Encoding.PEM,
29
                serialization.PublicFormat.SubjectPublicKeyInfo).decode(),
30
            "encrypted_private_key": base64.b64encode(encrypted_key['CiphertextBlob']).decode(),
31
            "created_at": datetime.utcnow(),
32
            "expires_at": datetime.utcnow() + self.rotation_interval + self.overlap_period,
33
            "status": "active"
34
        }
35
        self.db.insert("signing_keys", key_record)
36
        # 4. 将旧密钥标记为"仅验证"
37
        self.db.update("signing_keys",
38
            {"status": "active", "kid": {"$ne": kid}}, {"status": "verify_only"})
39
        # 5. 发布 JWKS
40
        self._publish_jwks()
41
        return key_record
42

43
    def _publish_jwks(self):
44
        active_keys = self.db.find("signing_keys", {
45
            "status": {"$in": ["active", "verify_only"]},
46
            "expires_at": {"$gt": datetime.utcnow()}
47
        })
48
        self._cache_jwks({"keys": [self._to_jwk(k) for k in active_keys]})

1
# 安全审计日志 — 不可变哈希链
2
import structlog, json, hashlib
3
from datetime import datetime
4

5
class AuditLogger:
6
    """写入后不可修改或删除，哈希链防篡改"""
7
    def __init__(self, storage):  # storage: 仅追加存储（如 Append-Only S3）
8
        self.storage = storage
9

10
    def log(self, event_type: str, actor: str, resource: str,
11
            action: str, result: str, metadata: dict = None):
12
        entry = {
13
            "timestamp": datetime.utcnow().isoformat(),
14
            "event_type": event_type, "actor": actor,
15
            "resource": resource, "action": action, "result": result,
16
            "metadata": metadata or {}, "trace_id": get_current_trace_id(),
17
        }
18
        prev_hash = self.storage.get_last_hash()
19
        entry["prev_hash"] = prev_hash
20
        entry["hash"] = hashlib.sha256(
21
            json.dumps(entry, sort_keys=True).encode()).hexdigest()
22
        self.storage.append(entry)
23

24
    def verify_integrity(self) -> bool:
25
        entries = self.storage.get_all()
26
        for i in range(1, len(entries)):
27
            expected = hashlib.sha256(
28
                json.dumps(entries[i-1], sort_keys=True).encode()).hexdigest()
29
            if entries[i]["prev_hash"] != expected:
30
                return False
31
        return True

1
# 滑动窗口分布式限流
2
import redis, time
3

4
class RateLimiter:
5
    def __init__(self, redis_client: redis.Redis):
6
        self.redis = redis_client
7

8
    def check_rate(self, key: str, max_requests: int, window: int) -> tuple[bool, int]:
9
        now = time.time()
10
        pipe = self.redis.pipeline()
11
        pipe.zremrangebyscore(key, 0, now - window)
12
        pipe.zadd(key, {str(now): now})
13
        pipe.zcard(key)
14
        pipe.expire(key, window)
15
        count = pipe.execute()[2]
16
        return (count <= max_requests, max(0, max_requests - count))
17

18
RATE_LIMITS = {
19
    "/auth/login": (10, 300),  # 5 分钟 10 次
20
    "/auth/token": (30, 60),   # 1 分钟 30 次
21
    "/api/*":     (100, 60),   # 1 分钟 100 次
22
}

1
# FastAPI CORS 安全配置
2
from fastapi import FastAPI
3
from fastapi.middleware.cors import CORSMiddleware
4

5
app = FastAPI()
6
app.add_middleware(
7
    CORSMiddleware,
8
    allow_origins=["https://app.example.com", "https://admin.example.com"],
9
    allow_credentials=True,
10
    allow_methods=["GET", "POST", "PUT", "DELETE"],
11
    allow_headers=["Authorization", "Content-Type"],
12
    max_age=3600, expose_headers=["X-Request-ID"],
13
)

1
// auth_server.go — 认证服务核心逻辑
2
package main
3

4
import (
5
    "crypto/ecdsa"
6
    "fmt"
7
    "time"
8
    "github.com/golang-jwt/jwt/v5"
9
    "github.com/gin-gonic/gin"
10
)
11

12
type AuthServer struct {
13
    signingKey  *ecdsa.PrivateKey
14
    verifyKey   *ecdsa.PublicKey
15
    keyID       string
16
    tokenTTL    time.Duration
17
    rateLimiter *RateLimiter
18
    auditLogger *AuditLogger
19
}
20

21
func NewAuthServer(key *ecdsa.PrivateKey, kid string) *AuthServer {
22
    return &AuthServer{
23
        signingKey: key, verifyKey: &key.PublicKey,
24
        keyID: kid, tokenTTL: 15 * time.Minute,
25
        rateLimiter: NewRateLimiter(), auditLogger: NewAuditLogger(),
26
    }
27
}
28

29
// Token 签发端点
30
func (s *AuthServer) IssueToken(c *gin.Context) {
31
    if !s.rateLimiter.Allow(c.ClientIP(), "/auth/token") {
32
        c.JSON(429, gin.H{"error": "rate_limit_exceeded"}); return
33
    }
34
    claims, err := s.validateAuthCode(c.PostForm("code"))
35
    if err != nil {
36
        c.JSON(401, gin.H{"error": "invalid_grant"}); return
37
    }
38
    now := time.Now()
39
    token := jwt.NewWithClaims(jwt.SigningMethodES256, jwt.MapClaims{
40
        "sub": claims.UserID, "roles": claims.Roles,
41
        "iss": "https://auth.example.com", "aud": "https://api.example.com",
42
        "exp": now.Add(s.tokenTTL).Unix(), "iat": now.Unix(), "jti": generateUUID(),
43
    })
44
    token.Header["kid"] = s.keyID
45
    tokenString, _ := token.SignedString(s.signingKey)
46
    s.auditLogger.Log("token_issue", c.ClientIP(), claims.UserID, "issue", "success", nil)
47
    c.JSON(200, gin.H{"access_token": tokenString, "token_type": "Bearer",
48
        "expires_in": int(s.tokenTTL.Seconds())})
49
}
50

51
// Token 验证中间件 — 强制算法白名单
52
func (s *AuthServer) AuthMiddleware() gin.HandlerFunc {
53
    return func(c *gin.Context) {
54
        token, err := jwt.Parse(
55
            strings.TrimPrefix(c.GetHeader("Authorization"), "Bearer "),
56
            func(t *jwt.Token) (any, error) {
57
                if _, ok := t.Method.(*jwt.SigningMethodECDSA); !ok {
58
                    return nil, fmt.Errorf("unexpected method: %v", t.Header["alg"])
59
                }
60
                return s.verifyKey, nil
61
            }, jwt.WithAudience("https://api.example.com"),
62
               jwt.WithIssuer("https://auth.example.com"))
63
        if err != nil || !token.Valid {
64
            c.AbortWithStatusJSON(401, gin.H{"error": "invalid_token"}); return
65
        }
66
        claims := token.Claims.(jwt.MapClaims)
67
        c.Set("user_id", claims["sub"]); c.Set("roles", claims["roles"])
68
        c.Next()
69
    }
70
}

1
#!/bin/bash
2
# test_security.sh — 安全验证自动化脚本
3
set -euo pipefail
4

5
echo "=== 1. TLS 配置验证 ==="
6
echo | openssl s_client -connect localhost:8443 -tls1_3 2>/dev/null | grep "Protocol\|Cipher"
7

8
echo "=== 2. mTLS 验证 ==="
9
if openssl s_client -connect auth-server:8443 -CAfile certs/ca.crt 2>&1 | grep -q "no peer certificate"; then
10
  echo "mTLS 强制验证通过"
11
fi
12

13
echo "=== 3. OAuth 2.0 + PKCE 测试 ==="
14
CODE_VERIFIER=$(openssl rand -base64 32 | tr -d '=')
15
CODE_CHALLENGE=$(echo -n "$CODE_VERIFIER" | openssl dgst -sha256 -binary | openssl base64 -A | tr '+/' '-_' | tr -d '=')
16
AUTH_CODE=$(curl -s -k https://localhost:8443/authorize \
17
  -d "response_type=code&client_id=test&code_challenge=$CODE_CHALLENGE&code_challenge_method=S256" | jq -r '.code')
18
TOKEN=$(curl -s -k https://localhost:8443/token \
19
  -d "grant_type=authorization_code&code=$AUTH_CODE&code_verifier=$CODE_VERIFIER")
20
echo "OAuth 2.0 + PKCE 验证通过"
21

22
echo "=== 4. JWT 算法验证 ==="
23
ALG=$(echo "$TOKEN" | jq -r '.access_token' | cut -d. -f1 | base64 -d 2>/dev/null | jq -r '.alg')
24
[ "$ALG" = "ES256" ] && echo "JWT 签名算法 ES256 验证通过"
25

26
echo "=== 5. 限流测试 ==="
27
for i in $(seq 1 12); do
28
  STATUS=$(curl -s -o /dev/null -w "%{http_code}" -k https://localhost:8443/auth/login -d "user=test&pass=wrong")
29
  [ "$STATUS" = "429" ] && echo "限流在第 $i 次请求触发" && break
30
done
31

32
echo "=== 安全验证完成 ==="