事不过三——如何实现恶意登陆的自动防护

背景介绍

近些年，随着网络技术的发展，恶意的网络入侵事件时有发生，如何保障云上网络安全引起了越来越多客户的关注。同时，国内外相关的法律细则也在愈发的完善，其中都对恶意登陆控制台的相关防护提出了指导性的要求。具体来看，大致分为两点：

是否配置并启用了登录失败处理功能。
是否配置并启用了限制非法登录功能，非法登录达到一定次数后采取特定动作（如：登录失败处理次数限制，账户锁定等）。

本文便是针对上述的诉求，探讨如何在AWS侧实现相关的功能。

分析

对于绝大多数的客户来说，进行AWS控制台的登陆无外乎两条路，1）直接使用AWS的IAM用户进行登陆2）自建SSO平台，通过登陆第三方认证系统，然后登陆进AWS平台。那么相对应我们的方案思路也有两种：

方案1: 使用第三方SSO服务，例如ADFS，Authing等，在其侧进行相关用户锁定控制
方案2: 利用Cloudtrail中出现的用户登陆失败事件触发用户锁定功能

本文由于篇幅的限制，接下来会主要探讨基于方案2的实现和演进。

实践分析

项目架构参考：

实验说明：

使用IAM 用户A进行登陆，故意输入3次错误密码
对应的用户登陆错误事件会发送到CloudTrail中，配置trail监听发送日志到CloudWatch
CloudWatch对于2中收到用户登陆的日志
CLoudWatch中配置登陆失败事件订阅，目标为lambda函数
Lambda触发后会在DDB中插入登陆失败事件
Lambda会检查是否N分钟登陆失败超过三次
如果发现N分钟内超过三次失败，则禁用对应IAM 控制台登陆的权限

具体流程

（1）CloudTrail配置
登陆CloudTrail界面，点击Trail进行创建，因为本实验并不会进行基于S3的查询，所以S3可以选择任意一个bucket或者新建一个bucket使用。配置CloudWatch是否监听，选择启动，并创建新的日志组如名称：ct-cw-log；点击下一步进入事件监听，events项目只选择Management events，类别选择Write，然后点击下一步最终生成Trail监听。效果如下图所示：

（2）DynamoDB配置

配置如下的DynamoDB table：

其中：

Partition key： username，string类型
Sort key： randomUUID，string类型

（3）Lambda函数配置

进入Lambda界面，点击新建函数，选择运行环境为python3.7，选择一个具有DynamoDB读写，IAM login profile 删除权限的role（需要自建）；创建完毕后把下面的代码copy到code处，点击部署。

import gzip
import json
import base64¬
import boto3
import uuid
import time

DDBclient = boto3.client('DynamoDB')
IAMclient = boto3.client('iam')

## insert a iterm with username(pk) and random sort-k, and set ttl to target like 3 min
def diable_iam_function(userName):  
    ticks = time.time()
    print('current time is ', time.asctime( time.localtime(ticks) ))
    expiryDateTime = ticks + 180
    print('expire time is ', time.asctime( time.localtime(expiryDateTime) ))
    try:
        data = DDBclient.put_item(
            TableName='login_counter',
            Item={
                'username': {'S': userName},
                'randomUUID': { 'S': str(uuid.uuid4()) },
                'ttl': { 'N': str(expiryDateTime) } 
            }
        )
    except Exception as e:
        print('Exception: ', e) 
        
``` query username and get count ,if >=3 then disable the iam user
    record will expire after 3mins, and will modify this parm to be custmized
```

    data = DDBclient.query(
      TableName='login_counter',
      KeyConditionExpression='#name = :value',
      FilterExpression = '#t > :ttl', 
      ExpressionAttributeValues={
        ':value': {
          'S': userName
        },
        ':ttl': { 
          'N': str(time.time()), 
        }
        
      },
      ExpressionAttributeNames={
         '#name': 'username',
         '#t': 'ttl', 
      }
    )

