Merge pull request #685 from cisco-open/extended_community_list_v1_model

Extended community list v1 model

.pre-commit-config.yaml

@@ -48,7 +48,7 @@ repos:
 
 
   - repo: https://github.com/PyCQA/flake8
-    rev: 5.0.4
+    rev: 7.0.0
     hooks:
       - id: flake8
         additional_dependencies: [

catalystwan/api/policy_api.py

@@ -57,6 +57,7 @@
 from catalystwan.endpoints.configuration.policy.list.data_ipv6_prefix import ConfigurationPolicyDataIPv6PrefixList
 from catalystwan.endpoints.configuration.policy.list.data_prefix import ConfigurationPolicyDataPrefixList
 from catalystwan.endpoints.configuration.policy.list.expanded_community import ConfigurationPolicyExpandedCommunityList
+from catalystwan.endpoints.configuration.policy.list.extended_community import ConfigurationPolicyExtendedCommunityList
 from catalystwan.endpoints.configuration.policy.list.fqdn import ConfigurationPolicyFQDNList, FQDNListInfo
 from catalystwan.endpoints.configuration.policy.list.geo_location import ConfigurationPolicyGeoLocationList
 from catalystwan.endpoints.configuration.policy.list.ips_signature import ConfigurationPolicyIPSSignatureList
@@ -111,6 +112,7 @@
     DataIPv6PrefixList,
     DataPrefixList,
     ExpandedCommunityList,
+    ExtendedCommunityList,
     FQDNList,
     GeoLocationList,
     IPSSignatureList,
@@ -173,7 +175,11 @@
 from catalystwan.models.policy.definition.zone_based_firewall import ZoneBasedFWPolicy, ZoneBasedFWPolicyGetResponse
 from catalystwan.models.policy.list.app_probe import AppProbeClassListInfo
 from catalystwan.models.policy.list.class_map import ClassMapListInfo
-from catalystwan.models.policy.list.communities import CommunityListInfo, ExpandedCommunityListInfo
+from catalystwan.models.policy.list.communities import (
+    CommunityListInfo,
+    ExpandedCommunityListInfo,
+    ExtendedCommunityListInfo,
+)
 from catalystwan.models.policy.list.data_ipv6_prefix import DataIPv6PrefixListInfo
 from catalystwan.models.policy.list.data_prefix import DataPrefixListInfo
 from catalystwan.models.policy.list.geo_location import GeoLocationListInfo
@@ -218,6 +224,7 @@
     DataIPv6PrefixList: ConfigurationPolicyDataIPv6PrefixList,
     DataPrefixList: ConfigurationPolicyDataPrefixList,
     ExpandedCommunityList: ConfigurationPolicyExpandedCommunityList,
+    ExtendedCommunityList: ConfigurationPolicyExtendedCommunityList,
     FQDNList: ConfigurationPolicyFQDNList,
     GeoLocationList: ConfigurationPolicyGeoLocationList,
     IPSSignatureList: ConfigurationPolicyIPSSignatureList,
@@ -441,6 +448,10 @@ def get(self, type: Type[DataPrefixList]) -> DataSequence[DataPrefixListInfo]:
     def get(self, type: Type[ExpandedCommunityList]) -> DataSequence[ExpandedCommunityListInfo]:
         ...
 
+    @overload
+    def get(self, type: Type[ExtendedCommunityList]) -> DataSequence[ExtendedCommunityListInfo]:
+        ...
+
     @overload
     def get(self, type: Type[FQDNList]) -> DataSequence[FQDNListInfo]:
         ...
@@ -571,6 +582,10 @@ def get(self, type: Type[DataPrefixList], id: UUID) -> DataPrefixListInfo:
     def get(self, type: Type[ExpandedCommunityList], id: UUID) -> ExpandedCommunityListInfo:
         ...
 
+    @overload
+    def get(self, type: Type[ExtendedCommunityList], id: UUID) -> ExtendedCommunityListInfo:
+        ...
+
     @overload
     def get(self, type: Type[FQDNList], id: UUID) -> FQDNListInfo:
         ...

catalystwan/endpoints/configuration/policy/list/extended_community.py

@@ -0,0 +1,52 @@
+# Copyright 2023 Cisco Systems, Inc. and its affiliates
+
+# mypy: disable-error-code="empty-body"
+from uuid import UUID
+
+from catalystwan.endpoints import APIEndpoints, delete, get, post, put
+from catalystwan.endpoints.configuration.policy.abstractions import PolicyListEndpoints
+from catalystwan.models.policy.list.communities import (
+    ExtendedCommunityList,
+    ExtendedCommunityListEditPayload,
+    ExtendedCommunityListInfo,
+)
+from catalystwan.models.policy.policy_list import InfoTag, PolicyListId, PolicyListPreview
+from catalystwan.typed_list import DataSequence
+
+
+class ConfigurationPolicyExtendedCommunityList(APIEndpoints, PolicyListEndpoints):
+    @post("/template/policy/list/extcommunity")
+    def create_policy_list(self, payload: ExtendedCommunityList) -> PolicyListId:
+        ...
+
+    @delete("/template/policy/list/extcommunity/{id}")
+    def delete_policy_list(self, id: UUID) -> None:
+        ...
+
+    @delete("/template/policy/list/extcommunity")
+    def delete_policy_lists_with_info_tag(self, params: InfoTag) -> None:
+        ...
+
+    @put("/template/policy/list/extcommunity/{id}")
+    def edit_policy_list(self, id: UUID, payload: ExtendedCommunityListEditPayload) -> None:
+        ...
+
+    @get("/template/policy/list/extcommunity/{id}")
+    def get_lists_by_id(self, id: UUID) -> ExtendedCommunityListInfo:
+        ...
+
+    @get("/template/policy/list/extcommunity", "data")
+    def get_policy_lists(self) -> DataSequence[ExtendedCommunityListInfo]:
+        ...
+
+    @get("/template/policy/list/extcommunity/filtered", "data")
+    def get_policy_lists_with_info_tag(self, params: InfoTag) -> DataSequence[ExtendedCommunityListInfo]:
+        ...
+
+    @post("/template/policy/list/extcommunity/preview")
+    def preview_policy_list(self, payload: ExtendedCommunityList) -> PolicyListPreview:
+        ...
+
+    @get("/template/policy/list/extcommunity/preview/{id}")
+    def preview_policy_list_by_id(self, id: UUID) -> PolicyListPreview:
+        ...

catalystwan/models/policy/__init__.py

@@ -16,6 +16,8 @@
     CommunityListInfo,
     ExpandedCommunityList,
     ExpandedCommunityListInfo,
+    ExtendedCommunityList,
+    ExtendedCommunityListInfo,
 )
 from catalystwan.models.policy.list.data_ipv6_prefix import DataIPv6PrefixList, DataIPv6PrefixListInfo
 from catalystwan.models.policy.list.data_prefix import DataPrefixList, DataPrefixListInfo
