17｜自动化注入神器（四）：sqlmap的核心功能解析

你好，我是王昊天。

在上节课中，我们重点学习了sqlmap中一个非常重要的算法——页面相似度算法。相信你对页面相似度这个概念会有更加清晰的认知，不但知道它是什么含义，而且知道它是如何计算出来的。解决了这个大难点之后，我在上节课的结尾提出了一个空连接检测功能，有了它，sqlmap就可以大大提高执行效率。完成了检测，sqlmap就进入到实际的SQL注入测试阶段了。

在SQL注入测试阶段，系统首先会检测有哪些注入点，然后对这些注入点逐一发送合适的payload，检测注入是否成功。如果注入成功，那么系统会将注入点存储下来，最后对它们进行输出。

这节课，我们就来正式学习sqlmap的SQL注入测试过程。

注入点检测

在SQL正式注入测试之前，sqlmap会对每个目标的参数进行过滤。将那些非动态的，不存在注入可能的参数剔除掉，留下可能的注入点。这样sqlmap仅需要对这些可能的注入点进行正式的注入测试即可。

动态参数检测

我们首先来看sqlmap是如何检测动态参数的。这部分代码依旧在start函数中，紧接着空连接检测出现。

# sqlmap首先对所有可用于注入测试的参数进行简单的优先级排序。
       parameters = list(conf.parameters.keys())
    # 定义测试列表的顺序。（从后到前）
       orderList = (PLACE.CUSTOM_POST, PLACE.CUSTOM_HEADER, PLACE.URI, PLACE.POST, PLACE.GET)
    # 对测试参数排好序之后，系统开始对参数进行过滤操作。
       proceed = True
       for place in parameters:
           skip = # ...
           if skip:
               continue
           if place not in conf.paramDict:
               continue
           paramDict = conf.paramDict[place]
           paramType = conf.method if conf.method not in (None, HTTPMETHOD.GET, HTTPMETHOD.POST) else place
    # ...
           for parameter, value in paramDict.items():
               if not proceed:
                   break
    # 经过过滤，将该参数加入到测试过的参数中，防止重复测试。
               kb.testedParams.add(paramKey)

我们可以结合代码中的注释，来理解参数的过滤。首先sqlmap会对待测参数进行一个优先级排序。在排序完成之后，系统会根据用户的配置信息，对这些参数进行过滤操作。这里我举一个例子来让你更加容易理解这一步骤。例如，当用户配置的检测level小于2时，那么系统就会跳过对cookie参数的检测过程。

过滤完成之后，我们就会进入到你最熟悉的一步——SQL注入测试过程。让我们结合代码，分析sqlmap是如何进行SQL注入测试的。

if testSqlInj:
    # 开始注入测试
        try:
    # ...
    # 进入启发式注入测试。
            check = heuristicCheckSqlInjection(place, parameter)
    # 当启发式注入测试失败，就跳过该参数。
            if check != HEURISTIC_TEST.POSITIVE:
                if conf.smart or (kb.ignoreCasted and check == HEURISTIC_TEST.CASTED):
    # ... 
                   continue
    # ...
    # 通过启发式注入测试后，就会进入到SQL注入测试阶段。
            injection = checkSqlInjection(place, parameter, value)

启发式注入测试

如果一个参数被检测为注入点，那我们就可以对它进行注入测试。为了提高注入测试的效率，系统会过滤一些注入成功率较低的注入点，这需要首先对它进行一个启发式注入测试。下面让我们结合代码，对启发式注入测试有个更具体的理解。

