init

porkbun-refresher.py

@@ -0,0 +1,96 @@
+import http.client
+from types import MappingProxyType
+import typing as t
+import json
+import os
+
+# Need to define the PRIMARY_DOMAIN, otherwise it will fail
+PRIMARY_DOMAIN: t.Final = os.environ["PRIMARY_DOMAIN"]
+PORKBUN_DOMAIN: t.Final = os.environ.get("PORKBUN_DOMAIN" , "api.porkbun.com")
+REFERENTIAL_DOMAIN: t.Final = os.environ.get("REFERENTIAL_DOMAIN", "ifconfig.me")
+PORKBUN_DNS_PREFIX: t.Final = os.environ.get("PORKBUN_DNS_PREFIX", "/api/json/v3/dns")
+SECRET_FILE: t.Final = os.environ.get("SECRET_FILE", "/run/agenix/porkbun")
+
+with open(SECRET_FILE) as f:
+    KEYS: t.Final = MappingProxyType(dict(x.strip().split("=") for x in f.readlines()))
+
+
+def get_current_ip(domain: str):
+    conn = http.client.HTTPSConnection(domain)
+    conn.request(method="GET", url="/ip")
+    res = conn.getresponse()
+    data = res.read().decode()
+    if res.status != 200:
+        raise RuntimeError(f"couldn't get ip addr: {res.status}: {res.reason}")
+    conn.close()
+    return data
+
+
+class PorkbunRecord(t.TypedDict):
+    id: str
+    name: str
+    type: str
+    content: str
+    ttl: str
+    prio: str
+    notes: str
+
+
+class PorkbunResponse(t.TypedDict):
+    status: str
+    cloudflare: t.Literal["enabled", "disabled"]
+    records: list[PorkbunRecord]
+
+
+def get_porkbun_record(domain: str) -> PorkbunResponse:
+    body = KEYS.copy()
+    conn = http.client.HTTPSConnection(PORKBUN_DOMAIN)
+    conn.request(
+        method="POST",
+        url=f"{PORKBUN_DNS_PREFIX}/retrieveByNameType/{domain}/A/",
+        body=json.dumps(body),
+    )
+    res = conn.getresponse()
+    data = res.read().decode()
+    if res.status != 200:
+        raise RuntimeError(f"couldn't get ip addr: {res.status}: {res.reason}")
+    conn.close()
+    return json.loads(data)
+
+
+def set_porkbun_record(domain: str, ip_addr: str, id: str):
+    body = KEYS.copy()
+    body.update(type="A", content=ip_addr)
+    conn = http.client.HTTPSConnection(PORKBUN_DOMAIN)
+    conn.request(
+        method="POST",
+        url=f"{PORKBUN_DNS_PREFIX}/edit/{domain}/{id}",
+        body=json.dumps(body),
+    )
+    res = conn.getresponse()
+    data = res.read().decode()
+    if res.status != 200:
+        raise RuntimeError(f"couldn't set ip addr: {res.status}: {res.reason}")
+    conn.close()
+    return json.loads(data)
+
+
+def main(domain: str):
+    current_ip = get_current_ip(REFERENTIAL_DOMAIN)
+
+    if (porkbun_record := get_porkbun_record(domain=domain))["status"] != "SUCCESS" or len(
+        porkbun_record["records"]
+    ) != 1:
+        raise RuntimeError("Something went very wrong")
+
+    if (porkbun_ip := porkbun_record["records"][0]["content"]) == current_ip:
+        print("the ip hasn't changed. Nothing left to be done...")
+    else:
+        print(f"Current IP: {current_ip} Porkbun IP: {porkbun_ip}. Updating record...")
+        id = porkbun_record["records"][0]["id"]
+        set_porkbun_record(domain=domain, ip_addr=current_ip, id=id)
+
+
+if __name__ == "__main__":
+    main(PRIMARY_DOMAIN)
+