申请Let's Encrypt免费SSL证书API接口详细教程

<?php
/**
 * Let's Encrypt SSL证书申请PHP示例
 */
class LetsEncryptAPI {
    private $apiUrl = 'https://cn.apihz.cn/api/ssl/letsencrypt.php';
    private $userId = '88888888'; // 替换为您的用户ID
    private $userKey = '88888888'; // 替换为您的用户KEY
    
    /**
     * 申请证书订单
     * @param string $domain 要申请证书的域名
     * @return array 返回结果数组
     */
    public function applyOrder($domain) {
        $params = [
            'id' => $this->userId,
            'key' => $this->userKey,
            'type' => 1,
            'domain' => $domain,
            'order' => ''
        ];
        
        $url = $this->apiUrl . '?' . http_build_query($params);
        $response = file_get_contents($url);
        
        return json_decode($response, true);
    }
    
    /**
     * 获取证书
     * @param string $orderId 订单号
     * @return array 返回结果数组
     */
    public function getCertificate($orderId) {
        $params = [
            'id' => $this->userId,
            'key' => $this->userKey,
            'type' => 2,
            'domain' => '',
            'order' => $orderId
        ];
        
        $url = $this->apiUrl . '?' . http_build_query($params);
        $response = file_get_contents($url);
        
        return json_decode($response, true);
    }
    
    /**
     * 查询历史订单
     * @param string $orderId 订单号
     * @return array 返回结果数组
     */
    public function queryOrder($orderId) {
        $params = [
            'id' => $this->userId,
            'key' => $this->userKey,
            'type' => 3,
            'domain' => '',
            'order' => $orderId
        ];
        
        $url = $this->apiUrl . '?' . http_build_query($params);
        $response = file_get_contents($url);
        
        return json_decode($response, true);
    }
    
    /**
     * 保存证书文件
     * @param array $certData 证书数据
     * @param string $domain 域名
     * @return bool 是否保存成功
     */
    public function saveCertificate($certData, $domain) {
        if (!isset($certData['data'])) {
            return false;
        }
        
        $sanitizedDomain = preg_replace('/[^a-zA-Z0-9\.\-]/', '_', $domain);
        $timestamp = date('YmdHis');
        
        // 保存私钥
        $privateKeyFile = "{$sanitizedDomain}_{$timestamp}.key";
        file_put_contents($privateKeyFile, $certData['data']['zskey']);
        
        // 保存证书
        $certificateFile = "{$sanitizedDomain}_{$timestamp}.pem";
        file_put_contents($certificateFile, $certData['data']['zspem']);
        
        // 保存证书信息
        $infoFile = "{$sanitizedDomain}_{$timestamp}.info";
        $info = [
            'domain' => $certData['data']['domain'],
            'expires_at' => $certData['data']['expires_at'],
            'files' => [
                'private_key' => $privateKeyFile,
                'certificate' => $certificateFile
            ]
        ];
        file_put_contents($infoFile, json_encode($info, JSON_PRETTY_PRINT));
        
        return true;
    }
}
// 使用示例
$api = new LetsEncryptAPI();
// 示例1：申请证书订单
$domain = 'example.com';
$applyResult = $api->applyOrder($domain);
if ($applyResult['code'] == 200) {
    echo "申请成功！\n";
    echo "订单号: " . $applyResult['order'] . "\n";
    echo "DNS记录: " . $applyResult['dnsurl'] . "\n";
    echo "DNS类型: " . $applyResult['dnstype'] . "\n";
    echo "DNS值: " . $applyResult['dnsvalue'] . "\n";
    
    // 提示用户添加DNS解析
    echo "\n请到域名服务商处添加以下DNS解析：\n";
    echo "类型: " . $applyResult['dnstype'] . "\n";
    echo "主机记录: " . $applyResult['dnsurl'] . "\n";
    echo "记录值: " . $applyResult['dnsvalue'] . "\n";
    
    // 保存订单号供后续使用
    $orderId = $applyResult['order'];
    file_put_contents('order_id.txt', $orderId);
} else {
    echo "申请失败: " . $applyResult['msg'] . "\n";
}
// 示例2：获取证书（在DNS验证通过后执行）
$orderId = file_get_contents('order_id.txt');
if ($orderId) {
    // 可以多次尝试获取，直到成功
    $maxRetries = 10;
    $retryCount = 0;
    
    while ($retryCount < $maxRetries) {
        $certResult = $api->getCertificate($orderId);
        
        if ($certResult['code'] == 200 && isset($certResult['data'])) {
            echo "获取证书成功！\n";
            echo "域名: " . $certResult['data']['domain'] . "\n";
            echo "到期时间: " . $certResult['data']['expires_at'] . "\n";
            
            // 保存证书文件
            if ($api->saveCertificate($certResult, $domain)) {
                echo "证书文件已保存\n";
            }
            break;
        } else {
            echo "第" . ($retryCount + 1) . "次尝试: " . $certResult['msg'] . "\n";
            $retryCount++;
            sleep(30); // 等待30秒后重试
        }
    }
    
    if ($retryCount >= $maxRetries) {
        echo "获取证书失败，请检查DNS解析是否生效\n";
    }
}

