Router tools

cloudflare_ddns.py

#!/usr/bin/env python3

import json
import os
import sys
import requests

CLOUDFLARE_API_BASE = "https://api.cloudflare.com/client/v4"

def get_public_ipv4():
    res = requests.get("https://api.ipify.org?format=json")
    return res.json()["ip"]

def update_dns_record(token: str, zone_id: str, name: str):
    ipv4 = get_public_ipv4()
    cloudflare_headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    dns_records_res = requests.get(
        f"{CLOUDFLARE_API_BASE}/zones/{zone_id}/dns_records",
        params={ "name": name },
        headers=cloudflare_headers,
    ).json()
    if not dns_records_res.get("success"):
        return None
    dns_records = dns_records_res["result"]
    dns_record = dns_records[0]
    dns_record_id = dns_record["id"]
    patch_header = cloudflare_headers | { "Content-Type": "application/json" }
    res = requests.patch(
        f"{CLOUDFLARE_API_BASE}/zones/{zone_id}/dns_records/{dns_record_id}",
        data=json.dumps({ "content": ipv4 }),
        headers=patch_header,
    )
    return res.json()

if __name__ == "__main__":
    token = sys.argv[1] if 1 < len(sys.argv) else os.environ.get("CLOUDFLARE_EDIT_TOKEN")
    zone_id = sys.argv[2] if 2 < len(sys.argv) else os.environ.get("CLOUDFLARE_ZONE_ID")
    record_name = sys.argv[3] if 3 < len(sys.argv) else os.environ.get("CLOUDFLARE_RECORD_NAME")
    if zone_id and token and record_name:
        res = update_dns_record(token, zone_id, record_name)
        if res and res.get("success"):
            print("Success")
        else:
            sys.exit(1)

hosts.py

#!/usr/bin/env python3

import os
import sys
from datetime import datetime
from notion_tools import query_device_list, IPV4_PROPERTY_NAME, FQDN_PROPERTY_NAME

NOTION_TOKEN = os.environ.get("NOTION_TOKEN")
NOTION_DB_ID = os.environ.get("NOTION_DB_ID")

def get_hosts(token: str, db_id: str) -> str:
    query = {
        "filter": {
            "and": [
                {
                    "property": IPV4_PROPERTY_NAME,
                    "rich_text": {
                        "is_not_empty": True,
                    },
                },
                {
                    "property": FQDN_PROPERTY_NAME,
                    "rich_text": {
                        "is_not_empty": True,
                    },
                }
            ]
        }
    }
    devices = query_device_list(token, db_id, query=query)
    devices.reverse()
    res = ""
    for device in devices:
        if device.fqdn and device.ipv4_address:
            res += f"{device.ipv4_address} {device.fqdn}\n"
    return res

if __name__ == "__main__":
    if not NOTION_TOKEN:
        print("Notion Token not found", file=sys.stderr)
    elif not NOTION_DB_ID:
        print("Notion DB not found", file=sys.stderr)
    else:
        print(get_hosts(NOTION_TOKEN, NOTION_DB_ID))

notion_tools.py

#!/usr/bin/env python3

import os
import sys
import json
import requests

NOTION_TOKEN = os.environ.get("NOTION_TOKEN")
NOTION_DB_ID = os.environ.get("NOTION_DB_ID")
NOTION_API_BASE = "https://api.notion.com"
NOTION_VERSION = "2022-06-28"
NAME_PROPERTY_NAME = "Name"
KIND_PROPERTY_NAME = "Kind"
FQDN_PROPERTY_NAME = "FQDN"
IPV4_PROPERTY_NAME = "IPv4 address"
URL_PROPERTY_NAME = "URL"
CONNECTED_TO_PROPERTY_NAME = "Connected to"

class Device:
    id: str
    name: str
    icon_url: str | None
    kind: list[str]
    fqdn: str | None
    ipv4_address: str | None
    url: str | None
    connected_to: list[str]
    def __init__(
        self,
        id: str,
        name: str,
        icon_url: str | None,
        kind: list[str],
        fqdn: str | None,
        ipv4_address: str | None,
        url: str | None,
        connected_to: list[str],
    ):
        self.id = id
        self.name = name
        self.icon_url = icon_url
        self.kind = kind
        self.fqdn = fqdn
        self.ipv4_address = ipv4_address
        self.url = url
        self.connected_to = connected_to


def ipv4_sort_key(ipv4: str) -> str:
    ipe = ipv4.split(".")
    return f"{ipe[0]:03d}.{ipe[1]:03d}.{ipe[2]:03d}.{ipe[3]:03d}"

def db_to_device_lsit(db) -> list[Device]:
    res = []
    results = db.get("results")
    for r in results:
        obj_id = r.get("id")
        properties = r.get("properties")
        name = properties.get(NAME_PROPERTY_NAME).get("title")[0].get("plain_text")
        icon_obj = r.get("icon")
        if icon_obj and icon_obj.get("type") == "file":
            icon_url = icon_obj.get("file").get("url")
        else:
            icon_url = None
        kind_objs = properties.get(KIND_PROPERTY_NAME).get("multi_select")
        kinds = list(map(lambda x: x.get("name"), kind_objs))
        fqdn_rich_text = properties.get(FQDN_PROPERTY_NAME).get("rich_text")
        if 0 < len(fqdn_rich_text):
            fqdn = fqdn_rich_text[0].get("plain_text")
        else:
            fqdn = None
        ipv4_rich_text = properties.get(IPV4_PROPERTY_NAME).get("rich_text")
        if 0 < len(ipv4_rich_text):
            ipv4 = ipv4_rich_text[0].get("plain_text")
        else:
            ipv4 = None
        url = properties.get(URL_PROPERTY_NAME).get("url")
        relation_objs = properties.get(CONNECTED_TO_PROPERTY_NAME).get("relation")
        relations = list(map(lambda x: x.get("id"), relation_objs))
        res.append(Device(obj_id, name, icon_url, kinds, fqdn, ipv4, url, relations))
    return res

def query_device_list(token: str, db_id: str, query={}) -> list[Device]:
    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": NOTION_VERSION,
        "Content-Type": "application/json",
    }
    res = requests.post(f"{NOTION_API_BASE}/v1/databases/{db_id}/query", data=json.dumps(query), headers=headers)
    db = res.json()
    return db_to_device_lsit(db)


if __name__ == "__main__":
    if not NOTION_TOKEN:
        print("Notion Token not found", file=sys.stderr)
    elif not NOTION_DB_ID:
        print("Notion DB not found", file=sys.stderr)
    else:
        print(query_device_list(NOTION_TOKEN, NOTION_DB_ID))