def heuristicCheckSqlInjection(place, parameter):
    
    # 如果配置中设置了跳过启发式注入测试，就返回结果None，当使用者没有特殊配置conf.start这个配置项为false，就会跳过该参数的注入检测。
        if conf.skipHeuristics:
            return None
    
    # 初始化参数，并根据用户设置的偏好制作payload。
        origValue = conf.paramDict[place][parameter]
        paramType = conf.method if conf.method not in (None, HTTPMETHOD.GET, HTTPMETHOD.POST) else place
    
        prefix = ""
        suffix = ""
        randStr = ""
    
        if conf.prefix or conf.suffix:
            if conf.prefix:
                prefix = conf.prefix
    
            if conf.suffix:
                suffix = conf.suffix
    
        while randStr.count('\'') != 1 or randStr.count('\"') != 1:
            randStr = randomStr(length=10, alphabet=HEURISTIC_CHECK_ALPHABET)
    
        kb.heuristicMode = True
    
        payload = "%s%s%s" % (prefix, randStr, suffix)
        payload = agent.payload(place, parameter, newValue=payload)
    
    # 利用payload 请求目标页面的响应内容。
        page, _, _ = Request.queryPage(payload, place, content=True, raise404=False)
    
        kb.heuristicPage = page
        kb.heuristicMode = False

系统首先会判断，用户是否设置跳过启发式注入测试，如果设置了，则返回None。如果没有设置，那么系统就会获取到用户设置的偏好prefix以及suffix，然后据此构造出合适的payload，并发送给目标，获取到响应内容page。

# 检测请求目标的响应中是否有数据库错误。
       parseFilePaths(page)
       result = wasLastResponseDBMSError()
       infoMsg = "heuristic (basic) test shows that %sparameter '%s' might " % ("%s " % paramType if paramType != parameter else "", parameter)
    # 检测page中是否有。
       def _(page):
           return any(_ in (page or "") for _ in FORMAT_EXCEPTION_STRINGS)
       casting = _(page) and not _(kb.originalPage)

系统会根据获取到的内容，判断其中的报错信息。其中，如果为数据库报错信息，那么result的值为True。如果是设置在

sqlmap/lib/core/settings.py文件中FORMAT_EXCEPTION_SRTINGS配置项中定义的类型转化错误信息，那么就会用casting来储存错误内容。

#     ...
    # 当存在定义的问题时，发出报错信息。
        if casting:
            errMsg = "possible %s casting detected (e.g. '" % ("integer" if origValue.isdigit() else "type")
    
            platform = conf.url.split('.')[-1].lower()
            if platform == WEB_PLATFORM.ASP:
                errMsg += "%s=CInt(request.querystring(\"%s\"))" % (parameter, parameter)
            elif platform == WEB_PLATFORM.ASPX:
                errMsg += "int.TryParse(Request.QueryString[\"%s\"], out %s)" % (parameter, parameter)
            elif platform == WEB_PLATFORM.JSP:
                errMsg += "%s=Integer.parseInt(request.getParameter(\"%s\"))" % (parameter, parameter)
            else:
                errMsg += "$%s=intval($_REQUEST[\"%s\"])" % (parameter, parameter)
    
            errMsg += "') at the back-end web application"
            logger.error(errMsg)
    
            if kb.ignoreCasted is None:
                message = "do you want to skip those kind of cases (and save scanning time)? %s " % ("[Y/n]" if conf.multipleTargets else "[y/N]")
                kb.ignoreCasted = readInput(message, default='Y' if conf.multipleTargets else 'N', boolean=True)
    
    # 当数据库报错时，判断出注入漏洞很可能存在。
        elif result:
            infoMsg += "be injectable"
            if Backend.getErrorParsedDBMSes():
                infoMsg += " (possible DBMS: '%s')" % Format.getErrorParsedDBMSes()
            logger.info(infoMsg)
    
    # 否则判定为不存在注入漏洞。
        else:
            infoMsg += "not be injectable"
            logger.warn(infoMsg)
    
        kb.heuristicMode = True
        kb.disableHtmlDecoding = True

最后，函数会根据casting以及result中的内容进行输出。我在这里画了一个它的流程图，帮助你对它的作用进行理解。

图中启发式注入结果分为三种，其中阳性代表该参数大概率可以注入，类型转换和阴性都代表了该参数大概率不可以注入。我们会发现，想要判断是否可以注入，只需要判断有无数据库报错信息就可以了，有的话就认为该参数可注入，否则就认为不可注入。

