大概半年之前发现了家里的网络具有公网IP,于是就有了在家里部署一台服务器的想法。
最近由于学校还没有开学,只能居家“学习”,又正好之前自己淘汰了一个笔记本电脑,于是立马把笔记本重装成了 Ubuntu22,又在路由器上设置内网的固定IP,设置DMZ主机为我的笔记本电脑。这样,我在笔记本上部署的所有服务就都可以在公网访问了。
但是,由于家里的公网IP是动态的,不是固定的,每次光猫或者路由器断电,重新拨号就会导致IP变更,即使你能保证家里的路由器和光猫不重启,运营商也会每隔一段时间刷新一次的。所以为了能让域名动态的解析到家里的笔记本电脑。就需要DDNS服务了,即动态域名解析。
我的性格自然不会去买花钱的服务的😏,于是想到了利用 DNS 服务商提供的 API 去设置域名解析,在本地设置好定时任务,就可以实现动态域名解析了。
由于 Cloudflare 已经不再对 freenom 上的免费域名提供 API 支持了,所以我这里使用阿里云的 DNS API。
再设置一个企业微信的推送。就完美了✨

完整代码如下:

from aliyunsdkcore.client import AcsClient
from aliyunsdkalidns.request.v20150109.DescribeSubDomainRecordsRequest import DescribeSubDomainRecordsRequest
from aliyunsdkalidns.request.v20150109.DeleteSubDomainRecordsRequest import DeleteSubDomainRecordsRequest
from urllib.request import urlopen

import json,requests,socket,base64,psutil,os,time
from datetime import datetime


# 企业微信推送相关 具体可参考 https://github.com/easychen/wecomchan
qiye_id = ''
AgentId = ''
Secret = ''

def send_to_wecom(text,wecom_cid,wecom_aid,wecom_secret,wecom_touid='@all'):
    get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
    response = requests.get(get_token_url).content
    access_token = json.loads(response).get('access_token')
    if access_token and len(access_token) > 0:
        send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
        data = {
            "touser":wecom_touid,
            "agentid":wecom_aid,
            "msgtype":"text",
            "text":{
                "content":text
            },
            "duplicate_check_interval":600
        }
        response = requests.post(send_msg_url,data=json.dumps(data)).content
        return response
    else:
        return False

def send_to_wecom_image(base64_content,wecom_cid,wecom_aid,wecom_secret,wecom_touid='@all'):
    get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
    response = requests.get(get_token_url).content
    access_token = json.loads(response).get('access_token')
    if access_token and len(access_token) > 0:
        upload_url = f'https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={access_token}&type=image'
        upload_response = requests.post(upload_url, files={
            "picture": base64.b64decode(base64_content)
        }).json()
        if "media_id" in upload_response:
            media_id = upload_response['media_id']
        else:
            return False
        send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
        data = {
            "touser":wecom_touid,
            "agentid":wecom_aid,
            "msgtype":"image",
            "image":{
                "media_id": media_id
            },
            "duplicate_check_interval":600
        }
        response = requests.post(send_msg_url,data=json.dumps(data)).content
        return response
    else:
        return False

def send_to_wecom_markdown(text,wecom_cid,wecom_aid,wecom_secret,wecom_touid='@all'):
    get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
    response = requests.get(get_token_url).content
    access_token = json.loads(response).get('access_token')
    if access_token and len(access_token) > 0:
        send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
        data = {
            "touser":wecom_touid,
            "agentid":wecom_aid,
            "msgtype":"markdown",
            "markdown":{
                "content":text
            },
            "duplicate_check_interval":600
        }
        response = requests.post(send_msg_url,data=json.dumps(data)).content
        return response
    else:
        return False


def judgeprocess(processname):
    pl = psutil.pids()
    for pid in pl:
        if psutil.Process(pid).name() == processname:
            print(processname + "正在运行,pid=" + str(pid))
            return True
    else:
        return False

def getip():
    """
    查询本机ip地址
    :return: ip
    """
    res = requests.get("https://ifconfig.me/ip")
    return res.text


class Dnscontroller:
    access_key_id = ""
    #阿里云的新建账户id
    access_key_secret = ""
    region = "cn-hangzhou"
    #时区
    record_type = "A"

    domain = "oh-my.ga"
    #上面是你的一级域名
    name_ipv4 = ["@","*"]
    # 我解析的根域名 和 一个泛解析域名

    def __init__(self):
        self.client = AcsClient(
            self.access_key_id,
            self.access_key_secret,
            self.region
        )

    # 添加新的域名解析记录
    def add(self, DomainName, RR, Type, Value):
        from aliyunsdkalidns.request.v20150109.AddDomainRecordRequest import AddDomainRecordRequest
        request = AddDomainRecordRequest()
        request.set_accept_format('json')
        request.set_DomainName(DomainName)
        request.set_RR(RR)
        request.set_Type(Type)
        request.set_Value(Value)
        response = self.client.do_action_with_exception(request)

    # 修改域名解析记录
    def update(self, RecordId, RR, Type, Value):
        from aliyunsdkalidns.request.v20150109.UpdateDomainRecordRequest import UpdateDomainRecordRequest
        request = UpdateDomainRecordRequest()
        request.set_accept_format('json')
        request.set_RecordId(RecordId)
        request.set_RR(RR)
        request.set_Type(Type)
        request.set_Value(Value)
        response = self.client.do_action_with_exception(request)

    # 获取公网IP并进行修改
    def Get_IPv4(self):
        # 获取 IP
        ipv4 = getip()
        print("获取到IPv4地址:%s" % ipv4)

        request = DescribeSubDomainRecordsRequest()
        request.set_accept_format('json')
        request.set_DomainName(self.domain)

        for item in self.name_ipv4:
            request.set_SubDomain(item + '.' + self.domain)
            response = self.client.do_action_with_exception(request)
            domain_list = json.loads(response)

            if domain_list['TotalCount'] == 0:
                self.add(self.domain, item, self.record_type, ipv4)
                print("新建域名:" + item + ",解析成功")

            elif domain_list['TotalCount'] == 1:
                if domain_list['DomainRecords']['Record'][0]['Value'].strip() != ipv4.strip():
                    self.update(domain_list['DomainRecords']['Record'][0]['RecordId'], item, self.record_type, ipv4)
                    ret = send_to_wecom("宏碁笔记本新IP: " + ipv4, qiye_id, AgentId, Secret)
                    print(ret)
                    print("修改域名:" + item + ",解析成功")
                else:
                    print("IPv4地址没变")

            elif domain_list['TotalCount'] > 1:
                # 删除所有解析过的记录,并重新添加新的解析记录
                request = DeleteSubDomainRecordsRequest()
                request.set_accept_format('json')
                request.set_DomainName(self.domain)
                request.set_RR(item)
                response = self.client.do_action_with_exception(request)
                self.add(self.domain, item, self.record_type, ipv4)
                ret = send_to_wecom("宏碁笔记本新IP: " + ipv4, qiye_id, AgentId, Secret)
                print(ret)
                print("修改域名解析成功")

if __name__ == "__main__":
    print(datetime.now())
    # 检测服务器上的代理服务是否开启,如果开启应该先将它关闭,否则会影响到自身 IP 的获取
    # 我这里为了翻墙装了个 v2raya 
    if judgeprocess('v2raya'):
        os.system('systemctl stop v2raya')
        print('关闭v2raya ---')
        time.sleep(3)
        Dnscontroller().Get_IPv4()
        os.system('systemctl start v2raya')
        print('启动v2raya ---')
    else:
        Dnscontroller().Get_IPv4()
    print()