1
# WAAP 规则配置
2
rules:
3
  # 正则规则
4
  - id: R1001
5
    name: "SQL 注入检测"
6
    type: regex
7
    priority: 0
8
    enabled: true
9
    conditions:
10
      - field: request.params
11
        operator: regex
12
        pattern: "(?i)(union\\s+select|or\\s+1\\s*=\\s*1|drop\\s+table)"
13
        transform:
14
          - url_decode
15
          - html_decode
16
          - base64_decode
17
    action: block
18
    severity: critical
19
    tags: [sqli, injection]
20

21
  # 语义分析规则
22
  - id: R2001
23
    name: "编码绕过检测"
24
    type: semantic
25
    priority: 1
26
    enabled: true
27
    conditions:
28
      - field: request.body
29
        operator: semantic_match
30
        attack_type: sqli
31
        decode_chain:
32
          - url_decode
33
          - unicode_decode
34
          - html_entity_decode
35
    action: block
36
    severity: high
37

38
  # 协议合规规则
39
  - id: R3001
40
    name: "HTTP 请求走私防护"
41
    type: protocol
42
    priority: 2
43
    enabled: true
44
    conditions:
45
      - field: request.headers.content_length
46
        operator: mismatch
47
        reference: request.body.actual_length
48
      - field: request.headers.transfer_encoding
49
        operator: contains
50
        pattern: "chunked"
51
    action: block
52
    severity: critical
53

54
  # 自定义规则
55
  - id: R4001
56
    name: "管理接口访问控制"
57
    type: custom
58
    priority: 3
59
    enabled: true
60
    conditions:
61
      - field: request.uri
62
        operator: prefix
63
        pattern: "/admin"
64
      - field: request.headers.x_forwarded_for
65
        operator: not_in
66
        values: ["10.0.0.0/8", "172.16.0.0/12"]
67
    action: block
68
    severity: medium

1
// RuleEngine 规则引擎核心逻辑
2
type RuleEngine struct {
3
    rules    []Rule
4
    matcher  Matcher
5
    updater  HotUpdater
6
}
7

8
type MatchResult struct {
9
    Hit      bool
10
    RuleID   string
11
    Action   string
12
    Severity string
13
}
14

15
// Match 按优先级依次匹配规则
16
func (e *RuleEngine) Match(ctx *RequestContext) *MatchResult {
17
    for _, rule := range e.rules {
18
        if !rule.Enabled {
19
            continue
20
        }
21

22
        hit := true
23
        for _, cond := range rule.Conditions {
24
            value := ctx.GetField(cond.Field, cond.Transform...)
25
            if !e.matcher.Match(value, cond) {
26
                hit = false
27
                break
28
            }
29
        }
30

31
        if hit {
32
            return &MatchResult{
33
                Hit:      true,
34
                RuleID:   rule.ID,
35
                Action:   rule.Action,
36
                Severity: rule.Severity,
37
            }
38
        }
39
    }
40

41
    return &MatchResult{Hit: false}
42
}
43

44
// HotUpdate 规则热更新
45
func (e *RuleEngine) HotUpdate(newRules []Rule) error {
46
    // 双缓冲切换：先加载新规则到备用区，再原子切换指针
47
    backup := make([]Rule, len(e.rules))
48
    copy(backup, e.rules)
49

50
    e.rules = newRules
51
    // 切换失败则回滚
52
    if err := e.validate(); err != nil {
53
        e.rules = backup
54
        return fmt.Errorf("hot update failed: %w", err)
55
    }
56
    return nil
57
}

1
# WAAP AI 检测特征提取
2
import math
3
from collections import Counter
4

5
def shannon_entropy(text: str) -> float:
6
    """计算字符串的 Shannon 熵，用于检测编码/混淆攻击"""
7
    if not text:
8
        return 0.0
9
    freq = Counter(text)
10
    length = len(text)
11
    entropy = -sum(
12
        (count / length) * math.log2(count / length)
13
        for count in freq.values()
14
    )
15
    return entropy
16

17
def extract_request_features(request: dict) -> dict:
18
    """从请求中提取 AI 模型所需的特征向量"""
19
    features = {
20
        # 请求频率特征
21
        "qps": request.get("qps", 0),
22
        "burst_ratio": request.get("burst_ratio", 0),
23
        "time_window_variance": request.get("time_var", 0),
24

25
        # URL 分布特征
26
        "path_depth": request["uri"].count("/"),
27
        "param_count": len(request.get("params", {})),
28
        "path_entropy": shannon_entropy(request["uri"]),
29

30
        # 参数熵值特征
31
        "param_entropy": max(
32
            shannon_entropy(v) for v in request.get("params", {}).values()
33
        ) if request.get("params") else 0,
34

35
        # User-Agent 特征
36
        "ua_length": len(request.get("user_agent", "")),
37
        "ua_known_tool": _check_tool_signature(request.get("user_agent", "")),
38

39
        # 行为序列特征
40
        "request_interval_mean": request.get("interval_mean", 0),
41
        "request_interval_std": request.get("interval_std", 0),
42
    }