除了进行启发式SQL注入检测之外，sqlmap还会做一些不属于它的工作，包括进行简单的xss检测和文件包含检测。

# 更换payload，检测xss以及文件包含。
    randStr1, randStr2 = randomStr(NON_SQLI_CHECK_PREFIX_SUFFIX_LENGTH), randomStr(NON_SQLI_CHECK_PREFIX_SUFFIX_LENGTH)
    value = "%s%s%s" % (randStr1, DUMMY_NON_SQLI_CHECK_APPENDIX, randStr2)
    payload = "%s%s%s" % (prefix, "'%s" % value, suffix)
    payload = agent.payload(place, parameter, newValue=payload)
    page, _, _ = Request.queryPage(payload, place, content=True, raise404=False)
    
    paramType = conf.method if conf.method not in (None, HTTPMETHOD.GET, HTTPMETHOD.POST) else place
    
    # 进行xss检测。
    if value.upper() in (page or "").upper():
        infoMsg = "heuristic (XSS) test shows that %sparameter '%s' might be vulnerable to cross-site scripting (XSS) attacks" % ("%s " % paramType if paramType != parameter else "", parameter)
        logger.info(infoMsg)
    
        if conf.beep:
            beep()
    
    # 进行文件包含检测。
    for match in re.finditer(FI_ERROR_REGEX, page or ""):
        if randStr1.lower() in match.group(0).lower():
            infoMsg = "heuristic (FI) test shows that %sparameter '%s' might be vulnerable to file inclusion (FI) attacks" % ("%s " % paramType if paramType != parameter else "", parameter)
            logger.info(infoMsg)
    
            if conf.beep:
                beep()
    
            break
    
    kb.disableHtmlDecoding = False
    kb.heuristicMode = False
    
    return kb.heuristicTest

最终的检测结果都会在全局变量kb中保存起来，这个全局变量我们在之前的课程中学习过。到此，启发式注入检测的函数已经完成，接下来会进入真正的SQL注入检测，这是sqlmap最核心的功能，没有之一！

checkSqlInjection函数

sqlmap对启发式注入的检测结果进行简单地判断后，程序就会进入sqlmap最核心的函数checkSqlInjection中。这个函数用于实现注入检测的核心功能，包括布尔注入、联合注入、报错注入、堆注入等检测。

下面让我们观察它的代码来理解这个注入检测功能。

def checkSqlInjection(place, parameter, value):
    
    # 根据参数的类型选择 boundary 。
        injection = InjectionDict()
    
        threadData = getCurrentThreadData()
    
        if isDigit(value):
            kb.cache.intBoundaries = kb.cache.intBoundaries or sorted(copy.deepcopy(conf.boundaries), key=lambda boundary: any(_ in (boundary.prefix or "") or _ in (boundary.suffix or "") for _ in ('"', '\'')))
            boundaries = kb.cache.intBoundaries
        elif value.isalpha():
            kb.cache.alphaBoundaries = kb.cache.alphaBoundaries or sorted(copy.deepcopy(conf.boundaries), key=lambda boundary: not any(_ in (boundary.prefix or "") or _ in (boundary.suffix or "") for _ in ('"', '\'')))
            boundaries = kb.cache.alphaBoundaries
        else:
            boundaries = conf.boundaries

这个函数首先会判断参数的类型，然后根据参数的不同类型设置合适的闭合方式。解决完寻找注入点以及闭合参数这个问题后，下面让我们进入到payload的选择中。

我们知道，payload的选择和数据库的类型有很大的关系，所以sqlmap在构造payload前，会先尝试探测目标数据库的类型。

