cert-manager/app/main.py

import asyncio
import logging
from contextlib import asynccontextmanager
from datetime import datetime, timezone

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from fastapi import FastAPI, HTTPException, Request, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel

from . import apisix, certbot, cloudflare
from .alert import send_discord_dm
from .config import AppConfig, load_config, validate_domain

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

config: AppConfig = None  # type: ignore
scheduler = AsyncIOScheduler()
security = HTTPBearer()


# --- Auth ---

async def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)) -> str:
    if not config.api_token:
        raise HTTPException(status_code=500, detail="API token not configured")
    if credentials.credentials != config.api_token:
        raise HTTPException(status_code=401, detail="Invalid token")
    return credentials.credentials


# --- Request/Response models ---

class DomainRequest(BaseModel):
    domain: str
    instances: list[str] | None = None


class SyncResponse(BaseModel):
    results: list[dict]


# --- Alert helper ---

async def _alert(message: str):
    """Discord DM으로 알림. 실패해도 무시."""
    try:
        await send_discord_dm(config.discord_bot_token, config.discord_alert_user_id, message)
    except Exception as e:
        logger.warning("Alert send failed: %s", e)


# --- Scheduled task ---

async def auto_renew():
    """만료 30일 이내 인증서 자동 갱신 + APISIX 재배포."""
    global config
    # 설정 리로드 (Vault 토큰 만료 대응)
    try:
        config = load_config()
    except Exception as e:
        logger.error("Config reload failed: %s", e)
        await _alert(f"[cert-manager] Config reload failed: {e}")
        return

    logger.info("Starting auto-renewal check")
    certs = certbot.list_certificates(config)
    now = datetime.now(timezone.utc)
    failures = []

    for cert in certs:
        if cert["days_remaining"] <= 30:
            domain = cert["domain"]
            logger.info("Renewing %s (expires in %d days)", domain, cert["days_remaining"])
            result = await certbot.issue_certificate(domain, config)
            if result["success"]:
                # domain_instance_map에 따라 대상 인스턴스 결정
                target_instances = _resolve_instances(domain)
                await apisix.deploy_certificate(domain, config, instances=target_instances)
            else:
                error = result.get("error", "Unknown")
                logger.error("Renewal failed for %s: %s", domain, error)
                failures.append(f"{domain}: {error}")

    if failures:
        msg = "[cert-manager] Renewal failures:\n" + "\n".join(failures)
        await _alert(msg)

    logger.info("Auto-renewal check completed")


def _resolve_instances(domain: str) -> list | None:
    """domain_instance_map에서 도메인에 맞는 APISIX 인스턴스 목록 반환."""
    if not config.domain_instance_map:
        return None  # 매핑 없으면 전체

    for pattern, instance_names in config.domain_instance_map.items():
        if domain == pattern or domain.endswith(f".{pattern.lstrip('*.')}"):
            matched = [i for i in config.apisix_instances if i.name in instance_names]
            if matched:
                return matched
    return None  # 매핑에 없으면 전체


# --- Lifespan ---

@asynccontextmanager
async def lifespan(app: FastAPI):
    global config
    config = load_config()
    scheduler.add_job(auto_renew, CronTrigger(hour=3, minute=0), id="auto_renew")
    scheduler.start()
    logger.info("Scheduler started")
    yield
    scheduler.shutdown()


app = FastAPI(title="cert-manager", lifespan=lifespan)


# --- Endpoints ---

@app.get("/health")
async def health():
    return {"status": "ok", "timestamp": datetime.now(timezone.utc).isoformat()}


@app.get("/domains")
async def get_domains(token: str = Security(verify_token)):
    try:
        domains = await cloudflare.list_domains(config)
        return {"domains": domains, "count": len(domains)}
    except Exception as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.get("/certificates")
async def get_certificates(token: str = Security(verify_token)):
    certs = certbot.list_certificates(config)
    return {"certificates": certs, "count": len(certs)}


@app.get("/certificates/{domain}")
async def get_certificate_detail(domain: str, token: str = Security(verify_token)):
    """특정 도메인의 인증서 정보 + PEM 내용 반환."""
    error = validate_domain(domain)
    if error:
        raise HTTPException(status_code=400, detail=error)

    info = certbot.get_certificate_info(domain, config)
    if not info:
        raise HTTPException(status_code=404, detail=f"Certificate not found for {domain}")
    from pathlib import Path
    cert_pem = Path(info["cert_path"]).read_text()
    key_pem = Path(info["key_path"]).read_text()
    return {
        **info,
        "cert": cert_pem,
        "key": key_pem,
    }


@app.post("/certificates/issue")
async def issue_certificate(req: DomainRequest, token: str = Security(verify_token)):
    error = validate_domain(req.domain)
    if error:
        raise HTTPException(status_code=400, detail=error)

    result = await certbot.issue_certificate(req.domain, config)
    if not result["success"]:
        raise HTTPException(status_code=500, detail=result.get("error", "Unknown error"))
    return result


@app.post("/certificates/deploy")
async def deploy_certificate(req: DomainRequest, token: str = Security(verify_token)):
    error = validate_domain(req.domain)
    if error:
        raise HTTPException(status_code=400, detail=error)

    targets = None
    if req.instances:
        targets = [i for i in config.apisix_instances if i.name in req.instances]
        if not targets:
            raise HTTPException(status_code=400, detail=f"Unknown instances: {req.instances}")
    results = await apisix.deploy_certificate(req.domain, config, instances=targets)
    failures = [r for r in results if not r["success"]]
    if failures and len(failures) == len(results):
        raise HTTPException(status_code=500, detail="All deployments failed")
    return {"domain": req.domain, "results": results}


async def _sync_one(domain: str) -> dict:
    """단일 도메인 발급 + 배포."""
    logger.info("Syncing %s", domain)
    issue_result = await certbot.issue_certificate(domain, config)
    if not issue_result["success"]:
        return {"domain": domain, "issue": issue_result, "deploy": None}
    target_instances = _resolve_instances(domain)
    deploy_results = await apisix.deploy_certificate(domain, config, instances=target_instances)
    return {"domain": domain, "issue": issue_result, "deploy": deploy_results}


@app.post("/certificates/sync")
async def sync_all(token: str = Security(verify_token)):
    """전체 도메인 조회 → 3개씩 병렬 발급+배포."""
    domains = await cloudflare.list_domains(config)
    sem = asyncio.Semaphore(3)

    async def _limited(domain: str):
        async with sem:
            return await _sync_one(domain)

    results = await asyncio.gather(
        *[_limited(z["name"]) for z in domains],
        return_exceptions=True,
    )

    # 예외를 dict로 변환
    final = []
    failures = []
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            domain = domains[i]["name"]
            final.append({"domain": domain, "error": str(r)})
            failures.append(f"{domain}: {r}")
        else:
            final.append(r)

    if failures:
        await _alert("[cert-manager] Sync failures:\n" + "\n".join(failures))

    return {"results": final, "total": len(final)}