-- 查看当前支持的认证插件
SELECT PLUGIN_NAME, PLUGIN_STATUS
FROM INFORMATION_SCHEMA.PLUGINS
WHERE PLUGIN_TYPE = 'AUTHENTICATION';

-- 查看用户的认证方式
SELECT user, host, plugin FROM mysql.user;

"""
mysql_native_password 完整认证实现
"""
import hashlib
import struct

def sha1_binary(data: bytes) -> bytes:
    """计算 SHA1 哈希，返回 20 字节二进制"""
    return hashlib.sha1(data).digest()

def sha1_hex(data: bytes) -> str:
    """计算 SHA1 哈希，返回 40 字符十六进制"""
    return hashlib.sha1(data).hexdigest().upper()

def mysql_native_password_auth(password: str, scramble: bytes) -> bytes:
    """
    mysql_native_password 客户端认证计算
    输入:
      password  - 明文密码
      scramble  - 服务器发送的 20 字节随机数 (取前 20 字节)
    输出:
      20 字节的认证响应
    """
    if not password:
        return b''

    # Step 1: SHA1(password)
    stage1 = sha1_binary(password.encode('utf-8'))

    # Step 2: SHA1(SHA1(password))
    stage2 = sha1_binary(stage1)

    # Step 3: SHA1(scramble + SHA1(SHA1(password)))
    stage3 = sha1_binary(scramble + stage2)

    # Step 4: XOR(SHA1(password), SHA1(scramble + SHA1(SHA1(password))))
    result = bytes(a ^ b for a, b in zip(stage1, stage3))

    return result  # 20 字节


def mysql_native_password_verify(password: str, scramble: bytes, response: bytes) -> bool:
    """
    服务器端验证
    输入:
      password  - 明文密码 (用于演示; 实际服务器存储的是 hash)
      scramble  - 发送给客户端的随机数
      response  - 客户端返回的认证响应
    """
    # 服务器端的验证逻辑 (使用存储的 hash)
    stored_hash = sha1_binary(sha1_binary(password.encode('utf-8')))

    # 预期的客户端响应
    expected_stage3 = sha1_binary(scramble + stored_hash)
    expected_stage1 = sha1_binary(password.encode('utf-8'))
    expected_response = bytes(a ^ b for a, b in zip(expected_stage1, expected_stage3))

    # 验证方式: SHA1(response) 应该等于 SHA1(expected_response)
    # 实际上服务器更高效地比较: SHA1(response) == SHA1(scramble + stored_hash)
    return sha1_binary(response) == expected_stage3


# 完整演示
def demo():
    password = "MySecurePass123!"
    scramble = bytes.fromhex("4a5b6c7d8e9fa0b1c2d3e4f5a6b7c8d9e0f1a2b3")

    print(f"密码: {password}")
    print(f"Scramble: {scramble.hex()}")
    print()

    # 客户端计算
    response = mysql_native_password_auth(password, scramble)
    print(f"认证响应: {response.hex()}")

    # 服务器验证
    is_valid = mysql_native_password_verify(password, scramble, response)
    print(f"验证结果: {'✓ 通过' if is_valid else '✗ 失败'}")

    # 错误密码测试
    wrong_response = mysql_native_password_auth("WrongPassword", scramble)
    is_valid_wrong = mysql_native_password_verify("WrongPassword", scramble, wrong_response)
    print(f"错误密码验证: {'✓ 通过' if is_valid_wrong else '✗ 失败'}")

    # 交叉验证（错误密码的响应不应该通过正确密码的验证）
    is_cross = mysql_native_password_verify(password, scramble, wrong_response)
    print(f"交叉验证: {'✓ 通过' if is_cross else '✗ 失败（符合预期）'}")


if __name__ == '__main__':
    demo()

"""
caching_sha2_password 完整认证实现
"""
import hashlib
import os
import struct

try:
    from cryptography.hazmat.primitives.asymmetric import rsa, padding
    from cryptography.hazmat.primitives import hashes, serialization
    HAS_CRYPTO = True
except ImportError:
    HAS_CRYPTO = False
    print("[!] 请安装 cryptography: pip install cryptography")


def sha256_binary(data: bytes) -> bytes:
    return hashlib.sha256(data).digest()


def xor_bytes(a: bytes, b: bytes) -> bytes:
    """两个等长字节序列的异或"""
    return bytes(x ^ y for x, y in zip(a, b))


def caching_sha2_password_fast_auth(password: str, scramble: bytes) -> bytes:
    """
    caching_sha2_password 快速认证（挑战-应答模式）
    这是客户端发送的第一轮认证数据

    算法:
      hash1 = SHA256(password)
      hash2 = SHA256(hash1)
      hash3 = SHA256(hash2 + scramble)
      response = hash1 XOR hash3

    注意: 响应长度为 32 字节 (SHA256 输出长度)
    """
    if not password:
        return b''

    hash1 = sha256_binary(password.encode('utf-8'))       # SHA256(password)
    hash2 = sha256_binary(hash1)                           # SHA256(SHA256(password))
    hash3 = sha256_binary(hash2 + scramble[:])             # SHA256(SHA256(SHA256(password)) + scramble)
    response = xor_bytes(hash1, hash3)                     # XOR

    return response  # 32 字节


