hmk
104 lines
3.5 KiB
Python
Executable File
104 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import sys
|
|
import requests
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
api_url = "https://pi.hole/api"
|
|
verify = "/etc/pihole/tls_ca.crt"
|
|
default_password_file = "/etc/pihole/api_password"
|
|
|
|
parser = argparse.ArgumentParser(description="List non-static (dynamic) Pi-hole DHCP leases.")
|
|
parser.add_argument("-p", "--password-file", default=default_password_file,
|
|
help=f"Path to file containing the Pi-hole API password (default: {default_password_file})")
|
|
parser.add_argument("-t", "--table", action="store_true",
|
|
help="Print a human-readable aligned table (with local time expiry)")
|
|
parser.add_argument("-j", "--json", action="store_true",
|
|
help="Print machine-readable output as JSON instead of the default CSV")
|
|
args = parser.parse_args()
|
|
|
|
if args.table and args.json:
|
|
parser.error("--table and --json are mutually exclusive")
|
|
|
|
|
|
def read_password(path):
|
|
try:
|
|
return Path(path).read_text().strip()
|
|
except OSError as e:
|
|
raise SystemExit(f"Could not read password file '{path}': {e}")
|
|
|
|
|
|
def format_expiry_human(value):
|
|
try:
|
|
return datetime.fromtimestamp(int(value)).strftime("%Y-%m-%d %H:%M:%S")
|
|
except (TypeError, ValueError):
|
|
return str(value)
|
|
|
|
|
|
auth_payload = {"password": read_password(args.password_file)}
|
|
|
|
with requests.post(f"{api_url}/auth", json=auth_payload, verify=verify) as auth:
|
|
auth_json = auth.json()
|
|
session = auth_json.get('session')
|
|
if not session or not session.get('sid'):
|
|
raise SystemExit(f"Authentication failed: {auth_json}")
|
|
|
|
headers = {
|
|
"X-FTL-SID": session.get('sid'),
|
|
"X-FTL-CSRF": session.get('sid')
|
|
}
|
|
|
|
try:
|
|
# Get static DHCP host mappings to know which MACs are static
|
|
with requests.get(f"{api_url}/config", headers=headers, verify=verify) as config:
|
|
config_json = config.json()
|
|
static_macs = set()
|
|
for entry in config_json.get('config').get('dhcp').get('hosts'):
|
|
mac = entry.split(",")[0].lower()
|
|
static_macs.add(mac)
|
|
|
|
# Get current active DHCP leases
|
|
with requests.get(f"{api_url}/dhcp/leases", headers=headers, verify=verify) as leases_resp:
|
|
leases_json = leases_resp.json()
|
|
leases = leases_json.get('leases', [])
|
|
|
|
non_static = [l for l in leases if l.get('hwaddr', '').lower() not in static_macs]
|
|
|
|
records = []
|
|
for lease in non_static:
|
|
records.append({
|
|
"ip": lease.get('ip', '-'),
|
|
"mac": lease.get('hwaddr', '-'),
|
|
"hostname": lease.get('name') or '-',
|
|
"expires": lease.get('expires', '-'),
|
|
})
|
|
|
|
if args.table:
|
|
if not records:
|
|
print("No non-static (dynamic) leases found.")
|
|
else:
|
|
rows = [(r["ip"], r["mac"], r["hostname"], format_expiry_human(r["expires"])) for r in records]
|
|
ip_w = max(len(r[0]) for r in rows + [("IP", "", "", "")])
|
|
mac_w = max(len(r[1]) for r in rows + [("", "MAC", "", "")])
|
|
name_w = max(len(r[2]) for r in rows + [("", "", "Hostname", "")])
|
|
|
|
print(f"{'IP':<{ip_w}} {'MAC':<{mac_w}} {'Hostname':<{name_w}} Expires (local time)")
|
|
print(f"{'-' * ip_w} {'-' * mac_w} {'-' * name_w} {'-' * 19}")
|
|
for ip, mac, name, expires in rows:
|
|
print(f"{ip:<{ip_w}} {mac:<{mac_w}} {name:<{name_w}} {expires}")
|
|
|
|
elif args.json:
|
|
print(json.dumps(records, indent=2))
|
|
|
|
else:
|
|
# Default: machine-readable CSV, raw (unix timestamp) expiry
|
|
writer = csv.DictWriter(sys.stdout, fieldnames=["ip", "mac", "hostname", "expires"])
|
|
writer.writeheader()
|
|
writer.writerows(records)
|
|
finally:
|
|
requests.delete(f"{api_url}/auth", headers=headers, verify=verify)
|
|
|