首次提交
This commit is contained in:
@@ -0,0 +1,213 @@
|
||||
"""
|
||||
CertCenter 端到端测试
|
||||
用 Let's Encrypt Staging 环境测试完整的证书签发流程
|
||||
|
||||
用法:
|
||||
1. 先在 WebUI 中配置好 ACME（使用 Staging 环境 + 阿里云 DNS 凭据）
|
||||
2. 添加一台服务器和一个域名
|
||||
3. 执行: python -m tests.test_e2e --domain example.com
|
||||
"""
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
import requests
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000"
|
||||
|
||||
|
||||
def log(step, msg, status="info"):
|
||||
icons = {"info": "ℹ️", "ok": "✅", "fail": "❌", "warn": "⚠️"}
|
||||
print(f" {icons.get(status, ' ')} [{step}] {msg}")
|
||||
|
||||
|
||||
def test_api_alive():
|
||||
"""测试 API 是否存活"""
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/admin/api/stats", timeout=5)
|
||||
return resp.status_code == 200
|
||||
except:
|
||||
return False
|
||||
|
||||
|
||||
def get_or_create_server(name, platform, token):
|
||||
"""获取或创建服务器"""
|
||||
resp = requests.get(f"{BASE_URL}/admin/api/servers")
|
||||
for s in resp.json():
|
||||
if s["name"] == name:
|
||||
return s["id"]
|
||||
|
||||
resp = requests.post(f"{BASE_URL}/admin/api/servers", json={
|
||||
"name": name,
|
||||
"platform": platform,
|
||||
"token": token,
|
||||
})
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
def get_or_create_domain(server_id, domain, cert_dir, check_cmd, reload_cmd):
|
||||
"""获取或创建域名"""
|
||||
resp = requests.get(f"{BASE_URL}/admin/api/domains")
|
||||
for d in resp.json():
|
||||
if d["domain"] == domain:
|
||||
return d["id"]
|
||||
|
||||
resp = requests.post(f"{BASE_URL}/admin/api/domains", json={
|
||||
"server_id": server_id,
|
||||
"domain": domain,
|
||||
"cert_dir": cert_dir,
|
||||
"check_cmd": check_cmd,
|
||||
"reload_cmd": reload_cmd,
|
||||
})
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
def configure_acme(email, dns_provider, credentials):
|
||||
"""配置 ACME"""
|
||||
resp = requests.put(f"{BASE_URL}/admin/api/acme/config", json={
|
||||
"acme_server": "https://acme-staging-v02.api.letsencrypt.org/directory",
|
||||
"email": email,
|
||||
"dns_provider": dns_provider,
|
||||
"dns_credentials": json.dumps(credentials),
|
||||
"renew_days": 30,
|
||||
})
|
||||
return resp.status_code == 200
|
||||
|
||||
|
||||
def issue_certificate(domain_id):
|
||||
"""申请证书"""
|
||||
resp = requests.post(f"{BASE_URL}/admin/api/acme/issue/{domain_id}", timeout=180)
|
||||
return resp.json()
|
||||
|
||||
|
||||
def check_cert_info(domain_id):
|
||||
"""检查证书信息"""
|
||||
resp = requests.get(f"{BASE_URL}/admin/api/cert-info/{domain_id}")
|
||||
return resp.json()
|
||||
|
||||
|
||||
def test_script_download(domain, server_name):
|
||||
"""测试脚本下载"""
|
||||
resp = requests.get(f"{BASE_URL}/api/script/{domain}", params={"server_name": server_name})
|
||||
return resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
def test_version_api(domain):
|
||||
"""测试版本接口"""
|
||||
resp = requests.get(f"{BASE_URL}/api/version/{domain}")
|
||||
return resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
def test_cert_download(domain, token):
|
||||
"""测试证书下载"""
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
resp = requests.get(f"{BASE_URL}/api/cert/{domain}/fullchain", headers=headers)
|
||||
return resp.status_code == 200
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="CertCenter 端到端测试")
|
||||
parser.add_argument("--domain", required=True, help="测试域名（需已使用阿里云 DNS）")
|
||||
parser.add_argument("--email", default="test@example.com", help="ACME 注册邮箱")
|
||||
parser.add_argument("--ak", required=True, help="阿里云 Access Key")
|
||||
parser.add_argument("--sk", required=True, help="阿里云 Access Secret")
|
||||
args = parser.parse_args()
|
||||
|
||||
print(f"\n{'=' * 50}")
|
||||
print(f" 端到端测试: {args.domain}")
|
||||
print(f"{'=' * 50}\n")
|
||||
|
||||
# Step 0: 检查 API
|
||||
log("0", "检查 API 服务...")
|
||||
if not test_api_alive():
|
||||
log("0", "API 服务未启动，请先执行 python -m backend.main", "fail")
|
||||
return 1
|
||||
log("0", "API 服务正常", "ok")
|
||||
|
||||
# Step 1: 配置 ACME
|
||||
log("1", "配置 ACME (Staging 环境)...")
|
||||
credentials = {"access_key": args.ak, "access_secret": args.sk}
|
||||
if configure_acme(args.email, "aliyun", credentials):
|
||||
log("1", "ACME 配置完成", "ok")
|
||||
else:
|
||||
log("1", "ACME 配置失败", "fail")
|
||||
return 1
|
||||
|
||||
# Step 2: 创建测试服务器
|
||||
log("2", "创建测试服务器...")
|
||||
test_token = "tok_e2e_test_" + str(int(time.time()))
|
||||
server_id = get_or_create_server("e2e-test", "linux", test_token)
|
||||
log("2", f"服务器 ID: {server_id}", "ok")
|
||||
|
||||
# Step 3: 创建测试域名
|
||||
log("3", "创建测试域名配置...")
|
||||
cert_dir = f"/tmp/cert-e2e-test/{args.domain}"
|
||||
domain_id = get_or_create_domain(
|
||||
server_id, args.domain,
|
||||
cert_dir=cert_dir,
|
||||
check_cmd="true", # 测试环境用 true 代替 nginx -t
|
||||
reload_cmd="true",
|
||||
)
|
||||
log("3", f"域名 ID: {domain_id}", "ok")
|
||||
|
||||
# Step 4: 申请证书
|
||||
log("4", "申请证书（这需要 1-2 分钟，等待 DNS 验证）...")
|
||||
result = issue_certificate(domain_id)
|
||||
if result.get("success"):
|
||||
log("4", f"证书申请成功: {result.get('message')}", "ok")
|
||||
else:
|
||||
log("4", f"证书申请失败: {result.get('message')}", "fail")
|
||||
log("4", "请检查：1) 域名 DNS 是否托管在阿里云 2) AK/SK 是否正确 3) 网络是否通畅", "warn")
|
||||
return 1
|
||||
|
||||
# Step 5: 检查证书信息
|
||||
log("5", "检查证书信息...")
|
||||
cert_info = check_cert_info(domain_id)
|
||||
if cert_info.get("cert_info"):
|
||||
info = cert_info["cert_info"]
|
||||
log("5", f"签发者: {info.get('issuer')}", "ok")
|
||||
log("5", f"有效期: {info.get('not_before')} ~ {info.get('not_after')}", "ok")
|
||||
log("5", f"域名: {info.get('san')}", "ok")
|
||||
log("5", f"剩余天数: {cert_info.get('days_left')}", "ok")
|
||||
else:
|
||||
log("5", "无法读取证书信息", "warn")
|
||||
|
||||
# Step 6: 测试版本接口
|
||||
log("6", "测试版本接口...")
|
||||
ok, version = test_version_api(args.domain)
|
||||
if ok:
|
||||
log("6", f"版本号: {version}", "ok")
|
||||
else:
|
||||
log("6", "版本接口异常", "fail")
|
||||
|
||||
# Step 7: 测试脚本生成
|
||||
log("7", "测试脚本生成...")
|
||||
ok, script = test_script_download(args.domain, "e2e-test")
|
||||
if ok and len(script) > 100:
|
||||
log("7", f"脚本生成成功，长度: {len(script)} 字符", "ok")
|
||||
else:
|
||||
log("7", "脚本生成失败", "fail")
|
||||
|
||||
# Step 8: 测试证书下载
|
||||
log("8", "测试证书下载...")
|
||||
if test_cert_download(args.domain, test_token):
|
||||
log("8", "证书下载成功", "ok")
|
||||
else:
|
||||
log("8", "证书下载失败", "fail")
|
||||
|
||||
# 完成
|
||||
print(f"\n{'=' * 50}")
|
||||
print(f" ✅ 端到端测试完成!")
|
||||
print(f"{'=' * 50}")
|
||||
print(f"\n 证书位置: {cert_dir}/")
|
||||
print(f" Staging 证书不受浏览器信任，仅用于验证流程。")
|
||||
print(f" 确认流程正常后，在 WebUI 中切换到生产环境重新签发。\n")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,242 @@
|
||||
"""
|
||||
CertCenter 部署验证脚本
|
||||
用法: python -m tests.test_setup
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import requests
|
||||
from pathlib import Path
|
||||
|
||||
# 添加项目根目录到 path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
PASS = "✅"
|
||||
FAIL = "❌"
|
||||
SKIP = "⏭️"
|
||||
|
||||
results = []
|
||||
|
||||
|
||||
def check(name, func):
|
||||
"""执行检查并记录结果"""
|
||||
try:
|
||||
ok, msg = func()
|
||||
status = PASS if ok else FAIL
|
||||
results.append((name, status, msg))
|
||||
print(f" {status} {name}: {msg}")
|
||||
except Exception as e:
|
||||
results.append((name, FAIL, str(e)))
|
||||
print(f" {FAIL} {name}: {e}")
|
||||
|
||||
|
||||
def test_python_deps():
|
||||
"""检查 Python 依赖"""
|
||||
deps = ["fastapi", "uvicorn", "sqlalchemy", "acme", "cryptography", "jinja2"]
|
||||
missing = []
|
||||
for dep in deps:
|
||||
try:
|
||||
__import__(dep)
|
||||
except ImportError:
|
||||
missing.append(dep)
|
||||
if missing:
|
||||
return False, f"缺少依赖: {', '.join(missing)}，执行 pip install -r requirements.txt"
|
||||
return True, "所有依赖已安装"
|
||||
|
||||
|
||||
def test_certbot():
|
||||
"""检查 certbot 是否可用"""
|
||||
import subprocess
|
||||
try:
|
||||
result = subprocess.run(["certbot", "--version"], capture_output=True, text=True, timeout=5)
|
||||
if result.returncode == 0:
|
||||
return True, result.stdout.strip()
|
||||
return False, "certbot 命令不可用"
|
||||
except FileNotFoundError:
|
||||
return False, "certbot 未安装，执行 pip install certbot"
|
||||
|
||||
|
||||
def test_certbot_dns_plugin():
|
||||
"""检查 certbot DNS 插件"""
|
||||
try:
|
||||
import certbot_dns_alicloud
|
||||
return True, "certbot-dns-alicloud 已安装"
|
||||
except ImportError:
|
||||
try:
|
||||
import certbot_dns_cloudflare
|
||||
return True, "certbot-dns-cloudflare 已安装"
|
||||
except ImportError:
|
||||
return False, "未检测到 DNS 插件，执行 pip install certbot-dns-alicloud"
|
||||
|
||||
|
||||
def test_env_file():
|
||||
"""检查 .env 文件"""
|
||||
env_path = Path(__file__).parent.parent / ".env"
|
||||
if not env_path.exists():
|
||||
return False, ".env 文件不存在，执行 cp .env .env"
|
||||
return True, ".env 文件存在"
|
||||
|
||||
|
||||
def test_database():
|
||||
"""检查数据库连接"""
|
||||
from backend.config import get_settings
|
||||
settings = get_settings()
|
||||
db_path = settings.database_url.replace("sqlite+aiosqlite:///", "")
|
||||
if "sqlite" in settings.database_url:
|
||||
return True, f"SQLite 数据库: {db_path}"
|
||||
return True, f"数据库: {settings.database_url}"
|
||||
|
||||
|
||||
def test_frontend_dist():
|
||||
"""检查前端构建产物"""
|
||||
dist = Path(__file__).parent.parent / "frontend" / "dist"
|
||||
if dist.exists() and (dist / "index.html").exists():
|
||||
return True, f"前端已构建: {dist}"
|
||||
return False, f"前端未构建，执行 cd frontend && npm install && npm run build"
|
||||
|
||||
|
||||
def test_cert_dir():
|
||||
"""检查证书存储目录"""
|
||||
from backend.config import get_settings
|
||||
settings = get_settings()
|
||||
cert_dir = Path(settings.cert_dir)
|
||||
if cert_dir.exists():
|
||||
return True, f"证书目录存在: {cert_dir}"
|
||||
try:
|
||||
cert_dir.mkdir(parents=True, exist_ok=True)
|
||||
return True, f"证书目录已创建: {cert_dir}"
|
||||
except Exception as e:
|
||||
return False, f"无法创建证书目录: {e}"
|
||||
|
||||
|
||||
def test_base_url():
|
||||
"""检查 BASE_URL 配置"""
|
||||
from backend.config import get_settings
|
||||
settings = get_settings()
|
||||
if "example.com" in settings.base_url:
|
||||
return False, f"BASE_URL 仍为默认值: {settings.base_url}，请修改 .env"
|
||||
return True, f"BASE_URL: {settings.base_url}"
|
||||
|
||||
|
||||
def test_letsencrypt_staging():
|
||||
"""测试 Let's Encrypt 测试环境连通性"""
|
||||
try:
|
||||
resp = requests.get(
|
||||
"https://acme-staging-v02.api.letsencrypt.org/directory",
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
return True, f"Staging 环境可达, endpoints: {len(data)} 个"
|
||||
return False, f"HTTP {resp.status_code}"
|
||||
except Exception as e:
|
||||
return False, f"无法连接 Let's Encrypt Staging: {e}"
|
||||
|
||||
|
||||
def test_letsencrypt_production():
|
||||
"""测试 Let's Encrypt 生产环境连通性"""
|
||||
try:
|
||||
resp = requests.get(
|
||||
"https://acme-v02.api.letsencrypt.org/directory",
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
return True, "生产环境可达"
|
||||
return False, f"HTTP {resp.status_code}"
|
||||
except Exception as e:
|
||||
return False, f"无法连接 Let's Encrypt: {e}"
|
||||
|
||||
|
||||
def test_dns_credentials():
|
||||
"""检查 DNS 凭据是否配置"""
|
||||
from backend.database import engine, async_session
|
||||
from backend.models import AcmeConfig
|
||||
from sqlalchemy import select
|
||||
import asyncio
|
||||
|
||||
async def _check():
|
||||
async with async_session() as session:
|
||||
result = await session.execute(select(AcmeConfig).where(AcmeConfig.id == 1))
|
||||
config = result.scalar_one_or_none()
|
||||
if not config:
|
||||
return False, "ACME 未配置，请在 WebUI 中配置"
|
||||
if not config.email:
|
||||
return False, "邮箱未配置"
|
||||
creds = json.loads(config.dns_credentials or "{}")
|
||||
if not creds.get("access_key") and not creds.get("api_token"):
|
||||
return False, "DNS 凭据未配置"
|
||||
return True, f"DNS 提商: {config.dns_provider}, 邮箱: {config.email}"
|
||||
|
||||
return asyncio.run(_check())
|
||||
|
||||
|
||||
def test_api_server():
|
||||
"""测试 FastAPI 服务是否能启动"""
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
[sys.executable, "-m", "backend.main"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
cwd=str(Path(__file__).parent.parent),
|
||||
)
|
||||
time.sleep(3)
|
||||
|
||||
# 尝试访问
|
||||
try:
|
||||
resp = requests.get("http://127.0.0.1:8000/admin/api/stats", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
proc.terminate()
|
||||
return True, "服务启动成功，API 正常响应"
|
||||
proc.terminate()
|
||||
return False, f"服务启动但 API 返回 {resp.status_code}"
|
||||
except requests.ConnectionError:
|
||||
proc.terminate()
|
||||
return False, "服务启动失败，端口 8000 无响应"
|
||||
except Exception as e:
|
||||
return False, f"启动测试失败: {e}"
|
||||
|
||||
|
||||
def main():
|
||||
print("\n" + "=" * 50)
|
||||
print(" CertCenter 部署验证")
|
||||
print("=" * 50)
|
||||
|
||||
print("\n📦 环境检查:")
|
||||
check("Python 依赖", test_python_deps)
|
||||
check("Certbot", test_certbot)
|
||||
check("DNS 插件", test_certbot_dns_plugin)
|
||||
check(".env 配置", test_env_file)
|
||||
check("数据库", test_database)
|
||||
check("前端构建", test_frontend_dist)
|
||||
check("证书目录", test_cert_dir)
|
||||
check("BASE_URL", test_base_url)
|
||||
|
||||
print("\n🌐 网络检查:")
|
||||
check("Let's Encrypt Staging", test_letsencrypt_staging)
|
||||
check("Let's Encrypt Production", test_letsencrypt_production)
|
||||
|
||||
print("\n⚙️ 服务检查:")
|
||||
check("FastAPI 服务", test_api_server)
|
||||
|
||||
# 汇总
|
||||
print("\n" + "=" * 50)
|
||||
passed = sum(1 for _, s, _ in results if s == PASS)
|
||||
failed = sum(1 for _, s, _ in results if s == FAIL)
|
||||
print(f" 结果: {passed} 通过, {failed} 失败")
|
||||
|
||||
if failed:
|
||||
print("\n 请修复以上 ❌ 标记的问题后重试。")
|
||||
print(" 建议先使用 Let's Encrypt Staging 环境测试。")
|
||||
else:
|
||||
print("\n 所有检查通过！可以开始使用 CertCenter。")
|
||||
print(" 建议首次测试使用 Staging 环境，确认无误后再切换到生产环境。")
|
||||
print("=" * 50 + "\n")
|
||||
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user