| import argparse |
| from pathlib import Path |
| from at.dhcp import parse_isc_dhcpd_leases |
| from time import time |
| |
| def list(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument("leases", type=Path, help="leases file") |
| parser.add_argument("--timeout", type=int, default=None, help="timeout in minutes") |
| args = parser.parse_args() |
| |
| with open(args.leases) as f: |
| offset, devices = parse_isc_dhcpd_leases(f) |
| if args.timeout is not None: |
| devices.purge_stale(args.timeout * 60) |
| print("Found devices:") |
| for device in devices._devices.values(): |
| print(device._replace(atime = time() - device.atime)) |
| |
| import grpc |
| from .tracker_pb2 import ClientsRequest |
| from .tracker_pb2_grpc import DhcpTrackerStub |
| |
| def format_mac(raw: bytes) -> str: |
| return ':'.join(f'{b:02x}' for b in raw) |
| |
| def tracker_list(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument("address", default='unix:///tmp/checkinator.sock', nargs='?', help="tracker grpc address") |
| parser.add_argument("--timeout", type=int, default=None, help="timeout in minutes") |
| args = parser.parse_args() |
| with grpc.insecure_channel(args.address) as channel: |
| stub = DhcpTrackerStub(channel) |
| response = stub.GetClients(ClientsRequest()) |
| for client in response.clients: |
| print(format_mac(client.hw_address), client.last_seen, client.ip_address, client.client_hostname ) |