draft: system profile update with device access parcel is working

catalystwan/api/templates/device_template/device_template.py

@@ -4,7 +4,8 @@
 
 import logging
 from pathlib import Path
-from typing import TYPE_CHECKING, ClassVar, List
+from typing import TYPE_CHECKING, ClassVar, List, Optional
+from uuid import UUID
 
 from jinja2 import DebugUndefined, Environment, FileSystemLoader, meta  # type: ignore
 from pydantic import BaseModel, ConfigDict, Field, field_validator
@@ -15,6 +16,13 @@
 logger = logging.getLogger(__name__)
 
 
+def str_to_uuid(s: str) -> Optional[UUID]:
+    try:
+        return UUID(s)
+    except ValueError:
+        return None
+
+
 class GeneralTemplate(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
@@ -54,6 +62,12 @@ class DeviceTemplate(BaseModel):
     )
     policy_id: str = Field(default="", serialization_alias="policyId", validation_alias="policyId")
 
+    def get_security_policy_uuid(self) -> Optional[UUID]:
+        return str_to_uuid(self.security_policy_id)
+
+    def get_policy_uuid(self) -> Optional[UUID]:
+        return str_to_uuid(self.policy_id)
+
     def generate_payload(self) -> str:
         env = Environment(
             loader=FileSystemLoader(self.payload_path.parent),

catalystwan/models/configuration/config_migration.py

@@ -137,6 +137,17 @@ class UX1Templates(BaseModel):
         validation_alias="deviceTemplates",
     )
 
+    def create_device_template_by_policy_id_lookup(self) -> Dict[Literal["policy", "security"], Dict[UUID, UUID]]:
+        lookup: Dict[Literal["policy", "security"], Dict[UUID, UUID]] = {"policy": {}, "security": {}}
+        for dt in self.device_templates:
+            policy_id = dt.get_policy_uuid()
+            security_policy_id = dt.get_security_policy_uuid()
+            if policy_id is not None:
+                lookup["policy"][policy_id] = UUID(dt.template_id)
+            if security_policy_id is not None:
+                lookup["security"][security_policy_id] = UUID(dt.template_id)
+        return lookup
+
 
 class UX1Config(BaseModel):
     # All UX1 Configuration items - Mega Model
