使用AWS KMS非对称密钥构建动态JWKS服务

本文详细介绍了如何将AWS KMS非对称密钥动态暴露为JWKS的技术方案,涵盖密钥选择、公钥转换到JWK格式,以及通过Lambda函数提供服务,解决服务间认证中的公钥轮换问题。

将AWS KMS非对称密钥暴露为JWKS

在Benchling,服务间交互是我们业务的重要组成部分,从员工与日常业务使用的软件即服务产品交互,一直到构成Benchling应用平台本身的服务间交互。服务的安全认证和授权是行业中长期存在的问题,但由于OAuth 2.0和OpenID Connect(OIDC)等现代标准的广泛采用,近年来有所改善。

使用OIDC进行现代认证

我们的特定身份服务供应商为其API提供了两种认证选项:管理员可以生成的API令牌,或作为应用程序进行交互。API令牌虽然非常易于使用,但由于两个原因并不是理想选择:

首先,它们是静态密钥,必须小心处理并频繁轮换以减轻密钥泄露风险,这会带来显著的管理开销。

其次,身份服务供应商将API令牌的特权和身份与生成它的管理员不可分割地联系在一起。这导致使用该密钥的操作被归因于管理员,并且也无法实现最小权限原则。

解决方案:AWS KMS非对称密钥

对于在云环境中管理私钥,AWS在2019年向KMS添加非对称密钥功能时为我们解决了这个问题。KMS非对称密钥功能允许使用已有的云基础设施工具配置公钥/私钥对。私钥永远不会离开AWS基础设施,但公钥可以导出和共享。

挑战:公钥轮换

KMS非对称密钥仍然需要轮换以实现最大安全性,但这意味着需要与身份提供商共享新的公钥。OIDC Discovery标准提供了一种使用标准Web技术发现OpenID提供商信息的通用方法,包括提供指向包含OpenID提供商公钥的JSON Web密钥集(JWKS)的URL。

JWKS构建和服务

我们的JWKS服务基本设计是一个Lambda函数,前面是相对较新的AWS Lambda函数URL功能。函数URL允许提供单函数微服务(如我们的JWKS导出器),而无需API网关的额外基础设施。

密钥选择

由于AWS API的ListKeys函数响应不包含在密钥上定义的标签,唯一合理的方法是使用密钥别名来识别要在JWKS中导出的密钥。

将公钥渲染为JWK

现在我们有要导出的KMS中的非对称密钥资源列表,需要将公钥转换为可以组合成JWKS的JWK结构。我们使用JWCrypto库来处理JWK功能,包括从PEM格式转换。

完整实现代码

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import os
import logging
import boto3
from cryptography.hazmat.primitives import serialization
from jwcrypto import jwk

kms = boto3.client('kms')

for var in ["ALIAS_START"]:
    if not var in os.environ:
        logging.critical(f'{var} environment variable not set')
        exit()

def find_enabled_keys(starts_with):
    alias_paginator = kms.get_paginator('list_aliases')
    alias_iterator = alias_paginator.paginate().search(f'Aliases[?TargetKeyId != null && starts_with(AliasName, `{starts_with}`)].TargetKeyId')
    key_ids = [page for page in alias_iterator]
    enabled_keys = [
        key_id
        for key_id in key_ids
        if kms.describe_key(KeyId=key_id)['KeyMetadata']['KeyState']
        == 'Enabled'
    ]
    return enabled_keys

def get_jwk(key_id):
    response = kms.get_public_key(KeyId=key_id)
    pubkey = serialization.\
        load_der_public_key(response['PublicKey'])
    pub_pem = pubkey.public_bytes(
         encoding=serialization.Encoding.PEM,
         format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    key = jwk.JWK.from_pem(pub_pem)
    key.update(use=response['KeyUsage'][0:3].lower(), kid=key_id)
    return key

def lambda_handler(event, context):
    jwks = {
         "keys": [
             get_jwk(key_id)
             for key_id
             in find_enabled_keys(os.environ['ALIAS_START'])
         ]
    }
    return({
        'statusCode': 200,
        'headers': {
            "Content-Type": "application/json",
        },
        'body': jwks
    })

生产环境增强

本文提供的代码应仅视为概念验证。如果需要在生产中使用此模式,应考虑您的使用场景。如果需要频繁重新加载此JWKS,更可扩展的方法是在发生更改时将其写入对象存储,并直接从对象存储或CDN提供JWKS。

comments powered by Disqus
使用 Hugo 构建
主题 StackJimmy 设计