Skip to content

deploy_task

FW deployment task

logger module-attribute #

logger = getLogger('fortimanager_template_sync.deploy_task')

FMGDeployTask #

Bases: CommonTask

Fortimanager Deployment Task

Steps of this task:

1. check firewall statuses
2. deploy changes to firewalls in protected group only
3. check firewall statuses again

Attributes:

Name Type Description
settings FMGSyncSettings

task settings to use
fmg FMGSync

FMG instance

_deploy_changes #

_deploy_changes(to_deploy: Dict[str, List[str]])

Deploy changes to firewalls

Source code in fortimanager_template_sync/deploy_task.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def _deploy_changes(self, to_deploy: Dict[str, List[str]]):
    """Deploy changes to firewalls"""
    def log_install(percent, log):
        nonlocal last_log, last_percent
        if percent == last_percent and last_log == log:
            return
        last_percent = percent
        last_log = log
        logger.debug(f"{percent}%: {log}")

    last_log = ""
    last_percent = 0
    scopes = []
    for fw, vdoms in to_deploy.items():
        for vdom in vdoms:
            scopes.append(Scope(name=fw, vdom=vdom))
    if scopes:
        if not self.settings.prod_run:
            logger.info("TEST - to deploy to %s", to_deploy)
            return
        logger.debug(f"Deploying to {scopes}")
        task = self.fmg.get_obj(InstallDeviceTask, adom=self.fmg.adom, flags=["auto_lock_ws"], scope=scopes)
        result = task.exec()
        if result.success:
            logger.info(f"Running install for {len(scopes)} items")
            result.wait_for_task(timeout=len(scopes)*120,
                                 callback=log_install)
        else:
            logger.error(f"Error by installation: {result.data}")
    else:
        logger.info("No firewalls/VDOMs to install templates to")

_get_deployable_firewalls staticmethod #

_get_deployable_firewalls(statuses: Dict[str, Dict[str, Any]]) -> Dict[str, List[str]]

Get list of firewall names which are to be deployed.

Example input - statuses 
{
    'FW01': {
        'cli_status': {
            'root': {
                'name': 'test_global', 'status': 'modified', 'type': 'cli'
            },
            'VDOM2': {
                'name': 'test_global', 'status': 'modified', 'type': 'cli'
            }
        },
        'conf_status': 'insync', 'db_status': 'nomod', 'dev_status': 'installed'
    },
    'FW01': {
        'cli_status': {
            'root': {
                'name': 'test_global', 'status': 'modified', 'type': 'cli'
            },
        },
        'conf_status': 'insync', 'db_status': 'nomod', 'dev_status': 'installed'
    }
}
Example output 
{
    "FW1": ["root", "VDOM2"],
    "FW2": ["root"],
}

Returns:

Type Description
Dict[str, List[str]]

List of Dicts with firewall name as key and VDOM as value

Raises:

Type Description
FMGSyncInvalidStatusException

In case there is a firewall with conf or db modified, error out.
Source code in fortimanager_template_sync/deploy_task.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@staticmethod
def _get_deployable_firewalls(statuses: Dict[str, Dict[str, Any]]) -> Dict[str, List[str]]:
    """Get list of firewall names which are to be deployed.

    Example input - statuses:
        ```python
        {
            'FW01': {
                'cli_status': {
                    'root': {
                        'name': 'test_global', 'status': 'modified', 'type': 'cli'
                    },
                    'VDOM2': {
                        'name': 'test_global', 'status': 'modified', 'type': 'cli'
                    }
                },
                'conf_status': 'insync', 'db_status': 'nomod', 'dev_status': 'installed'
            },
            'FW01': {
                'cli_status': {
                    'root': {
                        'name': 'test_global', 'status': 'modified', 'type': 'cli'
                    },
                },
                'conf_status': 'insync', 'db_status': 'nomod', 'dev_status': 'installed'
            }
        }
        ```

    Example output:
        ```python
        {
            "FW1": ["root", "VDOM2"],
            "FW2": ["root"],
        }
        ```

    Returns:
        List of Dicts with firewall name as key and VDOM as value

    Raises:
        (FMGSyncInvalidStatusException): In case there is a firewall with conf or db modified, error out.
    """
    to_deploy = {}
    num_of_vdoms = 0
    for fw, status in statuses.items():
        if status.get("conf_status") == "outofsync" or status.get("db_status") == "mod":
            raise FMGSyncInvalidStatusException(f"Firewall {fw} has modified configuration or database")
        modified_vdoms = [vdom for vdom in status.get("cli_status") if status["cli_status"][vdom].get("status") == "modified"]
        if modified_vdoms:
            to_deploy[fw] = modified_vdoms
            num_of_vdoms += len(to_deploy[fw])
    logger.info(f"Found {num_of_vdoms} firewall/VDOMs to deploy")
    return to_deploy

run #

run() -> bool

Run deployment task

Returns:

Type Description
bool

True if task succeeded, False otherwise
Source code in fortimanager_template_sync/deploy_task.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def run(self) -> bool:
    """Run deployment task

    Returns:
        (bool): True if task succeeded, False otherwise
    """
    success = True
    try:
        if not self.fmg:
            self.fmg = FMGSync(
                base_url=self.settings.fmg_url,
                username=self.settings.fmg_user,
                password=self.settings.fmg_pass,
                adom=self.settings.fmg_adom,
                verify=self.settings.fmg_verify,
            ).open()
        # 1. check firewall statuses
        statuses = self._get_firewall_statuses(self.settings.protected_fw_group)

        # 2. find firewalls with applicable status
        to_deploy = self._get_deployable_firewalls(statuses)

        # 3. deploy changes to firewalls in protected group only
        if to_deploy:
            self._deploy_changes(to_deploy)

        # 4. check firewall statuses again
        if self.settings.prod_run and to_deploy:
            statuses = self._get_firewall_statuses(self.settings.protected_fw_group)
            to_deploy = self._get_deployable_firewalls(statuses)
            if to_deploy:
                logger.warning("The following firewalls are still not updated: %s", list(to_deploy.keys()))
                success = False
            else:
                logger.info("CLI template install task ran successfully")
        else:
            logger.info("No checking required")
    finally:
        self.fmg.close()
        return success