Source code for dhcpkit.ipv6.server.extensions.timing_limits

"""
Handlers that limit the t1/t2 values in replies
"""

from typing import Iterable, List, Optional, Union

from dhcpkit.ipv6 import INFINITY
from dhcpkit.ipv6.extensions.prefix_delegation import IAPDOption, IAPrefixOption
from dhcpkit.ipv6.options import IAAddressOption, IANAOption, Option
from dhcpkit.ipv6.server.handlers import Handler
from dhcpkit.ipv6.server.transaction_bundle import TransactionBundle


[docs]class TimingLimitsHandler(Handler): """ A handler that limits the t1/t2 values in an option """ def __init__(self, min_t1: int = 0, max_t1: int = INFINITY, factor_t1: Optional[float] = 0.5, min_t2: int = 0, max_t2: int = INFINITY, factor_t2: Optional[float] = 0.8): super().__init__() # These are the outer limits self.min_t1 = max(0, min_t1) self.max_t2 = min(max_t2, INFINITY) # If t2 has a max then t1's max must be at least as small self.max_t1 = min(max_t1, self.max_t2) # If t1 has a min then t2's min must be at least as large self.min_t2 = max(self.min_t1, min_t2) # Store the factors to auto-calculate t1/t2 based on the shortest preferred lifetime if factor_t1 is not None: self.factor_t1 = min(max(0.0, factor_t1), 1) else: self.factor_t1 = None if factor_t2 is not None: self.factor_t2 = min(max(0.0, factor_t2), 1) else: self.factor_t2 = None # Do some basic checks for impossible values if self.min_t1 > self.max_t2: raise ValueError("min_t1 must be smaller than max_t2") if self.factor_t1 and self.factor_t2 and self.factor_t1 > self.factor_t2: raise ValueError("factor_t1 must be smaller than factor_t2") def __str__(self): return "{} with t1={},{},{} t2={},{},{}".format(self.__class__.__name__, self.min_t1, self.max_t1, self.factor_t1, self.min_t2, self.max_t2, self.factor_t2)
[docs] @staticmethod def filter_options(options: Iterable[Option]) -> List[Union[IANAOption, IAPDOption]]: """ Extract the options that we want to set the t1/t2 values of. :param options: The options in the response message :returns: The relevant options of the response message :rtype: list[IANAOption] """ return []
[docs] @staticmethod def extract_preferred_lifetime(option: Option) -> Optional[int]: """ Extract the preferred lifetime from the given (sub)option. Returns None if this option doesn't contain a preferred lifetime. :param option: The option to extract the preferred lifetime from :returns: The preferred lifetime, if any """ return None
[docs] def handle(self, bundle: TransactionBundle): """ Make sure the T1/T2 values are within the set limits. :param bundle: The transaction bundle """ # Make a list of IAIDs in the response for option in self.filter_options(bundle.response.options): # Find the shortest preferred lifetime shortest_preferred = INFINITY + 1 for suboption in option.options: preferred_lifetime = self.extract_preferred_lifetime(suboption) if preferred_lifetime is not None: shortest_preferred = min(shortest_preferred, preferred_lifetime) # Don't mess with the timers if there are no addresses if shortest_preferred > INFINITY: continue # Calculate t1 if option.t1 == 0 and self.factor_t1 is not None: if shortest_preferred == INFINITY: option.t1 = INFINITY else: option.t1 = int(shortest_preferred * self.factor_t1) # Calculate t2 if option.t2 == 0 and self.factor_t2 is not None: if shortest_preferred == INFINITY: option.t2 = INFINITY else: option.t2 = int(shortest_preferred * self.factor_t2) # Now limit to the min/max bounds, making sure that t1 is not larger than t2 # This will ignore the given boundaries if they conflict with shortest_preferred lifetime option.t2 = min(max(self.min_t2, option.t2), self.max_t2, shortest_preferred) option.t1 = min(max(self.min_t1, option.t1), self.max_t1, option.t2)
[docs]class IANATimingLimitsHandler(TimingLimitsHandler): """ A handler that limits the t1/t2 values in an IANAOption """
[docs] @staticmethod def filter_options(options: Iterable[Option]) -> List[IANAOption]: """ Extract the IANAOptions that we want to set the t1/t2 values of. :param options: The options in the response message :returns: The relevant options of the response message :rtype: list[IANAOption] """ # noinspection PyTypeChecker return [option for option in options if isinstance(option, IANAOption)]
[docs] @staticmethod def extract_preferred_lifetime(option: Option) -> Optional[int]: """ Extract the preferred lifetime from the given (sub)option. Returns None if this option doesn't contain a preferred lifetime. :param option: The option to extract the preferred lifetime from :returns: The preferred lifetime, if any """ if isinstance(option, IAAddressOption): return option.preferred_lifetime else: return None
[docs]class IAPDTimingLimitsHandler(TimingLimitsHandler): """ A handler that limits the t1/t2 values in an IANAOption """
[docs] @staticmethod def filter_options(options: Iterable[Option]) -> List[IAPDOption]: """ Extract the IAPDOptions that we want to set the t1/t2 values of. :param options: The options in the response message :returns: The relevant options of the response message :rtype: list[IAPDOption] """ # noinspection PyTypeChecker return [option for option in options if isinstance(option, IAPDOption)]
[docs] @staticmethod def extract_preferred_lifetime(option: Option) -> Optional[int]: """ Extract the preferred lifetime from the given (sub)option. Returns None if this option doesn't contain a preferred lifetime. :param option: The option to extract the preferred lifetime from :returns: The preferred lifetime, if any """ if isinstance(option, IAPrefixOption): return option.preferred_lifetime else: return None