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, db_add
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.scalars().first()
    if not server:
        raise HTTPException(status_code=401, detail="Invalid token")
    return server


async def _find_domain(db: AsyncSession, domain: str, server_id: int | None = None, acme_config_id: int | None = None) -> Domain:
    """根据条件查找域名记录，acme_config_id 精确匹配"""
    conditions = [Domain.domain == domain]
    if server_id is not None:
        conditions.append(Domain.server_id == server_id)
    if acme_config_id is not None:
        conditions.append(Domain.acme_config_id == acme_config_id)
    result = await db.execute(select(Domain).where(*conditions))
    d = result.scalars().first()
    if not d:
        raise HTTPException(status_code=404, detail="Domain not found")
    return d


@router.get("/version")
async def get_version(domain: str, acme_config_id: int | None = None, db: AsyncSession = Depends(get_db)):
    """返回当前证书版本号（无需认证，方便客户端轻量检查）"""
    d = await _find_domain(db, domain, acme_config_id=acme_config_id)
    return PlainTextResponse(d.version)


def _cert_store_dir(domain: str) -> str:
    """泛域名用裸域名作为存储目录"""
    return domain[2:] if domain.startswith("*.") else domain


@router.get("/cert/fullchain")
async def get_fullchain(
    domain: str,
    acme_config_id: int | None = None,
    server: Server = Depends(verify_token),
    db: AsyncSession = Depends(get_db),
):
    """下载 fullchain.pem"""
    await _find_domain(db, domain, server_id=server.id, acme_config_id=acme_config_id)
    cert_path = Path(settings.cert_dir) / _cert_store_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/private")
async def get_private_key(
    domain: str,
    acme_config_id: int | None = None,
    server: Server = Depends(verify_token),
    db: AsyncSession = Depends(get_db),
):
    """下载 private.key"""
    await _find_domain(db, domain, server_id=server.id, acme_config_id=acme_config_id)
    key_path = Path(settings.cert_dir) / _cert_store_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")
async def generate_script(
    domain: str,
    server_name: str,
    acme_config_id: int | None = None,
    db: AsyncSession = Depends(get_db),
):
    """根据 server_name 生成对应的部署脚本"""
    # 查找服务器
    result = await db.execute(select(Server).where(Server.name == server_name))
    server = result.scalars().first()
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    # 查找域名配置
    d = await _find_domain(db, domain, server_id=server.id, acme_config_id=acme_config_id)

    # 选择模板
    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,
        acme_config_id=d.acme_config_id or "",
    )

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


@router.post("/report")
async def report_deploy(
    domain: str,
    status: str,
    message: str = "",
    hostname: str = "",
    acme_config_id: int | None = None,
    server: Server = Depends(verify_token),
    db: AsyncSession = Depends(get_db),
):
    """客户端上报部署结果"""
    d = await _find_domain(db, domain, server_id=server.id, acme_config_id=acme_config_id)

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