Cognito Identity Pool （1）

AWS Cognito提供了身份验证授权服务以及用户管理服务。

• 用户池 （中国区目前不支持）
• 身份池

App以第三方身份登录以后，可以通过Cognito identity pool以assume role的方式获取访问AWS资源的临时身份，应用端根据这个临时身份访问所需要的AWS资源。

架构图

1. App从微信登录
2. 请求经过API Gateway,从Lambda发起微信身份验证，并获取openid
3. 查找dynamodb是否存在openid对应的identity id
   1. 如果存在，获取identity id并从cognito获取token
   2. 如果不存在，调用GetOpenIdTokenForDeveloperIdentity创建identity id以及获取token并保存identity id到dynamodb里
4. 通过第3步中获取的token调用STS获取临时权限
5. 通过第4步中获取的临时权限去访问AWS内部资源

创建身份池

这里需要注意的是身份验证选Custom，我们并不能用Facebook或者Google登录，目前也不支持微信，QQ，支付宝登录。

通过用户ID获取Token

在AWS环境中运行(Lambda+API Gateway)

首先需要从第三方身份登录服务商那边获取到用户的唯一标识，如微信的UnionID或者OpenID，这里假设我们已经拿到了这个OpenID

为这个OpenID在身份池里创建相对应的Identity。AWS提供GetOpenIdTokenForDeveloperIdentity这个API来创建新的Identity或者绑定用户到到已存在的Identity上。

这里需要注意的是，因为目前Cognito不支持国内主流的三方认证方式，所以选择经过开发人员验证的身份，也就是第一步设置的自定义身份提供商(login.wechat.pocapp)

用户信息可以存在dynamodb

import boto3
client = boto3.client('cognito-identity')
def get_token_for_user(user_id):
    response = client.get_open_id_token_for_developer_identity(
        IdentityPoolId='cn-north-1:xxxxxx-identity-poolId-in-step-1',
        Logins={
            'login.wechat.pocapp': 'user_id'
        },
        TokenDuration=120
    )
    print(response)
    return response
get_token_for_user('test-user-1')

API返回：

{
    'IdentityId': 'string',
    'Token': 'string'
}

从identity browser里可以看到新的identity已经创建好了

通过Token从STS获取临时身份

在AWS环境中运行(Lambda+API Gateway)

我们可以使用AssumeRoleWithWebIdentity这个API来获取扮演这个角色的临时身份。
调用这个方法我们需要找到在创建身份池时候绑定的角色的ARN以及上一步获取到的Token。

sts_client = boto3.client('sts')
def get_sts_credential(token_response):
    sts_response = sts_client.assume_role_with_web_identity(
        RoleArn='arn:aws-cn:iam::your-account-id:role/Cognito_WeChatAppPOCAuth_Role',
        RoleSessionName='test-session-1',
        WebIdentityToken=token_response['Token'],
        DurationSeconds=900
    )
    print(sts_response)
    return sts_response
get_sts_credential(get_token_for_user('test-user-1'))

API返回：

{
    'Credentials': {
        'AccessKeyId': 'string',
        'SecretAccessKey': 'string',
        'SessionToken': 'string',
        'Expiration': datetime(2015, 1, 1)
    },
    'SubjectFromWebIdentityToken': 'string',
    'AssumedRoleUser': {
        'AssumedRoleId': 'string',
        'Arn': 'string'
    },
    'PackedPolicySize': 123,
    'Provider': 'string',
    'Audience': 'string'
}

测试临时权限访问S3服务

应用端运行(Android/IOS)

从上一步中获取到的临时权限中创建一个会话，然后由这个会话再去创建S3的Client，最后打印出有权限的S3桶。

def test_s3(sts_credentials):
    credential = sts_credentials['Credentials']
    session = boto3.Session(
        aws_access_key_id=credential['AccessKeyId'],
        aws_secret_access_key=credential['SecretAccessKey'],
        aws_session_token=credential['SessionToken'],
    )
    s3_client = session.client('s3')
    response = s3_client.list_buckets()
    print(response)
test_s3(get_sts_credential(get_token_for_user('test-user-1')))

因为我创建的Cognito角色并没有给访问S3的策略，所以返回了个Access Denied

Github Source Code