## disable iam user if failed count >= 3   
    if data['Count'] >= 3:
        print('user', userName, 'is being deleted due to more than 3 times retry')
        try:
            response = IAMclient.delete_login_profile(
                UserName=userName,
            )
            
            result = "delete user", response
            return result
        except Exception as e:
            print('Exception: ', e)
    else:
        
        result = 'user', userName, 'current failed count is ', data['Count']
        print(result)
        return result
        
def lambda_handler(event, context):
    cw_data = event['awslogs']['data']
    compressed_payload = base64.b64decode(cw_data)
    uncompressed_payload = gzip.decompress(compressed_payload)
    payload = json.loads(uncompressed_payload)
    print('how many events:', len(payload['logEvents']))
    log_events = payload['logEvents']


#   iterate all log events    
    for event in log_events:
        output = event['message']
        data = json.loads(output)
        userName = data['userIdentity']['userName']
        diable_iam_function(userName)

代码逻辑解释：

每一条登陆失败的信息都会被记录在DynamoDB中，其主键为用户名，sort key为随机值，以当前时间+N（N取决于客户对于登陆失败限定的时间）设定其TTL。
当有新的失败触发时都会进行DynamoDB事件的插入，同时基于TTL进行查询是否已经有三次登陆失败发生，如果有那么通过delete IAM login profile进行console登陆的禁止。
查询时还需要基于TTL进行查询，主要因为在DynamoDB中的TTL时软删除，其实效性并没有保障，参考：https://docs.aws.amazon.com/amazonDynamoDB/latest/developerguide/TTL.html

（4） CloudWatch配置

进入到CloudWatch界面，搜寻在步骤1中创建的log group, 点击其下方的subscription filter，选择创建Lambda subscription。选择destination为步骤三中创建的lambda函数，在配置Configure log format and filters处按如下填写：

Log format：从下拉框选择Amazon CloudTrail
Subscription filter pattern：
- { ($.eventName = ConsoleLogin) && ($.errorMessage = “Failed authentication”) }
Subscription filter name：failed auth filter

然后点击Start Streaming 按钮。

（5）测试过程：

使用测试用户A，登陆失败三次后，连续一分钟内继续登陆登出操作，发现仍旧能正常操作；
15分钟过后发现不可登陆，用其他IAM用户登陆，发现测试用户A的console权限已经被禁止；
恢复用户A的console权限，连续错误登陆两次，三分钟后再错误登陆一次，15分钟后再进行登陆，登陆顺利。

（6）待需要解决问题：

CloudTrail 日志到CloudWatch 最大可能达到15分钟延迟（并无官方数据），导致了无法立即禁用掉错误登陆的用户，参考：https://summitroute.com/blog/2018/08/07/aws_cloudtrail_vs_cloudwatch_events_vs_event_history/
配置Trail监听会产额外的s3费用

（7）其他注意

最初测试时发现短时间内错误登陆多次，但是Lambda只触发了一次。查看日志后发现，从CloudWatch传过来的日志可能一条日志中包含了多个登陆错误事件，所以改成如上的循环读取logevent模式。

改进方案

项目架构参考：

与原始方案相比，主要的区别为用EventBridge监听CloudTrail新事件替换掉了CloudWatch监听CloudTrail日志的触发流程。主要解决的问题：

错误登陆事件可以立即在CloudTrail中观察到，Eventbridge也能立即得到监听，从而解决了不能立即禁用多次错误登陆IAM用户的问题
不需要新创建trail，节省了对应s3的费用

具体流程

（1）DynamoDB配置

配置如下的DynamoDB table，与上述实验相同

（2）Lambda函数配置

进入Lambda界面，点击新建函数，选择运行环境为python3.7，选择一个具有DDB读写，IAM login profile删除权限的role；创建完毕后把下面的代码copy到code处，点击部署

import gzip
import json
import base64
import boto3
import uuid
import time


DDBclient = boto3.client('DynamoDB')
IAMclient = boto3.client('iam')


