CloudflareをALB(AWS)の前に配置して、CloudflareのAccessでAzureAD認証してみました。
どのクラウドでも実装できるはずなので、ちょっと認証かけたい時などなかなか便利です。

構成

Cloudflare以外からのアクセスは、ALBのセキュリティーグループでブロックします。
自分のCloudflare以外のCloudflareからのアクセスは、Lambdaでjwtを検証してブロックします。

必要なもの

Cloudflareアカウント
AWSアカウント
ドメイン2個
- ドメインA
- ドメインB
ドメインBに対しAWS ACMで取得したワイルドカード証明書1個

設定

Azure AD認証の対象リソースを作成する
Cloudflare DNSを設定する
Cloudflare Accessを設定する
CloudfraleのIPアドレスをALBのセキュリティグループに設定する
テストする

1. Azure AD認証の対象リソースを作成する

AWS側でALBのターゲット用のLambdaを作成する
ALBを作成する
1. HTTPSのリスナーを作成しドメインBの証明書を紐づける
2. ターゲットグループを作成して、Lambdaをターゲットに設定する
Route53に、ALBのエイリアスレコードをドメインBのサブドメインとして設定する

jwtの検証は以下を参考にしました。
- https://developers.cloudflare.com/access/setting-up-access/validate-jwt-tokens/
Lambda作成時のハマりポイント
- Amazon Linux上でpip installしてLambdaのパッケージをzipで固めます。
- Amazon Linuxの現在のyumリポジトリのPythonのバージョンは3.6なので、Lambda側も3.6に合わせます。
- Lambdaのパッケージをzipで固める時、".libs_cffi_backend"配下も忘れずに含めます。
- Cloudflareからのjwtを検証するためALBからのヘルスチェックは失敗するので、ヘルスチェックを外します。

以下Lambda内で指定している値は、Cloudflareの設定後に確認します。

Lambda内で指定している定数	値
CERTS_URL	CloudflareのAccessのLogin Page Domain＋"/cdn-cgi/access/certs"
AUDIENCE_TAG	CloudflareのAccessのAccess PoliciesのAudience Tag

requirements.txt

pyjwt
cryptography
cffi

Lambda_function.py

import jwt
import json
import urllib.request

CERTS_URL = 'https://*******************/cdn-cgi/access/certs'
AUDIENCE_TAG = '**************************************'

def get_public_keys():
    headers = {
        'User-Agent': 'curl/7.61.1'
    }
    req = urllib.request.Request(CERTS_URL, headers=headers)
    with urllib.request.urlopen(req) as response:
        res = response.read()
    public_keys = []
    jwk_set = json.loads(res)

    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 validate(header):
    is_valid = False
    token = ''
    if 'cf-access-jwt-assertion' in header:
        token = header['cf-access-jwt-assertion']
    else:
        return is_valid

    keys = get_public_keys()

    for key in keys:
        try:
            decoded_token = jwt.decode(token, key=key, audience=AUDIENCE_TAG)
            # print(decoded_token)
            is_valid = True
            break
        except Exception as e:
            print(e)
            pass
    return is_valid


def lambda_handler(event, context):
    is_valid = validate(event['headers'])

    if is_valid:
        res = {
            "isBase64Encoded": False,
            "statusCode": 200,
            "statusDescription": "200 OK",
            "headers": {
                "Set-cookie": "cookies",
                "Content-Type": "text/html"
            }
        }

        res['body'] = """<html>
        <head>
        <title>hello world</title>
        </head>
        <body>
        <h1>Hello World</h1>
        </body>
        </html>"""
    else:
        res = {
            "isBase64Encoded": False,
            "statusCode": 403,
            "statusDescription": "403"
        }

    return res