Crawl/backend/app/utils/apikey_crypto.py

"""
API Key 加密/解密工具（纯标准库，无第三方依赖）

算法：XOR 密钥流 + 字节循环位移 + Base64 URL-safe 编码

加密流程：
  1. 用 SECRET_KEY 通过 SHA-256 派生密钥流（hashlib，标准库）
  2. 明文每字节与密钥流对应字节 XOR
  3. 每字节循环左移 (i % 5 + 1) 位
  4. Base64 URL-safe 编码（无 = 填充）

解密流程（完全对称，逆序）：
  1. Base64 URL-safe 解码
  2. 每字节循环右移 (i % 5 + 1) 位
  3. 与相同密钥流 XOR
  4. UTF-8 解码

解密示例（JavaScript）：
  见文件末尾注释
"""
from __future__ import annotations

import base64
import hashlib
import os
from typing import Optional


def _derive_keystream(key: str, length: int) -> bytes:
    """
    从 key 派生指定长度的密钥流。
    原理：反复对 key + 块索引 做 SHA-256，拼接直到够用。
    """
    stream = bytearray()
    block = 0
    key_bytes = key.encode("utf-8")
    while len(stream) < length:
        h = hashlib.sha256(key_bytes + block.to_bytes(4, "big")).digest()
        stream.extend(h)
        block += 1
    return bytes(stream[:length])


def _rotl8(byte: int, n: int) -> int:
    """8 位循环左移 n 位。"""
    n = n % 8
    return ((byte << n) | (byte >> (8 - n))) & 0xFF


def _rotr8(byte: int, n: int) -> int:
    """8 位循环右移 n 位。"""
    n = n % 8
    return ((byte >> n) | (byte << (8 - n))) & 0xFF


def _get_key() -> str:
    key = os.environ.get("APIKEY_ENCRYPT_KEY", "")
    if not key:
        raise RuntimeError("APIKEY_ENCRYPT_KEY 未配置")
    return key


def encrypt_api_key(plaintext: str) -> str:
    """将 API Key 明文加密为 Base64 URL-safe 密文字符串。"""
    data = plaintext.encode("utf-8")
    keystream = _derive_keystream(_get_key(), len(data))
    result = bytearray(len(data))
    for i, byte in enumerate(data):
        xored = byte ^ keystream[i]
        result[i] = _rotl8(xored, i % 5 + 1)
    return base64.urlsafe_b64encode(result).rstrip(b"=").decode("ascii")


def decrypt_api_key(ciphertext: str) -> str:
    """将 Base64 URL-safe 密文解密为 API Key 明文。"""
    # 补回 Base64 padding
    rem = len(ciphertext) % 4
    if rem:
        ciphertext += "=" * (4 - rem)
    data = base64.urlsafe_b64decode(ciphertext)
    keystream = _derive_keystream(_get_key(), len(data))
    result = bytearray(len(data))
    for i, byte in enumerate(data):
        unshifted = _rotr8(byte, i % 5 + 1)
        result[i] = unshifted ^ keystream[i]
    return result.decode("utf-8")


def try_encrypt(plaintext: Optional[str]) -> Optional[str]:
    """安全封装：None 直接返回 None，加密失败返回 None。"""
    if not plaintext:
        return None
    try:
        return encrypt_api_key(plaintext)
    except Exception:
        return None


# ─────────────────────────────────────────────────────────────────────────────
# JavaScript 解密实现（供调用方参考）
# ─────────────────────────────────────────────────────────────────────────────
#
# async function deriveKeystream(key, length) {
#   const enc = new TextEncoder();
#   const keyBytes = enc.encode(key);
#   const stream = [];
#   let block = 0;
#   while (stream.length < length) {
#     const blockBytes = new Uint8Array(4);
#     new DataView(blockBytes.buffer).setUint32(0, block, false); // big-endian
#     const input = new Uint8Array([...keyBytes, ...blockBytes]);
#     const hash = await crypto.subtle.digest('SHA-256', input);
#     stream.push(...new Uint8Array(hash));
#     block++;
#   }
#   return stream.slice(0, length);
# }
#
# function rotr8(byte, n) {
#   n = n % 8;
#   return ((byte >>> n) | (byte << (8 - n))) & 0xFF;
# }
#
# function base64urlDecode(str) {
#   str = str.replace(/-/g, '+').replace(/_/g, '/');
#   while (str.length % 4) str += '=';
#   return Uint8Array.from(atob(str), c => c.charCodeAt(0));
# }
#
# async function decryptApiKey(ciphertext, secretKey) {
#   const data = base64urlDecode(ciphertext);
#   const keystream = await deriveKeystream(secretKey, data.length);
#   const result = new Uint8Array(data.length);
#   for (let i = 0; i < data.length; i++) {
#     const unshifted = rotr8(data[i], i % 5 + 1);
#     result[i] = unshifted ^ keystream[i];
#   }
#   return new TextDecoder().decode(result);
# }
#
# // 使用示例：
# // const apiKey = await decryptApiKey(encryptedValue, 'your_secret_key');
#
# ─────────────────────────────────────────────────────────────────────────────
# Python 解密示例（调用方如果用 Python）
# ─────────────────────────────────────────────────────────────────────────────
#
# import base64, hashlib
#
# def decrypt_api_key(ciphertext: str, secret_key: str) -> str:
#     rem = len(ciphertext) % 4
#     if rem:
#         ciphertext += '=' * (4 - rem)
#     data = base64.urlsafe_b64decode(ciphertext)
#     # 派生密钥流
#     stream, block = bytearray(), 0
#     key_bytes = secret_key.encode()
#     while len(stream) < len(data):
#         stream.extend(hashlib.sha256(key_bytes + block.to_bytes(4, 'big')).digest())
#         block += 1
#     keystream = bytes(stream[:len(data)])
#     # 解密
#     result = bytearray(len(data))
#     for i, byte in enumerate(data):
#         n = i % 5 + 1
#         unshifted = ((byte >> n) | (byte << (8 - n))) & 0xFF
#         result[i] = unshifted ^ keystream[i]
#     return result.decode('utf-8')