Introduction

Intrusion policies are able to monitor traffic when associated with an access rule. The access rule properties must include ruleAction=PERMIT, and eventLogAction=LOG_BOTH in order for the intrusion policy to work correctly, and for intrusion events to be created.

Threat license entitlement must be enabled before intrusion policies can be attached to access rules.

To Configure an Access Rule with an Intrusion Policy

Step 1

Send an intrusion policy GET request to obtain the policy.

In this example, the request URL includes a filter parameter so that only the policy with name="Security Over Connectivity" is returned.

Bash_Request

Python_Request

Response

Copycurl -H 'Accept: application/json'  -H "Authorization: Bearer <ACCESS_TOKEN_GOES_HERE>"  "https://ftd.example/api/fdm/latest/policy/intrusionpolicies?filter=name:Security%20Over%20Connectivity"

Copydef get_intrusion_policy(host, port, access_token, policy_name):
    """
    Requires Python v3.0 or greater and requests lib.
    Send an intrusion policy GET request.
    :param host: ftd host address
    :param port: ftd port
    :param access_token: OAUTH token for device access
    :param policy_name: URL encoded name of the policy to retrieve
    :return: intrusion policy object
    """
    headers = {
        "Accept": "application/json",
        "Authorization": "Bearer {}".format(access_token)
    }
    intrusion_policy = None
    intrusion_policy_url = 'api/fdm/latest/policy/intrusionpolicies?filter=name:{}'.format(policy_name)
    response = requests.get(
        'https://{host}:{port}/{url}'.format(host=host, port=port, url=intrusion_policy_url),
        verify=False, headers=headers)
    if response.status_code != 200:
        print("Failed GET intrusion policy response {} {}".format(response.status_code, response.json()))
    else:
        intrusion_policy = response.json().get('items')[0]
        print('Intrusion policy found: {}'.format(str(intrusion_policy)))
    return intrusion_policy

Copy{
  "items" : [ {
    "version" : "oq7siofnpzswb",
    "name" : "Security Over Connectivity",
    "description" : "Security Over Connectivity Layer",
    "rules" : {
      "links" : {
        "self" : "https://ftd.example/api/fdm/latest/policy/intrusionpolicies/cea80e1c-d662-11e9-8607-6fbc78ab2e13/intrusionrules"
      }
    },
    "id" : "cea80e1c-d662-11e9-8607-6fbc78ab2e13",
    "type" : "intrusionpolicy",
    "links" : {
      "self" : "https://ftd.example/api/fdm/latest/policy/intrusionpolicies/cea80e1c-d662-11e9-8607-6fbc78ab2e13"
    }
  } ],
  "paging" : {
    "prev" : [ ],
    "next" : [ ],
    "limit" : 10,
    "offset" : 0,
    "count" : 1,
    "pages" : 0
  }
}

Step 2

Send an access rule POST request to create a PERMIT access rule with attached intrusion policy. Event logging should be set to LOG_BOTH. Include the URL parameter at=0 to position the rule at the top of the list, to ensure that all traffic uses the intrusion policy.

Bash_Request

Python_Request

JSON_Text

Response

Copycurl -H "Accept: application/json" -H "Content-Type: application/json" -H "Authorization: Bearer <ACCESS_TOKEN_GOES_HERE>" -d @json.txt "https://ftd.example/api/fdm/latest/policy/accesspolicies/default/accessrules?at=0"

Copydef post_access_rule(host, port, access_token, access_rule):
    """
    Requires Python v3.0 or greater and requests lib.
    Send an access rules POST request.
    :param host: ftd host address
    :param port: ftd port
    :param access_token: OAUTH token for device access
    :param access_rule: object representing the access rule
    :return: True if successful, otherwise False
    """
    result = True
    headers = {
        "Accept": "application/json",
        "Authorization": "Bearer {}".format(access_token)
    }
    access_rule_url = 'api/fdm/latest/policy/accesspolicies/default/accessrules?at=0'
    response = requests.post(
        'https://{host}:{port}/{url}'.format(host=host, port=port, url=access_rule_url),
        data=json.dumps(access_rule), verify=False, headers=headers)
    if response.status_code != 200:
        print("Failed POST access rule response {} {}".format(response.status_code, response.json()))
        result = False
    else:
        print(response.json())
    return result

