多云身份管控平台构建——第二篇 基于动态策略实现Auth0与亚马逊云控制台的权限集成

 	{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "ec2:Describe*",
                "ec2:List*"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:StartInstances"
            ],
            "Resource": "arn:aws:ec2:{{region}}:{{accountid}}:instance/*",
            "Condition": {
                "StringEquals": {
                    "aws:ResourceTag/environment": "{{project}}"
                }
            }
        }
    ]
}

exports.onExecutePostLogin = async (event, api) => {
  const namespace = 'https://mytesttest.com';
  const project = event.user.user_metadata.project;
  const role = event.user.user_metadata.role;

  //if (event.authorization) { 这里为简便起见忽略了验证部分，实际项目中可根据安全需要对applications ->api -> permission中限定的scope进行验证，允许访问相应的内容时，才将用户的相关信息放入IDToken
    // Set claims 
    api.idToken.setCustomClaim(`${namespace}/project`, event.user.user_metadata.project);
    api.idToken.setCustomClaim(`${namespace}/role`, event.user.user_metadata.role);
  //}
};

def lambda_handler(event, context):
    #从Event中获取APIGateway中传来的IDToken
    token = event['authorizationToken'].split(" ")
    if (token[0] != 'Bearer'):
        raise Exception('Authorization header should have a format Bearer <JWT> Token')
    jwt_bearer_token = token[1]  
    unauthorized_claims = jwt.get_unverified_claims(jwt_bearer_token) 
    #并从IDToken中获得project和role的属性
    project_attri = unauthorized_claims['https://mytesttest.com/project']
    accessrole_attri = unauthorized_claims['https://mytesttest.com/role']
    
    #从Auth0中创建的API Application中获得jwt签名的公共密钥
    keys_url = 'https://dev-gzu7esjn.us.auth0.com/.well-known/jwks.json' 
    with urllib.request.urlopen(keys_url) as f:
        response = f.read()
    keys = json.loads(response.decode('utf-8'))['keys']

    #用公共密钥校验IDToen中的签名部分，并核对IDToken中的aud是否为Auth0中创建的API Application对应的ClientID，是则认为用户校验通过。该部分代码篇幅原因删略，如需要请参见github链接
    response = validateJWT(jwt_bearer_token, appclientid, keys)
    
    #get authenticated claims
    if (response == False):
        raise Exception('Unauthorized')
    else:
        principal_id = response["sub"]
    
    #生成APIGateway的policy--authResponse，通过Authorizer lambda 返回给apigw，决定是否允许调用APIGateway
    tmp = event['methodArn'].split(':')
    api_gateway_arn_tmp = tmp[5].split('/')
    aws_account_id = tmp[4]  
    policy = AuthPolicy(principal_id, aws_account_id)
    policy.restApiId = api_gateway_arn_tmp[0]
    policy.region = tmp[3]
    policy.stage = api_gateway_arn_tmp[1]
    policy.allowAllMethods() 
    authResponse = policy.build()
    
    #同时，获取IDToken中的用户属性信息，结合策略模版为后端的业务lambda生成动态policy：deliverpolicy，通过上下文传给后端业务lambda
    deliverpolicy = policyGenerator(project_attri, accessrole_attri)
    ##将生成的动态policy向后端业务传递
    context = {
        'policy': deliverpolicy
    }
    authResponse['context'] = context
    return authResponse
    
## 根据用户的属性信息从DynamoDB中取出策略模版，使用策略模版动态生成policy。模版替换使用mastache的python版本实现。    
def policyGenerator(project, projrole):
    dynamodb = boto3.client('dynamodb')
    response = dynamodb.get_item(
        TableName="policytemplate",
        Key={
        'project': {'S': 'project'},
        'role': {'S': 'projrole'}
        }
    )
    thecontext = {'region':REGIONID,'accountid':ACCOUNTID,'project':project}
    temppolicy = pystache.render(response['Item']['policy']['S'],thecontext)
    return temppolicy

def lambda_handler(event, context):
  ##由于Amazon console不支持角色串联（参见https://aws.amazon.com/cn/premiumsupport/knowledge-center/iam-role-chaining-limit/），因此示例中使用了IAM user的AKSK再通过assumerole的方式挂载policy
    session = boto3.session.Session(aws_access_key_id="xxxxx", aws_secret_access_key="xxxx")
    sts_connection = session.client('sts')
    ##在APIGateway传递来的event中获取由authorizer lambda生成的动态policy
    thepolicy = event['requestContext']['authorizer']['policy']
    
    ##由于assume_role最终生成的临时凭证是基础role_arn和policy的合集，因此示例中选用的较大权限。实际生产中可以根据需要选择合理范围。同时注意对role_arn创建对指定aksk的用户的信任策略，以便assumerole具有权限。
    role_arn = "arn:aws:iam::{}:role/adminrole".format(aws_account_id)
    assumed_role = sts_connection.assume_role(
        RoleArn=role_arn,
        RoleSessionName="console-session",
        Policy=thepolicy
    )
    credentials = assumed_role["Credentials"]
    # 获得临时凭证的AKSK和session Token
    url_credentials = {}
    url_credentials['sessionId'] = credentials['AccessKeyId']
    url_credentials['sessionKey'] = credentials['SecretAccessKey']
    url_credentials['sessionToken'] = credentials["SessionToken"]
    json_string_with_temp_credentials = json.dumps(url_credentials)
    
    #用临时凭证构建Amazon console
    request_parameters = "?Action=getSigninToken"
    request_parameters += "&SessionDuration=43200"
    if sys.version_info[0] < 3:
        def quote_plus_function(s):
            return urllib.quote_plus(s)
    else:
        def quote_plus_function(s):
            return urllib.parse.quote_plus(s)
    
    request_parameters += "&Session=" + quote_plus_function(json_string_with_temp_credentials)
    request_url = "https://signin.aws.amazon.com/federation" + request_parameters
    r = requests.get(request_url)
    signin_token = json.loads(r.text)

    request_parameters = "?Action=login"
    request_parameters += "&Issuer=Example.org"
    request_parameters += "&Destination=" + quote_plus_function("https://console.aws.amazon.com/")
    request_parameters += "&SigninToken=" + signin_token["SigninToken"]
    request_url = "https://signin.aws.amazon.com/federation" + request_parameters
    
    return {
        'statusCode': 200,
        'body': json.dumps(request_url)
    }