diff options
Diffstat (limited to 'src/op_mode/show_dhcpv6.py')
-rwxr-xr-x | src/op_mode/show_dhcpv6.py | 220 |
1 files changed, 0 insertions, 220 deletions
diff --git a/src/op_mode/show_dhcpv6.py b/src/op_mode/show_dhcpv6.py deleted file mode 100755 index b34b730e6..000000000 --- a/src/op_mode/show_dhcpv6.py +++ /dev/null @@ -1,220 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2018-2021 VyOS maintainers and contributors -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 or later as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see <http://www.gnu.org/licenses/>. -# -# TODO: merge with show_dhcp.py - -from json import dumps -from argparse import ArgumentParser -from ipaddress import ip_address -from tabulate import tabulate -from sys import exit -from collections import OrderedDict -from datetime import datetime - -from isc_dhcp_leases import Lease, IscDhcpLeases - -from vyos.base import Warning -from vyos.config import Config -from vyos.util import is_systemd_service_running - -lease_file = "/config/dhcpdv6.leases" -pool_key = "shared-networkname" - -lease_display_fields = OrderedDict() -lease_display_fields['ip'] = 'IPv6 address' -lease_display_fields['state'] = 'State' -lease_display_fields['last_comm'] = 'Last communication' -lease_display_fields['expires'] = 'Lease expiration' -lease_display_fields['remaining'] = 'Remaining' -lease_display_fields['type'] = 'Type' -lease_display_fields['pool'] = 'Pool' -lease_display_fields['iaid_duid'] = 'IAID_DUID' - -lease_valid_states = ['all', 'active', 'free', 'expired', 'released', 'abandoned', 'reset', 'backup'] - -def in_pool(lease, pool): - if pool_key in lease.sets: - if lease.sets[pool_key] == pool: - return True - - return False - -def format_hex_string(in_str): - out_str = "" - - # if input is divisible by 2, add : every 2 chars - if len(in_str) > 0 and len(in_str) % 2 == 0: - out_str = ':'.join(a+b for a,b in zip(in_str[::2], in_str[1::2])) - else: - out_str = in_str - - return out_str - -def utc_to_local(utc_dt): - return datetime.fromtimestamp((utc_dt - datetime(1970,1,1)).total_seconds()) - -def get_lease_data(lease): - data = {} - - # isc-dhcp lease times are in UTC so we need to convert them to local time to display - try: - data["expires"] = utc_to_local(lease.end).strftime("%Y/%m/%d %H:%M:%S") - except: - data["expires"] = "" - - try: - data["last_comm"] = utc_to_local(lease.last_communication).strftime("%Y/%m/%d %H:%M:%S") - except: - data["last_comm"] = "" - - try: - data["remaining"] = lease.end - datetime.utcnow() - # negative timedelta prints wrong so bypass it - if (data["remaining"].days >= 0): - # substraction gives us a timedelta object which can't be formatted with strftime - # so we use str(), split gets rid of the microseconds - data["remaining"] = str(data["remaining"]).split('.')[0] - else: - data["remaining"] = "" - except: - data["remaining"] = "" - - # isc-dhcp records lease declarations as ia_{na|ta|pd} IAID_DUID {...} - # where IAID_DUID is the combined IAID and DUID - data["iaid_duid"] = format_hex_string(lease.host_identifier_string) - - lease_types_long = {"na": "non-temporary", "ta": "temporary", "pd": "prefix delegation"} - data["type"] = lease_types_long[lease.type] - - data["state"] = lease.binding_state - data["ip"] = lease.ip - - try: - data["pool"] = lease.sets[pool_key] - except: - data["pool"] = "" - - return data - -def get_leases(config, leases, state, pool=None, sort='ip'): - leases = IscDhcpLeases(lease_file).get() - - # filter leases by state - if 'all' not in state: - leases = list(filter(lambda x: x.binding_state in state, leases)) - - # filter leases by pool name - if pool is not None: - if config.exists_effective("service dhcp-server shared-network-name {0}".format(pool)): - leases = list(filter(lambda x: in_pool(x, pool), leases)) - else: - print("Pool {0} does not exist.".format(pool)) - exit(0) - - # should maybe filter all state=active by lease.valid here? - - # sort by last_comm time to dedupe (newest lease overrides older) - leases = sorted(leases, key = lambda lease: lease.last_communication) - - # dedupe by converting to dict - leases_dict = {} - for lease in leases: - # dedupe by IP - leases_dict[lease.ip] = lease - - # convert the lease data - leases = list(map(get_lease_data, leases_dict.values())) - - # apply output/display sort - if sort == 'ip': - leases = sorted(leases, key = lambda k: int(ip_address(k['ip'].split('/')[0]))) - else: - leases = sorted(leases, key = lambda k: k[sort]) - - return leases - -def show_leases(leases): - lease_list = [] - for l in leases: - lease_list_params = [] - for k in lease_display_fields.keys(): - lease_list_params.append(l[k]) - lease_list.append(lease_list_params) - - output = tabulate(lease_list, lease_display_fields.values()) - - print(output) - -if __name__ == '__main__': - parser = ArgumentParser() - - group = parser.add_mutually_exclusive_group() - group.add_argument("-l", "--leases", action="store_true", help="Show DHCPv6 leases") - group.add_argument("-s", "--statistics", action="store_true", help="Show DHCPv6 statistics") - group.add_argument("--allowed", type=str, choices=["pool", "sort", "state"], help="Show allowed values for argument") - - parser.add_argument("-p", "--pool", type=str, help="Show lease for specific pool") - parser.add_argument("-S", "--sort", type=str, default='ip', help="Sort by") - parser.add_argument("-t", "--state", type=str, nargs="+", default=["active"], help="Lease state to show (can specify multiple with spaces)") - parser.add_argument("-j", "--json", action="store_true", default=False, help="Produce JSON output") - - args = parser.parse_args() - - conf = Config() - - if args.allowed == 'pool': - if conf.exists_effective('service dhcpv6-server'): - print(' '.join(conf.list_effective_nodes("service dhcpv6-server shared-network-name"))) - exit(0) - elif args.allowed == 'sort': - print(' '.join(lease_display_fields.keys())) - exit(0) - elif args.allowed == 'state': - print(' '.join(lease_valid_states)) - exit(0) - elif args.allowed: - parser.print_help() - exit(1) - - if args.sort not in lease_display_fields.keys(): - print(f'Invalid sort key, choose from: {list(lease_display_fields.keys())}') - exit(0) - - if not set(args.state) < set(lease_valid_states): - print(f'Invalid lease state, choose from: {lease_valid_states}') - exit(0) - - # Do nothing if service is not configured - if not conf.exists_effective('service dhcpv6-server'): - print("DHCPv6 service is not configured") - exit(0) - - # if dhcp server is down, inactive leases may still be shown as active, so warn the user. - if not is_systemd_service_running('isc-dhcp-server6.service'): - Warning('DHCPv6 server is configured but not started. Data may be stale.') - - if args.leases: - leases = get_leases(conf, lease_file, args.state, args.pool, args.sort) - - if args.json: - print(dumps(leases, indent=4)) - else: - show_leases(leases) - elif args.statistics: - print("DHCPv6 statistics option is not available") - else: - parser.print_help() - exit(1) |