Copy{
    "type" : "accessrule",
    "name" : "myName",
    "ruleAction" : "PERMIT",
    "eventLogAction" : "LOG_BOTH",
    "intrusionPolicy" : {
        "type" : "intrusionpolicy",
        "id" : "cea80e1c-d662-11e9-8607-6fbc78ab2e13"
    }
}

Copy{
  "version" : "j4gfbq2t3w3fa",
  "name" : "myName",
  "ruleId" : 268435464,
  "sourceZones" : [ ],
  "destinationZones" : [ ],
  "sourceNetworks" : [ ],
  "destinationNetworks" : [ ],
  "sourcePorts" : [ ],
  "destinationPorts" : [ ],
  "ruleAction" : "PERMIT",
  "eventLogAction" : "LOG_BOTH",
  "identitySources" : [ ],
  "users" : [ ],
  "embeddedAppFilter" : null,
  "urlFilter" : null,
  "intrusionPolicy" : {
    "version" : "oq7siofnpzswb",
    "name" : "Security Over Connectivity",
    "id" : "cea80e1c-d662-11e9-8607-6fbc78ab2e13",
    "type" : "intrusionpolicy"
  },
  "filePolicy" : null,
  "logFiles" : false,
  "syslogServer" : null,
  "destinationDynamicObjects" : [ ],
  "sourceDynamicObjects" : [ ],
  "id" : "02b0d027-dbbb-11e9-88f6-7bc30e26b0d9",
  "type" : "accessrule",
  "links" : {
    "self" : "https://ftd.example/api/fdm/latest/policy/accesspolicies/default/accessrules/02b0d027-dbbb-11e9-88f6-7bc30e26b0d9"
  }
}

Step 3

Perform a deployment to commit the intrusion policy changes.

Complete Python Access Rule With Intrusion Policy Script

Combining all of the Python functions from the previous steps

Complete_Python_Script

Output

Copy'''
Copyright (c) 2019 Cisco and/or its affiliates.

This software is licensed to you under the terms of the Cisco Sample
Code License, Version 1.1 (the "License"). A copy of the License
can be found in the LICENSE.TXT file of this software or at

https://developer.cisco.com/site/license/cisco-sample-code-license/

All use of the material herein must be in accordance with the terms of
the License. All rights not expressly granted by the License are
reserved. Unless required by applicable law or agreed to separately in
writing, software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
express or implied.

'''

import json
import requests


def get_access_token(host, port, user, passwd):
    """
    Requires Python v3.0 or greater and requests lib.
    Login to FTD device and obtain an access token. The access token is required so that the user can
    connect to the device to send REST API requests.
    :param host: ftd host address
    :param port: ftd port
    :param user: login user name
    :param passwd: login password
    :return: OAUTH access token
    """
    access_token = None
    requests.packages.urllib3.disable_warnings()
    payload = '{{"grant_type": "password", "username": "{}", "password": "{}"}}'.format(user, passwd)
    auth_headers = {"Content-Type": "application/json", "Accept": "application/json"}
    try:
        response = requests.post("https://{}:{}/api/fdm/latest/fdm/token".format(host, port),
                                 data=payload, verify=False, headers=auth_headers)
        if response.status_code == 200:
            access_token = response.json().get('access_token')
            print("Login successful, access_token obtained {}".format(access_token))
        else:
            print("Login failed {} {}".format(response.status_code, response.json()))
    except Exception as e:
        print("Exception in POST access token request: {}".format(str(e)))
    return access_token


def get_smart_licenses(host, port, access_token):
    """
    Requires Python v3.0 or greater and requests lib.
    Send a smart licenses GET request.
    :param host: ftd host address
    :param port: ftd port
    :param access_token: OAUTH token for device access
    :return: smart licenses
    """
    headers = {
        "Accept": "application/json",
        "Authorization": "Bearer {}".format(access_token)
    }
    smart_licenses = None
    smart_license_url = 'api/fdm/latest/license/smartlicenses'
    response = requests.get(
        'https://{host}:{port}/{url}'.format(host=host, port=port, url=smart_license_url),
        verify=False, headers=headers)
    if response.status_code != 200:
        print("Failed GET smart license response {} {}".format(response.status_code, response.json()))
    else:
        smart_licenses = response.json().get('items')
        for smart_license in smart_licenses:
            print('smart license found: {}'.format(str(smart_license)))
    return smart_licenses


