# coding: utf-8 # ------------------------------------------------------------------- # 宝塔Linux面板 # ------------------------------------------------------------------- # Copyright (c) 2015-2099 宝塔软件(http://bt.cn) All rights reserved. # ------------------------------------------------------------------- # Author: cjxin # ------------------------------------------------------------------- # 备份 # ------------------------------ import os, sys, re, json, shutil, psutil, time import uuid from sslModel.base import sslBase import public from BTPanel import project class main(sslBase): def __init__(self): super().__init__() # self.__init_data() self.__create_table() def run_fun(self,get): """ @name 执行指定函数 @param get.def_name 函数名 @param get.dns_type dns类型 """ if 'fun_name' not in get: return public.returnMsg(False,'缺少参数 fun_name') if not get.fun_name in ['delete_dns_record','create_dns_record','get_dns_record','update_dns_record','set_dns_record_status', 'get_domain_list']: return public.returnMsg(False,'参数错误，未知的函数名') dns_type = False try: dns_data = self.get_dns_data(get) dns_type = dns_data[get.dns_id]['dns_type'] except: pass if not dns_type: return public.returnMsg(False,'未知的dns_id') get.dns_type = dns_type try: res = self.func_models(get, get.fun_name) return res except Exception as e: return public.returnMsg(False,'执行失败:{}'.format(str(e))) def get_domain_list(self,get): """ @name 获取域名列表 """ # self.__init_data() filter_sql = '' if get.get('search'): filter_sql += ' and domain like {}'.format("'%"+get.search+"%'") public.set_search_history('ssl', 'get_domain_list', get.search) if get.get('type_id'): filter_sql += " and type_id = '{}'".format(get.type_id) p = 1 if 'p' in get: p = int(get.p) collback = '' if 'collback' in get: collback = get.collback limit = 20 if 'limit' in get: limit = int(get.limit) # count = public.M('ssl_domains').where("dns_id not like 'cloud_id=%' {}".format(filter_sql), ()).count() # page_data = public.get_page(count, p, limit, collback) dns_data = self.get_dns_data(get) data = public.M('ssl_domains').field('id,domain,type_id,dns_id,endtime,ps').where("dns_id not like 'cloud_id=%' {}".format(filter_sql), ()).select() record_data = self.get_record_data() from mod.project.push import taskMod report_data = taskMod.main().get_task_list().get('data') report_data_dic = {} if report_data: report_data_dic = {i["keyword"]: i["id"] for i in report_data if i["source"] == "domain_endtime" and i["status"]} domain_list = [] _data = [] remove_ids = [] for val in data: if val['domain'] in domain_list: remove_ids.append(str(val['id'])) continue if self._is_ip(val['domain']): continue domain_list.append(val['domain']) val["report_id"] = report_data_dic.get(val['domain'], "") or "" if val['dns_id'] == '0': val['dns_id'] = '' val['dns'] = dns_data.get(val['dns_id'],{}) # 获取子域名数量 val['sbd_count'] = record_data.get(val["domain"], {}).get('info', {}).get('record_total', '-') if val['dns_id'] != '': _data.append(val) else: remove_ids.append(str(val['id'])) count = len(_data) begin = (p - 1) * limit end = begin + limit _data = _data[begin:end] page_data = public.get_page(count, p, limit, collback) try: public.M('ssl_domains').where("id in ({})".format(",".join(remove_ids)), ()).delete() except Exception as e: print(e) search_history = public.get_search_history('ssl', 'get_domain_list') page_data.update({'data': _data, 'search_history': search_history}) public.set_module_logs('ssl', 'get_domain_list', 1) return page_data def set_domain_ps(self,get): """ @name 设置域名备注 @param get.id 域名id @param get.ps 备注 """ id = get.id ps = get.ps res = public.M('ssl_domains').where('id=?',(id,)).setField('ps',ps) if not res: return public.returnMsg(False,'设置失败!') return public.returnMsg(True,'设置成功!') def set_domain_dns(self,get): """ @name 给域名设置dns @param get.ids 域名id [1,2,3] @param get.dns_id dns_id """ try: ids = json.loads(get.ids) except: pass if not ids: return public.returnMsg(False,'请选择要设置的域名!') dns_id = get.dns_id dns_data = self.get_dns_data(get) if dns_id not in dns_data: return public.returnMsg(False,'指定DNS不存在!') params = [] for id in ids: params.append((dns_id,id)) res = public.M('ssl_domains').executemany('update ssl_domains set dns_id=? where id=?',params) if type(res) != int: return public.returnMsg(False,'设置失败!') if res != len(ids): insert_res = public.M('ssl_domains').executemany('insert into ssl_domains (dns_id,domain) values (?,?)', params) return public.returnMsg(True, '设置成功，已更新 {} 条记录，新增 {} 条记录!'.format(res, insert_res)) return public.returnMsg(True,'设置成功，已更新 {} 条记录!'.format(res)) def set_domain_endtime(self,get): """ @name 设置域名到期时间 @param get.ids 域名id [1,2,3] @param get.endtime 到期时间 """ try: ids = json.loads(get.ids) except: pass if not ids: return public.returnMsg(False,'请选择要设置的域名!') endtime = get.endtime params = [] for id in ids: params.append((endtime,id)) res = public.M('ssl_domains').executemany('update ssl_domains set endtime=? where id=?',params) if type(res) != int: return public.returnMsg(False,'设置失败!') return public.returnMsg(True,'设置成功，已更新 {} 条记录!'.format(res)) def set_domain_type(self,get): """ @设置域名到期时间 @param get.ids 域名id [1,2,3] @param get.type_id 类型id """ try: ids = json.loads(get.ids) except: pass if not ids: return public.returnMsg(False,'请选择要设置的域名!') type_id = get.type_id params = [] for id in ids: params.append((type_id,id)) res = public.M('ssl_domains').executemany('update ssl_domains set type_id=? where id=?',params) if type(res) != int: return public.returnMsg(False,'设置失败!') return public.returnMsg(True,'设置成功，已更新 {} 条记录!'.format(res)) def get_domain_type(self,get): """ @name 获取域名类型 """ data = [{'type_id': 0, 'name': '默认分组'}] try: sfile = '{}/data/domains_type.json'.format(public.get_panel_path()) data.extend(json.loads(public.readFile(sfile))) except:pass return data def add_domain_type(self,get): """ @name 添加域名分类 @param get.name 分类名称 @param get.type_id 分类类型 """ sfile = '{}/data/domains_type.json'.format(public.get_panel_path()) try: data = json.loads(public.readFile(sfile)) except: data = [] type_id = str(uuid.uuid4().hex) if 'type_id' in get: type_id = get.type_id for i in data: if get.name == i["name"] or get.name == '默认分组': return public.returnMsg(False, '此类型已存在') data.append({'name':get.name,'type_id':type_id}) public.writeFile(sfile,json.dumps(data)) return public.returnMsg(True,'添加成功') def del_domain_type(self, get): sfile = '{}/data/domains_type.json'.format(public.get_panel_path()) try: data = json.loads(public.readFile(sfile)) i = 0 while i < len(data): if data[i]["type_id"] == get.type_id: del data[i] break i += 1 public.writeFile(sfile, json.dumps(data)) return public.returnMsg(True, '删除成功') except: return public.returnMsg(True, '删除失败') # def __init_data(self): # """ # @name 同步数据 # """ # # 检查表是否存在 # self.__create_table() # # # 根据文件修改时间判断是否要同步 # check_path = "/www/server/panel/data/sync_domains.pl" # m_time = str(os.stat("/www/server/panel/data/db/site.db").st_mtime) # check_path_docker = "/www/server/panel/data/sync_domains_docker.pl" # m_time_docker = str(os.stat("/www/server/panel/data/db/docker.db").st_mtime) # if os.path.exists(check_path) and os.path.exists(check_path_docker): # if m_time == public.readFile(check_path) and m_time_docker == public.readFile(check_path_docker): # return # public.writeFile(check_path, m_time) # public.writeFile(check_path_docker, m_time) # # # 获取手动删除的域名 # del_path = '{}/config/del_domains.pl'.format(public.get_panel_path()) # try: # skip_domains = json.loads(public.readFile(del_path)) # except: # skip_domains = [] # # root_domains = {i['domain'] for i in public.M('ssl_domains').field('domain').select()} # site_domains = set() # for i in public.M('domain').field('name').select(): # if not self._is_ip(i['name']): # site_domains.add(self.extract_zone(i['name'])[0]) # # docker_site_domains = set() # for i in public.M('docker_domain').field('name').select(): # if not self._is_ip(i['name']): # docker_site_domains.add(self.extract_zone(i['name'])[0]) # add_domains = (site_domains | docker_site_domains) - root_domains - set(skip_domains) # # del_domains = root_domains - (docker_site_domains | site_domains) # # # 添加域名 # for domain in add_domains: # public.M('ssl_domains').add('domain,dns_id,type_id,endtime,ps', (domain, 0, 0, 0, '')) # # # 删除域名 # # for domain in del_domains: # # public.M('ssl_domains').where("domain=?", (domain,)).delete() def del_domains(self, get): """ @name 删除域名 @param get.domains 域名列表，以逗号分隔 """ domains = get.domains.split(',') try: public.M('ssl_domains').where("domain in ('{}')".format("','".join(domains)), ()).delete() path = '{}/config/del_domains.pl'.format(public.get_panel_path()) try: del_domains = json.loads(public.readFile(path)) del_domains.extend(domains) public.writeFile(path, json.dumps(del_domains)) except: public.writeFile(path, json.dumps(domains)) return public.returnMsg(True, '删除成功') except: return public.returnMsg(False, '删除失败') # # 手动同步域名 # def sync_domains(self, get): # """ # @name 手动同步域名 # """ # root_domains = {i['domain'] for i in public.M('ssl_domains').field('domain').select()} # site_domains = set() # for i in public.M('domain').field('name').select(): # if not self._is_ip(i['name']): # site_domains.add(self.extract_zone(i['name'])[0]) # # docker_site_domains=set() # for i in public.M('docker_domain').field('name').select(): # if not self._is_ip(i['name']): # docker_site_domains.add(self.extract_zone(i['name'])[0]) # # add_domains = (site_domains | docker_site_domains) - root_domains # del_domains = root_domains - (docker_site_domains | site_domains) # # # 添加域名 # for domain in add_domains: # public.M('ssl_domains').add('domain,dns_id,type_id,endtime,ps', (domain, 0, 0, 0, '')) # # 删除域名 # for domain in del_domains: # public.M('ssl_domains').where("domain=?", (domain,)).delete() # # 删除文件 # try: # os.remove('{}/config/del_domains.pl'.format(public.get_panel_path())) # except: pass # return public.returnMsg(True, '同步成功') @staticmethod def _is_ip(data: str): import ipaddress try: ipaddress.ip_address(data) except: return False return True def __create_table(self): """ @name 检查表是否存在 """ public.check_table('ssl_domains',"""CREATE TABLE IF NOT EXISTS `ssl_domains` ( `id` INTEGER PRIMARY KEY AUTOINCREMENT, `domain` TEXT, `dns_id` TEXT, `type_id` INTEGER, `endtime` INTEGER, `ps` TEXT ) """) def check_table(self): self.__create_table() def get_objectModel(self): ''' 获取模型对象 ''' from panelController import Controller project_obj = Controller() return project_obj def func_models(self,get,def_name): ''' 获取模型对象 ''' if 'dns_type' not in get: raise Exception('缺少参数 dns_type') sfile = '{}/class/sslModel/{}Model.py'.format(public.get_panel_path(),get.dns_type) if not os.path.exists(sfile): raise Exception('模块文件{}不存在'.format(sfile)) obj_main = self.get_objectModel() args = public.dict_obj() args['data'] = get args['mod_name'] = get.dns_type args['def_name'] = def_name return obj_main.model(args) def get_domain_dns_config(self, get): # self.__init_data() dns_data = self.get_dns_data(get) root_data = public.M('ssl_domains').field('id,domain,dns_id').select() where_sql = '1=1' param = [] if 'domain_name' in get: where_sql += " AND name like ?" param.append('%{}%'.format(get.domain_name)) if 'site_id' in get: where_sql += " AND pid in ({})".format(get.site_id) data = public.M('domain').where(where_sql, param).field('name,pid as site_id').select() elif 'docker_site_id' in get: where_sql += " AND pid in ({})".format(get.docker_site_id) data = public.M('docker_domain').where(where_sql, param).field('name,pid as site_id').select() else: data = public.M('domain').where(where_sql, param).field('name,pid as site_id').select() + public.M('docker_domain').where(where_sql, param).field('name,pid as site_id').select() for i in data: root, sub_domain, _ = self.extract_zone(i['name']) i["status"] = 0 i["domain_id"] = root for j in root_data: if j['domain'] == root: i["domain_id"] = j["id"] i["status"] = 1 if j["dns_id"].find("cloud_id=")!=-1 or j["dns_id"] in dns_data.keys()else 0 break if 'auto_wildcard' in get and get.auto_wildcard == '1': new_data = [] for i in data: root, _, _ = self.extract_zone(i['name']) # 创建新对象以避免修改原数据 new_item = i.copy() new_item['name'] = root if new_item not in new_data: new_data.append(new_item) # 处理通配符域名 wildcard_item = new_item.copy() wildcard_item['name'] = "*." + root if wildcard_item not in new_data: new_data.append(wildcard_item) data = new_data return data def get_site_list(self, get): # 自动部署不兼容docker网站 # return public.M('sites').field("name,id,'site_id' as type").select() + public.M('docker_sites').field("name,id,'docker_site_id' as type").select() from mod.base.web_conf.ssl import RealSSLManger from datalistModel import sitesModel sites_model = sitesModel.main() # 获取所有域名 domain_data = public.M('domain').field('pid,name').select() domain_dic = {} for item in domain_data: if item['pid'] not in domain_dic: domain_dic[item['pid']] = [] domain_dic[item['pid']].append(item['name']) data = public.M('sites').field("name,id,'site_id' as type,project_type").select() for item in data: if item['project_type'] in ['PHP', 'proxy', 'WP2']: item['ssl'] = sites_model.get_site_ssl_info(item['name']) else: item['ssl'] = RealSSLManger('{}_'.format(item['project_type'].lower())).get_site_ssl_info(item['name']) if not item['ssl']: item['ssl'] = -1 item['domains'] = domain_dic.get(item['id'], []) return data def create_report_task(self, get): from mod.base.push_mod import manager sender_lsit = get.sender.split(",") task_data = {"task_data":{"tid":"70","type":"domain_endtime","title":"域名到期","status":True,"count":0,"interval":600,"project":get.domain,"cycle":int(get.cycle)},"sender":sender_lsit,"number_rule":{"day_num":0,"total":int(get.total)},"time_rule":{"send_interval":0,"time_range":[0,86399]}} get.template_id = "70" get.task_data = json.dumps(task_data) return manager.PushManager().set_task_conf(get) def remove_report_task(self, get): from mod.base.push_mod import manager return manager.PushManager().remove_task_conf(get) def add_dns_value_by_domain(self, domain, dns_value, record_type="TXT", is_let_txt=False, mx=10): root, sub, subd = self.extract_zone(domain, is_let_txt) domain_name = subd+'.'+root if is_let_txt else domain try: data = public.M('ssl_domains').field('dns_id').where("domain=?", (root,)).select() dns_id = data[0]['dns_id'] except: dns_id = '' if dns_id.find("cloud_id=") != -1: 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":subd if is_let_txt else sub,"value":dns_value,"type":record_type,"mx":mx,"ttl":600,"remark":"","view_id":0}) if not rep["status"] and '记录已存在' not in rep["msg"]: public.print_log(rep) raise Exception("设置解析记录失败：{}".format(rep["msg"])) else: return rep else: for dns_config in self.get_dns_data(public.dict_obj()).values(): if root in dns_config.get("domains", []) or str(dns_id) == dns_config["id"]: args = { "fun_name": "create_dns_record", "dns_id": dns_id, "domain_dns_value": dns_value, "record_type": record_type, "domain_name": domain_name, "mx": mx, } _return = self.run_fun(public.to_dict_obj(args)) if not _return["status"] and '记录已存在' not in _return["msg"]: raise Exception("设置解析记录失败：{}".format(_return["msg"])) else: return _return raise Exception("设置解析记录失败：没有找到域名为{}的有效的DNS API密钥信息".format(domain)) def get_dns_value_by_domain(self, domain, record_type="TXT", is_let_txt=False): root, sub, subd = self.extract_zone(domain, is_let_txt) domain_name = subd + '.' + root if is_let_txt else domain try: data = public.M('ssl_domains').field('dns_id').where("domain=?", (root,)).select() dns_id = data[0]['dns_id'] except: dns_id = '' if dns_id.find("cloud_id=") != -1: domain_id = dns_id.split("cloud_id=")[1] from mod.project.domain import domainMod domain_obj = domainMod.main() record_data = domain_obj.request({"url": "/api/v1/dns/record/list", "domain_type": 1, "domain_id": domain_id, "searchKey": "record", "searchValue": subd if is_let_txt else sub}) _return = [] for record in record_data["data"].get("data", []): if record["type"] == record_type and record["record"] == (subd if is_let_txt else sub): _return.append(record["record_id"]) return _return, dns_id else: for dns_config in self.get_dns_data(public.dict_obj()).values(): if root in dns_config.get("domains", []) or str(dns_id) == dns_config["id"]: args = { "fun_name": "get_dns_record", "dns_id": dns_id, "domain_name": domain, } record_data = self.run_fun(public.to_dict_obj(args)) _return = [] for record in record_data.get("list", []): if record["type"] == record_type and record["name"] == domain_name: _return.append(record["RecordId"]) return _return, dns_id return [], "" def del_dns_value_by_domain(self, domain, record_type="TXT", is_let_txt=False): ids, dns_id = self.get_dns_value_by_domain(domain, record_type, is_let_txt) if dns_id.find("cloud_id=") != -1: domain_id = dns_id.split("cloud_id=")[1] from mod.project.domain import domainMod domain_obj = domainMod.main() for record_id in ids: rep = domain_obj.request({"url": "/api/v1/dns/record/delete", "domain_type": 1, "domain_id": domain_id, "record_id": record_id}) if not rep["status"]: raise Exception("删除解析记录失败：{}".format(rep["msg"])) return else: for record_id in ids: args = { "fun_name": "delete_dns_record", "dns_id": dns_id, "RecordId": record_id, "domain_name": domain, } _return = self.run_fun(public.to_dict_obj(args)) if not _return["status"]: raise Exception("删除解析记录失败：{}".format(_return["msg"])) def get_sub_domains(self, get): import socket root_domain = get.root_domain domains = public.M('domain').where("name like ?", ('%'+root_domain,)).field('name,pid').select() for domain in domains: site_data = public.M('sites').where("id = ?", (domain['pid'],)).find() domain.update({"site_name": site_data['name']}) domain.update({"record_a": socket.gethostbyname(domain['name'])}) return domains def add_domain(self, get): domain_name = get.domain_name dns_id = get.dns_id # 检查是否为顶级域名 root, sub, _ = self.extract_zone(domain_name) if sub != '': return public.returnMsg(False, '只能添加顶级域名') data = public.M('ssl_domains').field('id').where("domain=?", (domain_name,)).select() if data: return public.returnMsg(False, '域名已存在') public.M('ssl_domains').add('domain,dns_id,type_id,endtime,ps', (domain_name, dns_id, 0, 0, '')) return public.returnMsg(True, '添加成功')