Policy support examples

.github/workflows/linters.yml

@@ -43,7 +43,7 @@ jobs:
         uses: pre-commit/action@2c7b3805fd2a0fd8c1884dcaebf91fc102a13ecd # v3.0.1
 
       - name: Run ansible-lint
-        uses: ansible/ansible-lint@95382d398ea1744bf6bfa47b030f14c38b3f6957 # v24.7.0
+        uses: ansible/ansible-lint@v25.7.0
 
       - name: Install detect-secrets
         run: pip install detect-secrets==1.4.0

plugins/module_utils/policy_templates/definition.py

@@ -5,6 +5,7 @@
     AclPolicy,
     AdvancedInspectionProfilePolicy,
     AdvancedMalwareProtectionPolicy,
+    AppRoutePolicy,
     CflowdPolicy,
     ControlPolicy,
     DeviceAccessIPv6Policy,
@@ -31,6 +32,7 @@
     "access_control_policy_ipv6": AclIPv6Policy,
     "aip": AdvancedInspectionProfilePolicy,
     "amp": AdvancedMalwareProtectionPolicy,
+    "app_route": AppRoutePolicy,
     "cflowd": CflowdPolicy,
     "control": ControlPolicy,
     "device_access": DeviceAccessPolicy,
@@ -64,6 +66,7 @@
                 "default": "feature",
             },
             "definition": {"type": "dict"},
+            "sequences": {"type": "list"},
         },
     }
 }

plugins/module_utils/policy_templates/security.py

@@ -0,0 +1,15 @@
+policy_security_definition = {
+    "security": {
+        "default": None,
+        "required": False,
+        "type": "dict",
+        "options": {
+            "type": {
+                "type": "str",
+                "choices": ["feature", "cli"],
+                "default": "feature",
+            },
+            "definition": {"type": "dict"},
+        },
+    }
+}

plugins/modules/cli_templates_info.py