<?php
/**
 * POST方式调用示例
 */
function postRequest($data) {
    $apiUrl = 'https://cn.apihz.cn/api/ssl/letsencrypt.php';
    
    $options = [
        'http' => [
            'header' => "Content-type: application/x-www-form-urlencoded\r\n",
            'method' => 'POST',
            'content' => http_build_query($data),
            'timeout' => 30 // 30秒超时
        ]
    ];
    
    $context = stream_context_create($options);
    $result = file_get_contents($apiUrl, false, $context);
    
    return json_decode($result, true);
}
// 申请订单
$applyData = [
    'id' => '88888888',
    'key' => '88888888',
    'type' => 1,
    'domain' => 'ssl.apihz.cn',
    'order' => ''
];
$result = postRequest($applyData);
print_r($result);

"""
Let's Encrypt SSL证书申请Python示例
支持Python 3.6+
"""
import json
import time
import requests
from typing import Dict, Optional, Tuple
import re
from datetime import datetime
class LetsEncryptAPIClient:
    """Let's Encrypt API客户端"""
    
    def __init__(self, user_id: str, user_key: str):
        """
        初始化客户端
        
        Args:
            user_id: 用户ID
            user_key: 用户通讯秘钥
        """
        self.api_url = "https://cn.apihz.cn/api/ssl/letsencrypt.php"
        self.user_id = user_id
        self.user_key = user_key
        self.session = requests.Session()
        self.session.timeout = 30  # 30秒超时
        
    def _make_request(self, params: Dict) -> Dict:
        """
        发送API请求
        
        Args:
            params: 请求参数
            
        Returns:
            解析后的JSON响应
        """
        try:
            # 同时支持GET和POST
            response = self.session.post(self.api_url, data=params)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {
                "code": 400,
                "msg": f"请求失败: {str(e)}"
            }
        except json.JSONDecodeError:
            return {
                "code": 400,
                "msg": "响应解析失败"
            }
    
    def apply_order(self, domain: str) -> Dict:
        """
        申请证书订单
        
        Args:
            domain: 要申请证书的域名
            
        Returns:
            申请结果
        """
        params = {
            "id": self.user_id,
            "key": self.user_key,
            "type": "1",
            "domain": domain,
            "order": ""
        }
        return self._make_request(params)
    
    def get_certificate(self, order_id: str) -> Dict:
        """
        获取证书
        
        Args:
            order_id: 订单号
            
        Returns:
            证书信息
        """
        params = {
            "id": self.user_id,
            "key": self.user_key,
            "type": "2",
            "domain": "",
            "order": order_id
        }
        return self._make_request(params)
    
    def query_order(self, order_id: str) -> Dict:
        """
        查询历史订单
        
        Args:
            order_id: 订单号
            
        Returns:
            订单信息
        """
        params = {
            "id": self.user_id,
            "key": self.user_key,
            "type": "3",
            "domain": "",
            "order": order_id
        }
        return self._make_request(params)
    
    def save_certificate(self, cert_data: Dict, domain: str) -> Tuple[bool, str]:
        """
        保存证书文件
        
        Args:
            cert_data: 证书数据
            domain: 域名
            
        Returns:
            (是否成功, 错误信息)
        """
        if "data" not in cert_data:
            return False, "证书数据不存在"
        
        try:
            # 清理域名用于文件名
            safe_domain = re.sub(r'[^a-zA-Z0-9\.\-]', '_', domain)
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            
            # 保存私钥
            private_key_file = f"{safe_domain}_{timestamp}.key"
            with open(private_key_file, 'w', encoding='utf-8') as f:
                f.write(cert_data['data']['zskey'])
            
            # 保存证书
            certificate_file = f"{safe_domain}_{timestamp}.pem"
            with open(certificate_file, 'w', encoding='utf-8') as f:
                f.write(cert_data['data']['zspem'])
            
            # 保存证书信息
            info_file = f"{safe_domain}_{timestamp}.info.json"
            info = {
                "domain": cert_data['data']['domain'],
                "expires_at": cert_data['data']['expires_at'],
                "files": {
                    "private_key": private_key_file,
                    "certificate": certificate_file
                },
                "saved_at": datetime.now().isoformat()
            }
            
            with open(info_file, 'w', encoding='utf-8') as f:
                json.dump(info, f, indent=2, ensure_ascii=False)
            
            return True, f"证书已保存: {private_key_file}, {certificate_file}"
            
        except Exception as e:
            return False, f"保存失败: {str(e)}"
    
    def auto_certificate_flow(self, domain: str, max_retries: int = 10, 
                             retry_interval: int = 30) -> Dict:
        """
        自动化证书申请流程
        
        Args:
            domain: 域名
            max_retries: 最大重试次数
            retry_interval: 重试间隔(秒)
            
        Returns:
            最终结果
        """
        result = {
            "success": False,
            "message": "",
            "order_id": "",
            "dns_info": {},
            "certificate_files": []
        }
        
        # 1. 申请订单
        print(f"正在为域名 {domain} 申请证书订单...")
        apply_result = self.apply_order(domain)
        
        if apply_result.get("code") != 200:
            result["message"] = f"申请失败: {apply_result.get('msg', '未知错误')}"
            return result
        
        order_id = apply_result.get("order")
        result["order_id"] = order_id
        result["dns_info"] = {
            "dnsurl": apply_result.get("dnsurl"),
            "dnstype": apply_result.get("dnstype"),
            "dnsvalue": apply_result.get("dnsvalue")
        }
        
        print(f"✅ 申请成功！订单号: {order_id}")
        print("\n📋 DNS验证信息:")
        print(f"   记录类型: {apply_result.get('dnstype')}")
        print(f"   主机记录: {apply_result.get('dnsurl')}")
        print(f"   记录值: {apply_result.get('dnsvalue')}")
        print("\n请到您的域名服务商处添加以上DNS解析记录，然后按回车键继续...")
        input()
        
        # 2. 获取证书（带重试）
        print(f"\n正在尝试获取证书，最多尝试 {max_retries} 次...")
        
        for attempt in range(1, max_retries + 1):
            print(f"第 {attempt} 次尝试...")
            
            cert_result = self.get_certificate(order_id)
            
            if cert_result.get("code") == 200 and "data" in cert_result:
                print("✅ 获取证书成功！")
                
                # 保存证书文件
                success, message = self.save_certificate(cert_result, domain)
                if success:
                    result["success"] = True
                    result["message"] = "证书申请并保存成功"
                    result["certificate_files"] = [
                        f"{domain}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.key",
                        f"{domain}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pem"
                    ]
                    
                    print(f"📁 {message}")
                    print(f"🌐 域名: {cert_result['data']['domain']}")
                    print(f"⏰ 到期时间: {cert_result['data']['expires_at']}")
                else:
                    result["message"] = f"证书获取成功但保存失败: {message}"
                
                break
            else:
                print(f"❌ 尝试失败: {cert_result.get('msg', '未知错误')}")
                
                if attempt < max_retries:
                    print(f"等待 {retry_interval} 秒后重试...")
                    time.sleep(retry_interval)
                else:
                    result["message"] = f"获取证书失败，已尝试 {max_retries} 次"
        
        return result
