MagicDNSを無効化して、Tailscaleのnameserverに依存しないようにした

#!/usr/bin/env python3
import json
import os
import subprocess
import sys
from typing import Dict, Tuple, List

import requests

# ===== 設定 =====

# AdGuard Home のベースURL
ADGUARD_BASE_URL = os.environ.get("ADGUARD_BASE_URL", "http://127.0.0.1")

# AdGuard Home のログイン情報（Basic認証想定）
ADGUARD_USERNAME = os.environ.get("ADGUARD_USERNAME", "FILL_YOUR_USERNAME")
ADGUARD_PASSWORD = os.environ.get("ADGUARD_PASSWORD", "FILL_YOUR_PASSWORD")

# Tailscale コマンド
TAILSCALE_BIN = os.environ.get("TAILSCALE_BIN", "tailscale")

# 自前ゾーンのサフィックス（例: ".home.arpa"）
DOMAIN_SUFFIX = os.environ.get("DOMAIN_SUFFIX", "FILL_YOUR_SUFFIX").strip()

# ドライラン（True の場合は API を叩かずに stdout にだけ出力）
DRY_RUN = os.environ.get("DRY_RUN", "0") == "1"

# ===== 内部処理 =====


def run_tailscale_status() -> dict:
    """tailscale status --json を実行して dict を返す"""
    try:
        out = subprocess.check_output(
            [TAILSCALE_BIN, "status", "--json"],
            stderr=subprocess.STDOUT,
        )
    except subprocess.CalledProcessError as e:
        print(f"[ERROR] tailscale status failed: {e.output.decode(errors='ignore')}", file=sys.stderr)
        sys.exit(1)

    return json.loads(out)


def build_desired_records(ts_status: dict) -> Dict[str, str]:
    """
    tailscale status --json の結果から
    {domain: ip} という dict を生成する。
    domain は "hostname + DOMAIN_SUFFIX"。
    """
    records: Dict[str, str] = {}

    def add_entry(hostname: str, ip: str) -> None:
        if not hostname or not ip:
            return
        name = hostname.split(".")[0].lower()  # "host.tailnet.ts.net" → "host"
        domain = f"{name}{DOMAIN_SUFFIX}"
        # 既に登録済みなら上書きしない（最初の IP を優先）
        if domain not in records:
            records[domain] = ip

    # Self
    self_info = ts_status.get("Self", {})
    for ip in self_info.get("TailscaleIPs", []):
        if "." in ip:  # IPv4 だけ使う
            add_entry(self_info.get("DNSName"), ip)
            break

    # Peers
    for peer in ts_status.get("Peer", {}).values():
        for ip in peer.get("TailscaleIPs", []):
            if "." in ip:
                add_entry(peer.get("DNSName"), ip)
                break

    return records


def is_ipv4(addr: str) -> bool:
    """ざっくり IPv4 アドレスかどうかを判定"""
    parts = addr.split(".")
    if len(parts) != 4:
        return False
    for p in parts:
        if not p.isdigit():
            return False
        n = int(p)
        if not (0 <= n <= 255):
            return False
    return True


def adguard_get_user_rules() -> List[str]:
    """
    AdGuard Home から現在の user_rules（カスタムフィルタ）を取得する。
    """
    url = f"{ADGUARD_BASE_URL}/control/filtering/status"
    auth = (ADGUARD_USERNAME, ADGUARD_PASSWORD)
    resp = requests.get(url, auth=auth, timeout=5)
    resp.raise_for_status()
    data = resp.json()
    rules = data.get("user_rules", [])
    if not isinstance(rules, list):
        print(f"[WARN] unexpected user_rules type: {type(rules)}", file=sys.stderr)
        return []
    return rules


def adguard_set_user_rules(rules: List[str]) -> None:
    """
    AdGuard Home の user_rules 全体を置き換える。
    """
    url = f"{ADGUARD_BASE_URL}/control/filtering/set_rules"
    auth = (ADGUARD_USERNAME, ADGUARD_PASSWORD)
    payload = {"rules": rules}
    if DRY_RUN:
        print("[DRY-RUN] SET user_rules:")
        for line in rules:
            print(f"  {line}")
        return
    resp = requests.post(url, auth=auth, json=payload, timeout=5)
    if resp.status_code != 200:
        print(f"[ERROR] set_rules failed: {resp.status_code} {resp.text}", file=sys.stderr)


def split_static_and_dynamic_rules(rules: List[str]) -> Tuple[List[str], Dict[str, str]]:
    """
    user_rules を
      - static_rules: そのまま残すルール（広告ブロックなど）
      - dynamic_map:  *.DOMAIN_SUFFIX な hosts 形式エントリ {domain: ip}
    に分割する。

    dynamic の対象とする形式:
      " "
    かつ domain.endswith(DOMAIN_SUFFIX)
    """
    static_rules: List[str] = []
    dynamic_map: Dict[str, str] = {}

    for line in rules:
        raw = line.rstrip("\n")
        stripped = raw.strip()

        # 空行・コメントは static 扱い
        if not stripped or stripped.startswith("#"):
            static_rules.append(raw)
            continue

        parts = stripped.split()
        if len(parts) == 2:
            ip, domain = parts
            if is_ipv4(ip) and domain.endswith(DOMAIN_SUFFIX):
                # 動的に管理するエントリ
                dynamic_map[domain] = ip
                continue

        # それ以外は static
        static_rules.append(raw)

    return static_rules, dynamic_map


def diff_and_apply(desired: Dict[str, str], current_rules: List[str]) -> Tuple[int, int]:
    """
    desired: {domain: ip} （tailscale から生成した理想状態）
    current_rules: AdGuard Home の現在の user_rules

    - current_rules を static/dynamic に分割
    - dynamic 部分は DOMAIN_SUFFIX の hosts 形式だけを対象にする
    - dynamic 部分を desired から再生成し、set_rules で丸ごと更新

    戻り値: (num_added, num_deleted)
    """
    static_rules, current_dynamic = split_static_and_dynamic_rules(current_rules)

    # 差分計算用
    desired_dynamic = desired
    added = 0
    deleted = 0

    # 追加・更新: domain が新規 or IP 変更
    for d, ip in desired_dynamic.items():
        old_ip = current_dynamic.get(d)
        if old_ip != ip:
            added += 1

    # 削除: current_dynamic にあるが desired に無いもの
    for d in current_dynamic.keys():
        if d not in desired_dynamic:
            deleted += 1

    print(
        "[INFO] desired_dynamic=%d current_dynamic=%d add=%d del=%d"
        % (len(desired_dynamic), len(current_dynamic), added, deleted)
    )

    # 実際に適用する dynamic hosts 行を組み立てる
    # （domain でソートして安定した順序にしておく）
    new_dynamic_lines: List[str] = []
    for domain in sorted(desired_dynamic.keys()):
        ip = desired_dynamic[domain]
        new_dynamic_lines.append(f"{ip} {domain}")

    new_rules = static_rules + new_dynamic_lines

    # 適用
    adguard_set_user_rules(new_rules)

    return added, deleted


def main() -> None:
    ts_status = run_tailscale_status()
    desired = build_desired_records(ts_status)
    current_rules = adguard_get_user_rules()
    added, deleted = diff_and_apply(desired, current_rules)
    if DRY_RUN:
        print("[INFO] DRY-RUN finished.")
    else:
        print(f"[INFO] applied: added={added}, deleted={deleted}")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"[FATAL] {e}", file=sys.stderr)
        sys.exit(1)