## insert a iterm with username(pk) and random sort-k, and set ttl to target like 3 min
def diable_iam_function(userName):  
    ticks = time.time()
    print('current time is ', time.asctime( time.localtime(ticks) ))
    expiryDateTime = ticks + 180
    print('expire time is ', time.asctime( time.localtime(expiryDateTime) ))
    try:
        data = DDBclient.put_item(
            TableName='login_counter',
            Item={
                'username': {'S': userName},
                'randomUUID': { 'S': str(uuid.uuid4()) },
                'ttl': { 'N': str(expiryDateTime) } 
            }
        )
    except Exception as e:
        print('Exception: ', e) 
        
## query user from pk and get count ,if >=3 then disable the iam user
    data = DDBclient.query(
      TableName='login_counter',
      KeyConditionExpression='#name = :value',
      FilterExpression = '#t > :ttl', 
      ExpressionAttributeValues={
        ':value': {
          'S': userName
        },
        ':ttl': { 
          'N': str(time.time()), 
        }
        
      },
      ExpressionAttributeNames={
         '#name': 'username',
         '#t': 'ttl', 
      }
    )

# disable iam user if failed count >= 3   
    if data['Count'] >= 3:
        print('user', userName, 'is being deleted due to more than 3 times retry')
        try:
            response = IAMclient.delete_login_profile(
                UserName=userName,
            )¬¬
            
            result = "delete user", response
            return result
        except Exception as e:
            print('Exception: ', e)
    else:
        
        result = 'user', userName, 'current failed count is ', data['Count']
        print(result)
        return result
        
def lambda_handler(event, context):
    userName = event
    diable_iam_function(userName)

与原始方案相比，代码逻辑基本没有变化，唯一的变化是通过EvenBridge传过来的event是单条的，并且直接是错误登陆的用户名，从而代码中做了相应的调整。这个变化主要是由于EventBridge和CloudWatch传递事件的机制不同。

（3）EventBridge配置

进入EventBridge界面，点击新创建Rule。在Define rule detail界面，填写Name为: Falied auth trigger，然后点击下一步；拉到页面最下方Event Pattern处，选择Custom Pattern，复制如下内容

{
  "source": ["aws.signin"],
  "detail-type": ["AWS Console Sign In via CloudTrail"],
  "detail":{
    "responseElements": {
        "ConsoleLogin": [ { "anything-but": [ "Success" ] } ]
      }
   } 
}

点击下一步，select a target处选择Lambda，然后选择我们在步骤2中创建的Lambda函数，然后展开additional setting，配置如下：

Configure target input: 从下拉框选择Part of matched event
Specify the part of the matched event: 填写如下
- $.detail.userIdentity.userName

一路点击下一步，直到创建完成。

（4）测试过程：

使用测试用户A，登陆失败三次后，间隔5秒进行正确密码登陆，发现无法登陆
使用其他账号进行登陆，发现用户A已经被禁止console login的权限
恢复用户A的console权限，连续错误登陆两次，三分钟后再错误登陆一次，间隔5，10，180秒后登陆发现可以继续登陆

总结

通过改进后的方案，目前可以实现实时和事后两种方式对于IAM用户的锁定，其中实时方案更为贴近等保和客户的要求，在整体流程也更为简单，成本也较低。此外，关于为什么要使用DynamoDB作为登陆数据的存储，作者也进行了一系列调研，其及简要的优劣势分析如下所示：

基于事件流直接进行分析
- KDS + KDA，基于滑动窗口进行分析
  - 优点：基于SQL进行查询相对简单
  - 缺点：需要长时间运行，成本较高, 滑动窗口范围可能导致漏判
基于No-SQL进行存储
- Redis
  - 优点：其TTL为实时生效，减小了范围查询导致的延迟
  - 缺点：需要维护额外的半托管Elasticache集群，其成本也较高
- Dynamodb
  - 优点：按需使用，价格较低
  - 缺点：“软”TTL，查询效率相对低
基于时序数据库进行存储
- TimeStream
  - 优点：相对更为贴合本场景
  - 缺点：中国区暂时没有

亚马逊AWS官方博客

事不过三——如何实现恶意登陆的自动防护

背景介绍

分析

实践分析

改进方案

总结

本篇作者