如何实现无硬编码、不暴露凭证的Python数据库连接认证？

以下是几种完全避免数据库凭证落地存储（包括文件、.env等）的可行方案，结合你的异步Python客户端场景给出具体实现：

方案1：运行时交互式输入（桌面端首选） #

启动程序时让用户手动输入数据库凭证，仅在内存中临时持有，连接完成后立即销毁，全程不写入任何文件。

修改你的连接逻辑：

import getpass
import asyncpg
import logging
import asyncio

# 程序启动时同步获取凭证，仅存于内存
def get_db_credentials():
    user = input("请输入数据库用户名: ")
    password = getpass.getpass("请输入数据库密码: ")
    host = input("请输入数据库主机地址: ")
    port = input("请输入数据库端口: ")
    database = input("请输入数据库名称: ")
    return user, password, host, port, database

async def create_db_connection():
    # 按需获取凭证
    user, password, host, port, database = get_db_credentials()
    while True:
        try:
            conn = await asyncpg.connect(
                user=user,
                password=password,
                host=host,
                port=port,
                database=database
            )
            # 连接成功后立即销毁内存中的凭证
            del user, password, host, port, database
            return conn
        except Exception as e:
            logging.error(f"\n\tError connecting to the database: {e}")
            await asyncio.sleep(0.7)

async def log_off_database(token):
    while True:
        try:
            conn = await create_db_connection()
            await conn.execute(
                'UPDATE "Tokens" SET "Status" = $1, "UsedBy" = $2 WHERE "Token" = $3',
                'Free', '-', token
            )
        except Exception as e:
            logging.error(f"Error logging off: {e}")
        finally:
            await conn.close()
        await asyncio.sleep(9)

优点：完全无落地存储，实现简单；缺点：每次启动需手动输入，适合个人或小量用户。

方案2：调用系统密钥管理服务（桌面端自动化场景） #

利用Windows凭证管理器、macOS钥匙串、Linux秘密服务等系统内置安全存储保存凭证，程序运行时临时读取到内存，不落地到自定义文件。

使用keyring库实现跨平台调用：
先安装依赖：pip install keyring

修改代码：

import keyring
import asyncpg
import logging
import asyncio

# 首次配置时执行（仅一次）：将凭证存入系统密钥管理
# keyring.set_password("my_db_client", "db_user", "你的用户名")
# keyring.set_password("my_db_client", "db_pass", "你的密码")
# keyring.set_password("my_db_client", "db_host", "你的主机地址")
# keyring.set_password("my_db_client", "db_port", "你的端口")
# keyring.set_password("my_db_client", "db_name", "你的数据库名")

async def create_db_connection():
    while True:
        try:
            # 运行时从系统安全存储临时读取
            user = keyring.get_password("my_db_client", "db_user")
            password = keyring.get_password("my_db_client", "db_pass")
            host = keyring.get_password("my_db_client", "db_host")
            port = keyring.get_password("my_db_client", "db_port")
            database = keyring.get_password("my_db_client", "db_name")
            
            conn = await asyncpg.connect(
                user=user,
                password=password,
                host=host,
                port=port,
                database=database
            )
            # 销毁内存中的凭证
            del user, password, host, port, database
            return conn
        except Exception as e:
            logging.error(f"\n\tError connecting to the database: {e}")
            await asyncio.sleep(0.7)

async def log_off_database(token):
    while True:
        try:
            conn = await create_db_connection()
            await conn.execute(
                'UPDATE "Tokens" SET "Status" = $1, "UsedBy" = $2 WHERE "Token" = $3',
                'Free', '-', token
            )
        except Exception as e:
            logging.error(f"Error logging off: {e}")
        finally:
            await conn.close()
        await asyncio.sleep(9)

优点：无需手动输入，凭证存储在系统安全容器中；缺点：首次需手动配置凭证。

方案3：集中身份认证服务（企业级场景） #

