bt-source/panel/class/sslModel/autodeployModel.py

import os, sys, json, psutil
import time

os.chdir('/www/server/panel')
if not 'class/' in sys.path:
    sys.path.insert(0,'class/')

import public

from sslModel.base import sslBase


class main(sslBase):
    __APIURL2 = public.GetConfigValue('home') + '/api/v2/synchron'
    # __APIURL2 = "https://dev.bt.cn" + '/api/v2/synchron'
    __PDATA = None
    _check_url = None
    __request_url = None
    __UPATH = 'data/userInfo.json'
    __userInfo = None
    def __init__(self):
        super().__init__()
        if not os.path.exists(public.get_panel_path() + '/data/auto_deploy'):
            os.makedirs(public.get_panel_path() + '/data/auto_deploy')

        # 本地任务表
        self.M("").execute("""CREATE TABLE IF NOT EXISTS `autodeploy_task` (
          `id` INTEGER  PRIMARY KEY AUTOINCREMENT,        
          `task_name` varchar(320) NOT NULL,    -- 任务名称
          `cloud_id` INTEGER NOT NULL,    -- 云端任务ID
          `access_id` INTEGER NOT NULL,    -- 接入ID
          `private_key` varchar(320) NOT NULL,    -- 密钥
          `sites` varchar(320) NOT NULL,    -- 网站
          `cycle` INTEGER NOT NULL DEFAULT 12,    -- 检测周期
          `deploy_time` INTEGER NOT NULL DEFAULT 0,    -- 最近部署时间
          `deploy_status` INTEGER NOT NULL DEFAULT 0,    -- 最近部署状态 0等待部署 1=部署成功 -1部署失败
          `ssl_hash` varchar(320) NOT NULL DEFAULT '',   -- 证书hash
          `ssl_cloud_id` varchar(320) NOT NULL DEFAULT ''   -- 云端证书id
          );""", ())

        self.get_user_info()

        # 云端任务表
    def request(self, dname):

        request_url = self.__APIURL2 + '/' + dname
        self.__request_url = public.get_home_node(request_url)
        msg = '接口请求失败（{}）'.format(self.__request_url)
        try:

            res = public.httpPost(request_url, self.__PDATA)
            if res == False:
                raise public.error_conn_cloud(msg)
        except Exception as ex:
            raise public.error_conn_cloud(str(ex))

        result = public.returnMsg(False, msg)
        try:
            result = json.loads(res)
        except:
            pass
        return result

    def get_user_info(self):
        upath = public.get_panel_path() + "/" + self.__UPATH
        try:
            self.__userInfo = json.loads(public.readFile(upath))
        except:
            self.__userInfo = {}
        self.__PDATA = self.__userInfo

    def M(self, table_name):
        import db
        sql = db.Sql()
        sql._Sql__DB_FILE = public.get_panel_path() + '/data/db/ssl_data.db'
        sql._Sql__encrypt_keys = []
        return sql.table(table_name)

    def plane_synchron_list(self, get):
        res = self.request('plane_synchron_list')
        if not res["success"]:
            public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def plane_synchron_access_list(self):
        # path = public.get_panel_path() + '/data/auto_deploy/plane_synchron_list.json'
        # try:
        #     data = json.loads(public.readFile(path))
        #     return public.returnMsg(True, data)
        # except:
        #     pass
        res = self.request('plane_synchron_access_list')

        if not res["success"]:
            public.returnMsg(False, res["res"])
        # public.writeFile(path, json.dumps(res["res"]))
        return public.returnMsg(True, res["res"])

    def plane_synchron_access_set(self, get):
        self.__PDATA["key"] = get.private_key
        self.__PDATA["synid"] = int(get.cloud_id)
        # self.__PDATA["url"] = "https://www.bt.cn"
        res = self.request('plane_synchron_access_set')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def get_task_list(self, get=None):

        p = 1
        if 'p' in get:
            p = int(get.p)
        collback = ''
        if 'collback' in get:
            collback = get.collback
        limit = 999999999
        if 'limit' in get:
            limit = int(get.limit)
        f_sql = '1=1'
        if 'search' in get and get.search:
            f_sql = "task_name like '%{}%' or sites like '%{}%' ".format(get.search, get.search)

        count = self.M("autodeploy_task").where(f_sql, ()).count()
        page_data = public.get_page(count, p, limit, collback)

        data = self.M("autodeploy_task").where(f_sql, ()).limit(page_data["shift"]+","+page_data["row"]).order("id desc").select()
        # public.print_log(data)

        try:
            cloud_data = self.plane_synchron_access_list()
        except:
            cloud_data = None
        # public.print_log(cloud_data)

        deploy = []
        for i in data:
            i["access_state"] = 0
            i["synchron_info"] = {}
            if not cloud_data:
                continue
            for j in cloud_data["msg"]:
                if i["access_id"] == j["id"]:
                    i["access_state"] = j["access_state"]
                    if j["access_state"] == 2 and str(i["ssl_cloud_id"]) != str(j["synchron_info"]["ca_id"]):
                        self.M("autodeploy_task").where("cloud_id=?", (i["cloud_id"],)).update({"ssl_cloud_id": j["synchron_info"]["ca_id"]})
                        # 写需要更新证书的任务到队列
                        deploy.append(i["id"])
                    i["synchron_info"] = j["synchron_info"]
        if deploy:
            public.ExecShell("nohup btpython {}/class/sslModel/autodeployModel.py --task_ids={} > /dev/null 2>&1 &".format(public.get_panel_path(), ",".join(str(i) for i in deploy)))
        page_data["data"] = data

        return {"status": True, "msg": page_data, "deploy": deploy}

    def add_task(self, get):
        task_name = get.task_name
        cloud_id = get.cloud_id
        private_key = get.private_key
        sites = get.sites
        # cycle = get.cycle
        cycle = 1
        data = self.M("autodeploy_task").where("cloud_id=?", (cloud_id,)).find()
        if data:
            return public.returnMsg(False, "选择的证书部署任务在本服务器已存在对应的任务【{}】请勿重复添加！".format(data["task_name"]))

        res = self.plane_synchron_access_set(get)
        if not res["status"]:
            return public.returnMsg(False, res["msg"])

        self.M("autodeploy_task").add('task_name,cloud_id,access_id,private_key,sites,cycle', (task_name, cloud_id, res["msg"]["id"], private_key, sites, cycle))
        if not res["status"]:
            return public.returnMsg(False, res["msg"])
        return public.returnMsg(True, "添加成功")

    def plane_synchron_access_delte(self, get):
        self.__PDATA["id"] = int(get.access_id)
        res = self.request('plane_synchron_access_delte')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def del_task(self, get):
        data = self.M("autodeploy_task").where("id=?", (get.task_id,)).find()
        if not data:
            return public.returnMsg(False, "未找到任务")
        self.M("autodeploy_task").where("id=?", (get.task_id,)).delete()
        self.plane_synchron_access_delte(public.to_dict_obj({"access_id": data["access_id"]}))
        return public.returnMsg(True, "删除成功")

    def batch_del_task(self, get):
        task_id_list = get.task_ids.split(",")
        data = self.M("autodeploy_task").where("id in ({})".format(get.task_ids), ()).select()
        self.M("autodeploy_task").where("id in ({})".format(get.task_ids), ()).delete()

        total = len(data)
        faild = 0
        success = 0
        faildList = []
        successList = []
        for i in data:
            if str(i["id"]) not in task_id_list:
                faild += 1
                faildList.append({"task_id": i["id"], "task_name":i["task_name"],"error_msg": "任务不存在", "status": False})
                continue
            self.plane_synchron_access_delte(public.to_dict_obj({"access_id": i["access_id"]}))
            success += 1
            successList.append({"task_id": i["id"], "task_name":i["task_name"],"error_msg": "删除成功", "status": True})

        return public.returnMsg(True, {"total": total, "success": success, "faild": faild, "faildList": faildList, "successList": successList})

    def edit_task(self, get):
        task_id = get.task_id
        task_name = get.task_name
        cloud_id = int(get.cloud_id)
        private_key = get.private_key
        sites = get.sites
        # cycle = get.cycle
        cycle = 1

        data = self.M("autodeploy_task").where("id=?", (task_id,)).find()
        if not data:
            return public.returnMsg(False, "未找到任务")
        update_data = {"task_name": task_name, "cloud_id": cloud_id, "private_key": private_key, "sites": sites, "cycle": cycle}

        if data["cloud_id"] != cloud_id:
            cloud_data = self.M("autodeploy_task").where("cloud_id=?", (cloud_id)).find()
            if cloud_data:
                return public.returnMsg(False, "选择的证书部署任务在本服务器已存在对应的任务【{}】请勿重复添加！".format(cloud_data["task_name"]))
            res = self.plane_synchron_access_set(get)
            if not res["status"]:
                return public.returnMsg(False, res["msg"])
            update_data["access_id"] = res["msg"]["id"]
        elif data["private_key"] != private_key:
            res = self.plane_synchron_access_set(get)
            if not res["status"]:
                return public.returnMsg(False, res["msg"])

        if data["sites"] != sites:
            update_data["ssl_cloud_id"] = ""

        self.M("autodeploy_task").where("id=?", (get.task_id,)).update(update_data)
        return public.returnMsg(True, "修改成功")

    def plane_synchron_get_cert(self, access_id):
        self.__PDATA["id"] = int(access_id)
        res = self.request('plane_synchron_get_cert')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def set_cert_to_database(self, access_id):
        cert_data = self.plane_synchron_get_cert(access_id)
        if not cert_data["status"]:
            return public.returnMsg(False, cert_data["msg"])
        cert_data = cert_data["msg"]["cert"]

        from sslModel import certModel
        get = public.dict_obj()
        get.key = cert_data["privateCert"]
        get.csr = cert_data["cert"]

        try:
            cert = certModel.main().save_cert(get)
            if not cert["status"]:
                return False
        except:
            return False
        self.M("autodeploy_task").where("access_id=?", (access_id,)).update({"ssl_hash": cert["ssl_hash"]})
        return True

    def deploy_cert(self, get):
        data = self.M("autodeploy_task").where("id=?", (int(get.task_id),)).find()
        if not data:
            return public.returnMsg(False, "未找到任务")
        sites = data["sites"].split(",")
        ssl_hash = data["ssl_hash"]
        get.BatchInfo = json.dumps([
            {"ssl_hash": ssl_hash, "siteName": i}
            for i in sites
        ])
        import panelSSL
        res = panelSSL.panelSSL().SetBatchCertToSite(get)

        deploy_time = int(time.time())
        if res["total"] == res["success"]:
            self.write_deploy_detail(get.task_id, json.dumps(res), data["access_id"], 1, 1)
            deploy_status = 1
        else:
            self.write_deploy_detail(get.task_id, json.dumps(res), data["access_id"], -1, 1)
            deploy_status = -1
        self.M("autodeploy_task").where("id=?", (int(get.task_id),)).update({"deploy_time": deploy_time, "deploy_status": deploy_status})

    def upload_deploy_msg(self, get):
        self.__PDATA["msg"] = get.error_msg
        self.__PDATA["id"] = get.access_id
        self.__PDATA["state"] = get.state
        self.__PDATA["type"] = get.msg_type

        # print(self.__PDATA)

        res = self.request('plane_synchron_deploy_state')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def write_deploy_detail(self, task_id, error_msg, access_id, state, msg_type):
        path = public.get_panel_path() + '/data/auto_deploy/deploy_detail'
        if not os.path.exists(path):
            os.makedirs(path)
        public.writeFile(path + "/{}.json".format(task_id), error_msg)
        # print(access_id, state, msg_type)
        res = self.upload_deploy_msg(public.to_dict_obj({"access_id": access_id, "error_msg": error_msg, "state": state, "msg_type": msg_type}))
        # print(res)

    def get_task_deploy_detail(self, get):
        try:
            res = json.loads(public.readFile(public.get_panel_path() + '/data/auto_deploy/deploy_detail/{}.json'.format(get.task_id)))
        except:
            res = None
        return public.returnMsg(True, res)



