基于AWS服务的特定用户IP访问控制方案问询（Cognito/IAM/API Gateway）

这是个很典型的基于身份+网络的细粒度访问控制需求，结合你现有的Cognito+IAM架构，有两种高效的实现方式，我给你拆解一下：

核心思路铺垫 #

首先得把用户的身份标签和对应的网络权限绑定起来，最直接的方式是在Cognito用户池里做配置：

• 用Cognito的**用户组（Groups）**来分类：创建Individual、CompanyA、CompanyB、CompanyC这几个组，把用户分到对应的组里（比给每个用户单独加自定义属性更便于批量管理）。
• 提前整理好网络A、B的IP段（CIDR格式），比如网络A是10.0.0.0/16，网络B是172.16.0.0/12。

方案一：自定义Lambda授权器（灵活可控） #

这是最适配你需求的方案，能直接把用户身份、请求源IP做联动验证，逻辑完全自定义：

1. 创建自定义授权器
   在API Gateway里创建一个Token类型的自定义授权器，触发条件可选“每个请求都触发”（或根据业务场景配置缓存），关联到你的Lambda函数。

2. 编写授权器Lambda逻辑
   核心代码逻辑大概是这样（伪代码示例）：

   import jwt
   import ipaddress
   from jose import jwk, jwt
   from jose.utils import base64url_decode
   
   def lambda_handler(event, context):
       # 1. 解析并验证Cognito Token的合法性
       token = event['authorizationToken'].split(' ')[1]
       # 实际项目中要通过Cognito公钥验证签名，这里简化示例
       decoded_token = jwt.decode(token, options={"verify_signature": False})
       user_groups = decoded_token.get('cognito:groups', [])
       user_category = next((g for g in user_groups if g in ['CompanyA','CompanyB','CompanyC']), 'Individual')
   
       # 2. 获取请求的真实源IP
       source_ip = event['requestContext']['identity']['sourceIp']
   
       # 3. 按规则验证权限
       if user_category in ['Individual', 'CompanyC']:
           return generate_allow_policy(decoded_token['sub'], event['methodArn'])
       elif user_category == 'CompanyA':
           allowed_cidr = ipaddress.ip_network('10.0.0.0/16')
           if ipaddress.ip_address(source_ip) in allowed_cidr:
               return generate_allow_policy(decoded_token['sub'], event['methodArn'])
           else:
               return generate_deny_policy(decoded_token['sub'], event['methodArn'])
       elif user_category == 'CompanyB':
           allowed_cidr = ipaddress.ip_network('172.16.0.0/12')
           if ipaddress.ip_address(source_ip) in allowed_cidr:
               return generate_allow_policy(decoded_token['sub'], event['methodArn'])
           else:
               return generate_deny_policy(decoded_token['sub'], event['methodArn'])
       else:
           return generate_deny_policy(decoded_token['sub'], event['methodArn'])
   
   # 辅助函数：生成允许/拒绝的IAM策略
   def generate_allow_policy(principal_id, resource):
       return {
           'principalId': principal_id,
           'policyDocument': {
               'Version': '2012-10-17',
               'Statement': [{
                   'Action': 'execute-api:Invoke',
                   'Effect': 'Allow',
                   'Resource': resource
               }]
           }
       }
   
   def generate_deny_policy(principal_id, resource):
       return {
           'principalId': principal_id,
           'policyDocument': {
               'Version': '2012-10-17',
               'Statement': [{
                   'Action': 'execute-api:Invoke',
                   'Effect': 'Deny',
                   'Resource': resource
               }]
           }
       }
   

3. 关联API资源
   把这个授权器绑定到你需要保护的API方法上，这样每个请求都会先经过授权器验证，通过后才会转发到Lambda业务函数。

方案二：AWS WAF + Cognito用户组（轻量配置） #

如果你的逻辑比较固定（只是IP+用户组的匹配），可以用WAF来实现，不用写太多代码，性能也更好：

1. 创建WAF IP集
   在AWS WAF里创建两个IP集：NetworkA_IPs和NetworkB_IPs，分别填入对应的CIDR段。

2. 创建WAF规则
   针对API Gateway的阶段创建WAF Web ACL，添加以下规则：

   • 规则1：拦截CompanyA用户的非网络A请求
     条件：Cognito用户组等于CompanyA AND 请求IP不在NetworkA_IPs
     动作：拦截（Block）
   • 规则2：拦截CompanyB用户的非网络B请求
     条件：Cognito用户组等于CompanyB AND 请求IP不在NetworkB_IPs
     动作：拦截（Block）
   • 规则3：允许其他所有请求
     条件：默认（所有请求）
     动作：允许（Allow）

3. 关联WAF到API Gateway
   把创建好的Web ACL关联到API Gateway的部署阶段，这样所有请求都会先经过WAF的规则校验。

优化与注意事项 #

• 授权器缓存：如果用自定义授权器，记得配置缓存（比如5分钟），减少Lambda调用次数，降低成本同时提高性能。
• 真实IP获取：如果API Gateway前面挂了CloudFront或者在VPC内部署，要确保能拿到真实的客户端IP，API Gateway的requestContext.identity.sourceIp在大多数场景下是准确的，但如果有代理，可能需要检查X-Forwarded-For头。
• Token验证：自定义授权器里一定要验证Cognito Token的签名和有效期，不要像伪代码那样跳过签名验证，避免伪造Token的风险。
• 用户组管理：用Cognito用户组分类比单个用户属性更高效，后续调整权限直接修改组配置或移动用户组即可。

内容的提问来源于stack exchange，提问作者rhythm