综合实用脚本，防守漏洞怎么识别定位？

wen 实用脚本 2026-11-13 1

防守漏洞识别与定位

防守漏洞识别是安全评估、红蓝对抗、渗透测试中的核心能力，下面给出一套分层、可落地的实用脚本思路与代码框架,覆盖从信息收集到漏洞定位的完整链路。

综合实用脚本，防守漏洞怎么识别定位？

漏洞识别的方法论（先想清楚再写脚本）

防守漏洞通常分四类,识别方式不同：

漏洞类型	典型场景	识别手段
配置类	弱口令、默认配置、开放端口、暴露服务	端口扫描 + 指纹识别 + 配置核查
代码/应用类	SQL注入、XSS、命令执行、文件上传	主动扫描 + 流量分析 + 代码审计
权限类	越权、未授权访问、IDOR	多角色对比测试 + 接口枚举
架构/边界类	内网横向、边界突破、供应链	资产测绘 + 攻击面分析

定位的关键：漏洞不是"扫出来"的，是"对比出来"的——对比预期行为 vs 实际行为、对比正常请求 vs 异常请求。

综合脚本框架（Python，模块化）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
综合防守漏洞识别脚本框架
用法: python vuln_scan.py -t target.com --mode full
"""
import argparse, socket, requests, json, re, sys
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urljoin
requests.packages.urllib3.disable_warnings()
class VulnScanner:
    def __init__(self, target, threads=20):
        self.target = target
        self.threads = threads
        self.findings = []   # 漏洞结果，含定位信息
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": "Mozilla/5.0 (SecScan)"})
        self.session.verify = False
    # ---------- 1. 端口与服务识别 ----------
    def port_scan(self, ports=(21,22,23,80,443,3306,6379,8080,9200,27017)):
        open_ports = []
        for p in ports:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(1.5)
            try:
                if s.connect_ex((self.target, p)) == 0:
                    open_ports.append(p)
            finally:
                s.close()
        for p in open_ports:
            self.findings.append({
                "type": "open_port", "severity": "info",
                "location": f"{self.target}:{p}",
                "desc": f"端口 {p} 开放"
            })
        return open_ports
    # ---------- 2. Web 指纹与暴露面 ----------
    def web_probe(self, base_url):
        paths = ["/", "/admin", "/.git/config", "/.env", "/phpinfo.php",
                 "/actuator/env", "/swagger-ui.html", "/api-docs"]
        for path in paths:
            url = urljoin(base_url, path)
            try:
                r = self.session.get(url, timeout=5)
                # 关键：用"特征对比"定位漏洞，而非盲目匹配
                if path == "/.git/config" and "repositoryformatversion" in r.text:
                    self._add("exposed_git", "high", url, "Git 源码泄露")
                elif path == "/.env" and re.search(r"(DB_|SECRET|KEY)", r.text):
                    self._add("exposed_env", "critical", url, "环境变量泄露")
                elif path == "/actuator/env" and "spring" in r.text.lower():
                    self._add("actuator_exposed", "high", url, "Spring Actuator 未授权")
                elif r.status_code == 200 and "phpinfo" in r.text.lower():
                    self._add("phpinfo", "medium", url, "phpinfo 暴露")
            except requests.RequestException:
                continue
    # ---------- 3. 弱口令探测（示例：HTTP Basic/表单） ----------
    def weak_auth_check(self, url, users=("admin","test"), pwds=("admin","123456","password")):
        for u in users:
            for p in pwds:
                try:
                    r = self.session.get(url, auth=(u, p), timeout=5)
                    if r.status_code == 200 and "login" not in r.text.lower():
                        self._add("weak_credential", "critical", url,
                                  f"弱口令 {u}:{p}")
                        return
                except requests.RequestException:
                    continue
    # ---------- 4. 未授权/越权检测（对比法） ----------
    def authz_check(self, url):
        """核心：携带/不携带凭证对比响应差异"""
        try:
            r_noauth = self.session.get(url, timeout=5)
            r_auth = self.session.get(url, headers={"Authorization": "Bearer test"})
            # 无凭证却返回敏感数据 → 未授权
            if r_noauth.status_code == 200 and len(r_noauth.text) > 200:
                if any(k in r_noauth.text.lower() for k in ("token","user","email")):
                    self._add("unauthorized_access", "high", url,
                              "接口未授权即可访问敏感数据")
        except requests.RequestException:
            pass
    # ---------- 5. 常见注入点探测（参数回显对比） ----------
    def injection_probe(self, url, param="id"):
        payloads = {
            "sqli": "' OR '1'='1",
            "xss": "<script>alert(1)</script>",
        }
        base = self.session.get(url, timeout=5).text
        for name, payload in payloads.items():
            try:
                r = self.session.get(url, params={param: payload}, timeout=5)
                if payload in r.text and payload not in base:
                    self._add(f"{name}_reflected", "medium", url,
                              f"参数 {param} 存在 {name} 反射")
                if name == "sqli" and re.search(r"SQL syntax|mysql_fetch", r.text, re.I):
                    self._add("sqli_error", "high", url, "SQL 报错注入特征")
            except requests.RequestException:
                continue
    # ---------- 工具方法 ----------
    def _add(self, vtype, sev, loc, desc):
        self.findings.append({
            "type": vtype, "severity": sev,
            "location": loc, "desc": desc
        })
        print(f"[{sev.upper():8}] {vtype:20} @ {loc}  -> {desc}")
    def run(self, base_url):
        print(f"[*] 目标: {self.target}")
        print("[*] 阶段1: 端口扫描")
        self.port_scan()
        print("[*] 阶段2: Web 暴露面")
        self.web_probe(base_url)
        print("[*] 阶段3: 认证与授权")
        self.weak_auth_check(base_url + "/admin")
        self.authz_check(base_url + "/api/user")
        print("[*] 阶段4: 参数注入")
        self.injection_probe(base_url + "/product")
        return self.report()
    def report(self):
        summary = {}
        for f in self.findings:
            summary[f["severity"]] = summary.get(f["severity"], 0) + 1
        result = {"target": self.target, "summary": summary, "findings": self.findings}
        with open("vuln_report.json", "w", encoding="utf-8") as fp:
            json.dump(result, fp, ensure_ascii=False, indent=2)
        print(f"\n[+] 结果已保存 vuln_report.json  统计: {summary}")
        return result
if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("-t", "--target", required=True)
    ap.add_argument("--url", default=None)
    args = ap.parse_args()
    base = args.url or f"http://{args.target}"
    VulnScanner(args.target).run(base)

"识别定位"的三个关键技巧

差异对比定位（最核心）

漏洞 = 预期行为与实际行为的偏差,脚本设计上要处处做对比：

带凭证 vs 不带凭证 → 定位越权
正常参数 vs 恶意参数 → 定位注入
存在 vs 不存在路径 → 定位敏感文件
状态码 200 vs 403/302 → 判断访问控制

特征指纹库（定位是什么漏洞）

SIGNATURES = {
    "sqli":   [r"SQL syntax.*MySQL", r"ORA-\d{5}", r"Microsoft.*ODBC"],
    "lfi":    [r"root:x:0:0", r"\[boot loader\]"],
    "rce":    [r"uid=\d+\(.*\)", r"Windows IP Configuration"],
    "xss":    [r"<script>alert\(1\)</script>"],
}
# 命中即定位："在哪个URL、哪个参数、什么类型漏洞"

攻击链串联定位（从点到面）

单个开放端口不是闭环漏洞，要串联成攻击路径：

开放 6379(Redis) + 未授权 → 写 SSH key / 计划任务 → 定位到"主机沦陷路径"

脚本里可以加一层关联分析：

def correlate(findings):
    ports = {f["location"] for f in findings if f["type"]=="open_port"}
    risks = []
    if any("6379" in p for p in ports) and any(f["type"]=="redis_unauth" for f in findings):
        risks.append("Redis 未授权可导致 RCE（高危攻击链）")
    return risks