KeyVaultBrtueForce

KeyVaultBrtueForce.py

import argparse
import threading
import requests
import queue
import sys
from colorama import init, Fore

# Disable SSL warnings
requests.packages.urllib3.disable_warnings()

# Global config
API_VERSION = "7.3"
LOCK = threading.Lock()
init(autoreset=True)  # initialize colorama


def check_vault(token, vault_name):
    """Check if we can access the vault's secrets endpoint"""
    url = f"https://{vault_name}.vault.azure.net/secrets?api-version={API_VERSION}"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    try:
        resp = requests.get(url, headers=headers, timeout=5)
        if resp.status_code == 200 and "value" in resp.json():
            with LOCK:
                print(Fore.GREEN + f"[+] ACCESS: {vault_name} ({len(resp.json()['value'])} secrets)")
    except Exception:
        pass


def worker(token, q):
    while not q.empty():
        name = q.get()
        check_vault(token, name.strip())
        q.task_done()


def main():
    parser = argparse.ArgumentParser(description="Brute-force accessible Azure Key Vaults")
    parser.add_argument("--wordlist", required=True, help="Path to wordlist of vault names")
    parser.add_argument("--token", required=True, help="Access token for https://vault.azure.net")
    parser.add_argument("--threads", type=int, default=20, help="Number of threads (default: 20)")
    args = parser.parse_args()

    token = args.token

    try:
        with open(args.wordlist, "r") as f:
            names = [line.strip() for line in f if line.strip()]
    except Exception as e:
        print(f"[!] Error reading wordlist: {e}")
        sys.exit(1)

    q = queue.Queue()
    for name in names:
        q.put(name)

    threads = []
    for _ in range(args.threads):
        t = threading.Thread(target=worker, args=(token, q))
        t.daemon = True
        t.start()
        threads.append(t)

    q.join()
    print("[*] Done.")


if __name__ == "__main__":
    main()