验证 JWT

当 Cloudflare 向您的源站发送请求时，该请求将在 Cf-Access-Jwt-Assertion 请求标头中包含一个应用程序令牌。通过浏览器发起的请求还将以 CF_Authorization cookie 的形式传递该令牌。

Cloudflare 使用您账户特有的密钥对来签署令牌。您应该使用您的公钥来验证令牌，以确保请求来自 Access，而不是恶意的第三方。我们建议验证 Cf-Access-Jwt-Assertion 标头，而不是 CF_Authorization cookie，因为不保证一定会传递该 cookie。

Access 签名密钥

签名密钥对的公钥位于 https://<your-team-name>.cloudflareaccess.com/cdn-cgi/access/certs，其中 <your-team-name> 是您的 Cloudflare One 团队名称 (team name)。

默认情况下，Access 每 6 周轮换一次签名密钥。这意味着随着密钥的轮换，您需要以编程方式或手动更新您的密钥。轮换后的旧密钥在 7 天内仍然有效，以便为您预留时间来进行更新。

您也可以使用 API 手动轮换密钥。这可以出于测试或安全目的而进行。

如下面示例所示，https://<your-team-name>.cloudflareaccess.com/cdn-cgi/access/certs 包含两个公钥：用于签署所有新令牌的当前密钥，以及已被轮换掉的旧密钥。

keys：JWK 格式的两个密钥
public_cert：PEM 格式的当前密钥
public_certs：PEM 格式的两个密钥

{
  "keys": [
    {
      "kid": "1a1c3986a44ce6390be42ec772b031df8f433fdc71716db821dc0c39af3bce49",
      "kty": "RSA",
      "alg": "RS256",
      "use": "sig",
      "e": "AQAB",
      "n": "5PKw-...-AG7MyQ"
    },
    {
      "kid": "6c3bffef71bb0a90c9cbef3b7c0d4a1c7b4b8b76b80292a623afd9dac45d1c65",
      "kty": "RSA",
      "alg": "RS256",
      "use": "sig",
      "e": "AQAB",
      "n": "pwVn...AA6Hw"
    }
  ],
  "public_cert": {
    "kid": "6c3bffef71bb0a90c9cbef3b7c0d4a1c7b4b8b76b80292a623afd9dac45d1c65",
    "cert": "-----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- "
  },
  "public_certs": [
    {
      "kid": "1a1c3986a44ce6390be42ec772b031df8f433fdc71716db821dc0c39af3bce49",
      "cert": "-----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- "
    },
    {
      "kid": "6c3bffef71bb0a90c9cbef3b7c0d4a1c7b4b8b76b80292a623afd9dac45d1c65",
      "cert": "-----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- "
    }
  ]
}

手动验证 JWT

要手动验证令牌：

从 Cf-Access-Jwt-Assertion 请求标头中复制 JWT。
转到 jwt.io ↗。
选择 RS256 算法。
将 JWT 粘贴到 Encoded 文本框中。
在 Payload 文本框中，确保 iss 字段指向您的团队域（https://<your-team-name>.cloudflareaccess.com）。jwt.io 使用 iss 值来获取用于验证令牌的公钥。
确保页面显示 Signature Verified。

现在，您可以信任此请求是由 Access 发送的。

编程验证

您可以在您的源站服务器上运行自动脚本以验证传入的请求。提供的示例代码从请求中获取应用程序令牌，并将其签名与您的公钥进行校验。您需要在示例代码中插入您自己的团队域和应用程序受众 (AUD) 标签。

获取您的 AUD 标签

Cloudflare Access 为每个应用程序分配一个唯一的 AUD 标签。令牌有效负载中的 aud 声明指定了该 JWT 针对哪个应用程序有效。

要获取 AUD 标签：

在 Cloudflare 仪表板 ↗中，转到 Zero Trust > Access controls（访问控制）> Applications（应用程序）。
选择您应用程序的 Configure（配置）。
从 **Additional settings（其他设置）**中，复制 Application Audience (AUD) Tag（应用程序受众 (AUD) 标签）。

您现在可以将 AUD 标签粘贴到您的令牌验证脚本中。除非您删除或重新创建 Access 应用程序，否则 AUD 标签将永远不会改变。

Cloudflare Workers 示例