def post_smart_license(host, port, access_token, smart_license):
    """
    Requires Python v3.0 or greater and requests lib.
    Send a smart license POST request
    :param host: ftd host address
    :param port: ftd port
    :param access_token: OAUTH token for device access
    :param smart_license: the smart license to be created
    :return: True if successful, otherwise False
    """
    result = True
    headers = {
        "Accept": "application/json",
        "Authorization": "Bearer {}".format(access_token)
    }
    smart_license_url = 'api/fdm/latest/license/smartlicenses'
    response = requests.post(
        'https://{host}:{port}/{url}'.format(host=host, port=port, url=smart_license_url),
        data=json.dumps(smart_license), verify=False, headers=headers)
    if response.status_code != 200 and response.status_code != 204:
        print("Failed POST smart license response {} {}".format(response.status_code, response.json()))
        result = False
    elif response.status_code == 200:
        print(response.json())
    return result


def get_intrusion_policy(host, port, access_token, policy_name):
    """
    Requires Python v3.0 or greater and requests lib.
    Send an intrusion policy GET request.
    :param host: ftd host address
    :param port: ftd port
    :param access_token: OAUTH token for device access
    :param policy_name: URL encoded name of the policy to retrieve
    :return: intrusion policy object
    """
    headers = {
        "Accept": "application/json",
        "Authorization": "Bearer {}".format(access_token)
    }
    intrusion_policy = None
    intrusion_policy_url = 'api/fdm/latest/policy/intrusionpolicies?filter=name:{}'.format(policy_name)
    response = requests.get(
        'https://{host}:{port}/{url}'.format(host=host, port=port, url=intrusion_policy_url),
        verify=False, headers=headers)
    if response.status_code != 200:
        print("Failed GET intrusion policy response {} {}".format(response.status_code, response.json()))
    else:
        intrusion_policy = response.json().get('items')[0]
        print('Intrusion policy found: {}'.format(str(intrusion_policy)))
    return intrusion_policy


def post_access_rule(host, port, access_token, access_rule):
    """
    Requires Python v3.0 or greater and requests lib.
    Send an access rules POST request.
    :param host: ftd host address
    :param port: ftd port
    :param access_token: OAUTH token for device access
    :param access_rule: object representing the access rule
    :return: True if successful, otherwise False
    """
    result = True
    headers = {
        "Accept": "application/json",
        "Authorization": "Bearer {}".format(access_token)
    }
    access_rule_url = 'api/fdm/latest/policy/accesspolicies/default/accessrules?at=0'
    response = requests.post(
        'https://{host}:{port}/{url}'.format(host=host, port=port, url=access_rule_url),
        data=json.dumps(access_rule), verify=False, headers=headers)
    if response.status_code != 200:
        print("Failed POST access rule response {} {}".format(response.status_code, response.json()))
        result = False
    else:
        print(response.json())
    return result


def main():
    """
    End to end example of code that creates an access rule with an intrusion policy.
    Requires Python v3.0 or greater and the reqeusts library.
    You must update the values for host, port, user, and password to connect to your device.
    """
    host = 'ftd.example'
    port = '443'
    user = 'admin'
    passwd = 'Admin123'
    access_token = get_access_token(host, port, user, passwd)
    if not access_token:
        print("Unable to obtain an access token. Did you remember to set host, port, user, and password?")
        return
    smart_licenses = get_smart_licenses(host, port, access_token)
    for smart_license in smart_licenses:
        if smart_license['licenseType'] == 'THREAT':
            print('threat license found')
            break
    else:
        threat_license = {
            'type': 'license',
            'count': 1,
            'licenseType': 'THREAT'
        }
        result = post_smart_license(host, port, access_token, threat_license)
        if not result:
            print('Unable to post threat license')
    if not smart_licenses:
        print('Unable to get smart licenses')
        return
    intrusion_policy = get_intrusion_policy(host, port, access_token, 'Security%20Over%20Connectivity')
    if not intrusion_policy:
        print('Unable to get intrusion policy')
        return
    access_rule = {
        "type": "accessrule",
        "name": "myName",
        "ruleAction": "PERMIT",
        "eventLogAction": "LOG_BOTH",
        "intrusionPolicy": {
            "type": "intrusionpolicy",
            "id": intrusion_policy['id']
        }
    }
    result = post_access_rule(host, port, access_token, access_rule)
    if not result:
        print('Unable to post access rule')
        return