@@ -123,6 +125,7 @@
         DataIPv6PrefixList,
         DataPrefixList,
         ExpandedCommunityList,
+        ExtendedCommunityList,
         FQDNList,
         GeoLocationList,
         IPSSignatureList,
@@ -161,6 +164,7 @@
         DataIPv6PrefixListInfo,
         DataPrefixListInfo,
         ExpandedCommunityListInfo,
+        ExtendedCommunityListInfo,
         FQDNListInfo,
         GeoLocationListInfo,
         IPSSignatureListInfo,
@@ -242,6 +246,7 @@
     "DnsSecurityPolicy",
     "DNSTypeEntryType",
     "ExpandedCommunityList",
+    "ExtendedCommunityList",
     "FQDNList",
     "GeoLocationList",
     "HubAndSpokePolicy",

catalystwan/models/policy/list/communities.py

@@ -1,6 +1,7 @@
 # Copyright 2024 Cisco Systems, Inc. and its affiliates
 
-from typing import List, Literal
+from ipaddress import IPv4Address, IPv6Address
+from typing import List, Literal, Optional, Union
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -14,6 +15,11 @@ class CommunityListEntry(BaseModel):
     community: str = Field(examples=["1000:10000", "internet", "local-AS"])
 
 
+class ExtendedCommunityListEntry(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    community: str = Field(examples=["soo 1.2.3.4:1000", "rt 10:100"])
+
+
 class CommunityListBase(PolicyListBase):
     entries: List[CommunityListEntry] = []
 
@@ -46,3 +52,32 @@ class ExpandedCommunityListEditPayload(ExpandedCommunityList, PolicyListId):
 
 class ExpandedCommunityListInfo(ExpandedCommunityList, PolicyListInfo):
     pass
+
+
+class ExtendedCommunityList(PolicyListBase):
+    entries: List[ExtendedCommunityListEntry] = []
+    type: Literal["extCommunity"] = "extCommunity"
+
+    def add_site_of_origin_community(
+        self, ip_address: Union[IPv4Address, IPv6Address], port: int, name: Optional[str] = None
+    ):
+        entry = f"soo {ip_address}:{port}"
+        self.entries.append(ExtendedCommunityListEntry(community=self._append_name(entry, name)))
+
+    def add_route_target_community(self, as_number: int, community_number: int, name: Optional[str] = None):
+        entry = f"rt {as_number}:{community_number}"
+        self.entries.append(ExtendedCommunityListEntry(community=self._append_name(entry, name)))
+
+    def _append_name(self, entry: str, name: Optional[str] = None) -> str:
+        if not name:
+            return entry
+
+        return f"{name} {entry}"
+
+
+class ExtendedCommunityListEditPayload(ExtendedCommunityList, PolicyListId):
+    pass
+
+
+class ExtendedCommunityListInfo(ExtendedCommunityList, PolicyListInfo):
+    pass