def caching_sha2_password_full_auth(password: str, rsa_public_key_pem: bytes) -> bytes:
    """
    caching_sha2_password 完整认证（RSA 加密模式）
    当快速认证失败（缓存未命中）时使用

    算法:
      xor_key = SHA256(password)
      encrypted = RSA_OAEP_Encrypt(xor_key XOR password_bytes, rsa_public_key)
    """
    if not HAS_CRYPTO:
        raise RuntimeError("需要 cryptography 库")

    # XOR 密码与 SHA256(password)
    password_bytes = password.encode('utf-8')
    xor_key = sha256_binary(password_bytes)

    # 对齐到 8 字节边界
    padded = password_bytes + b'\x00' * (8 - len(password_bytes) % 8) if len(password_bytes) % 8 != 0 else password_bytes

    # XOR 操作（循环使用 xor_key）
    xor_result = bytearray(len(padded))
    for i in range(len(padded)):
        xor_result[i] = padded[i] ^ xor_key[i % len(xor_key)]

    # RSA OAEP 加密
    public_key = serialization.load_pem_public_key(rsa_public_key_pem)
    encrypted = public_key.encrypt(
        bytes(xor_result),
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )

    return encrypted


def simulate_full_auth_flow():
    """模拟完整的认证流程（含缓存未命中场景）"""
    print("=" * 60)
    print("模拟 caching_sha2_password 完整认证流程")
    print("=" * 60)

    password = "MySecurePass123!"
    scramble = os.urandom(20)  # 20 字节随机数

    print(f"\n密码: {password}")
    print(f"Scramble: {scramble.hex()}")

    # === 第一步：快速认证尝试 ===
    print("\n--- 第一步：快速认证 ---")
    fast_response = caching_sha2_password_fast_auth(password, scramble)
    print(f"快速认证响应: {fast_response.hex()}")
    print(f"响应长度: {len(fast_response)} 字节")

    # 假设缓存未命中
    print("\n[服务器] 缓存未命中, 请求完整认证...")

    # === 第二步：完整认证 ===
    print("\n--- 第二步：RSA 公钥交换 ---")

    if HAS_CRYPTO:
        # 生成 RSA 密钥对（演示用）
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048
        )
        public_key = private_key.public_key()

        # 导出公钥 PEM
        public_key_pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
        print(f"RSA 公钥 (PEM, {len(public_key_pem)} 字节):")
        print(public_key_pem.decode()[:200] + "...")

        # 客户端使用公钥加密
        encrypted = caching_sha2_password_full_auth(password, public_key_pem)
        print(f"加密后的认证数据: {encrypted[:32].hex()}...")
        print(f"加密数据长度: {len(encrypted)} 字节")

        # 服务端使用私钥解密
        decrypted = private_key.decrypt(
            encrypted,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )

        # 提取密码（去掉 padding）
        decrypted_password = decrypted.rstrip(b'\x00')
        print(f"\n[服务器] 解密得到: {decrypted_password}")
        print(f"验证: {'✓ 成功' if decrypted_password == password.encode() else '✗ 失败'}")
    else:
        print("[!] 跳过 RSA 演示 (需要 cryptography 库)")


def compare_auth_methods():
    """对比两种认证方法"""
    print("\n" + "=" * 60)
    print("认证方法对比")
    print("=" * 60)

    password = "TestPassword!"
    scramble = os.urandom(20)

    # mysql_native_password
    import hashlib
    h1 = hashlib.sha1(password.encode()).digest()
    h2 = hashlib.sha1(h1).digest()
    h3 = hashlib.sha1(scramble + h2).digest()
    native_response = bytes(a ^ b for a, b in zip(h1, h3))

    # caching_sha2_password
    sha2_response = caching_sha2_password_fast_auth(password, scramble)

    print(f"\n密码: {password}")
    print(f"Scramble: {scramble.hex()}")
    print(f"\nmysql_native_password:")
    print(f"  响应长度: {len(native_response)} 字节")
    print(f"  哈希算法: SHA1")
    print(f"  响应值: {native_response.hex()}")
    print(f"\ncaching_sha2_password:")
    print(f"  响应长度: {len(sha2_response)} 字节")
    print(f"  哈希算法: SHA256")
    print(f"  响应值: {sha2_response.hex()}")


if __name__ == '__main__':
    compare_auth_methods()
    simulate_full_auth_flow()

-- 查看缓存大小
SHOW VARIABLES LIKE 'caching_sha2_password%';
-- caching_sha2_password_auto_generate_rsa_keys  ON
-- caching_sha2_password_private_key_path        private_key.pem
-- caching_sha2_password_public_key_path         public_key.pem
-- caching_sha2_password_rsa_key_size            2048

