环境准备：
1.安装python3.7和相关的依赖
并安装redis缓存数据库
pip install aliyun-python-sdk-core pip install aliyun-python-sdk-slb pip intall IPy pip intall redis pip intall paramiko
2.添加ram访问控制的编程接口用户
3.添加slb的访问控制策略并和需要频控的slb进行绑定
redis封堵ip的格式
脚本程序目录
Aliyun_SLB_Manager ├── helpers │ ├── common.py │ ├── email.py │ ├── remote.py │ └── slb.py ├── logs │ └── run_20210204.log └── run.py
# 程序核心就是使用shell命令对nginx的日志中出现的ip地址 和 访问的接口进行过滤，找出访问频繁的那些程序加入slb黑名单，同时加入redis缓存，因为slb有封堵ip个数限制，redis中存储的ip需要设置过期时间，对比后删除slb中封堵的Ip
# grep 04/Feb/2021:15:4 /data/www/logs/nginx_log/access/masterapi.chinasoft.cn_access.log | grep \'/api\' | awk \'{print $1}\' | sort | uniq -c | sort -r -n | head -200
2454 114.248.45.15
1576 47.115.122.23
1569 47.107.239.148
269 112.32.217.52
grep 04/Feb/2021:14:5 /data/www/logs/nginx_log/access/masterapi.chinasoft.cn_access.log | grep \'/api\' | awk \'{print $1}\' | awk -F \':\' \'{print $2}\' | sort | uniq -c | sort -r -n | head -200 | awk \'{if ($1 >15)print $1,$2}\'
[root@alisz-edraw-api-server-web01:~]# grep 04/Feb/2021:15:4 /data/www/logs/nginx_log/access/masterapi.chinasoft.cn_access.log | grep \'/api\' | awk \'{print $1}\' | sort | uniq -c | sort -r -n | head -3
2454 114.248.45.15
1576 47.115.122.23
1569 47.107.239.148
python脚本
主入口程序
run.py
import time
from helpers.email import send_mail
from helpers.remote import get_black_ips
from helpers.common import is_white_ip,get_ban_ip_time,set_ban_ip_time,groups
from helpers.slb import slb_add_host,slb_del_host,slb_get_host
if __name__ == \"__main__\":
# aliyun 访问控制针对 slb 的管理用户
# 用户登录名称 slb-frequency-user@xxx.onaliyun.com
accessKeyId = \'id\'
accessSecret = \'pass\'
# slb 访问控制策略id
acl_id = \'acl-slb\'
# reginid 查询地址：https://help.aliyun.com/document_detail/40654.html?spm=a2c6h.13066369.0.0.54a17471VmN3kA
region_id = \'cn-shenzhen\'
# 黑名单限制个数 300
slb_limit = 200
# 每10分钟访问限制阈值
threshold = 50
# 接收邮箱
mails = [\'reblue520@chinasoft.cn\']
# 远程ssh执行grep过滤出可疑ip
res = get_black_ips(threshold)
deny_host_list = res[0]
hosts_with_count = res[1]
hosts_with_count = sorted(hosts_with_count.items(), key=lambda x: x[1] , reverse=True)
print(hosts_with_count)
# exit()
# 等待被ban的ip , 过滤掉ip白名单
deny_hosts = []
for host in deny_host_list:
if (is_white_ip(host) == False):
deny_hosts.append(host + \'/32\')
# 获取所有已经被ban的ip
response = slb_get_host(accessKeyId , accessSecret , acl_id , region_id)
denied_hosts = []
if(\'AclEntrys\' in response.keys()):
for item in response[\'AclEntrys\'][\'AclEntry\']:
denied_hosts.append(item[\'AclEntryIP\'])
# 被ban超过2天，首先移除
must_del_hosts = []
denied_hosts_clone = denied_hosts.copy()
for host in denied_hosts:
if (get_ban_ip_time(host) == 0 or (get_ban_ip_time(host) < int(round(time.time())) - 2* 24 * 3600)):
must_del_hosts.append(host)
denied_hosts_clone.remove(host)
# 排除相同的
deny_hosts_new = []
for item in deny_hosts:
if(item not in denied_hosts_clone):
deny_hosts_new.append(item)
# 两者和超过300的限制
if((len(denied_hosts_clone)+len(deny_hosts_new))>slb_limit):
denied_hosts_detail = {}
for host in denied_hosts_clone:
denied_hosts_detail[host] = get_ban_ip_time(host)
# 需要排除的数量
num = len(denied_hosts_clone) + len(deny_hosts_new) - slb_limit
denied_hosts_detail = sorted(denied_hosts_detail.items(), key=lambda x: x[1])
denied_hosts_detail = denied_hosts_detail[:num]
for item in denied_hosts_detail:
must_del_hosts.append(item[0])
print(\"denied:\",denied_hosts)
print(\"delete:\",must_del_hosts)
print(\"add:\",deny_hosts_new)
# exit()
# 先删除一部分 must_del_hosts
if(len(must_del_hosts)>0):
if (len(must_del_hosts)>50):
must_del_hosts_clone = groups(must_del_hosts,50)
for item in must_del_hosts_clone:
slb_del_host(item, accessKeyId, accessSecret, acl_id, region_id)
time.sleep(1)
else :
slb_del_host(must_del_hosts, accessKeyId, accessSecret, acl_id, region_id)
# 再新增 deny_hosts_new
if(len(deny_hosts_new)>0):
if(len(deny_hosts_new)>50):
deny_hosts_new_clone = groups(deny_hosts_new,50)
for item in deny_hosts_new_clone:
slb_add_host(item, accessKeyId, accessSecret, acl_id, region_id)
time.sleep(1)
else:
slb_add_host(deny_hosts_new, accessKeyId, accessSecret, acl_id, region_id)
# 记录ip被禁时间
for host in deny_hosts_new:
set_ban_ip_time(host)
if (len(deny_hosts_new) >= 1):
mail_content = \'\'
if(len(must_del_hosts) > 0):
mail_content += \"以下黑名单已被解禁(\"+str(len(must_del_hosts))+\"):\\n\"+\"\\n\".join(must_del_hosts) + \"\\n\"
mail_content += \"\\n新增以下ip黑名单(\"+str(len(deny_hosts_new))+\"):\\n\"+\"\\n\".join(deny_hosts_new)
mail_content += \"\\n\\n10分钟访问超过15次(\"+str(len(hosts_with_count))+\"):\\n\"
for item in hosts_with_count:
mail_content += str(item[1]) + \" \" + str(item[0]) + \"\\n\"
mail_content += \"\\n\\n黑名单(\"+str(len(denied_hosts))+\"个):\\n\"
for item in denied_hosts:
mail_content += str(item) + \"\\n\"
send_mail(mail_content , mails)
slb操作相关的脚本
slb.py
import logging , json
from aliyunsdkcore.client import AcsClient
from aliyunsdkslb.request.v20140515.AddAccessControlListEntryRequest import AddAccessControlListEntryRequest
from aliyunsdkslb.request.v20140515.RemoveAccessControlListEntryRequest import RemoveAccessControlListEntryRequest
from aliyunsdkslb.request.v20140515.DescribeAccessControlListAttributeRequest import DescribeAccessControlListAttributeRequest
# 阿里云slb访问控制里添加ip
def slb_add_host(hosts, accessKeyId, accessSecret, acl_id, region_id):
client = AcsClient(accessKeyId, accessSecret, region_id)
request = AddAccessControlListEntryRequest()
request.set_accept_format(\'json\')
logging.info(\"正在封印IP：%s\" % \",\".join(hosts))
try:
add_hosts = []
for host in hosts:
add_hosts.append({\"entry\": host, \"comment\": \"deny\"})
request.set_AclEntrys(add_hosts)
request.set_AclId(acl_id)
response = client.do_action_with_exception(request)
print(response)
except BaseException as e:
logging.error(\"添加黑名单失败，原因：%s\" % e)
# slb删除ip
def slb_del_host(hosts, accessKeyId, accessSecret, acl_id , region_id = \'us-west-1\'):
logging.info(\"正在解封IP：%s\" % \",\".join(hosts))
try:
del_hosts = []
for host in hosts:
del_hosts.append({\"entry\": host, \"comment\": \"deny\"})
client = AcsClient(accessKeyId, accessSecret, region_id)
request = RemoveAccessControlListEntryRequest()
request.set_accept_format(\'json\')
request.set_AclEntrys(del_hosts)
request.set_AclId(acl_id)
client.do_action_with_exception(request)
logging.info(\"slb删除IP:%s成功\" % \",\".join(hosts)) # 查看调用接口结果
logging.info(\"slb删除IP:%s成功\" % \",\".join(hosts)) # 查看调用接口结果
except BaseException as e:
logging.error(\"移出黑名单失败，原因：%s\" % e)
# 阿里云slb获取IP黑名单列表
def slb_get_host(accessKeyId, accessSecret, acl_id, region_id):
client = AcsClient(accessKeyId, accessSecret, region_id)
request = DescribeAccessControlListAttributeRequest()
request.set_accept_format(\'json\')
try:
request.set_AclId(acl_id)
response = client.do_action_with_exception(request)
data_sub = json.loads((response.decode(\"utf-8\")))
return data_sub
except BaseException as e:
logging.error(\"获取黑名单失败，原因：%s\" % e)
远程操作日志的脚本
remote.py
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import datetime
import re
import paramiko
def get_black_ips(threshold = 100):
# file = \'/data/www/logs/nginx_log/access/*api*_access.log\'
file = \'/data/www/logs/nginx_log/access/masterapi.chinasoft.cn_access.log\'
# 可以 ssh 访问服务器 nginx 日志的用户信息
username = \'apache\'
passwd = \'pass\'
ten_min_time = (datetime.datetime.now() - datetime.timedelta(minutes=10)).strftime(\"%d/%b/%Y:%H:%M\")
ten_min_time = ten_min_time[:-1]
# 线上 需要对日志进行过滤的目标服务器，一般是内网ip，本地调试时可以直接使用外网ip方便调试
ssh_hosts = [\'1.1.1.1\']
deny_host_list = []
for host in ssh_hosts:
\'\'\'
# 过滤日志文件，需要显示如下效果，次数 ip地址,需要定位具体的api接口，否则误伤率极高
# grep 04/Feb/2021:15:2 /data/www/logs/nginx_log/access/masterapi.chinasoft.cn_access.log | grep \'/api\' | awk \'{print $1}\' | sort | uniq -c | sort -r -n | head -5 | awk \'{if ($1 >15)print $1,$2}\'
2998 116.248.89.2
2381 114.248.45.15
1639 47.107.239.148
1580 47.115.122.23
245 59.109.149.45
\'\'\'
shell = (
# \"grep %s %s | grep \'/index.php?submod=checkout&method=index&pid\' | awk \'{print $1}\' | awk -F \':\' \'{print $2}\' | sort | uniq -c | sort -r -n | head -200 | awk \'{if ($1 >15)print $1,$2}\'\") % (
# grep 04/Feb/2021:14:5 /data/www/logs/nginx_log/access/masterapi.chinasoft.cn_access.log | grep \'/api/user\' | awk \'{print $1}\' | awk -F \':\' \'{print $2}\' | sort | uniq -c | sort -r -n | head -200 | awk \'{if ($1 >15)print $1,$2}\'
\"grep %s %s | grep \'/api\' | awk \'{print $1}\' | sort | uniq -c | sort -r -n | head -200 | awk \'{if ($1 >2000)print $1,$2}\'\") % (
ten_min_time, file)
print(shell)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(host, port=2020, username=username, password=passwd)
stdin, stdout, stderr = ssh.exec_command(shell)
result = stdout.read().decode(encoding=\"utf-8\")
deny_host_re = re.compile(r\'\\d{1,99} \\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\')
deny_host_re = deny_host_re.findall(result)
deny_host_list = deny_host_list + deny_host_re
uniq_host = {}
for host_str in deny_host_list:
tmp = host_str.split(\' \')
if tmp[1] in uniq_host:
uniq_host[tmp[1]] += int(tmp[0])
else:
uniq_host[tmp[1]] = int(tmp[0])
deny_host_list = []
for v in uniq_host:
if (uniq_host[v] > threshold):
deny_host_list.append(v)
return [deny_host_list , uniq_host]
发送邮件的脚本
email.py
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import smtplib
from email.mime.text import MIMEText
from email.header import Header
import logging
def send_mail(host , receivers):
# 发送邮件的服务器，用户信息
mail_host = \"smtpdm-ap-southeast-1.aliyun.com\"
mail_user = \"admin@mail.chinasoft.com\"
mail_pass = \"pass\"
sender = \'admin@mail.chinasoft.com\'
message = MIMEText(\'chinasoft国内接口被刷,单个IP最近10分钟内访问超过阈值100次会收到此邮件告警!!!!\\n%s\' % (host), \'plain\', \'utf-8\')
message[\'From\'] = Header(\"chinasoft国内接口被刷\", \'utf-8\')
subject =\'[DDOS]购买链接接口异常链接!!\'
message[\'Subject\'] = Header(subject, \'utf-8\')
try:
smtpObj = smtplib.SMTP(mail_host, 80)
smtpObj.login(mail_user, mail_pass)
smtpObj.sendmail(sender, receivers, message.as_string())
logging.info(\"邮件发送成功\")
except smtplib.SMTPException as e:
logging.error(\"发送邮件失败，原因：%s\" % e)
配置文件
common.py
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import IPy
from functools import reduce
import redis,time
def groups(L1,len1):
groups=zip(*(iter(L1),)*len1)
L2=[list(i) for i in groups]
n=len(L1) % len1
L2.append(L1[-n:]) if n !=0 else L2
return L2
def ip_into_int(ip):
return reduce(lambda x, y: (x << 8) + y, map(int, ip.split(\'.\')))
# 过滤掉内网ip
def is_internal_ip(ip):
ip = ip_into_int(ip)
net_a = ip_into_int(\'10.255.255.255\') >> 24
net_b = ip_into_int(\'172.31.255.255\') >> 20
net_c = ip_into_int(\'192.168.255.255\') >> 16
return ip >> 24 == net_a or ip >> 20 == net_b or ip >> 16 == net_c
# 是否为白名单ip （公司内网+集群内网ip+slb和需要互访的服务器ip避免误杀）
def is_white_ip(ip):
if (is_internal_ip(ip)):
return True
white_hosts = [
# web-servers
\'1.1.1.1\',
\'1.1.1.2\',
];
for white in white_hosts:
if (ip in IPy.IP(white)):
return True
return False
def get_ban_ip_time(ip):
pool = redis.ConnectionPool(host=\'127.0.0.1\', port=6379, db=1)
client = redis.Redis(connection_pool=pool)
key = \'slb_ban_\'+ip
val = client.get(key)
if val == None:
return 0
else :
return int(val)
def set_ban_ip_time(ip):
pool = redis.ConnectionPool(host=\'127.0.0.1\', port=6379, db=1)
client = redis.Redis(connection_pool=pool)
key = \'slb_ban_\'+ip
timestamp = time.time()
timestamp = int(round(timestamp))
return client.set(key , timestamp , 86400)
本地可以直接运行run.py进行调试
© 版权声明
THE END
暂无评论内容