初衷:
在家里搭建了一個Centos服務器,和一個Nas服務器。為了遠程訪問家里的服務器,下載文件,特別向電信申請了公網ip
問題:
電信為了限制個人用於部署服務,盡管會分配公網ip,但是ip經常會改變。
於是又去阿里雲去購買了一個域名,想將本地公網ip動態解析到域名,盡管會有域名生效的間隔期,但是也能基本滿足日常的需要了。
軟路由市場中有直接支持阿里ddns設置的插件,感覺別人寫的,總是不太放心。
於是自己造輪子:查閱資料,通過python自己來實現。
整體思路:
獲取自己的公網ip,域名解析ip是否一致
如果一致,等待一段時間,循環重復
如果不一致,調用阿里雲接口設置,並等待生效。
關於公網ip的獲取方式,采用了好多種嘗試:
方案一:通過訪問一些網頁,從中抓取自己的公網ip(網上教程大多數都是這種)
弊端,網頁服務不穩定,有可能掛掉,有可能
有可能返回錯誤ip,特別是使用軟路由的情況下。
為了解決這些問題,采用多個網站獲取ip,並取交集的方式來減小錯誤幾率。
網上有很多代碼,我的代碼已經不在了,需要的自行百度即可找到。
方案二:用軟路由撥號,通過ssh客戶端返回wan口ip:
需要使用到python的 paramiko庫,直接上代碼:
ssh_paras = { "hostname": '192.168.2.1', "port": 9999, "username": 'root', "password": '***********' } import paramiko def get_ssh_client(): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(**ssh_paras) return ssh def get_current_wan_ip(): cmd = 'ifconfig pppoe-WAN' ssh_client = get_ssh_client() _, stdout, _ = ssh_client.exec_command(cmd) result = stdout.read() ssh_client.close() if result: result = result.decode("UTF-8") if find := re.search(r"inet addr:(\d+\.\d+\.\d+\.\d+)", result): ip = find.group(1) return ip
方案三:用光貓撥號,通過requests庫,模擬登錄,通過http返回結果解析;
通過web登錄,點擊操作獲取wan口公網ip, 通過Wireshark抓包,分析http登錄流程機制。
注意不同的路由器,光貓可能登錄流程不一致,但思路是雷同的。(我的光貓的是華為的MA5671)
COOKIE = None def get_wan_ip(): global COOKIE while True: if COOKIE is None: COOKIE = get_cookie() t = time.time() res3 = requests.get("http://192.168.1.1/html/bbsp/common/wan_list.asp", headers={ "DNT": "1", "Accept": "*/*", "Cookie": "Cookie=" + COOKIE, }) text = res3.text or "" wan_ip_find = re.search("INTERNET_R_VID.*?\"(\d+\.\d+\.\d+\.\d+)\"", text) if wan_ip_find: print(f"used time {time.time() - t}") return wan_ip_find.group(1) else: time.sleep(1) print(time.asctime(), " cookie is outof time!") def get_cookie(): res1 = requests.post("http://192.168.1.1/asp/GetRandCount.asp") rand = res1.text[3:] data = {"x.X_HW_Token": rand} headers = { "Cookie": "Cookie=UserName:<****>:PassWord:<****>:Language:chinese:id=-1", "Content-Type": "application/x-www-form-urlencoded" }
# <****> 代表賬號和密碼,需要注意密碼是轉碼之后的形式,通過抓包工具可以看到自己的。 res2 = requests.post("http://192.168.1.1/login.cgi", data, headers=headers) cookie = res2.cookies.get("Cookie") return cookie
方案四:通過telnet登錄光貓,通過telnet交互獲取wan口公網ip:
方案三中通過requests庫模擬web登錄,會造成登錄的用戶過多,
然后就無法通過瀏覽器進入光貓了(我的模擬代碼太簡陋,沒有考慮推出登錄的代碼實現)
並且返回wan口公網ip的同時,會返回很多無用信息,每次執行比較耗時,浪費光貓資源。
所以想通過ssh登錄光貓,去獲取密碼,發現這款光貓里面卻沒有內置ssh服務。
發現有telnet服務,於是通過telnet實現公網ip的獲取:
注意:設備支持哪些telnet命令,可以通過“?”命令查詢,自己在根據命令猜測其作用。
實現代碼:
需要注意不同的設備,telnet的截取關鍵字兒可能不同,需要自己根據設備修改。
# coding=UTF-8 import re import time import telnetlib def get_wan_ip(): while True: try: tel = telnetlib.Telnet("192.168.1.1", port=23) tel.read_until(b"Login:") tel.write(b"root\n") tel.read_until(b"Password:") tel.write(b"youpassword\n") tel.read_until(b"WAP>") except: time.sleep(1) continue else: while True: try: tel.write(b"ip route\n") data = tel.read_until(b"WAP>").decode() if ip_find:= re.search("ppp.*?src (\d+\.\d+\.\d+\.\d+)", data): yield ip_find.group(1) except: time.sleep(1) break for i in get_wan_ip(): print(i) time.sleep(1)
最后:附上通過aliyun python ddns接口實現域名動態解析的代碼。
requirements.txt
aliyun-python-sdk-core==2.13.35
aliyun-python-sdk-alidns==2.6.32
aliyun-python-sdk-domain==3.14.6
dnspython==2.1.0
paramiko==2.8.0
# coding=UTF-8 import re import json import time import telnetlib from aliyunsdkcore.client import AcsClient from aliyunsdkalidns.request.v20150109.DescribeDomainRecordsRequest import DescribeDomainRecordsRequest from aliyunsdkalidns.request.v20150109.UpdateDomainRecordRequest import UpdateDomainRecordRequest import dns.resolver aliyun_id = "wkelwXModwxPOIBD假的YTEEFDDD0DDDD" aliyun_secret = "ADDFDSDEEEFD錯的dCCCCxclaldoerC" aliyun_account = "aliyun賬號名"
# 以上信息需要登錄阿里雲官網去申請,是通過這個就能訪問阿里雲里你的賬戶,設置你的購買的域名解析。
domain_lst = ["baidu.com", "你購買的域名", ] def get_wan_ip(): while True: try: tel = telnetlib.Telnet("192.168.1.1", port=23) tel.read_until(b"Login:") tel.write(b"root\n") tel.read_until(b"Password:") tel.write(b"000000000\n") tel.read_until(b"WAP>") except: time.sleep(1) continue else: while True: try: tel.write(b"ip route\n") data = tel.read_until(b"WAP>").decode() if ip_find := re.search("ppp.*?src (\d+\.\d+\.\d+\.\d+)", data): yield ip_find.group(1) except: time.sleep(1) break def set_dns_ip(ip): client = AcsClient(aliyun_id, aliyun_secret) for domain in domain_lst: request = DescribeDomainRecordsRequest() request.set_accept_format('json') request.set_DomainName(domain) response = client.do_action_with_exception(request) json_data = json.loads(str(response, encoding='utf-8')) for record in json_data["DomainRecords"]["Record"]: if record["Value"] == ip: continue request = UpdateDomainRecordRequest() request.set_accept_format('json') request.set_Value(ip) request.set_Type(record["Type"]) request.set_TTL(600) request.set_RR(record["RR"]) request.set_RecordId(record["RecordId"]) client.do_action_with_exception(request) def get_domain_ip(domain): resolver = dns.resolver.Resolver() resolver.nameservers = ['61.139.2.69', '218.6.200.139', '202.98.96.68'] return resolver.resolve(domain, 'A')[0].to_text() def run(): ip_getter = get_wan_ip() while True: # noinspection PyBroadException try: wan_ip = next(ip_getter) domain_ip = get_domain_ip(domain_lst[0]) if wan_ip and domain_ip and wan_ip != domain_ip: set_dns_ip(wan_ip) except Exception: pass finally: time.sleep(30)
備注:
代碼是基於Python 3.8.10開發。自己用實現有些簡陋,請自行完善。