# -*- coding: utf-8 -*- """ @File : issue_certificate_api.py @Date : 2023-07-23 """ import json import requests from flask import g, request from playhouse.shortcuts import model_to_dict, chunked from domain_admin.enums.role_enum import RoleEnum from domain_admin.model.dns_model import DnsModel from domain_admin.model.domain_model import DomainModel from domain_admin.model.host_model import HostModel from domain_admin.model.issue_certificate_model import IssueCertificateModel, ChallengeDeployTypeEnum, \ SSLDeployTypeEnum, DeployStatusEnum from domain_admin.service import issue_certificate_service, auth_service from domain_admin.utils import ip_util, domain_util, fabric_util, datetime_util, validate_util from domain_admin.utils.acme_util.challenge_type import ChallengeType from domain_admin.utils.acme_util.key_type_enum import KEY_TYPE_OPTIONS, KeyTypeEnum from domain_admin.utils.acme_util.directory_type_enum import DIRECTORY_URL_OPTIONS, DirectoryTypeEnum from domain_admin.utils.flask_ext.app_exception import AppException, DataNotFoundAppException from domain_admin.utils.open_api import aliyun_domain_api from domain_admin.utils.open_api.aliyun_domain_api import RecordTypeEnum @auth_service.permission(role=RoleEnum.USER) def issue_certificate(): """ 发起申请 :return: """ current_user_id = g.user_id domains = request.json['domains'] directory_type = request.json.get('directory_type') or DirectoryTypeEnum.LETS_ENCRYPT key_type = request.json.get('key_type') or KeyTypeEnum.RSA issue_certificate_id = issue_certificate_service.issue_certificate( domains=domains, user_id=current_user_id, directory_type=directory_type, key_type=key_type ) issue_certificate_row = IssueCertificateModel.get_by_id(issue_certificate_id) return issue_certificate_row.to_dict() @auth_service.permission(role=RoleEnum.USER) def verify_certificate(): """ 通知验证 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] challenge_type = request.json['challenge_type'] # 验证文件部署信息 可选参数 verify_deploy_path = request.json.get('verify_deploy_path') host_id = request.json.get('host_id') dns_id = request.json.get('dns_id') # 验证成功后，自动添加到证书监控列表 issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() issue_certificate_service.verify_certificate( issue_certificate_id=issue_certificate_id, challenge_type=challenge_type, verify_deploy_path=verify_deploy_path, host_id=host_id, dns_id=dns_id ) if host_id and verify_deploy_path: IssueCertificateModel.update( challenge_deploy_type_id=ChallengeDeployTypeEnum.SSH, challenge_deploy_id=host_id, deploy_verify_path=verify_deploy_path, challenge_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() if dns_id: IssueCertificateModel.update( challenge_deploy_type_id=ChallengeDeployTypeEnum.DNS, challenge_deploy_id=dns_id, challenge_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() issue_certificate_service.renew_certificate(issue_certificate_id) # fix: 过滤通配符的域名 lst = [ { 'domain': domain, 'root_domain': domain_util.get_root_domain(domain), 'port': 443, 'alias': '', 'user_id': current_user_id, 'group_id': 0, } for domain in issue_certificate_row.domains if validate_util.is_domain(domain) ] for batch in chunked(lst, 10): DomainModel.insert_many(batch).on_conflict_ignore().execute() @auth_service.permission(role=RoleEnum.USER) def get_certificate_challenges(): current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() lst = issue_certificate_service.get_certificate_challenges(issue_certificate_id) return { 'total': len(lst), 'list': lst } @auth_service.permission(role=RoleEnum.USER) def get_domain_host(): domain = request.json['domain'] host = ip_util.get_domain_ip(domain) return { 'domain': domain, 'host': host } @auth_service.permission(role=RoleEnum.USER) def deploy_verify_file(): """ 部署验证文件 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] verify_deploy_path = request.json['verify_deploy_path'] challenges = request.json['challenges'] host_id = request.json['host_id'] if not verify_deploy_path.endswith("/"): raise AppException("verify_deploy_path must endswith '/'") # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() # deploy issue_certificate_service.deploy_verify_file( host_id=host_id, verify_deploy_path=verify_deploy_path, challenges=challenges ) IssueCertificateModel.update( challenge_deploy_type_id=ChallengeDeployTypeEnum.SSH, challenge_deploy_id=host_id, deploy_verify_path=verify_deploy_path, challenge_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() @auth_service.permission(role=RoleEnum.USER) def deploy_certificate_file(): """ ssh方式部署证书文件 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] host_id = request.json['host_id'] key_deploy_path = request.json['key_deploy_path'] pem_deploy_path = request.json['pem_deploy_path'] reload_cmd = request.json['reloadcmd'] # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() # host_row = HostModel.get_by_id(host_id) # data check host_row = HostModel.select().where( HostModel.id == host_id, HostModel.user_id == current_user_id, ).first() if not host_row: raise DataNotFoundAppException() host = host_row.host user = host_row.user password = host_row.password # issue_certificate_row = IssueCertificateModel.get_by_id(issue_certificate_id) if not issue_certificate_row.ssl_certificate: issue_certificate_service.renew_certificate(issue_certificate_id) # issue_certificate_row = IssueCertificateModel.get_by_id(issue_certificate_id) # deploy key issue_certificate_row = IssueCertificateModel.get_by_id(issue_certificate_id) issue_certificate_service.deploy_certificate_file( host_id=host_id, key_content=issue_certificate_row.ssl_certificate_key, pem_content=issue_certificate_row.ssl_certificate, key_deploy_path=key_deploy_path, pem_deploy_path=pem_deploy_path, reload_cmd=reload_cmd ) # update only support file verify # if issue_certificate_row.challenge_type == ChallengeType.HTTP01: # is_auto_renew = True # else: # is_auto_renew = False IssueCertificateModel.update( deploy_type_id=SSLDeployTypeEnum.SSH, deploy_host_id=host_id, deploy_key_file=key_deploy_path, deploy_fullchain_file=pem_deploy_path, deploy_reloadcmd=reload_cmd, ssl_deploy_status=DeployStatusEnum.SUCCESS, ).where( IssueCertificateModel.id == issue_certificate_id ).execute() # 验证成功后, check_auto_renew issue_certificate_service.check_auto_renew( issue_certificate_id=issue_certificate_id ) @auth_service.permission(role=RoleEnum.USER) def renew_certificate(): """ 发起申请 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() issue_certificate_service.renew_certificate(issue_certificate_id) issue_certificate_row = IssueCertificateModel.get_by_id(issue_certificate_id) if not issue_certificate_row: raise AppException('数据不存在') return issue_certificate_row.to_dict() @auth_service.permission(role=RoleEnum.USER) def get_issue_certificate_list(): """ 发起申请 :return: """ current_user_id = g.user_id page = request.json.get('page', 1) size = request.json.get('size', 10) keyword = request.json.get('keyword') query = IssueCertificateModel.select().where( IssueCertificateModel.user_id == current_user_id ) if keyword: query = query.where(IssueCertificateModel.domain_raw.contains(keyword)) total = query.count() rows = query.order_by( IssueCertificateModel.update_time.desc(), IssueCertificateModel.id.desc() ).paginate(page, size) lst = [] for row in rows: item = model_to_dict( row, extra_attrs=[ 'domains', 'create_time_label', 'update_time_label', 'start_date', 'expire_date', 'has_ssl_certificate', 'can_auto_renew', ], exclude=[ IssueCertificateModel.ssl_certificate, IssueCertificateModel.ssl_certificate_key ]) # 关联信息 item['challenge_deploy_name'] = '-' if row.challenge_deploy_type_id == ChallengeDeployTypeEnum.SSH: host = HostModel.get_or_none(HostModel.id == row.challenge_deploy_id) if host: item['challenge_deploy_name'] = host.host elif row.challenge_deploy_type_id == ChallengeDeployTypeEnum.DNS: dns = DnsModel.get_or_none(DnsModel.id == row.challenge_deploy_id) if dns: item['challenge_deploy_name'] = dns.name item['deploy_host_name'] = '-' if row.deploy_type_id == SSLDeployTypeEnum.SSH: host = HostModel.get_or_none(HostModel.id == row.deploy_host_id) if host: item['deploy_host_name'] = host.host elif row.deploy_type_id in [SSLDeployTypeEnum.OSS, SSLDeployTypeEnum.CDN, SSLDeployTypeEnum.DCDN]: dns = DnsModel.get_or_none(DnsModel.id == row.deploy_host_id) if dns: item['deploy_host_name'] = dns.name lst.append(item) return { 'list': lst, 'total': total, } @auth_service.permission(role=RoleEnum.USER) def get_issue_certificate_by_id(): """ 获取 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] # check data issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id ).first() if not issue_certificate_row: raise DataNotFoundAppException() data = issue_certificate_row.to_dict() data['deploy_dns'] = None data['deploy_host'] = None data['cert_deploy_host'] = None data['cert_deploy_dns'] = None if issue_certificate_row.challenge_deploy_type_id == ChallengeDeployTypeEnum.SSH: host_row = HostModel.get_or_none(HostModel.id == issue_certificate_row.challenge_deploy_id) data['deploy_host'] = host_row if host_row and host_row.dns_id: data['deploy_dns'] = DnsModel.get_or_none(DnsModel.id == host_row.dns_id) elif issue_certificate_row.challenge_deploy_type_id == ChallengeDeployTypeEnum.DNS: data['deploy_dns'] = DnsModel.get_or_none(DnsModel.id == issue_certificate_row.challenge_deploy_id) if issue_certificate_row.deploy_type_id == SSLDeployTypeEnum.SSH: data['cert_deploy_host'] = HostModel.get_or_none(HostModel.id == issue_certificate_row.deploy_host_id) elif issue_certificate_row.deploy_type_id in [SSLDeployTypeEnum.OSS, SSLDeployTypeEnum.CDN, SSLDeployTypeEnum.DCDN]: data['cert_deploy_dns'] = DnsModel.get_or_none(DnsModel.id == issue_certificate_row.deploy_host_id) return data @auth_service.permission(role=RoleEnum.USER) def delete_issue_certificate_by_id(): """ 获取 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] # check data issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id ).first() if not issue_certificate_row: raise DataNotFoundAppException() IssueCertificateModel.delete_by_id(issue_certificate_row.id) @auth_service.permission(role=RoleEnum.USER) def delete_certificate_by_batch(): """ 批量删除 @since v1.2.16 :return: """ current_user_id = g.user_id ids = request.json['ids'] IssueCertificateModel.delete().where( IssueCertificateModel.id.in_(ids), IssueCertificateModel.user_id == current_user_id ).execute() @auth_service.permission(role=RoleEnum.USER) def renew_issue_certificate_by_id(): """ 手动续期SSL证书 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] # check data issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id ).first() if not issue_certificate_row: raise DataNotFoundAppException() issue_certificate_service.renew_certificate_row(issue_certificate_row) @auth_service.permission(role=RoleEnum.USER) def get_allow_commands(): """ 命令白名单 :return: """ return fabric_util.allow_commands @auth_service.permission(role=RoleEnum.USER) def notify_web_hook(): """ 用户调用webhook :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] url = request.json['url'] headers = request.json.get('headers') # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() ret = issue_certificate_service.deploy_ssl_by_web_hook( issue_certificate_id=issue_certificate_id, url=url, headers=headers, ) # 更新验证信息 IssueCertificateModel.update( deploy_type_id=SSLDeployTypeEnum.WEB_HOOK, deploy_url=url, deploy_header_raw=json.dumps(headers or {}), ssl_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() # 验证成功后, check_auto_renew issue_certificate_service.check_auto_renew( issue_certificate_id=issue_certificate_id ) return ret @auth_service.permission(role=RoleEnum.USER) def deploy_cert_to_oss(): """ 部署证书到阿里云oss :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] dns_id = request.json['dns_id'] # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() ret = issue_certificate_service.deploy_cert_to_oss( issue_certificate_id=issue_certificate_id, dns_id=dns_id, ) # 更新验证信息 IssueCertificateModel.update( deploy_type_id=SSLDeployTypeEnum.OSS, deploy_host_id=dns_id, ssl_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() # 验证成功后, check_auto_renew issue_certificate_service.check_auto_renew( issue_certificate_id=issue_certificate_id ) return ret @auth_service.permission(role=RoleEnum.USER) def deploy_cert_to_cdn(): """ 部署证书到阿里云cdn :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] dns_id = request.json['dns_id'] # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() ret = issue_certificate_service.deploy_cert_to_cdn( issue_certificate_id=issue_certificate_id, dns_id=dns_id, ) # 更新验证信息 IssueCertificateModel.update( deploy_type_id=SSLDeployTypeEnum.CDN, deploy_host_id=dns_id, ssl_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() # 验证成功后, check_auto_renew issue_certificate_service.check_auto_renew( issue_certificate_id=issue_certificate_id ) return ret @auth_service.permission(role=RoleEnum.USER) def deploy_cert_to_dcdn(): """ 部署证书到阿里云dcdn :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] dns_id = request.json['dns_id'] # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() ret = issue_certificate_service.deploy_cert_to_dcdn( issue_certificate_id=issue_certificate_id, dns_id=dns_id, ) # 更新验证信息 IssueCertificateModel.update( deploy_type_id=SSLDeployTypeEnum.DCDN, deploy_host_id=dns_id, ssl_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() # 验证成功后, check_auto_renew issue_certificate_service.check_auto_renew( issue_certificate_id=issue_certificate_id ) return ret @auth_service.permission(role=RoleEnum.USER) def add_dns_domain_record(): """ 添加dns记录 :return: """ current_user_id = g.user_id dns_id = request.json['dns_id'] issue_certificate_id = request.json['issue_certificate_id'] # data check issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id, ).first() if not issue_certificate_row: raise DataNotFoundAppException() challenges = issue_certificate_service.get_certificate_challenges(issue_certificate_id) # 添加txt记录 issue_certificate_service.add_dns_domain_record( dns_id=dns_id, challenges=challenges ) # 更新验证信息 IssueCertificateModel.update( challenge_deploy_type_id=ChallengeDeployTypeEnum.DNS, challenge_deploy_id=dns_id, challenge_deploy_status=DeployStatusEnum.SUCCESS ).where( IssueCertificateModel.id == issue_certificate_id ).execute() @auth_service.permission(role=RoleEnum.USER) def update_row_auto_renew(): """ 修改自动更新字段 :return: """ current_user_id = g.user_id issue_certificate_id = request.json['issue_certificate_id'] is_auto_renew = request.json['is_auto_renew'] # check data issue_certificate_row = IssueCertificateModel.select().where( IssueCertificateModel.id == issue_certificate_id, IssueCertificateModel.user_id == current_user_id ).first() if not issue_certificate_row: raise DataNotFoundAppException() if issue_certificate_row and issue_certificate_row.can_auto_renew: # 更新验证信息 IssueCertificateModel.update( is_auto_renew=is_auto_renew ).where( IssueCertificateModel.id == issue_certificate_id ).execute() else: raise AppException("不支持自动续期") @auth_service.permission(role=RoleEnum.USER) def get_issue_certificate_options(): """ 获取常量 :return: """ return { 'KEY_TYPE_OPTIONS': KEY_TYPE_OPTIONS, 'DIRECTORY_URL_OPTIONS': DIRECTORY_URL_OPTIONS }