您现在的位置是：首页 > 其他

当前栏目

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

自动

2023-02-26 12:35:45 时间

背景介绍

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

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

本文便是针对上述的诉求，探讨如何在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
  - 优点：相对更为贴合本场景
  - 缺点：中国区暂时没有

本篇作者

猜你喜欢

最长无重复子串
写技术博客的一些心得分享
Java 多线程（七）：线程池
Java 多线程（五）：锁（三）
Java 多线程（四）：锁（二）
Java 多线程（三）：锁（一）
Java 多线程（二）：并发编程的三大特性
线性时间非比较类排序
Java 多线程（一）：基础
合并k个已排序的链表
HDFS 高可用分布式环境搭建
合并两个有序数组
连续子数组的最大和
HDFS 分布式环境搭建
容器盛水问题
大数加法
HDFS 伪分布式环境搭建
设计LRU缓存结构
两数之和
使用单调栈来解决的一些问题

zl程序教程