from sslModel.base import sslBase import public import hmac import hashlib import json import time import requests from typing import Dict, Any, Optional, Tuple class main(sslBase): def __init__(self): super().__init__() def __init_data(self, data): self.account_id = data["AccountID"] self.access_key = data["AccessKey"] self.secret_key = data["SecretKey"] self.base_url = "https://dmp.bt.cn" def _generate_signature(self, method: str, path: str, body: str) -> Tuple[str, str]: """生成 API 签名""" timestamp = str(int(time.time())) signing_string = f"{self.account_id}\n{timestamp}\n{method.upper()}\n{path}\n{body}" signature = hmac.new( self.secret_key.encode('utf-8'), signing_string.encode('utf-8'), hashlib.sha256 ).hexdigest() return timestamp, signature def _request(self, method: str, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """发起 API 请求""" url = self.base_url + path body_str = "" body_bytes = None if data is not None: body_str = json.dumps(data, separators=(',', ':')) body_bytes = body_str.encode('utf-8') timestamp, signature = self._generate_signature(method, path, body_str) headers = { "Content-Type": "application/json", "X-Account-ID": self.account_id, "X-Access-Key": self.access_key, "X-Timestamp": timestamp, "X-Signature": signature } response = requests.request( method=method.upper(), url=url, data=body_bytes, headers=headers ) result = response.json() if not result.get("status"): raise Exception(f"API 请求失败: {result.get('msg')}") return result def create_dns_record(self, get): domain_name = get.domain_name domain_dns_value = get.domain_dns_value record_type = 'TXT' if 'record_type' in get: record_type = get.record_type mx = 0 if record_type == 'MX': if not get.get('mx'): return public.returnMsg(False, 'MX记录类型必须填写MX值') mx = int(get.mx) domain_name, sub_domain, _ = self.extract_zone(domain_name) if not sub_domain: sub_domain = '@' domain_data = self.get_domain_list(get) if not domain_data['status']: return domain_data domain_list = domain_data['data'] domain_id = None for d in domain_list: if d['name'] == domain_name: domain_id = d['id'] break if not domain_id: return public.returnMsg(False, '域名不存在，请先添加域名') try: body = { "domain_type": 1, "domain_id": domain_id, "record": sub_domain, "type": record_type, "value": domain_dns_value, "TTL": int(get.ttl) if 'ttl' in get else 600, "MX": mx, "line": "默认", "remark": get.remark if 'remark' in get else "", "view_id": 0, } res = self._request("POST", "/api/v1/dns/record/create", body) except Exception as e: public.print_log(e) return public.returnMsg(False, "添加解析记录失败: {}".format(e)) if not res.get("status"): return public.returnMsg(False, "添加解析记录失败: {}".format(res.get("msg"))) return public.returnMsg(True, "添加解析记录成功") def delete_dns_record(self, get): domain_name = get.domain_name domain_name, sub_domain, _ = self.extract_zone(domain_name) domain_data = self.get_domain_list(get) if not domain_data['status']: return domain_data domain_list = domain_data['data'] domain_id = None for d in domain_list: if d['name'] == domain_name: domain_id = d['id'] break if not domain_id: return public.returnMsg(False, '域名不存在，请先添加域名') record_id = get.RecordId try: body = { "record_id": record_id, "domain_type":1, "domain_id":domain_id } res = self._request("POST", "/api/v1/dns/record/delete", body) except Exception as e: public.print_log(e) return public.returnMsg(False, "删除解析记录失败: {}".format(e)) if not res.get("status"): return public.returnMsg(False, "删除解析记录失败: {}".format(res.get("msg"))) return public.returnMsg(True, "删除解析记录成功") def update_dns_record(self, get): domain_name = get.domain_name domain_dns_value = get.domain_dns_value record_type = 'TXT' if 'record_type' in get: record_type = get.record_type mx = 0 if record_type == 'MX': if not get.get('mx'): return public.returnMsg(False, 'MX记录类型必须填写MX值') mx = int(get.mx) domain_name, sub_domain, _ = self.extract_zone(domain_name) if not sub_domain: sub_domain = '@' domain_data = self.get_domain_list(get) if not domain_data['status']: return domain_data domain_list = domain_data['data'] domain_id = None for d in domain_list: if d['name'] == domain_name: domain_id = d['id'] break if not domain_id: return public.returnMsg(False, '域名不存在，请先添加域名') record_id = get.RecordId try: body = { "record_id": record_id, "domain_type": 1, "domain_id": domain_id, "record": sub_domain, "type": record_type, "value": domain_dns_value, "TTL": int(get.ttl) if 'ttl' in get else 600, "MX": mx, "line": "默认", "remark": get.remark if 'remark' in get else "", "view_id": 0, } res = self._request("POST", "/api/v1/dns/record/update", body) except Exception as e: public.print_log(e) return public.returnMsg(False, "修改解析记录失败: {}".format(e)) if not res.get("status"): return public.returnMsg(False, "修改解析记录失败: {}".format(res.get("msg"))) return public.returnMsg(True, "修改解析记录成功") def set_dns_record_status(self, get): domain_name = get.domain_name domain_name, sub_domain, _ = self.extract_zone(domain_name) domain_data = self.get_domain_list(get) if not domain_data['status']: return domain_data domain_list = domain_data['data'] domain_id = None for d in domain_list: if d['name'] == domain_name: domain_id = d['id'] break if not domain_id: return public.returnMsg(False, '域名不存在，请先添加域名') record_id = get.RecordId try: body = { "record_id": record_id, "domain_type":1, "domain_id":domain_id, } res = self._request("POST", "/api/v1/dns/record/{}".format("start" if get.status == '0' else "pause"), body) except Exception as e: public.print_log(e) return public.returnMsg(False, "设置解析记录状态失败: {}".format(e)) if not res.get("status"): return public.returnMsg(False, "设置解析记录状态失败: {}".format(res.get("msg"))) return public.returnMsg(True, "设置解析记录状态成功") def get_dns_record(self, get): domain_name = get.domain_name domain_data = self.get_domain_list(get) if not domain_data['status']: return domain_data domain_list = domain_data['data'] domain_id = None for d in domain_list: if d['name'] == domain_name: domain_id = d['id'] break if not domain_id: return public.returnMsg(False, '域名不存在，请先添加域名') try: body = { "domain_id": domain_id, "searchKey": "record", "domain_type": 1, "p": get.p, "row": get.limit, } if "search" in get and get.search: body["searchValue"] = get.search res = self._request("POST", "/api/v1/dns/record/list", body) except Exception as e: public.print_log(e) return public.returnMsg(False, "获取解析记录失败: {}".format(e)) data = {"info": {"record_total": res.get("data", {}).get("count", 0)}, "list": []} for i in res.get("data", {}).get("data", []): data["list"].append({ "RecordId": i["record_id"], "name": i["record"]+"."+domain_name if i["record"] != "@" else domain_name, "type": i["type"], "value": i["value"], "ttl": i["TTL"], "mx": i.get("MX", 0), "status": "暂停" if i["state"] == 2 else "启用", "line": i.get("line", "默认"), "remark": i.get("remark", ""), }) return data def get_domain_list(self, get): self.__init_data(self.get_dns_data(None)[get.dns_id]) try: res = self._request("POST", "/api/v1/domain/manage/list", {"p":1,"rows":10, "status":1}) except Exception as e: public.print_log(e) return public.returnMsg(False, "获取域名列表失败: {}".format(e)) data = [] local_domain_list = [d['domain'] for d in public.M('ssl_domains').field('domain').select()] for i in res.get("data", {}).get("data", []): data.append({ "id": i["id"], "name": i["full_domain"], "sync": 0 if i["full_domain"] in local_domain_list else 1, }) return {"status": True, "msg": "获取成功", "data": data}