if __name__ == '__main__':
    main()

CopyLogin successful, access_token obtained eyJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1NzA4MTc3NDAsInN1YiI6ImFkbWluIiwianRpIjoiMjE5ZWIzMWUtZWM1My0xMWU5LWEwYzItZTE2OGM5ZDliOGEzIiwibmJmIjoxNTcwODE3NzQwLCJleHAiOjE1NzA4MTk1NDAsInJlZnJlc2hUb2tlbkV4cGlyZXNBdCI6MTU3MDgyMDE0MDIwMCwidG9rZW5UeXBlIjoiSldUX0FjY2VzcyIsInVzZXJVdWlkIjoiMjcxYmNkOTEtYWNjYy0xMWU5LWIxOTUtNDlmNzk4YTg1NTk2IiwidXNlclJvbGUiOiJST0xFX0FETUlOIiwib3JpZ2luIjoicGFzc3dvcmQiLCJ1c2VybmFtZSI6ImFkbWluIn0.yE3zzE0xRJX-gHEedaVdlEp5_ItI8xCdVz_J_i1Nfrc
smart license found: {'links': {'self': 'ftd.example/api/fdm/latest/license/smartlicenses/fa99fd6e-ad52-11e9-a0c2-019387955800'}, 'type': 'license', 'version': 'jlp2ytr2rklej', 'count': 1, 'licenseType': 'BASE', 'compliant': True, 'id': 'fa99fd6e-ad52-11e9-a0c2-019387955800'}
{'links': {'self': 'https://ftd.example/api/fdm/latest/license/smartlicenses/23045da0-ec53-11e9-a0c2-7d169fa07622'}, 'type': 'license', 'version': 'iez6zkh7xx5qm', 'count': 1, 'licenseType': 'THREAT', 'compliant': True, 'id': '23045da0-ec53-11e9-a0c2-7d169fa07622'}
Intrusion policy found: {'links': {'self': 'https://ftd.example/api/fdm/latest/policy/intrusionpolicies/9cc5d466-accb-11e9-9a9e-77482da9acbd'}, 'type': 'intrusionpolicy', 'rules': {'links': {'self': 'https://ftd.example/api/fdm/latest/policy/intrusionpolicies/9cc5d466-accb-11e9-9a9e-77482da9acbd/intrusionrules'}}, 'version': 'drtz5kpzg3x66', 'id': '9cc5d466-accb-11e9-9a9e-77482da9acbd', 'name': 'Security Over Connectivity', 'description': 'Security Over Connectivity Layer'}
{'destinationPorts': [], 'embeddedAppFilter': None, 'destinationNetworks': [], 'syslogServer': None, 'ruleId': 268435459, 'id': '23425103-ec53-11e9-a0c2-27cb60e11963', 'sourceZones': [], 'ruleAction': 'PERMIT', 'sourceDynamicObjects': [], 'destinationDynamicObjects': [], 'users': [], 'type': 'accessrule', 'links': {'self': 'https://ftd.example/api/fdm/latest/policy/accesspolicies/default/accessrules/23425103-ec53-11e9-a0c2-27cb60e11963'}, 'sourceNetworks': [], 'logFiles': False, 'identitySources': [], 'filePolicy': None, 'version': 'k34hyz6b5qb4r', 'sourcePorts': [], 'destinationZones': [], 'eventLogAction': 'LOG_BOTH', 'intrusionPolicy': {'type': 'intrusionpolicy', 'id': '9cc5d466-accb-11e9-9a9e-77482da9acbd', 'name': 'Security Over Connectivity', 'version': 'drtz5kpzg3x66'}, 'urlFilter': None, 'name': 'myName'}