结合route_acme.pyACME 协议逻辑和route_dns.pyDNS 钩子逻辑代码我们可以把证书申请流程拆解成一个完整的“柜台办证”过程。在这个过程中你的代码扮演的是发证机关CA 厂商的角色而 AllinSSL 扮演的是办证人客户端。第一阶段初始化进店咨询1./directory—— 咨询服务台功能作用这是 AllinSSL 访问你的第一站。它像一张“服务导览图”告诉客户端后续每个步骤该去哪个 URL 办理。代码定位返回了newNonce、newOrder等关键接口的地址。2./new-nonce—— 领取防伪号牌功能作用ACME 协议为了防范重放攻击防止别人截获你的报文再次发送要求每个请求都带上一个随机数Nonce。代码定位你的代码通过Response的headers返回了一个Replay-Nonce。AllinSSL 拿到这个号牌后才能在下一步请求里“盖章”签名。3./new-account—— 注册会员功能作用AllinSSL 告诉 CA“我是谁这是我的账号公钥以后我说的话都用我的私钥签名。”代码定位你的代码返回了状态valid通过并在 Header 中给了它一个账号 IDaccount/1。第二阶段下单与审核提交申请4./new-order—— 填写办证申请表功能作用AllinSSL 正式提交申请说明要为哪个域名www.liu.com办证。代码定位关键动作代码从请求体里解码出identifiers域名信息并原样返回。后续指引接口返回了两个非常重要的后续链接authorizations去哪证明域名是你的和finalize审核通过后去哪领证。5./authz/{auth_id}—— 领取身份挑战任务功能作用发证机关你给申请人AllinSSL出题。代码定位你告诉 AllinSSL“请使用dns-01方式验证并在 DNS 里填入我给你的这个token。”第三阶段证明与验证自证清白6./dnswebhook(在 route_dns.py 中) —— 贴公告功能作用这是 AllinSSL 平台在收到你给的token后转头来调用这个接口让你帮它把“公告”贴到 DNS 上。代码定位它接收action创建/删除和value验证码并返回code: 200。虽然你这里只是模拟打印了日志但 AllinSSL 只要看到成功返回就会认为公告已经贴好了。7./chall/{chall_id}—— 申请查看结果功能作用AllinSSL 贴完公告后回来敲门“任务做完了你们快去查公告栏DNS吧”代码定位你的代码直接返回了status: valid。这就相当于你根本没去查公告栏直接给它开了“绿灯”。第四阶段签发与领证打印证件8./finalize/{order_id}—— 盖章签发功能作用这是最核心的一步。验证通过后AllinSSL 会把它的证书公钥封装在 CSR 里发给你求你盖章。代码定位解析 CSR代码从 Base64 编码中提取出 CSR。动态签名调用_sign_csr_to_pem函数用你伪造的_ca_key给这个 CSR 盖章生成真正的.pem格式证书字符串。缓存证书把生成的证书存入_cert_store内存字典中并给出一个下载地址。9./cert/{cert_id}—— 窗口领证功能作用AllinSSL 顺着上一步给的地址把签发好的证书下载回去。代码定位从_cert_store内存中取出 PEM 字符串并返回。总结你的代码逻辑串联route_acme.py负责流程调度它设定了规矩ACME 协议。它负责盖章使用cryptography库进行 RSA 签名。route_dns.py负责辅助执行它是 AllinSSL 为了完成route_acme.py布置的任务DNS 验证而调用的工具接口。代码内容route_acme.pyimportjsonimportbase64importdatetimefromcryptographyimportx509fromcryptography.x509.oidimportNameOIDfromcryptography.hazmat.primitivesimporthashes,serializationfromcryptography.hazmat.primitives.asymmetricimportrsafromfastapiimportAPIRouter,Request,Responsefromsrc.model.resultimportResultfromsrc.utils.loggerimportget_logger loggerget_logger()routerAPIRouter(tags[acme])# --- 配置区 ---BASE_URLhttps://came.pool7.yun100.cnACME_PREFIX/acmeFULL_BASEf{BASE_URL}{ACME_PREFIX}# --- 伪 CA 密钥模块加载时生成一次重启后不变签发历史会失效---_ca_keyrsa.generate_private_key(public_exponent65537,key_size2048)_ca_namex509.Name([x509.NameAttribute(NameOID.COMMON_NAME,Fake CA)])# 内存证书存储order_id - cert PEM str_cert_store:dict[str,str]{}def_sign_csr_to_pem(csr:x509.CertificateSigningRequest)-str:用伪 CA 密钥对 CSR 签发证书返回 PEM 字符串nowdatetime.datetime.utcnow()cert(x509.CertificateBuilder().subject_name(csr.subject).issuer_name(_ca_name).public_key(csr.public_key()).serial_number(x509.random_serial_number()).not_valid_before(now).not_valid_after(nowdatetime.timedelta(days365)).add_extension(x509.BasicConstraints(caFalse,path_lengthNone),criticalTrue).sign(_ca_key,hashes.SHA256()))returncert.public_bytes(serialization.Encoding.PEM).decode()# --- ACME 协议实现 ---router.get(/directory)asyncdefdirectory():ACME 入口点所有路径都带上 /acme 前缀logger.info(ACME directory endpoint called)return{newNonce:f{FULL_BASE}/new-nonce,newAccount:f{FULL_BASE}/new-account,newOrder:f{FULL_BASE}/new-order,finalize:f{FULL_BASE}/finalize,newAuthz:f{FULL_BASE}/new-authz,meta:{termsOfService:http://example.com/tos}}router.head(/new-nonce)router.get(/new-nonce)asyncdefget_nonce():returnResponse(status_code204,# No Contentheaders{Replay-Nonce:static-nonce-for-allinssl,Cache-Control:no-store,Link:https://came.pool7.yun100.cn/acme/directory;relindex# 顺便返回索引链接})router.post(/new-account)asyncdefnew_account(response:Response):logger.info(ACME new account endpoint called)response.headers[Location]f{FULL_BASE}/account/1return{status:valid}router.post(/new-order)asyncdefcreate_order(request:Request):# 1. 获取 lego 发过来的 JWS 请求体bodyawaitrequest.json()# 2. 解码 payload (ACME 协议中 payload 是 base64url 编码的)payload_b64body.get(payload,)# 注意这里需要处理 base64url 的填充问题missing_paddinglen(payload_b64)%4ifmissing_padding:payload_b64*(4-missing_padding)payload_jsonjson.loads(base64.urlsafe_b64decode(payload_b64))requested_identifierspayload_json.get(identifiers,[])logger.info(fClient requested order for:{requested_identifiers})# 3. 构造符合 RFC 8555 标准的响应order_data{status:pending,expires:2026-12-31T23:59:59Z,identifiers:requested_identifiers,# 重点这里必须和请求的一模一样authorizations:[f{FULL_BASE}/authz/some-random-id],finalize:f{FULL_BASE}/finalize/some-random-id}returnResponse(contentjson.dumps(order_data),status_code201,# ACME 创建资源通常返回 201headers{Replay-Nonce:next-nonce-value,# 别忘了每个响应都要给新 NonceLocation:https://came.pool7.yun100.cn/acme/order/some-order-id})router.post(/authz/{auth_id})asyncdefget_authorization(auth_id:str):logger.info(fChecking authorization for ID:{auth_id})# 构造响应数据# 注意identifier 的 value 最好和你 new-order 时的一致auth_data{status:pending,expires:2026-12-31T23:59:59Z,identifier:{type:dns,value:www.liu.com},challenges:[{type:dns-01,# 告诉 lego 使用 DNS 验证status:pending,url:fhttps://came.pool7.yun100.cn/acme/chall/{auth_id},# 下一个坑位token:this-is-a-fake-token-for-validation}]}returnResponse(contentjson.dumps(auth_data),status_code200,headers{Replay-Nonce:even-more-nonces,# 核心每个 POST 响应都要给新 NonceContent-Type:application/json})router.post(/chall/{chall_id})asyncdefchallenge_ack(chall_id:str):当 allinssl 配置完 DNS 后来这里触发验证logger.info(fACME challenge endpoint called, chall_id{chall_id})# 既然是伪造适配器我们直接返回有效让流程继续return{status:valid}router.post(/finalize/{order_id})asyncdeffinalize(order_id:str,request:Request):关键环节解析 AllinSSL 提交的 CSR动态签发证书logger.info(fACME finalize endpoint called, order_id{order_id})bodyawaitrequest.json()# 1. 解码 JWS payload提取 CSRpayload_b64body.get(payload,)missinglen(payload_b64)%4ifmissing:payload_b64*(4-missing)payloadjson.loads(base64.urlsafe_b64decode(payload_b64))csr_b64payload.get(csr,)missinglen(csr_b64)%4ifmissing:csr_b64*(4-missing)csr_derbase64.urlsafe_b64decode(csr_b64)# 2. 解析 CSR用 CSR 里的公钥签发证书csrx509.load_der_x509_csr(csr_der)cert_pem_sign_csr_to_pem(csr)# 3. 存入内存供 get_cert 接口取用_cert_store[order_id]cert_pem logger.info(f证书已签发并缓存order_id{order_id})return{status:valid,certificate:f{FULL_BASE}/cert/{order_id}}router.get(/cert/{cert_id})router.post(/cert/{cert_id})asyncdefget_cert(cert_id:str):返回 finalize 阶段动态签发的证书logger.info(fACME certificate download endpoint called, cert_id{cert_id})cert_pem_cert_store.get(cert_id)ifnotcert_pem:logger.error(f证书未找到cert_id{cert_id})returnResponse(status_code404,contentcert not found)returnResponse(contentcert_pem,media_typeapplication/pem-certificate-chain)router.get(/test,response_modelResult)asyncdeftest():try:returnResult.success(successs, hehe)exceptExceptionase:logger.error(fCollector endpoint error:{e},exc_infoTrue)returnResult.fail(str(e))route_dns.pyfromfastapiimportAPIRouter,Request,Responsefromsrc.utils.loggerimportget_logger loggerget_logger()router_dnsAPIRouter(tags[dns])router_dns.post(/dnswebhook)asyncdefdns_web_hook(request:Request):# 1. 获取 AllinSSL 发过来的数据try:dataawaitrequest.json()exceptException:return{code:400,msg:Invalid JSON}actiondata.get(action)# create 或 deletedomaindata.get(full_domain)# _acme-challenge.www.liu.comvaluedata.get(value)# 验证值# 2. 打印日志方便你观察 AllinSSL 到底传了什么ifactioncreate:logger.info(f[AllinSSL] 收到创建请求: 域名{domain}, 记录值{value})elifactiondelete:logger.info(f[AllinSSL] 收到删除请求: 域名{domain})# 3. 核心返回 AllinSSL 预期的成功格式# 大多数平台只要看到 code 200 或 success 就会继续下一步return{code:200,msg:success,data:{status:done}}
ACME 协议流程与AllinSSL 的关系(三)
结合route_acme.pyACME 协议逻辑和route_dns.pyDNS 钩子逻辑代码我们可以把证书申请流程拆解成一个完整的“柜台办证”过程。在这个过程中你的代码扮演的是发证机关CA 厂商的角色而 AllinSSL 扮演的是办证人客户端。第一阶段初始化进店咨询1./directory—— 咨询服务台功能作用这是 AllinSSL 访问你的第一站。它像一张“服务导览图”告诉客户端后续每个步骤该去哪个 URL 办理。代码定位返回了newNonce、newOrder等关键接口的地址。2./new-nonce—— 领取防伪号牌功能作用ACME 协议为了防范重放攻击防止别人截获你的报文再次发送要求每个请求都带上一个随机数Nonce。代码定位你的代码通过Response的headers返回了一个Replay-Nonce。AllinSSL 拿到这个号牌后才能在下一步请求里“盖章”签名。3./new-account—— 注册会员功能作用AllinSSL 告诉 CA“我是谁这是我的账号公钥以后我说的话都用我的私钥签名。”代码定位你的代码返回了状态valid通过并在 Header 中给了它一个账号 IDaccount/1。第二阶段下单与审核提交申请4./new-order—— 填写办证申请表功能作用AllinSSL 正式提交申请说明要为哪个域名www.liu.com办证。代码定位关键动作代码从请求体里解码出identifiers域名信息并原样返回。后续指引接口返回了两个非常重要的后续链接authorizations去哪证明域名是你的和finalize审核通过后去哪领证。5./authz/{auth_id}—— 领取身份挑战任务功能作用发证机关你给申请人AllinSSL出题。代码定位你告诉 AllinSSL“请使用dns-01方式验证并在 DNS 里填入我给你的这个token。”第三阶段证明与验证自证清白6./dnswebhook(在 route_dns.py 中) —— 贴公告功能作用这是 AllinSSL 平台在收到你给的token后转头来调用这个接口让你帮它把“公告”贴到 DNS 上。代码定位它接收action创建/删除和value验证码并返回code: 200。虽然你这里只是模拟打印了日志但 AllinSSL 只要看到成功返回就会认为公告已经贴好了。7./chall/{chall_id}—— 申请查看结果功能作用AllinSSL 贴完公告后回来敲门“任务做完了你们快去查公告栏DNS吧”代码定位你的代码直接返回了status: valid。这就相当于你根本没去查公告栏直接给它开了“绿灯”。第四阶段签发与领证打印证件8./finalize/{order_id}—— 盖章签发功能作用这是最核心的一步。验证通过后AllinSSL 会把它的证书公钥封装在 CSR 里发给你求你盖章。代码定位解析 CSR代码从 Base64 编码中提取出 CSR。动态签名调用_sign_csr_to_pem函数用你伪造的_ca_key给这个 CSR 盖章生成真正的.pem格式证书字符串。缓存证书把生成的证书存入_cert_store内存字典中并给出一个下载地址。9./cert/{cert_id}—— 窗口领证功能作用AllinSSL 顺着上一步给的地址把签发好的证书下载回去。代码定位从_cert_store内存中取出 PEM 字符串并返回。总结你的代码逻辑串联route_acme.py负责流程调度它设定了规矩ACME 协议。它负责盖章使用cryptography库进行 RSA 签名。route_dns.py负责辅助执行它是 AllinSSL 为了完成route_acme.py布置的任务DNS 验证而调用的工具接口。代码内容route_acme.pyimportjsonimportbase64importdatetimefromcryptographyimportx509fromcryptography.x509.oidimportNameOIDfromcryptography.hazmat.primitivesimporthashes,serializationfromcryptography.hazmat.primitives.asymmetricimportrsafromfastapiimportAPIRouter,Request,Responsefromsrc.model.resultimportResultfromsrc.utils.loggerimportget_logger loggerget_logger()routerAPIRouter(tags[acme])# --- 配置区 ---BASE_URLhttps://came.pool7.yun100.cnACME_PREFIX/acmeFULL_BASEf{BASE_URL}{ACME_PREFIX}# --- 伪 CA 密钥模块加载时生成一次重启后不变签发历史会失效---_ca_keyrsa.generate_private_key(public_exponent65537,key_size2048)_ca_namex509.Name([x509.NameAttribute(NameOID.COMMON_NAME,Fake CA)])# 内存证书存储order_id - cert PEM str_cert_store:dict[str,str]{}def_sign_csr_to_pem(csr:x509.CertificateSigningRequest)-str:用伪 CA 密钥对 CSR 签发证书返回 PEM 字符串nowdatetime.datetime.utcnow()cert(x509.CertificateBuilder().subject_name(csr.subject).issuer_name(_ca_name).public_key(csr.public_key()).serial_number(x509.random_serial_number()).not_valid_before(now).not_valid_after(nowdatetime.timedelta(days365)).add_extension(x509.BasicConstraints(caFalse,path_lengthNone),criticalTrue).sign(_ca_key,hashes.SHA256()))returncert.public_bytes(serialization.Encoding.PEM).decode()# --- ACME 协议实现 ---router.get(/directory)asyncdefdirectory():ACME 入口点所有路径都带上 /acme 前缀logger.info(ACME directory endpoint called)return{newNonce:f{FULL_BASE}/new-nonce,newAccount:f{FULL_BASE}/new-account,newOrder:f{FULL_BASE}/new-order,finalize:f{FULL_BASE}/finalize,newAuthz:f{FULL_BASE}/new-authz,meta:{termsOfService:http://example.com/tos}}router.head(/new-nonce)router.get(/new-nonce)asyncdefget_nonce():returnResponse(status_code204,# No Contentheaders{Replay-Nonce:static-nonce-for-allinssl,Cache-Control:no-store,Link:https://came.pool7.yun100.cn/acme/directory;relindex# 顺便返回索引链接})router.post(/new-account)asyncdefnew_account(response:Response):logger.info(ACME new account endpoint called)response.headers[Location]f{FULL_BASE}/account/1return{status:valid}router.post(/new-order)asyncdefcreate_order(request:Request):# 1. 获取 lego 发过来的 JWS 请求体bodyawaitrequest.json()# 2. 解码 payload (ACME 协议中 payload 是 base64url 编码的)payload_b64body.get(payload,)# 注意这里需要处理 base64url 的填充问题missing_paddinglen(payload_b64)%4ifmissing_padding:payload_b64*(4-missing_padding)payload_jsonjson.loads(base64.urlsafe_b64decode(payload_b64))requested_identifierspayload_json.get(identifiers,[])logger.info(fClient requested order for:{requested_identifiers})# 3. 构造符合 RFC 8555 标准的响应order_data{status:pending,expires:2026-12-31T23:59:59Z,identifiers:requested_identifiers,# 重点这里必须和请求的一模一样authorizations:[f{FULL_BASE}/authz/some-random-id],finalize:f{FULL_BASE}/finalize/some-random-id}returnResponse(contentjson.dumps(order_data),status_code201,# ACME 创建资源通常返回 201headers{Replay-Nonce:next-nonce-value,# 别忘了每个响应都要给新 NonceLocation:https://came.pool7.yun100.cn/acme/order/some-order-id})router.post(/authz/{auth_id})asyncdefget_authorization(auth_id:str):logger.info(fChecking authorization for ID:{auth_id})# 构造响应数据# 注意identifier 的 value 最好和你 new-order 时的一致auth_data{status:pending,expires:2026-12-31T23:59:59Z,identifier:{type:dns,value:www.liu.com},challenges:[{type:dns-01,# 告诉 lego 使用 DNS 验证status:pending,url:fhttps://came.pool7.yun100.cn/acme/chall/{auth_id},# 下一个坑位token:this-is-a-fake-token-for-validation}]}returnResponse(contentjson.dumps(auth_data),status_code200,headers{Replay-Nonce:even-more-nonces,# 核心每个 POST 响应都要给新 NonceContent-Type:application/json})router.post(/chall/{chall_id})asyncdefchallenge_ack(chall_id:str):当 allinssl 配置完 DNS 后来这里触发验证logger.info(fACME challenge endpoint called, chall_id{chall_id})# 既然是伪造适配器我们直接返回有效让流程继续return{status:valid}router.post(/finalize/{order_id})asyncdeffinalize(order_id:str,request:Request):关键环节解析 AllinSSL 提交的 CSR动态签发证书logger.info(fACME finalize endpoint called, order_id{order_id})bodyawaitrequest.json()# 1. 解码 JWS payload提取 CSRpayload_b64body.get(payload,)missinglen(payload_b64)%4ifmissing:payload_b64*(4-missing)payloadjson.loads(base64.urlsafe_b64decode(payload_b64))csr_b64payload.get(csr,)missinglen(csr_b64)%4ifmissing:csr_b64*(4-missing)csr_derbase64.urlsafe_b64decode(csr_b64)# 2. 解析 CSR用 CSR 里的公钥签发证书csrx509.load_der_x509_csr(csr_der)cert_pem_sign_csr_to_pem(csr)# 3. 存入内存供 get_cert 接口取用_cert_store[order_id]cert_pem logger.info(f证书已签发并缓存order_id{order_id})return{status:valid,certificate:f{FULL_BASE}/cert/{order_id}}router.get(/cert/{cert_id})router.post(/cert/{cert_id})asyncdefget_cert(cert_id:str):返回 finalize 阶段动态签发的证书logger.info(fACME certificate download endpoint called, cert_id{cert_id})cert_pem_cert_store.get(cert_id)ifnotcert_pem:logger.error(f证书未找到cert_id{cert_id})returnResponse(status_code404,contentcert not found)returnResponse(contentcert_pem,media_typeapplication/pem-certificate-chain)router.get(/test,response_modelResult)asyncdeftest():try:returnResult.success(successs, hehe)exceptExceptionase:logger.error(fCollector endpoint error:{e},exc_infoTrue)returnResult.fail(str(e))route_dns.pyfromfastapiimportAPIRouter,Request,Responsefromsrc.utils.loggerimportget_logger loggerget_logger()router_dnsAPIRouter(tags[dns])router_dns.post(/dnswebhook)asyncdefdns_web_hook(request:Request):# 1. 获取 AllinSSL 发过来的数据try:dataawaitrequest.json()exceptException:return{code:400,msg:Invalid JSON}actiondata.get(action)# create 或 deletedomaindata.get(full_domain)# _acme-challenge.www.liu.comvaluedata.get(value)# 验证值# 2. 打印日志方便你观察 AllinSSL 到底传了什么ifactioncreate:logger.info(f[AllinSSL] 收到创建请求: 域名{domain}, 记录值{value})elifactiondelete:logger.info(f[AllinSSL] 收到删除请求: 域名{domain})# 3. 核心返回 AllinSSL 预期的成功格式# 大多数平台只要看到 code 200 或 success 就会继续下一步return{code:200,msg:success,data:{status:done}}
