initial commit

src/config.py

@@ -0,0 +1,7 @@
+import tomllib
+from pathlib import Path
+
+
+def load_config(path: str = "config.toml") -> dict:
+    with Path(path).open("rb") as f:
+        return tomllib.load(f)

src/ip.py

@@ -0,0 +1,36 @@
+import subprocess
+
+SERVERS = [
+    ("resolver1.opendns.com", "myip.opendns.com"),
+    ("resolver2.opendns.com", "myip.opendns.com"),
+    ("resolver3.opendns.com", "myip.opendns.com"),
+    ("resolver4.opendns.com", "myip.opendns.com"),
+    ("ns1-1.akamaitech.net", "whoami.akamai.net"),
+]
+
+
+def get_public_ip():
+    for server, host in SERVERS:
+        try:
+            result = subprocess.run(
+                [
+                    "dig",
+                    "+short",
+                    host,
+                    f"@{server}",
+                ],
+                capture_output=True,
+                text=True,
+                check=True,
+                timeout=5,
+            )
+
+            ip = result.stdout.strip()
+
+            if ip:
+                return ip
+
+        except Exception:
+            pass
+
+    raise RuntimeError("cannot determine public IP")

src/main.py

@@ -0,0 +1,24 @@
+from config import load_config
+from ip import get_public_ip
+from state import load_state, save_success, save_failure
+from nsupdate import update_dns
+
+
+cfg = load_config()
+
+ip = get_public_ip()
+
+state = load_state()
+
+if state["ip"] == ip:
+    print("IP unchanged.")
+    raise SystemExit(0)
+
+try:
+    update_dns(cfg, ip)
+    save_success(ip)
+    print("DNS updated.")
+
+except Exception:
+    save_failure()
+    raise

src/nsupdate.py

@@ -0,0 +1,33 @@
+import subprocess
+
+
+def update_dns(cfg: dict, ip: str) -> None:
+    commands = f"""\
+server {cfg["server"]}
+zone {cfg["zone"]}
+update delete {cfg["hostname"]} A
+update add {cfg["hostname"]} {cfg["ttl"]} A {ip}
+send
+"""
+
+    print("===== nsupdate input =====")
+    print(commands)
+
+    result = subprocess.run(
+        ["nsupdate", "-k", cfg["keyfile"]],
+        input=commands,
+        text=True,
+        capture_output=True,
+    )
+
+    print("===== return code =====")
+    print(result.returncode)
+
+    print("===== stdout =====")
+    print(result.stdout)
+
+    print("===== stderr =====")
+    print(result.stderr)
+
+    if result.returncode != 0:
+        raise RuntimeError("nsupdate failed")

src/state.py

@@ -0,0 +1,41 @@
+# src/state.py
+
+from pathlib import Path
+from datetime import datetime
+import json
+
+
+STATE_FILE = Path("state.json")
+
+
+def load_state() -> dict:
+    if not STATE_FILE.exists():
+        return {
+            "ip": "",
+            "updated_at": "",
+            "failure_count": 0,
+        }
+
+    return json.loads(STATE_FILE.read_text())
+
+
+def save_success(ip: str):
+    state = {
+        "ip": ip,
+        "updated_at": datetime.now().astimezone().isoformat(),
+        "failure_count": 0,
+    }
+
+    STATE_FILE.write_text(
+        json.dumps(state, indent=4)
+    )
+
+
+def save_failure():
+    state = load_state()
+
+    state["failure_count"] += 1
+
+    STATE_FILE.write_text(
+        json.dumps(state, indent=4)
+    )

src/wireguard.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import ipaddress
+import shutil
+import socket
+import subprocess
+from dataclasses import dataclass
+
+
+class WireGuardUnavailable(RuntimeError):
+    """WireGuard 터널을 통해 DDNS 서버에 도달할 수 없을 때 발생."""
+
+
+@dataclass(frozen=True)
+class WireGuardStatus:
+    interface: str
+    server_ip: str
+    route_ok: bool
+    dns_ok: bool
+
+
+def _run(command: list[str], timeout: float = 3.0) -> str:
+    try:
+        result = subprocess.run(
+            command,
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    except FileNotFoundError as exc:
+        raise WireGuardUnavailable(
+            f"명령을 찾을 수 없습니다: {command[0]}"
+        ) from exc
+    except subprocess.TimeoutExpired as exc:
+        raise WireGuardUnavailable(
+            f"명령 실행 시간이 초과되었습니다: {' '.join(command)}"
+        ) from exc
+    except subprocess.CalledProcessError as exc:
+        stderr = exc.stderr.strip() if exc.stderr else "오류 내용 없음"
+        raise WireGuardUnavailable(
+            f"명령 실행 실패: {' '.join(command)}: {stderr}"
+        ) from exc
+
+    return result.stdout.strip()
+
+
+def check_interface(interface: str = "wg0") -> None:
+    if shutil.which("wg") is None:
+        raise WireGuardUnavailable("wg 명령이 설치되어 있지 않습니다.")
+
+    interfaces = _run(["wg", "show", "interfaces"]).split()
+
+    if interface not in interfaces:
+        raise WireGuardUnavailable(
+            f"WireGuard 인터페이스 {interface!r}가 활성화되지 않았습니다."
+        )
+
+
+def check_route(server_ip: str, interface: str = "wg0") -> None:
+    ipaddress.ip_address(server_ip)
+
+    output = _run(["ip", "route", "get", server_ip])
+
+    if f"dev {interface}" not in output:
+        raise WireGuardUnavailable(
+            f"{server_ip} 경로가 {interface}를 사용하지 않습니다: {output}"
+        )
+
+
+def check_dns_port(server_ip: str, port: int = 53) -> None:
+    # TCP 53 확인. 실제 nsupdate는 UDP 또는 TCP를 사용할 수 있으므로
+    # 이후 dig/nsupdate 테스트도 별도로 수행한다.
+    try:
+        with socket.create_connection((server_ip, port), timeout=3.0):
+            return
+    except OSError as exc:
+        raise WireGuardUnavailable(
+            f"{server_ip}:{port}에 연결할 수 없습니다."
+        ) from exc
+
+
+def require_wireguard(
+    server_ip: str = "10.77.0.1",
+    interface: str = "wg0",
+) -> WireGuardStatus:
+    check_interface(interface)
+    check_route(server_ip, interface)
+    check_dns_port(server_ip)
+
+    return WireGuardStatus(
+        interface=interface,
+        server_ip=server_ip,
+        route_ok=True,
+        dns_ok=True,
+    )