diff --git a/catalystwan/api/configuration_groups/parcel.py b/catalystwan/api/configuration_groups/parcel.py index 792088226..4490f4bdb 100644 --- a/catalystwan/api/configuration_groups/parcel.py +++ b/catalystwan/api/configuration_groups/parcel.py @@ -1,13 +1,23 @@ from enum import Enum from typing import Any, Dict, Generic, Literal, Optional, TypeVar, get_origin -from pydantic import AliasPath, BaseModel, ConfigDict, Field, PrivateAttr, model_serializer +from pydantic import ( + AliasPath, + BaseModel, + ConfigDict, + Field, + PrivateAttr, + SerializerFunctionWrapHandler, + model_serializer, +) T = TypeVar("T") class _ParcelBase(BaseModel): - model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True, populate_by_name=True) + model_config = ConfigDict( + extra="allow", arbitrary_types_allowed=True, populate_by_name=True, # json_schema_mode_override="validation" + ) parcel_name: str = Field( min_length=1, max_length=128, @@ -21,11 +31,18 @@ class _ParcelBase(BaseModel): validation_alias="description", description="Set the parcel description", ) - # data: Optional[Any] = None _parcel_data_key: str = PrivateAttr(default="data") - @model_serializer(mode="wrap", when_used="json") - def envelope_parcel_data(self, handler) -> Dict[str, Any]: + @model_serializer(mode="wrap") + def envelope_parcel_data(self, handler: SerializerFunctionWrapHandler) -> Dict[str, Any]: + """ + serializes model fields with respect to field validation_alias, + sub-classing parcel fields can be defined like following: + >>> entries: List[SecurityZoneListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) + + "data" is default _parcel_data_key which must match validation_alias prefix, + this attribute can be overriden in sub-class when needed + """ model_dict = handler(self) model_dict[self._parcel_data_key] = {} remove_keys = [] diff --git a/catalystwan/endpoints/__init__.py b/catalystwan/endpoints/__init__.py index a4e1da9a4..2dca31aec 100644 --- a/catalystwan/endpoints/__init__.py +++ b/catalystwan/endpoints/__init__.py @@ -50,6 +50,7 @@ Sequence, Set, Tuple, + Type, TypeVar, Union, runtime_checkable, @@ -114,6 +115,47 @@ def json(cls) -> TypeSpecifier: def model_union(cls, models: Sequence[type]) -> TypeSpecifier: return TypeSpecifier(present=True, payload_union_model_types=models) + @classmethod + def resolve_nested_base_model_unions( + cls, annotation: Any, models_types: List[Union[Type[BaseModelV1], Type[BaseModelV2]]] + ) -> List[Union[Type[BaseModelV1], Type[BaseModelV2]]]: + type_origin = get_origin(annotation) + if isclass(annotation): + try: + if issubclass(annotation, (BaseModelV1, BaseModelV2)): + return [annotation] + raise APIEndpointError(f"Expected: {PayloadType}") + except TypeError: + raise APIEndpointError(f"Expected: {PayloadType}") + # Check if Annnotated[Union[PayloadModelType, ...]], only unions of pydantic models allowed + elif type_origin == Annotated: + if annotated_origin := get_args(annotation): + if (len(annotated_origin) >= 1) and get_origin(annotated_origin[0]) == Union: + type_args = get_args(annotated_origin[0]) + if all(isclass(t) for t in type_args) and all( + issubclass(t, (BaseModelV1, BaseModelV2)) for t in type_args + ): + models_types.extend(list(type_args)) + return models_types + else: + non_models = [t for t in type_args if not isclass(t)] + for non_model in non_models: + models_types.extend(cls.resolve_nested_base_model_unions(non_model, models_types)) + return models_types + + # Check if Union[PayloadModelType, ...], only unions of pydantic models allowed + elif type_origin == Union: + type_args = get_args(annotation) + if all(isclass(t) for t in type_args) and all(issubclass(t, (BaseModelV1, BaseModelV2)) for t in type_args): + models_types.extend(list(type_args)) + return models_types + else: + non_models = [t for t in type_args if not isclass(t)] + for non_model in non_models: + models_types.extend(cls.resolve_nested_base_model_unions(non_model, models_types)) + return models_types + raise APIEndpointError(f"Expected: {PayloadType}") + @dataclass class APIEndpointRequestMeta: @@ -451,27 +493,10 @@ def specify_payload_type(self) -> TypeSpecifier: and issubclass(type_args[0], (BaseModelV1, BaseModelV2)) ): return TypeSpecifier(True, type_origin, type_args[0], None, False, is_optional) - # Check if Annnotated[Union[PayloadModelType, ...]], only unions of pydantic models allowed - elif type_origin == Annotated: - if annotated_origin := get_args(annotation): - if (len(annotated_origin) >= 1) and get_origin(annotated_origin[0]) == Union: - if ( - (type_args := get_args(annotated_origin[0])) - and all(isclass(t) for t in type_args) - and all(issubclass(t, (BaseModelV1, BaseModelV2)) for t in type_args) - ): - return TypeSpecifier.model_union(models=list(type_args)) - # Check if Union[PayloadModelType, ...], only unions of pydantic models allowed - elif type_origin == Union: - if ( - (type_args := get_args(annotation)) - and all(isclass(t) for t in type_args) - and all(issubclass(t, (BaseModelV1, BaseModelV2)) for t in type_args) - ): - return TypeSpecifier.model_union(models=list(type_args)) - raise APIEndpointError(f"Expected: {PayloadType} but found payload {annotation}") - else: - raise APIEndpointError(f"Expected: {PayloadType} but found payload {annotation}") + else: + models = TypeSpecifier.resolve_nested_base_model_unions(annotation, []) + return TypeSpecifier.model_union(models) + raise APIEndpointError(f"'payload' param must be annotated with supported type: {PayloadType}") def check_params(self): """Checks params in decorated method definition diff --git a/catalystwan/models/configuration/config_migration.py b/catalystwan/models/configuration/config_migration.py index d29db50a2..36cf8673f 100644 --- a/catalystwan/models/configuration/config_migration.py +++ b/catalystwan/models/configuration/config_migration.py @@ -1,11 +1,13 @@ -from typing import List +from typing import List, Union from pydantic import BaseModel, ConfigDict, Field +from typing_extensions import Annotated -from catalystwan.api.configuration_groups.parcel import _ParcelBase from catalystwan.api.template_api import DeviceTemplateInformation, FeatureTemplateInformation from catalystwan.endpoints.configuration_group import ConfigGroup from catalystwan.models.configuration.feature_profile.common import FeatureProfileCreationPayload +from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel +from catalystwan.models.configuration.feature_profile.sdwan.system import AnySystemParcel from catalystwan.models.policy import ( AnyPolicyDefinition, AnyPolicyList, @@ -14,6 +16,14 @@ SecurityPolicy, ) +AnyParcel = Annotated[ + Union[ + AnySystemParcel, + AnyPolicyObjectParcel, + ], + Field(discriminator="type_"), +] + class UX1Policies(BaseModel): model_config = ConfigDict(populate_by_name=True) @@ -58,6 +68,6 @@ class UX2Config(BaseModel): feature_profiles: List[FeatureProfileCreationPayload] = Field( default=[], serialization_alias="featureProfiles", validation_alias="featureProfiles" ) - profile_parcels: List[_ParcelBase] = Field( + profile_parcels: List[AnyParcel] = Field( default=[], serialization_alias="profileParcels", validation_alias="profileParcels" ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/__init__.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/__init__.py index 1ac7c2da8..8a07d1fb7 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/__init__.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/__init__.py @@ -32,35 +32,42 @@ from .security.url import BaseURLListEntry, URLAllowParcel, URLBlockParcel from .security.zone import SecurityZoneListEntry, SecurityZoneListParcel +AnyURLParcel = Annotated[ + Union[ + URLAllowParcel, + URLBlockParcel, + ], + Field(discriminator="parcel_type"), +] + AnyPolicyObjectParcel = Annotated[ Union[ - AppProbeParcel, + AnyURLParcel, ApplicationListParcel, + AppProbeParcel, ColorParcel, DataPrefixParcel, ExpandedCommunityParcel, FowardingClassParcel, + FQDNDomainParcel, + GeoLocationListParcel, + IPSSignatureParcel, IPv6DataPrefixParcel, IPv6PrefixListParcel, - PrefixListParcel, + LocalDomainParcel, PolicierParcel, PreferredColorGroupParcel, - SLAClassParcel, - TlocParcel, - StandardCommunityParcel, - LocalDomainParcel, - FQDNDomainParcel, - IPSSignatureParcel, - URLAllowParcel, - URLBlockParcel, - SecurityPortParcel, + PrefixListParcel, ProtocolListParcel, - GeoLocationListParcel, - SecurityZoneListParcel, SecurityApplicationListParcel, SecurityDataPrefixParcel, + SecurityPortParcel, + SecurityZoneListParcel, + SLAClassParcel, + StandardCommunityParcel, + TlocParcel, ], - Field(discriminator="type"), + Field(discriminator="type_"), ] POLICY_OBJECT_PAYLOAD_ENDPOINT_MAPPING: Mapping[type, str] = { diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/app_probe.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/app_probe.py index 2a82cd91c..2b1182860 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/app_probe.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/app_probe.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field, field_validator @@ -26,6 +26,7 @@ class AppProbeEntry(BaseModel): class AppProbeParcel(_ParcelBase): + type_: Literal["app-probe"] = Field(default="app-probe", exclude=True) entries: List[AppProbeEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_fowarding_class(self, forwarding_class_name: str): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/application_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/application_list.py index eeda66374..376c0263c 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/application_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/application_list.py @@ -1,4 +1,4 @@ -from typing import List, Union +from typing import List, Literal, Union from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -16,6 +16,7 @@ class ApplicationFamilyListEntry(BaseModel): class ApplicationListParcel(_ParcelBase): + type_: Literal["app-list"] = Field(default="app-list", exclude=True) entries: List[Union[ApplicationListEntry, ApplicationFamilyListEntry]] = Field( default=[], validation_alias=AliasPath("data", "entries") ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/color_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/color_list.py index 4917deea8..7ef55f793 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/color_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/color_list.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, Field @@ -11,6 +11,7 @@ class ColorEntry(BaseModel): class ColorParcel(_ParcelBase): + type_: Literal["color"] = Field(default="color", exclude=True) entries: List[ColorEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_color(self, color: TLOCColor): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/data_prefix.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/data_prefix.py index b549a3316..d243c9e94 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/data_prefix.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/data_prefix.py @@ -1,5 +1,5 @@ from ipaddress import IPv4Address, IPv4Network -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -20,6 +20,7 @@ def from_ipv4_network(ipv4_network: IPv4Network) -> "DataPrefixEntry": class DataPrefixParcel(_ParcelBase): + type_: Literal["data-prefix"] = Field(default="data-prefix", exclude=True) entries: List[DataPrefixEntry] = Field(default_factory=list, validation_alias=AliasPath("data", "entries")) def add_data_prefix(self, ipv4_network: IPv4Network): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/expanded_community_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/expanded_community_list.py index 7ceedd0da..0aafb2244 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/expanded_community_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/expanded_community_list.py @@ -1,9 +1,12 @@ +from typing import Literal + from pydantic import AliasPath, ConfigDict, Field, field_validator from catalystwan.api.configuration_groups.parcel import Global, _ParcelBase, as_global class ExpandedCommunityParcel(_ParcelBase): + type_: Literal["expanded-community"] = Field(default="expanded-community", exclude=True) model_config = ConfigDict(populate_by_name=True) expandedCommunityList: Global[list] = Field( default=as_global([]), diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/fowarding_class.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/fowarding_class.py index 769072b39..3500cffcb 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/fowarding_class.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/fowarding_class.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, Field, field_validator @@ -16,6 +16,7 @@ def check_burst(cls, queue: Global): class FowardingClassParcel(_ParcelBase): + type_: Literal["class"] = Field(default="class", exclude=True) entries: List[FowardingClassQueueEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_queue(self, queue: int): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_data_prefix.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_data_prefix.py index 4e6c58603..3e99b980e 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_data_prefix.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_data_prefix.py @@ -1,5 +1,5 @@ from ipaddress import IPv6Address, IPv6Network -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -13,6 +13,7 @@ class IPv6DataPrefixEntry(BaseModel): class IPv6DataPrefixParcel(_ParcelBase): + type_: Literal["data-ipv6-prefix"] = Field(default="data-ipv6-prefix", exclude=True) entries: List[IPv6DataPrefixEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_prefix(self, ipv6_network: IPv6Network): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_prefix_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_prefix_list.py index f80c0687b..6e5c6982d 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_prefix_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/ipv6_prefix_list.py @@ -1,5 +1,5 @@ from ipaddress import IPv6Address, IPv6Network -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -13,6 +13,7 @@ class IPv6PrefixListEntry(BaseModel): class IPv6PrefixListParcel(_ParcelBase): + type_: Literal["ipv6-prefix"] = Field(default="ipv6-prefix", exclude=True) entries: List[IPv6PrefixListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_prefix(self, ipv6_network: IPv6Network): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/policier.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/policier.py index 23c952ab3..bae02f15d 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/policier.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/policier.py @@ -1,9 +1,13 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field, field_validator from catalystwan.api.configuration_groups.parcel import Global, _ParcelBase, as_global -from catalystwan.models.policy.lists_entries import PolicerExceedAction + +PolicerExceedAction = Literal[ + "drop", + "remark", +] class PolicierEntry(BaseModel): @@ -26,6 +30,7 @@ def check_rate(cls, rate_str: Global): class PolicierParcel(_ParcelBase): + type_: Literal["policer"] = Field(default="policer", exclude=True) entries: List[PolicierEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_entry(self, burst: int, exceed: PolicerExceedAction, rate: int): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefered_group_color.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefered_group_color.py index 7d65463a9..a96e0b4c5 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefered_group_color.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefered_group_color.py @@ -1,10 +1,15 @@ -from typing import List, Optional +from typing import List, Literal, Optional from pydantic import AliasPath, BaseModel, ConfigDict, Field, model_validator from catalystwan.api.configuration_groups.parcel import Global, _ParcelBase, as_global from catalystwan.models.common import TLOCColor -from catalystwan.models.policy.lists_entries import PathPreference + +PathPreference = Literal[ + "direct-path", + "multi-hop-path", + "all-paths", +] class Preference(BaseModel): @@ -35,6 +40,7 @@ def check_passwords_match(self) -> "PreferredColorGroupEntry": class PreferredColorGroupParcel(_ParcelBase): + type_: Literal["preferred-color-group"] = Field(default="preferred-color-group", exclude=True) entries: List[PreferredColorGroupEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_primary(self, color_preference: List[TLOCColor], path_preference: PathPreference): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefix_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefix_list.py index 3c74105a6..ac6c9f629 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefix_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/prefix_list.py @@ -1,5 +1,5 @@ from ipaddress import IPv4Address, IPv4Network -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -13,6 +13,7 @@ class PrefixListEntry(BaseModel): class PrefixListParcel(_ParcelBase): + type_: Literal["prefix"] = Field(default="prefix", exclude=True) entries: List[PrefixListEntry] = Field(default_factory=list, validation_alias=AliasPath("data", "entries")) def add_prefix(self, ipv4_network: IPv4Network): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/sla_class.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/sla_class.py index 310f30484..90e6b7adc 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/sla_class.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/sla_class.py @@ -113,6 +113,7 @@ class SLAClassListEntry(BaseModel): class SLAClassParcel(_ParcelBase): + type_: Literal["sla-class"] = Field(default="sla-class", exclude=True) entries: List[SLAClassListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_entry( diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/standard_community.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/standard_community.py index 82ea892d5..bd21cd899 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/standard_community.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/standard_community.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -14,6 +14,7 @@ class StandardCommunityEntry(BaseModel): class StandardCommunityParcel(_ParcelBase): + type_: Literal["standard-community"] = Field(default="standard-community", exclude=True) entries: List[StandardCommunityEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_community(self, standard_community: WellKnownBGPCommunities): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/tloc_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/tloc_list.py index 876c09b6e..8b4207a1b 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/tloc_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/policy/tloc_list.py @@ -1,11 +1,15 @@ from ipaddress import IPv4Address -from typing import List, Optional +from typing import List, Literal, Optional from pydantic import AliasPath, BaseModel, ConfigDict, Field, field_validator from catalystwan.api.configuration_groups.parcel import Global, _ParcelBase, as_global from catalystwan.models.common import TLOCColor -from catalystwan.models.policy.lists_entries import EncapType + +EncapType = Literal[ + "ipsec", + "gre", +] class TlocEntry(BaseModel): @@ -26,6 +30,7 @@ def ensure_correct_preference_value(cls, v: Global): class TlocParcel(_ParcelBase): + type_: Literal["tloc"] = Field(default="tloc", exclude=True) entries: List[TlocEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_entry( diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/application_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/application_list.py index 50d808d4f..39ce368b3 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/application_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/application_list.py @@ -1,4 +1,4 @@ -from typing import List, Union +from typing import List, Literal, Union from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -16,6 +16,7 @@ class SecurityApplicationFamilyListEntry(BaseModel): class SecurityApplicationListParcel(_ParcelBase): + type_: Literal["security-localapp"] = Field(default="security-localapp", exclude=True) entries: List[Union[SecurityApplicationFamilyListEntry, SecurityApplicationListEntry]] = Field( default=[], validation_alias=AliasPath("data", "entries") ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/data_prefix.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/data_prefix.py index 263a77758..30f8fea53 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/data_prefix.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/data_prefix.py @@ -1,5 +1,5 @@ from ipaddress import IPv4Network -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -12,6 +12,7 @@ class SecurityDataPrefixEntry(BaseModel): class SecurityDataPrefixParcel(_ParcelBase): + type_: Literal["security-data-ip-prefix"] = Field(default="security-data-ip-prefix", exclude=True) entries: List[SecurityDataPrefixEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_prefix(self, ip_prefix: IPv4Network): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/fqdn.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/fqdn.py index e80dbe170..5c6826218 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/fqdn.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/fqdn.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -13,6 +13,7 @@ class FQDNListEntry(BaseModel): class FQDNDomainParcel(_ParcelBase): + type_: Literal["security-fqdn"] = Field(default="security-fqdn", exclude=True) entries: List[FQDNListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def from_fqdns(self, fqdns: List[str]): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/geolocation_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/geolocation_list.py index d2e3ce69b..2ec36d68d 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/geolocation_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/geolocation_list.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from typing import List, Literal, Optional from pydantic import AliasPath, BaseModel, Field, model_validator @@ -19,6 +19,7 @@ def check_country_xor_continent(self): class GeoLocationListParcel(_ParcelBase): + type_: Literal["security-geolocation"] = Field(default="security-geolocation", exclude=True) entries: List[GeoLocationListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_country(self, country: str): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/ips_signature.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/ips_signature.py index 4c8f821ca..22684710e 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/ips_signature.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/ips_signature.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field, field_validator @@ -28,6 +28,7 @@ def check_signature_id(cls, signature_id: Global): class IPSSignatureParcel(_ParcelBase): + type_: Literal["security-ipssignature"] = Field(default="security-ipssignature", exclude=True) entries: List[IPSSignatureListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_signature(self, signature: str): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/local_domain.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/local_domain.py index 92e63f4e8..ba2c81477 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/local_domain.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/local_domain.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -13,6 +13,7 @@ class LocalDomainListEntry(BaseModel): class LocalDomainParcel(_ParcelBase): + type_: Literal["security-localdomain"] = Field(default="security-localdomain", exclude=True) entries: List[LocalDomainListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def from_local_domains(self, domains: List[str]): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/protocol_list.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/protocol_list.py index fcff212d2..5105d6446 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/protocol_list.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/protocol_list.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -11,6 +11,7 @@ class ProtocolListEntry(BaseModel): class ProtocolListParcel(_ParcelBase): + type_: Literal["security-protocolname"] = Field(default="security-protocolname", exclude=True) entries: List[ProtocolListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_protocol(self, protocol: str): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/security_port.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/security_port.py index dd7b00914..6829c021f 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/security_port.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/security_port.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field, field_validator @@ -27,6 +27,7 @@ def check_port(cls, port: Global[str]): class SecurityPortParcel(_ParcelBase): + type_: Literal["security-port"] = Field(default="security-port", exclude=True) entries: List[SecurityPortListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_port(self, port: str): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/url.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/url.py index a42a4d561..69ca9163c 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/url.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/url.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Literal from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -11,6 +11,7 @@ class BaseURLListEntry(BaseModel): class BaseURLParcel(_ParcelBase): + type_: Literal["security-urllist"] = Field(default="security-urllist", exclude=True) entries: List[BaseURLListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_url(self, pattern: str): @@ -18,8 +19,14 @@ def add_url(self, pattern: str): class URLAllowParcel(BaseURLParcel): - parcel_type: str = Field(default="urlallowed", validation_alias="type", serialization_alias="type") + type_: Literal["security-urllist"] = Field(default="security-urllist", exclude=True) + parcel_type: Literal["urlallowed"] = Field( + default="urlallowed", validation_alias="type", serialization_alias="type" + ) class URLBlockParcel(BaseURLParcel): - parcel_type: str = Field(default="urlblocked", validation_alias="type", serialization_alias="type") + type_: Literal["security-urllist"] = Field(default="security-urllist", exclude=True) + parcel_type: Literal["urlblocked"] = Field( + default="urlblocked", validation_alias="type", serialization_alias="type" + ) diff --git a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/zone.py b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/zone.py index 8730264d0..1e6315322 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/zone.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/policy_object/security/zone.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from typing import List, Literal, Optional from pydantic import AliasPath, BaseModel, Field, field_validator, model_validator @@ -23,6 +23,7 @@ def check_vpn_xor_interface(self): class SecurityZoneListParcel(_ParcelBase): + type_: Literal["security-zone"] = Field(default="security-zone", exclude=True) entries: List[SecurityZoneListEntry] = Field(default=[], validation_alias=AliasPath("data", "entries")) def add_interface(self, interface: InterfaceType): diff --git a/catalystwan/models/configuration/feature_profile/sdwan/system/aaa.py b/catalystwan/models/configuration/feature_profile/sdwan/system/aaa.py index c18dd2b80..21b4afdc2 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/system/aaa.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/system/aaa.py @@ -1,5 +1,5 @@ from ipaddress import IPv4Address, IPv6Address -from typing import List, Optional, Union +from typing import List, Literal, Optional, Union from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -269,6 +269,7 @@ class AuthorizationRuleItem(BaseModel): class AAA(_ParcelBase): + type_: Literal["aaa"] = Field(default="aaa", exclude=True) authentication_group: Union[DefaultGlobalBool, Variable, Global[bool], Default[bool]] = Field( default=as_default(False), validation_alias=AliasPath("data", "authenticationGroup"), diff --git a/catalystwan/models/configuration/feature_profile/sdwan/system/bfd.py b/catalystwan/models/configuration/feature_profile/sdwan/system/bfd.py index 574a4ed53..123e9ccff 100644 --- a/catalystwan/models/configuration/feature_profile/sdwan/system/bfd.py +++ b/catalystwan/models/configuration/feature_profile/sdwan/system/bfd.py @@ -1,4 +1,4 @@ -from typing import List, Optional, Union +from typing import List, Literal, Optional, Union from pydantic import AliasPath, BaseModel, ConfigDict, Field @@ -30,6 +30,7 @@ class Color(BaseModel): class BFD(_ParcelBase): + type_: Literal["bfd"] = Field(default="bfd", exclude=True) model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) multiplier: Optional[Global[int]] = Field( diff --git a/catalystwan/models/policy/lists.py b/catalystwan/models/policy/lists.py index 5686c3f66..ea6944b12 100644 --- a/catalystwan/models/policy/lists.py +++ b/catalystwan/models/policy/lists.py @@ -4,12 +4,14 @@ from pydantic import BaseModel, Field -from catalystwan.api.configuration_groups.parcel import _ParcelBase from catalystwan.models.common import InterfaceType, TLOCColor, WellKnownBGPCommunities +from catalystwan.models.configuration.feature_profile.sdwan.policy_object import AnyPolicyObjectParcel from catalystwan.models.configuration.feature_profile.sdwan.policy_object.policy.data_prefix import ( DataPrefixEntry, DataPrefixParcel, ) +from catalystwan.models.configuration.feature_profile.sdwan.policy_object.policy.tloc_list import TlocParcel +from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.zone import SecurityZoneListParcel from catalystwan.models.policy.lists_entries import ( AppListEntry, AppProbeClassListEntry, @@ -60,8 +62,8 @@ def _add_entry(self, entry: Any, single: bool = False) -> None: else: self.entries.append(entry) - def to_policy_object_parcel(self) -> _ParcelBase: - return _ParcelBase(parcel_name=self.name, parcel_description=self.description) + def to_policy_object_parcel(self) -> Optional[AnyPolicyObjectParcel]: + return None class DataPrefixList(PolicyListBase): @@ -72,9 +74,11 @@ def add_prefix(self, ip_prefix: IPv4Network) -> None: self._add_entry(DataPrefixListEntry(ip_prefix=ip_prefix)) def to_policy_object_parcel(self) -> DataPrefixParcel: - parcel = DataPrefixParcel(parcel_name=self.name, parcel_description=self.description) - parcel.entries = [DataPrefixEntry.from_ipv4_network(i.ip_prefix) for i in self.entries] - return parcel + return DataPrefixParcel( + parcel_name=self.name, + parcel_description=self.description, + entries=[DataPrefixEntry.from_ipv4_network(i.ip_prefix) for i in self.entries], + ) class SiteList(PolicyListBase): @@ -89,6 +93,9 @@ def add_site_range(self, site_range: Tuple[int, int]): entry = SiteListEntry(site_id=f"{site_range[0]}-{site_range[1]}") self._add_entry(entry) + def to_policy_object_parcel(self) -> None: + return None + class VPNList(PolicyListBase): type: Literal["vpn"] = "vpn" @@ -102,6 +109,9 @@ def add_vpn_range(self, vpn_range: Tuple[int, int]): entry = VPNListEntry(vpn=f"{vpn_range[0]}-{vpn_range[1]}") self._add_entry(entry) + def to_policy_object_parcel(self) -> None: + return None + class ZoneList(PolicyListBase): type: Literal["zone"] = "zone" @@ -113,6 +123,18 @@ def assign_vpns(self, vpns: Set[int]) -> None: def assign_interfaces(self, ifs: Set[InterfaceType]) -> None: self.entries = [ZoneListEntry(interface=interface) for interface in ifs] + def to_policy_object_parcel(self) -> SecurityZoneListParcel: + parcel = SecurityZoneListParcel( + parcel_name=self.name, + parcel_description=self.description, + ) + for e in self.entries: + if e.vpn is not None: + parcel.add_vpn(e.vpn) + if e.interface is not None: + parcel.add_interface(e.interface) + return parcel + class FQDNList(PolicyListBase): type: Literal["fqdn"] = "fqdn" @@ -285,6 +307,15 @@ def add_tloc(self, tloc: IPv4Address, color: TLOCColor, encap: EncapType, prefer _preference = str(preference) if preference is not None else None self.entries.append(TLOCListEntry(tloc=tloc, color=color, encap=encap, preference=_preference)) + def to_policy_object_parcel(self) -> TlocParcel: + parcel = TlocParcel( + parcel_name=self.name, + parcel_description=self.description, + ) + for i in self.entries: + parcel.add_entry(i.tloc, i.color, i.encap, i.preference) + return parcel + class PreferredColorGroupList(PolicyListBase): type: Literal["preferredColorGroup"] = "preferredColorGroup" diff --git a/catalystwan/workflows/config_migration.py b/catalystwan/workflows/config_migration.py index bcbaa3512..3ae015ed3 100644 --- a/catalystwan/workflows/config_migration.py +++ b/catalystwan/workflows/config_migration.py @@ -8,6 +8,8 @@ logger = logging.getLogger(__name__) +SUPPORTED_TEMPLATE_TYPES = ["cedge_aaa", "cedge_aaa"] + def log_progress(task: str, completed: int, total: int) -> None: logger.info(f"{task} {completed}/{total}") @@ -15,8 +17,13 @@ def log_progress(task: str, completed: int, total: int) -> None: def transform(ux1: UX1Config) -> UX2Config: ux2 = UX2Config() - ux2.profile_parcels.extend([lst.to_policy_object_parcel() for lst in ux1.policies.policy_lists]) - ux2.profile_parcels.extend([create_parcel_from_template(ft) for ft in ux1.templates.features]) + for ft in ux1.templates.features: + if ft.template_type in SUPPORTED_TEMPLATE_TYPES: + ux2.profile_parcels.append(create_parcel_from_template(ft)) + # Policy Lists + for policy_list in ux1.policies.policy_lists: + if (parcel := policy_list.to_policy_object_parcel()) is not None: + ux2.profile_parcels.append(parcel) return ux2 @@ -55,8 +62,6 @@ def collect_ux1_config(session: ManagerSession, progress: Callable[[str, int, in ux1.policies.localized_policies.append(policy_api.localized.get(id=lpid)) progress("Collecting Localized Policies", i + 1, len(localized_policy_ids)) - ux1.policies.policy_lists = policy_api.lists.get_all() - """Collect Templates""" template_api = session.api.templates progress("Collecting Templates Info", 0, 2) diff --git a/endpoints-md.py b/endpoints-md.py index 64cd581b9..721f36994 100644 --- a/endpoints-md.py +++ b/endpoints-md.py @@ -199,7 +199,7 @@ def md(self) -> str: # this instantiates APIEndpoints classes triggering method decorators # endpoints not attached to container will be not documented ! - _ = APIEndpointContainter(MagicMock()) + APIEndpointContainter(MagicMock()) endpoint_registry = EndpointRegistry( meta_lookup=request.request_lookup, diff --git a/pyproject.toml b/pyproject.toml index 69efdd4f0..0ab064150 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "catalystwan" -version = "0.31.0dev1" +version = "0.31.0dev2" description = "Cisco Catalyst WAN SDK for Python" authors = ["kagorski "] readme = "README.md"