# 判断是否配置数据库类型。
    if conf.dbms is None:
    
    # 探测目标数据库类型。
        if not injection.dbms and PAYLOAD.TECHNIQUE.BOOLEAN in injection.data:
            if not Backend.getIdentifiedDbms() and kb.heuristicDbms is None and not kb.droppingRequests:
                kb.heuristicDbms = heuristicCheckDbms(injection)
    
    # 根据探测结果输出提示信息。
        if kb.reduceTests is None and not conf.testFilter and (intersect(Backend.getErrorParsedDBMSes(), SUPPORTED_DBMS, True) or kb.heuristicDbms or injection.dbms):
            msg = "it looks like the back-end DBMS is '%s'. " % (Format.getErrorParsedDBMSes() or kb.heuristicDbms or joinValue(injection.dbms, '/'))
            msg += "Do you want to skip test payloads specific for other DBMSes? [Y/n]"
            kb.reduceTests = (Backend.getErrorParsedDBMSes() or [kb.heuristicDbms]) if readInput(msg, default='Y', boolean=True) else []

如果用户在配置中指定了目标数据库的类型，那么就无需探测，用指定类型即可。否则需要用heuristicCheckDbms(injection)函数来判断目标数据库类型。它的判断方法是，发送一些payload给测试目标，然后根据获得的响应判断数据库的类型。

判断出目标数据库的类型之后，系统会根据获得的数据库类型以及用户的配置，挑选适合的测试用例，然后根据这些测试用例以及之前配置的boundary，构造适合的payload。

# 配置联合查询的信息。
    if stype == PAYLOAD.TECHNIQUE.UNION:
        configUnion(test.request.char)
    
        if "[CHAR]" in title:
            if conf.uChar is None:
                continue
            else:
                title = title.replace("[CHAR]", conf.uChar)
    # ...
    # 用户指定了测试方法的配置。
    if conf.technique and isinstance(conf.technique, list) and stype not in conf.technique:
        debugMsg = "skipping test '%s' because user " % title
        debugMsg += "specified testing of only "
        debugMsg += "%s techniques" % " & ".join(PAYLOAD.SQLINJECTION[_] for _ in conf.technique)
        logger.debug(debugMsg)
        continue
    
    # ...
    # 根据指定的数据库以及用户的配置信息，对payload进行筛选。
    if conf.technique and isinstance(conf.technique, list) and stype not in conf.technique:
        debugMsg = "skipping test '%s' because user " % title
        debugMsg += "specified testing of only "
        debugMsg += "%s techniques" % " & ".join(PAYLOAD.SQLINJECTION[_] for _ in conf.technique)
        logger.debug(debugMsg)
        continue
    
    # ...
    # 对payload去重。
    if fstPayload:
        boundPayload = agent.prefixQuery(fstPayload, prefix, where, clause)
        boundPayload = agent.suffixQuery(boundPayload, comment, suffix, where)
        reqPayload = agent.payload(place, parameter, newValue=boundPayload, where=where)

sqlmap准备完payload之后，就到了你最期待的注入测试环节，这个过程和我们手动测试类似，系统会使用不同的注入测试方法，包括布尔注入、报错注入、时延注入以及联合注入。

# 布尔注入
    if method == PAYLOAD.METHOD.COMPARISON:
        def genCmpPayload():
            sndPayload = agent.cleanupPayload(test.response.comparison, origValue=value if place not in (PLACE.URI, PLACE.CUSTOM_POST, PLACE.CUSTOM_HEADER) and BOUNDED_INJECTION_MARKER not in (value or "") else None)
    
    # ...
    # 报错注入
    elif method == PAYLOAD.METHOD.GREP:
        try:
            page, headers, _ = Request.queryPage(reqPayload, place, content=True, raise404=False)
            output = extractRegexResult(check, page, re.DOTALL | re.IGNORECASE)
            output = output or extractRegexResult(check, threadData.lastHTTPError[2] if wasLastResponseHTTPError() else None, re.DOTALL | re.IGNORECASE)
    
    # ...
    # 时延注入
    elif method == PAYLOAD.METHOD.TIME:
        trueResult = Request.queryPage(reqPayload, place, timeBasedCompare=True, raise404=False)
        trueCode = threadData.lastCode
    
    # ...
    # 联合注入
    elif method == PAYLOAD.METHOD.UNION:
        configUnion(test.request.char, test.request.columns)

