761 lines
26 KiB
Python
761 lines
26 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Header, Cookie, Request
|
||
from fastapi.responses import JSONResponse, PlainTextResponse
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.orm import selectinload
|
||
from pydantic import BaseModel
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import logging
|
||
import time
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
from backend.database import get_db
|
||
from backend.models import Server, Domain, DeployLog, AcmeConfig, AcmeLog
|
||
from backend.config import get_settings
|
||
|
||
router = APIRouter(tags=["admin-api"])
|
||
|
||
|
||
# ──────────────── Auth ────────────────
|
||
|
||
def generate_token(username: str, secret_key: str) -> str:
|
||
"""生成简单的认证 token"""
|
||
expire = int(time.time()) + 86400 * 7 # 7 天过期
|
||
payload = f"{username}:{expire}"
|
||
signature = hmac.new(secret_key.encode(), payload.encode(), hashlib.sha256).hexdigest()[:16]
|
||
return f"{payload}:{signature}"
|
||
|
||
|
||
def verify_token(token: str, secret_key: str) -> bool:
|
||
"""验证 token"""
|
||
try:
|
||
parts = token.split(":")
|
||
if len(parts) != 3:
|
||
return False
|
||
username, expire_str, signature = parts
|
||
expire = int(expire_str)
|
||
if time.time() > expire:
|
||
return False
|
||
payload = f"{username}:{expire_str}"
|
||
expected = hmac.new(secret_key.encode(), payload.encode(), hashlib.sha256).hexdigest()[:16]
|
||
return hmac.compare_digest(signature, expected)
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
async def require_auth(authorization: str = Header(None), admin_token: str = Cookie(None)):
|
||
"""认证依赖，检查 Header 或 Cookie 中的 token"""
|
||
settings = get_settings()
|
||
|
||
# 从 Header 获取
|
||
token = None
|
||
if authorization and authorization.startswith("Bearer "):
|
||
token = authorization[7:]
|
||
# 从 Cookie 获取
|
||
elif admin_token:
|
||
token = admin_token
|
||
|
||
if not token or not verify_token(token, settings.secret_key):
|
||
raise HTTPException(status_code=401, detail="未登录或登录已过期")
|
||
return True
|
||
|
||
|
||
class LoginRequest(BaseModel):
|
||
username: str
|
||
password: str
|
||
|
||
|
||
@router.post("/login")
|
||
async def login(data: LoginRequest):
|
||
settings = get_settings()
|
||
if data.username != settings.admin_username or data.password != settings.admin_password:
|
||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||
|
||
token = generate_token(data.username, settings.secret_key)
|
||
response = JSONResponse(content={"ok": True, "token": token})
|
||
response.set_cookie(
|
||
key="admin_token",
|
||
value=token,
|
||
max_age=86400 * 7,
|
||
httponly=True,
|
||
samesite="lax",
|
||
)
|
||
return response
|
||
|
||
|
||
@router.post("/logout")
|
||
async def logout():
|
||
response = JSONResponse(content={"ok": True})
|
||
response.delete_cookie("admin_token")
|
||
return response
|
||
|
||
|
||
@router.get("/me")
|
||
async def check_auth(_: bool = Depends(require_auth)):
|
||
return {"ok": True}
|
||
|
||
|
||
# ──────────────── Schemas ────────────────
|
||
|
||
class ServerCreate(BaseModel):
|
||
name: str
|
||
platform: str = "linux"
|
||
token: str
|
||
ip: str | None = None
|
||
|
||
|
||
class ServerUpdate(BaseModel):
|
||
name: str | None = None
|
||
platform: str | None = None
|
||
token: str | None = None
|
||
ip: str | None = None
|
||
|
||
|
||
class DomainCreate(BaseModel):
|
||
server_id: int
|
||
domain: str
|
||
acme_config_id: int | None = None
|
||
cert_dir: str | None = None
|
||
check_cmd: str = "nginx -t"
|
||
reload_cmd: str = "systemctl reload nginx"
|
||
|
||
|
||
class DomainUpdate(BaseModel):
|
||
server_id: int | None = None
|
||
acme_config_id: int | None = None
|
||
domain: str | None = None
|
||
cert_dir: str | None = None
|
||
check_cmd: str | None = None
|
||
reload_cmd: str | None = None
|
||
version: str | None = None
|
||
|
||
|
||
# ──────────────── Stats ────────────────
|
||
|
||
@router.get("/stats")
|
||
async def get_stats(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
"""仪表盘统计数据"""
|
||
server_count = (await db.execute(select(func.count(Server.id)))).scalar() or 0
|
||
domain_count = (await db.execute(select(func.count(Domain.id)))) .scalar() or 0
|
||
|
||
# 即将过期（7 天内）
|
||
now = datetime.utcnow()
|
||
soon = now.replace(day=now.day + 7) if now.day <= 24 else now.replace(month=now.month + 1, day=now.day + 7 - 30)
|
||
expiring = (await db.execute(
|
||
select(func.count(Domain.id)).where(
|
||
Domain.cert_not_after.isnot(None),
|
||
Domain.cert_not_after <= soon,
|
||
)
|
||
)).scalar() or 0
|
||
|
||
# 最近日志
|
||
result = await db.execute(
|
||
select(DeployLog)
|
||
.options(selectinload(DeployLog.domain), selectinload(DeployLog.domain).selectinload(Domain.server))
|
||
.order_by(DeployLog.created_at.desc())
|
||
.limit(10)
|
||
)
|
||
logs = result.scalars().all()
|
||
|
||
return {
|
||
"server_count": server_count,
|
||
"domain_count": domain_count,
|
||
"expiring_count": expiring,
|
||
"recent_logs": [
|
||
{
|
||
"id": log.id,
|
||
"domain": log.domain.domain if log.domain else None,
|
||
"server": log.domain.server.name if log.domain and log.domain.server else None,
|
||
"status": log.status,
|
||
"message": log.message,
|
||
"created_at": log.created_at.isoformat(),
|
||
}
|
||
for log in logs
|
||
],
|
||
}
|
||
|
||
|
||
# ──────────────── Servers ────────────────
|
||
|
||
@router.get("/servers")
|
||
async def list_servers(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(Server).options(selectinload(Server.domains)).order_by(Server.id))
|
||
servers = result.scalars().all()
|
||
return [
|
||
{
|
||
"id": s.id,
|
||
"name": s.name,
|
||
"platform": s.platform,
|
||
"token": s.token,
|
||
"ip": s.ip,
|
||
"domain_count": len(s.domains),
|
||
"created_at": s.created_at.isoformat(),
|
||
}
|
||
for s in servers
|
||
]
|
||
|
||
|
||
@router.get("/servers/{server_id}")
|
||
async def get_server(server_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(Server).where(Server.id == server_id))
|
||
s = result.scalar_one_or_none()
|
||
if not s:
|
||
raise HTTPException(404, "Server not found")
|
||
return {"id": s.id, "name": s.name, "platform": s.platform, "token": s.token, "ip": s.ip}
|
||
|
||
|
||
@router.post("/servers", status_code=201)
|
||
async def create_server(data: ServerCreate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
server = Server(**data.model_dump())
|
||
db.add(server)
|
||
await db.flush()
|
||
await db.refresh(server)
|
||
return {"id": server.id, "name": server.name}
|
||
|
||
|
||
@router.put("/servers/{server_id}")
|
||
async def update_server(server_id: int, data: ServerUpdate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(Server).where(Server.id == server_id))
|
||
server = result.scalar_one_or_none()
|
||
if not server:
|
||
raise HTTPException(404, "Server not found")
|
||
for key, value in data.model_dump(exclude_unset=True).items():
|
||
setattr(server, key, value)
|
||
return {"id": server.id}
|
||
|
||
|
||
@router.delete("/servers/{server_id}")
|
||
async def delete_server(server_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(Server).where(Server.id == server_id))
|
||
server = result.scalar_one_or_none()
|
||
if not server:
|
||
raise HTTPException(404, "Server not found")
|
||
await db.delete(server)
|
||
return {"ok": True}
|
||
|
||
|
||
# ──────────────── Domains ────────────────
|
||
|
||
@router.get("/domains")
|
||
async def list_domains(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(
|
||
select(Domain).options(selectinload(Domain.server), selectinload(Domain.acme_config)).order_by(Domain.id)
|
||
)
|
||
domains = result.scalars().all()
|
||
return [
|
||
{
|
||
"id": d.id,
|
||
"server_id": d.server_id,
|
||
"server_name": d.server.name if d.server else None,
|
||
"acme_config_id": d.acme_config_id,
|
||
"acme_config_name": d.acme_config.name if d.acme_config else None,
|
||
"domain": d.domain,
|
||
"cert_dir": d.cert_dir,
|
||
"check_cmd": d.check_cmd,
|
||
"reload_cmd": d.reload_cmd,
|
||
"version": d.version,
|
||
"cert_not_after": d.cert_not_after.isoformat() if d.cert_not_after else None,
|
||
}
|
||
for d in domains
|
||
]
|
||
|
||
|
||
@router.get("/domains/{domain_id}")
|
||
async def get_domain(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(Domain).where(Domain.id == domain_id))
|
||
d = result.scalar_one_or_none()
|
||
if not d:
|
||
raise HTTPException(404, "Domain not found")
|
||
return {
|
||
"id": d.id,
|
||
"server_id": d.server_id,
|
||
"acme_config_id": d.acme_config_id,
|
||
"domain": d.domain,
|
||
"cert_dir": d.cert_dir,
|
||
"check_cmd": d.check_cmd,
|
||
"reload_cmd": d.reload_cmd,
|
||
"version": d.version,
|
||
}
|
||
|
||
|
||
@router.post("/domains", status_code=201)
|
||
async def create_domain(data: DomainCreate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
dump = data.model_dump()
|
||
if not dump.get("cert_dir"):
|
||
dump["cert_dir"] = dump["domain"].lstrip("*.")
|
||
domain = Domain(**dump)
|
||
db.add(domain)
|
||
await db.flush()
|
||
await db.refresh(domain)
|
||
return {"id": domain.id, "domain": domain.domain}
|
||
|
||
|
||
@router.put("/domains/{domain_id}")
|
||
async def update_domain(domain_id: int, data: DomainUpdate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(Domain).where(Domain.id == domain_id))
|
||
domain = result.scalar_one_or_none()
|
||
if not domain:
|
||
raise HTTPException(404, "Domain not found")
|
||
for key, value in data.model_dump(exclude_unset=True).items():
|
||
setattr(domain, key, value)
|
||
return {"id": domain.id}
|
||
|
||
|
||
@router.delete("/domains/{domain_id}")
|
||
async def delete_domain(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(Domain).where(Domain.id == domain_id))
|
||
domain = result.scalar_one_or_none()
|
||
if not domain:
|
||
raise HTTPException(404, "Domain not found")
|
||
await db.delete(domain)
|
||
return {"ok": True}
|
||
|
||
|
||
# ──────────────── Logs ────────────────
|
||
|
||
@router.get("/logs")
|
||
async def list_logs(status: str | None = None, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
query = (
|
||
select(DeployLog)
|
||
.options(selectinload(DeployLog.domain), selectinload(DeployLog.domain).selectinload(Domain.server))
|
||
.order_by(DeployLog.created_at.desc())
|
||
.limit(100)
|
||
)
|
||
if status:
|
||
query = query.where(DeployLog.status == status)
|
||
result = await db.execute(query)
|
||
logs = result.scalars().all()
|
||
return [
|
||
{
|
||
"id": log.id,
|
||
"domain": log.domain.domain if log.domain else None,
|
||
"server": log.domain.server.name if log.domain and log.domain.server else None,
|
||
"status": log.status,
|
||
"message": log.message,
|
||
"created_at": log.created_at.isoformat(),
|
||
}
|
||
for log in logs
|
||
]
|
||
|
||
|
||
# ──────────────── ACME 配置（多配置）────────────────
|
||
|
||
class AcmeConfigCreate(BaseModel):
|
||
name: str = "默认配置"
|
||
acme_server: str = "https://acme-v02.api.letsencrypt.org/directory"
|
||
email: str = ""
|
||
dns_provider: str = "aliyun"
|
||
dns_credentials: str = "{}"
|
||
renew_days: int = 30
|
||
|
||
|
||
class AcmeConfigUpdate(BaseModel):
|
||
name: str | None = None
|
||
acme_server: str | None = None
|
||
email: str | None = None
|
||
dns_provider: str | None = None
|
||
dns_credentials: str | None = None
|
||
renew_days: int | None = None
|
||
|
||
|
||
def _serialize_acme_config(c: AcmeConfig) -> dict:
|
||
return {
|
||
"id": c.id,
|
||
"name": c.name,
|
||
"acme_server": c.acme_server,
|
||
"email": c.email,
|
||
"dns_provider": c.dns_provider,
|
||
"dns_credentials": c.dns_credentials,
|
||
"renew_days": c.renew_days,
|
||
"has_account_key": c.account_key is not None,
|
||
"domain_count": len(c.domains) if c.domains else 0,
|
||
}
|
||
|
||
|
||
@router.get("/acme/configs")
|
||
async def list_acme_configs(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(AcmeConfig).options(selectinload(AcmeConfig.domains)).order_by(AcmeConfig.id))
|
||
configs = result.scalars().all()
|
||
return [_serialize_acme_config(c) for c in configs]
|
||
|
||
|
||
@router.get("/acme/configs/{config_id}")
|
||
async def get_acme_config(config_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == config_id))
|
||
c = result.scalar_one_or_none()
|
||
if not c:
|
||
raise HTTPException(404, "ACME config not found")
|
||
return {
|
||
"id": c.id,
|
||
"name": c.name,
|
||
"acme_server": c.acme_server,
|
||
"email": c.email,
|
||
"dns_provider": c.dns_provider,
|
||
"dns_credentials": c.dns_credentials,
|
||
"renew_days": c.renew_days,
|
||
"has_account_key": c.account_key is not None,
|
||
}
|
||
|
||
|
||
@router.post("/acme/configs", status_code=201)
|
||
async def create_acme_config(data: AcmeConfigCreate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
config = AcmeConfig(**data.model_dump())
|
||
db.add(config)
|
||
await db.flush()
|
||
await db.refresh(config)
|
||
return {"id": config.id, "name": config.name}
|
||
|
||
|
||
@router.put("/acme/configs/{config_id}")
|
||
async def update_acme_config(config_id: int, data: AcmeConfigUpdate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == config_id))
|
||
config = result.scalar_one_or_none()
|
||
if not config:
|
||
raise HTTPException(404, "ACME config not found")
|
||
for key, value in data.model_dump(exclude_unset=True).items():
|
||
setattr(config, key, value)
|
||
return {"id": config.id}
|
||
|
||
|
||
@router.delete("/acme/configs/{config_id}")
|
||
async def delete_acme_config(config_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == config_id))
|
||
config = result.scalar_one_or_none()
|
||
if not config:
|
||
raise HTTPException(404, "ACME config not found")
|
||
# 解除关联的域名
|
||
result = await db.execute(select(Domain).where(Domain.acme_config_id == config_id))
|
||
for d in result.scalars().all():
|
||
d.acme_config_id = None
|
||
await db.delete(config)
|
||
return {"ok": True}
|
||
|
||
|
||
# ──────────────── ACME 操作 ────────────────
|
||
|
||
async def _get_domain_acme_config(domain: Domain, db: AsyncSession) -> AcmeConfig:
|
||
"""获取域名关联的 ACME 配置，未关联则使用第一个可用配置"""
|
||
if domain.acme_config_id:
|
||
result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == domain.acme_config_id))
|
||
config = result.scalar_one_or_none()
|
||
if config:
|
||
return config
|
||
# 回退：使用第一个有邮箱的配置
|
||
result = await db.execute(select(AcmeConfig).where(AcmeConfig.email != "").order_by(AcmeConfig.id))
|
||
config = result.scalars().first()
|
||
if not config:
|
||
raise HTTPException(400, "无可用的 ACME 配置，请先配置邮箱")
|
||
return config
|
||
|
||
|
||
@router.post("/acme/issue/{domain_id}")
|
||
async def issue_cert(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
"""为指定域名申请证书"""
|
||
from backend.acme_service import AcmeService
|
||
from backend.config import get_settings
|
||
|
||
# 获取域名
|
||
result = await db.execute(select(Domain).where(Domain.id == domain_id))
|
||
domain = result.scalar_one_or_none()
|
||
if not domain:
|
||
raise HTTPException(404, "Domain not found")
|
||
|
||
# 获取该域名关联的 ACME 配置
|
||
config = await _get_domain_acme_config(domain, db)
|
||
|
||
settings = get_settings()
|
||
service = AcmeService(config, settings.cert_dir)
|
||
|
||
# 记录开始
|
||
log = AcmeLog(domain_id=domain_id, action="issue", status="pending", message=f"Issuing certificate for {domain.domain}")
|
||
db.add(log)
|
||
await db.flush()
|
||
|
||
try:
|
||
success, msg = service.issue_certificate(domain.domain)
|
||
log.status = "success" if success else "failed"
|
||
log.message = msg
|
||
|
||
if success:
|
||
info = service.get_cert_info(domain.domain)
|
||
if info:
|
||
domain.cert_not_after = datetime.fromisoformat(info["not_after"])
|
||
domain.version = str(int(domain.version or "0") + 1)
|
||
|
||
await db.flush()
|
||
return {"success": success, "message": msg}
|
||
except Exception as e:
|
||
log.status = "failed"
|
||
log.message = str(e)
|
||
await db.flush()
|
||
raise HTTPException(500, str(e))
|
||
|
||
|
||
@router.post("/acme/renew/{domain_id}")
|
||
async def renew_cert(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
"""续签指定域名的证书"""
|
||
return await issue_cert(domain_id, db)
|
||
|
||
|
||
@router.post("/acme/auto-renew")
|
||
async def auto_renew_all(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
"""自动续签所有即将过期的证书"""
|
||
from backend.acme_service import AcmeService
|
||
from backend.config import get_settings
|
||
|
||
settings = get_settings()
|
||
|
||
# 获取所有域名
|
||
result = await db.execute(select(Domain).options(selectinload(Domain.acme_config)))
|
||
domains = result.scalars().all()
|
||
|
||
# 按 ACME 配置分组，为每个配置创建一个 service 实例
|
||
services: dict[int, AcmeService] = {}
|
||
|
||
results = []
|
||
for d in domains:
|
||
try:
|
||
config = await _get_domain_acme_config(d, db)
|
||
except HTTPException:
|
||
results.append({"domain": d.domain, "success": False, "message": "无可用 ACME 配置", "days_left": 0})
|
||
continue
|
||
|
||
if config.id not in services:
|
||
services[config.id] = AcmeService(config, settings.cert_dir)
|
||
service = services[config.id]
|
||
|
||
need_renew, days_left = service.check_expiry(d.domain)
|
||
if need_renew:
|
||
log = AcmeLog(domain_id=d.id, action="renew", status="pending", message=f"Auto-renewing {d.domain} (config: {config.name})")
|
||
db.add(log)
|
||
await db.flush()
|
||
|
||
success, msg = service.renew_certificate(d.domain)
|
||
log.status = "success" if success else "failed"
|
||
log.message = msg
|
||
|
||
if success:
|
||
info = service.get_cert_info(d.domain)
|
||
if info:
|
||
d.cert_not_after = datetime.fromisoformat(info["not_after"])
|
||
d.version = str(int(d.version or "0") + 1)
|
||
|
||
results.append({"domain": d.domain, "success": success, "message": msg, "days_left": days_left})
|
||
else:
|
||
results.append({"domain": d.domain, "success": True, "message": f"Skipped, {days_left} days left", "days_left": days_left})
|
||
|
||
return {"results": results}
|
||
|
||
|
||
# ──────────────── ACME 日志 ────────────────
|
||
|
||
@router.get("/acme/logs")
|
||
async def list_acme_logs(status: str | None = None, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
query = (
|
||
select(AcmeLog)
|
||
.options(selectinload(AcmeLog.domain))
|
||
.order_by(AcmeLog.created_at.desc())
|
||
.limit(100)
|
||
)
|
||
if status:
|
||
query = query.where(AcmeLog.status == status)
|
||
result = await db.execute(query)
|
||
logs = result.scalars().all()
|
||
return [
|
||
{
|
||
"id": log.id,
|
||
"domain": log.domain.domain if log.domain else None,
|
||
"action": log.action,
|
||
"status": log.status,
|
||
"message": log.message,
|
||
"detail": log.detail,
|
||
"created_at": log.created_at.isoformat(),
|
||
}
|
||
for log in logs
|
||
]
|
||
|
||
|
||
# ──────────────── 部署脚本 ────────────────
|
||
|
||
async def _render_script(domain: str, server_name: str, os: str, db: AsyncSession) -> tuple[str, Server, Domain]:
|
||
"""渲染部署脚本，返回 (script, server, domain_obj)"""
|
||
from jinja2 import Environment, FileSystemLoader
|
||
|
||
template_dir = Path(__file__).parent.parent / "templates_cert"
|
||
jinja_env = Environment(loader=FileSystemLoader(str(template_dir)))
|
||
|
||
result = await db.execute(select(Server).where(Server.name == server_name))
|
||
server = result.scalar_one_or_none()
|
||
if not server:
|
||
raise HTTPException(404, "Server not found")
|
||
|
||
result = await db.execute(
|
||
select(Domain).where(Domain.domain == domain, Domain.server_id == server.id)
|
||
)
|
||
d = result.scalar_one_or_none()
|
||
if not d:
|
||
raise HTTPException(404, "Domain not found for this server")
|
||
|
||
settings = get_settings()
|
||
template_name = "deploy-cert.ps1.j2" if os == "windows" else "deploy-cert.sh.j2"
|
||
tpl = jinja_env.get_template(template_name)
|
||
script = tpl.render(
|
||
domain=domain,
|
||
base_url=settings.base_url,
|
||
token=server.token,
|
||
cert_dir=d.cert_dir,
|
||
check_cmd=d.check_cmd,
|
||
reload_cmd=d.reload_cmd,
|
||
)
|
||
return script, server, d
|
||
|
||
|
||
@router.get("/script")
|
||
async def admin_generate_script(
|
||
domain: str,
|
||
server_name: str,
|
||
os: str = "linux",
|
||
db: AsyncSession = Depends(get_db),
|
||
_: bool = Depends(require_auth),
|
||
):
|
||
"""管理端生成部署脚本（JSON）"""
|
||
script, server, d = await _render_script(domain, server_name, os, db)
|
||
return {
|
||
"script": script,
|
||
"token": server.token,
|
||
"cert_dir": d.cert_dir,
|
||
"check_cmd": d.check_cmd,
|
||
"reload_cmd": d.reload_cmd,
|
||
}
|
||
|
||
|
||
@router.get("/script/download")
|
||
async def admin_download_script(
|
||
domain: str,
|
||
server_name: str,
|
||
os: str = "linux",
|
||
db: AsyncSession = Depends(get_db),
|
||
_: bool = Depends(require_auth),
|
||
):
|
||
"""下载脚本文件（纯文本，供 curl 使用）"""
|
||
script, _, _ = await _render_script(domain, server_name, os, db)
|
||
ext = "ps1" if os == "windows" else "sh"
|
||
filename = f"deploy-{domain.replace('*', '_')}.{ext}"
|
||
return PlainTextResponse(
|
||
script,
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
@router.get("/script/setup")
|
||
async def admin_setup_script(
|
||
request: Request,
|
||
domain: str,
|
||
server_name: str,
|
||
os: str = "linux",
|
||
token: str = "",
|
||
db: AsyncSession = Depends(get_db),
|
||
_: bool = Depends(require_auth),
|
||
):
|
||
"""生成一键安装脚本（纯文本下载）"""
|
||
from jinja2 import Environment, FileSystemLoader
|
||
|
||
template_dir = Path(__file__).parent.parent / "templates_cert"
|
||
jinja_env = Environment(loader=FileSystemLoader(str(template_dir)))
|
||
|
||
from urllib.parse import quote
|
||
settings = get_settings()
|
||
base_url = str(request.base_url).rstrip("/")
|
||
ext = "ps1" if os == "windows" else "sh"
|
||
download_url = f"{base_url}/admin/api/script/download?domain={quote(domain)}&server_name={quote(server_name)}&os={os}"
|
||
|
||
template_name = "setup-cert.ps1.j2" if os == "windows" else "setup-cert.sh.j2"
|
||
tpl = jinja_env.get_template(template_name)
|
||
script = tpl.render(
|
||
domain=domain,
|
||
server_name=server_name,
|
||
download_url=download_url,
|
||
admin_token=token,
|
||
)
|
||
filename = f"setup-{domain.replace('*', '_')}.{ext}"
|
||
return PlainTextResponse(
|
||
script,
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
# ──────────────── 证书信息 ────────────────
|
||
|
||
@router.get("/cert-info/{domain_id}")
|
||
async def get_cert_info(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
"""获取证书详细信息"""
|
||
from backend.acme_service import AcmeService
|
||
from backend.config import get_settings
|
||
|
||
result = await db.execute(select(Domain).where(Domain.id == domain_id))
|
||
domain = result.scalar_one_or_none()
|
||
if not domain:
|
||
raise HTTPException(404, "Domain not found")
|
||
|
||
config = await _get_domain_acme_config(domain, db)
|
||
|
||
settings = get_settings()
|
||
service = AcmeService(config, settings.cert_dir)
|
||
|
||
info = service.get_cert_info(domain.domain)
|
||
need_renew, days_left = service.check_expiry(domain.domain)
|
||
|
||
return {
|
||
"domain": domain.domain,
|
||
"cert_info": info,
|
||
"need_renew": need_renew,
|
||
"days_left": days_left,
|
||
}
|
||
|
||
|
||
# ──────────────── 证书下载 ────────────────
|
||
|
||
def _cert_store_dir(domain: str) -> str:
|
||
"""泛域名用裸域名作为存储目录"""
|
||
return domain[2:] if domain.startswith("*.") else domain
|
||
|
||
|
||
@router.get("/cert-download/{domain_id}/{file_type}")
|
||
async def download_cert_file(domain_id: int, file_type: str, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
|
||
"""下载证书文件，file_type: fullchain 或 private"""
|
||
from fastapi.responses import FileResponse
|
||
|
||
if file_type not in ("fullchain", "private"):
|
||
raise HTTPException(400, "file_type must be fullchain or private")
|
||
|
||
result = await db.execute(select(Domain).where(Domain.id == domain_id))
|
||
domain = result.scalar_one_or_none()
|
||
if not domain:
|
||
raise HTTPException(404, "Domain not found")
|
||
|
||
settings = get_settings()
|
||
store_dir = _cert_store_dir(domain.domain)
|
||
|
||
if file_type == "fullchain":
|
||
file_path = Path(settings.cert_dir) / store_dir / "fullchain.pem"
|
||
filename = f"{store_dir}-fullchain.pem"
|
||
else:
|
||
file_path = Path(settings.cert_dir) / store_dir / "private.key"
|
||
filename = f"{store_dir}-private.key"
|
||
|
||
logger.debug(f"下载证书: domain={domain.domain}, store_dir={store_dir}, file_path={file_path}, exists={file_path.exists()}")
|
||
|
||
if not file_path.exists():
|
||
raise HTTPException(404, f"文件不存在 ({file_path})，请先申请证书")
|
||
|
||
return FileResponse(
|
||
str(file_path),
|
||
media_type="application/x-pem-file",
|
||
filename=filename,
|
||
)
|