-
-
[原创]渗透测试之一次红包激励模块逻辑漏洞测试总结
-
发表于: 2022-5-1 22:56
-
下图是本次App逻辑漏洞测试个人总结的指导原则和测试方法，指导原则后面对应的是发现的漏洞，有些漏洞描述做了脱敏泛化处理，根据这些漏洞基本可以反推出测试用例。这些是本人认为App逻辑漏洞测试应该重点关注的，可能不全，如有补充欢迎探讨（其他sql注入、XSS等其他类型漏洞不是本次测试的重点，测试的对象App，另外看了后端代码都是用的orm框架一般也没有什么问题）。
图片看不清可以看下面这个：
（1）不同任务活跃时长可以直接发包绕过、直接达到活跃时长上限
（2）可以直接发包获取所有奖励（部分奖励有活跃时长等要求，但这些要求也可以直接发包完成）
（3）提现接口满足金额要求就可以发包提现，染过客户端特定金额按钮提现限制方式
（4）经验证在有提现微信账号openid的前提下是可以全程脱离APP通过发包实现提现1元到账的
（1）任务id是连续数字，可被发包枚举完成，与当前用户身份不匹配的任务也可以被完成并获取额外奖励(如新用户完成了老用户才有的任务等)
（2）不同金额的提现接口参数也是连续可枚举的，可以发包完成不同金额的提现，从而绕过客户端提现金额限制
（3）App核心内容也是连续可枚举的，相关内容可以被爬虫枚举爬取
（1）任务奖励次数通过重复发包(重置、完成任务、获取奖励接口)可以绕过约定的次数限制
（2）红包奖励接口没有加锁、存在并发漏洞导致奖励超发(阅读代码发现只有一个提现接口加锁了、其他都没加锁)
（3）绑定微信openid提现账号没有加锁存在并发漏洞，导致一个账号可以绑定多个微信账号(虽然进一步测试对提现并没有什么影响)
（4）提现接口通过发包可以绕过提现次数和金额的限制
（1）可以虚假号码接收短信注册为新用户并立马发包获取新用户的红包提现奖励(好在目前对微信提现账号有提现次数限制，微信提现账号要实名认证后才可以提现)
（2）账号注册后生成没后有效期token并缓存本地，只要有token就可以发包绑定微信提现，如果集成了恶意的SDK或者其他漏洞导致缓存的token被泄露，那么用户账号的余额就可以被不知不觉被转走(好在token是随机的uuid不可枚举否则就惨了，另外虽然一个微信绑定账号数有限制，但是有解绑功能，可以发包迅速解绑并重新绑定，而解绑和重新绑定并没有次数限制)
（1）sign算法通过jni反射调用java的标准MD5方法，极容易hook跟踪猜测
（2）so未做加壳混淆等保护破解相对容易
（3）也没有做so脱离app防调用以及防frida等注入app后rpc服务远程调用处理
（1）抓包、理清接口和重要参数的含义(有源码可以结合源码)
（2）编写burpsuite插件自动sign计算，便于重放等各种测试（有源码可直接看源码的算法，没有就只能破解）
（3）burpsuite中进行包重放、intruder（暴力枚举等）、并发（插件Turbo Intruder）测试(有源码可结合代码审计快速判断是否存在并发漏洞等)
公司不同App产品线的红包激励模块曾好几次被破解，出现过几次针对我司产品的autojs破解协议apk，本人作为风控部门专业做逆向的技术人员，曾逆向分析过这些黑产app，大部分都是抓包获取用户token后直接发包获取奖励，绕过看广告、做任务等环节。
而本次任务的主要目的就是主动出击，对公司核心产品的红包激励模块进行逻辑漏洞测试，看是否有容易被薅羊毛的的逻辑漏洞，对发现的漏洞督促业务线进行整改。虽然本人一直做的逆向方向的，但也长期关注安全相关的其他领域，对渗透测试的方法和工具等也是耳濡目染。本文主要是做本人本次所做渗透的测试简要总结。
通常需要编写burpsuite插件实现自动sign计算，这样在修改参数重放或并发测试时不需要关心sign计算问题，再评估完sign算法保护强度后，可以直接看代码算法实现，而不需要完全逆向还原可以提高测试效率。
很多漏洞，如并发漏洞等是可以通过代码审计一眼看出来的，不需要完全黑盒测试，这也可以极大提高测试效率。当然也有由于对代码逻辑不熟或代码逻辑过于复杂，导致看代码还不如直接黑盒测试来的快得情况。个人觉得两者是相辅相成的。
服务端对客户端请求都是并发处理的，而且很多服务端还有负载均衡多节点处理，这样当对数据库某字段进行加减操作（如红包发放）如果没有加锁就会导致竞争/并发漏洞，分布式锁一般用数据库锁或redis锁。
本次测试的一个重点就是看能否完全脱离app直接通过发包就可以完成提现的，经验证在有提现微信账号openid的前提下是可以全程脱离APP通过发包实现提现1元到账的。但微信的openid是跟app绑定的，不同app对应的openid不同，需要获取微信的openid，需要调⽤微信授权登录接⼝来获取，并且调⽤时会校验调⽤⽅app的签名，若app的签名与后台配置的签名不⼀致，⽆法成功调⽤，试过各种签名欺骗未生效，估计不是在app内校验签名的而是在微信里校验的，⽬前openid是抓包获取的。
想象一下如果可以完全脱离app就可以全程发包获取奖励并提现（如支付号手机号直接提现），那么即使限制了每个账号的提现次数那也是很恐怖的。比如开发一个在线的薅羊毛工具，然后到处传播，每个人只要扫个码或者填个手机号就可以提现一元，传播量大也很恐怖。
解决办法就是事先在app内做好各个用户行为的埋点数据和用户环境检测数据等，然后在提现重要关卡验证这些数据是否正常，不正常就拒绝提现。
关于如何使用burpsuite来进行测试，看官方帮助文档或网上资料基本就能学会这边就不展开细说，附上处理过burpsuite插件脚本供参考。
# -*- coding: utf-8 -*-from burp import IBurpExtender
from burp import IHttpListener
from burp import IRequestInfo
from burp import IParameter
from burp import IBurpExtenderCallbacks
from java.io import PrintWriter
import base64
from hashlib import md5
import time
HOST_FROM = "yy.xx.com"
# https://portswigger.net/burp/extender/api/index.htmlclass BurpExtender(IBurpExtender, IHttpListener):
#
# implement IBurpExtender
#
def registerExtenderCallbacks(self, callbacks):
# obtain an extension helpers object
self._helpers = callbacks.getHelpers()
self._stdout = PrintWriter(callbacks.getStdout(), True)
# set our extension name
callbacks.setExtensionName("zz resign plugin")
# register ourselves as an HTTP listener
callbacks.registerHttpListener(self)
#
# implement IHttpListener
#
def processHttpMessage(self, toolFlag, messageIsRequest, messageInfo):
# only process requests
if not messageIsRequest:
return
# get the HTTP service for the request
httpService = messageInfo.getHttpService()
# only process the host that is HOST_FROM
if (HOST_FROM != httpService.getHost()):
return
# self._stdout.println(toolFlag)
# self._stdout.println(IBurpExtenderCallbacks.TOOL_REPEATER)
if (IBurpExtenderCallbacks.TOOL_REPEATER != toolFlag) or (IBurpExtenderCallbacks.TOOL_INTRUDER != toolFlag):
return
requestInfo = self._helpers.analyzeRequest(messageInfo)
path = requestInfo.getUrl().getPath()
self._stdout.println("path=%s" % path)
if not path.startswith("/xx/"):
return
self.resign(messageInfo, check_sign=False)
def resign(self, messageInfo, check_sign=False):
requestInfo = self._helpers.analyzeRequest(messageInfo)
method = requestInfo.getMethod()
# self._stdout.println("method=%s" % method)
path = requestInfo.getUrl().getPath()
# self._stdout.println("path=%s" % path)
parameters = requestInfo.getParameters()
body_offset = requestInfo.getBodyOffset()
url_param_dist = {}
for param in parameters:
if IParameter.PARAM_URL == param.getType():
k = param.getName()
v = param.getValue()
v = self._helpers.urlDecode(v) # 需要转化一下有些%号没有decode，签名算法是错误的
self._stdout.println("%s=%s" % (k, v))
# ...
request = messageInfo.getRequest()
body = request[body_offset:]
body = self._helpers.bytesToString(body)
self._stdout.println("body=%s" % body)
s = '{}-{}'.format(method, path)
# params
old_sign = ""
new_time_sign = None
new_ts = str(int(time.time()))
if not check_sign:
url_param_dist["_ts"] = new_ts #更新时间戳
if "XX" in url_param_dist:
s = str(url_param_dist["XX"]) + time.strftime('%Y%m%d')
new_time_sign = md5(s.encode()).hexdigest()
url_param_dist["sign"] = new_time_sign #更新sign
arguments = url_param_dist
keys = list(arguments.keys())
keys.sort()
for k in keys:
if k == '_sign':
old_sign = arguments[k]
continue
val = arguments[k]
if type(val) == str or type(val) == unicode or type(val) == int:
s += '&{}={}'.format(k, val)
elif type(val) == list:
for v in val:
s += '&{}={}'.format(k, v)
else:
self._stdout.println('params unknown type:{}, key:{}'.format(type(val), k))
return
# json body
s += '&json={}'.format(body or '')
digest = md5(s.encode()).hexdigest()
new_sign = base64.urlsafe_b64encode(digest)[:-2]
# self._stdout.println('new_sign:{}, old_sign:{}'.format(new_sign, old_sign))
request = self._helpers.urlDecode(request)
if not check_sign:
new_request = self._helpers.updateParameter(request, self._helpers.buildParameter("ts", new_ts, IParameter.PARAM_URL))
new_request = self._helpers.updateParameter(new_request, self._helpers.buildParameter("_sign", new_sign, IParameter.PARAM_URL))
if new_time_sign:
new_request = self._helpers.updateParameter(new_request, self._helpers.buildParameter("sign", new_time_sign, IParameter.PARAM_URL))
# self._stdout.println(self._helpers.bytesToString(new_request))
messageInfo.setRequest(new_request)
else:
if new_sign != old_sign:
self._stdout.println("error! new_sign != old_sign")
else:
self._stdout.println("success! new_sign == old_sign")
# messageInfo.setHttpService(self._helpers.buildHttpService(HOST_TO,
# httpService.getPort(), httpService.getProtocol()))
# -*- coding: utf-8 -*-from burp import IBurpExtender
from burp import IHttpListener
from burp import IRequestInfo
from burp import IParameter
from burp import IBurpExtenderCallbacks
from java.io import PrintWriter
import base64
from hashlib import md5
import time
HOST_FROM = "yy.xx.com"
# https://portswigger.net/burp/extender/api/index.htmlclass BurpExtender(IBurpExtender, IHttpListener):
#
# implement IBurpExtender
#
def registerExtenderCallbacks(self, callbacks):
# obtain an extension helpers object
self._helpers = callbacks.getHelpers()
self._stdout = PrintWriter(callbacks.getStdout(), True)
# set our extension name
callbacks.setExtensionName("zz resign plugin")
# register ourselves as an HTTP listener
callbacks.registerHttpListener(self)
#
# implement IHttpListener
#
def processHttpMessage(self, toolFlag, messageIsRequest, messageInfo):
# only process requests
if not messageIsRequest:
return
# get the HTTP service for the request
httpService = messageInfo.getHttpService()
# only process the host that is HOST_FROM
if (HOST_FROM != httpService.getHost()):
return
# self._stdout.println(toolFlag)
# self._stdout.println(IBurpExtenderCallbacks.TOOL_REPEATER)
if (IBurpExtenderCallbacks.TOOL_REPEATER != toolFlag) or (IBurpExtenderCallbacks.TOOL_INTRUDER != toolFlag):
return
requestInfo = self._helpers.analyzeRequest(messageInfo)
path = requestInfo.getUrl().getPath()
self._stdout.println("path=%s" % path)
if not path.startswith("/xx/"):
return
self.resign(messageInfo, check_sign=False)
def resign(self, messageInfo, check_sign=False):
[注意]传递专业知识、拓宽行业人脉——看雪讲师团队等你加入！