当 Cloudflare Access 位于您的 Worker 前面时，您的 Worker 仍需要验证 Cloudflare Access 添加到传入请求的 Cf-Access-Jwt-Assertion 标头中的 JWT。

以下代码将使用 jose NPM 包 ↗来验证 JWT：

import { jwtVerify, createRemoteJWKSet } from "jose";

export default {
	async fetch(request, env, ctx) {
		// Verify the POLICY_AUD environment variable is set
		if (!env.POLICY_AUD) {
			return new Response("Missing required audience", {
				status: 403,
				headers: { "Content-Type": "text/plain" },
			});
		}

		// Get the JWT from the request headers
		const token = request.headers.get("cf-access-jwt-assertion");

		// Check if token exists
		if (!token) {
			return new Response("Missing required CF Access JWT", {
				status: 403,
				headers: { "Content-Type": "text/plain" },
			});
		}

		try {
			// Create JWKS from your team domain
			const JWKS = createRemoteJWKSet(
				new URL(`${env.TEAM_DOMAIN}/cdn-cgi/access/certs`),
			);

			// Verify the JWT
			const { payload } = await jwtVerify(token, JWKS, {
				issuer: env.TEAM_DOMAIN,
				audience: env.POLICY_AUD,
			});

			// Token is valid, proceed with your application logic
			return new Response(`Hello ${payload.email || "authenticated user"}!`, {
				headers: { "Content-Type": "text/plain" },
			});
		} catch (error) {
			// Token verification failed
			const message = error instanceof Error ? error.message : "Unknown error";
			return new Response(`Invalid token: ${message}`, {
				status: 403,
				headers: { "Content-Type": "text/plain" },
			});
		}
	},
};

import { jwtVerify, createRemoteJWKSet } from "jose";

interface Env {
	POLICY_AUD: string;
	TEAM_DOMAIN: string;
}

export default {
	async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
		// Verify the POLICY_AUD environment variable is set
		if (!env.POLICY_AUD) {
			return new Response("Missing required audience", {
				status: 403,
				headers: { "Content-Type": "text/plain" },
			});
		}

		// Get the JWT from the request headers
		const token = request.headers.get("cf-access-jwt-assertion");

		// Check if token exists
		if (!token) {
			return new Response("Missing required CF Access JWT", {
				status: 403,
				headers: { "Content-Type": "text/plain" },
			});
		}

		try {
			// Create JWKS from your team domain
			const JWKS = createRemoteJWKSet(
				new URL(`${env.TEAM_DOMAIN}/cdn-cgi/access/certs`)
			);

			// Verify the JWT
			const { payload } = await jwtVerify(token, JWKS, {
				issuer: env.TEAM_DOMAIN,
				audience: env.POLICY_AUD,
			});

			// Token is valid, proceed with your application logic
			return new Response(
				`Hello ${payload.email || "authenticated user"}!`,
				{
					headers: { "Content-Type": "text/plain" },
				}
			);
		} catch (error) {
			// Token verification failed
			const message = error instanceof Error ? error.message : "Unknown error";
			return new Response(`Invalid token: ${message}`, {
				status: 403,
				headers: { "Content-Type": "text/plain" },
			});
		}
	},
};

必需的环境变量

将这些环境变量添加到您的 Worker 中：

POLICY_AUD：您的应用程序的 AUD 标签
TEAM_DOMAIN：https://<your-team-name>.cloudflareaccess.com，其中 <your-team-name> 替换为您的实际团队名称。

您可以通过将这些变量添加到您的 Worker 的 Wrangler 配置文件中来进行设置，也可以通过 Cloudflare 仪表板在 Workers & Pages > 您的 worker > Settings（设置）> **Environment Variables（环境变量）**下进行设置。

Golang 示例

package main

import (
    "context"
    "fmt"
    "net/http"

    "github.com/coreos/go-oidc/v3/oidc"
)

var (
    ctx        = context.TODO()
    teamDomain = "https://test.cloudflareaccess.com"
    certsURL   = fmt.Sprintf("%s/cdn-cgi/access/certs", teamDomain)

    // The Application Audience (AUD) tag for your application
    policyAUD = "4714c1358e65fe4b408ad6d432a5f878f08194bdb4752441fd56faefa9b2b6f2"

    config = &oidc.Config{
        ClientID: policyAUD,
    }
    keySet   = oidc.NewRemoteKeySet(ctx, certsURL)
    verifier = oidc.NewVerifier(teamDomain, keySet, config)
)