完全避开直接使用数据库凭证，通过OAuth2/OIDC或企业SSO获取短期数据库访问token，再用该token连接数据库（需数据库支持临时token认证，如PostgreSQL的pgtoken插件、云数据库IAM认证）。

核心流程代码示例：

import aiohttp
import asyncpg
import logging
import asyncio

async def get_temp_db_token():
    # 对接企业认证服务，获取短期数据库凭证
    async with aiohttp.ClientSession() as session:
        async with session.post("https://your-auth-service.com/get-db-token", 
                               json={"user_id": "当前用户ID", "scope": "db_access"}) as resp:
            data = await resp.json()
            return data["db_user"], data["db_token"], data["db_host"], data["db_port"], data["db_name"]

async def create_db_connection():
    while True:
        try:
            # 获取临时数据库凭证（有效期通常为1-24小时）
            user, token, host, port, database = await get_temp_db_token()
            # 用临时token连接数据库
            conn = await asyncpg.connect(
                user=user,
                password=token,
                host=host,
                port=port,
                database=database
            )
            # 销毁内存中的临时凭证
            del user, token, host, port, database
            return conn
        except Exception as e:
            logging.error(f"\n\tError connecting to the database: {e}")
            await asyncio.sleep(0.7)

async def log_off_database(token):
    while True:
        try:
            conn = await create_db_connection()
            await conn.execute(
                'UPDATE "Tokens" SET "Status" = $1, "UsedBy" = $2 WHERE "Token" = $3',
                'Free', '-', token
            )
        except Exception as e:
            logging.error(f"Error logging off: {e}")
        finally:
            await conn.close()
        await asyncio.sleep(9)

优点：完全不存储长期数据库凭证，安全性最高；缺点：依赖外部认证服务，数据库需支持临时token认证。

方案4：临时环境变量注入（服务器端脚本场景） #

启动程序时通过命令行临时传递环境变量，程序读取后立即清除，不写入任何文件：

启动命令示例 #

Linux/macOS：

DB_USER=你的用户名 DB_PASS=你的密码 DB_HOST=你的主机 DB_PORT=5432 DB_NAME=你的数据库 python your_client.py

Windows PowerShell：

$env:DB_USER="你的用户名"; $env:DB_PASS="你的密码"; $env:DB_HOST="你的主机"; $env:DB_PORT="5432"; $env:DB_NAME="你的数据库"; python your_client.py

修改代码 #

import os
import asyncpg
import logging
import asyncio

async def create_db_connection():
    while True:
        try:
            # 从环境变量读取凭证
            user = os.environ.get("DB_USER")
            password = os.environ.get("DB_PASS")
            host = os.environ.get("DB_HOST")
            port = os.environ.get("DB_PORT")
            database = os.environ.get("DB_NAME")
            
            # 读取后立即删除环境变量中的凭证
            del os.environ["DB_USER"]
            del os.environ["DB_PASS"]
            del os.environ["DB_HOST"]
            del os.environ["DB_PORT"]
            del os.environ["DB_NAME"]
            
            conn = await asyncpg.connect(
                user=user,
                password=password,
                host=host,
                port=port,
                database=database
            )
            # 销毁内存中的凭证
            del user, password, host, port, database
            return conn
        except Exception as e:
            logging.error(f"\n\tError connecting to the database: {e}")
            await asyncio.sleep(0.7)

async def log_off_database(token):
    while True:
        try:
            conn = await create_db_connection()
            await conn.execute(
                'UPDATE "Tokens" SET "Status" = $1, "UsedBy" = $2 WHERE "Token" = $3',
                'Free', '-', token
            )
        except Exception as e:
            logging.error(f"Error logging off: {e}")
        finally:
            await conn.close()
        await asyncio.sleep(9)

优点：无需修改系统配置，适合服务器端脚本；缺点：需注意避免shell历史记录泄露凭证（可禁用对应命令的历史记录）。

内容的提问来源于stack exchange，提问作者localboy