104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""上传 .env 文件到 Infisical"""
|
|
|
|
import os
|
|
import re
|
|
import requests
|
|
|
|
INFISICAL_URL = "https://infisical.seekkey.eu.org/api"
|
|
TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhdXRoTWV0aG9kIjoiZW1haWwiLCJhdXRoVG9rZW5UeXBlIjoiYWNjZXNzVG9rZW4iLCJ1c2VySWQiOiJkY2NkYzVhYS04OTU0LTQwMmEtOWJkMi00OGM0NmNmZTVkZWQiLCJ0b2tlblZlcnNpb25JZCI6IjE4ZTlhMWEwLThmZTItNGE1MS1hMGQ4LTJiMWVmOTcyNDNiNiIsImFjY2Vzc1ZlcnNpb24iOjEsIm9yZ2FuaXphdGlvbklkIjoiNTI3NjU4ZGQtNWUwZS00NzAzLThiMDktOGNkZmFhZTk2YmIwIiwiaWF0IjoxNzY5ODMyNzI5LCJleHAiOjE3NzA2OTY3Mjl9.OcAUDeioCFNCXx6OjuTjzlgg9AKWPS0EIJr4K87aH1Q"
|
|
WORKSPACE_ID = "9946b650-e08b-40d9-8e28-a55e3b19ec94"
|
|
ENVIRONMENT = "dev"
|
|
|
|
def parse_env_file(filepath):
|
|
"""解析 .env 文件，返回 key-value 字典"""
|
|
secrets = {}
|
|
with open(filepath, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
# 跳过空行和注释
|
|
if not line or line.startswith('#') or line.startswith('---'):
|
|
continue
|
|
# 匹配 KEY=VALUE 格式
|
|
match = re.match(r'^([A-Za-z_][A-Za-z0-9_]*)=(.*)$', line)
|
|
if match:
|
|
key = match.group(1)
|
|
value = match.group(2)
|
|
secrets[key] = value
|
|
return secrets
|
|
|
|
def upload_secret(key, value):
|
|
"""使用 Infisical API 上传单个 secret"""
|
|
headers = {
|
|
"Authorization": f"Bearer {TOKEN}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# 获取工作区的加密密钥
|
|
url = f"{INFISICAL_URL}/v3/secrets/{key}"
|
|
params = {
|
|
"workspaceId": WORKSPACE_ID,
|
|
"environment": ENVIRONMENT,
|
|
"type": "shared"
|
|
}
|
|
|
|
# 先检查 secret 是否存在
|
|
response = requests.get(url, headers=headers, params=params)
|
|
|
|
if response.status_code == 200:
|
|
# 更新现有 secret
|
|
secret_id = response.json().get('secret', {}).get('id')
|
|
update_url = f"{INFISICAL_URL}/v3/secrets/{secret_id}"
|
|
data = {
|
|
"workspaceId": WORKSPACE_ID,
|
|
"environment": ENVIRONMENT,
|
|
"type": "shared",
|
|
"secretValue": value
|
|
}
|
|
response = requests.patch(update_url, headers=headers, json=data)
|
|
action = "更新"
|
|
elif response.status_code == 404:
|
|
# 创建新 secret - 使用批量创建 API
|
|
create_url = f"{INFISICAL_URL}/v3/secrets/batch/raw"
|
|
data = {
|
|
"workspaceId": WORKSPACE_ID,
|
|
"environment": ENVIRONMENT,
|
|
"secrets": [{
|
|
"secretKey": key,
|
|
"secretValue": value,
|
|
"type": "shared"
|
|
}]
|
|
}
|
|
response = requests.post(create_url, headers=headers, json=data)
|
|
action = "创建"
|
|
else:
|
|
print(f"❌ 检查 {key} 失败: {response.status_code}")
|
|
return False
|
|
|
|
if response.status_code in [200, 201]:
|
|
print(f"✅ {action}: {key}")
|
|
return True
|
|
else:
|
|
print(f"❌ {action} {key} 失败: {response.status_code} - {response.text[:100]}")
|
|
return False
|
|
|
|
def main():
|
|
env_file = "/home/ben/terraform/.env"
|
|
secrets = parse_env_file(env_file)
|
|
|
|
print(f"📤 开始上传 {len(secrets)} 个 secrets 到 Infisical...")
|
|
print(f" 工作区: {WORKSPACE_ID}")
|
|
print(f" 环境: {ENVIRONMENT}")
|
|
print()
|
|
|
|
success_count = 0
|
|
for key, value in secrets.items():
|
|
if upload_secret(key, value):
|
|
success_count += 1
|
|
|
|
print()
|
|
print(f"🎉 完成! 成功上传 {success_count}/{len(secrets)} 个 secrets")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|