roof
ffe1dbeed6
Unit Tests / test (push) Failing after 8m57s
setup_cell.py: register_with_ddns() called at end of setup — detects public IP via api.ipify.org, generates TOTP code from DDNS_TOTP_SECRET, POSTs to DDNS /register, saves token to data/api/.ddns_token (mode 600). Idempotent: skips if token file already exists. Fails gracefully if DDNS_TOTP_SECRET is unset or network is unreachable. scripts/ddns_update.py: standalone script for periodic IP updates. Reads token from data/api/.ddns_token, fetches current public IP, compares to cached last IP (data/api/.ddns_last_ip) and calls /update only when the IP has actually changed. Makefile: add ddns-update (run update script) and ddns-register (force re-registration by removing old token then calling register_with_ddns). Usage: DDNS_TOTP_SECRET=<secret> make ddns-register Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Update the cell's DDNS record with the current public IP.
|
|
|
|
Called by: make ddns-update
|
|
systemd timer (optional, see scripts/pic-ddns-update.timer)
|
|
|
|
Reads the DDNS token from data/api/.ddns_token (written by setup_cell.py).
|
|
Exits 0 on success or if already up to date, non-zero on failure.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
DDNS_URL = os.environ.get('DDNS_URL', 'https://ddns.pic.ngo/api/v1')
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
TOKEN_FILE = os.path.join(ROOT, 'data', 'api', '.ddns_token')
|
|
IP_CACHE_FILE = os.path.join(ROOT, 'data', 'api', '.ddns_last_ip')
|
|
|
|
|
|
def get_public_ip() -> str:
|
|
return urllib.request.urlopen('https://api.ipify.org', timeout=5).read().decode().strip()
|
|
|
|
|
|
def read_token() -> str:
|
|
if not os.path.exists(TOKEN_FILE):
|
|
print('ERROR: DDNS token not found. Run "make setup" to register.', file=sys.stderr)
|
|
sys.exit(1)
|
|
return open(TOKEN_FILE).read().strip()
|
|
|
|
|
|
def read_last_ip() -> str:
|
|
try:
|
|
return open(IP_CACHE_FILE).read().strip()
|
|
except FileNotFoundError:
|
|
return ''
|
|
|
|
|
|
def write_last_ip(ip: str) -> None:
|
|
with open(IP_CACHE_FILE, 'w') as f:
|
|
f.write(ip)
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
public_ip = get_public_ip()
|
|
except Exception as e:
|
|
print(f'ERROR: Could not detect public IP: {e}', file=sys.stderr)
|
|
return 1
|
|
|
|
last_ip = read_last_ip()
|
|
if public_ip == last_ip:
|
|
print(f'DDNS: IP unchanged ({public_ip}) — no update needed')
|
|
return 0
|
|
|
|
token = read_token()
|
|
data = json.dumps({'token': token, 'ip': public_ip}).encode()
|
|
req = urllib.request.Request(
|
|
f'{DDNS_URL}/update',
|
|
data=data,
|
|
headers={'Content-Type': 'application/json'},
|
|
method='PUT',
|
|
)
|
|
try:
|
|
resp = urllib.request.urlopen(req, timeout=10)
|
|
result = json.loads(resp.read())
|
|
if result.get('updated'):
|
|
write_last_ip(public_ip)
|
|
print(f'DDNS: Updated to {public_ip}')
|
|
return 0
|
|
else:
|
|
print(f'ERROR: Unexpected response: {result}', file=sys.stderr)
|
|
return 1
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode()
|
|
print(f'ERROR: DDNS update failed ({e.code}): {body}', file=sys.stderr)
|
|
return 1
|
|
except Exception as e:
|
|
print(f'ERROR: DDNS update failed: {e}', file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|