Import documentation: README.md

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.

.github/workflows/pylint.yml

@@ -0,0 +1,23 @@
+name: Pylint
+
+on: [push]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.12"]
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install pylint
+    - name: Analysing the code with pylint
+      run: |
+        pylint $(git ls-files '*.py')

.pre-commit-config.yaml

@@ -0,0 +1,48 @@
+# See https://pre-commit.com for more information
+# See https://pre-commit.com/hooks.html for more hooks
+repos:
+    - repo: https://github.com/PyCQA/isort
+      rev: 5.13.2
+      hooks:
+          - id: isort
+            args: [--profile, black]
+
+    # Using this mirror lets us use mypyc-compiled black, which is about 2x faster
+    - repo: https://github.com/psf/black-pre-commit-mirror
+      rev: 24.4.2
+      hooks:
+          - id:
+                black
+                # It is recommended to specify the latest version of Python
+                # supported by your project here, or alternatively use
+                # pre-commit's default_language_version, see
+                # https://pre-commit.com/#top_level-default_language_version
+            language_version: python3.12
+            args: ["--line-length", "200", "--exclude", "migrations/"]
+
+    -   repo: https://github.com/astral-sh/ruff-pre-commit
+        rev: v0.0.285
+        hooks:
+        -   id: ruff
+            alias: autoformat
+            args: [--fix]
+
+    - repo: https://github.com/pycqa/flake8
+      rev: 7.1.0
+      hooks:
+          - id: flake8
+            exclude: ^tests/(data|examples)/
+
+    - repo: https://github.com/pre-commit/mirrors-mypy
+      rev: v1.10.1
+      hooks:
+          - id: mypy
+            args: [--ignore-missing-imports, --no-namespace-packages]
+
+    - repo: https://github.com/pre-commit/pre-commit-hooks
+      rev: v3.2.0
+      hooks:
+          - id: trailing-whitespace
+          - id: end-of-file-fixer
+          - id: check-yaml
+          - id: check-added-large-files

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+    "black-formatter.args": [
+        "--line-length",
+        "200"
+    ]
+}

CITATION.cff

@@ -0,0 +1,12 @@
+cff-version: 1.2.0
+message: "If you use this software, please cite it as below."
+authors:
+- family-names: "Hou"
+  given-names: "Junyi"
+  orcid: "https://orcid.org/0009-0003-0443-456X"
+
+title: "ChatGPT-API-Leakage"
+version: 1.5
+# doi: 10.5281/zenodo.1234
+date-released: 2024-02-21
+url: "https://github.com/Junyi-99/ChatGPT-API-Leakage"

README.md

