import sys sys.dont_write_bytecode = True import argparse import json import os import re from urllib.parse import urlsplit import requests sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from keycheck_candidates import extract_azure_foundry_parts from keycheck_common import ( append_jsonl, append_status, commit_status_transaction, default_input_file, default_proxy_file, ensure_output_files, keycheck_input_mode, iter_findings, iter_bounded_text_lines, load_checked_statuses, load_known_keys, load_proxies, mask_secret, recover_status_transaction, request_error_message, record_validation_result, require_provider_authority, service_output_dir, should_skip_key, write_keycheck_event, ) SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) PARENT_DIR = os.path.dirname(SCRIPT_DIR) SERVICE = "azure" 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, "azureChecked.txt") RESULTS_FILE = os.path.join(OUTPUT_DIR, "azureResults.jsonl") AZURE_OPENAI_LLM_FILE = os.path.join(OUTPUT_DIR, "azureOpenAILLM.txt") AZURE_OPENAI_PLAIN_FILE = os.path.join(OUTPUT_DIR, "azureOpenAI.txt") AZURE_FOUNDRY_PLAIN_FILE = os.path.join(OUTPUT_DIR, "azureFoundry.txt") STATUS_FILES = { "VALID": os.path.join(OUTPUT_DIR, "azureAlive.txt"), "DEAD": os.path.join(OUTPUT_DIR, "azureDead.txt"), "RESTRICTED": os.path.join(OUTPUT_DIR, "azureRestricted.txt"), "NETWORK": os.path.join(OUTPUT_DIR, "azureNetwork.txt"), "UNKNOWN": os.path.join(OUTPUT_DIR, "azureUnknown.txt"), "OPENAI_UNRESOLVED": os.path.join(OUTPUT_DIR, "azureOpenAIUnresolved.txt"), "OPENAI_BAD_ENDPOINT": os.path.join(OUTPUT_DIR, "azureOpenAIBadEndpoint.txt"), "FOUNDRY": os.path.join(OUTPUT_DIR, "azureFoundryLLM.txt"), "FOUNDRY_UNRESOLVED": os.path.join(OUTPUT_DIR, "azureFoundryUnresolved.txt"), "FOUNDRY_BAD_ENDPOINT": os.path.join(OUTPUT_DIR, "azureFoundryBadEndpoint.txt"), } AZURE_OPENAI_ENDPOINT_RE = re.compile(r"([a-z0-9-]+\.openai\.azure\.com)", re.IGNORECASE) AZURE_FOUNDRY_HOST_RE = r"[a-z0-9-]+(?:\.[a-z0-9-]+)*\.(?:models\.ai\.azure\.com|services\.ai\.azure\.com|inference\.ai\.azure\.com)" AZURE_FOUNDRY_ENDPOINT_RE = re.compile(r"((?:https?://)?" + AZURE_FOUNDRY_HOST_RE + r"(?:/[^\s:\"'<>\\]*)?)", re.IGNORECASE) AZURE_OPENAI_DEPLOYMENTS_API_VERSION = "2023-03-15-preview" AZURE_OPENAI_CHAT_API_VERSION = "2024-02-15-preview" AZURE_FOUNDRY_API_VERSION = "2024-05-01-preview" AZURE_FOUNDRY_KEY_ASSIGNMENT_RE = re.compile( r"(?is)(?:authorization|api[_-]?key|key|token|secret|credential|bearer)[^\n:=]{0,80}[:=]\s*[\"']?(?:bearer\s+)?([A-Za-z0-9_./+=\-]{20,512})" ) NON_FOUNDRY_KEY_PREFIXES = ( "sk-", "sk_", "sk-or-", "xai-", "ghp_", "gho_", "ghu_", "ghs_", "ghr_", "github_pat_", "glpat-", "glrt-", "hf_", "AIza", "AQ.", "zai-", "gsk_", "r8_", "nvapi-", ) def transaction_status_files(): return {**STATUS_FILES, "AUX_OPENAI_LLM": AZURE_OPENAI_LLM_FILE} def ensure_files(): ensure_output_files([CHECKED_FILE, RESULTS_FILE, AZURE_OPENAI_LLM_FILE, AZURE_OPENAI_PLAIN_FILE, AZURE_FOUNDRY_PLAIN_FILE, *STATUS_FILES.values()]) recover_status_transaction(CHECKED_FILE, transaction_status_files()) def parse_azure_sp(raw_v2): try: data = json.loads(raw_v2) except (TypeError, ValueError): return None client_secret = data.get("clientSecret") or data.get("client_secret") client_id = data.get("clientId") or data.get("client_id") tenant_id = data.get("tenantId") or data.get("tenant_id") if not all([client_secret, client_id, tenant_id]): return None return {"client_secret": client_secret, "client_id": client_id, "tenant_id": tenant_id} def scanner_context_text(finding): context = finding.get("ScannerContext") if isinstance(finding, dict) else None if isinstance(context, dict): return str(context.get("nearby") or "") return "" def foundry_keyish(value): text = re.sub(r"(?i)^bearer\s+", "", str(value or "").strip().strip('"\'`,;')).strip() lower = text.lower() if not (20 <= len(text) <= 512): return False if any(marker in lower for marker in ("http://", "https://", "{{", "${", "<", "azure.com")): return False if any(ch.isspace() for ch in text): return False if re.match(r"(?i)^(?:authorization|api[_-]?key|key|token|secret|credential|bearer)\s*[:=]", text): return False if text.startswith(NON_FOUNDRY_KEY_PREFIXES): return False return bool(re.search(r"[A-Za-z]", text) and re.search(r"[0-9]", text)) def normalize_foundry_endpoint(value): text = str(value or "").strip().strip('"\'`,;') if not text: return "" split_text = text if re.match(r"(?i)^https?://", text) else "https://" + text try: parsed = urlsplit(split_text) host = parsed.netloc or parsed.path.split("/", 1)[0] path = parsed.path if parsed.netloc else ("/" + parsed.path.split("/", 1)[1] if "/" in parsed.path else "") except Exception: host, path = re.sub(r"(?i)^https?://", "", text).split("/", 1)[0], "" path = path.rstrip(".,;:)]}/") terminal_routes = ( ("/models/chat/completions", ""), ("/openai/v1/chat/completions", "/openai/v1"), ("/v1/chat/completions", "/v1"), ("/chat/completions", ""), ("/v1/models", "/v1"), ("/models", ""), ) lower_path = path.lower() for suffix, replacement in terminal_routes: if lower_path.endswith(suffix): path = path[:-len(suffix)] + replacement break return (host + path).strip("/").lower() def split_foundry_endpoint_key(text): candidate = str(text or "").strip().split("\t", 1)[0].strip() if not candidate: return None endpoint_match = AZURE_FOUNDRY_ENDPOINT_RE.search(candidate) if not endpoint_match: return None endpoint = normalize_foundry_endpoint(endpoint_match.group(1)) before = candidate[:endpoint_match.start()].replace("https://", " ").replace("http://", " ").strip(" \t:=,;'\"/") after = candidate[endpoint_match.end():].replace("https://", " ").replace("http://", " ").strip(" \t:=,;'\"/") for key in (after, before): if foundry_keyish(key): return {"key": key, "endpoint": endpoint} return None def foundry_context_values(finding): values = [] for value in (finding.get("Raw"), finding.get("RawV2")) if isinstance(finding, dict) else (): if value: text = str(value) values.append(text) values.extend(str(item) for item in AZURE_FOUNDRY_KEY_ASSIGNMENT_RE.findall(text)) context = scanner_context_text(finding) if context: values.append(context) values.extend(str(item) for item in AZURE_FOUNDRY_KEY_ASSIGNMENT_RE.findall(context or "")) extra = finding.get("ExtraData") if isinstance(finding, dict) else None if isinstance(extra, dict): values.extend(str(value) for value in extra.values() if isinstance(value, str)) return values def parse_azure_openai(raw, raw_v2, finding): key = raw or "" endpoint = "" raw_v2 = raw_v2 or "" match = re.match(r"^([a-f0-9]{32}):(.+\.openai\.azure\.com)$", raw_v2, re.IGNORECASE) if match: key = match.group(1) endpoint = match.group(2) if not endpoint: context_match = AZURE_OPENAI_ENDPOINT_RE.search(scanner_context_text(finding)) if context_match: endpoint = context_match.group(1) if not key: return None return {"key": key, "endpoint": endpoint} def parse_azure_openai_line(line): text = str(line or "").strip() if not text: return None text = text.split("\t", 1)[0].strip() if ":" in text: endpoint, key = text.split(":", 1) if AZURE_OPENAI_ENDPOINT_RE.fullmatch(endpoint.strip()) and key.strip(): return {"endpoint": endpoint.strip(), "key": key.strip()} endpoint_match = AZURE_OPENAI_ENDPOINT_RE.search(text) key_match = re.search(r"\b[a-f0-9]{32}\b", text, re.IGNORECASE) if endpoint_match and key_match: return {"endpoint": endpoint_match.group(1), "key": key_match.group(0)} if key_match: return {"endpoint": "", "key": key_match.group(0)} return None def parse_azure_foundry(raw, raw_v2, finding): key = raw or "" endpoint = "" raw_v2 = raw_v2 or "" split = split_foundry_endpoint_key(raw_v2) or split_foundry_endpoint_key(raw) if not split: split = extract_azure_foundry_parts(raw, raw_v2) if split: key = split["key"] endpoint = split["endpoint"] if not endpoint: for endpoint_text in (raw, raw_v2, scanner_context_text(finding)): context_match = AZURE_FOUNDRY_ENDPOINT_RE.search(str(endpoint_text or "")) if context_match: endpoint = normalize_foundry_endpoint(context_match.group(1)) break if not foundry_keyish(key): for value in foundry_context_values(finding): if foundry_keyish(value): key = value.strip().strip('"\'`,;') break if not key or not foundry_keyish(key): return None return {"key": key.strip().strip('"\'`,;'), "endpoint": normalize_foundry_endpoint(endpoint)} def parse_azure_foundry_line(line): parts = str(line or "").strip().split("\t", 1) text = parts[0].strip() if not text: return None split = split_foundry_endpoint_key(text) parsed = split if split else ({"key": text, "endpoint": ""} if foundry_keyish(text) else None) if not parsed: return None if len(parts) > 1: try: metadata = json.loads(parts[1]) except ValueError: metadata = {} if isinstance(metadata, dict): parsed["finding_uid"] = metadata.get("finding_uid") or "" parsed["origin"] = metadata.get("origin") or "" return parsed def parse_azure_acr(raw_v2): try: data = json.loads(raw_v2) except (TypeError, ValueError): return None username = data.get("username") password = data.get("password") if not username or not password: return None return {"username": username, "password": password} def azure_openai_key(parsed): endpoint = parsed.get("endpoint") or "" return f"{endpoint}:{parsed['key']}" if endpoint else parsed["key"] def azure_acr_key(parsed): return f"{parsed['username']}:{parsed['password']}" def azure_sp_key(parsed): return f"{parsed['tenant_id']}:{parsed['client_id']}:{parsed['client_secret']}" def azure_foundry_key(parsed): endpoint = parsed.get("endpoint") or "" return f"{endpoint}:{parsed['key']}" if endpoint else parsed["key"] def extract_candidates(input_file): seen_plain = set() foundry_detectors = {"AzureFoundryEndpointBeforeKey", "AzureFoundryKeyBeforeEndpoint"} detector_names = ["AzureOpenAI", "AzureContainerRegistry", "Azure", *sorted(foundry_detectors)] for item in iter_findings(input_file, detector_names): candidate_kind = item.get('candidate_kind') or '' if keycheck_input_mode() == 'postgres' and candidate_kind: secret_text = item.get('credential_secret_text') or '' secret_json = item.get('credential_secret_json') or '' endpoint = item.get('credential_endpoint') or '' parsed = None detector = '' if candidate_kind == 'azure_service_principal': parsed = parse_azure_sp(secret_json) detector = 'Azure' key = azure_sp_key(parsed) if parsed else '' elif candidate_kind == 'azure_container_registry': parsed = parse_azure_acr(secret_json) detector = 'AzureContainerRegistry' key = azure_acr_key(parsed) if parsed else '' elif candidate_kind == 'azure_openai': parsed = {'key': secret_text, 'endpoint': endpoint} if secret_text else None detector = 'AzureOpenAI' key = azure_openai_key(parsed) if parsed else '' elif candidate_kind == 'azure_foundry': parsed = ( {'key': secret_text, 'endpoint': normalize_foundry_endpoint(endpoint)} if foundry_keyish(secret_text) and endpoint else parse_azure_foundry(secret_text, '', item.get('finding') or {}) ) if parsed and endpoint: parsed['endpoint'] = normalize_foundry_endpoint(endpoint) detector = 'AzureFoundry' key = azure_foundry_key(parsed) if parsed else '' else: key = '' if parsed and key: yield key, detector, item['source'], item['finding'], parsed continue unresolved_detector = { 'azure_openai': 'AzureOpenAI', 'azure_foundry': 'AzureFoundry', 'azure_container_registry': 'AzureContainerRegistry', 'azure_service_principal': 'Azure', }.get(candidate_kind, 'Azure') unresolved_key = secret_text or secret_json or candidate_kind if unresolved_key: yield unresolved_key, unresolved_detector, item['source'], item['finding'], { '_unresolved_candidate': True, 'candidate_kind': candidate_kind, } continue if item["detector"] == "AzureOpenAI": foundry = parse_azure_foundry(item["raw"], item["raw_v2"], item["finding"]) if foundry and foundry.get("endpoint"): key = azure_foundry_key(foundry) yield key, "AzureFoundry", item["source"], item["finding"], foundry parsed = parse_azure_openai(item["raw"], item["raw_v2"], item["finding"]) if not parsed: continue key = azure_openai_key(parsed) yield key, "AzureOpenAI", item["source"], item["finding"], parsed continue if item["detector"] == "AzureContainerRegistry": parsed = parse_azure_acr(item["raw_v2"]) if not parsed: continue key = azure_acr_key(parsed) yield key, "AzureContainerRegistry", item["source"], item["finding"], parsed continue if item["detector"] == "Azure": parsed = parse_azure_sp(item["raw_v2"]) if not parsed: continue key = azure_sp_key(parsed) yield key, "Azure", item["source"], item["finding"], parsed continue if item["detector"] in foundry_detectors: foundry = parse_azure_foundry(item.get("raw"), item.get("raw_v2"), item.get("finding")) if not foundry or not foundry.get("endpoint"): continue key = azure_foundry_key(foundry) yield key, "AzureFoundry", item["source"], item["finding"], foundry if keycheck_input_mode() == 'jsonl' and os.path.exists(AZURE_FOUNDRY_PLAIN_FILE): for line_num, line in enumerate(iter_bounded_text_lines(AZURE_FOUNDRY_PLAIN_FILE), 1): parsed = parse_azure_foundry_line(line) if not parsed: continue key = azure_foundry_key(parsed) if key not in seen_plain: seen_plain.add(key) yield key, "AzureFoundry", f"{AZURE_FOUNDRY_PLAIN_FILE}:{line_num}", {}, parsed if keycheck_input_mode() == 'jsonl' and os.path.exists(AZURE_OPENAI_PLAIN_FILE): for line_num, line in enumerate(iter_bounded_text_lines(AZURE_OPENAI_PLAIN_FILE), 1): parsed = parse_azure_openai_line(line) if not parsed: continue key = azure_openai_key(parsed) if key not in seen_plain: seen_plain.add(key) yield key, "AzureOpenAI", f"{AZURE_OPENAI_PLAIN_FILE}:{line_num}", {}, parsed def permission_matches(action, pattern): action = str(action or "").lower() pattern = str(pattern or "").lower() if pattern == "*": return True if pattern.endswith("/*"): return action.startswith(pattern[:-1]) return action == pattern def has_action(actions, wanted): return any(permission_matches(wanted, action) for action in actions) def probe_azure_rbac(access_token, proxy, timeout, max_subscriptions=3): if not access_token: return {"azure_rbac_level": "unknown", "message": "rbac_probe=no_access_token"} headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"} try: response = requests.get( "https://management.azure.com/subscriptions?api-version=2020-01-01", headers=headers, proxies=proxy, timeout=timeout, ) except requests.RequestException as exc: return {"azure_rbac_level": "network", "message": f"rbac_probe_network={str(exc)[:200]}"} if response.status_code == 403: return {"azure_rbac_level": "token_only", "azure_subscription_count": 0, "message": "rbac_probe=subscriptions_forbidden"} if response.status_code >= 400: return {"azure_rbac_level": "unknown", "azure_rbac_http_status": response.status_code, "message": f"rbac_probe_http={response.status_code}:{request_error_message(response)[:200]}"} payload = response.json() subscriptions = payload.get("value") if isinstance(payload, dict) else [] subscriptions = subscriptions or [] sub_ids = [item.get("subscriptionId") for item in subscriptions if isinstance(item, dict) and item.get("subscriptionId")] if not sub_ids: return {"azure_rbac_level": "token_only", "azure_subscription_count": 0, "message": "rbac_probe=no_subscriptions"} all_actions = set() all_not_actions = set() permission_errors = [] for sub_id in sub_ids[:max_subscriptions]: url = f"https://management.azure.com/subscriptions/{sub_id}/providers/Microsoft.Authorization/permissions?api-version=2022-04-01" try: perms_response = requests.get(url, headers=headers, proxies=proxy, timeout=timeout) except requests.RequestException as exc: permission_errors.append(f"{sub_id}:network:{str(exc)[:120]}") continue if perms_response.status_code >= 400: permission_errors.append(f"{sub_id}:http_{perms_response.status_code}:{request_error_message(perms_response)[:120]}") continue data = perms_response.json().get("value") or [] for item in data: for action in item.get("actions") or []: all_actions.add(str(action)) for action in item.get("notActions") or []: all_not_actions.add(str(action)) can_all = has_action(all_actions, "*") can_assign_roles = has_action(all_actions, "Microsoft.Authorization/roleAssignments/write") and not has_action(all_not_actions, "Microsoft.Authorization/roleAssignments/write") can_manage_cognitive = any( has_action(all_actions, action) for action in ( "Microsoft.CognitiveServices/accounts/write", "Microsoft.CognitiveServices/accounts/deployments/write", "Microsoft.CognitiveServices/*", ) ) or can_all can_manage_ml = any( has_action(all_actions, action) for action in ( "Microsoft.MachineLearningServices/workspaces/write", "Microsoft.MachineLearningServices/*", ) ) or can_all can_deploy_resources = has_action(all_actions, "Microsoft.Resources/deployments/write") or can_all can_manage_ai = can_manage_cognitive or can_manage_ml if can_all and can_assign_roles: level = "owner_like" elif can_all: level = "contributor_like" elif can_manage_ai: level = "ai_manager" elif all_actions: level = "limited" else: level = "subscriptions_visible" message = ( f"rbac_probe={level}; subscriptions={len(sub_ids)}; " f"can_manage_ai={can_manage_ai}; can_assign_roles={can_assign_roles}; can_deploy_resources={can_deploy_resources}" ) if permission_errors and not all_actions: message += "; permission_errors=" + " | ".join(permission_errors[:3]) return { "azure_rbac_level": level, "azure_subscription_count": len(sub_ids), "azure_subscription_ids": sub_ids[:10], "azure_can_manage_ai": can_manage_ai, "azure_can_manage_cognitive": can_manage_cognitive, "azure_can_manage_ml": can_manage_ml, "azure_can_assign_roles": can_assign_roles, "azure_can_deploy_resources": can_deploy_resources, "azure_permission_actions_sample": sorted(all_actions)[:40], "message": message, } def check_service_principal(parsed, proxy, timeout): url = f"https://login.microsoftonline.com/{parsed['tenant_id']}/oauth2/v2.0/token" payload = { "client_id": parsed["client_id"], "client_secret": parsed["client_secret"], "scope": "https://management.azure.com/.default", "grant_type": "client_credentials", } try: response = requests.post(url, data=payload, proxies=proxy, timeout=timeout) except requests.RequestException as exc: return {"status": "NETWORK", "message": str(exc)} if response.status_code == 200: data = response.json() rbac = probe_azure_rbac(data.get("access_token"), proxy, timeout) message = "token issued" if rbac.get("message"): message = f"{message}; {rbac.get('message')}" return { "status": "VALID", "tenant_id": parsed["tenant_id"], "client_id": parsed["client_id"], "expires_in": data.get("expires_in"), "message": message, **rbac, } message = request_error_message(response) lower = message.lower() if response.status_code in (400, 401) and ("invalid_client" in lower or "invalid_grant" in lower): return {"status": "DEAD", "http_status": response.status_code, "message": message} if response.status_code in (401, 403): return {"status": "RESTRICTED", "http_status": response.status_code, "message": message} return {"status": "UNKNOWN", "http_status": response.status_code, "message": message} def azure_openai_deployment_ids(payload): data = payload.get("data") if isinstance(payload, dict) else None if data is None and isinstance(payload, dict): data = payload.get("value") deployments = [] for item in data or []: if not isinstance(item, dict): continue deployment_id = item.get("id") or item.get("name") model = item.get("model") or item.get("modelName") or "" if deployment_id: deployments.append({"id": deployment_id, "model": model}) return deployments def endpoint_url(endpoint, path): endpoint = str(endpoint or "").strip().rstrip("/") if not endpoint.startswith("http://") and not endpoint.startswith("https://"): endpoint = "https://" + endpoint path = "/" + str(path or "").lstrip("/") parsed = urlsplit(endpoint) endpoint_path = parsed.path.rstrip("/") if endpoint_path and path.lower().startswith(endpoint_path.lower() + "/"): path = path[len(endpoint_path):] return endpoint + path def azure_model_ids(payload): data = payload.get("data") if isinstance(payload, dict) else payload if isinstance(payload, list) else [] if data is None and isinstance(payload, dict): data = payload.get("value") or payload.get("models") models = [] for item in data or []: if isinstance(item, str): models.append(item) elif isinstance(item, dict): model_id = item.get("id") or item.get("name") or item.get("model") or item.get("modelName") if model_id: models.append(str(model_id)) return models def endpoint_failure_status(error_text, status): lower = str(error_text or "").lower() if any(item in lower for item in ( "name resolution", "no such host", "failed to resolve", "getaddrinfo", "unexpected_eof", "eof occurred in violation of protocol", "ssleoferror", )): return status return "NETWORK" def probe_azure_openai_chat_route(endpoint, key, deployments, proxy, timeout): if not deployments: return {"route_probe": "no_deployments"} preferred = None for item in deployments: text = f"{item.get('id', '')} {item.get('model', '')}".lower() if any(marker in text for marker in ("gpt", "chat", "turbo", "4o")): preferred = item break deployment = preferred or deployments[0] deployment_id = deployment["id"] url = f"https://{endpoint}/openai/deployments/{deployment_id}/chat/completions?api-version={AZURE_OPENAI_CHAT_API_VERSION}" headers = {"api-key": key, "Content-Type": "application/json"} # Empty messages should fail validation after auth/deployment routing, without generating content. payload = {"messages": [], "max_tokens": 1} try: response = requests.post(url, headers=headers, json=payload, proxies=proxy, timeout=timeout) except requests.RequestException as exc: return {"route_probe": "network", "route_deployment": deployment_id, "route_message": str(exc)[:500]} message = request_error_message(response) if response.status_code in (200, 400): return { "route_probe": "accepted_auth_route", "route_deployment": deployment_id, "route_model": deployment.get("model", ""), "route_http_status": response.status_code, "route_message": message, } if response.status_code in (401, 403): return {"route_probe": "auth_failed", "route_deployment": deployment_id, "route_http_status": response.status_code, "route_message": message} if response.status_code == 404: return {"route_probe": "not_found", "route_deployment": deployment_id, "route_http_status": 404, "route_message": message} return {"route_probe": "unknown", "route_deployment": deployment_id, "route_http_status": response.status_code, "route_message": message} def check_azure_openai(parsed, proxy, timeout, probe_openai_route=False): endpoint = (parsed.get("endpoint") or "").strip().strip("/") key = parsed.get("key") if not endpoint: return {"status": "OPENAI_UNRESOLVED", "message": "AzureOpenAI key found without endpoint/resource name"} url = f"https://{endpoint}/openai/deployments?api-version={AZURE_OPENAI_DEPLOYMENTS_API_VERSION}" headers = {"api-key": key, "Content-Type": "application/json"} try: response = requests.get(url, headers=headers, proxies=proxy, timeout=timeout) except requests.RequestException as exc: first_error = str(exc) def endpoint_failure_status(error_text): lower = str(error_text or "").lower() if any(item in lower for item in ( "name resolution", "no such host", "failed to resolve", "getaddrinfo", "unexpected_eof", "eof occurred in violation of protocol", "ssleoferror", )): return {"status": "OPENAI_BAD_ENDPOINT", "endpoint": endpoint, "message": error_text} return None endpoint_status = endpoint_failure_status(first_error) if endpoint_status: return endpoint_status return {"status": "NETWORK", "endpoint": endpoint, "message": first_error} if response.status_code == 200: deployments = azure_openai_deployment_ids(response.json()) result = { "status": "VALID", "endpoint": endpoint, "deployment_count": len(deployments), "deployments": [item.get("id") for item in deployments[:20]], "message": f"deployments endpoint accepted key; deployments={len(deployments)}", } if probe_openai_route: result.update(probe_azure_openai_chat_route(endpoint, key, deployments, proxy, timeout)) return result message = request_error_message(response) if response.status_code in (401, 403): return {"status": "DEAD", "endpoint": endpoint, "http_status": response.status_code, "message": message} if response.status_code == 404: return {"status": "UNKNOWN", "endpoint": endpoint, "http_status": 404, "message": message} if response.status_code >= 500: return {"status": "NETWORK", "endpoint": endpoint, "http_status": response.status_code, "message": message} return {"status": "UNKNOWN", "endpoint": endpoint, "http_status": response.status_code, "message": message} def foundry_model_routes(endpoint): return [ endpoint_url(endpoint, f"/models?api-version={AZURE_FOUNDRY_API_VERSION}"), endpoint_url(endpoint, "/models"), endpoint_url(endpoint, "/v1/models"), ] def foundry_auth_headers(key): return [ {"api-key": key, "Content-Type": "application/json"}, {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, ] def check_foundry_models(endpoint, key, proxy, timeout): attempts = [] auth_failures = 0 attempted = 0 for url in foundry_model_routes(endpoint): for headers in foundry_auth_headers(key): attempted += 1 auth_kind = "bearer" if "Authorization" in headers else "api-key" try: response = requests.get(url, headers=headers, proxies=proxy, timeout=timeout) except requests.RequestException as exc: status = endpoint_failure_status(str(exc), "FOUNDRY_BAD_ENDPOINT") attempts.append(f"{url}:{auth_kind}:network:{str(exc)[:180]}") if status == "FOUNDRY_BAD_ENDPOINT": return {"status": status, "endpoint": endpoint, "message": str(exc)} continue message = request_error_message(response) if response.status_code == 200: models = azure_model_ids(response.json()) return { "status": "FOUNDRY", "endpoint": endpoint, "auth_scheme": auth_kind, "model_count": len(models), "models": models[:50], "message": f"models endpoint accepted key; auth={auth_kind}; models={len(models)}", } if response.status_code == 429: return { "status": "FOUNDRY", "endpoint": endpoint, "auth_scheme": auth_kind, "model_count": 0, "models": [], "message": f"models endpoint rate limited after auth; auth={auth_kind}; {message[:200]}", } if response.status_code in (401, 403): auth_failures += 1 attempts.append(f"{url}:{auth_kind}:auth_{response.status_code}:{message[:160]}") continue if response.status_code == 404: attempts.append(f"{url}:{auth_kind}:http_404:{message[:160]}") continue if response.status_code >= 500: attempts.append(f"{url}:{auth_kind}:server_{response.status_code}:{message[:160]}") continue attempts.append(f"{url}:{auth_kind}:http_{response.status_code}:{message[:160]}") if attempted and auth_failures == attempted: return {"status": "DEAD", "endpoint": endpoint, "message": "; ".join(attempts[:4])} return {"status": "UNKNOWN", "endpoint": endpoint, "message": "; ".join(attempts[:4])} def probe_foundry_route(endpoint, key, models, proxy, timeout): configured = [item.strip() for item in (models or []) if item.strip()] if not configured: return {"foundry_route_probe": "not_configured"} attempts = [] accepted = [] for model in configured: route_specs = [ (endpoint_url(endpoint, f"/models/chat/completions?api-version={AZURE_FOUNDRY_API_VERSION}"), {"model": model, "messages": [], "max_tokens": 1}), (endpoint_url(endpoint, "/chat/completions"), {"model": model, "messages": [], "max_tokens": 1}), (endpoint_url(endpoint, "/v1/chat/completions"), {"model": model, "messages": [], "max_tokens": 1}), (endpoint_url(endpoint, "/openai/v1/chat/completions"), {"model": model, "messages": [], "max_tokens": 1}), ] for url, payload in route_specs: for headers in foundry_auth_headers(key): auth_kind = "bearer" if "Authorization" in headers else "api-key" try: response = requests.post(url, headers=headers, json=payload, proxies=proxy, timeout=timeout) except requests.RequestException as exc: attempts.append(f"{model}:{auth_kind}:network:{str(exc)[:120]}") continue message = request_error_message(response) if response.status_code == 200: accepted.append(model) break if response.status_code == 400 and any(item in message.lower() for item in ("messages", "content", "validation", "empty")): accepted.append(model) break if response.status_code == 429: accepted.append(model) break if response.status_code in (401, 403, 404): attempts.append(f"{model}:{auth_kind}:http_{response.status_code}:{message[:160]}") continue attempts.append(f"{model}:{auth_kind}:http_{response.status_code}:{message[:160]}") if model in accepted: break if accepted: return {"foundry_route_probe": "accepted", "foundry_route_models": accepted, "foundry_route_message": "route accepted"} return {"foundry_route_probe": "not_accepted", "foundry_route_models": [], "foundry_route_message": "; ".join(attempts[:8])} def check_azure_foundry(parsed, proxy, timeout, probe_foundry_route_enabled=False, foundry_models=None): endpoint = (parsed.get("endpoint") or "").strip().strip("/") key = parsed.get("key") if not endpoint: return {"status": "FOUNDRY_UNRESOLVED", "message": "Azure Foundry key found without endpoint"} result = check_foundry_models(endpoint, key, proxy, timeout) if probe_foundry_route_enabled: route = probe_foundry_route(endpoint, key, foundry_models or [], proxy, timeout) if result.get("status") != "FOUNDRY" and route.get("foundry_route_probe") == "accepted": result = {"status": "FOUNDRY", "endpoint": endpoint, "model_count": 0, "models": [], "message": "route accepted without model-list support"} result.update(route) return result def is_azure_openai_llm(result): if result.get("status") != "VALID": return False if int(result.get("deployment_count") or 0) <= 0: return False route_probe = result.get("route_probe") if route_probe and route_probe != "accepted_auth_route": return False return True def append_azure_openai_llm(key, result, source): deployments = result.get("deployments") or [] deployment_text = ",".join(str(item) for item in deployments[:20]) details = " ".join(part for part in [ f"deployments={int(result.get('deployment_count') or 0)}", f"route_probe={result.get('route_probe') or ''}" if result.get("route_probe") else "", f"route_deployment={result.get('route_deployment') or ''}" if result.get("route_deployment") else "", f"route_model={result.get('route_model') or ''}" if result.get("route_model") else "", f"deployment_ids={deployment_text}" if deployment_text else "", ] if part) append_status(AZURE_OPENAI_LLM_FILE, key, result.get("status", "VALID"), details, source) def check_azure_acr(parsed, proxy, timeout): username = parsed["username"] password = parsed["password"] url = f"https://{username}.azurecr.io/v2/" try: response = requests.get(url, auth=(username, password), proxies=proxy, timeout=timeout) except requests.RequestException as exc: text = str(exc) if "no such host" in text.lower(): return {"status": "DEAD", "registry": username, "message": text} return {"status": "NETWORK", "registry": username, "message": text} if response.status_code == 200: return {"status": "VALID", "registry": username, "message": "ACR /v2 accepted basic auth"} message = request_error_message(response) if response.status_code == 401: return {"status": "DEAD", "registry": username, "http_status": 401, "message": message} if response.status_code == 403: return {"status": "RESTRICTED", "registry": username, "http_status": 403, "message": message} if response.status_code >= 500: return {"status": "NETWORK", "registry": username, "http_status": response.status_code, "message": message} return {"status": "UNKNOWN", "registry": username, "http_status": response.status_code, "message": message} def write_result(key, detector, result, source, finding): safe_finding = strip_finding_nearby_context(finding) write_keycheck_event(SERVICE, RESULTS_FILE, key, {"detector": detector, **result}, source, safe_finding, detector) commit_status_transaction( CHECKED_FILE, transaction_status_files(), key, result["status"], result.get("message", ""), source, ) if detector == "AzureOpenAI" and is_azure_openai_llm(result): append_azure_openai_llm(key, result, source) record_validation_result(SERVICE, key, {"detector": detector, **result}, source, safe_finding, detector) def strip_finding_nearby_context(finding): if not isinstance(finding, dict): return finding output = dict(finding) context = output.get("ScannerContext") if isinstance(context, dict) and "nearby" in context: output["ScannerContext"] = {key: value for key, value in context.items() if key != "nearby"} return output def parse_args(): parser = argparse.ArgumentParser(description="Azure key checker") parser.add_argument("--input", default=INPUT_FILE) parser.add_argument("--proxy-file", default=PROXY_FILE) parser.add_argument("--timeout", type=int, default=20) parser.add_argument("--max-keys", type=int, default=0) parser.add_argument("--retry-network", action="store_true") parser.add_argument("--retry-unknown", action="store_true") parser.add_argument("--retry-valid", action="store_true") parser.add_argument("--recheck-all", action="store_true") parser.add_argument("--probe-openai-route", action="store_true", help="Probe Azure OpenAI chat route with an invalid no-generation request after deployment listing succeeds") parser.add_argument("--probe-foundry-route", action="store_true", help="Probe Azure Foundry/MaaS chat route for configured models") parser.add_argument("--foundry-models", default="", help="Comma-separated Azure Foundry model IDs to route-probe") 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 = set() if args.retry_network: retry_statuses.update({"NETWORK", "FOUNDRY_BAD_ENDPOINT", "OPENAI_BAD_ENDPOINT"}) if args.retry_unknown: retry_statuses.update({"UNKNOWN", "FOUNDRY_UNRESOLVED", "OPENAI_UNRESOLVED"}) if args.retry_valid: retry_statuses.update({"VALID", "FOUNDRY"}) foundry_models = [item.strip() for item in str(args.foundry_models or "").split(",") if item.strip()] processed = 0 skipped = 0 for key, detector, source, finding, parsed in extract_candidates(args.input): if should_skip_key(key, checked, known, args, retry_statuses, service=SERVICE, source=source, finding=finding, detector=detector): skipped += 1 continue if args.max_keys and processed >= args.max_keys: break processed += 1 print(f"\n[{processed}] {detector} candidate {mask_secret(key)} from {source}") proxy = next(proxy_cycler) if proxy_cycler else None if parsed.get('_unresolved_candidate'): result = { 'status': ( 'FOUNDRY_UNRESOLVED' if detector == 'AzureFoundry' else 'OPENAI_UNRESOLVED' if detector == 'AzureOpenAI' else 'UNKNOWN' ), 'message': f"normalized {parsed.get('candidate_kind') or 'azure'} candidate is incomplete", } elif detector == "AzureOpenAI": result = check_azure_openai(parsed, proxy, args.timeout, args.probe_openai_route) elif detector == "AzureFoundry": result = check_azure_foundry(parsed, proxy, args.timeout, args.probe_foundry_route, foundry_models) if parsed.get("finding_uid"): result["finding_uid"] = parsed.get("finding_uid") elif detector == "AzureContainerRegistry": result = check_azure_acr(parsed, proxy, args.timeout) else: result = check_service_principal(parsed, proxy, args.timeout) print(f" STATUS: {result['status']} | {result.get('message', '')[:200]}") write_result(key, detector, result, source, finding) known.add(key) checked[key] = result["status"] print(f"\nDone. Processed={processed}, skipped={skipped}, results={RESULTS_FILE}") if __name__ == "__main__": main()