mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-17 21:25:47 +02:00
Jack
a24f2e23c6
1. Added Mail Marketing Automation trigger tasks 2. Added Mail Marketing Groups Import, Export, Merge 3. Added Mail Marketing Subscribers to support paste import 4. Added Mail Marketing Template Import Export Duplicate 5. Added Mail server add timed automatic reply 6. Added HTTPS Protection function for Website (disables automatic HTTP to HTTPS redirection when enabled) 7. Optimize Website Interface Button Integration 8. Optimize the response speed of WP Toolkit interface
1683 lines
62 KiB
Python
1683 lines
62 KiB
Python
# coding: utf-8
|
|
# +-------------------------------------------------------------------
|
|
# | aapanel x3
|
|
# +-------------------------------------------------------------------
|
|
# | Copyright (c) 2015-2099 aapanel(http://www.aapanel.com) All rights reserved.
|
|
# +-------------------------------------------------------------------
|
|
# | Author: bazoi <bazoi@bt.cn>
|
|
# +--------------------------------------------------------------------
|
|
# | 宝塔DNS管理平台
|
|
# +--------------------------------------------------------------------
|
|
import shutil
|
|
import sys
|
|
import os
|
|
import json
|
|
import random
|
|
import datetime
|
|
import hmac
|
|
import re
|
|
import base64
|
|
from urllib.parse import urljoin
|
|
from hashlib import sha1
|
|
from uuid import uuid4
|
|
from itertools import chain
|
|
from typing import Set, List, Optional, Union, Dict, Tuple
|
|
from mod.base import json_response
|
|
|
|
if "/www/server/panel/class" not in sys.path:
|
|
sys.path.insert(0, "/www/server/panel/class")
|
|
|
|
import public
|
|
|
|
try:
|
|
import requests
|
|
except:
|
|
public.ExecShell('pip install requests')
|
|
import requests
|
|
|
|
caa_value = '0 issue "letsencrypt.org"'
|
|
|
|
|
|
class ExtractZoneTool(object):
|
|
def __init__(self):
|
|
self.top_domain_list = [
|
|
'.ac.cn', '.ah.cn', '.bj.cn', '.com.cn', '.cq.cn', '.fj.cn', '.gd.cn',
|
|
'.gov.cn', '.gs.cn', '.gx.cn', '.gz.cn', '.ha.cn', '.hb.cn', '.he.cn',
|
|
'.hi.cn', '.hk.cn', '.hl.cn', '.hn.cn', '.jl.cn', '.js.cn', '.jx.cn',
|
|
'.ln.cn', '.mo.cn', '.net.cn', '.nm.cn', '.nx.cn', '.org.cn']
|
|
top_domain_list_data = public.readFile('{}/config/domain_root.txt'.format(public.get_panel_path()))
|
|
if top_domain_list_data:
|
|
self.top_domain_list = set(top_domain_list_data.strip().split('\n'))
|
|
|
|
def __call__(self, domain_name):
|
|
if domain_name.startswith('*.'):
|
|
domain_name = domain_name[2:]
|
|
domain_name_copy = domain_name
|
|
top_domain = "." + ".".join(domain_name.rsplit('.')[-2:])
|
|
new_top_domain = "." + top_domain.replace(".", "")
|
|
is_tow_top = False
|
|
if top_domain in self.top_domain_list:
|
|
is_tow_top = True
|
|
domain_name = domain_name[:-len(top_domain)] + new_top_domain
|
|
|
|
if domain_name.count(".") <= 1:
|
|
zone = ""
|
|
root = domain_name_copy
|
|
else:
|
|
zone, middle, last = domain_name.rsplit(".", 2)
|
|
if is_tow_top:
|
|
last = top_domain[1:]
|
|
root = ".".join([middle, last])
|
|
|
|
return root, zone
|
|
|
|
|
|
extract_zone = ExtractZoneTool()
|
|
|
|
|
|
class BaseDns(object):
|
|
def __init__(self):
|
|
self.dns_provider_name = self.__class__.__name__
|
|
|
|
@staticmethod
|
|
def load_response(response):
|
|
try:
|
|
log_body = response.json()
|
|
except ValueError:
|
|
log_body = response.content
|
|
return log_body
|
|
|
|
def add_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
raise NotImplementedError("add_record method must be implemented.")
|
|
|
|
def remove_record(self, domain: str, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
raise NotImplementedError("remove_record method must be implemented.")
|
|
|
|
def modify_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
raise NotImplementedError("modify_record method must be implemented.")
|
|
|
|
def get_record_list(self, domain, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
raise NotImplementedError("get_record_list must be implemented.")
|
|
|
|
def get_record_list_by_view(self, domain):
|
|
raise NotImplementedError("get_record_list must be implemented.")
|
|
|
|
def get_domain_list(self):
|
|
return []
|
|
|
|
@classmethod
|
|
def new(cls, conf_data):
|
|
raise NotImplementedError("new method must be implemented.")
|
|
|
|
def transform_record_type(self, r_type: str):
|
|
if r_type not in ("显性URL", "隐性URL"):
|
|
return r_type
|
|
self_cls = self.__class__
|
|
if self_cls is AliyunDns:
|
|
return "REDIRECT_URL" if r_type == "显性URL" else "FORWARD_URL"
|
|
elif self_cls is CloudFlareDns:
|
|
return "URI"
|
|
|
|
|
|
class DNSPodDns(BaseDns):
|
|
dns_provider_name = "dnspod"
|
|
|
|
def __init__(self, DNSPOD_ID, DNSPOD_API_KEY, DNSPOD_API_BASE_URL="https://dnsapi.cn/"):
|
|
self.DNSPOD_ID = DNSPOD_ID
|
|
self.DNSPOD_API_KEY = DNSPOD_API_KEY
|
|
self.DNSPOD_API_BASE_URL = DNSPOD_API_BASE_URL
|
|
self.HTTP_TIMEOUT = 65 # seconds
|
|
self.DNSPOD_LOGIN = "{0},{1}".format(self.DNSPOD_ID, self.DNSPOD_API_KEY)
|
|
|
|
if DNSPOD_API_BASE_URL[-1] != "/":
|
|
self.DNSPOD_API_BASE_URL = DNSPOD_API_BASE_URL + "/"
|
|
else:
|
|
self.DNSPOD_API_BASE_URL = DNSPOD_API_BASE_URL
|
|
super(DNSPodDns, self).__init__()
|
|
|
|
def add_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
url = urljoin(self.DNSPOD_API_BASE_URL, "Record.Create")
|
|
body = {
|
|
"record_type": r_type,
|
|
"domain": domain,
|
|
"sub_domain": host,
|
|
"value": value,
|
|
"record_line_id": "0",
|
|
"format": "json",
|
|
"weight": weight,
|
|
"ttl": ttl,
|
|
"login_token": self.DNSPOD_LOGIN,
|
|
}
|
|
add_response = requests.post(
|
|
url, data=body, timeout=self.HTTP_TIMEOUT
|
|
).json()
|
|
if add_response["status"]["code"] != "1":
|
|
raise ValueError(
|
|
"Error creating dnspod dns record: status_code={status_code} response={response}".format(
|
|
status_code=add_response["status"]["code"],
|
|
response=add_response["status"]["message"],
|
|
)
|
|
)
|
|
|
|
def get_record_list_by_view(self, domain):
|
|
records = self.get_record_list(domain)
|
|
if records is None:
|
|
return []
|
|
res = []
|
|
for record in records:
|
|
res.append({
|
|
"host": record['name'],
|
|
"type": record['type'],
|
|
"value": record['value'],
|
|
"ttl": record['ttl'],
|
|
"updated_on": datetime.datetime.strptime(record['updated_on'], "%Y-%m-%d %H:%M:%S").timestamp(),
|
|
})
|
|
return res
|
|
|
|
def get_record_list(self, domain, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
url = urljoin(self.DNSPOD_API_BASE_URL, "Record.List")
|
|
body = {
|
|
"login_token": self.DNSPOD_LOGIN,
|
|
"format": "json",
|
|
"domain": domain
|
|
}
|
|
if host is not None and isinstance(host, str):
|
|
body["subdomain"] = host
|
|
if r_type is not None and isinstance(r_type, str):
|
|
body["record_type"] = r_type
|
|
|
|
list_response = requests.post(url, data=body, timeout=self.HTTP_TIMEOUT).json()
|
|
if "records" in list_response:
|
|
return list_response["records"]
|
|
else:
|
|
return None
|
|
|
|
def remove_record(self, domain, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
records = self.get_record_list(domain, host, r_type)
|
|
if records is None:
|
|
return False, "未查询到记录信息，没有执行删除"
|
|
for record in records:
|
|
if record['name'] != host or record['type'] != r_type:
|
|
continue
|
|
record_id = record["id"]
|
|
url = urljoin(self.DNSPOD_API_BASE_URL, "Record.Remove")
|
|
body = {
|
|
"login_token": self.DNSPOD_LOGIN,
|
|
"format": "json",
|
|
"domain": domain,
|
|
"record_id": record_id,
|
|
}
|
|
requests.post(
|
|
url, data=body, timeout=self.HTTP_TIMEOUT
|
|
).json()
|
|
|
|
return True, "删除成功"
|
|
|
|
def modify_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
records = self.get_record_list(domain, host, r_type)
|
|
if records is None:
|
|
return False, "未查询到记录信息，无法执行修改"
|
|
|
|
for record in records:
|
|
if record['name'] != host and record['type'] != r_type:
|
|
continue
|
|
|
|
record_id = record["id"]
|
|
url = urljoin(self.DNSPOD_API_BASE_URL, "Record.Modify")
|
|
body = {
|
|
"record_type": r_type,
|
|
"record_id": record_id,
|
|
"domain": domain,
|
|
"sub_domain": host,
|
|
"value": value,
|
|
"record_line_id": "0",
|
|
"format": "json",
|
|
"weight": weight,
|
|
"ttl": ttl,
|
|
"login_token": self.DNSPOD_LOGIN,
|
|
}
|
|
add_response = requests.post(
|
|
url, data=body, timeout=self.HTTP_TIMEOUT
|
|
).json()
|
|
|
|
if add_response["status"]["code"] != "1":
|
|
raise ValueError(
|
|
"Error creating dnspod dns record: status_code={status_code} response={response}".format(
|
|
status_code=add_response["status"]["code"],
|
|
response=add_response["status"]["message"],
|
|
)
|
|
)
|
|
|
|
def get_domain_list(self):
|
|
url = urljoin(self.DNSPOD_API_BASE_URL, "Domain.List")
|
|
body = {
|
|
"format": "json",
|
|
"login_token": self.DNSPOD_LOGIN,
|
|
}
|
|
domain_list_resp = requests.post(
|
|
url, data=body, timeout=self.HTTP_TIMEOUT
|
|
).json()
|
|
if domain_list_resp["status"]["code"] != "1":
|
|
return {}
|
|
domains = domain_list_resp["domains"]
|
|
return {d["name"]: d for d in domains}
|
|
|
|
@classmethod
|
|
def new(cls, conf_data: dict) -> BaseDns:
|
|
key = conf_data.get("key", None) or conf_data.get("ID", "")
|
|
secret = conf_data.get("secret", None) or conf_data.get("Token", "")
|
|
base_url = "https://dnsapi.cn/"
|
|
|
|
return cls(key, secret, base_url)
|
|
|
|
|
|
class CloudFlareDns(BaseDns):
|
|
dns_provider_name = "cloudflare"
|
|
|
|
def __init__(
|
|
self,
|
|
CLOUDFLARE_EMAIL,
|
|
CLOUDFLARE_API_KEY,
|
|
CLOUDFLARE_API_BASE_URL="https://api.cloudflare.com/client/v4/",
|
|
):
|
|
self.CLOUDFLARE_EMAIL = CLOUDFLARE_EMAIL
|
|
self.CLOUDFLARE_API_KEY = CLOUDFLARE_API_KEY
|
|
self.CLOUDFLARE_API_BASE_URL = CLOUDFLARE_API_BASE_URL
|
|
self.HTTP_TIMEOUT = 65 # seconds
|
|
|
|
self._domain_zone_id_cache = {}
|
|
|
|
if CLOUDFLARE_API_BASE_URL[-1] != "/":
|
|
self.CLOUDFLARE_API_BASE_URL = CLOUDFLARE_API_BASE_URL + "/"
|
|
else:
|
|
self.CLOUDFLARE_API_BASE_URL = CLOUDFLARE_API_BASE_URL
|
|
super(CloudFlareDns, self).__init__()
|
|
|
|
def _get_auth_headers(self) -> dict:
|
|
if self.CLOUDFLARE_EMAIL is None and isinstance(self.CLOUDFLARE_API_KEY, str):
|
|
return {"Authorization": "Bearer " + self.CLOUDFLARE_API_KEY}
|
|
else:
|
|
return {"X-Auth-Email": self.CLOUDFLARE_EMAIL, "X-Auth-Key": self.CLOUDFLARE_API_KEY}
|
|
|
|
def find_dns_zone(self, domain):
|
|
for domain_key, zone_id in self._domain_zone_id_cache.items():
|
|
if domain_key in domain:
|
|
return self._domain_zone_id_cache[domain_key]
|
|
|
|
url = urljoin(self.CLOUDFLARE_API_BASE_URL, "zones?status=active&per_page=1000")
|
|
headers = self._get_auth_headers()
|
|
find_dns_zone_response = requests.get(url, headers=headers, timeout=self.HTTP_TIMEOUT)
|
|
if find_dns_zone_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error creating cloudflare dns record: status_code={status_code} response={response}".format(
|
|
status_code=find_dns_zone_response.status_code,
|
|
response=self.load_response(find_dns_zone_response),
|
|
)
|
|
)
|
|
|
|
result = find_dns_zone_response.json()["result"]
|
|
self._domain_zone_id_cache = {}
|
|
for i in result:
|
|
self._domain_zone_id_cache[i["name"]] = i["id"]
|
|
|
|
for domain_key, zone_id in self._domain_zone_id_cache.items():
|
|
if domain_key in domain:
|
|
return self._domain_zone_id_cache[domain_key]
|
|
|
|
raise ValueError(
|
|
("Error unable to get DNS zone for domain_name={domain_name}: "
|
|
"status_code={status_code} response={response}").format(
|
|
domain_name=domain,
|
|
status_code=find_dns_zone_response.status_code,
|
|
response=self.load_response(find_dns_zone_response),
|
|
)
|
|
)
|
|
|
|
def get_domain_list(self):
|
|
url = urljoin(self.CLOUDFLARE_API_BASE_URL, "zones?status=active&per_page=1000")
|
|
headers = self._get_auth_headers()
|
|
find_dns_zone_response = requests.get(url, headers=headers, timeout=self.HTTP_TIMEOUT)
|
|
if find_dns_zone_response.status_code != 200:
|
|
return {}
|
|
domains = find_dns_zone_response.json()["result"]
|
|
return {d["name"]: d for d in domains}
|
|
|
|
def add_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
zone_id = self.find_dns_zone(domain)
|
|
url = urljoin(
|
|
self.CLOUDFLARE_API_BASE_URL,
|
|
"zones/{0}/dns_records".format(zone_id),
|
|
)
|
|
headers = self._get_auth_headers()
|
|
body = {
|
|
"type": r_type,
|
|
"name": host,
|
|
"content": "{0}".format(value),
|
|
"ttl": ttl
|
|
}
|
|
|
|
create_cloudflare_dns_record_response = requests.post(
|
|
url, headers=headers, json=body, timeout=self.HTTP_TIMEOUT
|
|
)
|
|
if create_cloudflare_dns_record_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error creating cloudflare dns record: status_code={status_code} response={response}".format(
|
|
status_code=create_cloudflare_dns_record_response.status_code,
|
|
response=self.load_response(create_cloudflare_dns_record_response),
|
|
)
|
|
)
|
|
|
|
def get_record_list_by_view(self, domain):
|
|
records = self.get_record_list(domain)
|
|
if records is None:
|
|
return []
|
|
res = []
|
|
for record in records:
|
|
res.append({
|
|
"host": record['name'],
|
|
"type": record['type'],
|
|
"value": record['content'],
|
|
"ttl": record['ttl'],
|
|
"modified_on": datetime.datetime.strptime(record['updated_on'], "%Y-%m-%dT%H:%M:%S.%fZ").timestamp(),
|
|
})
|
|
return res
|
|
|
|
def get_record_list(self, domain, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
zone_id = self.find_dns_zone(domain)
|
|
list_dns_url = urljoin(
|
|
self.CLOUDFLARE_API_BASE_URL,
|
|
"zones/{0}/dns_records?per_page=50000".format(zone_id),
|
|
)
|
|
list_dns_payload = {}
|
|
if host is not None and isinstance(host, str):
|
|
list_dns_payload["name"] = host
|
|
if r_type is not None and isinstance(r_type, str):
|
|
list_dns_payload["type"] = r_type
|
|
|
|
headers = self._get_auth_headers()
|
|
list_response = requests.get(
|
|
list_dns_url, params=list_dns_payload, headers=headers, timeout=self.HTTP_TIMEOUT
|
|
).json()
|
|
if "success" and list_response["success"] is False:
|
|
return None
|
|
if "result" in list_response:
|
|
return list_response["result"]
|
|
else:
|
|
return None
|
|
|
|
def remove_record(self, domain: str, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
headers = self._get_auth_headers()
|
|
zone_id = self.find_dns_zone(domain)
|
|
records = self.get_record_list(domain, host, r_type)
|
|
if records is None:
|
|
return False, "未查询到记录信息，无法执行修改"
|
|
|
|
for record in records:
|
|
if record['name'] != host or record['type'] != r_type:
|
|
continue
|
|
dns_record_id = record["id"]
|
|
url = urljoin(
|
|
self.CLOUDFLARE_API_BASE_URL,
|
|
"zones/{0}/dns_records/{1}".format(zone_id, dns_record_id),
|
|
)
|
|
requests.delete(
|
|
url, headers=headers, timeout=self.HTTP_TIMEOUT
|
|
)
|
|
|
|
def modify_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
records = self.get_record_list(domain, host, r_type)
|
|
if records is None:
|
|
return False, "未查询到记录信息，无法执行修改"
|
|
|
|
headers = self._get_auth_headers()
|
|
zone_id = self.find_dns_zone(domain)
|
|
|
|
for record in records:
|
|
if record['name'] != host and record['type'] != r_type:
|
|
continue
|
|
|
|
record_id = record["id"]
|
|
url = urljoin(
|
|
self.CLOUDFLARE_API_BASE_URL,
|
|
"zones/{0}/dns_records/{1}".format(zone_id, record_id),
|
|
)
|
|
|
|
body = {
|
|
"type": r_type,
|
|
"name": host,
|
|
"content": "{0}".format(value),
|
|
"ttl": ttl
|
|
}
|
|
modify_response = requests.put(
|
|
url, data=body, timeout=self.HTTP_TIMEOUT, headers=headers
|
|
).json()
|
|
|
|
if modify_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error creating cloudflare dns record: status_code={status_code} response={response}".format(
|
|
status_code=modify_response.status_code,
|
|
response=self.load_response(modify_response),
|
|
)
|
|
)
|
|
|
|
@classmethod
|
|
def new(cls, conf_data: dict) -> BaseDns:
|
|
key = conf_data.get("key", None) or conf_data.get("E-Mail", None)
|
|
secret = conf_data.get("secret", None) or conf_data.get("API Key", None)
|
|
base_url = "https://api.cloudflare.com/client/v4/"
|
|
|
|
if key is None and secret is None:
|
|
secret = conf_data.get("API Token", None) # 处理api - token的情况
|
|
if key is None and secret is None:
|
|
raise Exception("没有找到有效的DNS API密钥信息")
|
|
return cls(key, secret, base_url)
|
|
|
|
|
|
class AliyunDns(BaseDns):
|
|
dns_provider_name = "alidns"
|
|
|
|
def __init__(self, key, secret):
|
|
self.key = str(key).strip()
|
|
self.secret = str(secret).strip()
|
|
self.url = "https://alidns.aliyuncs.com"
|
|
super(AliyunDns, self).__init__()
|
|
|
|
def sign(self, accessKeySecret, parameters): # '''签名方法
|
|
def percent_encode(encodeStr):
|
|
import urllib.request
|
|
encodeStr = str(encodeStr)
|
|
res = urllib.request.quote(encodeStr, '')
|
|
res = res.replace('+', '%20')
|
|
res = res.replace('*', '%2A')
|
|
res = res.replace('%7E', '~')
|
|
return res
|
|
|
|
sortedParameters = sorted(parameters.items(), key=lambda parameters: parameters[0])
|
|
canonicalizedQueryString = ''
|
|
for (k, v) in sortedParameters:
|
|
canonicalizedQueryString += '&' + percent_encode(k) + '=' + percent_encode(v)
|
|
stringToSign = 'GET&%2F&' + percent_encode(canonicalizedQueryString[1:])
|
|
if sys.version_info[0] == 2:
|
|
h = hmac.new(accessKeySecret + "&", stringToSign, sha1)
|
|
else:
|
|
h = hmac.new(bytes(accessKeySecret + "&", encoding="utf8"), stringToSign.encode('utf8'), sha1)
|
|
signature = base64.encodebytes(h.digest()).strip()
|
|
return signature
|
|
|
|
def add_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
random_int = random.randint(11111111111111, 99999999999999)
|
|
now = datetime.datetime.utcnow()
|
|
otherStyleTime = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
paramsdata = {
|
|
"Action": "AddDomainRecord",
|
|
"Format": "json",
|
|
"Version": "2015-01-09",
|
|
"SignatureMethod": "HMAC-SHA1",
|
|
"Timestamp": otherStyleTime,
|
|
"SignatureVersion": "1.0",
|
|
"SignatureNonce": str(random_int),
|
|
"AccessKeyId": self.key,
|
|
"DomainName": domain,
|
|
"RR": host,
|
|
"Type": r_type,
|
|
"Value": value,
|
|
"TTL": ttl,
|
|
}
|
|
|
|
Signature = self.sign(self.secret, paramsdata)
|
|
paramsdata['Signature'] = Signature
|
|
req = requests.get(url=self.url, params=paramsdata)
|
|
if req.status_code != 200:
|
|
if req.json()['Code'] == 'IncorrectDomainUser' or req.json()['Code'] == 'InvalidDomainName.NoExist':
|
|
raise ValueError("这个阿里云账户下面不存在这个域名，添加解析失败")
|
|
elif req.json()['Code'] == 'InvalidAccessKeyId.NotFound' or req.json()['Code'] == 'SignatureDoesNotMatch':
|
|
raise ValueError("API密钥错误，添加解析失败")
|
|
else:
|
|
raise ValueError(req.json()['Message'])
|
|
|
|
def get_record_list(self, domain, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
random_int = random.randint(11111111111111, 99999999999999)
|
|
now = datetime.datetime.utcnow()
|
|
otherStyleTime = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
params_data = {
|
|
"Action": "DescribeDomainRecords",
|
|
"Format": "json",
|
|
"Version": "2015-01-09",
|
|
"SignatureMethod": "HMAC-SHA1",
|
|
"Timestamp": otherStyleTime,
|
|
"SignatureVersion": "1.0",
|
|
"SignatureNonce": str(random_int),
|
|
"AccessKeyId": self.key,
|
|
"DomainName": domain,
|
|
"PageNumber": 1,
|
|
"PageSize": 1000,
|
|
}
|
|
if host is not None and isinstance(host, str):
|
|
params_data["RRKeyWord"] = host
|
|
if r_type is not None and isinstance(r_type, str):
|
|
params_data["TypeKeyWord"] = r_type
|
|
|
|
Signature = self.sign(self.secret, params_data)
|
|
params_data['Signature'] = Signature
|
|
req_data = requests.get(url=self.url, params=params_data).json()
|
|
record_list = req_data.get("DomainRecords", {}).get("Record", [])
|
|
return record_list
|
|
|
|
def get_record_list_by_view(self, domain):
|
|
records = self.get_record_list(domain)
|
|
if records is None:
|
|
return []
|
|
res = []
|
|
for record in records:
|
|
res.append({
|
|
"host": record['RR'],
|
|
"type": record['Type'],
|
|
"value": record['Value'],
|
|
"ttl": record['ttl'],
|
|
"modified_on": record['UpdateTimestamp'],
|
|
})
|
|
return res
|
|
|
|
def remove_record(self, domain: str, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
records = self.get_record_list(domain, host, r_type)
|
|
if records is None:
|
|
return False, "未查询到记录信息，无法执行修改"
|
|
|
|
random_int = random.randint(11111111111111, 99999999999999)
|
|
now = datetime.datetime.utcnow()
|
|
otherStyleTime = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
params_data = {
|
|
"Action": "DeleteDomainRecord",
|
|
"Format": "json",
|
|
"Version": "2015-01-09",
|
|
"SignatureMethod": "HMAC-SHA1",
|
|
"Timestamp": otherStyleTime,
|
|
"SignatureVersion": "1.0",
|
|
"SignatureNonce": str(random_int),
|
|
"AccessKeyId": self.key,
|
|
"RecordId": None,
|
|
}
|
|
for record in records:
|
|
if record['RR'] != host or record['Type'] != r_type:
|
|
continue
|
|
record_id = record["RecordId"]
|
|
p_data = params_data.copy()
|
|
p_data["RecordId"] = record_id
|
|
|
|
Signature = self.sign(self.secret, p_data)
|
|
p_data['Signature'] = Signature
|
|
req = requests.get(url=self.url, params=p_data)
|
|
if req.status_code != 200:
|
|
raise ValueError("删除解析记录失败")
|
|
|
|
def modify_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
records = self.get_record_list(domain, host, r_type)
|
|
if records is None:
|
|
return False, "未查询到记录信息，无法执行修改"
|
|
|
|
random_int = random.randint(11111111111111, 99999999999999)
|
|
now = datetime.datetime.utcnow()
|
|
otherStyleTime = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
params_data = {
|
|
"Action": "UpdateDomainRecord",
|
|
"Format": "json",
|
|
"Version": "2015-01-09",
|
|
"SignatureMethod": "HMAC-SHA1",
|
|
"Timestamp": otherStyleTime,
|
|
"SignatureVersion": "1.0",
|
|
"SignatureNonce": str(random_int),
|
|
"AccessKeyId": self.key,
|
|
"RR": host,
|
|
"Type": r_type,
|
|
"Value": value,
|
|
"TTL": ttl,
|
|
"RecordId": None
|
|
}
|
|
|
|
for record in records:
|
|
if record['RR'] != host and record['Type'] != r_type:
|
|
continue
|
|
|
|
record_id = record["RecordId"]
|
|
p_data = params_data.copy()
|
|
p_data["RecordId"] = record_id
|
|
|
|
Signature = self.sign(self.secret, p_data)
|
|
p_data['Signature'] = Signature
|
|
req = requests.get(url=self.url, params=p_data)
|
|
if req.status_code != 200:
|
|
if req.json()['Code'] == 'IncorrectDomainUser' or req.json()['Code'] == 'InvalidDomainName.NoExist':
|
|
raise ValueError("这个阿里云账户下面不存在这个域名，添加解析失败")
|
|
elif req.json()['Code'] == 'InvalidAccessKeyId.NotFound' or req.json()['Code'] == 'SignatureDoesNotMatch':
|
|
raise ValueError("API密钥错误，添加解析失败")
|
|
else:
|
|
raise ValueError(req.json()['Message'])
|
|
|
|
def get_domain_list(self):
|
|
random_int = random.randint(11111111111111, 99999999999999)
|
|
now = datetime.datetime.utcnow()
|
|
otherStyleTime = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
paramsdata = {
|
|
"Action": "DescribeDomains",
|
|
"Format": "json",
|
|
"Version": "2015-01-09",
|
|
"SignatureMethod": "HMAC-SHA1",
|
|
"Timestamp": otherStyleTime,
|
|
"SignatureVersion": "1.0",
|
|
"SignatureNonce": str(random_int),
|
|
"AccessKeyId": self.key,
|
|
"PageNumber": 1,
|
|
"PageSize": 1000,
|
|
}
|
|
|
|
Signature = self.sign(self.secret, paramsdata)
|
|
paramsdata['Signature'] = Signature
|
|
req = requests.get(url=self.url, params=paramsdata)
|
|
if req.status_code != 200:
|
|
req_json = req.json()
|
|
if req_json['Code'] == 'IncorrectDomainUser' or req_json['Code'] == 'InvalidDomainName.NoExist':
|
|
raise ValueError("这个阿里云账户下面不存在这个域名，添加解析失败")
|
|
elif req_json['Code'] == 'InvalidAccessKeyId.NotFound' or req_json['Code'] == 'SignatureDoesNotMatch':
|
|
raise ValueError("API密钥错误，添加解析失败")
|
|
else:
|
|
raise ValueError(req_json['Message'])
|
|
|
|
domains = req.json()["Domains"]["Domain"]
|
|
return {d["DomainName"]: d for d in domains}
|
|
|
|
@classmethod
|
|
def new(cls, conf_data: dict) -> "AliyunDns":
|
|
key = conf_data.get("key", None) or conf_data.get("AccessKey", "")
|
|
secret = conf_data.get("secret", None) or conf_data.get("SecretKey", "")
|
|
|
|
return cls(key, secret)
|
|
|
|
|
|
# 未完善请勿使用
|
|
class GoDaddyDns(BaseDns):
|
|
|
|
_type = 0 # 0:lest 1：锐成
|
|
http_timeout = 65
|
|
|
|
def __init__(self, sso_key: str, sso_secret: str, base_url='https://api.godaddy.com'):
|
|
self.sso_key = sso_key
|
|
self.sso_secret = sso_secret
|
|
self.base_url = base_url
|
|
super(GoDaddyDns, self).__init__()
|
|
self._headers = None
|
|
|
|
def _get_auth_headers(self) -> dict:
|
|
if self._headers is not None:
|
|
return self._headers
|
|
self._headers = {
|
|
"Authorization": "sso-key {}:{}".format(self.sso_key, self.sso_secret)
|
|
}
|
|
return self._headers
|
|
|
|
|
|
def add_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
url = urljoin(
|
|
self.base_url,
|
|
"/v1/domains/{}/records".format(root),
|
|
)
|
|
headers = self._get_auth_headers()
|
|
|
|
body = [{
|
|
"type": s_type,
|
|
"name": host,
|
|
"data": "{0}".format(value),
|
|
}]
|
|
|
|
create_cloudflare_dns_record_response = requests.patch(
|
|
url, headers=headers, json=body, timeout=self.http_timeout
|
|
)
|
|
if create_cloudflare_dns_record_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error creating cloudflare dns record: status_code={status_code} response={response}".format(
|
|
status_code=create_cloudflare_dns_record_response.status_code,
|
|
response=self.load_response(create_cloudflare_dns_record_response),
|
|
)
|
|
)
|
|
|
|
def modify_record(self, domain: str, host: str, r_type: str, value: str, weight: int = 1, ttl=600):
|
|
pass
|
|
|
|
|
|
def remove_record(self, domain: str, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
headers = self._get_auth_headers()
|
|
|
|
list_dns_url = urljoin(
|
|
self.base_url,
|
|
"/v1/domains/{}/records/{}/{}".format(domain_name, s_type, dns_name),
|
|
)
|
|
|
|
dns_response = requests.delete(
|
|
list_dns_url, headers=headers, timeout=self.http_timeout
|
|
)
|
|
if dns_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error creating cloudflare dns record: status_code={status_code} response={response}".format(
|
|
status_code=dns_response.status_code,
|
|
response=self.load_response(dns_response),
|
|
)
|
|
)
|
|
|
|
def get_record_list(self, domain, host: Optional[str] = None, r_type: Optional[str] = None):
|
|
pass
|
|
|
|
def get_record_list_by_view(self, domain):
|
|
pass
|
|
|
|
|
|
@classmethod
|
|
def new(cls, conf_data: dict) -> BaseDns:
|
|
key = conf_data.get("key", None) or conf_data.get("Key", "")
|
|
secret = conf_data.get("secret", None) or conf_data.get("Secret", "")
|
|
base_url = "https://api.godaddy.com"
|
|
|
|
return cls(key, secret, base_url)
|
|
|
|
|
|
# 未上线
|
|
class DNSLADns(BaseDns):
|
|
dns_provider_name = "dnsla"
|
|
_type = 0 # 0:lest 1：锐成
|
|
|
|
def __init__(self, api_id, api_secret):
|
|
self.api_id = api_id
|
|
self.api_secret = api_secret
|
|
self.base_url = "https://api.dns.la"
|
|
self.http_timeout = 65 # seconds
|
|
self._token = None
|
|
self.domain_list = None
|
|
super(DNSLADns, self).__init__()
|
|
|
|
@classmethod
|
|
def new(cls, conf_data) -> BaseDns:
|
|
key = conf_data.get("key", None) or conf_data.get("APIID", "")
|
|
secret = conf_data.get("secret", None) or conf_data.get("API密钥", "")
|
|
|
|
return cls(key, secret)
|
|
|
|
def _get_auth_headers(self) -> dict:
|
|
if self._token is None:
|
|
self._token = base64.b64encode("{}:{}".format(self.api_id, self.api_secret).encode("utf-8")).decode("utf-8")
|
|
return {"Authorization": "Basic " + self._token}
|
|
|
|
def find_dns_zone(self, domain_name):
|
|
url = urljoin(self.base_url, "/api/domainList?pageIndex=1&pageSize=1000")
|
|
headers = self._get_auth_headers()
|
|
find_dns_zone_response = requests.get(url, headers=headers, timeout=self.http_timeout)
|
|
if find_dns_zone_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error creating DNS.LA domains: status_code={status_code} response={response}".format(
|
|
status_code=find_dns_zone_response.status_code,
|
|
response=self.load_response(find_dns_zone_response),
|
|
)
|
|
)
|
|
|
|
result = find_dns_zone_response.json()["data"]["results"]
|
|
self.domain_list = result
|
|
have = False
|
|
for domain_data in result:
|
|
if domain_data["domain"].rstrip(".") == domain_name:
|
|
have = True
|
|
break
|
|
|
|
if not have:
|
|
raise ValueError(
|
|
(
|
|
"Error unable to get DNS zone for domain_name={domain_name}: "
|
|
"status_code={status_code} response={response}"
|
|
).format(
|
|
domain_name=domain_name,
|
|
status_code=find_dns_zone_response.status_code,
|
|
response=self.load_response(find_dns_zone_response),
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _format_s_type_to_request(s_type):
|
|
trans = {
|
|
"A": 1,
|
|
"NS": 2,
|
|
"CNAME": 5,
|
|
"MX": 15,
|
|
"TXT": 16,
|
|
"AAAA": 28,
|
|
"SRV": 33,
|
|
"CAA": 257,
|
|
"URL": 256
|
|
}
|
|
|
|
if isinstance(s_type, (int, float)):
|
|
if int(s_type) in trans.values():
|
|
return int(s_type)
|
|
|
|
if isinstance(s_type, str):
|
|
if s_type in trans:
|
|
return trans[s_type]
|
|
return 16
|
|
|
|
def add_record(self, domain, s_type, host, value):
|
|
url = urljoin(self.base_url, "/api/record", )
|
|
headers = self._get_auth_headers()
|
|
|
|
domain_id = self._get_domain_id(domain)
|
|
|
|
body = {
|
|
"domainId": domain_id,
|
|
"type": self._format_s_type_to_request(s_type),
|
|
"host": host,
|
|
"data": value,
|
|
"ttl": 600,
|
|
}
|
|
|
|
create_dns_la_record_response = requests.post(
|
|
url, headers=headers, json=body, timeout=self.http_timeout
|
|
)
|
|
if create_dns_la_record_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error creating cloudflare dns record: status_code={status_code} response={response}".format(
|
|
status_code=create_dns_la_record_response.status_code,
|
|
response=self.load_response(create_dns_la_record_response),
|
|
)
|
|
)
|
|
|
|
def create_dns_record(self, domain_name, domain_dns_value):
|
|
domain_name = domain_name.lstrip("*.")
|
|
root, zone, acme_txt = extract_zone(domain_name)
|
|
self.find_dns_zone(root)
|
|
|
|
if self._type == 1:
|
|
return self.add_record(root, 'CNAME', acme_txt.replace('_acme-challenge.', ''), domain_dns_value)
|
|
else:
|
|
return self.add_record(root, 'TXT', acme_txt, domain_dns_value)
|
|
|
|
def add_record_for_creat_site(self, domain, server_ip):
|
|
root, zone, _ = extract_zone(domain)
|
|
self.add_record(root, "A", zone, server_ip)
|
|
|
|
def get_record_list(self, domain_id: str) -> list:
|
|
url = urljoin(self.base_url, "/api/recordList?pageIndex=1&pageSize=10&domainId={}".format(domain_id))
|
|
headers = self._get_auth_headers()
|
|
get_record_list_response = requests.get(url, headers=headers, timeout=self.http_timeout)
|
|
if get_record_list_response.status_code != 200:
|
|
raise ValueError(
|
|
"Error unable to get record list : status_code={status_code} response={response}".format(
|
|
status_code=get_record_list_response.status_code,
|
|
response=self.load_response(get_record_list_response),
|
|
)
|
|
)
|
|
|
|
result = get_record_list_response.json()["data"]["results"]
|
|
if isinstance(result, list):
|
|
return result
|
|
else:
|
|
return []
|
|
|
|
def _get_domain_id(self, domain_name: str) -> str:
|
|
if domain_name.count('.') > 2:
|
|
domain_name, _, _ = extract_zone(domain_name)
|
|
if self.domain_list is None:
|
|
self.find_dns_zone(domain_name)
|
|
domain_id = None
|
|
for domain_data in self.domain_list:
|
|
if domain_data["domain"].rstrip(".") == domain_name:
|
|
domain_id = domain_data["id"]
|
|
if domain_id is None:
|
|
raise ValueError(
|
|
"Error unable to get DNS zone for domain_name={domain_name}".format(domain_name=domain_name))
|
|
return domain_id
|
|
|
|
def remove_record(self, domain_name, dns_name, s_type):
|
|
domain_id = self._get_domain_id(domain_name)
|
|
record_list = self.get_record_list(domain_id)
|
|
trans_type = self._format_s_type_to_request(s_type)
|
|
remove_record_id_list = []
|
|
for record in record_list:
|
|
if record["type"] == trans_type and (record["host"] == dns_name or record["displayHost"] == dns_name):
|
|
remove_record_id_list.append(record["id"])
|
|
|
|
headers = self._get_auth_headers()
|
|
|
|
del_record_url_list = [urljoin(self.base_url, "/api/record?id={}".format(i)) for i in remove_record_id_list]
|
|
|
|
for del_record_url in del_record_url_list:
|
|
requests.delete(
|
|
del_record_url, headers=headers, timeout=self.http_timeout
|
|
)
|
|
|
|
def delete_dns_record(self, domain_name, domain_dns_value):
|
|
domain_name = domain_name.lstrip("*.")
|
|
root, zone, acme_txt = extract_zone(domain_name)
|
|
self.remove_record(root, acme_txt, 'TXT')
|
|
|
|
|
|
class DomainListCache(object):
|
|
_CACHE_TIP = '{}/plugin/dns_api_manager/cache.tip'.format(public.get_panel_path())
|
|
_CACHE_DATA = '{}/plugin/dns_api_manager/cache.json'.format(public.get_panel_path())
|
|
|
|
def __init__(self):
|
|
self._cache = None
|
|
self.read()
|
|
|
|
def clear(self):
|
|
self._cache = {}
|
|
if os.path.exists(self._CACHE_DATA):
|
|
os.remove(self._CACHE_DATA)
|
|
|
|
def set(self, cache_id, data: dict):
|
|
self._cache[cache_id] = data
|
|
|
|
def get(self, cache_id):
|
|
if cache_id in self._cache:
|
|
return self._cache[cache_id]
|
|
return None
|
|
|
|
def save(self):
|
|
now = int(datetime.datetime.now().timestamp())
|
|
if self._cache:
|
|
res = public.writeFile(self._CACHE_DATA, json.dumps(self._cache))
|
|
public.writeFile(self._CACHE_TIP, str(now + 60 * 60))
|
|
|
|
def _remove_cache(self):
|
|
now = int(datetime.datetime.now().timestamp())
|
|
last = int(public.ReadFile(self._CACHE_TIP))
|
|
if last < now:
|
|
if os.path.exists(self._CACHE_DATA):
|
|
os.remove(self._CACHE_DATA)
|
|
|
|
def read(self):
|
|
self._remove_cache()
|
|
if not os.path.exists(self._CACHE_DATA):
|
|
self._cache = {}
|
|
|
|
try:
|
|
data = json.loads(public.readFile(self._CACHE_DATA))
|
|
if not isinstance(data, dict):
|
|
data = {}
|
|
except:
|
|
data = {}
|
|
|
|
self._cache = data
|
|
|
|
|
|
class RealDnsMager(object):
|
|
"""
|
|
config = {
|
|
"CloudFlareDns": [
|
|
{
|
|
"E-Mail": "122456944@qq.com",
|
|
"API Key": "dsgvfcdkjausvgfkjasdfgakj",
|
|
"ps": "xxx",
|
|
"id": 1,
|
|
"domains": [ # domains 可以不存在，内容是根域名
|
|
]
|
|
}
|
|
]
|
|
........
|
|
}"""
|
|
|
|
CONF_FILE = "{}/config/dns_mager.conf".format(public.get_panel_path())
|
|
CLS_MAP: Dict = {
|
|
"AliyunDns": AliyunDns,
|
|
"DNSPodDns": DNSPodDns,
|
|
"CloudFlareDns": CloudFlareDns,
|
|
"GoDaddyDns": GoDaddyDns,
|
|
"DNSLADns": DNSLADns,
|
|
}
|
|
|
|
NOT_USED_LIST: list = ["GoDaddyDns", "DNSLADns"]
|
|
|
|
RULE_MAP: Dict[str, List[str]] = {
|
|
"AliyunDns": ["AccessKey", "SecretKey"],
|
|
"DNSPodDns": ["ID", "Token"],
|
|
"CloudFlareDns": ["E-Mail", "API Key"],
|
|
"GoDaddyDns": ["Key", "Secret"],
|
|
"DNSLADns": ["APIID", "API密钥"]
|
|
}
|
|
|
|
def __init__(self):
|
|
self._config: Optional[Dict[str, Dict[str, Union[int, str]]]] = None
|
|
|
|
@staticmethod
|
|
def _get_new_id() -> str:
|
|
return uuid4().hex
|
|
|
|
@classmethod
|
|
def _read_config_old(cls) -> Optional[Dict[str, List[Dict[str, Union[str, list]]]]]:
|
|
old_config_file = "{}/config/dns_api.json".format(public.get_panel_path())
|
|
if os.path.isfile(old_config_file):
|
|
try:
|
|
data = json.loads(public.readFile(old_config_file))
|
|
except json.JSONDecodeError:
|
|
return None
|
|
res = {}
|
|
rule_list = ("AliyunDns", "DNSPodDns", "CloudFlareDns", "GoDaddyDns")
|
|
if isinstance(data, list):
|
|
for d in data:
|
|
if d["name"] not in rule_list:
|
|
continue
|
|
|
|
conf_data = d.get("data", None)
|
|
if isinstance(data, list):
|
|
tmp = {i["name"]: i["value"] for i in conf_data if i["value"].strip()}
|
|
tmp["ps"] = "默认账户"
|
|
tmp["id"] = cls._get_new_id()
|
|
if len(tmp) > 2:
|
|
res[d["name"]] = [tmp]
|
|
|
|
if res:
|
|
return res
|
|
return None
|
|
|
|
@staticmethod
|
|
def _get_acme_dns_api() -> Dict[str, Dict[str, str]]:
|
|
path = '/root/.acme.sh'
|
|
if not os.path.exists(path + '/account.conf'):
|
|
path = "/.acme.sh"
|
|
account = public.readFile(path + '/account.conf')
|
|
if not account:
|
|
return {}
|
|
rule_map: Dict[str, Dict[str, str]] = {
|
|
"AliyunDns": {
|
|
"AccessKey": "SAVED_Ali_Key",
|
|
"SecretKey": "SAVED_Ali_Secret",
|
|
},
|
|
"DNSPodDns": {
|
|
"ID": "SAVED_DP_Id",
|
|
"Token": "SAVED_DP_Key"
|
|
},
|
|
"CloudFlareDns": {
|
|
"E-Mail": "SAVED_CF_MAIL",
|
|
"API Key": "SAVED_CF_KEY",
|
|
},
|
|
"GoDaddyDns": {
|
|
"Key": "SAVED_GD_Key",
|
|
"Secret": "SAVED_GD_Secret",
|
|
},
|
|
"DNSLADns": {
|
|
"APIID": "SAVED_LA_Id",
|
|
"API密钥": "SAVED_LA_Key"
|
|
}
|
|
}
|
|
res = {}
|
|
for rule_name, rule in rule_map.items():
|
|
tmp = {}
|
|
for r_key, r_value in rule.items():
|
|
account_res = re.search(r_value + r"\s*=\s*'(.+)'", account)
|
|
if account_res:
|
|
tmp[r_key] = account_res.groups()[0]
|
|
|
|
if len(tmp) == len(rule):
|
|
res[rule_name] = tmp
|
|
return res
|
|
|
|
@property
|
|
def config(self) -> dict:
|
|
if self._config is not None:
|
|
return self._config
|
|
|
|
change = False
|
|
if not os.path.exists(self.CONF_FILE):
|
|
change = True
|
|
old_config = self._read_config_old()
|
|
if old_config is not None:
|
|
self._config = old_config
|
|
else:
|
|
self._config = {}
|
|
|
|
l_data = self._get_config_data_from_letsencrypt_data()
|
|
if l_data is not None:
|
|
for tmp_conf in l_data:
|
|
key = tmp_conf["dns_name"]
|
|
if key not in self._config:
|
|
self._config[key] = []
|
|
for v in self._config[key]:
|
|
if all([v.get(n, None) == m for n, m in tmp_conf["conf_data"].items()]):
|
|
break
|
|
else:
|
|
tmp_data = {
|
|
"ps": "配置中的默认账户",
|
|
"id": self._get_new_id(),
|
|
}
|
|
tmp_data.update(tmp_conf["conf_data"])
|
|
self._config[key].append(tmp_data)
|
|
else:
|
|
try:
|
|
config_data = json.loads(public.readFile(self.CONF_FILE))
|
|
except json.JSONDecodeError:
|
|
self._config = {}
|
|
else:
|
|
if isinstance(config_data, dict):
|
|
self._config = config_data
|
|
else:
|
|
self._config = {}
|
|
|
|
acme_conf = self._get_acme_dns_api()
|
|
if acme_conf:
|
|
for key, value in acme_conf.items():
|
|
if key not in self._config:
|
|
self._config[key] = []
|
|
for v in self._config[key]:
|
|
if all([v.get(n, None) == m for n, m in value.items()]):
|
|
break
|
|
else:
|
|
change = True
|
|
value["ps"] = "检测到的acme_dns"
|
|
value["id"] = self._get_new_id()
|
|
self._config[key].append(value)
|
|
|
|
if change:
|
|
self.save_config()
|
|
|
|
return self._config
|
|
|
|
def save_config(self):
|
|
if self._config is None:
|
|
_ = self.config
|
|
public.writeFile(self.CONF_FILE, json.dumps(self._config))
|
|
|
|
def get_dns_obj_by_domain(self, domain) -> BaseDns:
|
|
root, _ = extract_zone(domain)
|
|
for key, value in self.config.items():
|
|
for dns_config in value:
|
|
if root in dns_config.get("domains", []):
|
|
return self.CLS_MAP[key].new(dns_config)
|
|
raise Exception("没有找到域名为{}的有效的DNS API密钥信息".format(domain))
|
|
|
|
def add_conf(self,
|
|
dns_type: str,
|
|
conf_data: List[dict],
|
|
ps: str,
|
|
domains: List[str],
|
|
force_domain: Optional[str]):
|
|
|
|
if dns_type in self.NOT_USED_LIST:
|
|
return False, "当前DNS平台还未完全支持，请等待后续更新"
|
|
|
|
if dns_type not in self.CLS_MAP:
|
|
return False, "不支持的DNS平台"
|
|
|
|
f, data = self._parse_data(conf_data, dns_type)
|
|
if not f:
|
|
return False, data
|
|
|
|
if not isinstance(domains, list):
|
|
return False, "The domain name parameter is incorrectly formatted"
|
|
|
|
if dns_type not in self.config:
|
|
self.config[dns_type] = []
|
|
|
|
for v in self.config[dns_type]:
|
|
if all([v.get(n, None) == m for n, m in data.items()]):
|
|
return False, "该通行凭证已添加过"
|
|
|
|
data["ps"] = ps
|
|
data["id"] = self._get_new_id()
|
|
root_list = self.paser_domains_list_to_root_list(domains)
|
|
all_domains = self._get_all_root(with_out=None)
|
|
for root in root_list:
|
|
if root in all_domains:
|
|
return False, "域名{}已绑定其他api账户，不能添加".format(root)
|
|
|
|
if force_domain is not None and not isinstance(force_domain, str):
|
|
return False, "The domain name parameter is incorrectly formatted"
|
|
if force_domain is not None:
|
|
force_root = self.paser_domains_list_to_root_list([force_domain])[0]
|
|
if force_root not in root_list:
|
|
if force_root in all_domains:
|
|
self.remove_domains_by_root(force_root)
|
|
|
|
data["domains"] = root_list
|
|
self.config[dns_type].append(data)
|
|
self.save_config()
|
|
return True, "Saved successfully"
|
|
|
|
def _parse_data(self, conf_data: List[dict], dns_type: str) -> Tuple[bool, Union[dict, str]]:
|
|
data = {}
|
|
if not isinstance(conf_data, list):
|
|
return False, "Parameter format error"
|
|
for conf in conf_data:
|
|
if isinstance(conf, dict) and "name" in conf and "value" in conf:
|
|
data[conf.get("name")] = str(conf.get("value")).strip()
|
|
if not data:
|
|
return False, "Parameter format error,没有指定参数"
|
|
if dns_type == "CloudFlareDns" and len(data) == 1 and "API Token" in data:
|
|
return True, data
|
|
|
|
for n in self.RULE_MAP[dns_type]:
|
|
if n not in data:
|
|
return False, "Parameter format error,参数名称与平台不对应"
|
|
|
|
return True, data
|
|
|
|
def modify_conf(self,
|
|
api_id: str,
|
|
dns_type: str,
|
|
conf_data: Optional[List[Dict]],
|
|
ps: Optional[str],
|
|
domains: Optional[List[str]],
|
|
force_domain: Optional[str]): # 强制添加的域名
|
|
|
|
if dns_type in self.NOT_USED_LIST:
|
|
return False, "当前DNS平台还未完全支持，请等待后续更新"
|
|
|
|
if dns_type not in self.CLS_MAP:
|
|
return False, "不支持的DNS平台"
|
|
|
|
target_idx = -1
|
|
if dns_type not in self.config:
|
|
self.config[dns_type] = []
|
|
|
|
for idx, v in enumerate(self.config[dns_type]):
|
|
if api_id == v.get("id", None):
|
|
target_idx = idx
|
|
|
|
if target_idx == -1:
|
|
return False, "不存在这个通行凭证"
|
|
|
|
if conf_data is not None:
|
|
f, data = self._parse_data(conf_data, dns_type)
|
|
if not f:
|
|
return False, data
|
|
|
|
self.config[dns_type][target_idx].update(**data)
|
|
if ps is not None:
|
|
self.config[dns_type][target_idx].update(ps=ps)
|
|
|
|
if domains is not None and not isinstance(domains, list):
|
|
return False, "The domain name parameter is incorrectly formatted"
|
|
|
|
if domains is not None:
|
|
root_list = self.paser_domains_list_to_root_list(domains)
|
|
all_domains = self._get_all_root(with_out=self.config[dns_type][target_idx].get("domains"))
|
|
for root in root_list:
|
|
if root in all_domains:
|
|
return False, "域名{}已绑定其他api账户，不能添加".format(root)
|
|
self.config[dns_type][target_idx]["domains"] = root_list
|
|
|
|
if force_domain is not None and not isinstance(force_domain, str):
|
|
return False, "The domain name parameter is incorrectly formatted"
|
|
if force_domain is not None:
|
|
root = self.paser_domains_list_to_root_list([force_domain])[0]
|
|
self.remove_domains_by_root(root)
|
|
if "domains" not in self.config[dns_type][target_idx]:
|
|
self.config[dns_type][target_idx]["domains"] = []
|
|
self.config[dns_type][target_idx]["domains"].append(root)
|
|
|
|
self.save_config()
|
|
return True, "修改成功"
|
|
|
|
def remove_domains_by_root(self, root: str):
|
|
for key, value in self.config.items():
|
|
for dns_config in value:
|
|
domains = dns_config.get("domains", None)
|
|
if domains is not None and root in domains:
|
|
domains.remove(root)
|
|
|
|
def _get_all_root(self, with_out: Optional[List[str]]) -> Set[str]:
|
|
all_domains = set(
|
|
chain(*[c.get("domains", []) for c in
|
|
chain(*[c_list for c_list in self.config.values()])]
|
|
)
|
|
)
|
|
if with_out is not None:
|
|
return all_domains - set(with_out)
|
|
return all_domains
|
|
|
|
def test_domains_api(self, domains: List[str]) -> List[dict]:
|
|
res = [{}] * len(domains)
|
|
for idx, domain in enumerate(domains):
|
|
root = self.paser_domains_list_to_root_list([domain])[0]
|
|
for key, conf in self.config.items():
|
|
for c in conf:
|
|
if root in c.get("domains", []):
|
|
res[idx] = {
|
|
"dns_name": key,
|
|
"conf": c,
|
|
"rooot": root,
|
|
"domain": domain
|
|
}
|
|
return res
|
|
|
|
def remove_conf(self, api_id: str, dns_type: str):
|
|
if dns_type in self.NOT_USED_LIST:
|
|
return False, "当前DNS平台还未完全支持，请等待后续更新"
|
|
|
|
if dns_type not in self.CLS_MAP:
|
|
return False, "不支持的DNS平台"
|
|
|
|
if dns_type not in self.config:
|
|
self.config[dns_type] = []
|
|
|
|
target_idx = -1
|
|
for idx, v in enumerate(self.config[dns_type]):
|
|
if api_id == v.get("id", None):
|
|
target_idx = idx
|
|
|
|
if target_idx == -1:
|
|
return False, "不存在这个通行凭证"
|
|
|
|
del self.config[dns_type][target_idx]
|
|
self.save_config()
|
|
return True, "删除成功"
|
|
|
|
@classmethod
|
|
def paser_auth_to(cls, auth_to_string: str) -> Tuple[Optional[str], Optional[Dict[str, str]]]:
|
|
tmp = auth_to_string.split('|')
|
|
dns_name = tmp[0]
|
|
if dns_name not in cls.CLS_MAP:
|
|
return None, None
|
|
if len(tmp) != 3:
|
|
return None, None
|
|
|
|
if tmp[2] == "":
|
|
key = None
|
|
secret = tmp[1]
|
|
else:
|
|
key = tmp[1]
|
|
secret = tmp[2]
|
|
|
|
if dns_name == "CloudFlareDns" and key is None:
|
|
return "CloudFlareDns", {"API Token": secret}
|
|
elif key and secret:
|
|
return dns_name, dict(zip(cls.RULE_MAP[dns_name], [key, secret]))
|
|
return None, None
|
|
|
|
@classmethod
|
|
def paser_domains_list_to_root_list(cls, domains_list: List[str]) -> List[str]:
|
|
res = []
|
|
for domain in domains_list:
|
|
root, _ = extract_zone(domain)
|
|
if root in res:
|
|
continue
|
|
res.append(root)
|
|
return res
|
|
|
|
@classmethod
|
|
def _get_config_data_from_letsencrypt_data(cls) -> Optional[List[Dict[str, Union[str, list, dict]]]]:
|
|
conf_file = "{}/config/letsencrypt.json".format(public.get_panel_path())
|
|
if not os.path.exists(conf_file):
|
|
return None
|
|
tmp_config = public.readFile(conf_file)
|
|
try:
|
|
orders = json.loads(tmp_config)["orders"]
|
|
except (json.JSONDecodeError, KeyError):
|
|
return None
|
|
|
|
res = {}
|
|
for order in orders:
|
|
if 'auth_type' in order and order['auth_type'] == "dns":
|
|
if order["auth_to"].find("|") == -1 or order["auth_to"].find("/") != -1: # 文件验证跳过
|
|
continue
|
|
if order["auth_to"] in res:
|
|
tmp_conf = res[order["auth_to"]]
|
|
else:
|
|
dns_name, conf_dict = cls.paser_auth_to(order["auth_to"])
|
|
if dns_name is None:
|
|
continue
|
|
tmp_conf = {
|
|
"dns_name": dns_name,
|
|
"conf_data": conf_dict,
|
|
"domains": []
|
|
}
|
|
res[order["auth_to"]] = tmp_conf
|
|
root_list = order.get("domains", [])
|
|
for root in root_list:
|
|
if root not in tmp_conf["domains"]:
|
|
tmp_conf["domains"].append(root)
|
|
|
|
if len(res) == 0:
|
|
return None
|
|
return list(res.values())
|
|
|
|
|
|
class DNSApiManager:
|
|
_DEFAULT_ERROR = (
|
|
"执行中发生了错误，您可以尝试：<br>"
|
|
"1. 检查API通行证是否有效，或是否填写错误<br>"
|
|
"2. 检查域名是否托管在该平台下<br>"
|
|
"3. 检查解析值是否正确<br>"
|
|
)
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
@staticmethod
|
|
def get_dns_api_conf(get=None):
|
|
api_init = '{}/config/dns_api_init_v2.json'.format(public.get_panel_path())
|
|
|
|
apis = json.loads(public.ReadFile(api_init))
|
|
m = RealDnsMager()
|
|
result = []
|
|
|
|
for data in apis:
|
|
if data["name"] == "dns":
|
|
continue
|
|
if data["name"] in m.NOT_USED_LIST:
|
|
continue
|
|
if data["name"] in m.CLS_MAP:
|
|
conf_list = m.config.get(data["name"], None)
|
|
tmp = []
|
|
if conf_list:
|
|
for conf in conf_list:
|
|
tmp_dict = {
|
|
"ps": conf.pop("ps", ""),
|
|
"domains": conf.pop("domains", []),
|
|
"id": conf.pop("id")
|
|
}
|
|
for table in data["add_table"]:
|
|
if table["fields"][0] in conf:
|
|
tmp_dict["conf"] = [{"name": f, "value": conf.get(f, "")} for f in table["fields"]]
|
|
if "conf" in tmp_dict:
|
|
tmp.append(tmp_dict)
|
|
data["data"] = tmp
|
|
result.append(data)
|
|
|
|
return json_response(status=True, data=result)
|
|
|
|
@staticmethod
|
|
def add_dns_api(get):
|
|
try:
|
|
dns_type = get.dns_type.strip()
|
|
ps = get.ps.strip()
|
|
conf_data = json.loads(get.pdata.strip())
|
|
domains = json.loads(get.domains.strip())
|
|
force_domain = None
|
|
if "force_domain" in get:
|
|
force_domain = get.force_domain.strip()
|
|
except (json.JSONDecodeError, AttributeError, KeyError):
|
|
return json_response(status=False, msg="Parameter error")
|
|
|
|
f, msg = RealDnsMager().add_conf(dns_type, conf_data, ps, domains, force_domain)
|
|
return json_response(status=f, msg=msg)
|
|
|
|
@staticmethod
|
|
def set_dns_api(get):
|
|
try:
|
|
dns_type = get.dns_type.strip()
|
|
api_id = get.api_id.strip()
|
|
ps = None
|
|
conf_data = None
|
|
domains = None
|
|
force_domain = None
|
|
if "ps" in get:
|
|
ps = get.ps.strip()
|
|
if "force_domain" in get:
|
|
force_domain = get.force_domain.strip()
|
|
if "pdata" in get:
|
|
conf_data = json.loads(get.pdata.strip())
|
|
if "domains" in get:
|
|
domains = json.loads(get.domains.strip())
|
|
except (json.JSONDecodeError, AttributeError, KeyError):
|
|
return json_response(status=False, msg="Parameter error")
|
|
try:
|
|
f, msg = RealDnsMager().modify_conf(api_id, dns_type, conf_data, ps, domains, force_domain)
|
|
return json_response(status=f, msg=msg)
|
|
except:
|
|
public.print_log(public.get_error_info())
|
|
|
|
@staticmethod
|
|
def remove_dns_api(get):
|
|
try:
|
|
dns_type = get.dns_type.strip()
|
|
api_id = get.api_id.strip()
|
|
except (json.JSONDecodeError, AttributeError, KeyError):
|
|
return json_response(status=False, msg="Parameter error")
|
|
f, msg = RealDnsMager().remove_conf(api_id, dns_type)
|
|
return json_response(status=f, msg=msg)
|
|
|
|
@staticmethod
|
|
def remove_domain(get):
|
|
try:
|
|
domain = get.domain.strip()
|
|
except (json.JSONDecodeError, AttributeError, KeyError):
|
|
return json_response(status=False, msg="Parameter error")
|
|
root, _ = extract_zone(domain)
|
|
m = RealDnsMager()
|
|
m.remove_domains_by_root(root)
|
|
m.save_config()
|
|
return json_response(status=True, msg="Successfully delete")
|
|
|
|
def query_dns(self, get):
|
|
domain = get.domain
|
|
dns_type = get.dns_type
|
|
res = public.query_dns(domain, dns_type)
|
|
if not res:
|
|
return json_response(status=False, msg="Parameter error")
|
|
|
|
return json_response(status=True, data=res)
|
|
|
|
@staticmethod
|
|
def get_record_list(get):
|
|
"""
|
|
获取解析记录列表
|
|
"""
|
|
try:
|
|
domain = get.domain.strip()
|
|
except (json.JSONDecodeError, AttributeError, KeyError):
|
|
return json_response(status=False, msg="Parameter error")
|
|
|
|
m = RealDnsMager()
|
|
try:
|
|
dns_onj = m.get_dns_obj_by_domain(domain)
|
|
return json_response(status=True, data=dns_onj.get_record_list_by_view(domain))
|
|
except Exception as e:
|
|
return json_response(status=False, msg=str(e))
|
|
|
|
@classmethod
|
|
def create_record(cls, get):
|
|
"""
|
|
@创建解析记录
|
|
"""
|
|
try:
|
|
domain = get.domain.strip()
|
|
host = get.host.strip()
|
|
r_type = get.r_type.strip()
|
|
value = get.value.strip()
|
|
ttl = int(get.ttl.strip())
|
|
except (json.JSONDecodeError, AttributeError, KeyError):
|
|
return json_response(status=False, msg="Parameter error")
|
|
domain, _ = extract_zone(domain)
|
|
|
|
try:
|
|
m = RealDnsMager()
|
|
dns_obj = m.get_dns_obj_by_domain(domain)
|
|
r_type = dns_obj.transform_record_type(r_type)
|
|
dns_obj.add_record(domain, host, r_type, value, ttl=ttl)
|
|
return json_response(status=True, msg="添加成功")
|
|
except:
|
|
# return public.returnMsg(False, public.get_error_info())
|
|
return json_response(status=False, msg=cls._DEFAULT_ERROR)
|
|
|
|
@classmethod
|
|
def delete_record(cls, get):
|
|
"""
|
|
@删除解析记录
|
|
@domain String 解析记录所在的域名
|
|
@recordId Int 解析记录 ID
|
|
"""
|
|
try:
|
|
domain = get.domain.strip()
|
|
host = get.host.strip()
|
|
r_type = get.r_type.strip()
|
|
except (json.JSONDecodeError, AttributeError, KeyError):
|
|
return json_response(status=False, msg="Parameter error")
|
|
domain, _ = extract_zone(domain)
|
|
|
|
try:
|
|
m = RealDnsMager()
|
|
dns_obj = m.get_dns_obj_by_domain(domain)
|
|
dns_obj.remove_record(domain, host, r_type)
|
|
return json_response(status=True, msg="Successfully delete")
|
|
except:
|
|
# return public.returnMsg(False, public.get_error_info())
|
|
return json_response(status=False, msg=cls._DEFAULT_ERROR)
|
|
|
|
@staticmethod
|
|
def get_domain_list(get=None):
|
|
"""
|
|
@name 获取域名列表
|
|
@param search 搜索关键字
|
|
"""
|
|
res = []
|
|
cache = DomainListCache()
|
|
cache_change = False
|
|
domain_change = False
|
|
m = RealDnsMager()
|
|
for key, value in m.config.items():
|
|
if key in m.NOT_USED_LIST:
|
|
continue
|
|
for dns_config in value:
|
|
config_id = dns_config["id"]
|
|
dns_obj: BaseDns = m.CLS_MAP[key].new(dns_config)
|
|
cloud_domain_data = cache.get(config_id)
|
|
if cloud_domain_data is None:
|
|
try:
|
|
cloud_domain_data = dns_obj.get_domain_list()
|
|
except Exception:
|
|
public.print_log(public.get_error_info())
|
|
cloud_domain_data = {}
|
|
cache_change = True
|
|
cache.set(config_id, cloud_domain_data)
|
|
|
|
if "domains" not in dns_config:
|
|
dns_config['domains'] = []
|
|
|
|
domains = dns_config.get("domains")
|
|
for d in domains:
|
|
tmp = {
|
|
"name": d,
|
|
"api_id": config_id,
|
|
"api_type": key,
|
|
"cloud_have": False,
|
|
"ps": dns_config["ps"]
|
|
}
|
|
if d in cloud_domain_data:
|
|
tmp["cloud_have"] = True
|
|
|
|
res.append(tmp)
|
|
|
|
for cloud_d in cloud_domain_data.keys():
|
|
if cloud_d not in domains:
|
|
m.remove_domains_by_root(cloud_d) # 删除其他地方的数据
|
|
dns_config["domains"].append(cloud_d)
|
|
domain_change = True
|
|
res.append({
|
|
"name": cloud_d,
|
|
"api_id": config_id,
|
|
"api_type": key,
|
|
"cloud_have": True,
|
|
"ps": dns_config["ps"]
|
|
})
|
|
|
|
if cache_change:
|
|
cache.save()
|
|
|
|
if domain_change:
|
|
m.save_config()
|
|
|
|
return json_response(status=True, data=res)
|