bt-source/panel/mod/project/ssl/sslMod.py

import json
import os.path

import public
from sslModel import base, dataModel
class main:

    # 获取dns-api列表和托管的域名
    def get_dns_api_and_domains(self, get=None):
        # 获取域名列表
        domain_list = public.M("ssl_domains").select()
        # 获取dns-api列表
        ssl_base_obj = base.sslBase()
        dns_api_data = ssl_base_obj.get_dns_data(None)

        # 宝塔域名
        dns_api_data.update({"bt": {"dns_name": "宝塔DNS", "id": "bt", "dns_type": "宝塔DNS", "domains": []}})

        for domain in domain_list:
            if not dns_api_data.get(domain["dns_id"], None):
                if "cloud_id=" in domain["dns_id"]:
                    dns_api_data["bt"]["domains"].append(domain["domain"])
                continue
            if not dns_api_data[domain["dns_id"]].get("domains", None):
                dns_api_data[domain["dns_id"]]["domains"] = []

            dns_api_data[domain["dns_id"]]["domains"].append(domain["domain"])

        dns_api_list = []
        for k in dns_api_data:
            if not dns_api_data[k].get("domains", None):
                dns_api_data[k]["domains"] = []
            # 移除敏感数据
            dns_api_data[k].pop("user_name", None)
            dns_api_data[k].pop("api_password", None)
            dns_api_data[k].pop("secret_id", None)
            dns_api_data[k].pop("secret_key", None)
            dns_api_data[k].pop("ID", None)
            dns_api_data[k].pop("Token", None)
            dns_api_data[k].pop("SecretKey", None)
            dns_api_data[k].pop("AccessKey", None)
            dns_api_data[k].pop("project_id", None)
            dns_api_data[k].pop("API Key", None)
            dns_api_data[k].pop("E-Mail", None)
            dns_api_list.insert(0, dns_api_data[k])

        dns_api_list.insert(0, {"dns_name": "手动解析", "id": "manual", "dns_type": "手动解析", "domains": []})

        return dns_api_list

    # 设置解析记录
    def set_dns_record(self, domain_name, dns_value, dns_id=None, record_type="TXT",sub_domain=""):
        if sub_domain:
            domain_name="{}.{}".format(sub_domain,domain_name)
        # 宝塔域名
        if dns_id == "bt":
            # 查询域名云端ID
            root, sub = public.split_domain_sld(domain_name)
            domain_data = public.M("ssl_domains").where("domain=?", (root,)).find()
            if not domain_data:
                return public.returnMsg(False, "域名【{}】非宝塔域名，请确定域名是在宝塔域名注册或已转入宝塔域名后，在域名管理中刷新本地缓存后重试！".format(root))
            dns_id = domain_data["dns_id"]
            if dns_id.find("cloud_id=") == -1:
                return public.returnMsg(False, "域名【{}】非宝塔域名，请确定域名是在宝塔域名注册或已转入宝塔域名后，在域名管理中刷新本地缓存后重试！".format(root))
            domain_id = dns_id.split("cloud_id=")[1]
            from mod.project.domain import domainMod
            domain_obj = domainMod.main()
            rep = domain_obj.request({"url": "/api/v1/dns/record/create", "domain_type": 1, "domain_id": domain_id,
                                      "record": sub, "value": dns_value, "type": record_type,
                                      "mx": 10, "ttl": 600, "remark": "", "view_id": 0})
            return rep
        elif dns_id == "manual":
            return public.returnMsg(True, "请手动添加DNS解析记录，解析类型：{}，解析值：{}".format(record_type, dns_value))
        else:
            ssl_data_obj = dataModel.main()
            args = {
                "fun_name": "create_dns_record",
                "dns_id": dns_id,
                "domain_dns_value": dns_value,
                "record_type": record_type,
                "domain_name": domain_name,
                "mx": 10,
            }
            return ssl_data_obj.run_fun(public.to_dict_obj(args))

    # 写验证文件
    def write_verify_file(self, domain_name, file_name, file_value):
        # 获取域名所在网站
        domain_data = public.M("domain").where("name=?", (domain_name,)).find()
        if not domain_data:
            return public.returnMsg(False, "域名【{}】未绑定到任何网站，请先将域名解析到本服务器并绑定网站后重试！".format(domain_name))
        site_data = public.M("sites").where("id=?", (domain_data["pid"],)).find()
        if not site_data:
            return public.returnMsg(False, "域名【{}】绑定的网站不存在，请先将域名解析到本服务器并绑定网站后重试！".format(domain_name))
        site_path = site_data["path"]
        if not site_path or not os.path.exists(site_path):
            return public.returnMsg(False, "域名【{}】绑定的网站根目录不存在，请先将域名解析到本服务器并绑定网站后重试！".format(domain_name))
        verify_path = os.path.join(site_path, file_name)
        if not os.path.exists(os.path.dirname(verify_path)):
            os.makedirs(os.path.dirname(verify_path), exist_ok=True)
        try:
            public.writeFile(verify_path, file_value)
            return public.returnMsg(True, "success")
        except Exception as e:
            return public.returnMsg(False, "写入验证文件【{}】失败，错误信息：{}".format(verify_path, str(e)))

    # 自动设置验证
    def auto_set_ssl_verify(self, get=None):
        if not "verify_type" in get or not get.verify_type:
            return public.returnMsg(False, "验证类型不能为空！")
        verify_type = get.verify_type
        if not "verify_data" in get or not get.verify_data:
            return public.returnMsg(False, "验证数据不能为空！")
        verify_data = get.verify_data.strip()
        # 是否设置验证值
        set_verify_value = False
        if "set_verify_value" in get:
           set_verify_value = get.set_verify_value
        try:
            verify_info = json.loads(verify_data)
        except Exception as e:
            return public.returnMsg(False, "验证数据格式错误，必须为JSON格式！")
        result = []
        if verify_type == "dns":
            for item in verify_info:
                if not item.get("domain", None):
                    return public.returnMsg(False, "域名不能为空！")
                domain_name = item["domain"].strip()
                # 特殊处理垦派多级域名的情况
                sub_domain = item.get("sub_domain", "")
                if domain_name.startswith("*."):
                    domain_name = domain_name[2:]
                root, sub = public.split_domain_sld(domain_name)
                if "_certum" in sub_domain and len(sub_domain.split(".")) > 1 and sub != "":
                    sub_domain = "_certum"

                record_type = item.get("record_type", "TXT")
                dns_value = item.get("dns_value", "")
                dns_id = item.get("dns_id", None)

                domain_name,_ = public.split_domain_sld(domain_name)

                # 本地验证
                local_verify = False
                try:
                    import dns.resolver
                    ns = dns.resolver.query(sub_domain+"."+domain_name, record_type)
                    for j in ns.response.answer:
                        for i in j.items:
                            txt_value = i.to_text().replace('"', '').strip()
                            # 特殊处理CNAME
                            if record_type.lower() == "cname" and txt_value[-1] == ".":
                                txt_value = txt_value[:-1]

                            public.print_log(txt_value)
                            if txt_value == dns_value:
                                local_verify = True
                except Exception as e:
                    pass
                rep = {"local_verify": local_verify, "status": None, "msg": None}
                if not local_verify and set_verify_value in (True, "True", "true"):
                    rep.update(self.set_dns_record(domain_name, dns_value, dns_id=dns_id, record_type=record_type, sub_domain=sub_domain))
                    try:
                        import dns.resolver
                        ns = dns.resolver.query(sub_domain + "." + domain_name, record_type)
                        for j in ns.response.answer:
                            for i in j.items:
                                txt_value = i.to_text().replace('"', '').strip()
                                if txt_value == dns_value:
                                    rep["local_verify"] = True
                    except Exception as e:
                        pass
                result.append({"domain": domain_name, "rep": rep})
        elif verify_type == "file":
            for item in verify_info:
                if not item.get("domain", None):
                    return public.returnMsg(False, "域名不能为空！")
                domain_name = item["domain"].strip()
                file_name = item.get("file_name", "")
                file_value = item.get("file_value", "")

                # 本地验证
                local_verify = False
                import requests
                url = "http://{}/{}".format(domain_name, file_name)
                try:
                    response = requests.get(url, timeout=5)

                    if response.status_code == 200:
                        if response.text.strip() == file_value.strip():
                            local_verify = True
                except Exception as e:
                    pass
                rep = {"local_verify": local_verify, "status": None, "msg": None}
                if not local_verify:
                    rep.update(self.write_verify_file(domain_name, file_name, file_value))
                else:
                    rep.update(public.returnMsg(True, "验证文件已存在且内容正确，无需重复写入！"))
                result.append({"domain": domain_name, "rep": rep})
        else:
            return public.returnMsg(False, "不支持的验证类型：{}！".format(verify_type))
        return {"status": True, "msg": "success", "data": result}