Merge pull request #787 from cisco-open/fr/policy-groups-aggregation

dev: Fr/policy groups aggregation

catalystwan/api/policy_api.py

@@ -369,9 +369,9 @@ def create(self, policy: AnySecurityPolicy) -> UUID:
         # POST does not return anything! we need to list all after creation and find by name to get id
         self._endpoints.create_security_template(policy)
         policy_infos = [
-            info.root
-            for info in self._endpoints.generate_security_template_list()
-            if info.root.policy_name == policy.policy_name
+            info
+            for info in self._endpoints.generate_security_template_list().root
+            if info.policy_name == policy.policy_name
         ]
         assert len(policy_infos) == 1
         return policy_infos[0].policy_id

catalystwan/api/template_api.py

@@ -279,7 +279,7 @@ def delete(self, template: Type[FeatureTemplate], name: str) -> bool:  # type: i
     def delete(self, template: Type[CLITemplate], name: str) -> bool:  # type: ignore
         ...
 
-    def delete(self, template, name):
+    def delete(self, template, name) -> bool:
         status = False
 
         if template is FeatureTemplate:

catalystwan/api/templates/device_template/device_template.py

@@ -113,6 +113,17 @@ def get(self, name: str, session: ManagerSession) -> DeviceTemplate:
         resp = session.get(f"dataservice/template/device/object/{device_template.id}").json()
         return DeviceTemplate(**resp)
 
+    def associate_feature_template(self, template_type: str, template_uuid: UUID) -> None:
+        self.general_templates.append(
+            GeneralTemplate(name="", template_id=str(template_uuid), template_type=template_type)
+        )
+
+    def associate_security_policy(self, security_policy_uuid: UUID) -> None:
+        self.security_policy_id = str(security_policy_uuid)
+
+    def associate_policy(self, policy_uuid: UUID) -> None:
+        self.policy_id = str(policy_uuid)
+
     model_config = ConfigDict(populate_by_name=True, use_enum_values=True)
 
 

catalystwan/api/templates/models/cisco_secure_internet_gateway.py

@@ -4,7 +4,8 @@
 from pathlib import Path
 from typing import ClassVar, List, Literal, Optional
 
-from pydantic import ConfigDict, Field
+from pydantic import BeforeValidator, ConfigDict, Field
+from typing_extensions import Annotated
 
 from catalystwan.api.templates.feature_template import FeatureTemplate, FeatureTemplateValidator
 
@@ -48,11 +49,24 @@
 SvcType = Literal["sig"]
 
 
+def is_private_ipv4_address(value: ipaddress.IPv4Interface) -> ipaddress.IPv4Interface:
+    # https://www.iana.org/assignments/iana-ipv4-special-registry/iana-ipv4-special-registry.xhtml
+    assert value.is_private, "IPv4 address is not private"
+    return value
+
+
+PrivateIPv4Address = Annotated[
+    ipaddress.IPv4Interface,
+    BeforeValidator(is_private_ipv4_address),
+]
+
+
 class Interface(FeatureTemplateValidator):
     if_name: str = Field(
         ...,  # Ellipsis indicates a required field
+        pattern="ipsec(?:25[0-5]|2[0-4][0-9]|1[0-9]{2}|[1-9][0-9]|[1-9])",
         json_schema_extra={"vmanage_key": "if-name"},
-        description="Name of the interface.",
+        description="Name of the interface. Ipsec1..255 allowed.",
     )
     auto: bool = Field(..., description="Flag to indicate if the interface should be automatically configured.")
     shutdown: bool = Field(..., description="Flag to indicate if the interface is administratively down (shutdown).")
@@ -319,11 +333,14 @@ class Tracker(FeatureTemplateValidator):
     name: str = Field(..., description="Name of the tracker.")
     endpoint_api_url: str = Field(
         ...,
+        pattern=r"^http:\/\/[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)*(\.[a-zA-Z]{2,})(\/\S*)?$",
         json_schema_extra={"vmanage_key": "endpoint-api-url"},
         description="URL of the endpoint API used by the tracker for health checks.",
     )
     threshold: Optional[int] = Field(
-        default=DEFAULT_TRACKER_THRESHOLD, description="Threshold value for the tracker to trigger an alert or action."
+        ge=100,
+        default=DEFAULT_TRACKER_THRESHOLD,
+        description="Threshold value for the tracker to trigger an alert or action.",
     )
     interval: Optional[int] = Field(
         default=DEFAULT_TRACKER_INTERVAL, description="Interval at which the tracker performs health checks."
@@ -354,7 +371,7 @@ class CiscoSecureInternetGatewayModel(FeatureTemplate):
     )
     interface: List[Interface] = Field(description="List of interface configurations associated with the service.")
     service: List[Service] = Field(description="List of service configurations for the Cisco Secure Internet Gateway.")
-    tracker_src_ip: Optional[ipaddress.IPv4Interface] = Field(
+    tracker_src_ip: Optional[PrivateIPv4Address] = Field(
         default=None,
         json_schema_extra={"vmanage_key": "tracker-src-ip"},
         description="Source IP address used by the tracker for sending health check packets.",

catalystwan/endpoints/configuration/policy_group.py

@@ -3,8 +3,8 @@
 # mypy: disable-error-code="empty-body"
 from uuid import UUID
 
-from catalystwan.endpoints import APIEndpoints, delete, get, post, versions
-from catalystwan.models.configuration.policy_group import PolicyGroup, PolicyGroupId
+from catalystwan.endpoints import APIEndpoints, delete, get, post, put, versions
+from catalystwan.models.configuration.policy_group import PolicyGroup, PolicyGroupId, PolicyGroupInfo
 from catalystwan.typed_list import DataSequence
 
 
@@ -16,7 +16,12 @@ def create_policy_group(self, payload: PolicyGroup) -> PolicyGroupId:
 
     @get("/v1/policy-group")
     @versions(">=20.12")
-    def get_all(self) -> DataSequence[PolicyGroupId]:
+    def get_all(self) -> DataSequence[PolicyGroupInfo]:
+        ...
+
+    @put("/v1/policy-group/{group_id}")
+    @versions(">=20.12")
+    def update(self, group_id: UUID, payload: PolicyGroup) -> None:
         ...
 
     @delete("/v1/policy-group/{group_id}")