成为一个程序员

Python 实现阿里云 DDNS 动态域名解析

DDNS(Dynamic Domain Name Server)，它的作用是将用户的动态IP绑定到一个域名上去。

这样就算你的服务器IP发生了变化，用户仍然可以使用你的域名找到你的服务器。

阿里云提供了一套API，可以让你用编程的方式实现 DDNS，但是需要你的域名是在阿里云上申请的。

感谢我的室友借用给我测试用的域名。

# 一些可能用到的库

1
2
3

pip install aliyun-python-sdk-core
pip install aliyun-python-sdk-alidns
pip install pyyaml

# 获取和缓存 IP 地址

先写一个简单的工具类，可以获取当前电脑的 公网IP 地址，有很多提供这类服务的网站，本例程采用www.3322.org/dyndns/getip。

获取 IP 之后最好再把它缓存在一个文件中。

之所以需要缓存是因为阿里云更新两条一样的IP时会报错，我们可以提前缓存，然后下次调用更新服务之前借用缓存的内容，判断当前的 IP 是否无变化。

# 定义 IPManager 类

定义一个 IPManager 类，可以获取本机的 公网IP 地址，并使用文件进行缓存。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

from urllib.request import urlopen
import logging


class IPManager:
    def __init__(self, file_cache=".ipbuffer"):
        self.ip = ""
        self.file_cache = file_cache

    def get_current_ip(self):
        # 获取当前的 IP
        with urlopen('http://www.3322.org/dyndns/getip') as response:
            self.ip = str(response.read(), encoding='utf-8').replace("\n", "")
            logging.info("current ip: " + self.ip)
        return self.ip

    def sync_cache(self):
        # 同步当前的 IP 到缓存
        with open(self.file_cache, "w") as f:
            f.write(self.ip)
            logging.info("sync cache ip: " + self.ip)

    def get_cache(self):
        # 获取缓存的内容
        with open(self.file_cache, "r") as f:
            old_ip = f.read()
            logging.info("get cache ip: " + self.ip)
        return old_ip

程序默认使用 .ipbuffer 文件存储 IP，我觉得我们还需要先创建这个文件，不然运行的时候可能会报错。

可以使用下面的函数检查和创建一个文件，支持递归创建:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import os


def check_file(filename):
    # 获取父文件夹
    dirname = os.path.dirname(filename)
    if not os.path.exists(dirname) and dirname != "":
        # 递归创建父文件夹
        os.makedirs(dirname)
    # 创建文件
    with open(filename, "w") as f:
        f.write("")

# IPManager 的简单使用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def main():
    # 创建一个 IPManager
    ip_manager = IPManager()
    # 获取当前的 IP
    current_ip = ip_manager.get_current_ip()
    # 如果 IP 已经缓存就返回
    if ip_manager.get_cache() == current_ip:
        return
    # 更新 IP 缓存
    ip_manager.sync_cache()

这个程序可以 获取IP 并且在 IP无缓存 或者 IP更新 的时候更新缓存。

# 获取 accessKeyId 和 accessKeySecret

  1. 云账号登录RAM控制台。
  2. 在左侧导航栏的人员管理菜单下，单击用户。
  3. 在用户登录名称/显示名称列表下，单击目标RAM用户名称。
  4. 在用户AccessKey区域下，单击创建新的AccessKey。

摘抄自 阿里云文档

# 创建连接阿里云的客户端

1
2
3
4
5
6
7
8

from aliyunsdkcore.client import AcsClient

profile = {
    "accessKeyId": "xxx",
    "accessKeySecret": "xxx",
    "regionId": "cn-hangzhou"
}
client = AcsClient(profile["accessKeyId"], profile["accessKeySecret"], profile["regionId"])

把上一步的 accessKeyId、accessKeySecret 填进去。

在 regionId 填写你的区域号，关于 regionId 的说明，可以见 阿里云官方文档。

我们需要借助 client.do_action_with_exception 这个函数来发送请求到阿里云。

# 域名解析记录查询

之所以需要加一步域名解析记录查询是为了校验我们的域名是否已经被其他的 IP 绑定了。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from aliyunsdkalidns.request.v20150109.DescribeDomainRecordsRequest import DescribeDomainRecordsRequest
import json
import logging