@@ -0,0 +1,147 @@
+# ChatGPT-API-Scanner
+
+This tool scans GitHub for available OpenAI API Keys.
+
+![Result Demo 1](pics/demo.png)
+
+> [!WARNING]
+> **⚠️ DISCLAIMER**
+>
+> THIS PROJECT IS ONLY FOR ***SECURITY RESEARCH*** AND REMINDS OTHERS TO PROTECT THEIR PROPERTY, DO NOT USE IT ILLEGALLY!!
+>
+> The project authors are not responsible for any consequences resulting from misuse.
+
+> [!NOTE]
+> As of `August 21, 2024`, GitHub has enabled push protection to prevent API key leakage, which could significantly impact this repository.
+
+> [!NOTE]
+> As of `March 11, 2024`, secret scanning and push protection will be enabled by default for all new user-owned public repositories that you create.
+> Check this announcement [here](https://docs.github.com/en/code-security/getting-started/quickstart-for-securing-your-repository).
+
+## Keeping Your API Key Safe
+
+It's important to keep it safe to prevent unauthorized access. Here are some useful resources:
+
+- [Best Practices for API Key Safety](https://help.openai.com/en/articles/5112595-best-practices-for-api-key-safety)
+
+- [My API is getting leaked.. need advice!](https://community.openai.com/t/my-api-is-getting-leaked-need-advice/280564)
+
+- [My OpenAI API Key Leaked! What Should I Do?](https://www.gitguardian.com/remediation/openai-key)
+
+## Prerequisites
+
+This project has been tested and works perfectly on macOS, Windows and WSL2 (see [Run Linux GUI apps on the Windows Subsystem for Linux](https://learn.microsoft.com/en-us/windows/wsl/tutorials/gui-apps))
+
+Ensure you have the following installed on your system:
+
+- Google Chrome
+- [uv](https://docs.astral.sh/uv/) (Python package manager; handles Python install automatically)
+
+## Installation
+
+1. Clone the repository:
+
+    ```bash
+    git clone https://github.com/Junyi-99/ChatGPT-API-Scanner
+
+    cd ChatGPT-API-Scanner
+    ```
+
+2. Install dependencies (uv reads `pyproject.toml` / `uv.lock` and provisions the matching Python version from `.python-version`):
+
+    ```bash
+    uv sync
+    ```
+
+    To include dev tools (pylint, flake8, ruff):
+
+    ```bash
+    uv sync --all-groups
+    ```
+
+## Usage
+
+1. Run the main script:
+
+    ```bash
+    uv run main.py
+    ```
+
+2. You will be prompted to log in to your GitHub account in the browser. Please do so.
+
+That's it! The script will now scan GitHub for available OpenAI API Keys.
+
+## Command Line Arguments
+
+The script supports several command line arguments for customization:
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `--from-iter` | Start scanning from a specific iteration | `None` |
+| `--debug` | Enable debug mode for detailed logging | `False` |
+| `-ceko, --check-existed-keys-only` | Only check existing keys in the database | `False` |
+| `-k, --keywords` | Specify a list of search keywords | Default keyword list |
+| `-l, --languages` | Specify a list of programming languages to search | Default language list |
+
+Examples:
+
+```bash
+# Start scanning from iteration 100
+uv run main.py --from-iter 100
+
+# Only check existing keys
+uv run main.py --check-existed-keys-only
+
+# Use custom keywords and languages
+uv run main.py -k "openai" "chatgpt" -l python javascript
+```
+
+## Results
+
+The results are stored in the `github.db` SQLite database, which is created in the same directory as the script.
+
+You can view the contents of this database using any SQLite database browser of your choice.
+
+<figure>
+  <img
+  src="pics/demo2.png"
+  alt="Running Demo">
+  <p align="center">
+    Running Demo
+  </p>
+</figure>
+
+<figure>
+  <img
+  src="pics/db.png"
+  alt="Result in DB">
+  <p align="center">
+    Result stored in SQLite (different API Key status)
+  </p>
+</figure>
+
+## FAQ
+
+**Q: Why are you using Selenium instead of the GitHub Search API?**
+
+A: We use regex search to have the best search results. However, the official GitHub search API does not support regex search, only web-based search does.
+
+**Q: Why are you limiting the programming language in the search instead of searching all languages?**
+
+A: There are many API keys available. However, the web-based search only provides the first 5 pages of results. By limiting the language, we can break down the search results and obtain more keys.
+
+**Q: Why don't you use multithreading?**
+
+A: Because GitHub searches and OpenAI are rate-limited. Using multithreading does not significantly increase efficiency.
+
+**Q: Why is the API Key provided in your repository not working?**
+
+A: The screenshots in this repo demonstrate the tool's ability to scan for available API keys. However, these keys may expire within hours or days. Please use the tool to scan for your own keys instead of relying on the provided examples.
+
+**Q: What's the push protection?**
+
+A: Push protection is a feature provided by GitHub to prevent users from accidentally pushing sensitive information, such as API keys, to their repositories. When push protection is enabled, GitHub scans the code being pushed for known patterns of sensitive information and blocks the push if any are found. (see picture below)
+
+<p align="center">
+    <kbd><img src="pics/warning1.png" alt="GitHub Push Protection" width="400"> </kbd>
+</p>

main.py

@@ -0,0 +1,9 @@
+"""Entry point that delegates to the scanner package under src/."""
+
+import runpy
+import sys
+from pathlib import Path
+
+SRC = Path(__file__).parent / "src"
+sys.path.insert(0, str(SRC))
+runpy.run_path(str(SRC / "main.py"), run_name="__main__")

src/configs.py

@@ -0,0 +1,99 @@
+"""
+This module is used to store the configurations.
+"""
+
+import re
+
+# Keywords are not enabled by current version.
+KEYWORDS = [
+    "CoT",
+    "DPO",
+    "RLHF",
+    "agent",
+    "ai model",
+    "aios",
+    "api key",
+    "apikey",
+    "artificial intelligence",
+    "chain of thought",
+    "chatbot",
+    "chatgpt",
+    "competitor analysis",
+    "content strategy",
+    "conversational AI",
+    "data analysis",
+    "deep learning",
+    "direct preference optimization",
+    "experiment",
+    "gpt",
+    "gpt-3",
+    "gpt-4",
+    "gpt4",
+    "key",
+    "keyword clustering",
+    "keyword research",
+    "lab",
+    "language model experimentation",
+    "large language model",
+    "llama.cpp",
+    "llm",
+    "long-tail keywords",
+    "machine learning",
+    "multi-agent",
+    "multi-agent systems",
+    "natural language processing",
+    "openai",
+    "personalized AI",
+    "project",
+    "rag",
+    "reinforcement learning from human feedback",
+    "retrieval-augmented generation",
+    "search intent",
+    "semantic search",
+    "thoughts",
+    "virtual assistant",
+    "实验",
+    "密钥",
+    "测试",
+    "语言模型",
+]
+
+LANGUAGES = [
+    "Dotenv",
+    "Text",
+    "JavaScript",
+    "Python",
+    "TypeScript",
+    "Dockerfile",
+    "Markdown",
+    '"Jupyter Notebook"',
+    "Shell",
+    "Java",
+    "Go",
+    "C%2B%2B",
+    "PHP",
+]
+
+PATHS = [
+    "path:.xml OR path:.json OR path:.properties OR path:.sql OR path:.txt OR path:.log OR path:.tmp OR path:.backup OR path:.bak OR path:.enc",
+    "path:.yml OR path:.yaml OR path:.toml OR path:.ini OR path:.config OR path:.conf OR path:.cfg OR path:.env OR path:.envrc OR path:.prod",
+    "path:.secret OR path:.private OR path:*.key",
+]
+
+# regex, have_many_results (if true, using AND to filter out keywords), result_too_lang (if true, the result needs to be expanded)
+REGEX_LIST = [
+    # Named Project API Key (no matter normal or restricted) still valid until Dec 2, 2024
+    (re.compile(r"sk-proj-[A-Za-z0-9-_]{74}T3BlbkFJ[A-Za-z0-9-_]{73}A"), True, True),
+
+    # Service Account key (new format, valid until Oct 13, 2025)
+    (re.compile(r"sk-svcacct-[A-Za-z0-9-_]{74}T3BlbkFJ[A-Za-z0-9-_]{73}A"), False, True),
+
+    # Old Project API Key (as of Oct 13, 2025, not sure if still valid)
+    (re.compile(r"sk-proj-[A-Za-z0-9-_]{58}T3BlbkFJ[A-Za-z0-9-_]{58}"), True, True),
+
+    # Service Account Key (not valid since Oct 13, 2025)
+    # (re.compile(r"sk-svcacct-[A-Za-z0-9-_]\+T3BlbkFJ[A-Za-z0-9-_]+"), False, False), # No search results on Oct 14, 2025
+
+    (re.compile(r"sk-proj-[A-Za-z0-9]{20}T3BlbkFJ[A-Za-z0-9]{20}"), True, False),
+    # (re.compile(r"sk-[a-zA-Z0-9]{48}"), True, False), # OpenAI deprecated these format, because the keys are only "sk-proj" "sk-svcacct-"
+]

src/main.py

@@ -0,0 +1,344 @@
+"""
+Scan GitHub for available OpenAI API Keys
+"""
+
+import argparse
+import logging
+import os
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import rich
+from selenium import webdriver
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.support.ui import WebDriverWait
+from tqdm import tqdm
+
+from configs import KEYWORDS, LANGUAGES, PATHS, REGEX_LIST
+from manager import CookieManager, DatabaseManager, ProgressManager
+from utils import check_key
+
+FORMAT = "%(message)s"
+logging.basicConfig(level=logging.INFO, format=FORMAT, datefmt="[%X]")
+log = logging.getLogger("ChatGPT-API-Leakage")
+httpx_logger = logging.getLogger("httpx")
+httpx_logger.setLevel(logging.WARNING)
+
+
+class APIKeyLeakageScanner:
+    """
+    Scan GitHub for available OpenAI API Keys
+    """
+
+    def __init__(self, db_file: str, keywords: list, languages: list):
+        self.db_file = db_file
+        self.driver: webdriver.Chrome | None = None
+        self.cookies: CookieManager | None = None
+        rich.print(f"📂 Opening database file {self.db_file}")
+
+        self.dbmgr = DatabaseManager(self.db_file)
+
+        self.keywords = keywords
+        self.languages = languages
+        self.candidate_urls = []
+        for regex, too_many_results, _ in REGEX_LIST:
+            # Add the paths to the search query
+            for path in PATHS:
+                self.candidate_urls.append(f"https://github.com/search?q=(/{regex.pattern}/)+AND+({path})&type=code&ref=advsearch")
+
+            for language in self.languages:
+                if too_many_results:  # if the regex is too many results, then we need to add AND condition
+                    self.candidate_urls.append(f"https://github.com/search?q=(/{regex.pattern}/)+language:{language}&type=code&ref=advsearch")
+                else:  # if the regex is not too many results, then we just need the regex
+                    self.candidate_urls.append(f"https://github.com/search?q=(/{regex.pattern}/)&type=code&ref=advsearch")
+
+    def login_to_github(self):
+        """
+        Login to GitHub
+        """
+        rich.print("🌍 Opening Chrome ...")
+
+        options = webdriver.ChromeOptions()
+        options.add_argument("--ignore-certificate-errors")
+        options.add_argument("--ignore-ssl-errors")
+
+        self.driver = webdriver.Chrome(options=options)
+        self.driver.implicitly_wait(3)
+
+        self.cookies = CookieManager(self.driver)
+
+        cookie_exists = os.path.exists("cookies.pkl")
+        self.driver.get("https://github.com/login")
+
+        if not cookie_exists:
+            rich.print("🤗 No cookies found, please login to GitHub first")
+            input("Press Enter after you logged in: ")
+            self.cookies.save()
+        else:
+            rich.print("🍪 Cookies found, loading cookies")
+            self.cookies.load()
+
+        self.cookies.verify_user_login()
+
+    def _expand_all_code(self):
+        """
+        Expand all the code in the current page
+        """
+        elements = self.driver.find_elements(by=By.XPATH, value="//*[contains(text(), 'more match')]")
+        for element in elements:
+            element.click()
+
+    def _find_urls_and_apis(self) -> tuple[list[str], list[str]]:
+        """
+        Find all the urls and apis in the current page
+        """
+        apis_found = []
+        urls_need_expand = []
+
+        codes = self.driver.find_elements(by=By.CLASS_NAME, value="code-list")  # type: ignore
+        for element in codes:
+            apis = []
+            # Check all regex for each code block
+            for regex, _, too_long in REGEX_LIST[2:]:
+                if not too_long:
+                    apis.extend(regex.findall(element.text))
+
+            if len(apis) == 0:
+                # Need to show full code. (because the api key is too long)
+                # get the <a> tag
+                a_tag = element.find_element(by=By.XPATH, value=".//a")
+                urls_need_expand.append(a_tag.get_attribute("href"))
+            apis_found.extend(apis)
+
+        return apis_found, urls_need_expand
+
+    def _process_url(self, url: str):
+        """
+        Process a search query url
+        """
+        if self.driver is None:
+            raise ValueError("Driver is not initialized")
+
+        self.driver.get(url)
+
+        while True:  # Loop until all the pages are processed
+            # If current webpage is reached the rate limit, then wait for 30 seconds
+            if self.driver.find_elements(by=By.XPATH, value="//*[contains(text(), 'You have exceeded a secondary rate limit')]"):
+                for _ in tqdm(range(30), desc="⏳ Rate limit reached, waiting ..."):
+                    time.sleep(1)
+                self.driver.refresh()
+                continue
+
+            self._expand_all_code()
+
+            apis_found, urls_need_expand = self._find_urls_and_apis()
+            rich.print(f"    🌕 There are {len(urls_need_expand)} urls waiting to be expanded")
+
+            try:
+                next_buttons = self.driver.find_elements(by=By.XPATH, value="//a[@aria-label='Next Page']")
+                rich.print("🔍 Clicking next page")
+                WebDriverWait(self.driver, 5).until(EC.presence_of_element_located((By.XPATH, "//a[@aria-label='Next Page']")))
+                next_buttons = self.driver.find_elements(by=By.XPATH, value="//a[@aria-label='Next Page']")
+                next_buttons[0].click()
+            except Exception:  # pylint: disable=broad-except
+                rich.print("⚪️ No more pages")
+                break
+
+        # Handle the expand_urls
+        for u in tqdm(urls_need_expand, desc="🔍 Expanding URLs ..."):
+            if self.driver is None:
+                raise ValueError("Driver is not initialized")
+
+            with self.dbmgr as mgr:
+                if mgr.get_url(u):
+                    rich.print(f"    🔑 skipping url '{u[:10]}...{u[-10:]}'")
+                    continue
+
+            self.driver.get(u)
+            time.sleep(3)  # TODO: find a better way to wait for the page to load # pylint: disable=fixme
+
+            retry = 0
+            while retry <= 3:
+                matches = []
+                for regex, _, _ in REGEX_LIST:
+                    matches.extend(regex.findall(self.driver.page_source))
+                matches = list(set(matches))
+
+                if len(matches) == 0:
+                    rich.print(f"    ⚪️ No matches found in the expanded page, retrying [{retry}/3]...")
+                    retry += 1
+                    time.sleep(3)
+                    continue
+
+                with self.dbmgr as mgr:
+                    new_apis = [api for api in matches if not mgr.key_exists(api)]
+                    new_apis = list(set(new_apis))
+                apis_found.extend(new_apis)
+                rich.print(f"    🔬 Found {len(matches)} matches in the expanded page, adding them to the list")
+                for match in matches:
+                    rich.print(f"        '{match[:10]}...{match[-10:]}'")
+
+                with self.dbmgr as mgr:
+                    mgr.insert_url(url)
+                break
+
+        self.check_api_keys_and_save(apis_found)
+
+    def check_api_keys_and_save(self, keys: list[str]):
+        """
+        Check a list of API keys
+        """
+        with self.dbmgr as mgr:
+            unique_keys = list(set(keys))
+            unique_keys = [api for api in unique_keys if not mgr.key_exists(api)]
+
+        with ThreadPoolExecutor(max_workers=10) as executor:
+            results = list(executor.map(check_key, unique_keys))
+            with self.dbmgr as mgr:
+                for idx, result in enumerate(results):
+                    mgr.insert(unique_keys[idx], result)
+
+    def search(self, from_iter: int | None = None):
+        """
+        Search for API keys, and save the results to the database
+        """
+        progress = ProgressManager()
+        total = len(self.candidate_urls)
+        pbar = tqdm(
+            enumerate(self.candidate_urls),
+            total=total,
+            desc="🔍 Searching ...",
+        )
+        if from_iter is None:
+            from_iter = progress.load(total=total)
+
+        for idx, url in enumerate(self.candidate_urls):
+            if idx < from_iter:
+                pbar.update()
+                time.sleep(0.05)  # let tqdm print the bar
+                log.debug("⚪️ Skip %s", url)
+                continue
+            self._process_url(url)
+            progress.save(idx, total)
+            log.debug("🔍 Finished %s", url)
+            pbar.update()
+        pbar.close()
+
+    def deduplication(self):
+        """
+        Deduplicate the database
+        """
+        with self.dbmgr as mgr:
+            mgr.deduplicate()
+
+    def update_existed_keys(self):
+        """
+        Update previously checked API keys in the database with their current status
+        """
+        with self.dbmgr as mgr:
+            rich.print("🔄 Updating existed keys")
+            keys = mgr.all_keys()
+            for key in tqdm(keys, desc="🔄 Updating existed keys ..."):
+                result = check_key(key[0])
+                mgr.delete(key[0])
+                mgr.insert(key[0], result)
+
+    def update_iq_keys(self):
+        """
+        Update insuffcient quota keys
+        """
+        with self.dbmgr as mgr:
+            rich.print("🔄 Updating insuffcient quota keys")
+            keys = mgr.all_iq_keys()
+            for key in tqdm(keys, desc="🔄 Updating insuffcient quota keys ..."):
+                result = check_key(key[0])
+                mgr.delete(key[0])
+                mgr.insert(key[0], result)
+
+    def all_available_keys(self) -> list:
+        """
+        Get all available keys
+        """
+        with self.dbmgr as mgr:
+            return mgr.all_keys()
+
+    def __del__(self):
+        if hasattr(self, "driver") and self.driver is not None:
+            self.driver.quit()
+
+
+def main(from_iter: int | None = None, check_existed_keys_only: bool = False, keywords: list | None = None, languages: list | None = None, check_insuffcient_quota: bool = False):
+    """
+    Main function to scan GitHub for available OpenAI API Keys
+    """
+    keywords = KEYWORDS.copy() if keywords is None else keywords
+    languages = LANGUAGES.copy() if languages is None else languages
+
+    leakage = APIKeyLeakageScanner("github.db", keywords, languages)
+
+    if not check_existed_keys_only:
+        leakage.login_to_github()
+        leakage.search(from_iter=from_iter)
+
+    if check_insuffcient_quota:
+        leakage.update_iq_keys()
+
+    leakage.update_existed_keys()
+    leakage.deduplication()
+    keys = leakage.all_available_keys()
+
+    rich.print(f"🔑 [bold green]Available keys ({len(keys)}):[/bold green]")
+    for key in keys:
+        rich.print(f"[bold green]{key[0]}[/bold green]")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--from-iter", type=int, default=None, help="Start from the specific iteration")
+    parser.add_argument(
+        "--debug",
+        action="store_true",
+        default=False,
+        help="Enable debug mode, otherwise INFO mode. Default is False (INFO mode)",
+    )
+    parser.add_argument(
+        "-ceko",
+        "--check-existed-keys-only",
+        action="store_true",
+        default=False,
+        help="Only check existed keys",
+    )
+    parser.add_argument(
+        "-ciq",
+        "--check-insuffcient-quota",
+        action="store_true",
+        default=False,
+        help="Check and update status of the insuffcient quota keys",
+    )
+    parser.add_argument(
+        "-k",
+        "--keywords",
+        nargs="+",
+        default=KEYWORDS,
+        help="Keywords to search",
+    )
+    parser.add_argument(
+        "-l",
+        "--languages",
+        nargs="+",
+        default=LANGUAGES,
+        help="Languages to search",
+    )
+    args = parser.parse_args()
+
+    if args.debug:
+        logging.getLogger().setLevel(logging.DEBUG)
+
+    main(
+        from_iter=args.from_iter,
+        check_existed_keys_only=args.check_existed_keys_only,
+        keywords=args.keywords,
+        languages=args.languages,
+        check_insuffcient_quota=args.check_insuffcient_quota,
+    )

src/manager.py

@@ -0,0 +1,309 @@
+"""
+Progress and Cookie Management Module
+
+This module provides functionality for managing application progress, cookies,
+and database operations.
+
+Classes:
+    ProgressManager: Handles progress tracking and persistence
+    CookieManager: Manages browser cookie operations
+    DatabaseManager: Handles database interactions
+"""
+
+import logging
+import os
+import pickle
+import sqlite3
+import sys
+import time
+from datetime import date
+
+from selenium.common.exceptions import UnableToSetCookieException
+from selenium.webdriver.common.by import By
+
+LOGGER_NAME = "ChatGPT-API-Leakage"
+LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, datefmt="[%X]")
+logger = logging.getLogger(LOGGER_NAME)
+
+
+class ProgressManagerError(Exception):
+    """Custom exception for ProgressManager class errors"""
+
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class ProgressManager:
+    """
+    Manages and persists progress information for long-running operations.
+
+    Attributes:
+        progress_file (Path): Path to the progress file
+
+    Methods:
+        save: Saves current progress
+        load: Loads saved progress
+    """
+
+    def __init__(self, progress_file=".progress.txt"):
+        """
+        Initialize the ProgressManager with a specified progress file.
+
+        Args:
+            progress_file (str): The file where progress data is stored.
+        """
+        self.progress_file = progress_file
+
+    def save(self, from_iter: int, total: int):
+        """
+        Saves the current progress to a file.
+
+        Args:
+            from_iter (int): The current iteration progress.
+            total (int): The total number of iterations.
+        """
+        with open(self.progress_file, "w", encoding="utf-8") as file:
+            file.write(f"{from_iter}/{total}/{time.time()}")
+
+    def load(self, total: int) -> int:
+        """
+        Loads the previously saved progress if available and valid.
+
+        Args:
+            total (int): The total number of iterations for the current process.
+
+        Returns:
+            int: The iteration number to continue from.
+        """
+        if not os.path.exists(self.progress_file):
+            return 0
+
+        with open(self.progress_file, "r", encoding="utf-8") as file:
+            last_, totl_, tmst_ = file.read().strip().split("/")
+            last, totl = int(last_), int(totl_)
+
+        if time.time() - float(tmst_) < 3600 and totl == total:
+            action = input(f"🔍 Progress found, do you want to continue from the last progress ({last}/{totl})? [yes] | no: ").lower()
+            if action in {"yes", "y", ""}:
+                return last
+
+        return 0
+
+
+class CookieManager:
+    """
+    Manages browser cookie operations.
+
+    Methods:
+        save: Saves cookies to a file
+        load: Loads cookies from a file
+        verify_user_login: Checks if the user is currently logged in
+    """
+
+    def __init__(self, driver):
+        """
+        Initialize the CookieManager with a Selenium WebDriver instance.
+
+        Args:
+            driver (WebDriver): The Selenium WebDriver for cookie operations.
+        """
+        self.driver = driver
+
+    def save(self):
+        """
+        Save cookies from the current browser session to a file.
+        """
+        cookies = self.driver.get_cookies()
+        with open("cookies.pkl", "wb") as file:
+            pickle.dump(cookies, file)
+            logger.info("🍪 Cookies saved")
+
+    def load(self):
+        """
+        Load cookies from a file and attempt to add them to the current browser session.
+        """
+        try:
+            with open("cookies.pkl", "rb") as file:
+                cookies = pickle.load(file)
+                for cookie in cookies:
+                    try:
+                        self.driver.add_cookie(cookie)
+                    except UnableToSetCookieException:
+                        logger.debug("🟡 Warning, unable to set a cookie %s", cookie)
+        except (EOFError, pickle.UnpicklingError):
+            if os.path.exists("cookies.pkl"):
+                os.remove("cookies.pkl")
+            logger.error("🔴 Error, unable to load cookies, invalid cookies has been removed, please restart.")
+
+    def verify_user_login(self):
+        """
+        Test if the user is really logged in by navigating to GitHub and checking login status.
+        """
+        logger.info("🤗 Redirecting ...")
+        self.driver.get("https://github.com/")
+
+        if self.driver.find_elements(by=By.XPATH, value="//*[contains(text(), 'Sign in')]"):
+            if os.path.exists("cookies.pkl"):
+                os.remove("cookies.pkl")
+            logger.error("🔴 Error, you are not logged in, please restart and try again.")
+            sys.exit(1)
+        return True
+
+
+class DatabaseManager:
+    """
+    This class is used to manage the database, including creating tables and handling data interactions.
+    """
+
+    def __init__(self, db_filename: str):
+        """
+        Initialize the DatabaseManager with the specified database filename.
+
+        Args:
+            db_filename (str): Path to the SQLite database file.
+        """
+        self.db_filename = db_filename
+        self.con = None
+        self.cur = None
+
+    def __enter__(self):
+        """
+        Enter the runtime context related to this object, initializing the database if needed.
+        """
+        if not os.path.exists(self.db_filename):
+            logging.info("Creating database github.db")
+
+        self.con = sqlite3.connect(self.db_filename)
+        self.cur = self.con.cursor()
+
+        self.cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='APIKeys'")
+        if self.cur.fetchone() is None:
+            logging.info("Creating table APIKeys")
+            self.cur.execute("CREATE TABLE APIKeys(apiKey, status, lastChecked)")
+
+        self.cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='URLs'")
+        if self.cur.fetchone() is None:
+            logging.info("Creating table URLs")
+            self.cur.execute("CREATE TABLE URLs(url, key)")
+
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """
+        Exit the runtime context and close the database connection.
+        """
+        if self.con:
+            self.con.close()
+
+    def all_iq_keys(self) -> list:
+        """
+        Get all keys with the status 'insufficient_quota'.
+
+        Returns:
+            list: A list of tuples containing API keys.
+        """
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        self.cur.execute("SELECT apiKey FROM APIKeys WHERE status='insufficient_quota'")
+        return self.cur.fetchall()
+
+    def all_keys(self) -> list:
+        """
+        Get all keys with the status 'yes'.
+
+        Returns:
+            list: A list of tuples containing API keys.
+        """
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        self.cur.execute("SELECT apiKey FROM APIKeys WHERE status='yes'")
+        return self.cur.fetchall()
+
+    def deduplicate(self) -> None:
+        """
+        Deduplicate the 'APIKeys' table by retaining only the latest record for each key.
+        """
+        if self.con is None:
+            raise ValueError("Connection is not initialized")
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        self.cur.execute("CREATE TABLE temp_table as SELECT apiKey, status, MAX(lastChecked) as lastChecked FROM APIKeys GROUP BY apiKey;")
+        self.cur.execute("DROP TABLE APIKeys;")
+        self.cur.execute("ALTER TABLE temp_table RENAME TO APIKeys;")
+        self.con.commit()
+
+    def delete(self, api_key: str) -> None:
+        """
+        Delete a specific API key from the database.
+
+        Args:
+            api_key (str): The unique API key to remove.
+        """
+        if self.con is None:
+            raise ValueError("Connection is not initialized")
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        self.cur.execute("DELETE FROM APIKeys WHERE apiKey=?", (api_key,))
+        self.con.commit()
+
+    def insert(self, api_key: str, status: str):
+        """
+        Insert a new API key and status into the database.
+
+        Args:
+            api_key (str): The API key to insert.
+            status (str): The status of the API key.
+        """
+        if self.con is None:
+            raise ValueError("Connection is not initialized")
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        today = date.today()
+        self.cur.execute("INSERT INTO APIKeys(apiKey, status, lastChecked) VALUES(?, ?, ?)", (api_key, status, today))
+        self.con.commit()
+
+    def key_exists(self, api_key: str) -> bool:
+        """
+        Check if a given API key exists in the database.
+
+        Args:
+            api_key (str): The API key to search for.
+
+        Returns:
+            bool: True if the API key exists, False otherwise.
+        """
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        self.cur.execute("SELECT apiKey FROM APIKeys WHERE apiKey=?", (api_key,))
+        return self.cur.fetchone() is not None
+
+    def insert_url(self, url: str) -> None:
+        """
+        Insert a new URL into the 'URLs' table.
+
+        Args:
+            url (str): The URL to add.
+        """
+        if self.con is None:
+            raise ValueError("Connection is not initialized")
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        self.cur.execute("INSERT INTO URLs(url, key) VALUES(?, ?)", (url, 1))
+        self.con.commit()
+
+    def get_url(self, url: str) -> str | None:
+        """
+        Retrieve the 'key' associated with the given URL.
+
+        Args:
+            url (str): The URL to look up.
+
+        Returns:
+            str | None: The key if it exists, None if not.
+        """
+        if self.cur is None:
+            raise ValueError("Cursor is not initialized")
+        self.cur.execute("SELECT key FROM URLs WHERE url=?", (url,))
+        fetch = self.cur.fetchone()
+        return fetch[0] if fetch else None

src/utils.py

@@ -0,0 +1,45 @@
+"""
+This module provides a function to check the validity of an OpenAI API key by making a test request
+to a chosen model using the OpenAI client. It captures various exceptions and prints error details.
+"""
+
+import rich
+from openai import APIStatusError, AuthenticationError, OpenAI, RateLimitError
+
+
+def check_key(key, model="gpt-4o-mini") -> str | None:
+    """
+    Check if the API key is valid.
+    """
+    try:
+        client = OpenAI(api_key=key)
+
+        completion = client.chat.completions.create(
+            model=model,
+            messages=[
+                {
+                    "role": "system",
+                    "content": "You are a yeser, you only output lowercase yes.",
+                },
+                {"role": "user", "content": "yes or no? say yes"},
+            ],
+        )
+        result = completion.choices[0].message.content
+        rich.print(f"🔑 [bold green]available key[/bold green]: [orange_red1]'{key}'[/orange_red1] ({result})\n")
+        return "yes"
+    except AuthenticationError as e:
+        rich.print(f"[deep_sky_blue1]{e.body['code']} ({e.status_code})[/deep_sky_blue1]: '{key[:10]}...{key[-10:]}'")  # type: ignore
+        return e.body["code"]  # type: ignore
+    except RateLimitError as e:
+        rich.print(f"[deep_sky_blue1]{e.body['code']} ({e.status_code})[/deep_sky_blue1]: '{key[:10]}...{key[-10:]}'")  # type: ignore
+        return e.body["code"]  # type: ignore
+    except APIStatusError as e:
+        rich.print(f"[bold red]{e.body['code']} ({e.status_code})[/bold red]: '{key[:10]}...{key[-10:]}'")  # type: ignore
+        return e.body["code"]  # type: ignore
+    except Exception as e:  # pylint: disable=broad-except
+        rich.print(f"[bold red]{e}[/bold red]: '{key[:10]}...{key[-10:]}'")  # type: ignore
+        return "Unknown Error"
+
+
+if __name__ == "__main__":
+    check_key("sk-proj-12345")