// VerifyToken is a middleware to verify a CF Access token
func VerifyToken(next http.Handler) http.Handler {
    fn := func(w http.ResponseWriter, r *http.Request) {
        headers := r.Header

        // Make sure that the incoming request has our token header
        //  Could also look in the cookies for CF_AUTHORIZATION
        accessJWT := headers.Get("Cf-Access-Jwt-Assertion")
        if accessJWT == "" {
            w.WriteHeader(http.StatusUnauthorized)
            w.Write([]byte("No token on the request"))
            return
        }

        // Verify the access token
        ctx := r.Context()
        _, err := verifier.Verify(ctx, accessJWT)
        if err != nil {
            w.WriteHeader(http.StatusUnauthorized)
            w.Write([]byte(fmt.Sprintf("Invalid token: %s", err.Error())))
            return
        }
        next.ServeHTTP(w, r)
    }
    return http.HandlerFunc(fn)
}

func MainHandler() http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("welcome"))
    })
}

func main() {
    http.Handle("/", VerifyToken(MainHandler()))
    http.ListenAndServe(":3000", nil)
}

Python 示例

pip 安装以下包：

flask
requests
PyJWT
cryptography

from flask import Flask, request
import requests
import jwt
import json
import os
app = Flask(__name__)


# The Application Audience (AUD) tag for your application
POLICY_AUD = os.getenv("POLICY_AUD")

# Your CF Access team domain
TEAM_DOMAIN = os.getenv("TEAM_DOMAIN")
CERTS_URL = "{}/cdn-cgi/access/certs".format(TEAM_DOMAIN)

def _get_public_keys():
    """
    Returns:
        List of RSA public keys usable by PyJWT.
    """
    r = requests.get(CERTS_URL)
    public_keys = []
    jwk_set = r.json()
    for key_dict in jwk_set['keys']:
        public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(key_dict))
        public_keys.append(public_key)
    return public_keys

def verify_token(f):
    """
    Decorator that wraps a Flask API call to verify the CF Access JWT
    """
    def wrapper():
				# Check for the POLICY_AUD environment variable
				if not POLICY_AUD:
					return "missing required audience", 403

        token = ''
        if 'CF_Authorization' in request.cookies:
            token = request.cookies['CF_Authorization']
        else:
            return "missing required cf authorization token", 403
        keys = _get_public_keys()

        # Loop through the keys since we can't pass the key set to the decoder
        valid_token = False
        for key in keys:
            try:
                # decode returns the claims that has the email when needed
                jwt.decode(token, key=key, audience=POLICY_AUD, algorithms=['RS256'])
                valid_token = True
                break
            except:
                pass
        if not valid_token:
            return "invalid token", 403

        return f()
    return wrapper


@app.route('/')
@verify_token
def hello_world():
    return 'Hello, World!'


if __name__ == '__main__':
    app.run()

JavaScript (Node.js) 示例

const express = require("express");
const jose = require("jose");

// The Application Audience (AUD) tag for your application
const AUD = process.env.POLICY_AUD;

// Your CF Access team domain
const TEAM_DOMAIN = process.env.TEAM_DOMAIN;
const CERTS_URL = `${TEAM_DOMAIN}/cdn-cgi/access/certs`;

const JWKS = jose.createRemoteJWKSet(new URL(CERTS_URL));

// verifyToken is a middleware to verify a CF authorization token
const verifyToken = async (req, res, next) => {
	// Check for the AUD environment variable
	if (!AUD) {
		return res.status(403).send({
			status: false,
			message: "missing required audience",
		});
	}

	const token = req.headers["cf-access-jwt-assertion"];

	// Make sure that the incoming request has our token header
	if (!token) {
		return res.status(403).send({
			status: false,
			message: "missing required cf authorization token",
		});
	}

	try {
		const result = await jose.jwtVerify(token, JWKS, {
			issuer: TEAM_DOMAIN,
			audience: AUD,
		});

		req.user = result.payload;
		next();
	} catch (err) {
		return res.status(403).send({
			status: false,
			message: "invalid token",
		});
	}
};

const app = express();

app.use(verifyToken);

app.get("/", (req, res) => {
	res.send("Hello World!");
});

app.listen(3333);