亚马逊云科技异常流量监控告警

{
  "vpc_endpoint_id": {
    "S": "vpce-0593f03c52cc203aa"
  },
  "hostzone_id": {
    "S": "Z0546917VILAMMMMMMMM"
  },
  "record_create_time": {
    "S": "2024-11-04T02:54:13.971082"
  },
  "record_delete_time": {
    "S": "9999-12-31T00:00:00"
  },
  "record_name": {
    "S": "monitor.dev.aws.example.cn."
  },
  "target_ips": {
    "L": [
      {
        "S": "100.65.249.139"
      },
      {
        "S": "100.65.248.112"
      }
    ]
  }
}

# 从DynamoDB表中读取数据
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(dynamodb_mapping_table)
response = table.scan(
    FilterExpression=Attr('record_delete_time').eq('9999-12-31T00:00:00')
)
items = response['Items']

# 将DynamoDB中读取的数据转换为Spark DataFrame,并重新分区为1个分区
dynamodb_schema = StructType([
    StructField("vpc_endpoint_id", StringType(), True),
    StructField("record_name", StringType(), True),
    StructField("hostzone_id", StringType(), True),
    StructField("record_create_time", StringType(), True),
    StructField("record_delete_time", StringType(), True),
    StructField("target_ips", ArrayType(StringType()), True)
])
dynamodb_df = spark.createDataFrame(items, schema=dynamodb_schema)
dynamodb_df = dynamodb_df.repartition(1)

# 从AWS Glue Data Catalog中读取源表,创建DynamicFrame
datasource0 = glueContext.create_dynamic_frame.from_catalog(
    database= catalog_database,
    table_name= source_catalog_table,
    transformation_ctx="datasource0"
)

# 使用DynamoDB表中的target_ips列过滤VPC Flow Logs数据
filtered_vpc_flow_logs_df = filtered_data_frame.join(
    dynamodb_df.select("target_ips"),
    (array_contains(col("target_ips"), col("srcaddr")) |
     array_contains(col("target_ips"), col("dstaddr"))),
    how="inner"
)

# 将过滤后的数据写回AWS Glue Data Catalog中的目标表,并设置分区键和增量更新策略
catalogsink = glueContext.write_dynamic_frame.from_catalog(
    frame=DynamicFrame.fromDF(joined_df, glueContext, "filtered_applymapping"),
    database= catalog_database,
    table_name= target_catalog_table,
    transformation_ctx="write_catalog",
    additional_options={
        "enableUpdateCatalog": True,
        "updateBehavior": "UPDATE_IN_DATABASE",
        "partitionKeys": ["aws-account-id", "aws-service", "aws-region", "year", "month", "day"]
    }
)

{ "requestId":"$context.requestId", "extendedRequestId":"$context.extendedRequestId","ip": "$context.identity.sourceIp", "caller":"$context.identity.caller", "user":"$context.identity.user", "requestTime":"$context.requestTime", "httpMethod":"$context.httpMethod", "domainName":"$context.domainName", "domainPrefix":"$context.domainPrefix", "resourcePath":"$context.resourcePath", "status":"$context.status", "protocol":"$context.protocol", "responseLength":"$context.responseLength"}

# 从Lambda环境变量中获取API跟踪日志的白名单路径
API_TRACELOGS_WHITELISTS = os.environ['API_TRACELOGS_WHITELISTS']

# 缓存已判断过的IP地址,避免重复判断影响执行效率
ip_cache = {}
# 从亚马逊云科技 Systems Manager Parameter Store中获取白名单IP列表
def get_whitelist_ips(whitelist_paths):
    ssm_client = boto3.client('ssm')
    whitelist_ips = []
    whitelist_paths = whitelist_paths.split(',')
    for path in whitelist_paths:
        try:
            response = ssm_client.get_parameter(Name=path, WithDecryption=True)
            whitelist_ips.extend(response['Parameter']['Value'].split('\n'))
        except Exception as e:
            print(f"Error getting whitelist from {path}: {e}")
    return whitelist_ips
# 预处理白名单IP列表,将IP地址转换为网络对象,并去除重叠的网络
def preprocess_whitelist(whitelist_ips):
    whitelist_networks = set()
    for ip_str in whitelist_ips:
        try:
            network = ipaddress.ip_network(ip_str)
            if any(network.subnet_of(existing_network) for existing_network in whitelist_networks):
                continue
            whitelist_networks.add(network)
        except ValueError:
            continue
    return whitelist_networks