做完这些注入测试后，系统会收到响应。我们平时会通过观察响应来判断注入是否成功，但是系统要如何判断呢？聪明的你或许想到了，这就是之前我们学习的页面相似度，我们在学习sqlmap判断waf时就用到了它。其实，根据注入方式的不同，sqlmap对于注入结果的判断方式也是不同的。

在报错注入中，系统会通过对页面的响应结果进行正则匹配，判断响应中是否有报错信息，如果有就判断注入成功，否则判断注入失败。

# 报错注入判断注入是否成功。
    page, headers, _ = Request.queryPage(reqPayload, place, content=True, raise404=False)
    output = extractRegexResult(check, page, re.DOTALL | re.IGNORECASE)
    output = output or extractRegexResult(check, threadData.lastHTTPError[2] if wasLastResponseHTTPError() else None, re.DOTALL | re.IGNORECASE)
    # ...
    injectable = True

在布尔注入中，系统会判断返回页面的相似度，如果结果为假，那么说明系统会根据错误结果进行不同的响应，这就意味着布尔注入是成功的。

falseResult = Request.queryPage(genCmpPayload(), place, raise404=False)
    
    if not falseResult:
        # ...
    
        injectable = True

在时延注入中，sqlmap会发送sleep([random])的请求，判断请求时间是否大于“平均时间+7*标准差”，注意这里的标准差是一个时间阈值，如果大于就认为存在时延注入。

if trueResult:
    
        if SLEEP_TIME_MARKER in reqPayload:
            falseResult = Request.queryPage(reqPayload.replace(SLEEP_TIME_MARKER, "0"), place, timeBasedCompare=True, raise404=False)
            if falseResult:
                continue
    
    # ...   
            injectable = True

在联合注入中，系统会通过unionTest函数来判断联合注入是否存在。它的实现原理比较复杂，我们可以将它简化一下，只需要比较联合注入得到的响应和原本内容是否一致，就可以做出判断，如果不一致，则说明存在联合注入问题。

reqPayload, vector = unionTest(comment, place, parameter, value, prefix, suffix)
    
    if isinstance(reqPayload, six.string_types):
        infoMsg = "%sparameter '%s' is '%s' injectable" % ("%s " % paramType if paramType != parameter else "", parameter, title)
        logger.info(infoMsg)
    
        injectable = True

最后系统将结果记录下来，并且输出给使用者，这就是我们在使用sqlmap时看到的结果信息。

至此，经过四讲的学习，我们终于学完了这款自动化注入测试神器，希望你可以了解sqlmap的底层原理，从而更好的使用这款工具。

总结

在这节课里，我们深入研究了sqlmap的真正SQL注入过程。为了你能更好的理解，我们主要通过观察它的源代码对它进行学习。

在这个过程中，我们首先学习了sqlmap对于注入点的检测，其中包括了动态参数的检测以及启发式注入测试。在实际注入测试的过程中，我们只会对通过检测的参数进行注入的探测。通过这个过程筛选参数，可以提高sqlmap的运行效率。

最后我们进入到最重要的一步中，即真正的注入测试，我们了解了它的测试过程。其中有payload的配置、对目标数据库信息的探测、筛选合适的payload以及实际的注入测试过程。完成测试，系统会根据页面相似度来判断注入结果，而对于不同的注入方式，sqlmap的判断方式也是不同的。我们将联合注入、报错注入、时延注入以及布尔注入的判断方法一一展开，对它们分别进行了介绍。

截止到目前，你已经完成了对SQL注入原理、攻击方式、防御方案以及自动化注入工具sqlmap的学习，结合对sqlmap原理的学习，快去自己尝试一下自动化注入的威力吧！

思考

sqlmap在实现中有什么值得改进的地方吗？

欢迎在评论区留下你的思考，我们下节课再见。

大师兄

17｜自动化注入神器（四）：sqlmap的核心功能解析

注入点检测

动态参数检测

启发式注入测试

checkSqlInjection函数

总结

思考