43

44
    return features
45

46
def _check_tool_signature(ua: str) -> bool:
47
    """检测 User-Agent 是否包含已知自动化工具指纹"""
48
    signatures = ["python-requests", "curl/", "httpclient", "scanner"]
49
    return any(sig in ua.lower() for sig in signatures)

1
# API Schema 校验配置
2
api_schema_protection:
3
  # OpenAPI 规范路径
4
  spec_url: /api/openapi.json
5

6
  # 校验策略
7
  validation:
8
    # 请求校验
9
    request:
10
      - check_path: true          # 路径必须匹配 API 定义
11
      - check_method: true        # HTTP 方法必须匹配
12
      - check_content_type: true  # Content-Type 必须匹配
13
      - check_body_schema: true   # 请求体必须符合 Schema
14
      - check_required_fields: true  # 必填字段不可缺失
15

16
    # 响应校验（防止数据泄露）
17
    response:
18
      - check_status_code: true
19
      - check_body_schema: true
20
      - check_sensitive_fields: true
21

22
  # 异常检测
23
  anomaly_detection:
24
    # 未定义路径访问
25
    undefined_path:
26
      action: block
27
      log: true
28

29
    # 参数类型不匹配
30
    type_mismatch:
31
      action: block
32
      log: true
33

34
    # 额外参数注入
35
    extra_parameters:
36
      action: strip_and_log

1
# WAAP 部署最佳实践 checklist
2
deployment_best_practices:
3
  # 第一阶段：观察模式（1-2 周）
4
  phase1_observe:
5
    - name: "全量规则设为观察模式"
6
      action: detect_only
7
      duration: 14d
8
    - name: "收集误报数据"
9
      action: log_false_positives
10
    - name: "建立流量基线"
11
      action: baseline_learning
12
    - name: "梳理业务 API 清单"
13
      action: api_inventory
14

15
  # 第二阶段：调优阶段（2-4 周）
16
  phase2_tune:
17
    - name: "处理误报 Top 20 规则"
18
      action: adjust_threshold
19
    - name: "添加业务白名单"
20
      action: whitelist
21
    - name: "核心规则切换为阻断"
22
      action: block_critical
23
    - name: "验证业务功能无损"
24
      action: regression_test
25

26
  # 第三阶段：全面防护
27
  phase3_protect:
28
    - name: "全部规则切换为阻断"
29
      action: block_all
30
    - name: "开启 AI 检测"
31
      action: enable_ai
32
    - name: "配置 Bot 管理策略"
33
      action: bot_management
34
    - name: "启用 API Schema 校验"
35
      action: schema_validation
36

37
  # 持续运维
38
  ongoing:
39
    - name: "每周审查误报日志"
40
      frequency: weekly
41
    - name: "每月更新规则库"
42
      frequency: monthly
43
    - name: "每季度评估 AI 模型效果"
44
      frequency: quarterly
45
    - name: "重大活动前压力测试"
46
      frequency: ad_hoc

1
# 白名单管理规范
2
whitelist_management:
3
  # 白名单审批流程
4
  approval:
5
    - security_team_review: required
6
    - business_owner_ack: required
7
    - expiration_date: mandatory  # 所有白名单必须设过期时间
8

9
  # 白名单最小化原则
10
  principles:
11
    - "优先使用路径 + 参数白名单，而非全量放行"
12
    - "IP 白名单仅限内网段，禁止公网 IP"
13
    - "规则白名单需标注关联的业务场景"
14
    - "过期白名单自动失效，需重新审批"
15

16
  # 白名单审计
17
  audit:
18
    - frequency: monthly
19
    - check_expired: true
20
    - check_orphaned: true  # 关联业务已下线的白名单

1
# WAAP 性能优化配置
2
performance:
3
  # 规则引擎优化
4
  rule_engine:
5
    - skip_rules_for_static_resources: true  # 静态资源跳过规则检测
6
    - cache_regex_compilation: true          # 正则预编译缓存
7
    - early_termination: true                # 命中即返回
8

9
  # AI 检测优化
10
  ai_detection:
11
    - sample_rate: 0.1          # 正常流量 10% 采样
12
    - anomaly_boost: 1.0        # 异常流量全量检测
13
    - model_cache_ttl: 300s     # 模型推理结果缓存
14

15
  # 连接管理
16
  connection:
17
    - keepalive_timeout: 75s
18
    - max_connections: 65535
19
    - upstream_timeout: 30s