251 lines
7.9 KiB
Python
251 lines
7.9 KiB
Python
"""
|
||
阿里云 DNS 解析管理服务
|
||
通过阿里云 OpenAPI 管理 zgitm.com 的 DNS 解析记录
|
||
使用 HMAC-SHA1 签名，兼容 Python 3.6+
|
||
"""
|
||
import os
|
||
import hmac
|
||
import hashlib
|
||
import base64
|
||
import logging
|
||
import urllib.request
|
||
import urllib.parse
|
||
import urllib.error
|
||
import json
|
||
import time
|
||
import uuid
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 阿里云 API 配置
|
||
ALIDNS_ENDPOINT = 'https://alidns.aliyuncs.com'
|
||
API_VERSION = '2015-01-09'
|
||
SIGNATURE_METHOD = 'HMAC-SHA1'
|
||
SIGNATURE_VERSION = '1.0'
|
||
|
||
|
||
class DnsService:
|
||
"""阿里云 DNS 解析管理"""
|
||
|
||
def __init__(self):
|
||
self.access_key_id = os.environ.get('Ali_Key', '')
|
||
self.access_key_secret = os.environ.get('Ali_Secret', '')
|
||
|
||
def _sign(self, method, params):
|
||
"""
|
||
HMAC-SHA1 签名（阿里云 API 签名 v1.0）
|
||
"""
|
||
# 1. 按参数名排序
|
||
sorted_keys = sorted(params.keys())
|
||
sorted_params = [(k, params[k]) for k in sorted_keys]
|
||
|
||
# 2. 构造规范化查询字符串
|
||
canonicalized = urllib.parse.urlencode(sorted_params)
|
||
|
||
# 3. 构造签名源字符串
|
||
string_to_sign = method + '&' + urllib.parse.quote('/', safe='') + '&' + urllib.parse.quote(canonicalized, safe='')
|
||
|
||
# 4. HMAC-SHA1 签名
|
||
key = (self.access_key_secret + '&').encode('utf-8')
|
||
h = hmac.new(key, string_to_sign.encode('utf-8'), hashlib.sha1)
|
||
signature = base64.b64encode(h.digest()).decode('utf-8')
|
||
|
||
return signature
|
||
|
||
def _call(self, action, extra_params=None):
|
||
"""
|
||
调用阿里云 OpenAPI
|
||
|
||
Args:
|
||
action: API 名称 (如 DescribeDomainRecords)
|
||
extra_params: 额外参数
|
||
|
||
Returns:
|
||
dict: API 响应
|
||
"""
|
||
if not self.access_key_id or not self.access_key_secret:
|
||
raise RuntimeError('阿里云 AccessKey 未配置')
|
||
|
||
# 公共参数
|
||
params = {
|
||
'AccessKeyId': self.access_key_id,
|
||
'Action': action,
|
||
'Format': 'JSON',
|
||
'Version': API_VERSION,
|
||
'SignatureMethod': SIGNATURE_METHOD,
|
||
'SignatureVersion': SIGNATURE_VERSION,
|
||
'SignatureNonce': str(uuid.uuid4()),
|
||
'Timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
|
||
}
|
||
if extra_params:
|
||
params.update(extra_params)
|
||
|
||
# 签名
|
||
params['Signature'] = self._sign('GET', params)
|
||
|
||
# 构造 URL
|
||
url = ALIDNS_ENDPOINT + '/?' + urllib.parse.urlencode(params)
|
||
|
||
# 发送请求
|
||
try:
|
||
req = urllib.request.Request(url, method='GET')
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
result = json.loads(resp.read().decode('utf-8'))
|
||
return result
|
||
except urllib.error.HTTPError as e:
|
||
body = e.read().decode('utf-8') if e.fp else ''
|
||
logger.error(f'DNS API HTTP {e.code}: {body}')
|
||
raise RuntimeError(f'DNS API 请求失败: {e.code} {body}')
|
||
except Exception as e:
|
||
logger.error(f'DNS API 异常: {e}')
|
||
raise RuntimeError(f'DNS API 异常: {e}')
|
||
|
||
def list_records(self, domain_zone, rr_keyword=None, record_type=None):
|
||
"""
|
||
获取域名解析记录列表（自动分页，拉取全部记录）
|
||
|
||
Args:
|
||
domain_zone: 主域名 (如 zgitm.com)
|
||
rr_keyword: 主机记录筛选 (如 www)
|
||
record_type: 记录类型筛选 (如 A, CNAME)
|
||
|
||
Returns:
|
||
list[dict]: 解析记录列表
|
||
"""
|
||
params = {
|
||
'DomainName': domain_zone,
|
||
'PageSize': '100', # 每页最大100条
|
||
}
|
||
if rr_keyword:
|
||
params['RRKeyWord'] = rr_keyword
|
||
if record_type:
|
||
params['Type'] = record_type
|
||
|
||
all_records = []
|
||
page_number = 1
|
||
total_count = None
|
||
|
||
while True:
|
||
params['PageNumber'] = str(page_number)
|
||
resp = self._call('DescribeDomainRecords', params)
|
||
|
||
domain_records = resp.get('DomainRecords', {})
|
||
records = domain_records.get('Record', [])
|
||
|
||
if total_count is None:
|
||
total_count = domain_records.get('TotalCount', 0)
|
||
logger.info(f'DNS 记录总数: {total_count}, 开始分页拉取...')
|
||
|
||
# 统一转列表（单条记录时阿里云返回 dict）
|
||
if isinstance(records, dict):
|
||
records = [records]
|
||
|
||
for r in records:
|
||
all_records.append({
|
||
'record_id': r.get('RecordId', ''),
|
||
'rr': r.get('RR', ''),
|
||
'type': r.get('Type', 'A'),
|
||
'value': r.get('Value', ''),
|
||
'ttl': r.get('TTL', 600),
|
||
'line': r.get('Line', 'default'),
|
||
'status': r.get('Status', 'ENABLE'),
|
||
'full_domain': (r.get('RR', '') + '.' + domain_zone).strip('.'),
|
||
})
|
||
|
||
# 判断是否还有下一页
|
||
if len(all_records) >= total_count or len(records) < 100:
|
||
break
|
||
page_number += 1
|
||
|
||
logger.info(f'DNS 记录拉取完成: zone={domain_zone}, 共 {len(all_records)} 条')
|
||
return all_records
|
||
|
||
def add_record(self, domain_zone, rr, record_type, value, ttl=600):
|
||
"""
|
||
添加解析记录
|
||
|
||
Args:
|
||
domain_zone: 主域名
|
||
rr: 主机记录 (如 webssd, @, www)
|
||
record_type: 记录类型 (A, CNAME, TXT 等)
|
||
value: 记录值 (IP地址或域名)
|
||
ttl: TTL 秒数
|
||
|
||
Returns:
|
||
dict: {'record_id': ..., 'full_domain': ...}
|
||
"""
|
||
params = {
|
||
'DomainName': domain_zone,
|
||
'RR': rr,
|
||
'Type': record_type.upper(),
|
||
'Value': value,
|
||
'TTL': str(ttl),
|
||
}
|
||
resp = self._call('AddDomainRecord', params)
|
||
record_id = resp.get('RecordId', '')
|
||
full_domain = (rr + '.' + domain_zone).strip('.').replace('@.', '')
|
||
logger.info(f'DNS 记录已添加: {full_domain} {record_type} {value} (ID: {record_id})')
|
||
return {'record_id': record_id, 'full_domain': full_domain}
|
||
|
||
def delete_record(self, record_id):
|
||
"""
|
||
删除解析记录
|
||
|
||
Args:
|
||
record_id: 记录ID
|
||
|
||
Returns:
|
||
bool: 是否成功
|
||
"""
|
||
resp = self._call('DeleteDomainRecord', {'RecordId': str(record_id)})
|
||
logger.info(f'DNS 记录已删除: {record_id}')
|
||
return True
|
||
|
||
def update_record(self, record_id, rr, record_type, value, ttl=600):
|
||
"""
|
||
修改解析记录
|
||
|
||
Args:
|
||
record_id: 记录ID
|
||
rr: 主机记录
|
||
record_type: 记录类型
|
||
value: 记录值
|
||
ttl: TTL
|
||
|
||
Returns:
|
||
bool: 是否成功
|
||
"""
|
||
params = {
|
||
'RecordId': str(record_id),
|
||
'RR': rr,
|
||
'Type': record_type.upper(),
|
||
'Value': value,
|
||
'TTL': str(ttl),
|
||
}
|
||
try:
|
||
self._call('UpdateDomainRecord', params)
|
||
logger.info(f'DNS 记录已修改: {record_id} → {rr} {record_type} {value}')
|
||
except RuntimeError as e:
|
||
msg = str(e)
|
||
# 未做任何改动直接保存，阿里云返回 DomainRecordDuplicate，视为成功
|
||
if 'DomainRecordDuplicate' in msg:
|
||
logger.info(f'DNS 记录未变更，跳过: {record_id} {rr} {record_type} {value}')
|
||
else:
|
||
raise
|
||
return True
|
||
|
||
def set_record_status(self, record_id, status):
|
||
"""
|
||
启用/停用解析记录
|
||
|
||
Args:
|
||
record_id: 记录ID
|
||
status: Enable / Disable
|
||
"""
|
||
self._call('SetDomainRecordStatus', {
|
||
'RecordId': str(record_id),
|
||
'Status': status.upper(),
|
||
})
|
||
logger.info(f'DNS 记录状态已更新: {record_id} → {status}')
|
||
return True
|