acme-auto/backend/routers/api.py

from fastapi import APIRouter, Depends, HTTPException, Header
from fastapi.responses import PlainTextResponse, FileResponse
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from jinja2 import Environment, FileSystemLoader
from pathlib import Path
import os

from backend.database import get_db
from backend.models import Server, Domain, DeployLog
from backend.config import get_settings

router = APIRouter(tags=["client-api"])

settings = get_settings()

# Jinja2 环境，用于生成脚本
template_dir = Path(__file__).parent.parent / "templates_cert"
jinja_env = Environment(loader=FileSystemLoader(str(template_dir)))


async def verify_token(
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db),
):
    """验证 Bearer Token，返回对应的 Server 对象"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization header")
    token = authorization[7:]
    result = await db.execute(select(Server).where(Server.token == token))
    server = result.scalar_one_or_none()
    if not server:
        raise HTTPException(status_code=401, detail="Invalid token")
    return server


@router.get("/version/{domain}")
async def get_version(domain: str, db: AsyncSession = Depends(get_db)):
    """返回当前证书版本号（无需认证，方便客户端轻量检查）"""
    result = await db.execute(select(Domain).where(Domain.domain == domain))
    d = result.scalar_one_or_none()
    if not d:
        raise HTTPException(status_code=404, detail="Domain not found")
    return PlainTextResponse(d.version)


@router.get("/cert/{domain}/fullchain")
async def get_fullchain(
    domain: str,
    server: Server = Depends(verify_token),
    db: AsyncSession = Depends(get_db),
):
    """下载 fullchain.pem"""
    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(status_code=404, detail="Domain not found for this server")
    cert_path = Path(settings.cert_dir) / domain / "fullchain.pem"
    if not cert_path.exists():
        raise HTTPException(status_code=404, detail="Certificate file not found")
    return FileResponse(str(cert_path), media_type="application/x-pem-file", filename="fullchain.pem")


@router.get("/cert/{domain}/private")
async def get_private_key(
    domain: str,
    server: Server = Depends(verify_token),
    db: AsyncSession = Depends(get_db),
):
    """下载 private.key"""
    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(status_code=404, detail="Domain not found for this server")
    key_path = Path(settings.cert_dir) / domain / "private.key"
    if not key_path.exists():
        raise HTTPException(status_code=404, detail="Private key file not found")
    return FileResponse(str(key_path), media_type="application/x-pem-file", filename="private.key")


@router.get("/script/{domain}")
async def generate_script(
    domain: str,
    server_name: str,
    db: AsyncSession = Depends(get_db),
):
    """根据 server_name 生成对应的部署脚本"""
    # 查找服务器
    result = await db.execute(select(Server).where(Server.name == server_name))
    server = result.scalar_one_or_none()
    if not server:
        raise HTTPException(status_code=404, detail="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(status_code=404, detail="Domain not found for this server")

    # 选择模板
    if server.platform == "windows":
        template_name = "deploy-cert.ps1.j2"
    else:
        template_name = "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,
    )

    media_type = "text/plain" if server.platform == "windows" else "application/x-sh"
    return PlainTextResponse(script, media_type=media_type)


@router.post("/report/{domain}")
async def report_deploy(
    domain: str,
    status: str,
    message: str = "",
    server: Server = Depends(verify_token),
    db: AsyncSession = Depends(get_db),
):
    """客户端上报部署结果"""
    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(status_code=404, detail="Domain not found for this server")

    log = DeployLog(domain_id=d.id, server_id=server.id, status=status, message=message)
    db.add(log)
    return {"ok": True}