@@ -126,8 +126,6 @@
   type: list
 """
 
-from typing import Literal
-
 from catalystwan.api.template_api import CLITemplate
 from catalystwan.api.templates.device_template.device_template import DeviceTemplate
 from catalystwan.models.templates import DeviceTemplateInformation

plugins/modules/policy.py

@@ -110,10 +110,12 @@
 from catalystwan.models.policy import (
     AnyPolicyDefinition,
     AnyPolicyList,
+    AnySecurityPolicyInfo,
     CentralizedPolicy,
     CentralizedPolicyInfo,
     LocalizedPolicy,
     LocalizedPolicyInfo,
+    SecurityPolicy,
 )
 from catalystwan.models.policy.centralized import CentralizedPolicyEditPayload
 from catalystwan.session import ManagerHTTPError
@@ -123,6 +125,7 @@
 from ..module_utils.policy_templates.definition import policy_definition_definition, policy_definition_type_mapping
 from ..module_utils.policy_templates.list import policy_list_definition, policy_list_type_mapping
 from ..module_utils.policy_templates.localized import policy_localized_definition
+from ..module_utils.policy_templates.security import policy_security_definition
 from ..module_utils.result import ModuleResult
 from ..module_utils.vmanage_module import AnsibleCatalystwanModule
 
@@ -157,6 +160,7 @@ def run_module():
         **policy_definition_definition,
         **policy_centralized_definition,
         **policy_localized_definition,
+        **policy_security_definition,
     )
 
     result = ModuleResult()
@@ -169,6 +173,7 @@ def run_module():
                 "localized",
                 "list",
                 "definition",
+                "security",
             ),
         ],
         mutually_exclusive=[
@@ -177,23 +182,44 @@ def run_module():
                 "localized",
                 "list",
                 "definition",
+                "security",
             ),
         ],
     )
 
     object_name: str = module.params.get("name")
     object_description: str = module.params.get("description")
 
-    if module.params.get("centralized"):
-        object_pretty_name = "Centralized Policy"
-        object_type = CentralizedPolicy
-        object_endpoint = module.session.api.policy.centralized
-        all_centralized_policy: DataSequence[CentralizedPolicyInfo] = module.get_response_safely(object_endpoint.get)
-        filtered_definitions: Optional[DataSequence[CentralizedPolicyInfo]] = all_centralized_policy.filter(
-            policy_name=object_name
-        )
+    if module.params.get("centralized") or module.params.get("localized") or module.params.get("security"):
+        if module.params.get("centralized"):
+            object_pretty_name = "Centralized Policy"
+            object_type = CentralizedPolicy
+            object_edit_type = CentralizedPolicyEditPayload
+            object_endpoint = module.session.api.policy.centralized
+            policy_definition = module.params.get("centralized")
+        elif module.params.get("localized"):
+            object_pretty_name = "Localized Policy"
+            object_type = LocalizedPolicy
+            object_edit_type = object_type
+            object_endpoint = module.session.api.policy.localized
+            policy_definition = module.params.get("localized")
+        elif module.params.get("security"):
+            object_pretty_name = "Security Policy"
+            object_type = SecurityPolicy
+            object_edit_type = object_type
+            object_endpoint = module.session.api.policy.security
+            policy_definition = module.params.get("security")
+
+        all_policies: DataSequence[
+            CentralizedPolicyInfo | LocalizedPolicyInfo | AnySecurityPolicyInfo
+        ] = module.get_response_safely(object_endpoint.get)
+        if module.params.get("security"):
+            all_policies = all_policies.security
+        filtered_definitions: Optional[
+            DataSequence[CentralizedPolicyInfo | LocalizedPolicyInfo | AnySecurityPolicyInfo]
+        ] = all_policies.filter(policy_name=object_name)
         if filtered_definitions:
-            existing_object: DataSequence[CentralizedPolicy] = [
+            existing_object: DataSequence[CentralizedPolicy | LocalizedPolicy | SecurityPolicy] = [
                 module.get_response_safely(object_endpoint.get, id=filtered_definitions[0].policy_id)
             ]
             existing_object_id = filtered_definitions[0].policy_id
@@ -204,52 +230,20 @@ def run_module():
             object_to_create = object_type(
                 policy_name=object_name,
                 policy_description=object_description,
-                policy_type=module.params.get("centralized").get("type"),
-                policy_definition=module.params.get("centralized").get("definition"),
+                policy_type=policy_definition.get("type"),
+                policy_definition=policy_definition.get("definition"),
                 is_policy_activated=module.params.get("state") == "active",
             )
         elif module.params.get("state") in ("active", "present"):
-            object_to_create = CentralizedPolicyEditPayload(
+            object_to_create = object_edit_type(
                 policy_name=object_name,
                 policy_description=object_description,
-                policy_type=module.params.get("centralized").get("type"),
-                policy_definition=module.params.get("centralized").get("definition"),
+                policy_type=policy_definition.get("type"),
+                policy_definition=policy_definition.get("definition"),
                 is_policy_activated=module.params.get("state") == "active",
                 policy_id=existing_object_id,
             )
 
-    elif module.params.get("localized"):
-        object_pretty_name = "Localized Policy"
-        object_type = LocalizedPolicy
-        object_endpoint = module.session.api.policy.localized
-        all_localized_policy: DataSequence[LocalizedPolicyInfo] = module.get_response_safely(object_endpoint.get)
-        filtered_definitions: Optional[DataSequence[LocalizedPolicyInfo]] = all_localized_policy.filter(
-            policy_name=object_name
-        )
-        if filtered_definitions:
-            existing_object: DataSequence[LocalizedPolicy] = [
-                module.get_response_safely(object_endpoint.get, id=filtered_definitions[0].policy_id)
-            ]
-            existing_object_id = filtered_definitions[0].policy_id
-        else:
-            existing_object = []
-
-        if module.params.get("state") in ("active", "present") and not existing_object:
-            object_to_create = object_type(
-                policy_name=object_name,
-                policy_description=object_description,
-                policy_type=module.params.get("localized").get("type"),
-                policy_definition=module.params.get("localized").get("definition"),
-            )
-        elif module.params.get("state") in ("active", "present"):
-            object_to_create = LocalizedPolicy(
-                policy_name=object_name,
-                policy_description=object_description,
-                policy_type=module.params.get("localized").get("type"),
-                policy_definition=module.params.get("localized").get("definition"),
-                policy_id=existing_object_id,
-            )
-
     elif module.params.get("definition"):
         object_pretty_name = "Policy Definition"
         object_type = policy_definition_type_mapping[module.params.get("definition").get("type")]
@@ -259,6 +253,7 @@ def run_module():
                 name=object_name,
                 description=object_description,
                 definition=module.params.get("definition").get("definition"),
+                sequences=module.params.get("definition").get("sequences"),
             )
         all_policy_definitions: DataSequence[AnyPolicyDefinition] = module.get_response_safely(
             object_endpoint.get, type=object_type
@@ -315,8 +310,8 @@ def run_module():
                             )
                         device_action.wait_for_completed()
                     object_endpoint.edit(policy=object_to_create)
-                elif module.params.get("localized"):
-                    object_endpoint.edit(policy=object_to_create)
+                elif module.params.get("localized") or module.params.get("security"):
+                    object_endpoint.edit(id=existing_object_id, policy=object_to_create)
                 elif module.params.get("definition"):
                     object_endpoint.edit(
                         id=existing_object_id,
@@ -340,7 +335,7 @@ def run_module():
         # object doesn't exist in Manager and needs to be created
         else:
             try:
-                if module.params.get("centralized") or module.params.get("localized"):
+                if module.params.get("centralized") or module.params.get("localized") or module.params.get("security"):
                     created_uuid: UUID = module.get_response_safely(object_endpoint.create, policy=object_to_create)
                 elif module.params.get("definition"):
                     created_uuid: UUID = module.get_response_safely(

roles/policies/README.md

@@ -0,0 +1,73 @@
+# Ansible Role: policies
+
+This Ansible role provides user with a few, ready to use, legacy policies scenarios. It handles creating legacy policies along will all required policies definitions and lists.
+
+## Role Description
+
+The `policies` role performs the following tasks:
+
+1. Create legacy policies based on user provided configuration. Supported scenarios:
+  - hub and spoke topology
+  - mesh topology
+  - application route policy
+  - acl policy
+  - geolocation blockade
+
+## Requirements
+
+- `cisco.catalystwan` collection installed.
+- Access details for the Cisco Manager instance must be provided.
+
+## Dependencies
+
+There are no external role dependencies. Only `cisco.catalystwan` collection is required.
+
+## Role Variables
+
+Variables expected by this role:
+
+- `vmanage_instances`: A list of vManage instances containing management IP, admin username, and admin password.
+- `policies`: A dictionary containing configuration of policies
+
+Example of `vmanage_instances`:
+
+```yaml
+vmanage_instances:
+  - hostname: 'vmanage01'
+    system_ip: '192.0.2.10'
+    mgmt_public_ip: '198.51.100.10'
+    admin_username: 'admin'
+    admin_password: 'password'
+```
+
+Example of `policies`:
+```yaml
+policies:
+  mesh:
+    - name: my_mesh_policy
+      vpns:
+        - 100
+        - 101
+      regions:
+        - name: mesh_region1
+          sites:
+            - 100
+        - name: mesh_region2
+          sites:
+            - 101
+  app_route:
+    - name: my_app_counter
+      match:
+        source_ip: 10.0.0.0/24
+        destination_port: 64534
+      action:
+        counter: my_counter
+```
+
+## License
+
+"GPL-3.0-only"
+
+## Author Information
+
+This role was created by Piotr Piwowarski <[email protected]>

roles/policies/defaults/main.yml

@@ -0,0 +1,19 @@
+# Copyright 2025 Cisco Systems, Inc. and its affiliates
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+---
+
+policies: {}
+default_policy:
+  name: "Ansible Managed policy"
+  hub_and_spoke: []
+  mesh: []
+  acl_policy: []
+  app_route: []
+  geolocation_block: []
+
+combined_policies: "{{ default_policy | combine(policies, recursive=True) }}"
+
+created_centralized_policies: []
+created_localized_policies: []
+created_security_policies: []

roles/policies/meta/main.yml

@@ -0,0 +1,15 @@
+---
+
+galaxy_info:
+  author: Piotr Piwowarski <[email protected]>
+  description: Allow user to configure pre defined policies in Cisco SD-WAN
+  license: GPL-3.0-or-later
+  min_ansible_version: "2.16.6"
+
+  galaxy_tags:
+    - cisco
+    - sdwan
+    - catalystwan
+    - networking
+
+dependencies: []

roles/policies/tasks/acl_policy.yml

@@ -0,0 +1,51 @@
+# Copyright 2025 Cisco Systems, Inc. and its affiliates
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+---
+- name: "Create acl policy {{ policy_item['name'] }}"
+  cisco.catalystwan.policy:
+    name: "{{ policy_item['name'] }}"
+    definition:
+      type: "access_control_list"
+      sequences: "{{ _sequences | from_yaml }}"
+    manager_credentials:
+      url: "{{ (vmanage_instances | first).mgmt_public_ip }}"
+      username: "{{ (vmanage_instances | first).admin_username }}"
+      password: "{{ (vmanage_instances | first).admin_password }}"
+  register: result_policy
+  vars:
+    _sequences: |
+      - sequenceId: 1
+      {% if "next_hop" in policy_item['action'] %}
+        actions:
+          - type: set
+            parameter:
+              - field: nextHop
+                value: "{{ policy_item['action']['next_hop'] }}"
+      {% endif %}
+        match:
+          entries:
+      {% if "source_port" in policy_item['match'] %}
+            - field: sourcePort
+              value: "{{ policy_item['match']['source_port'] }}"
+      {% endif %}
+      {% if "destination_port" in policy_item['match'] %}
+            - field: destinationPort
+              value: "{{ policy_item['match']['destination_port'] }}"
+      {% endif %}
+      {% if "source_ip" in policy_item['match'] %}
+            - field: sourceIp
+              value: "{{ policy_item['match']['source_ip'] }}"
+      {% endif %}
+      {% if "destination_ip" in policy_item['match'] %}
+            - field: destinationIp
+              value: "{{ policy_item['match']['destination_ip'] }}"
+      {% endif %}
+
+- name: Save policy id
+  ansible.builtin.set_fact:
+    created_localized_policies: "{{ created_localized_policies + [_created_policy] }}"
+  vars:
+    _created_policy:
+      type: acl
+      definitionId: "{{ result_policy['id'] }}"