def describe_domain_records(client, record_type, subdomain):
    logging.info("域名解析记录查询")
    request = DescribeDomainRecordsRequest()

    request.set_accept_format('json')
    request.set_Type(record_type)
    request.set_DomainName(subdomain)

    response = client.do_action_with_exception(request)
    response = str(response, encoding='utf-8')
    result = json.loads(response)
    logging.debug(result)
    return result

client 是上一步创建的客户端。

record_type 比较复杂，简单来说是 DNS域名解析 的解析类型。我们这里使用 A记录 就好了。

{% note info %}

常见的 DNS解析类型

A: 将主机名(或域名)指向一个 IPv4 地址

AAAA: 将主机名(或域名)指向一个 IPv6 地址

CNAME: 将域名指向另一个域名

{% endnote %}

subdomain 填你的域名就好了。

1
2

# 调用举例
describe_domain_records(client, "A", "tuenity.xyz")

# 添加域名解析记录

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from aliyunsdkalidns.request.v20150109.AddDomainRecordRequest import AddDomainRecordRequest
import logging
import json

def add_record(client, priority, ttl, record_type, value, rr, domain_name):
    logging.info("添加域名解析记录")
    request = AddDomainRecordRequest()

    request.set_accept_format('json')
    request.set_Priority(priority)
    request.set_TTL(ttl)
    request.set_Value(value)
    request.set_Type(record_type)
    request.set_RR(rr)
    request.set_DomainName(domain_name)

    response = client.do_action_with_exception(request)
    response = str(response, encoding='utf-8')
    result = json.loads(response)
    logging.debug(result)
    return result

priority 告诉域名解析服务，按照 priority 从小到大的顺序对记录搜索，搜索到匹配的记录后，就停止搜索 priority 值更大的记录，对于拥有相同 priority 的记录将通过 weight 再次选择 。

虽然阿里云并不提供 weight 的设置接口，但是你要知道它是个什么东西。

对于拥有相同 priority 的多条记录，weight 给出了选择某条记录的几率，值越大，被选中的概率就越大，合理的取值范围为 0-65535。

ttl( Time to live )，当用户在访问一个域名的时候，并不是每次都需要去解析一遍的，DNS服务器会在用户当地的递归DNS服务器上缓存一次，在 ttl 的时间长度内失效。一般设置 “600”。

record_type 同上一步。

value 就是你的 IP地址。

rr，阿里云的 rr 是主机头的意思，一般设置 “www”。

domain_name 就是你的域名。

# 更新域名解析记录

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from aliyunsdkalidns.request.v20150109.UpdateDomainRecordRequest import UpdateDomainRecordRequest
import logging
import json

def update_record(client, priority, ttl, record_type, value, rr, record_id):
    logging.info("更新域名解析记录")
    request = UpdateDomainRecordRequest()

    request.set_accept_format('json')
    request.set_Priority(priority)
    request.set_TTL(ttl)
    request.set_Value(value)
    request.set_Type(record_type)
    request.set_RR(rr)
    request.set_RecordId(record_id)

    response = client.do_action_with_exception(request)
    response = str(response, encoding='utf-8')
    logging.debug(response)
    return response

和上一步的函数接口几乎一摸一样，不过多解释了。

需要注意，不一样的是 record_id。这个需要 describe_domain_records 函数的返回值。

1

des_result = describe_domain_records(client, "A", "tuenity.xyz")

使用 des_result["TotalCount"] 就可以查处现在有多少条记录绑定在这个域名上了。

如果没有，我们就需要调用 add_record ，否则就调用 update_record。

record_id 可以通过 des_result["DomainRecords"]["Record"][0]["RecordId"] 获取。

# 改造、封装建议

  • 使用 yaml 来作为配置文件
  • 使用 python 自带的日志 logging
  • 把查询、更新、添加域名解析记录封装成一个类

# 获取完整的代码

Github 项目地址

© 2019 - 2024 nonlinearthink
使用 Hugo 构建
主题 Stack 由 Jimmy 设计