diff options
Diffstat (limited to 'src/op_mode')
| -rwxr-xr-x | src/op_mode/dhcp.py | 71 | ||||
| -rwxr-xr-x | src/op_mode/show_dhcp.py | 260 | ||||
| -rwxr-xr-x | src/op_mode/show_dhcpv6.py | 220 |
3 files changed, 42 insertions, 509 deletions
diff --git a/src/op_mode/dhcp.py b/src/op_mode/dhcp.py index 07e9b7d6c..b9e6e7bc9 100755 --- a/src/op_mode/dhcp.py +++ b/src/op_mode/dhcp.py @@ -15,13 +15,14 @@ # along with this program. If not, see <http://www.gnu.org/licenses/>. import sys -from ipaddress import ip_address import typing from datetime import datetime -from sys import exit -from tabulate import tabulate +from ipaddress import ip_address from isc_dhcp_leases import IscDhcpLeases +from tabulate import tabulate + +import vyos.opmode from vyos.base import Warning from vyos.configquery import ConfigTreeQuery @@ -30,19 +31,10 @@ from vyos.util import cmd from vyos.util import dict_search from vyos.util import is_systemd_service_running -import vyos.opmode - - config = ConfigTreeQuery() -pool_key = "shared-networkname" - - -def _in_pool(lease, pool): - if pool_key in lease.sets: - if lease.sets[pool_key] == pool: - return True - return False - +lease_valid_states = ['all', 'active', 'free', 'expired', 'released', 'abandoned', 'reset', 'backup'] +sort_valid_inet = ['end', 'mac', 'hostname', 'ip', 'pool', 'remaining', 'start', 'state'] +sort_valid_inet6 = ['end', 'iaid_duid', 'ip', 'last_communication', 'pool', 'remaining', 'state', 'type'] def _utc_to_local(utc_dt): return datetime.fromtimestamp((datetime.fromtimestamp(utc_dt) - datetime(1970, 1, 1)).total_seconds()) @@ -71,7 +63,7 @@ def _find_list_of_dict_index(lst, key='ip', value='') -> int: return idx -def _get_raw_server_leases(family, pool=None) -> list: +def _get_raw_server_leases(family='inet', pool=None, sorted=None, state=[]) -> list: """ Get DHCP server leases :return list @@ -79,9 +71,12 @@ def _get_raw_server_leases(family, pool=None) -> list: lease_file = '/config/dhcpdv6.leases' if family == 'inet6' else '/config/dhcpd.leases' data = [] leases = IscDhcpLeases(lease_file).get() - if pool is not None: - if config.exists(f'service dhcp-server shared-network-name {pool}'): - leases = list(filter(lambda x: _in_pool(x, pool), leases)) + + if pool is None: + pool = _get_dhcp_pools(family=family) + else: + pool = [pool] + for lease in leases: data_lease = {} data_lease['ip'] = lease.ip @@ -90,7 +85,7 @@ def _get_raw_server_leases(family, pool=None) -> list: data_lease['end'] = lease.end.timestamp() if family == 'inet': - data_lease['hardware'] = lease.ethernet + data_lease['mac'] = lease.ethernet data_lease['start'] = lease.start.timestamp() data_lease['hostname'] = lease.hostname @@ -110,8 +105,9 @@ def _get_raw_server_leases(family, pool=None) -> list: data_lease['remaining'] = '' # Do not add old leases - if data_lease['remaining'] != '': - data.append(data_lease) + if data_lease['remaining'] != '' and data_lease['pool'] in pool: + if not state or data_lease['state'] in state: + data.append(data_lease) # deduplicate checked = [] @@ -123,15 +119,20 @@ def _get_raw_server_leases(family, pool=None) -> list: idx = _find_list_of_dict_index(data, key='ip', value=addr) data.pop(idx) + if sorted: + if sorted == 'ip': + data.sort(key = lambda x:ip_address(x['ip'])) + else: + data.sort(key = lambda x:x[sorted]) return data -def _get_formatted_server_leases(raw_data, family): +def _get_formatted_server_leases(raw_data, family='inet'): data_entries = [] if family == 'inet': for lease in raw_data: ipaddr = lease.get('ip') - hw_addr = lease.get('hardware') + hw_addr = lease.get('mac') state = lease.get('state') start = lease.get('start') start = _utc_to_local(start).strftime('%Y/%m/%d %H:%M:%S') @@ -142,7 +143,7 @@ def _get_formatted_server_leases(raw_data, family): hostname = lease.get('hostname') data_entries.append([ipaddr, hw_addr, state, start, end, remain, pool, hostname]) - headers = ['IP Address', 'Hardware address', 'State', 'Lease start', 'Lease expiration', 'Remaining', 'Pool', + headers = ['IP Address', 'MAC address', 'State', 'Lease start', 'Lease expiration', 'Remaining', 'Pool', 'Hostname'] if family == 'inet6': @@ -256,16 +257,28 @@ def show_pool_statistics(raw: bool, family: str, pool: typing.Optional[str]): @_verify -def show_server_leases(raw: bool, family: str): +def show_server_leases(raw: bool, family: str, pool: typing.Optional[str], + sorted: typing.Optional[str], state: typing.Optional[str]): # 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-server.service'): Warning('DHCP server is configured but not started. Data may be stale.') - leases = _get_raw_server_leases(family) + v = 'v6' if family == 'inet6' else '' + if pool and pool not in _get_dhcp_pools(family=family): + raise vyos.opmode.IncorrectValue(f'DHCP{v} pool "{pool}" does not exist!') + + if state and state not in lease_valid_states: + raise vyos.opmode.IncorrectValue(f'DHCP{v} state "{state}" is invalid!') + + sort_valid = sort_valid_inet6 if family == 'inet6' else sort_valid_inet + if sorted and sorted not in sort_valid: + raise vyos.opmode.IncorrectValue(f'DHCP{v} sort "{sorted}" is invalid!') + + lease_data = _get_raw_server_leases(family=family, pool=pool, sorted=sorted, state=state) if raw: - return leases + return lease_data else: - return _get_formatted_server_leases(leases, family) + return _get_formatted_server_leases(lease_data, family=family) if __name__ == '__main__': diff --git a/src/op_mode/show_dhcp.py b/src/op_mode/show_dhcp.py deleted file mode 100755 index 4b1758eea..000000000 --- a/src/op_mode/show_dhcp.py +++ /dev/null @@ -1,260 +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_dhcpv6.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/dhcpd.leases" -pool_key = "shared-networkname" - -lease_display_fields = OrderedDict() -lease_display_fields['ip'] = 'IP address' -lease_display_fields['hardware_address'] = 'Hardware address' -lease_display_fields['state'] = 'State' -lease_display_fields['start'] = 'Lease start' -lease_display_fields['end'] = 'Lease expiration' -lease_display_fields['remaining'] = 'Remaining' -lease_display_fields['pool'] = 'Pool' -lease_display_fields['hostname'] = 'Hostname' - -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 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["start"] = utc_to_local(lease.start).strftime("%Y/%m/%d %H:%M:%S") - except: - data["start"] = "" - - try: - data["end"] = utc_to_local(lease.end).strftime("%Y/%m/%d %H:%M:%S") - except: - data["end"] = "" - - 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"] = "" - - # currently not used but might come in handy - # todo: parse into datetime string - for prop in ['tstp', 'tsfp', 'atsfp', 'cltt']: - if prop in lease.data: - data[prop] = lease.data[prop] - else: - data[prop] = '' - - data["hardware_address"] = lease.ethernet - data["hostname"] = lease.hostname - - 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'): - # get leases from file - 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 start time to dedupe (newest lease overrides older) - leases = sorted(leases, key = lambda lease: lease.start) - - # 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 lease: int(ip_address(lease['ip']))) - else: - leases = sorted(leases, key = lambda lease: lease[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) - -def get_pool_size(config, pool): - size = 0 - subnets = config.list_effective_nodes("service dhcp-server shared-network-name {0} subnet".format(pool)) - for s in subnets: - ranges = config.list_effective_nodes("service dhcp-server shared-network-name {0} subnet {1} range".format(pool, s)) - for r in ranges: - start = config.return_effective_value("service dhcp-server shared-network-name {0} subnet {1} range {2} start".format(pool, s, r)) - stop = config.return_effective_value("service dhcp-server shared-network-name {0} subnet {1} range {2} stop".format(pool, s, r)) - - # Add +1 because both range boundaries are inclusive - size += int(ip_address(stop)) - int(ip_address(start)) + 1 - - return size - -def show_pool_stats(stats): - headers = ["Pool", "Size", "Leases", "Available", "Usage"] - output = tabulate(stats, headers) - - print(output) - -if __name__ == '__main__': - parser = ArgumentParser() - - group = parser.add_mutually_exclusive_group() - group.add_argument("-l", "--leases", action="store_true", help="Show DHCP leases") - group.add_argument("-s", "--statistics", action="store_true", help="Show DHCP statistics") - group.add_argument("--allowed", type=str, choices=["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 == '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 dhcp-server'): - print("DHCP 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-server.service'): - Warning('DHCP 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: - pools = [] - - # Get relevant pools - if args.pool: - pools = [args.pool] - else: - pools = conf.list_effective_nodes("service dhcp-server shared-network-name") - - # Get pool usage stats - stats = [] - for p in pools: - size = get_pool_size(conf, p) - leases = len(get_leases(conf, lease_file, state='active', pool=p)) - - use_percentage = round(leases / size * 100) if size != 0 else 0 - - if args.json: - pool_stats = {"pool": p, "size": size, "leases": leases, - "available": (size - leases), "percentage": use_percentage} - else: - # For tabulate - pool_stats = [p, size, leases, size - leases, "{0}%".format(use_percentage)] - stats.append(pool_stats) - - # Print stats - if args.json: - print(dumps(stats, indent=4)) - else: - show_pool_stats(stats) - - else: - parser.print_help() - exit(1) 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) |