让 WAF 测试学会“看反馈”：用自适应循环发现固定规则漏掉的检测缺口

python -m venv .venv
. .venv/bin/activate
pip install requests
export TARGET_URL='https://staging.example.internal/waf-test'
export WAF_TEST_ACK='authorized-staging-only'
python waf_feedback_test.py

import json
import os
import time
from urllib.parse import quote

import requests

TARGET_URL = os.environ.get('TARGET_URL', '')
ACK = os.environ.get('WAF_TEST_ACK', '')

if ACK != 'authorized-staging-only':
    raise SystemExit('Refusing to run: set WAF_TEST_ACK for an authorized staging target')
if not TARGET_URL.startswith('https://'):
    raise SystemExit('TARGET_URL must be an HTTPS staging endpoint')

# 这些是测试标记。测试接口不得解释、执行、渲染或访问其中引用的资源。
PROBES = {
    'sqli': "' OR '1'='1 -- WAF_CANARY",
    'xss': '<script>console.log(\'WAF_CANARY\')</script>',
    'path_traversal': '../../waf-canary.txt',
    'command_injection': '; echo WAF_CANARY',
    'ssrf': 'https://waf-canary.invalid/check',
    'template_injection': '{{ 7 * 7 }} WAF_CANARY',
}

BLOCK_CODES = {403, 406, 429}
TIMEOUT_SECONDS = 5
MAX_REQUESTS = 18


def send_probe(category, value, profile):
    headers = {
        'User-Agent': 'authorized-waf-regression-test/1.0',
        'X-Security-Test': 'waf-feedback-loop',
    }

    if profile == 'json':
        response = requests.post(
            TARGET_URL,
            json={'category': category, 'value': value},
            headers=headers,
            timeout=TIMEOUT_SECONDS,
            allow_redirects=False,
        )
    elif profile == 'form':
        response = requests.post(
            TARGET_URL,
            data={'category': category, 'value': value},
            headers=headers,
            timeout=TIMEOUT_SECONDS,
            allow_redirects=False,
        )
    elif profile == 'json_percent_encoded':
        response = requests.post(
            TARGET_URL,
            json={'category': category, 'value': quote(value, safe='')},
            headers=headers,
            timeout=TIMEOUT_SECONDS,
            allow_redirects=False,
        )
    else:
        raise ValueError(f'Unknown profile: {profile}')

    action = response.headers.get('X-WAF-Action', '').lower()
    blocked = response.status_code in BLOCK_CODES or action == 'block'
    return {
        'category': category,
        'profile': profile,
        'status': response.status_code,
        'blocked': blocked,
        'waf_action': action or None,
    }


def main():
    request_count = 0
    results = []

    for category, seed in PROBES.items():
        if request_count >= MAX_REQUESTS:
            break

        first = send_probe(category, seed, 'json')
        request_count += 1
        results.append(first)
        print(json.dumps(first, ensure_ascii=False))

        # 根据反馈选择下一条请求，而不是无条件跑完所有组合。
        # 被拦截时检查规范化一致性；被放行时切换解析路径以确认缺口范围。
        next_profile = 'json_percent_encoded' if first['blocked'] else 'form'

        if request_count < MAX_REQUESTS:
            second = send_probe(category, seed, next_profile)
            request_count += 1
            results.append(second)
            print(json.dumps(second, ensure_ascii=False))

        time.sleep(0.25)

    with open('waf-results.json', 'w', encoding='utf-8') as output:
        json.dump(results, output, ensure_ascii=False, indent=2)

    passed = [item for item in results if not item['blocked']]
    print(f'Completed {len(results)} requests; {len(passed)} were not explicitly blocked')
    print('Review origin telemetry before classifying any result as exploitable.')


if __name__ == '__main__':
    main()