@@ -150,13 +161,17 @@ class UX1Config(BaseModel):
 
 
 class TransformHeader(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
     type: str = Field(
         description="Needed to push item to specific endpoint."
         "Type discriminator is not present in many UX2 item payloads"
     )
     origin: UUID = Field(description="Original UUID of converted item")
     origname: Optional[str] = None
     subelements: Set[UUID] = Field(default_factory=set)
+    localized_policy_subelements: Optional[Set[UUID]] = Field(
+        default=None, serialization_alias="localizedPolicySubelements", validation_alias="localizedPolicySubelements"
+    )
     status: ConvertOutputStatus = Field(default="complete")
     info: List[str] = Field(default_factory=list)
 
@@ -251,9 +266,29 @@ def insert_parcel_type_from_headers(cls, values: Dict[str, Any]):
             profile_parcel["parcel"]["type_"] = profile_parcel["header"]["type"]
         return values
 
-    def transformed_parcels_with_origin(self, origin: Set[UUID]) -> List[TransformedParcel]:
+    def list_transformed_parcels_with_origin(self, origin: Set[UUID]) -> List[TransformedParcel]:
         return [p for p in self.profile_parcels if p.header.origin in origin]
 
+    def add_subelement_in_config_group(
+        self, profile_type: ProfileType, device_template_id: UUID, subelement: UUID
+    ) -> bool:
+        profile_ids: Set[UUID] = set()
+        for config_group in self.config_groups:
+            if config_group.header.origin == device_template_id:
+                profile_ids = config_group.header.subelements
+                break
+        if not profile_ids:
+            return False
+        for feature_profile in self.feature_profiles:
+            if feature_profile.header.type == profile_type and feature_profile.header.origin in profile_ids:
+                head = feature_profile.header
+                if head.localized_policy_subelements is None:
+                    head.localized_policy_subelements = {subelement}
+                else:
+                    head.localized_policy_subelements.add(subelement)
+                return True
+        return False
+
 
 class ConfigTransformResult(BaseModel):
     # https://docs.pydantic.dev/2.0/usage/models/#fields-with-dynamic-default-values

catalystwan/models/configuration/feature_profile/sdwan/system/device_access.py

@@ -64,8 +64,8 @@ class Sequence(BaseModel):
 
     base_action: Union[Global[AcceptDropActionType], Default[Literal[AcceptDropActionType]]] = Field(
         default=Default[Literal[AcceptDropActionType]](value="accept"),
-        validation_alias="BasicPolicyActionType",
-        serialization_alias="BasicPolicyActionType",
+        validation_alias="baseAction",
+        serialization_alias="baseAction",
     )
     match_entries: MatchEntries = Field(
         validation_alias="matchEntries", serialization_alias="matchEntries", description="Define match conditions"

catalystwan/models/configuration/feature_profile/sdwan/system/device_access_ipv6.py

@@ -64,8 +64,8 @@ class Sequence(BaseModel):
 
     base_action: Union[Global[AcceptDropActionType], Default[Literal[AcceptDropActionType]]] = Field(
         default=Default[Literal[AcceptDropActionType]](value="accept"),
-        validation_alias="BasicPolicyActionType",
-        serialization_alias="BasicPolicyActionType",
+        validation_alias="baseAction",
+        serialization_alias="baseAction",
     )
     match_entries: MatchEntries = Field(
         validation_alias="matchEntries", serialization_alias="matchEntries", description="Define match conditions"

catalystwan/utils/config_migration/creators/config_pusher.py

@@ -22,6 +22,7 @@
 from catalystwan.models.configuration.feature_profile.sdwan.topology.mesh import MeshParcel
 from catalystwan.session import ManagerSession
 from catalystwan.utils.config_migration.creators.groups_of_interests_pusher import GroupsOfInterestPusher
+from catalystwan.utils.config_migration.creators.localized_policy_pusher import LocalizedPolicyPusher
 from catalystwan.utils.config_migration.creators.security_policy_pusher import SecurityPolicyPusher
 from catalystwan.utils.config_migration.factories.parcel_pusher import ParcelPusherFactory
 
@@ -52,6 +53,13 @@ def __init__(
             push_result=self._push_result,
             push_context=self._push_context,
         )
+        self._localized_policy_feature_updater = LocalizedPolicyPusher(
+            ux2_config=ux2_config,
+            session=session,
+            progress=progress,
+            push_result=self._push_result,
+            push_context=self._push_context,
+        )
         self._security_policy_pusher = SecurityPolicyPusher(
             ux2_config=ux2_config,
             session=session,
@@ -73,6 +81,7 @@ def push(self) -> UX2ConfigPushResult:
         self._create_cloud_credentials()
         self._create_config_groups()
         self._groups_of_interests_pusher.push()
+        self._localized_policy_feature_updater.push()
         self._security_policy_pusher.push()
         self._create_topology_groups(
             self._push_context.default_policy_object_profile_id
@@ -144,6 +153,7 @@ def _create_feature_profile_and_parcels(self, feature_profiles_ids: List[UUID])
                 profile = pusher.push(transformed_feature_profile.feature_profile, parcels, self._config_map.parcel_map)
                 feature_profiles.append(profile)
                 self._push_result.rollback.add_feature_profile(profile.profile_uuid, profile_type)
+                self._push_context.id_lookup[feature_profile_id] = profile.profile_uuid
             except ManagerHTTPError as e:
                 logger.error(f"Error occured during [{fp_name}] feature profile creation: {e}")
             except Exception:
@@ -205,7 +215,7 @@ def _create_topology_groups(self, default_policy_object_profile_id: Optional[UUI
                 logger.error(f"Error occured during topology group creation: {e}")
                 continue
 
-            for transformed_parcel in self._ux2_config.transformed_parcels_with_origin(origins):
+            for transformed_parcel in self._ux2_config.list_transformed_parcels_with_origin(origins):
                 parcel = transformed_parcel.parcel
                 if isinstance(parcel, (CustomControlParcel, HubSpokeParcel, MeshParcel)):
                     try:

catalystwan/utils/config_migration/creators/localized_policy_pusher.py

@@ -0,0 +1,139 @@
+import logging
+from typing import Callable, Dict, List, Tuple, Union, cast
+from uuid import UUID
+
+from pydantic import Field
+from typing_extensions import Annotated
+
+from catalystwan.endpoints.configuration_group import ConfigGroup
+from catalystwan.exceptions import ManagerHTTPError
+from catalystwan.models.configuration.config_migration import (
+    PushContext,
+    TransformedParcel,
+    UX2Config,
+    UX2ConfigPushResult,
+)
+from catalystwan.models.configuration.feature_profile.sdwan.acl.ipv4acl import Ipv4AclParcel
+from catalystwan.models.configuration.feature_profile.sdwan.acl.ipv6acl import Ipv6AclParcel
+from catalystwan.models.configuration.feature_profile.sdwan.system.device_access import DeviceAccessIPv4Parcel
+from catalystwan.models.configuration.feature_profile.sdwan.system.device_access_ipv6 import DeviceAccessIPv6Parcel
+from catalystwan.models.configuration.profile_type import ProfileType
+from catalystwan.session import ManagerSession
+from catalystwan.utils.config_migration.creators.references_updater import update_parcel_references
+
+LOCALIZED_POLICY_PARCEL_TYPES = [
+    "ipv4-device-access-policy",
+    "ipv6-device-access-policy",
+    "ipv4-acl",
+    "ipv6-acl",
+    "route",
+]
+AnyDeviceAccessParcel = Annotated[
+    Union[DeviceAccessIPv4Parcel, DeviceAccessIPv6Parcel],
+    Field(discriminator="type_"),
+]
+AnyAclParcel = Annotated[
+    Union[Ipv4AclParcel, Ipv6AclParcel],
+    Field(discriminator="type_"),
+]
+
+logger = logging.getLogger(__name__)
+
+
+class LocalizedPolicyPusher:
+    """
+    1. Associate selected Config Group with Default_Policy_Object_Profile
+    2. Update selected Feature Profiles with Parcels originating from Localized Policy items (eg. acl, route)
+    Update needs to be performed after Feature Profiles are already populated with VPN parcels
+    and Default_Policy_Object_Profile is populated with Groups of Interest
+    """
+
+    def __init__(
+        self,
+        ux2_config: UX2Config,
+        session: ManagerSession,
+        push_result: UX2ConfigPushResult,
+        push_context: PushContext,
+        progress: Callable[[str, int, int], None],
+    ) -> None:
+        self._ux2_config = ux2_config
+        self._fp_api = session.api.sdwan_feature_profiles
+        self._cg_api = session.api.config_group
+        self._system_api = self._fp_api.system
+        self._transport_api = self._fp_api.transport
+        self._service_api = self._fp_api.service
+        self.dns = session.api.sdwan_feature_profiles.dns_security
+        self._push_result: UX2ConfigPushResult = push_result
+        self._progress: Callable[[str, int, int], None] = progress
+        self.push_context = push_context
+        self._parcel_by_id = self._create_parcel_by_id_lookup()
+
+    def _create_parcel_by_id_lookup(self) -> Dict[UUID, TransformedParcel]:
+        lookup: Dict[UUID, TransformedParcel] = dict()
+        for transformed_parcel in self._ux2_config.profile_parcels:
+            if transformed_parcel.header.type in LOCALIZED_POLICY_PARCEL_TYPES:
+                lookup[transformed_parcel.header.origin] = transformed_parcel
+        return lookup
+
+    def get_parcels_to_push(self, parcel_ids: List[UUID]) -> List[TransformedParcel]:
+        result: List[TransformedParcel] = list()
+        for parcel_id in parcel_ids:
+            if parcel := self._parcel_by_id.get(parcel_id):
+                result.append(parcel)
+        return result
+
+    def find_config_groups_to_update(self) -> List[UUID]:
+        result: List[UUID] = list()
+        for transformed_cg in self._ux2_config.config_groups:
+            if transformed_cg.header.localized_policy_subelements is not None:
+                updated_id = self.push_context.id_lookup[transformed_cg.header.origin]
+                result.append(updated_id)
+        return result
+
+    def get_config_group_contents(self, cg_ids: List[UUID]) -> Dict[UUID, ConfigGroup]:
+        result: Dict[UUID, ConfigGroup] = dict()
+        for cg_id in cg_ids:
+            cg = self._cg_api.get(cg_id)
+            result[cg_id] = cg
+        return result
+
+    def find_profiles_to_update(self) -> List[Tuple[ProfileType, UUID, List[TransformedParcel]]]:
+        profiles: List[Tuple[ProfileType, UUID, List[TransformedParcel]]] = list()
+        for transformed_profile in self._ux2_config.feature_profiles:
+            if transformed_profile.header.localized_policy_subelements is not None:
+                updated_id = self.push_context.id_lookup[transformed_profile.header.origin]
+                parcels = self.get_parcels_to_push(list(transformed_profile.header.localized_policy_subelements))
+                profiles.append((cast(ProfileType, transformed_profile.header.type), updated_id, parcels))
+        return profiles
+
+    def update_system_profile(self, profile_id: UUID, device_access: AnyDeviceAccessParcel):
+        try:
+            self._system_api.create_parcel(
+                profile_id=profile_id, payload=update_parcel_references(device_access, self.push_context.id_lookup)
+            )
+        except ManagerHTTPError as e:
+            logger.error(f"Error occured during creation of {device_access.type_} {device_access.parcel_name}: {e}")
+
+    def associate_config_groups_with_default_policy_object_profile(self):
+        for cg_id, cg in self.get_config_group_contents(self.find_config_groups_to_update()).items():
+            profile_ids = [p.id for p in cg.profiles]
+            profile_ids.append(self.push_context.default_policy_object_profile_id)
+            try:
+                self._cg_api.edit(
+                    cg_id=str(cg_id),
+                    name=cg.name,
+                    description=cg.description,
+                    solution=cg.solution,
+                    profile_ids=profile_ids,
+                )
+            except ManagerHTTPError as e:
+                logger.error(f"Error occured during config group edit: {e}")
+
+    def push(self):
+        self.associate_config_groups_with_default_policy_object_profile()
+        for profile_tuple in self.find_profiles_to_update():
+            type_, uuid_, transformed_parcels = profile_tuple
+            for transformed_parcel in transformed_parcels:
+                parcel = transformed_parcel.parcel
+                if type_ == "system" and isinstance(parcel, (DeviceAccessIPv4Parcel, DeviceAccessIPv6Parcel)):
+                    self.update_system_profile(profile_id=uuid_, device_access=parcel)

catalystwan/workflows/config_migration.py

@@ -330,14 +330,15 @@ def transform(ux1: UX1Config, add_suffix: bool = True) -> ConfigTransformResult:
                             subtemplates_mapping[UUID(subtemplate_level_1.templateId)].add(
                                 UUID(subtemplate_level_2.templateId)
                             )
-
+        policy_id = dt.get_policy_uuid()
         transformed_cg = TransformedConfigGroup(
             header=TransformHeader(
                 type="config_group",
                 origin=UUID(dt.template_id),
                 subelements=set(
                     [fp_system_uuid, fp_other_uuid, fp_service_uuid, fp_transport_and_management_uuid, fp_cli_uuid]
                 ),
+                localized_policy_subelements=set([policy_id]) if policy_id else None,
             ),
             config_group=ConfigGroupCreationPayload(
                 name=dt.template_name,
@@ -445,6 +446,26 @@ def transform(ux1: UX1Config, add_suffix: bool = True) -> ConfigTransformResult:
             )
             ux2.profile_parcels.append(TransformedParcel(header=header, parcel=pd_parcel))
 
+    # Localized Policies
+    _lookup = ux1.templates.create_device_template_by_policy_id_lookup()
+    dt_by_policy_id = _lookup["policy"]
+    for localized_policy in ux1.policies.localized_policies:
+        if not isinstance(localized_policy.policy_definition, str):
+            if dt_id := dt_by_policy_id.get(localized_policy.policy_id):
+                for item in localized_policy.policy_definition.assembly:
+                    if item.type == "deviceaccesspolicy" or item.type == "deviceaccesspolicyv6":
+                        ux2.add_subelement_in_config_group(
+                            profile_type="system", device_template_id=dt_id, subelement=item.definition_id
+                        )
+                    elif item.type == "acl" or item.type == "aclv6":
+                        ux2.add_subelement_in_config_group(
+                            profile_type="transport", device_template_id=dt_id, subelement=item.definition_id
+                        )
+                    elif item.type == "vedgeRoute":
+                        ux2.add_subelement_in_config_group(
+                            profile_type="service", device_template_id=dt_id, subelement=item.definition_id
+                        )
+
     # Security policies
     for security_policy in ux1.policies.security_policies:
         sp_result = convert_security_policy(security_policy, security_policy.policy_id, policy_context)