import sys sys.dont_write_bytecode = True import argparse import json import os import re from urllib.parse import urlparse import requests sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, OSError, ValueError): pass from keycheck_common import ( combined_provider_routing_hint, commit_status_transaction, default_input_file, default_proxy_file, ensure_output_files, iter_findings, keycheck_input_mode, load_checked_statuses, load_known_keys, load_proxies, mask_secret, provider_routing_database_failed, read_plain_keys, record_validation_result, recover_status_transaction, require_provider_authority, service_output_dir, should_skip_key, write_keycheck_event, ) from keycheckers.provider_resolution import resolve_provider_key SERVICE = "qwen" DETECTOR = "QwenDashScope" OUTPUT_DIR = os.getenv("KEYCHECK_OUTPUT_DIR") or service_output_dir(SERVICE) INPUT_FILE = os.getenv("KEYCHECK_INPUT_FILE") or default_input_file() PROXY_FILE = os.getenv("KEYCHECK_PROXY_FILE") or default_proxy_file() CHECKED_FILE = os.path.join(OUTPUT_DIR, "qwenChecked.txt") RESULTS_FILE = os.path.join(OUTPUT_DIR, "qwenResults.jsonl") STATUS_FILES = { "VALID": os.path.join(OUTPUT_DIR, "qwenAlive.txt"), "DEAD": os.path.join(OUTPUT_DIR, "qwenDead.txt"), "RESTRICTED": os.path.join(OUTPUT_DIR, "qwenRestricted.txt"), "LIMITED": os.path.join(OUTPUT_DIR, "qwenLimited.txt"), "NO_BALANCE": os.path.join(OUTPUT_DIR, "qwenNoBalance.txt"), "NO_CONTEXT": os.path.join(OUTPUT_DIR, "qwenNoContext.txt"), "NETWORK": os.path.join(OUTPUT_DIR, "qwenNetwork.txt"), "UNKNOWN": os.path.join(OUTPUT_DIR, "qwenUnknown.txt"), } QWEN_DETECTOR_NAMES = {"qwendashscope", "qwen_dashscope", "dashscope", "qwen"} QWEN_EXPLICIT_DETECTOR_NAMES = {"qwendashscope", "qwen_dashscope"} DEEPSEEK_EXPLICIT_DETECTOR_NAMES = {"deepseekapikey", "deepseek_api_key"} KIMI_EXPLICIT_DETECTOR_NAMES = {"kimimoonshot", "moonshotai"} QWEN_KEY_MAX_BYTES = 512 QWEN_KEY_REGEX = re.compile( r"(? QWEN_KEY_MAX_BYTES: return f"candidate exceeds the {QWEN_KEY_MAX_BYTES}-byte key limit" if OPENAI_LEGACY_KEY_MARKER in value: return "candidate is a recognizable OpenAI legacy key" if value.count("sk-") != 1: return "candidate contains multiple concatenated key prefixes" if not QWEN_KEY_REGEX.fullmatch(value) or value.startswith(FOREIGN_QWEN_KEY_PREFIXES): return "candidate does not match the bounded Qwen key format" return "" def is_qwen_key(key): return not qwen_key_rejection_reason(key) def finding_has_oversized_qwen_key(finding, *raw_values): values = list(raw_values) if isinstance(finding, dict): values.extend((finding.get("Raw"), finding.get("RawV2"), finding.get("raw"), finding.get("raw_v2"))) nested = finding.get("finding") if isinstance(nested, dict): values.extend((nested.get("Raw"), nested.get("RawV2"), nested.get("raw"), nested.get("raw_v2"))) return any( QWEN_OVERSIZED_KEY_PREFIX_REGEX.search(str(value or "")) for value in values ) def warn_rejected_candidate(reason, source, key=""): reason = str(reason or "candidate rejected") source = str(source or "unknown") if key: reason = reason.replace(key, "***REDACTED***") source = source.replace(key, "***REDACTED***") reason = reason.replace("\r", " ").replace("\n", " ")[:300] source = source.replace("\r", " ").replace("\n", " ")[:300] print(f"Warning: skipped Qwen candidate from {source}: {reason}", flush=True) def warn_candidate_failure(reason, source, key=""): reason = str(reason or "candidate failure") source = str(source or "unknown") if key: reason = reason.replace(key, "***REDACTED***") source = source.replace(key, "***REDACTED***") reason = QWEN_KEY_REGEX.sub("***REDACTED***", reason).replace("\r", " ").replace("\n", " ")[:300] source = QWEN_KEY_REGEX.sub("***REDACTED***", source).replace("\r", " ").replace("\n", " ")[:300] print(f"Warning: Qwen candidate failure from {source}: {reason}", flush=True) def extract_key_from_finding(data): if not detector_name_from_finding(data): return "" if data.get("Raw") or data.get("RawV2"): return key_from_text(data.get("Raw"), data.get("RawV2")) if data.get("raw") or data.get("raw_v2"): return key_from_text(data.get("raw"), data.get("raw_v2")) finding = data.get("finding") if isinstance(finding, dict): return key_from_text(finding.get("Raw"), finding.get("RawV2")) return "" def finding_provider_routing_hint(finding): if not isinstance(finding, dict): return "" context = finding.get("ScannerContext") if isinstance(finding.get("ScannerContext"), dict) else {} persisted_hint = context.get("provider_hint") if ( context.get("provider_hint_source") == EXPLICIT_ASSIGNMENT_HINT_SOURCE and persisted_hint in (*GENERIC_SK_PROVIDERS, AMBIGUOUS_PROVIDER_HINT, AMBIGUOUS_GENERIC_SK_HINT) ): return persisted_hint parts = [str(context.get(key) or "") for key in ("nearby", "file")] metadata = finding.get("SourceMetadata") if isinstance(finding.get("SourceMetadata"), dict) else {} data = metadata.get("Data") if isinstance(metadata.get("Data"), dict) else {} for details in data.values(): if not isinstance(details, dict): continue parts.extend(str(details.get(key) or "") for key in ("file", "repository", "repo", "link", "image")) text = "\n".join(parts) evidence = set() if QWEN_CONTEXT_REGEX.search(text) or finding_has_explicit_detector(finding, QWEN_EXPLICIT_DETECTOR_NAMES): evidence.add("qwen") if DEEPSEEK_CONTEXT_REGEX.search(text) or finding_has_explicit_detector(finding, DEEPSEEK_EXPLICIT_DETECTOR_NAMES): evidence.add("deepseek") if KIMI_CONTEXT_REGEX.search(text) or finding_has_explicit_detector(finding, KIMI_EXPLICIT_DETECTOR_NAMES): evidence.add("kimi") if persisted_hint == AMBIGUOUS_PROVIDER_HINT: evidence.update(("qwen", "deepseek")) elif persisted_hint == AMBIGUOUS_GENERIC_SK_HINT: evidence.update(GENERIC_SK_PROVIDERS) elif persisted_hint in GENERIC_SK_PROVIDERS: evidence.add(persisted_hint) if len(evidence) > 1: return AMBIGUOUS_PROVIDER_HINT if evidence == {"qwen", "deepseek"} else AMBIGUOUS_GENERIC_SK_HINT return next(iter(evidence)) if evidence else "" def finding_has_ambiguous_provider_hint(finding): return finding_provider_routing_hint(finding) in (AMBIGUOUS_PROVIDER_HINT, AMBIGUOUS_GENERIC_SK_HINT) def ensure_files(): ensure_output_files([CHECKED_FILE, RESULTS_FILE, *STATUS_FILES.values()]) recover_status_transaction(CHECKED_FILE, STATUS_FILES) def iter_candidate_keys(input_file, plain_files, trusted_retry_files=None): seen_plain = set() seen_candidates = set() routing_decisions = {} detector_names = [ "QwenDashScope", "Qwen_DashScope", "qwendashscope", "qwen_dashscope", "Qwen", "DashScope", "qwen", "dashscope", "CustomRegex", ] for item in iter_findings(input_file, detector_names): data = dict(item.get("finding") or {}) data.pop(CANDIDATE_PROVIDER_ROUTE_FIELD, None) candidate_metadata = item.get("candidate_metadata") persisted_route = "" if keycheck_input_mode() == "postgres" and isinstance(candidate_metadata, dict): persisted_route = str(candidate_metadata.get("provider_hint") or "").lower() if persisted_route == SERVICE: data[CANDIDATE_PROVIDER_ROUTE_FIELD] = SERVICE if finding_has_oversized_qwen_key(data, item.get("raw"), item.get("raw_v2")): warn_rejected_candidate( f"candidate exceeds the {QWEN_KEY_MAX_BYTES}-byte key limit", item.get("source") or input_file, ) continue if persisted_route == SERVICE: key = key_from_text( item.get("raw"), item.get("raw_v2"), data.get("Raw"), data.get("RawV2"), ) else: key = extract_key_from_finding(data) if key and is_qwen_key(key): if not key.startswith("sk-sp-"): if persisted_route == SERVICE: hint, lookup_failed = SERVICE, False else: local_hint = finding_provider_routing_hint(data) if key in routing_decisions: hint, lookup_failed = routing_decisions[key] if local_hint == AMBIGUOUS_PROVIDER_HINT or ( local_hint and hint and local_hint != hint ): hint = AMBIGUOUS_PROVIDER_HINT elif not hint: hint = local_hint routing_decisions[key] = (hint, lookup_failed) else: hint = combined_provider_routing_hint(key, local_hint) lookup_failed = provider_routing_database_failed() routing_decisions[key] = (hint, lookup_failed) if lookup_failed: message = "provider routing evidence lookup failed closed" warn_candidate_failure(message, item.get("source") or input_file, key) raise RuntimeError(message) if hint != "qwen" and not ( keycheck_input_mode() == "postgres" and hint in (AMBIGUOUS_PROVIDER_HINT, AMBIGUOUS_GENERIC_SK_HINT) ): continue seen_candidates.add(key) yield key, item.get("source") or input_file, data for item in read_plain_keys(plain_files, QWEN_KEY_REGEX): key = item["key"] if not is_qwen_key(key) or not key.startswith("sk-sp-"): continue if key not in seen_plain: seen_plain.add(key) seen_candidates.add(key) yield key, item["source"], {} owned_retry_paths = { os.path.normcase(os.path.abspath(path)) for path in STATUS_FILES.values() } retry_files = [ path for path in (trusted_retry_files or []) if os.path.normcase(os.path.abspath(path)) in owned_retry_paths ] for item in read_plain_keys(retry_files, QWEN_KEY_REGEX): key = item["key"] if not is_qwen_key(key) or key in seen_candidates: continue if not key.startswith("sk-sp-"): hint = combined_provider_routing_hint(key, "qwen") if provider_routing_database_failed(): message = "provider routing evidence lookup failed closed" warn_candidate_failure(message, item["source"], key) raise RuntimeError(message) if hint != "qwen": continue seen_candidates.add(key) yield key, item["source"], {} def redact_text(text, key): redacted = str(text or "")[:1000] if key: redacted = redacted.replace(key, "***REDACTED***") return QWEN_KEY_REGEX.sub("***REDACTED***", redacted) def parse_error_response(response, key): try: payload = response.json() except ValueError: payload = {} error = payload.get("error") if isinstance(payload, dict) else {} if not isinstance(error, dict): error = {} message = error.get("message") or response.text[:500] return { "http_status": response.status_code, "code": error.get("code") or error.get("type") or "", "type": error.get("type") or "", "message": redact_text(message, key), } def classify_error(error): http_status = int(error.get("http_status") or 0) code = str(error.get("code") or "").lower() message = str(error.get("message") or "").lower() if http_status == 401 or "invalid_api_key" in code or "incorrect api key" in message: return "DEAD" if http_status == 402 or "arrearage" in code or any(item in message for item in ( "arrearage", "arrears", "billing", "balance", "overdue", "payment", "insufficient credit", "credit balance", )): return "NO_BALANCE" if http_status == 403: return "RESTRICTED" if http_status == 429: return "LIMITED" if 500 <= http_status <= 599: return "NETWORK" return "UNKNOWN" def choose_chat_model(models): models = [str(model or "").replace("models/", "") for model in models if model] by_lower = {model.lower(): model for model in models} for model in CHAT_MODEL_PRIORITY: if model.lower() in by_lower: return by_lower[model.lower()] for model in models: lowered = model.lower() if any(marker in lowered for marker in NON_CHAT_MODEL_MARKERS): continue if any(marker in lowered for marker in ("qwen", "qwq", "qvq")): return model return "" def probe_chat_completion(key, base_url, model, proxy, timeout, debug=False): if not model: return {"status": "NO_CONTEXT", "message": "no chat-capable model from /models", "model": ""} url = f"{normalize_base_url(base_url)}/chat/completions" headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"} payload = {"model": model, "messages": [{"role": "user", "content": "ping"}], "max_tokens": 1} try: response = requests.post(url, headers=headers, json=payload, proxies=proxy, timeout=timeout) except requests.RequestException as exc: return {"status": "NETWORK", "message": str(exc)[:1000], "model": model} if debug: print(f" DEBUG {endpoint_label(base_url)} chat ping {model}: HTTP {response.status_code}: {redact_text(response.text[:500], key)}") if response.status_code == 200: return {"status": "GENERATION_OK", "message": "chat completion accepted", "model": model} error = parse_error_response(response, key) return {"status": classify_error(error), "error": error, "message": error.get("message") or "", "model": model} def parse_models(payload): if not isinstance(payload, dict): return [], [] model_infos = payload.get("data") if not isinstance(model_infos, list): model_infos = payload.get("models") if isinstance(payload.get("models"), list) else [] models = [] for item in model_infos: if not isinstance(item, dict): continue model_id = item.get("id") or item.get("model") or item.get("name") if model_id: models.append(str(model_id).replace("models/", "")) return sorted(set(models)), model_infos def notable_models(models): notable = [] for model in models: lowered = model.lower() if any(marker in lowered for marker in MODEL_MARKERS): notable.append(model) return notable[:30] def check_base_url(key, base_url, proxy, timeout, debug=False): url = f"{normalize_base_url(base_url)}/models" headers = {"Authorization": f"Bearer {key}", "Accept": "application/json"} try: response = requests.get(url, headers=headers, proxies=proxy, timeout=timeout) except requests.RequestException as exc: return { "base_url": base_url, "region": endpoint_label(base_url), "status": "NETWORK", "message": str(exc)[:1000], } if debug: print(f" DEBUG {endpoint_label(base_url)} /models: HTTP {response.status_code}: {redact_text(response.text[:500], key)}") if response.status_code == 200: try: payload = response.json() except ValueError: payload = {} models, model_infos = parse_models(payload) chat_model = choose_chat_model(models) probe = probe_chat_completion(key, base_url, chat_model, proxy, timeout, debug) probe_status = probe.get("status") or "UNKNOWN" status = "VALID" if probe_status in ("GENERATION_OK", "NO_CONTEXT") else probe_status probe_message = probe.get("message") or json.dumps(probe.get("error") or {}, ensure_ascii=False)[:1000] return { "base_url": base_url, "region": endpoint_label(base_url), "status": status, "authenticated": True, "model_count": len(models), "models": notable_models(models), "all_model_count": len(models), "model_infos_count": len(model_infos), "llm_probe_status": probe_status, "llm_probe_model": probe.get("model", chat_model), "message": ( f"chat ping ok; model={chat_model}; models={len(models)}" if probe_status == "GENERATION_OK" else f"models authenticated; generation_probe={probe_status}; models={len(models)}; {probe_message}" )[:1000], "error": probe.get("error") or {}, } error = parse_error_response(response, key) return { "base_url": base_url, "region": endpoint_label(base_url), "status": classify_error(error), "error": error, "message": error.get("message") or "", } def choose_final_status(key, attempts, has_custom_base_urls): statuses = [attempt.get("status") for attempt in attempts] for status in ("VALID", "NO_BALANCE", "LIMITED", "RESTRICTED", "UNKNOWN", "NO_CONTEXT", "NETWORK"): if status in statuses: return status return "DEAD" def check_key(key, base_urls, has_custom_base_urls, proxy, timeout, debug=False): rejection = qwen_key_rejection_reason(key) if rejection: return {"status": "UNKNOWN", "message": rejection, "candidate_rejected": True} attempts = [] for base_url in base_urls: result = check_base_url(key, base_url, proxy, timeout, debug) attempts.append(result) if result.get("status") == "VALID": return { "status": "VALID", "region": result.get("region"), "base_url": result.get("base_url"), "model_count": result.get("model_count", 0), "models": result.get("models", []), "llm_probe_status": result.get("llm_probe_status", ""), "llm_probe_model": result.get("llm_probe_model", ""), "authenticated": bool(result.get("authenticated")), "attempts": attempts, "message": f"models={result.get('model_count', 0)} region={result.get('region')}", } status = choose_final_status(key, attempts, has_custom_base_urls) message = "" for attempt in attempts: if attempt.get("status") == status: message = attempt.get("message") or json.dumps(attempt.get("error") or {}, ensure_ascii=False)[:1000] break return {"status": status, "attempts": attempts, "message": message} def write_result(key, result, source, finding): rejection = qwen_key_rejection_reason(key) if rejection: raise ValueError(rejection) status = result.get("status") or "UNKNOWN" write_keycheck_event(SERVICE, RESULTS_FILE, key, result, source, finding, DETECTOR) if status == "VALID": extra = f"{result.get('region', '')};probe_model={result.get('llm_probe_model', '')};models={','.join(result.get('models', []))[:500]}" else: extra = source commit_status_transaction( CHECKED_FILE, STATUS_FILES, key, status, result.get("message", ""), extra, ) record_validation_result(SERVICE, key, result, source, finding, DETECTOR) def retry_statuses_from_args(args): retry_statuses = set() if args.retry_network: retry_statuses.add("NETWORK") if args.retry_limited: retry_statuses.add("LIMITED") if args.retry_unknown: retry_statuses.update({"UNKNOWN", "NO_CONTEXT"}) if args.retry_restricted: retry_statuses.add("RESTRICTED") if args.retry_no_balance: retry_statuses.add("NO_BALANCE") if args.retry_valid: retry_statuses.add("VALID") return retry_statuses def retry_input_files_from_args(args): if args.recheck_all: statuses = list(STATUS_FILES) else: statuses = [] if args.retry_network: statuses.append("NETWORK") if args.retry_limited: statuses.append("LIMITED") if args.retry_unknown: statuses.extend(("UNKNOWN", "NO_CONTEXT")) if args.retry_restricted: statuses.append("RESTRICTED") if args.retry_no_balance: statuses.append("NO_BALANCE") if args.retry_valid: statuses.append("VALID") return list(dict.fromkeys(STATUS_FILES[status] for status in statuses)) def base_urls_from_args(args): env_urls = split_csv(os.getenv("QWEN_BASE_URLS") or os.getenv("DASHSCOPE_BASE_URLS")) custom_urls = [] for value in args.base_url: custom_urls.extend(split_csv(value)) custom_urls.extend(env_urls) default_urls = [] if args.no_default_base_urls else DEFAULT_BASE_URLS return unique_ordered(custom_urls + default_urls), bool(custom_urls) def parse_args(): parser = argparse.ArgumentParser(description="Qwen/DashScope key checker") parser.add_argument("--input", default=INPUT_FILE) parser.add_argument("--plain", action="append", default=[]) parser.add_argument("--proxy-file", default=PROXY_FILE) parser.add_argument("--timeout", type=int, default=15) parser.add_argument("--max-keys", type=int, default=0) parser.add_argument("--base-url", action="append", default=[], help="Extra DashScope/OpenAI-compatible base URL; can be repeated") parser.add_argument("--no-default-base-urls", action="store_true") parser.add_argument("--retry-network", action="store_true") parser.add_argument("--retry-limited", action="store_true") parser.add_argument("--retry-unknown", action="store_true") parser.add_argument("--retry-restricted", action="store_true") parser.add_argument("--retry-no-balance", action="store_true") parser.add_argument("--retry-valid", action="store_true") parser.add_argument("--recheck-all", action="store_true") parser.add_argument("--debug", action="store_true") return parser.parse_args() def main(): require_provider_authority(SERVICE) args = parse_args() ensure_files() proxy_cycler = load_proxies(args.proxy_file) checked = load_checked_statuses(CHECKED_FILE) known = load_known_keys(CHECKED_FILE, STATUS_FILES) retry_statuses = retry_statuses_from_args(args) retry_input_files = retry_input_files_from_args(args) base_urls, has_custom_base_urls = base_urls_from_args(args) if not base_urls: raise SystemExit("No Qwen/DashScope base URLs configured") print("--- Qwen/DashScope key checker ---") print("base_urls: " + ", ".join(endpoint_label(url) for url in base_urls)) processed = 0 skipped = 0 for key, source, finding in iter_candidate_keys(args.input, args.plain, retry_input_files): finding = dict(finding or {}) candidate_route = str(finding.pop(CANDIDATE_PROVIDER_ROUTE_FIELD, "") or "").lower() rejection = qwen_key_rejection_reason(key) if rejection: skipped += 1 warn_rejected_candidate(rejection, source, key) continue try: should_skip = should_skip_key( key, checked, known, args, retry_statuses, service=SERVICE, source=source, finding=finding, detector=DETECTOR, ) except Exception as exc: warn_candidate_failure(f"candidate preparation failed: {exc}", source, key) raise if should_skip: skipped += 1 continue if args.max_keys and processed >= args.max_keys: break processed += 1 try: print(f"\n[{processed}] Qwen/DashScope candidate {mask_secret(key)} from {source}") proxy = next(proxy_cycler) if proxy_cycler else None routing_hint = "qwen" if keycheck_input_mode() == "postgres": if candidate_route == SERVICE: routing_hint = SERVICE else: routing_hint = combined_provider_routing_hint(key, finding_provider_routing_hint(finding)) if provider_routing_database_failed(): raise RuntimeError("provider routing evidence lookup failed closed") if routing_hint in (AMBIGUOUS_PROVIDER_HINT, AMBIGUOUS_GENERIC_SK_HINT): result = resolve_provider_key( key, finding, proxy, args.timeout, args.debug, hint=routing_hint, origin_service=SERVICE, ) else: result = check_key(key, base_urls, has_custom_base_urls, proxy, args.timeout, args.debug) print(f" STATUS: {result['status']} | {result.get('message', '')[:200]}") write_result(key, result, source, finding) known.add(key) checked[key] = result["status"] except Exception as exc: warn_candidate_failure(f"candidate processing failed: {exc}", source, key) raise print(f"\nDone. Processed={processed}, skipped={skipped}, results={RESULTS_FILE}") if __name__ == "__main__": main()