# 使用示例
if __name__ == "__main__":
    # 配置信息
    USER_ID = "88888888"  # 替换为您的用户ID
    USER_KEY = "88888888"  # 替换为您的用户KEY
    DOMAIN = "example.com"  # 要申请证书的域名
    
    # 创建客户端
    client = LetsEncryptAPIClient(USER_ID, USER_KEY)
    
    # 示例1：完整自动化流程
    print("=" * 50)
    print("Let's Encrypt SSL证书自动化申请")
    print("=" * 50)
    
    result = client.auto_certificate_flow(DOMAIN)
    
    print("\n" + "=" * 50)
    print("申请结果:")
    print(f"成功: {result['success']}")
    print(f"消息: {result['message']}")
    if result['order_id']:
        print(f"订单号: {result['order_id']}")
    print("=" * 50)
    
    # 示例2：分步操作
    print("\n\n" + "=" * 50)
    print("分步操作示例")
    print("=" * 50)
    
    # 步骤1：申请订单
    apply_result = client.apply_order("test.example.com")
    if apply_result["code"] == 200:
        print("申请成功!")
        print(f"订单号: {apply_result['order']}")
        print(f"DNS记录: {apply_result['dnsurl']}")
        print(f"记录值: {apply_result['dnsvalue']}")
        
        # 保存订单号
        order_id = apply_result["order"]
        
        # 这里可以提示用户手动添加DNS解析
        # 然后继续执行步骤2
        
        # 步骤2：获取证书（示例）
        # cert_result = client.get_certificate(order_id)
        # if cert_result["code"] == 200:
        #     print("证书获取成功!")
        #     client.save_certificate(cert_result, "test.example.com")
    else:
        print(f"申请失败: {apply_result['msg']}")
    
    # 示例3：查询历史订单
    print("\n\n" + "=" * 50)
    print("查询历史订单示例")
    print("=" * 50)
    
    # 假设有一个已存在的订单号
    existing_order = "8e15b1d070280456ee60461663409db8"  # 示例订单号
    query_result = client.query_order(existing_order)
    
    if query_result["code"] == 200 and "data" in query_result:
        print("查询成功!")
        print(f"域名: {query_result['data']['domain']}")
        print(f"到期时间: {query_result['data']['expires_at']}")
    else:
        print(f"查询失败: {query_result.get('msg', '未知错误')}")