# 全局变量,只在Lambda函数加载时初始化一次
whitelist_networks = preprocess_whitelist(get_whitelist_ips(API_TRACELOGS_WHITELISTS))
# 检查IP地址是否在白名单中
def is_ip_whitelisted(ip, whitelist_networks):
    if ip in ip_cache:
        return ip_cache[ip]
      try:
        ip_obj = ipaddress.ip_address(ip)
    except ValueError:
        ip_cache[ip] = False
        return False
        for network in whitelist_networks:
        if ip_obj in network:
            ip_cache[ip] = True
            return True
      ip_cache[ip] = False
    return False
### 使用新的解决方案进行IP地址检查和验证 ###
def transformLogEvent(log_event, owner, logGroup, subscriptionFilters):
    try:
        message = json.loads(log_event['message'])
        # 过滤不符合API网关日志格式的记录
        if not (message.get('requestTime') and message.get('ip') and message.get('domainName') and message.get('responseLength')): 
            print('Some records abandon while calling transformLogEvent, Need to redefine the API网关 log format according to the Abnormal Traffic Guideline.')
            print(f"Debug info: owner={owner}, logGroup={logGroup}, subscriptionFilters={subscriptionFilters}")
            print(f"Original message: {message}")
            return None
        
        message['owner'] = owner
        message['logGroup'] = logGroup
        message['subscriptionFilters'] = subscriptionFilters
        request_time = datetime.strptime(message['requestTime'], "%d/%b/%Y:%H:%M:%S %z")
        message['requestTime'] = int(time.mktime(request_time.timetuple()))
        # 检查IP地址是否在白名单中,并在返回的消息中添加新字段
        ip = message.get('ip')
        if ip:
            try:
                message['is_whitelisted'] = is_ip_whitelisted(ip, whitelist_networks)
            except ValueError:
                message['is_whitelisted'] = False
        return json.dumps(message) + '\n'
    except Exception as e:
        print(f"Error processing log event: {e}")
        return None

def create_or_update_partition(DATABASE_NAME, TABLE_NAME, source_key):
    """
    该函数用于在 AWS Glue 数据目录中创建或更新分区。
    参数:
    DATABASE_NAME (str): Glue 数据库名称
    TABLE_NAME (str): Glue 表名称
    source_key (str): 源数据路径,例如 's3://bucket/path/to/data'
    功能:
    1. 根据 source_key 提取 Glue 数据库和表名
    2. 从 source_key 中解析出分区值
    3. 初始化 Glue 客户端
    4. 尝试获取分区信息,如果分区已经存在,则跳过创建
    5. 如果分区不存在,则:
        a. 获取表的详细信息
        b. 创建自定义存储描述符,包含新的分区位置
        c. 在 Glue 数据目录中创建新的分区
    6. 如果创建成功,则打印成功消息
    7. 如果出现其他异常,则打印异常信息
    """
    # 根据 source_key 提取 Glue 数据库和表名
    TABLE_NAME = get_table_name(source_key)
    source_key = unquote(source_key)
    parts = source_key.split('/')[1:-1]
    partitions_values = [part.split('=')[1] for part in parts]
    # 使用 Boto 3 初始化 Glue 客户端
    glue_client = boto3.client('glue')
    try:
        # 检查分区是否已经存在。如果存在,则跳过创建
        get_partition_response = glue_client.get_partition(
            DatabaseName=DATABASE_NAME,
            TableName=TABLE_NAME,
            PartitionValues=partitions_values
        )
        print('Glue partition already exists.')
    except Exception as e:
        # 检查异常是否为 EntityNotFoundException。如果是,则继续创建分区
        if type(e).__name__ == 'EntityNotFoundException':
            print('Retrieve Table Details:')
            get_table_response = glue_client.get_table(
                DatabaseName=DATABASE_NAME,
                Name=TABLE_NAME
            )
            # 提取现有的存储描述符,并创建带有新分区位置的自定义存储描述符
            storage_descriptor = get_table_response['Table']['StorageDescriptor']
            custom_storage_descriptor = copy.deepcopy(storage_descriptor)
            custom_storage_descriptor['Location'] = f"{storage_descriptor['Location']}/{''.join([f'{part}/' for part in parts])}"
            # 在 Glue 数据目录中创建新的 Glue 分区
            create_partition_response = glue_client.create_partition(
                DatabaseName=DATABASE_NAME,
                TableName=TABLE_NAME,
                PartitionInput={
                    'Values': partitions_values,
                    'StorageDescriptor': custom_storage_descriptor
                }
            )
            print('Glue partition created successfully.')
        else:
            # 根据业务需求处理异常
            print(e)