-- 手动刷新缓存（通常在密码变更后）
FLUSH PRIVILEGES;

-- 创建 socket 认证用户
CREATE USER 'dbadmin'@'localhost' IDENTIFIED WITH auth_socket;

-- 允许 dbadmin 用户以 root 身份连接
CREATE USER 'root'@'localhost' IDENTIFIED WITH auth_socket AS 'root';

# 只有 OS root 用户可以以 root 身份连接 MySQL
sudo mysql -u root

# OS 用户 dbadmin 可以连接
mysql -u dbadmin   # 使用 dbadmin 的 OS 身份

-- 安装 LDAP 认证插件
INSTALL PLUGIN authentication_ldap_simple SONAME 'authentication_ldap_simple.so';

-- 配置 LDAP 服务器
SET GLOBAL authentication_ldap_simple_server_host = 'ldap.example.com';
SET GLOBAL authentication_ldap_simple_server_port = 389;
SET GLOBAL authentication_ldap_simple_bind_base_dn = 'ou=users,dc=example,dc=com';

-- 创建 LDAP 认证用户
CREATE USER 'ldapuser'@'%'
  IDENTIFIED WITH authentication_ldap_simple
  AS 'uid=ldapuser,ou=users,dc=example,dc=com';

-- 安装 PAM 认证插件
INSTALL PLUGIN authentication_pam SONAME 'authentication_pam.so';

-- 创建 PAM 认证用户
CREATE USER 'pamuser'@'%'
  IDENTIFIED WITH authentication_pam
  AS 'mysql';  -- 'mysql' 是 PAM 服务名

def handle_auth_more_data(data, password, scramble):
    """处理 AuthMoreData 响应"""
    status = data[0]

    if status == 0x01:
        # caching_sha2_password: 快速认证失败
        # 需要进行 RSA 完整认证
        print("[*] 快速认证失败，需要 RSA 公钥")

        # 发送公钥请求
        request_public_key = b'\x02'  # 特殊命令
        return request_public_key

    elif status == 0x04:
        # sha256_password: 服务器要求 RSA 加密
        print("[*] 需要 RSA 加密密码")

    else:
        # 后续字节是 RSA 公钥
        public_key_pem = data  # 整个数据是公钥
        print(f"[*] 收到 RSA 公钥 ({len(public_key_pem)} 字节)")
        # 使用公钥加密密码...

-- 将用户切换到 mysql_native_password（兼容旧客户端）
ALTER USER 'app_user'@'%'
  IDENTIFIED WITH mysql_native_password BY 'password123';

-- 将用户切换到 caching_sha2_password（推荐）
ALTER USER 'app_user'@'%'
  IDENTIFIED WITH caching_sha2_password BY 'password123';

-- 修改默认认证插件（全局）
-- 在 my.cnf 中设置:
-- [mysqld]
-- default_authentication_plugin = mysql_native_password

-- 安装密码验证组件
INSTALL COMPONENT 'file://component_validate_password';

-- 配置密码策略
SET GLOBAL validate_password.policy = MEDIUM;
SET GLOBAL validate_password.length = 12;
SET GLOBAL validate_password.mixed_case_count = 1;
SET GLOBAL validate_password.number_count = 1;
SET GLOBAL validate_password.special_char_count = 1;

-- 要求用户使用 TLS 连接
ALTER USER 'app_user'@'%' REQUIRE SSL;

-- 要求 TLS 版本
ALTER USER 'app_user'@'%' REQUIRE X509;

-- 全局 TLS 配置
-- my.cnf:
-- [mysqld]
-- require_secure_transport = ON
-- tls_version = TLSv1.2,TLSv1.3

-- 安全加固脚本
DELETE FROM mysql.user WHERE User = '';
DELETE FROM mysql.user WHERE User = 'root' AND Host NOT IN ('localhost', '127.0.0.1', '::1');
FLUSH PRIVILEGES;

-- 临时方案：将用户切换回旧的认证方式
ALTER USER 'app'@'%' IDENTIFIED WITH mysql_native_password BY 'password';

-- 长期方案：升级客户端驱动

SELECT
    User,
    Host,
    plugin AS '认证插件',
    authentication_string AS '认证数据',
    password_expired AS '密码过期',
    password_lifetime AS '密码有效期',
    account_locked AS '账户锁定',
    CASE
        WHEN plugin = 'mysql_native_password' THEN '⚠ 建议升级到 caching_sha2_password'
        WHEN plugin = 'caching_sha2_password' THEN '✓ 推荐'
        WHEN plugin = 'auth_socket' THEN '✓ 本地安全'
        ELSE '?'
    END AS '安全评估'
FROM mysql.user
WHERE User NOT LIKE 'mysql.%'
ORDER BY User, Host;