update

Dockerfile

@@ -0,0 +1,13 @@
+FROM ubuntu:22.04
+
+COPY ./dns /dns
+COPY ./server.py /server.py
+
+RUN apt update && apt -y upgrade && \
+    apt install -y python3 python3-pip && \
+    python3 -m pip install gradio fastapi uvicorn requests -i https://pypi.mirrors.ustc.edu.cn/simple
+
+CMD ["python3", "/server.py"]
+
+# docker build -t hky3535/hky3535:dns .
+# docker tag hky3535/hky3535:dns 

dns/__init__.py

@@ -0,0 +1 @@
+from .aliyun import AliyunDNS

dns/aliyun.py

@@ -0,0 +1,75 @@
+import requests
+import time
+import hmac
+import hashlib
+import base64
+import datetime
+import urllib
+
+
+class AliyunDNS:
+    def __init__(self, BASE_URL, ACCESS_KEY_ID, ACCESS_KEY_SECRET):
+        self.BASE_URL = BASE_URL
+        self.ACCESS_KEY_ID = ACCESS_KEY_ID
+        self.ACCESS_KEY_SECRET = ACCESS_KEY_SECRET
+
+    def params(self, params): # 请求体公共加密
+        params = {
+            'Format': 'JSON', 
+            'Version': '2015-01-09', 
+            'AccessKeyId': ACCESS_KEY_ID, 
+            'SignatureMethod': 'HMAC-SHA1', 
+            'SignatureVersion': '1.0',
+            'SignatureNonce': str(time.time()), 
+            'Timestamp': datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 
+            **params
+        }
+        # 加密验证签名部分
+        sign = sorted(params.items(), key=lambda x: x[0])
+        sign = '&'.join([urllib.parse.quote_plus(k) + '=' + urllib.parse.quote_plus(v) for k, v in sign])
+        sign = 'GET&%2F&' + urllib.parse.quote_plus(sign)
+        sign = base64.b64encode(
+            hmac.new(
+                ACCESS_KEY_SECRET.encode('utf-8') + b'&', 
+                sign.encode('utf-8'), 
+                hashlib.sha1
+            ).digest()
+        ).decode('utf-8')
+        params['Signature'] = sign
+        return params
+
+    def DescribeDomainRecords(self, DomainName, PageSize=100): # 查看所有解析记录
+        response = requests.get(
+            url=BASE_URL, 
+            params=self.params(params={
+                'Action': 'DescribeDomainRecords',
+                'DomainName': DomainName, 
+                "PageSize": str(PageSize)
+            })
+        )
+        assert response.status_code == 200
+        return response.json()
+
+    def AddDomainRecord(self, DomainName, RR, Type, Value, Line="default", TTL=600): # 新增解析记录
+        response = requests.get(
+            url=BASE_URL, 
+            params=self.params(params={
+                'Action': 'AddDomainRecord',
+                'DomainName': DomainName,
+                'RR': RR, 'Type': Type, 'Value': Value, 'Line': Line, 'TTL': str(TTL)
+            })
+        )
+        assert response.status_code == 200
+        return response.json()
+
+    def DeleteDomainRecord(self, RecordId): # 删除解析记录
+        response = requests.get(
+            url=BASE_URL, 
+            params=self.params(params={
+                'Action': 'DeleteDomainRecord',
+                'RecordId': RecordId, 
+            })
+        )
+        assert response.status_code == 200
+        return response.json()
+

server.py

@@ -0,0 +1,79 @@
+import gradio
+import fastapi
+import uvicorn
+
+from dns import AliyunDNS
+
+# Aliyun DNS 相关配置
+DOMAIN_NAMES =      str(os.getenv("DOMAIN_NAMES",       "")).split(",")
+BASE_URL =          str(os.getenv("BASE_URL",           "http://alidns.aliyuncs.com"))
+ACCESS_KEY_ID =     str(os.getenv("ACCESS_KEY_ID",      ""))
+ACCESS_KEY_SECRET = str(os.getenv("ACCESS_KEY_SECRET",  ""))
+
+# gradio 管理页面 相关配置
+USERNAME =          str(os.getenv("USERNAME",           "admin"))
+PASSWORD =          str(os.getenv("PASSWORD",           "password"))
+SERVER_HOST =       str(os.getenv("SERVER_HOST",        "0.0.0.0"))
+SERVER_PORT =       int(os.getenv("SERVER_PORT",        "60000"))
+
+
+class Server:
+    def __init__(self):
+        self.aliyun_dns = AliyunDNS(BASE_URL, ACCESS_KEY_ID, ACCESS_KEY_SECRET)
+        self.resolutions = {domain_name: list() for domain_name in DOMAIN_NAMES}
+        
+    def domain(self, domain):
+        self.resolutions[domain].clear()
+        records = self.aliyun_dns.DescribeDomainRecords(DomainName=domain)
+        for record in records["DomainRecords"]["Record"]:
+            self.resolutions[domain].append(
+                [record["RR"], record["Value"], record["RecordId"]])
+        self.resolutions[domain] = sorted(self.resolutions[domain], key=lambda x: x[0])
+        
+    def webui(self, app, path):
+        with gradio.Blocks() as webui:
+            domain = gradio.Dropdown(label="主域名", choices=DOMAIN_NAMES, value=None)
+            editor = gradio.DataFrame(label="编辑解析信息", headers=["子域名", "解析地址", "记录编码"], 
+                                      row_count=100, col_count=(3, "fixed"), datatype=["str", "str", "str"], type="pandas")
+            update = gradio.Button(value="更新解析信息")
+            
+            @domain.select(inputs=[domain], outputs=[editor])
+            def _(domain):
+                domain = str(domain)
+                self.domain(domain=domain) # 更新解析信息
+                return self.resolutions[domain]
+
+            @update.click(inputs=[domain, editor], outputs=[editor])
+            def _(domain, editor):
+                domain = str(domain)
+                editor = [line for line in editor.values.tolist() if "".join(line) != ""] # 去除空行
+
+                last = {tuple(line) for line in self.resolutions[domain]}
+                now = {tuple(line) for line in editor}
+                
+                for rr, ip, id in (now - last):
+                    if rr == "" or ip == "": continue # 子域名或解析地址为空则不做解析
+                    print(f"创建解析：{domain} {rr} {ip}")
+                    self.aliyun_dns.AddDomainRecord(DomainName=domain, RR=rr, Type="A", Value=ip, Line="default", TTL=600)
+                    
+                for rr, ip, id in (last - now):
+                    print(f"删除解析：{domain} {rr} {ip} {id}")
+                    self.aliyun_dns.DeleteDomainRecord(RecordId=id)
+                
+                self.domain(domain=domain) # 更新解析信息
+                return self.resolutions[domain]
+
+        gradio.mount_gradio_app(app, webui, path=path, auth=(USERNAME, PASSWORD))
+
+    def run(self, SERVER_HOST, SERVER_PORT):
+        app = fastapi.FastAPI()
+        self.webui(app=app, path="/")
+        uvicorn.run(app, host=SERVER_HOST, port=int(SERVER_PORT))
+
+
+if __name__ == "__main__":
+    server = Server()
+    server.run(
+        SERVER_HOST=SERVER_HOST, 
+        SERVER_PORT=SERVER_PORT
+    )