pyfortiapi.py

#!/usr/bin/env Python
__author__ = "James Simpson"
__copyright__ = "Copyright 2017, James Simpson"
__license__ = "MIT"
__version__ = "0.3.0"


import requests
import logging
import re

from requests.packages.urllib3.exceptions import InsecureRequestWarning

GROUP_SIZE_LIMIT_NAME = 'GROUP_SIZE_LIMIT'
FORTIGATE_LIMITS = {
    'default': {
        GROUP_SIZE_LIMIT_NAME: 300
    },
    'v7': {
        GROUP_SIZE_LIMIT_NAME: 600
    }
}

class FortiGate:
    def __init__(self, ipaddr, username, password, timeout=10, vdom="root", port="443", verify=False, keep_alive=False):

        self.ipaddr = ipaddr
        self.username = username
        self.password = password
        self.port = port
        self.urlbase = "https://{ipaddr}:{port}/".format(ipaddr=self.ipaddr,port=self.port)
        self.timeout = timeout
        self.vdom = vdom
        self.verify = verify
        self.version = None
        self.session = keep_alive and self.login() or None

    # Login / Logout Handlers
    def login(self):
        """
        Log in to FortiGate with info provided in during class instantiation


        :return: Open Session
        """
        session = requests.session()
        try:
            self._login(session)
            return session
        except Exception:
            session.close()
            raise

    def _login(self, session):
        if not self.verify:
            # Disable requests' warnings for insecure connections
            requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

        url = self.urlbase + 'logincheck'

        session.verify=self.verify

        # Login
        session.post(url,
                     data='username={username}&secretkey={password}'.format(username=self.username,
                                                                            password=self.password),
                     verify=self.verify,
                     timeout=self.timeout)

        # Get CSRF token from cookies, add to headers
        cookie_name = "ccsrftoken"

        cookies = [o for o in session.cookies if o and o.name.startswith(cookie_name)]

        if not cookies:
            raise ValueError("Invalid login credentials. Cookie 'ccsrftoken' is missing.")

        cookie = cookies[0]
        token = str(cookie.value).strip("\"")
        session.headers.update({"X-CSRFTOKEN": token})

        # Check whether login was successful
        login_check = session.get(self.urlbase + "api/v2/cmdb/system/vdom")

        login_check.raise_for_status()

        if not self.version:
            # Get FortiGate version number
            major, minor, patch = login_check.json()['version'].split('.')
            self.version = {'major': major, 'minor': minor, 'patch': patch}


    def logout(self, session):
        """
        Log out of device

        :param session: Session created by login method

        :return: None
        """
        url = self.urlbase + 'logout'
        session.get(url, verify=self.verify, timeout=self.timeout)
        session.close()
        logging.info("Session logged out.")


    def get_limits(self):
        if not self.version:
            self.login()
        major_version = self.version.get('major')
        if major_version in FORTIGATE_LIMITS:
            return FORTIGATE_LIMITS.get(major_version)
        return FORTIGATE_LIMITS.get('default')


    # General Logic Methods
    def does_exist(self, object_url):
        """
        GET URL to assert whether it exists within the firewall

        :param object_url: Object to locate

        :return: Bool - True if exists, False if not
        """
        session = self.session or self.login()
        request = session.get(object_url, verify=self.verify, timeout=self.timeout, params='vdom='+self.vdom)
        if not self.session:
            self.logout(session)
        if request.status_code == 200:
            return True
        else:
            return False

    # API Interaction Methods
    def get(self, url):
        """
        Perform GET operation on provided URL

        :param url: Target of GET operation

        :return: Request result if successful (type list), HTTP status code otherwise (type int)
        """
        session = self.session or self.login()
        request = session.get(url, verify=self.verify, timeout=self.timeout, params='vdom='+self.vdom)
        if not self.session:
            self.logout(session)

        if request.status_code == 200:
            return request.json()['results']
        else:
            return request.status_code

    def put(self, url, data):
        """
        Perform PUT operation on provided URL

        :param url: Target of PUT operation
        :param data: JSON data. MUST be a correctly formatted string. e.g. "{'key': 'value'}"

        :return: HTTP status code returned from PUT operation
        """
        session = self.session or self.login()
        result = session.put(url, data=data, verify=self.verify, timeout=self.timeout, params='vdom='+self.vdom).status_code
        if not self.session:
            self.logout(session)
        return result

    def post(self, url, data):
        """
        Perform POST operation on provided URL

        :param url: Target of POST operation
        :param data: JSON data. MUST be a correctly formatted string. e.g. "{'key': 'value'}"

        :return: HTTP status code returned from POST operation
        """
        session = self.session or self.login()
        result = session.post(url, data=data, verify=self.verify, timeout=self.timeout, params='vdom='+self.vdom).status_code
        if not self.session:
            self.logout(session)
        return result

    def delete(self, url):
        """
        Perform DELETE operation on provided URL

        :param url: Target of DELETE operation

        :return: HTTP status code returned from DELETE operation
        """
        session = self.session or self.login()
        result = session.delete(url, verify=self.verify, timeout=self.timeout, params='vdom='+self.vdom).status_code

        if not self.session:
            self.logout(session)
        return result

    # Firewall Address Methods
    def get_firewall_address(self, specific=False, filters=False):
        """
        Get address object information from firewall

        :param specific: If provided, a specific object will be returned. If not, all objects will be returned.
        :param filters: If provided, the raw filter is appended to the API call.

        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/address/"
        if specific:
            api_url += specific
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        return results

    def update_firewall_address(self, address, data):
        """
        Update firewall address record with provided data

        :param address: Address record being updated
        :param data: JSON Data with which to upate the address record

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/address/" + requests.utils.quote(address, safe='')
        # Check whether target object exists
        if not self.does_exist(api_url):
            logging.error('Requested address "{address}" does not exist in Firewall config.'.format(address=address))
            return 404
        result = self.put(api_url, data)
        return result

    def create_firewall_address(self, address, data, vdoms=None):
        """
        Create firewall address record

        :param address: Address record to be created
        :param data: JSON Data with which to create the address record

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/address/"
        # Check whether target object already exists
        if self.does_exist(api_url + address):
            return 424
        result = self.post(api_url, data)
        return result

    def delete_firewall_address(self, address):
        """
        Delete firewall address record

        :param address: Address record to be deleted

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/address/" + address
        result = self.delete(api_url)
        return result

    # Address Group Methods
    def get_address_group(self, specific=False, filters=False):
        """
        Get address group object information from firewall

        :param specific: If provided, a specific object will be returned. If not, all objects will be returned.
        :param filters: If provided, the raw filter is appended to the API call.

        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/addrgrp/"
        if specific:
            api_url += specific
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        return results

    def update_address_group(self, group_name, data):
        """
        Update address group with provided data

        :param group_name: Address group being updated
        :param data: JSON Data with which to upate the address group

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/addrgrp/" + group_name
        # Check whether target object already exists
        if not self.does_exist(api_url):
            logging.error('Requested address group "{group_name}" does not exist in Firewall config.'.format(
                group_name=group_name))
            return 404
        result = self.put(api_url, data)
        return result

    def create_address_group(self, group_name, data):
        """
        Create address group

        :param group_name: Address group to be created
        :param data: JSON Data with which to create the address group

        :return: HTTP Status Code.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/addrgrp/"
        if self.does_exist(api_url + group_name):
            return 424
        result = self.post(api_url, data)
        return result

    def delete_address_group(self, group_name):
        """
        Delete firewall address group

        :param group_name: Address group to be deleted

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/addrgrp/" + group_name
        result = self.delete(api_url)
        return result

    # Service Category Methods
    def get_service_category(self, specific=False, filters=False):
        """
        Get service category information from firewall

        :param specific: If provided, a specific object will be returned. If not, all objects will be returned.
        :param filters: If provided, the raw filter is appended to the API call.

        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/category/"
        if specific:
            api_url += specific
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        return results

    def update_service_category(self, category, data):
        """
        Update service category with provided data.

        :param category: Service category being updated
        :param data: JSON Data with which to upate the service category

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/category/" + category
        # Check whether target object already exists
        if not self.does_exist(api_url):
            logging.error('Requested service category "{category}" does not exist in Firewall config.'.format(
                category=category))
            return 404
        result = self.put(api_url, data)
        return result

    def create_service_category(self, category, data):
        """
        Create service category

        :param category: Service category to be created
        :param data: JSON Data with which to create the service category

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/category/"
        if self.does_exist(api_url + category):
            return 424
        result = self.post(api_url, data)
        return result

    def delete_service_category(self, category):
        """
        Delete firewall service category

        :param category: Service categrory to be deleted

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/category/" + category
        result = self.delete(api_url)
        return result

    # Service Group Methods
    def get_service_group(self, specific=False, filters=False):
        """
        Get service group information from firewall

        :param specific: If provided, a specific object will be returned. If not, all objects will be returned.
        :param filters: If provided, the raw filter is appended to the API call.

        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/group/"
        if specific:
            api_url += specific
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        return results

    def update_service_group(self, group_name, data):
        """
        Update service group with provided data

        :param group_name: Service group being updated
        :param data: JSON Data with which to upate the service group

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/group/" + group_name
        # Check whether target object already exists
        if not self.does_exist(api_url):
            logging.error('Requested service group "{group_name}" does not exist in Firewall config.'.format(
                group_name=group_name))
            return 404
        result = self.put(api_url, data)
        return result

    def create_service_group(self, group_name, data):
        """
        Create service group

        :param group_name: Service group to be created
        :param data: JSON Data with which to create the service group

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/group/"
        if self.does_exist(api_url + group_name):
            return 424
        result = self.post(api_url, data)
        return result

    def delete_service_group(self, group_name):
        """
        Delete firewall service group

        :param group_name: Service categrory to be deleted

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/group/" + group_name
        result = self.delete(api_url)
        return result

    # Firewall Service Methods
    def get_firewall_service(self, specific=False, filters=False):
        """
        Get service object information from firewall

        :param specific: If provided, a specific object will be returned. If not, all objects will be returned.
        :param filters: If provided, the raw filter is appended to the API call.

        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/custom/"
        if specific:
            api_url += specific
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        return results

    def update_firewall_service(self, service_name, data):
        """
        Update service with provided data

        :param service_name: Service  being updated
        :param data: JSON Data with which to upate the service

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/custom/" + service_name
        # Check whether target object already exists
        if not self.does_exist(api_url):
            logging.error('Requested service "{service_name}" does not exist in Firewall config.'.format(
                service_name=service_name))
            return 404
        result = self.put(api_url, data)
        return result

    def create_firewall_service(self, service_name, data):
        """
        Create service

        :param service_name: Service to be created
        :param data: JSON Data with which to create the service

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/custom/"
        if self.does_exist(api_url + service_name):
            return 424
        result = self.post(api_url, data)
        return result

    def delete_firewall_service(self, service_name):
        """
        Delete firewall service

        :param service_name: Service categrory to be deleted

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall.service/custom/" + service_name
        result = self.delete(api_url)
        return result

    # Firewall Policy Methods
    def get_firewall_policy(self, specific=False, filters=False):
        """
        Get firewall policy information from firewall

        :param specific: If provided, a specific object will be returned. If not, all objects will be returned.
            Specific can either be the policy name, or the policy ID.
        :param filters: If provided, the raw filter is appended to the API call.

        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/policy/"
        if specific:
            if type(specific) == int:
                api_url += str(specific)
            else:
                api_url += "?filter=name==" + specific
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        if type(results) == int:
            return results
        elif len(results) == 0:
            return 404
        else:
            return results

    def update_firewall_policy(self, policy_id, data):
        """
        Update firewall policy with provided data

        :param policy_id: ID of firewall policy to be updated
        :param data: Data with which to update the firewall policy

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/policy/" + str(policy_id)
        # Check whether target object already exists
        if not self.does_exist(api_url):
            logging.error('Requested Policy ID {policy_id} does not exist in Firewall Config.'.format(
                policy_id=str(policy_id)))
            return 404
        result = self.put(api_url, data)
        return result

    def move_firewall_policy(self, policy_id, position, neighbour):
        """
        Move firewall policy to new location

        :param policy_id: ID of firewall policy being moved
        :param position: "before" or "after"
        :param neighbour: ID of policy being used as positional reference

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/policy/" + str(policy_id)
        data = "{{'action': 'move', '{position}': {neighbour}}}".format(position=position, neighbour=neighbour)
        result = self.put(api_url, data)
        return result

    def create_firewall_policy(self, policy_id, data):
        """
        Create firewall Policy

        :param policy_id: ID of policy to be created
        :param data: Data with which to create policy

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/policy/"
        # Check whether object already exists
        if self.does_exist(api_url + str(policy_id)):
            return 424
        result = self.post(api_url, "{{'json': {data}}}".format(data=data))
        return result

    def delete_firewall_policy(self, policy_id):
        """
        Delete firewall policy

        :param policy_id: ID of policy to be deleted

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/policy/" + str(policy_id)
        result = self.delete(api_url)
        return result

    # SNMPv2 Community Methods
    def get_snmp_community(self, specific=False, filters=False):
        """
        Get SNMP community information from firewall

        :param specific: If provided, a specific object will be returned. If not, all objects will be returned.
            Specific can either be the Community string, or its internal ID.
        :param filters: If provided, the raw filter is appended to the API call.
        
        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/system.snmp/community/"
        if specific:
            if type(specific) == int:
                api_url += str(specific)
            else:
                api_url += "?filter=name==" + specific
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        return results

    def update_snmp_community(self, community_id, data):
        """
        Update SNMP community with provided data

        :param community_id: ID of community  being updated
        :param data: JSON Data with which to update the community

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/system.snmp/community/" + str(community_id)
        # Check whether target object already exists
        if not self.does_exist(api_url):
            logging.error('Requested SNMP Community ID "{community_id}" does not exist in Firewall config.'.format(
                community_id=community_id))
            return 404
        result = self.put(api_url, data)
        return result

    def create_snmp_community(self, community_id, data):
        """
        Create SNMP community

        :param community_id: ID of the SNMP Community to be created
        :param data: JSON Data with which to create the SNMP community

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/system.snmp/community/"
        if self.does_exist(api_url + str(community_id)):
            return 424
        result = self.post(api_url, data)
        return result

    def delete_snmp_community(self, community_id):
        """
        Delete SNMP community

        :param community_id: ID of the SNMP Community to be deleted

        :return: HTTP Status Code
        """
        api_url = self.urlbase + "api/v2/cmdb/system.snmp/community/" + str(community_id)
        result = self.delete(api_url)
        return result

    # ISDB read
    def get_internet_services(self, specific=False, filters=False):
        """
        Get ISDB (internet services database)

        :param specific: If provided, a specific object will be returned. 
        :param filters: If provided, the raw filter is appended to the API call.
        
        :return: JSON data for all objects in scope of request, nested in a list.
        """
        api_url = self.urlbase + "api/v2/cmdb/firewall/internet-service/"
        if specific:
            api_url += str(specific)
        elif filters:
            api_url += "?filter=" + filters
        results = self.get(api_url)
        return results