if __name__ == '__main__':
    try:
        if os.path.exists("/www/server/panel/data/auto_deploy.pl"):
            exit()
        public.writeFile("/www/server/panel/data/auto_deploy.pl", "")
        m = main()
        import argparse
        p = argparse.ArgumentParser()
        p.add_argument('--task_ids', default=None, help="任务id列表，用逗号分隔", dest="task_ids")
        args = p.parse_args()
        if not args.task_ids:
            exit()
        task_ids = args.task_ids.split(",")
        for i in task_ids:
            task_data = m.M("autodeploy_task").where("id=?", (int(i),)).find()
            if not task_data:
                continue
            set_cert_status = m.set_cert_to_database(task_data["access_id"])
            if not set_cert_status:
                sites = task_data["sites"].split(",")
                error_msg = json.dumps({
                    "total": len(sites),
                    "success": 0,
                    "faild": len(sites),
                    "faildList": [{"siteName": site, "error_msg": "证书保存失败", "status": False} for site in sites],
                    "successList": []
                })
                m.write_deploy_detail(i, error_msg, task_data["access_id"], -1, 1)
                m.M("autodeploy_task").where("id=?", (int(i),)).update(
                    {"deploy_time": int(time.time()), "deploy_status": -1})
                continue
            m.deploy_cert(public.to_dict_obj({"task_id": i}))
            print("部署成功：", task_data["task_name"])
        exit()
    finally:
        os.remove("/www/server/panel/data/auto_deploy.pl")