Janady13/AI_Brain / config/stripe_secrets.py

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/AI_Brain / config/stripe_secrets.py
Date	Not specified in source
Repository	Janady13/AI_Brain
Revision	cdb497ece2d1a770ca8c83eb1de8923cfaa9eceb
Branch	feature/seeder-engine-tests
Path or archive member	config/stripe_secrets.py
Original SHA-256	1b0e6325890fbacd41a7c6c9b7b58edae2c6c6ff1d93005bb4d79fe9a2dedcde
Git blob SHA	3b36727a1671e703c7d23b79d929f8dbe91a84bd
Reading-copy SHA-256	9d2827f74ab08d8745f7f4441e05a4a1b4826eab67ad2c016f58d69c7c3451b7

Original source ↗

Recovered source transcript

Download complete source text ↓134 lines · 5.0 KB

Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.

#!/usr/bin/env python3
from __future__ import annotations

import os
import re
import sqlite3
from typing import Dict, Optional, Tuple, List


_KEY_NAMES = (
    "STRIPE_SECRET_KEY",
    "STRIPE_API_KEY",
    "stripe_secret_key",
    "api_key",
)
_WEBHOOK_NAMES = (
    "STRIPE_WEBHOOK_SECRET",
    "webhook_secret",
    "stripe_webhook_secret",
)


def _looks_like_api_key(val: str) -> bool:
    return isinstance(val, str) and (val.startswith("sk_live_") or val.startswith("sk_test_"))


def _looks_like_webhook(val: str) -> bool:
    return isinstance(val, str) and val.startswith("whsec_")


def _read_sqlite(db_path: str) -> Tuple[Optional[str], Optional[str]]:
    try:
        con = sqlite3.connect(db_path)
        cur = con.cursor()
        cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = [r[0] for r in cur.fetchall()]

        api_key: [credential removed] = None
        whsec: Optional[str] = None

        # 1) canonical secrets table: key/value
        for t in tables:
            try:
                cols = [r[1] for r in cur.execute(f"PRAGMA table_info({t})").fetchall()]
            except Exception:
                continue
            lower_cols = [c.lower() for c in cols]
            if {"key", "value"} <= set(lower_cols):
                try:
                    for key_name in _KEY_NAMES:
                        row = cur.execute(
                            f"SELECT value FROM {t} WHERE key=? ORDER BY ROWID DESC LIMIT 1",
                            (key_name,),
                        ).fetchone()
                        if row and _looks_like_api_key(str(row[0])):
                            api_key = [credential removed]; break
                    for wh_name in _WEBHOOK_NAMES:
                        row = cur.execute(
                            f"SELECT value FROM {t} WHERE key=? ORDER BY ROWID DESC LIMIT 1",
                            (wh_name,),
                        ).fetchone()
                        if row and _looks_like_webhook(str(row[0])):
                            whsec = str(row[0]); break
                except Exception:
                    pass
            # 2) wide tables with api_key/webhook_secret columns
            key_cols = [c for c in cols if c.lower() in {n.lower() for n in _KEY_NAMES}]
            wh_cols = [c for c in cols if c.lower() in {n.lower() for n in _WEBHOOK_NAMES}]
            if key_cols:
                try:
                    row = cur.execute(f"SELECT {key_cols[0]} FROM {t} ORDER BY ROWID DESC LIMIT 1").fetchone()
                    if row and _looks_like_api_key(str(row[0])):
                        api_key = [credential removed] or str(row[0])
                except Exception:
                    pass
            if wh_cols:
                try:
                    row = cur.execute(f"SELECT {wh_cols[0]} FROM {t} ORDER BY ROWID DESC LIMIT 1").fetchone()
                    if row and _looks_like_webhook(str(row[0])):
                        whsec = whsec or str(row[0])
                except Exception:
                    pass
            # 3) heuristic scan: find any column value that looks like keys
            if api_key is None or whsec is None:
                try:
                    row = cur.execute(f"SELECT * FROM {t} ORDER BY ROWID DESC LIMIT 5").fetchall()
                    for r in row:
                        for v in r:
                            if isinstance(v, str):
                                if api_key is None and _looks_like_api_key(v):
                                    api_key = v
                                if whsec is None and _looks_like_webhook(v):
                                    whsec = v
                except Exception:
                    pass

        con.close()
        return api_key, whsec
    except Exception:
        return None, None


def apply_env_from_stripe_db(db_path: Optional[str] = None, force: bool = False) -> Dict[str, str]:
    """Load Stripe keys from a local sqlite DB and apply to env.
    Returns a mapping of keys set. Does not overwrite existing env unless force=True.
    """
    result: Dict[str, str] = {}
    # Respect existing env unless forced
    if not force and os.getenv("STRIPE_SECRET_KEY") and os.getenv("STRIPE_WEBHOOK_SECRET"):
        return result

    db_candidates: List[str] = []
    if db_path:
        db_candidates.append(os.path.expanduser(db_path))
    env_path = os.getenv("STRIPE_DB_PATH")
    if env_path:
        db_candidates.append(os.path.expanduser(env_path))
    db_candidates.append(os.path.expanduser('~/AI_Brain/db/stripe.db'))

    for path in db_candidates:
        if not os.path.isfile(path):
            continue
        api_key, whsec = _read_sqlite(path)
        if api_key and (force or not os.getenv("STRIPE_SECRET_KEY")):
            os.environ["STRIPE_SECRET_KEY"] = api_key
            result["STRIPE_SECRET_KEY"] = api_key
        if whsec and (force or not os.getenv("STRIPE_WEBHOOK_SECRET")):
            os.environ["STRIPE_WEBHOOK_SECRET"] = whsec
            result["STRIPE_WEBHOOK_SECRET"] = whsec
        if result:
            break
    return result