API Gateway LambdaオーソライザーでJWTトークン認証をやってみる
PythonのLambdaオーソライザー内でJWTトークンを検証して、検証OKなら後続のLambdaにトークンのペイロードを渡して呼び出すというのをやってみます。
まずはLambdaオーソライザーを作ります。
import logging
import jwt
logger = logging.getLogger()
logger.setLevel(logging.INFO)
SECRET = 'my-secret'
def lambda_handler(event, context):
logger.info(event)
token = event['authorizationToken']
try:
payload = jwt.decode(token, SECRET, algorithms=['HS256'])
logger.info(payload)
return generatePolicy('user', 'Allow', event['methodArn'], payload)
except DecodeError as e:
logger.error(e)
return generatePolicy('user', 'Deny', event['methodArn'], None)
def generatePolicy(principalId, effect, resource, context):
authResponse = {}
authResponse['principalId'] = principalId
if effect and resource:
policyDocument = {
'Version': '2012-10-17',
'Statement': [
{
'Sid': 'FirstStatement',
'Action': 'execute-api:Invoke',
'Effect': effect,
'Resource': resource
}
]
}
authResponse['policyDocument'] = policyDocument
if context:
authResponse['context'] = context
return authResponse出来上がった関数をAPI Gatewayのオーソライザーに登録します。
シークレットを違うものに変えれば署名検証が失敗してDenyが返ります。
次に認証を通った後に呼ばれるLambdaを作ります。
import json
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info(event)
context = event['requestContext']['authorizer']['foo']
return {
'statusCode': 200,
'body': json.dumps({
'context': context
})
}
$ curl https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/api/hello
{"message":"Unauthorized"}
$ curl https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/api/hello -H "Authorization:hoge"
{"Message":"User is not authorized to access this resource with an explicit deny"}
$ curl https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/api/hello \
-H "Authorization:eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJmb28iOiJiYXIifQ.9_46A6P4igXwgH4xY01wYFyBZjZCW5FOv_3tySVMC1Q"
{"context": "bar"}ちなみに後続のLambdaではコンテキストだけでなくプリンシパルIDも拾えます。