all-project/docs/设计文档/database/oss/bucket-password-helper.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Bucket密码管理工具
用于生成、验证和修改bucket访问密码
"""

import hashlib
import secrets
import string
import sys
import mysql.connector
from mysql.connector import Error

class BucketPasswordHelper:
    def __init__(self, host='localhost', user='root', password='', database='ry-vue'):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
    
    def connect_db(self):
        """连接数据库"""
        try:
            connection = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            return connection
        except Error as e:
            print(f"数据库连接失败: {e}")
            return None
    
    def generate_password(self, length=12):
        """生成强密码"""
        # 确保包含至少一个大写字母、小写字母、数字和特殊字符
        password = (
            secrets.choice(string.ascii_uppercase) +
            secrets.choice(string.ascii_lowercase) +
            secrets.choice(string.digits) +
            secrets.choice('!@#$%^&*') +
            ''.join(secrets.choice(string.ascii_letters + string.digits + '!@#$%^&*') 
                   for _ in range(length - 4))
        )
        # 打乱密码字符顺序
        password_list = list(password)
        secrets.SystemRandom().shuffle(password_list)
        return ''.join(password_list)
    
    def encrypt_password(self, password, bucket_name):
        """加密密码（MD5 + 盐值）"""
        salt = f"{bucket_name}_salt_2025"
        encrypted = hashlib.md5((password + salt).encode('utf-8')).hexdigest()
        return encrypted
    
    def update_bucket_password(self, bucket_name, new_password):
        """更新bucket密码"""
        connection = self.connect_db()
        if not connection:
            return False
        
        try:
            cursor = connection.cursor()
            
            # 加密新密码
            encrypted_password = self.encrypt_password(new_password, bucket_name)
            
            # 检查是否存在密码记录
            cursor.execute(
                "SELECT id FROM sys_bucket_access_password WHERE bucket_name = %s AND deleted = 0",
                (bucket_name,)
            )
            result = cursor.fetchone()
            
            if result:
                # 更新现有记录
                cursor.execute("""
                    UPDATE sys_bucket_access_password 
                    SET access_password = %s, update_time = NOW(), update_by = 1
                    WHERE bucket_name = %s AND deleted = 0
                """, (encrypted_password, bucket_name))
            else:
                # 插入新记录
                cursor.execute("""
                    INSERT INTO sys_bucket_access_password 
                    (bucket_name, access_password, password_salt, is_active, max_access_count, create_by, tenant_id)
                    VALUES (%s, %s, %s, 1, 0, 1, 0)
                """, (bucket_name, encrypted_password, f"{bucket_name}_salt_2025"))
            
            connection.commit()
            print(f"✅ Bucket '{bucket_name}' 密码更新成功")
            print(f"📝 新密码: {new_password}")
            print(f"🔐 加密后: {encrypted_password}")
            return True
            
        except Error as e:
            print(f"❌ 更新密码失败: {e}")
            return False
        finally:
            if connection.is_connected():
                cursor.close()
                connection.close()
    
    def verify_password(self, bucket_name, password):
        """验证密码"""
        connection = self.connect_db()
        if not connection:
            return False
        
        try:
            cursor = connection.cursor()
            encrypted_password = self.encrypt_password(password, bucket_name)
            
            cursor.execute("""
                SELECT access_password FROM sys_bucket_access_password 
                WHERE bucket_name = %s AND deleted = 0
            """, (bucket_name,))
            
            result = cursor.fetchone()
            if result and result[0] == encrypted_password:
                print(f"✅ 密码验证成功")
                return True
            else:
                print(f"❌ 密码验证失败")
                return False
                
        except Error as e:
            print(f"❌ 验证密码失败: {e}")
            return False
        finally:
            if connection.is_connected():
                cursor.close()
                connection.close()
    
    def get_password_info(self, bucket_name):
        """获取密码信息"""
        connection = self.connect_db()
        if not connection:
            return None
        
        try:
            cursor = connection.cursor()
            cursor.execute("""
                SELECT bucket_name, is_active, max_access_count, current_access_count, 
                       expire_time, update_time
                FROM sys_bucket_access_password 
                WHERE bucket_name = %s AND deleted = 0
            """, (bucket_name,))
            
            result = cursor.fetchone()
            if result:
                print(f"📊 Bucket '{bucket_name}' 密码信息:")
                print(f"   状态: {'启用' if result[1] == 1 else '禁用'}")
                print(f"   最大访问次数: {result[2] if result[2] > 0 else '无限制'}")
                print(f"   当前访问次数: {result[3]}")
                print(f"   过期时间: {result[4] if result[4] else '永不过期'}")
                print(f"   更新时间: {result[5]}")
                return result
            else:
                print(f"❌ Bucket '{bucket_name}' 密码配置不存在")
                return None
                
        except Error as e:
            print(f"❌ 获取密码信息失败: {e}")
            return None
        finally:
            if connection.is_connected():
                cursor.close()
                connection.close()

def main():
    """主函数"""
    print("🔐 Bucket密码管理工具")
    print("=" * 50)
    
    # 数据库配置
    host = input("数据库主机 (默认: localhost): ").strip() or "localhost"
    user = input("数据库用户名 (默认: root): ").strip() or "root"
    password = input("数据库密码: ").strip()
    database = input("数据库名称 (默认: ry-vue): ").strip() or "ry-vue"
    
    helper = BucketPasswordHelper(host, user, password, database)
    
    while True:
        print("\n请选择操作:")
        print("1. 生成强密码")
        print("2. 更新bucket密码")
        print("3. 验证bucket密码")
        print("4. 查看密码信息")
        print("5. 退出")
        
        choice = input("\n请输入选择 (1-5): ").strip()
        
        if choice == "1":
            length = int(input("密码长度 (默认: 12): ").strip() or "12")
            new_password = helper.generate_password(length)
            print(f"🔑 生成的强密码: {new_password}")
            
        elif choice == "2":
            bucket_name = input("Bucket名称: ").strip()
            if not bucket_name:
                print("❌ Bucket名称不能为空")
                continue
                
            use_generated = input("是否使用生成的强密码? (y/n): ").strip().lower()
            if use_generated == 'y':
                length = int(input("密码长度 (默认: 12): ").strip() or "12")
                new_password = helper.generate_password(length)
                print(f"🔑 生成的密码: {new_password}")
            else:
                new_password = input("新密码: ").strip()
                if not new_password:
                    print("❌ 密码不能为空")
                    continue
            
            helper.update_bucket_password(bucket_name, new_password)
            
        elif choice == "3":
            bucket_name = input("Bucket名称: ").strip()
            password = input("密码: ").strip()
            if bucket_name and password:
                helper.verify_password(bucket_name, password)
            else:
                print("❌ Bucket名称和密码不能为空")
                
        elif choice == "4":
            bucket_name = input("Bucket名称: ").strip()
            if bucket_name:
                helper.get_password_info(bucket_name)
            else:
                print("❌ Bucket名称不能为空")
                
        elif choice == "5":
            print("👋 再见!")
            break
            
        else:
            print("❌ 无效选择，请重新输入")

if __name__ == "__main__":
    main()