from ceph.deployment.utils import parse_combined_pem_file
from ceph.utils import is_hex
from ceph.smb import constants as smbconst
+from ceph.smb import network as smbnet
ServiceSpecT = TypeVar('ServiceSpecT', bound='ServiceSpec')
FuncT = TypeVar('FuncT', bound=Callable)
self.validate()
def validate(self) -> None:
- if self.address and self.network:
- raise SpecValidationError('only one of address or network may be given')
- if not (self.address or self.network):
- raise SpecValidationError('one of address or network is required')
- if self.address:
- # verify that address is an address
- try:
- ip_address(self.address)
- except ValueError as err:
- raise SpecValidationError(
- f'Cannot parse address {self.address}'
- ) from err
- # but we internallly store a list of networks
- # this is slight bit of YAGNI violation, but I actually plan on
- # adding IP ranges soon.
- addr = self.network if self.network else self.address
try:
- assert addr
- self._networks = [ip_network(addr)]
+ netval = smbnet.to_network(
+ network=self.network, address=self.address
+ )
+ self._networks = [netval]
except ValueError as err:
- raise SpecValidationError(
- f'Cannot parse network address {addr}'
- ) from err
+ raise SpecValidationError(str(err)) from err
def as_networks(self) -> List[Union[IPv4Network, IPv6Network]]:
"""Return a list of one or more IPv4 or IPv6 network objects."""
--- /dev/null
+# Shared functions for working with networking (addresses, etc) across
+# multiple ceph smb components.
+
+from typing import Optional, Union
+
+import ipaddress
+
+
+IPNetwork = Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
+
+
+def to_network(
+ *,
+ network: Optional[str] = None,
+ address: Optional[str] = None,
+) -> IPNetwork:
+ """Parse and convert an IP address or IP network string into a network
+ object so that calling code only needs to manage network objects (vs
+ seperate address and network objects). The input values may be for either
+ IPv4 or IPv6.
+ If the input value is an address rather than a network the resulting
+ network object will have a prefixlen value equal to the length of the
+ maximum prefixlen - /32 for IPv4 and /128 for IPv6.
+ Only keyword arguments are supported.
+ """
+ if address and network:
+ raise ValueError('only one of address or network may be given')
+ if not (address or network):
+ raise ValueError('one of address or network is required')
+
+ if address:
+ # verify that address is actually an address and doesn't contain a
+ # network value (that would be accepted by ip_network).
+ try:
+ ipaddress.ip_address(address)
+ except ValueError as err:
+ raise ValueError(f'Cannot parse address {address}') from err
+
+ addr = network if network else address
+ try:
+ assert addr
+ nw = ipaddress.ip_network(addr)
+ except ValueError as err:
+ raise ValueError(f'Cannot parse network address {addr}') from err
+ return nw