"""
高级功能：批量申请、自动续期监控
"""
import threading
from queue import Queue
import schedule
import time as time_module
from datetime import datetime, timedelta
class BatchCertificateManager:
    """批量证书管理器"""
    
    def __init__(self, client: LetsEncryptAPIClient):
        self.client = client
        self.results_queue = Queue()
    
    def batch_apply(self, domains: list, max_workers: int = 3):
        """
        批量申请证书
        
        Args:
            domains: 域名列表
            max_workers: 最大并发数
        """
        def worker(domain):
            try:
                result = client.auto_certificate_flow(domain, max_retries=5)
                self.results_queue.put((domain, result))
            except Exception as e:
                self.results_queue.put((domain, {
                    "success": False,
                    "message": f"异常: {str(e)}"
                }))
        
        threads = []
        for domain in domains:
            thread = threading.Thread(target=worker, args=(domain,))
            thread.start()
            threads.append(thread)
            
            # 控制并发数
            if len(threads) >= max_workers:
                for t in threads:
                    t.join()
                threads = []
        
        # 等待剩余线程
        for t in threads:
            t.join()
        
        # 收集结果
        results = []
        while not self.results_queue.empty():
            results.append(self.results_queue.get())
        
        return results
class CertificateRenewalMonitor:
    """证书续期监控器"""
    
    def __init__(self, client: LetsEncryptAPIClient, check_interval_days: int = 7):
        self.client = client
        self.check_interval = check_interval_days
        self.certificates_db = []  # 模拟证书数据库
        
    def load_certificates(self):
        """加载证书信息（示例）"""
        # 这里可以从文件或数据库加载
        self.certificates_db = [
            {"domain": "example1.com", "expires_at": "2026-07-01 12:00:00", "order_id": "xxx"},
            {"domain": "example2.com", "expires_at": "2026-07-15 14:30:00", "order_id": "yyy"},
        ]
    
    def check_expiring_certificates(self, days_threshold: int = 30):
        """
        检查即将过期的证书
        
        Args:
            days_threshold: 过期阈值（天）
            
        Returns:
            需要续期的证书列表
        """
        now = datetime.now()
        threshold_date = now + timedelta(days=days_threshold)
        
        expiring = []
        for cert in self.certificates_db:
            expires_at = datetime.strptime(cert["expires_at"], "%Y-%m-%d %H:%M:%S")
            
            if expires_at <= threshold_date:
                days_left = (expires_at - now).days
                expiring.append({
                    "domain": cert["domain"],
                    "expires_at": cert["expires_at"],
                    "days_left": days_left,
                    "order_id": cert.get("order_id")
                })
        
        return expiring
    
    def auto_renew_certificate(self, domain: str):
        """自动续期证书"""
        print(f"开始续期证书: {domain}")
        result = self.client.auto_certificate_flow(domain)
        
        if result["success"]:
            print(f"✅ 证书续期成功: {domain}")
            # 更新数据库
            self._update_certificate_in_db(domain, result)
        else:
            print(f"❌ 证书续期失败: {domain} - {result['message']}")
        
        return result
    
    def _update_certificate_in_db(self, domain: str, result: Dict):
        """更新证书信息到数据库"""
        for i, cert in enumerate(self.certificates_db):
            if cert["domain"] == domain:
                self.certificates_db[i]["expires_at"] = result.get("new_expires_at", "")
                self.certificates_db[i]["order_id"] = result.get("order_id", "")
                break
    
    def start_monitoring(self):
        """启动监控服务"""
        print("启动证书续期监控服务...")
        
        # 每天检查一次
        schedule.every().day.at("02:00").do(self._monitoring_task)
        
        try:
            while True:
                schedule.run_pending()
                time_module.sleep(60)  # 每分钟检查一次
        except KeyboardInterrupt:
            print("监控服务已停止")
    
    def _monitoring_task(self):
        """监控任务"""
        print(f"[{datetime.now()}] 执行证书检查...")
        
        expiring_certs = self.check_expiring_certificates(30)  # 30天内过期
        
        if expiring_certs:
            print(f"发现 {len(expiring_certs)} 个即将过期的证书:")
            for cert in expiring_certs:
                print(f"  - {cert['domain']}: {cert['days_left']}天后过期")
                
                # 自动续期（示例：只续期15天内过期的）
                if cert['days_left'] <= 15:
                    self.auto_renew_certificate(cert['domain'])
        else:
            print("没有即将过期的证书")
# 使用高级功能
if __name__ == "__main__":
    # 创建客户端
    client = LetsEncryptAPIClient("88888888", "88888888")
    
    # 批量申请示例
    print("批量申请示例")
    print("=" * 50)
    
    manager = BatchCertificateManager(client)
    domains_to_apply = [
        "site1.example.com",
        "site2.example.com",
        "site3.example.com"
    ]
    
    # 注意：实际使用时应控制并发数，避免触发频率限制
    # results = manager.batch_apply(domains_to_apply, max_workers=2)
    # for domain, result in results:
    #     print(f"{domain}: {'成功' if result['success'] else '失败'} - {result['message']}")
    
    # 续期监控示例
    print("\n\n续期监控示例")
    print("=" * 50)
    
    monitor = CertificateRenewalMonitor(client)
    monitor.load_certificates()
    
    # 检查即将过期的证书
    expiring = monitor.check_expiring_certificates(60)
    print(f"60天内过期的证书数量: {len(expiring)}")
    
    # 启动监控服务（在实际应用中可能需要以服务形式运行）
    # monitor.start_monitoring()