一、背景
最近有一个需求，公司内网的IP地址会发生变化，导致阿里云域名不能解析到新的IP地址，此时我们需要对阿里云的域名进行更新
二、实现
2.1 获取本地出口的公网IP
2.1.1 通过命令或网页 - 获取本地出口的公网IP**
获取本地IP的方式有很多，可以通过访问一些网站来获取本地的公网IP地址
1. curl ifconfig.me
2. curl -L ip.tool.lu
3. 以及一些在线查询网页然后解析内容来获取我们的公网IP地址
2.1.2 通过Nginx 获取我们的公网IP
我们通过在云端创建一个主机地址，然后通过nginx来获取访问者的公网IP，这样的好处是我们自己搭建的服务比较稳定可靠
-
1. 创建阿里云主机记录
在域名服务中通过访问域名列表点击解析
-
2. 新增获取访问者IP的主机记录
设置主机地址，记录值这里填写自己阿里云的IP地址，这是为了访问这个域名地址时，nginx返回访问者的IP地址。
-
3. 新增测试的主机记录
我们可以自己在新建一个记录，主机记录随意填写一个如dev，IP地址也可以随意填写一个。 -
4. Nginx 返回对应的域名时返回访问者的IP
新建一个nginx的配置文件，指定域名返回的内容root@rion:/etc/nginx/conf.d# pwd /etc/nginx/conf.d root@rion:/etc/nginx/conf.d# ls remote_ip.conf root@rion:/etc/nginx/conf.d# cat remote_ip.conf real_ip_recursive on; server { listen 80; # 指定访问端口，如果不想用80端口，需要在阿里云开启对应的端口 server_name ip.xxx.xx; # 省略其他配置.... # nginx直接返回客户端IP到body location /ip { default_type text/plain; return 200 "$remote_addr"; } } root@rion:/etc/nginx/conf.d# nginx -s reload访问对应的域名查看结果是否正确，返回的结果时正确的。
2.2 实现动态更新阿里云IP地址
2.2.1 获取AccessKey ID和AccessKey Secret
我们使用SDK时，需要使用到AccessKey ID和AccessKey Secret 。
参考链接： https://help.aliyun.com/document_detail/38738.html
2.2.2 代码实现
-
下载python包
aliyun-python-sdk-core-v3 aliyun-python-sdk-alidns -
代码
import requests
import traceback
import time
from datetime import datetime
import json
import sys
from aliyunsdkcore.client import AcsClient
from aliyunsdkalidns.request.v20150109.DescribeDomainRecordsRequest import DescribeDomainRecordsRequest
from aliyunsdkalidns.request.v20150109.UpdateDomainRecordRequest import UpdateDomainRecordRequest
class DNSRecord:
json_data = None # json 配置文件内容
client = None # 阿里云client
public_dic = {"time":None,"ip":None}
dns_record_dic = {"time":None,"ip":None,"record_id":None}
@classmethod
def init(cls):
"""初始化阿里云 client"""
try:
cls.client = AcsClient(cls.json_data['id'], cls.json_data['secret'], cls.json_data['region'])
print("Aliyun key 认证成功")
return True
except Exception as e:
print("Aliyun key 认证失败")
return False
@classmethod
def reload_config(cls):
"""加载配置文件"""
with open('./config.json',mode='r',encoding='utf-8') as fp:
cls.json_data = json.load(fp)
@classmethod
def get_dns_record(cls):
"""获取dns记录"""
try:
request = DescribeDomainRecordsRequest()
request.set_accept_format('json')
request.set_DomainName(cls.json_data['domain_name'])
response = cls.client.do_action_with_exception(request)
data = json.loads(str(response, encoding='utf-8'))
for record in data['DomainRecords']['Record']:
if cls.json_data['RR'] == record['RR']:
cls.dns_record_dic['time'] = datetime.now()
cls.dns_record_dic['ip'] = record['Value']
cls.dns_record_dic['record_id'] = record['RecordId']
return True
except:
traceback.print_exc()
return False
@classmethod
def get_public_ip(cls):
"""获取公网IP"""
for url in cls.json_data['public_url']:
try:
request_obj = requests.get(url,timeout=5)
if request_obj.status_code == 200:
cls.public_dic['time'] = datetime.now()
cls.public_dic['ip'] = request_obj.text
return True
except:
traceback.print_exc()
return False
@classmethod
def update_dns_record(cls,public_ip):
update_time = datetime.now()
update_time_str = update_time.strftime("%Y-%m-%d %H:%M:%S")
try:
request = UpdateDomainRecordRequest()
request.set_accept_format('json')
request.set_Value(public_ip)
request.set_Type(cls.json_data['type'])
request.set_RR(cls.json_data['RR'])
request.set_RecordId(cls.dns_record_dic['record_id'])
response = cls.client.do_action_with_exception(request)
cls.public_dic['time'] = update_time
cls.dns_record_dic['time'] = update_time
print(f"[更新成功 {update_time_str}] 域名:{cls.json_data['domain_name']} + \
主机:{cls.json_data['RR']} 记录类型:{cls.json_data['type']} + \
IP地址:{cls.dns_record_dic['ip']}")
return True
except Exception as e:
print(f"域名:{cls.json_data['domain_name']} 主机:{cls.json_data['RR']} 解析失败,错误:",e)
return False
@classmethod
def main(cls):
cls.reload_config()
cls.init()
while True:
public_time = cls.public_dic['time']
dns_time = cls.dns_record_dic['time']
if not dns_time:
ret = cls.get_dns_record() # 更新云端的IP、时间、record_id
if not ret:
time.sleep(10)
continue
dns_time = cls.dns_record_dic['time']
if not public_time:
ret = cls.get_public_ip() # 更新本地公网的IP、时间
if not ret:
time.sleep(10)
continue
public_time = cls.public_dic['time']
if (public_time - dns_time).total_seconds() < 300: # 本地时间和云端时间小于5分钟
print("本地更新时间与云端更新时间差小于5分钟")
time.sleep(10)
continue
else:
public_ret = cls.get_public_ip() # 更新本地公网的IP、时间
if not public_ret:
time.sleep(10)
continue
dns_ret = cls.get_dns_record() # 获取阿里云dns记录
if not dns_ret:
time.sleep(10)
continue
public_ip = cls.public_dic['ip']
dns_ip = cls.dns_record_dic['ip']
if public_ip != dns_ip: # 本地公网IP和云端IP不相等
ret = cls.update_dns_record(public_ip)
if not ret:
time.sleep(10)
else:
time.sleep(10)
if __name__ == "__main__":
dns_r = DNSRecord
dns_r.main()
- Json配置文件
{
"id":"xxxxxxxxxxxxxxxxxx",
"secret":"xxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"region":"cn-hangzhou",
"RR":"dev",
"type":"A",
"domain_name":"xxxxx.xx",
"public_url":["http://ip.xx.xx/ip","http://xx.xxx.xx:8899/ip"]
}
这里获取本地公网IP的对象为列表，若一个获取地址的URL挂掉了，我们还可以从另一个地址获取本地公网的IP，保证了一定的稳定性。
参考链接： https://help.aliyun.com/document_detail/124923.html文章来源：https://www.toymoban.com/news/detail-455348.html
以上是python实现动态更新阿里云DNS解析记录的代码，除了更新记录之外还可以增加记录，删除记录等。文章来源地址https://www.toymoban.com/news/detail-455348.html
到了这里，关于Python 实现动态解析阿里云DNS记录的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
