diff --git a/gso/products/product_blocks/router.py b/gso/products/product_blocks/router.py index d0d0fdcfad266a68bca2c1b1a5efb3bfa2f1e809..03be23a9b4f9ee38f1f1d5238d0db6a441871458 100644 --- a/gso/products/product_blocks/router.py +++ b/gso/products/product_blocks/router.py @@ -4,6 +4,7 @@ from typing import Optional from orchestrator.domain.base import ProductBlockModel from orchestrator.types import SubscriptionLifecycle, strEnum +from pydantic import ConstrainedInt from gso.products.product_blocks.site import SiteBlock, SiteBlockInactive, SiteBlockProvisioning @@ -28,13 +29,25 @@ class RouterRole(strEnum): """AMT router.""" +class PortNumber(ConstrainedInt): + """Constrained integer for valid port numbers. + + The range from 49152 to 65535 is marked as ephemeral, and can therefore not be selected for permanent allocation. + """ + + gt = 0 + """The lower bound of the valid port number range.""" + le = 49151 + """As mentioned earlier, the ephemeral port range should not be chosen, and is therefore not available.""" + + class RouterBlockInactive( ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="RouterBlock" ): """A router that's being currently inactive. See {class}`RouterBlock`.""" router_fqdn: Optional[str] = None - router_ts_port: Optional[int] = None + router_ts_port: Optional[PortNumber] = None router_access_via_ts: Optional[bool] = None router_lo_ipv4_address: Optional[ipaddress.IPv4Address] = None router_lo_ipv6_address: Optional[ipaddress.IPv6Address] = None @@ -52,7 +65,7 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy """A router that's being provisioned. See {class}`RouterBlock`.""" router_fqdn: str - router_ts_port: int + router_ts_port: PortNumber router_access_via_ts: Optional[bool] = None router_lo_ipv4_address: Optional[ipaddress.IPv4Address] = None router_lo_ipv6_address: Optional[ipaddress.IPv6Address] = None @@ -71,9 +84,8 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI router_fqdn: str """{term}`FQDN` of a router.""" - router_ts_port: int - """The port of the terminal server that this router is connected to. Used for the same reason as mentioned - previously.""" + router_ts_port: PortNumber + """The port of the terminal server that this router is connected to. Used to provide out of band access.""" router_access_via_ts: bool """Whether this router should be accessed through the terminal server, or through its loopback address.""" router_lo_ipv4_address: ipaddress.IPv4Address diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py index d519f8b8b549e96823570d4398bd5a3abb6d66b2..7db892d6173d65e9f226556203d4da5736cfc32b 100644 --- a/gso/products/product_blocks/site.py +++ b/gso/products/product_blocks/site.py @@ -25,6 +25,7 @@ class SnmpCoordinate(ConstrainedStr): The coordinate must match the format of `1.35`, `-12.3456`, etc. """ + regex = re.compile(r"^-?\d{1,2}\.\d+$") diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py index 3f4624caee8da28919fe200f4edf1b2a1fa57afc..1f7b96c480d0ddf72f28b2fb75a1874c23622cec 100644 --- a/gso/workflows/router/create_router.py +++ b/gso/workflows/router/create_router.py @@ -1,5 +1,5 @@ -import ipaddress import re +from typing import Optional # noinspection PyProtectedMember from orchestrator.forms import FormPage @@ -9,6 +9,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID from orchestrator.workflow import StepList, done, init, step, workflow from orchestrator.workflows.steps import resync, set_status, store_process_subscription from orchestrator.workflows.utils import wrap_create_initial_input_form +from products.product_blocks.router import PortNumber from gso.products.product_blocks import router as router_pb from gso.products.product_types import router @@ -16,10 +17,10 @@ from gso.products.product_types.router import RouterInactive, RouterProvisioning from gso.products.product_types.site import Site from gso.services import ipam, provisioning_proxy, subscriptions from gso.services.provisioning_proxy import pp_interaction -from gso.workflows.utils import customer_selector +from gso.workflows.utils import customer_selector, iso_from_ipv4 -def site_selector() -> Choice: +def _site_selector() -> Choice: site_subscriptions = {} for site_id, site_description in subscriptions.get_active_site_subscriptions( fields=["subscription_id", "description"] @@ -36,12 +37,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator: title = product_name customer: customer_selector() # type: ignore - router_site: site_selector() # type: ignore + router_site: _site_selector() # type: ignore hostname: str - ts_port: int + ts_port: PortNumber router_vendor: router_pb.RouterVendor router_role: router_pb.RouterRole - is_ias_connected: bool + is_ias_connected: Optional[bool] user_input = yield CreateRouterForm @@ -58,13 +59,6 @@ def create_subscription(product: UUIDstr, customer: UUIDstr) -> State: } -def iso_from_ipv4(ipv4_address: ipaddress.IPv4Address) -> str: - padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")] - joined_octets = "".join(padded_octets) - re_split = ".".join(re.findall("....", joined_octets)) - return ".".join(["49.51e5.0001", re_split, "00"]) - - @step("Get information from IPAM") def get_info_from_ipam(subscription: RouterProvisioning, is_ias_connected: bool) -> State: lo0_alias = re.sub(".geant.net", "", subscription.router.router_fqdn) @@ -93,7 +87,7 @@ def get_info_from_ipam(subscription: RouterProvisioning, is_ias_connected: bool) def initialize_subscription( subscription: router.RouterInactive, hostname: str, - ts_port: int, + ts_port: PortNumber, router_vendor: router_pb.RouterVendor, router_site: str, router_role: router_pb.RouterRole, diff --git a/gso/workflows/site/create_site.py b/gso/workflows/site/create_site.py index 94237b300528c23cb91512c3ae573fa7168f1b65..b16cf5b42eaf384f0cd9e960eb566437f11a5d36 100644 --- a/gso/workflows/site/create_site.py +++ b/gso/workflows/site/create_site.py @@ -1,13 +1,15 @@ import ipaddress +from typing import NoReturn +import pycountry from orchestrator.forms import FormPage from orchestrator.targets import Target from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr from orchestrator.workflow import StepList, done, init, step, workflow from orchestrator.workflows.steps import resync, set_status, store_process_subscription from orchestrator.workflows.utils import wrap_create_initial_input_form +from products.product_blocks.site import SnmpCoordinate from pydantic import validator -import pycountry from gso.products.product_blocks import site as site_pb from gso.products.product_types import site @@ -32,7 +34,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator: site_ts_address: str @validator("site_country_code", allow_reuse=True) - def country_code_must_exist(cls, country_code): + def country_code_must_exist(cls, country_code: str) -> str | NoReturn: try: _ = pycountry.countries.lookup(country_code) # Lookup succeeded, the country code is valid. @@ -42,23 +44,23 @@ def initial_input_form_generator(product_name: str) -> FormGenerator: raise ValueError("Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format.") @validator("site_latitude", allow_reuse=True) - def latitude_must_be_valid(cls, latitude): + def latitude_must_be_valid(cls, latitude: str) -> str | NoReturn: if -90 <= float(latitude) <= 90: # Check whether the value is a valid degree of latitude. return latitude - else: - raise ValueError("Entered latitude is not a valid value, must be between -90.0° and 90.0°.") + + raise ValueError("Entered latitude is not a valid value, must be between -90.0° and 90.0°.") @validator("site_longitude", allow_reuse=True) - def longitude_must_be_valid(cls, longitude): + def longitude_must_be_valid(cls, longitude: str) -> str | NoReturn: if -180 <= float(longitude) <= 180: # Check whether the value is a valid degree of longitude. return longitude - else: - raise ValueError("Entered longitude is not a valid value, must be between -180.0° and 180.0°.") + + raise ValueError("Entered longitude is not a valid value, must be between -180.0° and 180.0°.") @validator("site_ts_address", allow_reuse=True) - def ts_address_must_be_valid(cls, ts_address): + def ts_address_must_be_valid(cls, ts_address: str) -> str | NoReturn: try: ipaddress.ip_address(ts_address) # The address is valid @@ -88,8 +90,8 @@ def initialize_subscription( site_city: str, site_country: str, site_country_code: str, - site_latitude: float, - site_longitude: float, + site_latitude: SnmpCoordinate, + site_longitude: SnmpCoordinate, site_bgp_community_id: int, site_internal_id: int, site_ts_address: str, diff --git a/gso/workflows/utils.py b/gso/workflows/utils.py index d2c631377667abf1d31b9448c164f2dff96c1eaa..9b553e2e17c71e231c1e1d99043e8bac94e6769a 100644 --- a/gso/workflows/utils.py +++ b/gso/workflows/utils.py @@ -1,3 +1,6 @@ +import re +from ipaddress import IPv4Address + from orchestrator.forms.validators import Choice from gso.services.crm import all_customers @@ -9,3 +12,15 @@ def customer_selector() -> Choice: customers[customer["id"]] = customer["name"] return Choice("Select a customer", zip(customers.keys(), customers.items())) # type: ignore + + +def iso_from_ipv4(ipv4_address: IPv4Address) -> str: + """Calculate an ISO address, based on an IPv4 address. + + :param IPv4Address ipv4_address: The address that is to be converted + :returns: An ISO-formatted address. + """ + padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")] + joined_octets = "".join(padded_octets) + re_split = ".".join(re.findall("....", joined_octets)) + return ".".join(["49.51e5.0001", re_split, "00"])