CREATE EXTERNAL TABLE `vpc_flow_logs_origin`(
  `version` int, 
  `account_id` string, 
  `interface_id` string, 
  `srcaddr` string, 
  `dstaddr` string, 
  `srcport` int, 
  `dstport` int, 
  `protocol` bigint, 
  `packets` bigint, 
  `bytes` bigint, 
  `start` bigint, 
  `end` bigint, 
  `action` string, 
  `log_status` string, 
  `vpc_id` string, 
  `subnet_id` string, 
  `instance_id` string, 
  `tcp_flags` int, 
  `type` string, 
  `pkt_srcaddr` string, 
  `pkt_dstaddr` string, 
  `region` string, 
  `az_id` string, 
  `sublocation_type` string, 
  `sublocation_id` string, 
  `pkt_src_aws_service` string, 
  `pkt_dst_aws_service` string, 
  `flow_direction` string, 
  `traffic_path` int)
PARTITIONED BY ( 
  `aws-account-id` string, 
  `aws-service` string, 
  `aws-region` string, 
  `year` string, 
  `month` string, 
  `day` string)
ROW FORMAT DELIMITED 
  FIELDS TERMINATED BY ' ' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  's3://central-vpc-flow-logs-012345678901-cn-north-1/AWSLogs'
TBLPROPERTIES ('skip.header.line.count'='1')
CREATE EXTERNAL TABLE `vpc_flow_logs_extend`(
  `srcaddr` string, 
  `dstaddr` string, 
  `srcport` int, 
  `dstport` int, 
  `protocol` bigint, 
  `packets` bigint, 
  `bytes` bigint, 
  `start` bigint, 
  `end` bigint, 
  `action` string, 
  `log_status` string, 
  `record_name` string, 
  `vpc_endpoint_id` string)
PARTITIONED BY ( 
  `aws-account-id` string, 
  `aws-service` string, 
  `aws-region` string, 
  `year` string, 
  `month` string, 
  `day` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://flowlogs-abnormal-logs-target-012345678901-cn-north-1/AWSLogs'
TBLPROPERTIES (
  'classification'='parquet', 
  'useGlueParquetWriter'='true')

SELECT
  date_trunc('minute', from_unixtime(CAST(requesttime AS DOUBLE))) AS minute_trunc,
  ip,
  sum(CAST(responseLength AS BIGINT)) AS addr
FROM
  api_gateway_flow_logs_extend
WHERE
   $__timeFilter(from_unixtime(CAST(requesttime AS DOUBLE)))
   AND domainname = '${api_name}'
GROUP BY
  date_trunc('minute', from_unixtime(CAST(requesttime AS DOUBLE))),
  ip
ORDER BY
  minute_trunc ASC,
  ip

WITH base_query AS (
SELECT
    COALESCE(SUM(CAST(responseLength AS BIGINT)) / 1024, 0) AS total_volume_size,
    COALESCE(domainname, '') AS api_name,
    COALESCE(ip, '') AS src_ip,
    COALESCE(owner, '') AS awsid
FROM
    api_gateway_flow_logs_extend
WHERE
    $__timeFilter(from_unixtime(CAST(requesttime AS DOUBLE)))
GROUP BY
    domainname, ip, owner
)
SELECT * FROM base_query
UNION ALL
SELECT 0, '', '', ''
WHERE NOT EXISTS (SELECT 1 FROM base_query)

templates: 
  default_template: |
    {{ define "sns.default.subject" }}{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}{{ end }}
    {{ define "sns.default.api.subject" }}APIGateway AccessLogs Alert {{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}{{ end }}

    {{ define "sns.default.api.message" }} 
    APIGateway Download Alert
    {{ range .Alerts }} 
    • Stage: {{ .Labels.env }} 
    • API Name: {{ .Labels.api_name }} 
    • Alert Name: {{ .Labels.alertname }} 
    • AWS Account Id: {{ .Labels.awsid }}
    • Alert Time: {{ .StartsAt }}
    • End user Source IP: {{ .Labels.src_ip }} 
    {{ with .Annotations.__values__ }}{{ if . }}• End user download volume size per minute: {{ reReplaceAll "^([0-9]+\\.[0-9]{2}).*$" "${1}" (reReplaceAll ".*\"A\":\\s*([0-9.]+).*" "${1}" .) }} KB
    • Threshold of download volume size per minute: {{ reReplaceAll ".*\"C\":\\s*([0-9.]+).*" "${1}" . }} KB{{ end }}{{ end }} 
    -------------------------------------
    {{ end }} 
    More Info please visit to grafana url - dashboard Abnormal Traffic - APIGateway AccessLogs : https://grafana.loggin-${CURRENT_TENANT}.tenant.${PLATFORM_STAGE}-${PLATFORM_PREFIX}.cn/dashboards
    Please investigate this API Gateway alert immediately.
    Best regards,
    The Platform Team
    {{ end }}