Developer Interface v1.1.4#

PCE Interface#

PolicyComputeEngine#

The PolicyComputeEngine object provides the core interface for interacting with PCE API endpoints.

class illumio.PolicyComputeEngine[source]#

The REST client core for the Illumio Policy Compute Engine.

Contains request logic for API calls and handles the HTTP(S) connection to the PCE.

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=8443, org_id=12)
>>> pce.set_credentials('api_key', 'api_secret')
>>> workloads = pce.workloads.get(
...     params={
...         'managed': True,
...         'enforcement_mode': 'visibility_only'
...     }
... )
>>> workloads
[
    Workload(href='/orgs/12/workloads/c754a713-2bde-4427-af1f-bff145be509b', ...),
    ...
]
Parameters
  • url (str) – PCE URL. May include http:// or https:// as the scheme.

  • port (str, optional) – PCE http(s) port. Defaults to ‘443’.

  • version (str, optional) – The PCE API version to use. Defaults to ‘v2’.

  • org_id (str, optional) – The PCE organization ID. Defaults to ‘1’.

  • retry_count (int, optional) – Number of times to retry on failure. Defaults to 5.

  • request_timeout (int, optional) – HTTP request timeout in seconds. Defaults to 30.

base_url#

DEPRECATED in v1.0.3. The base URL for API calls to the PCE. Has the form http[s]://<DOMAIN_NAME>:<PORT>/api/<API_VERSION>

include_org#

flag denoting whether to prepend the orgs subpath to request endpoints by default. Defaults to True.

org_id#

the PCE organization ID.

set_credentials(username, password)[source]#

Sets username and password values to authenticate with the PCE.

The provided credentials can be either user credentials or an API key/secret pair.

Parameters
  • username (str) – username or API key.

  • password (str) – password or API secret.

Return type

None

set_proxies(http_proxy=None, https_proxy=None)[source]#

Sets HTTP proxies to use when connecting to the PCE.

Parameters
  • http_proxy (str, optional) – HTTP proxy URI. Defaults to None.

  • https_proxy (str, optional) – HTTPS proxy URI. Defaults to None.

Return type

None

set_timeout(timeout)[source]#

Sets the HTTP request timeout for PCE connections.

Parameters

timeout (int) – the new request timeout in seconds.

Return type

None

set_tls_settings(verify=True, cert=None)[source]#

Sets TLS settings for PCE connections.

Parameters
  • verify (Union[bool, str], optional) – set TLS verification. A value of True or False will enable/disable verification respectively. A string value with a path to a CA bundle can also be provided. Defaults to True.

  • cert (Union[str, tuple], optional) – set client-side certificate paths. The value can be a string with a path to a PEM containing both the certificate and private key, or a tuple of strings with paths to both files. Defaults to None.

get(endpoint, **kwargs)[source]#

Makes a GET call to a given PCE endpoint.

Additional keyword arguments are passed to the requests.Request object.

Parameters

endpoint (str) – the PCE endpoint to call.

Returns

the Response object returned from a successful request.

Return type

requests.Response

post(endpoint, **kwargs)[source]#

Makes a POST call to a given PCE endpoint.

Appends ‘Content-Type: application/json’ to the request headers by default. Additional keyword arguments are passed to the requests.Request object.

Parameters

endpoint (str) – the PCE endpoint to call.

Returns

the Response object returned from a successful request.

Return type

requests.Response

put(endpoint, **kwargs)[source]#

Makes a PUT call to a given PCE endpoint.

Appends ‘Content-Type: application/json’ to the request headers by default. Additional keyword arguments are passed to the requests.Request object.

Parameters

endpoint (str) – the PCE endpoint to call.

Returns

the Response object returned from a successful request.

Return type

requests.Response

delete(endpoint, **kwargs)[source]#

Makes a DELETE call to a given PCE endpoint.

Additional keyword arguments are passed to the requests.Request object.

Parameters

endpoint (str) – the PCE endpoint to call.

Returns

the Response object returned from a successful request.

Return type

requests.Response

get_collection(endpoint, **kwargs)[source]#

Uses the PCE’s asynchronous job logic to retrieve a collection of objects.

NOTE: for large collections (surpassing 10,000 objects), this function will be extremely slow - it is recommended that callers use query filters or the max_results parameter to limit the number of results in the collection.

Parameters

endpoint (str) – the PCE endpoint to call.

Raises

IllumioApiException – if an error message (>399) status code is returned.

Returns

the Response object returned from a successful request.

Return type

requests.Response

must_connect(**kwargs)[source]#

Checks the connection to the PCE.

Additional keyword arguments are passed to the requests call.

Raises

IllumioApiException – if the connection fails.

Return type

None

check_connection(**kwargs)[source]#

Checks the connection to the PCE.

Additional keyword arguments are passed to the requests call.

Returns

True if the call is successful, otherwise False.

Return type

bool

get_default_ip_list(**kwargs)[source]#

Retrieves the “Any (0.0.0.0/0 and ::/0)” default global IP list.

Returns

decoded object representing the default global IP list.

Return type

IPList

get_default_service(**kwargs)[source]#

Retrieves the “All Services” default global Service.

Returns

decoded object representing the default global Service.

Return type

Service

generate_pairing_key(pairing_profile_href, **kwargs)[source]#

Generates a pairing key using a pairing profile.

Parameters

pairing_profile_href (str) – the HREF of the pairing profile to use.

Returns

the pairing key value.

Return type

str

get_traffic_flows(traffic_query, **kwargs)[source]#

DEPRECATED (v1.0.0). Use get_traffic_flows_async instead.

Retrieves Explorer traffic flows using the provided query.

NOTE: this function is deprecated in the Illumio REST API, and is only provided for compatibility. The Illumio Explorer REST API documentation recommends using the async traffic flow query instead, provided here as PolicyComputeEngine.get_traffic_flows_async.

See https://docs.illumio.com/core/21.5/Content/Guides/rest-api/visualization/explorer.htm#TrafficAnalysisQueries for details on traffic query parameters.

Parameters

traffic_query (TrafficQuery) – TrafficQuery object representing the query parameters.

Returns

list of TrafficFlow objects found using the

provided query.

Return type

List[TrafficFlow]

get_traffic_flows_async(query_name, traffic_query, **kwargs)[source]#

Retrieves Explorer traffic flows using the provided query.

See https://docs.illumio.com/core/21.5/Content/Guides/rest-api/visualization/explorer.htm#AsynchronousQueriesforTrafficFlows for details on async traffic query parameters.

Usage:
>>> traffic_query = TrafficQuery.build(
...     start_date="2022-02-01T00:00:00Z",
...     end_date="2022-03-01T00:00:00Z",
...     include_services=[
...         {'port': 3389, 'proto': 'tcp'}
...     ],
...     policy_decisions=['potentially_blocked', 'unknown']
... )
>>> traffic_query
TrafficQuery(
    start_date='2022-02-01T00:00:00Z',
    end_date='2022-03-01T00:00:00Z',
    sources=TrafficQueryFilterBlock(
        include=[],
        exclude=[]
    ),
    destinations=TrafficQueryFilterBlock(
        include=[],
        exclude=[]
    ),
    services=TrafficQueryServiceBlock(
        include=[
            ServicePort(
                port=3389,
                proto=6,
                ...
            )
        ],
        exclude=[]
    ),
    policy_decisions=[
        'potentially_blocked',
        'unknown'
    ],
    ...
)
>>> traffic_flows = pce.get_traffic_flows_async(
...     query_name='rdp-traffic-feb-22',
...     traffic_query=traffic_query
... )
>>> traffic_flows
[
    TrafficFlow(
        src=TrafficNode(
            ip='10.0.9.14',
            "ip_lists": [
                {
                    "name": "IPL-INTERNAL",
                    "href": "/orgs/1/sec_policy/active/ip_lists/5",
                    ...
                }
            ],
            ...
        ),
        dst=TrafficNode(
            ip='10.0.6.63',
            workload=Workload(
                href='/orgs/1/workloads/6567900a-b49f-43cc-93a7-c892da39aad1',
                name='WIN-JUMPBOX',
                ...
            ),
            ...
        ),
        service=ServicePort(port=3389, proto=6),
        num_connections=44,
        state='closed',
        timestamp_range=TimestampRange(
            first_detected='2022-02-19T09:50:17Z',
            last_detected='2022-02-19T10:12:36Z'
        ),
        policy_decision='potentially_blocked',
        flow_direction='inbound',
        ...
    ),
    ...
]
Parameters
  • query_name (str) – name for the async query job.

  • traffic_query (TrafficQuery) – TrafficQuery object representing the query parameters.

Raises

IllumioApiException – if there is an error retrieving the async job results.

Returns

list of TrafficFlow objects found using the

provided query.

Return type

List[TrafficFlow]

provision_policy_changes(change_description, hrefs, **kwargs)[source]#

Provisions policy changes for draft objects with the given HREFs.

Usage:
>>> rule_set = pce.rule_sets.create(
...     RuleSet(name='RS-RINGFENCE')
... )
>>> changeset = pce.provision_policy_changes(
...     change_description='Provision ring-fence rule set',
...     hrefs=[rule_set.href]
... )
>>> changeset
PolicyVersion(
    href='/orgs/1/sec_policy/110',
    commit_message='Provision ring-fence rule set',
    version=110,
    workloads_affected=0,
    object_counts=PolicyObjectCounts(
        label_groups=17,
        services=8,
        ...
    )
)
Parameters
  • change_description (str) – the policy change description.

  • hrefs (List[str]) – the HREFs of the draft policy objects to provision.

Raises

IllumioException – if an invalid HREF is provided.

Returns

the decoded policy version object including the changeset.

Return type

PolicyVersion

PCE Object API#

This internal class is used to represent API objects within the PCE, and provides a common CRUD interface for them.

class PolicyComputeEngine._PCEObjectAPI[source]#

Generic API for registered PCE objects.

Each registered API exposes CRUD operation functions through this common interface.

get_by_reference(reference, **kwargs)[source]#

Retrieves an object from the PCE using its HREF.

Usage:
>>> ip_list = pce.ip_lists.get_by_reference('/orgs/1/sec_policy/active/ip_lists/1')
>>> ip_list
IPList(
    name='Any (0.0.0.0/0 and ::/0)',
    ...
)
Parameters
Returns

the object json, decoded to its python equivalent.

Return type

Reference

get_by_name(name, policy_version='draft', **kwargs)[source]#

Retrieves the object from the PCE with the given name.

Parameters
  • name (str) – the exact name of the object to retrieve.

  • policy_version (str) –

Returns

the decoded object, or None if an object with the

given name wasn’t found.

Return type

Reference

get(policy_version='draft', parent=None, **kwargs)[source]#

Retrieves objects from the PCE based on the given parameters.

Keyword arguments to this function are passed to the requests.get call. See https://docs.illumio.com/core/21.5/API-Reference/index.html for details on filter parameters for collection queries.

Usage:
>>> virtual_services = pce.virtual_services.get(
...     policy_version='active',
...     params={
...         'name': 'VS-'
...     }
... )
>>> virtual_services
[
    VirtualService(
        href='/orgs/1/sec_policy/active/virtual_services/9177c75f-7b21-4bf0-8c16-2c47c1ca3252',
        name='VS-LAB-SERVICES'
        ...
    ),
    ...
]
Parameters
  • policy_version (str, optional) – if fetching security policy objects, specifies whether to fetch ‘draft’ or ‘active’ objects. Defaults to ‘draft’.

  • parent (Union[str, Reference, dict], optional) – Reference to the object’s parent. Required for some object types, such as Security Rules which must be created as children of existing RuleSets.

Returns

the returned list of decoded objects.

Return type

List[Reference]

get_all(policy_version='draft', parent=None, **kwargs)[source]#

Retrieves all objects of a given type from the PCE.

This function makes two requests, using the X-Total-Count header in the response to set the max_results parameter on the follow-up request.

Parameters
  • policy_version (str, optional) – if fetching security policy objects, specifies whether to fetch ‘draft’ or ‘active’ objects. Defaults to ‘draft’.

  • parent (Union[str, Reference, dict], optional) – Reference to the object’s parent. Required for some object types, such as Security Rules which must be created as children of existing RuleSets.

Returns

the returned list of decoded objects.

Return type

List[Reference]

get_async(policy_version='draft', parent=None, **kwargs)[source]#

Retrieves objects asynchronously from the PCE based on the given parameters.

Parameters
  • policy_version (str, optional) – if fetching security policy objects, specifies whether to fetch ‘draft’ or ‘active’ objects. Defaults to ‘draft’.

  • parent (Union[str, Reference, dict], optional) – Reference to the object’s parent. Required for some object types, such as Security Rules which must be created as children of existing Rule Sets.

Returns

the returned list of decoded objects.

Return type

List[Reference]

create(body, parent=None, **kwargs)[source]#

Creates an object in the PCE.

See https://docs.illumio.com/core/21.5/API-Reference/index.html for details on POST body parameters when creating objects.

Usage:
>>> from illumio.policyobjects import Label
>>> label = Label(key='role', value='R-DB')
>>> label = pce.labels.create(label)
>>> label
Label(
    'href': '/orgs/1/labels/14',
    'key': 'role',
    'value': 'R-DB
)
Parameters
  • body (Any) – the parameters for the newly created object.

  • parent (Union[str, Reference, dict], optional) – Reference to the object’s parent. Required for some object types, such as Security Rules which must be created as children of existing RuleSets.

Returns

the created object.

Return type

Reference

update(reference, body, **kwargs)[source]#

Updates an object in the PCE.

Successful PUT requests return a 204 No Content response.

Usage:
>>> pairing_profiles = pce.pairing_profile.get_by_name('PP-Database')
>>> existing_profile = pairing_profiles[0]
>>> update = PairingProfile(
...     name='PP-DATABASE-VENS',
...     enabled=False  # disable this profile
... )
>>> pce.pairing_profile.update(existing_profile['href'], update)
Parameters
  • reference (Union[str, Reference, dict]) – the HREF of the pairing profile to update.

  • body (Any) – the update data.

Return type

None

delete(reference, **kwargs)[source]#

Deletes an object in the PCE.

Successful DELETE requests return a 204 No Content response.

Parameters

reference (Union[str, Reference, dict]) – the HREF of the object to delete.

Return type

None

bulk_create(objects_to_create, **kwargs)[source]#

Creates a set of objects in the PCE.

NOTE: Bulk creation can currently only be applied for Security Principals,

Virtual Services and Workloads.

Parameters

objects_to_create (List[Reference]) – list of objects to update.

Returns

a list containing HREFs of created objects

as well as any errors returned from the PCE. Has the following form:

>>> [
...     {
...         'href': {object_href},
...         'errors': [
...             {
...                 'token': {error_type},
...                 'message': {error_message}
...             }
...         ]
...     }
... ]

Return type

List[dict]

bulk_update(objects_to_update, **kwargs)[source]#

Updates a set of objects in the PCE.

NOTE: Bulk updates can currently only be applied for Virtual Services and Workloads.

Parameters

objects_to_update (List[Reference]) – list of objects to update.

Returns

a list containing HREFs of updated objects

as well as any errors returned from the PCE. Has the following form:

>>> [
...     {
...         'href': {object_href},
...         'errors': [
...             {
...                 'token': {error_type},
...                 'message': {error_message}
...             }
...         ]
...     }
... ]

Return type

List[dict]

bulk_delete(refs, **kwargs)[source]#

Deletes a set of objects in the PCE.

NOTE: Bulk updates can currently only be applied for Workloads.

Parameters

refs (List[Union[str, Reference, dict]]) – list of references to objects to delete.

Returns

a list containing any errors that occurred during

the bulk operation. Has the following form:

>>> [
...     {
...         'href': {object_href},
...         'errors': [
...             {
...                 'token': {error_type},
...                 'message': {error_message}
...             }
...         ]
...     }
... ]

Return type

List[dict]

PolicyComputeEngine API Attributes#

The PolicyComputeEngine class provides the following attributes:

container_clusters
container_workload_profiles
enforcement_boundaries
events
ip_lists
label_groups
labels
pairing_profiles
rule_sets
rules
security_principals
service_bindings
services
users
vens
virtual_services
workloads

Each represents a corresponding PCE API endpoint, implemented as a _PCEObjectAPI instance.

Exceptions#

The library uses two exception types to capture errors returned from the API or encountered in other library functions.

exception illumio.IllumioException[source]#

Superclass for Illumio library exceptions.

exception illumio.IllumioApiException[source]#

Superclass for exceptions generated by the Illumio API code.

exception illumio.IllumioIntegerValidationException[source]#

Raised when invalid integer values are provided.

Events#

class illumio.Event[source]#

Represents an event object in the PCE.

NOTE: Events are read-only via the PCE API.

See https://docs.illumio.com/core/21.5/Content/Guides/rest-api/pce-management/events.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> events = pce.events.get(params={'max_results': 5})
>>> events
[
    Event(
        href="/orgs/1/events/3764a636-4846-492f-a090-ae96cf33bddf",
        event_type="system_task.expire_service_account_api_keys",
        timestamp="2022-08-17T22:12:37.410Z",
        pce_fqdn="pce.company.com",
        severity="critical",
        status="success",
        created_by={"system": {}},
        action=ActionEvent(
            uuid="1ebad14e-bd87-42b1-ae9a-22433951fbd3",
            src_ip="FILTERED",
            ...
        ),
        resource_changes=[],
        notifications=[],
    ),
    ...
]

Workloads and VENs#

Workloads#

class illumio.workloads.Workload[source]#

Represents a workload in the PCE.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/workloads/_ch-workloads.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> role_label = pce.labels.create({'key': 'role', 'value': 'R-Web'})
>>> app_label = pce.labels.create({'key': 'app', 'value': 'A-App'})
>>> env_label = pce.labels.create({'key': 'env', 'value': 'E-Dev'})
>>> loc_label = pce.labels.create({'key': 'loc', 'value': 'L-NYC'})
>>> workload = illumio.Workload(
...     name='Web 01',
...     hostname='web01.lab.company.com',
...     public_ip='10.8.17.229',
...     labels=[role_label, app_label, env_label, loc_label],
...     interfaces=[
...         illumio.Interface(
...             name='lo0',
...             address='127.0.0.1',
...             link_state='up'
...         )
...     ],
...     enforcement_mode=illumio.EnforcementMode.SELECTIVE,
...     online=True
... )
>>> workload = pce.workloads.create(workload)
>>> workload
Workload(
    href='/orgs/1/workloads/572eb23e-a891-42b5-b488-cd9ffe3622f5',
    name='Web 01',
    hostname='web01.lab.company.com',
    public_ip='10.8.17.229',
    labels=[
        Reference(href='/orgs/1/labels/11'),
        ...
    ],
    interfaces=[
        Interface(
            name='lo0',
            address='127.0.0.1',
            link_state='up'
        )
    ],
    enforcement_mode='selective',
    online=True,
    ...
)

VENs#

class illumio.workloads.VEN[source]#

Represents a Virtual Enforcement Node object in the PCE.

The VEN object represents the agent on a workload that has been paired with the PCE.

NOTE: VENs are read-only via the PCE API.

See https://docs.illumio.com/core/21.5/Content/Guides/rest-api/workloads/ven-operations.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> vens = pce.vens.get()
>>> vens
[
    VEN(
        href="/orgs/1/vens/5fe88f13-d22c-427e-be47-b9c3369e1e32",
        hostname="web01.lab.company.com",
        status="active",
        os_id="centos-x86_64-8.0",
        os_detail="5.18.11-200.fc36.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Jul 12 22:52:35 UTC 2022 (CentOS Linux release 8.5.2111)",
        os_platform="linux",
        version="21.2.5-8017",
        activation_type="pairing_key",
        labels=[
            Reference(href="/orgs/1/labels/14"),
            ...
        ],
        interfaces=[
            Interface(
                href="/orgs/1/workloads/5fe88f13-d22c-427e-be47-b9c3369e1e32/interfaces/eth0",
                name="eth0",
                address="10.88.0.3",
                cidr_block=16,
                default_gateway_address="10.88.0.1",
                network=Reference(
                    href="/orgs/1/networks/40995e57-5e83-4fc5-9492-5b0e6df32a68"
                ),
                network_detection_mode="single_private_brn",
                loopback=False,
            ),
            ...
        ],
        workloads=[
            Reference(href="/orgs/1/workloads/5fe88f13-d22c-427e-be47-b9c3369e1e32")
        ],
        last_heartbeat_at="2022-08-01T17:14:08.830Z",
        last_goodbye_at=None,
        unpair_allowed=True,
        conditions=[
            VENCondition(
                first_reported_timestamp="2022-08-01T17:31:40.669Z",
                latest_event=Event(
                    href="/orgs/1/events/0d2cec5f-5025-4e62-b867-b6e607397105",
                    notification_type="agent.missed_heartbeats",
                    severity="warning",
                    info={
                        ...
                    },
                    timestamp="2022-08-01T17:31:40.669Z",
                ),
            ),
            ...
        ],
    ),
    ...
]

Pairing Profiles#

class illumio.workloads.PairingProfile[source]#

Represents a pairing profile in the PCE.

Pairing profiles are used to configure VEN defaults and generate keys for VEN pairing on workloads that will be managed by the PCE.

See https://docs.illumio.com/asp/20.1/Content/Guides/rest-api/workloads/pairing-profiles-and-pairing-keys.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> pairing_profile = illumio.PairingProfile(
...     name='PP-DATABASE-VENS',
...     enabled=True,
...     enforcement_mode='visibility_only',
...     visibility_level='flows_summary'
... )
>>> pairing_profile = pce.pairing_profiles.create(pairing_profile)
>>> pairing_profile
PairingProfile(
    href='/orgs/1/pairing_profiles/19',
    name='PP-DATABASE-VENS',
    enabled=True,
    enforcement_mode='visibility_only',
    visibility_level='flows_summary',
    ...
)

Security Policy#

Rule Sets#

class illumio.rules.RuleSet[source]#

Represents a rule set object in the PCE.

Rule sets provide scope boundaries for security policy rules. Scopes are defined using application, environment, and location labels. Rules within the set will default to applying to workloads with these labels.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/create-security-policy/rulesets.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> app_label = pce.labels.create({'key': 'app', 'value': 'A-App'})
>>> env_label = pce.labels.create({'key': 'env', 'value': 'E-Prod'})
>>> loc_label = pce.labels.create({'key': 'loc', 'value': 'L-AWS'})
>>> ruleset = illumio.RuleSet(
...     name='RS-RINGFENCE',
...     scopes=[
...         illumio.LabelSet(
...             labels=[app_label, env_label, loc_label]
...         )
...     ]
... )
>>> ruleset = pce.rule_sets.create(ruleset)
>>> ruleset
Ruleset(
    href='/orgs/1/sec_policy/draft/rule_sets/19',
    name='RS-RINGFENCE'
)

Rules#

class illumio.rules.Rule[source]#

Represents a security rule in the PCE.

Each security rule defines one or more services on which traffic will be allowed from the defined providers to the defined consumers.

Providers and consumers can be defined using static (workload HREF) or dynamic (label, IP list) references. By default, providers and consumers are resolved as workloads.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/create-security-policy/rules.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> any_ip_list = pce.get_default_ip_list()
>>> role_label = pce.labels.create({'key': 'role', 'value': 'R-Web'})
>>> app_label = pce.labels.create({'key': 'app', 'value': 'A-App'})
>>> env_label = pce.labels.create({'key': 'env', 'value': 'E-Prod'})
>>> loc_label = pce.labels.create({'key': 'loc', 'value': 'L-AWS'})
>>> ruleset = illumio.RuleSet(
...     name='RS-LAB-ALLOWLIST',
...     scopes=[
...         illumio.LabelSet(
...             labels=[app_label, env_label, loc_label]
...         )
...     ]
... )
>>> ruleset = pce.rule_sets.create(ruleset)
>>> rule = illumio.Rule.build(
...     providers=[role_label],
...     consumers=[any_ip_list],
...     ingress_services=[
...         {'port': 80, 'proto': 'tcp'},
...         {'port': 443, 'proto': 'tcp'}
...     ],
...     unscoped_consumers=True  # creates an extra-scope rule
... )
>>> rule = pce.rules.create(rule, parent=ruleset)
>>> rule
Rule(
    href='/orgs/1/sec_policy/rule_sets/19/rules/sec_rules/34',
    enabled=True,
    providers=[
        Actor(
            label=Reference(
                href='/orgs/1/labels/21'
            ),
            ...
        )
    ],
    consumers=[
        Actor(
            ip_list=Reference(
                href='/orgs/1/sec_policy/draft/ip_lists/1'
            ),
            ...
        )
    ],
    ingress_services=[
        ServicePort(port=80, proto=6, ...),
        ServicePort(port=443, proto=6, ...)
    ],
    resolve_labels_as=LabelResolutionBlock(
        providers=['workloads'],
        consumers=['workloads']
    ),
    unscoped_consumers=True,
    ...
)

Enforcement Boundaries#

class illumio.rules.EnforcementBoundary[source]#

Represents an enforcement boundary in the PCE.

Enforcement boundaries establish deny rules for workloads within their scope communicating on its defined services.

Workloads in selective enforcement mode that fall within an enforcement boundary will have policy rules apply to them as if they were in full enforcement.

Rules allowing traffic that would otherwise be denied by an enforcement boundary will override the boundary’s deny rule.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/policy-enforcement/enforcement-boundaries.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> any_ip_list = pce.get_default_ip_list()
>>> enforcement_boundary = illumio.EnforcementBoundary.build(
...     name='EB-BLOCK-RDP',
...     providers=[AMS],  # the special 'ams' literal denotes all workloads
...     consumers=[any_ip_list.href],
...     ingress_services=[
...         {'port': 3389, 'proto': 'tcp'},
...         {'port': 3389, 'proto': 'udp'},
...     ]
... )
>>> enforcement_boundary = pce.enforcement_boundaries.create(enforcement_boundary)
>>> enforcement_boundary
EnforcementBoundary(
    href='/orgs/1/sec_policy/draft/enforcement_boundary/8',
    name='EB-BLOCK-RDP',
    providers=[
        Actor(
            actors='ams',
            ...
        )
    ],
    consumers=[
        Actor(
            ip_list=Reference(
                href='/orgs/1/sec_policy/active/ip_lists/1'
            ),
            ...
        )
    ],
    ingress_services=[
        ServicePort(port=3389, proto=6),
        ServicePort(port=3389, proto=17)
    ],
    ...
)

Policy Objects#

IP Lists#

class illumio.policyobjects.IPList[source]#

Represents an IP list in the PCE.

IP lists are list of IP addresses, subnets, CIDR blocks, and/or FQDNs. They can be used in conjunction with other security policy objects to allow or deny traffic from these defined ranges.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/security-policy-objects/ip-lists.htm

Usage:
>>> from illumio import PolicyComputeEngine, IPList, IPRange
>>> pce = PolicyComputeEngine('my.pce.com')
>>> pce.set_credentials('api_key_username', 'api_key_secret')
>>> ip_list = IPList(
...     name='IPL-INTERNAL',
...     ip_ranges=IPRange(
...         from_ip='192.168.0.0/16'
...     )
... )
>>> ip_list = pce.ip_lists.create(ip_list)
>>> ip_list
IPList(
    href='/orgs/1/sec_policy/draft/ip_lists/22',
    name='IPL-INTERNAL',
    ip_ranges=IPRange(
        from_ip='192.168.0.0/16'
    ),
    ...
)

Labels#

class illumio.policyobjects.Label[source]#

Represents a label in the PCE.

Labels help to configure the reach of policy rules in a dynamic way, without relying on precise identifiers like IP addresses.

When fetching Labels from the PCE, a breakdown of the labels’ usage can be optionally included.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/security-policy-objects/labels-and-label-groups.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> label = illumio.Label(key='role', value='R-DB')
>>> label = pce.labels.create(label)
>>> label
Label(
    href='/orgs/1/labels/18',
    key='role',
    value='R-DB',
    ...
)
class illumio.policyobjects.LabelGroup[source]#

Represents a label group in the PCE.

Label groups can contain labels and other sub-groups to define broader categories that are often grouped when writing rules or otherwise referencing multiple labels.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/security-policy-objects/labels-and-label-groups.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> dev_label = pce.labels.create({'key': 'env', 'value': 'E-Dev'})
>>> stage_label = pce.labels.create({'key': 'env', 'value': 'E-Stage'})
>>> label_group = illumio.LabelGroup(
...     key='role',
...     name='LG-E-PreProd',
...     labels=[dev_label, stage_label]
... )
>>> label_group = pce.label_groups.create(label_group)
>>> label_group
LabelGroup(
    href='/orgs/1/sec_policy/draft/label_groups/5704a6f4-e051-4f88-9149-713ee22b5d41',
    key='role',
    value='R-DB',
    ...
)
class illumio.policyobjects.LabelSet[source]#

Represents a set of labels with distinct keys.

Used to define rule set scopes.

Parameters

labels (List[Reference], optional) – list of label and label group references in the set.

Services#

class illumio.policyobjects.Service[source]#

Represents a service in the PCE.

A service can be port-based or process-based (Windows services).

Each service contains one or more objects defining the port, protocol, and/or process name used by an application running on a workload.

Service objects are used to write rules or enforcement boundaries to allow or deny traffic on its defined ports and processes for workloads in the network.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/security-policy-objects/services.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> service = illumio.Service(
...     name='S-HTTP',
...     service_ports=[
...         illumio.ServicePort(port=80, proto='tcp'),
...         illumio.ServicePort(port=443, proto='tcp')
...     ]
... )
>>> service = pce.services.create(service)
>>> service
Service(
    href='/orgs/1/sec_policy/draft/services/15',
    name='S-HTTP',
    service_ports=[
        ServicePort(
            port=80,
            proto=6,
            ...
        ),
        ...
    ],
    ...
)

Virtual Services#

class illumio.policyobjects.VirtualService[source]#

Represents a virtual service object in the PCE.

Virtual services provide an abstraction for a service that can be bound to one or more workloads. This can be useful when a workload is running multiple services, or multiple instances of the same service for different apps.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/security-policy-objects/virtual-services.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> role_label = pce.labels.create({'key': 'role', 'value': 'R-Tomcat'})
>>> app_label = pce.labels.create({'key': 'app', 'value': 'A-App'})
>>> env_label = pce.labels.create({'key': 'env', 'value': 'E-Dev'})
>>> loc_label = pce.labels.create({'key': 'loc', 'value': 'L-NYC'})
>>> tomcat_svc = pce.services.create(illumio.Service(
...     name='S-Tomcat',
...     service_ports=[
...         {
...             'port': 80,
...             'proto': illumio.convert_protocol('tcp')
...         },
...         {
...             'port': 443,
...             'proto': illumio.convert_protocol('tcp')
...         }
...     ]
... )
>>> virtual_service = illumio.VirtualService(
...     name='VS-Tomcat',
...     apply_to=illumio.ApplyTo.HOST_ONLY,
...     service=tomcat_svc,
...     labels=[role_label, app_label, env_label, loc_label]
... )
>>> virtual_service = pce.virtual_services.create(virtual_service)
>>> virtual_service
VirtualService(
{
    "href": "/orgs/1/sec_policy/draft/virtual_services/14d7ff69-2fa4-458b-a299-e3f11ffa9b01",
    "created_at": "2021-10-05T12:34:56.789Z",
    "updated_at": "2021-10-05T12:34:56.789Z",
    "deleted_at": null,
    "created_by": {
        "href": "/users/1"
    },
    "updated_by": {
        "href": "/users/1"
    },
    "deleted_by": null,
    "update_type": "create",
    "name": "VS-INTERNAL",
    "description": null,
    "pce_fqdn": null,
    "service_ports": [
        {
            "port": 1234,
            "proto": 6
        },
        {
            "port": 80,
            "proto": 17,
            "to_port": 443
        }
    ],
    "service_addresses": [
        {
            "ip": "1.1.1.1",
            "port": 101
        },
        {
            "description": "test description",
            "fqdn": "*.illumio.com"
        }
    ],
    "labels": [
        {
            "href": "/orgs/1/labels/1",
            "key": "role",
            "value": "R-WEB"
        },
        {
            "href": "/orgs/1/labels/2",
            "key": "app",
            "value": "A-HRM"
        },
        {
            "href": "/orgs/1/labels/3",
            "key": "env",
            "value": "E-STAGE"
        },
        {
            "href": "/orgs/1/labels/4",
            "key": "loc",
            "value": "L-TOR"
        }
    ],
    "ip_overrides": [
        "1.2.3.4"
    ],
    "apply_to": "host_only",
    "caps": [
        "write",
        "provision",
        "delete"
    ]
}
)
Raises

IllumioException – if an invalid apply_to value is provided.

class illumio.policyobjects.ServiceBinding[source]#

Represents a service binding between a virtual service and a workload.

The /service_bindings POST method accepts a list of bindings and returns a list containing either the binding HREF (if the workload was successfully bound to the virtual service) or an error token and message.

NOTE: Service bindings can only be created using an active virtual service.

See https://docs.illumio.com/core/21.5/Content/Guides/security-policy/security-policy-objects/virtual-services.htm#BindaVirtualServicetoaWorkload

Usage:
>>> import illumio
>>> workload = pce.workloads.create(illumio.Workload(
...     name='haproxy0.company.com',
...     hostname='haproxy0.company.com',
...     public_ip='10.0.14.201'
... ))
>>> virtual_service = pce.virtual_services.create(illumio.VirtualService(
...     name='VS-HAProxy',
...     apply_to=illumio.ApplyTo.HOST_ONLY,
...     service_ports=[{'port': 443, 'proto': convert_protocol('tcp')}]
... ))
>>> policy_version = pce.provision_policy_changes(
...     change_description="Create HAProxy virtual service",
...     hrefs=[virtual_service.href]
... )
>>> virtual_service.href = convert_draft_href_to_active(virtual_service.href)
>>> service_binding = illumio.ServiceBinding(
...     virtual_service=virtual_service,
...     workload=workload,
...     port_overrides=[illumio.PortOverride(port=443, new_port=8443, proto='tcp')]
... )
>>> bindings = pce.service_bindings.create([service_binding])
{
    'service_bindings': [
        ServiceBinding(href='/orgs/1/service_bindings/bc98cf25-8f24-4989-853c-578fa14108cf')
    ],
    'errors': []
}

Infrastructure#

Container Clusters#

class illumio.infrastructure.ContainerCluster[source]#

Represents a container cluster object in the PCE.

Container clusters are abstract representations of container orchestration systems linked to the PCE.

NOTE: when a container cluster is created through the API, the container_cluster_token used by Kubelink and C-VEN containers to pair with the PCE is returned in the response. This token is only available after the initial POST request and cannot be retrieved via the API: make sure to store it in a persistent form after creating the cluster.

See https://docs.illumio.com/core/21.5/Content/LandingPages/Guides/kubernetes-and-openshift.htm

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> container_cluster = illumio.ContainerCluster(
...     name='CC-EKS-PROD',
...     description='Production Kubernetes cluster on AWS'
... )
>>> container_cluster = pce.container_clusters.create(container_cluster)
>>> container_cluster
ContainerCluster(
    href='/orgs/1/container_clusters/f5bef182-8c55-4219-b35b-0a50b707e434',
    name='CC-EKS-PROD',
    description='Production Kubernetes cluster on AWS',
    container_cluster_token='1_016dace1ab35fafe8e71c6dda6695e0881393f1f4c494e6cd70178f1e743b372',
    ...
)
class illumio.infrastructure.ContainerWorkloadProfile[source]#

Represents a workload profile within a container cluster object in the PCE.

Workload profiles define management and scope for container workloads under a cluster namespace defined by the profile.

The assign_labels field is DEPRECATED in favour of the more flexible labels for defining label assignments and restrictions on the profile. assign_labels is left in for compatibility with older PCE versions. In either case, label assignments can only be specified for managed workload profiles.

NOTE: though the enforcement_mode value for a workload profile can be set to selective, it is currently not supported and may result in unexpected behaviour. The only supported enforcement modes for workload profiles are idle, visibility_only, and full.

Usage:
>>> import illumio
>>> pce = illumio.PolicyComputeEngine('pce.company.com', port=443, org_id=1)
>>> pce.set_credentials('api_key', 'api_secret')
>>> container_cluster = illumio.ContainerCluster(
...     name='CC-EKS-PROD',
...     description='Production Kubernetes cluster on AWS'
... )
>>> container_cluster = pce.container_clusters.create(container_cluster)
>>> env_label = pce.labels.create({'key': 'env', 'value': 'Production'})
>>> loc_label = pce.labels.create({'key': 'loc', 'value': 'AWS'})
>>> container_workload_profile = illumio.ContainerWorkloadProfile(
...     name='illumio-system',
...     managed=True,
...     labels=[
...         illumio.LabelRestriction(key='env', assignment=env_label),
...         illumio.LabelRestriction(key='loc', assignment=loc_label)
...     ],
...     enforcement_mode='visibility_only'
... )
>>> container_workload_profile = pce.container_workload_profiles.create(
...     container_workload_profile, parent=container_cluster
... )
>>> container_workload_profile
ContainerWorkloadProfile(
    href='/orgs/1/container_clusters/f5bef182-8c55-4219-b35b-0a50b707e434/container_workload_profiles/d2d466b5-106d-48e9-ada9-68f6321d1da8',
    name='illumio-system',
    namespace=None,
    managed=True,
    labels=[
        LabelRestriction(
            key='env',
            assignment=Reference(href='/orgs/1/labels/23'),
            ...
        ),
        ...
    ],
    enforcement_mode='visibility_only'
    ...
)

Explorer#

Traffic Analysis#

class illumio.explorer.TrafficQuery[source]#

Represents a query against the PCE traffic flow database.

See https://docs.illumio.com/core/22.1/Content/Guides/rest-api/visualization/explorer.htm#TrafficAnalysisQueries

Raises

IllumioException – if any of the dates provided, the date range, or the policy_decision field value are invalid.

static build(start_date, end_date, include_sources=[[]], exclude_sources=[], include_destinations=[[]], exclude_destinations=[], include_services=[], exclude_services=[], policy_decisions=[], exclude_workloads_from_ip_list_query=True, max_results=100000, query_name=None)[source]#

Constructs a TrafficQuery object based on the provided parameters.

Empty include values indicate that all sources/destinations/services should be included, while empty exclude values indicate that none should be excluded.

Parameters
  • start_date (Optional[Union[str, int, float]]) – starting datetime of the search. Can be provided as a datetime string, or a Unix timestamp integer or float in either seconds or milliseconds.

  • end_date (Optional[Union[str, int, float]]) – ending datetime of the search. Can be provided as a datetime string, or a Unix timestamp integer or float in either seconds or milliseconds.

  • include_sources (list, optional) – sources to include in the search. Provided as a list of Reference objects or strings representing label, IP list, or workload HREFs, FQDNs, IP addresses, or transmission values. Defaults to [].

  • exclude_sources (list, optional) – sources to exclude from the search. Provided as a list of Reference objects or strings representing label, IP list, or workload HREFs, FQDNs, IP addresses, or transmission values. Defaults to [].

  • include_destinations (list, optional) – destinations to include in the search. Provided as a list of Reference objects or strings representing label, IP list, or workload HREFs, FQDNs, IP addresses, or transmission values. Defaults to [].

  • exclude_destinations (list, optional) – destinations to exclude from the search. Provided as a list of Reference objects or strings representing label, IP list, or workload HREFs, FQDNs, IP addresses, or transmission values. Defaults to [].

  • include_services (list, optional) – traffic flow ServicePort objects to include in the search. Defaults to [].

  • exclude_services (list, optional) – traffic flow ServicePort objects to exclude from the search. Defaults to [].

  • policy_decisions (list, optional) – filters the search based on flow policy decision. List elements must be one of allowed, blocked, potentially_blocked, or unknown. Defaults to [].

  • exclude_workloads_from_ip_list_query (bool, optional) – if True, exclude workload traffic when an IP list is included as a source or destination. Defaults to True.

  • max_results (int, optional) – maximum number of results to return from the search. Defaults to 100000.

  • query_name (str, optional) – query name. Required for async queries. Defaults to None.

Returns

the constructed traffic query.

Return type

TrafficQuery

class illumio.explorer.TrafficFlow[source]#

Represents a traffic flow to a workload in the PCE.

Raises

IllumioException – if invalid flow_direction, policy_decision, state, or transmission values are provided.

Access Management#

Users#

class illumio.accessmanagement.User[source]#

Represents a user object in the PCE.

Utilities#

Contains global constants, helper functions, and internal structures.

Constants#

illumio.util.constants.ACTIVE = 'active'#

Active policy version path literal.

illumio.util.constants.DRAFT = 'draft'#

Draft policy version path literal.

illumio.util.constants.AMS = 'ams'#

Used in rules and enforcement boundaries to denote that all workloads should be affected.

illumio.util.constants.RESOLVE_AS_WORKLOADS = 'workloads'#

Used in resolve_labels_as block in rule creation to denote that workloads matching the rule scope should be affected.

illumio.util.constants.RESOLVE_AS_VIRTUAL_SERVICES = 'virtual_services'#

Used in resolve_labels_as block in rule creation to denote that virtual services matching the rule scope should be affected.

illumio.util.constants.ANY_IP_LIST_NAME = 'Any (0.0.0.0/0 and ::/0)'#

Name of the default global IP list.

illumio.util.constants.ALL_SERVICES_NAME = 'All Services'#

Name of the default global Service.

illumio.util.constants.PORT_MAX = 65535#

Max port number.

illumio.util.constants.ICMP_CODE_MAX = 15#

Max value for the ICMP header Code field. See https://www.iana.org/assignments/icmp-parameters/icmp-parameters.xhtml#icmp-parameters-codes

illumio.util.constants.ICMP_TYPE_MAX = 255#

Max value for the ICMP header Type field. See https://www.iana.org/assignments/icmp-parameters/icmp-parameters.xhtml#icmp-parameters-types

illumio.util.constants.BULK_CHANGE_LIMIT = 1000#

Upper limit on the number of objects that can be sent to PCE bulk change endpoints in a single request.

enum illumio.util.constants.EnforcementMode(value)[source]#

Workload enforcement mode enumeration.

Member Type

str

Valid values are as follows:

IDLE = <EnforcementMode.IDLE: 'idle'>#
VISIBILITY_ONLY = <EnforcementMode.VISIBILITY_ONLY: 'visibility_only'>#
FULL = <EnforcementMode.FULL: 'full'>#
SELECTIVE = <EnforcementMode.SELECTIVE: 'selective'>#
enum illumio.util.constants.LinkState(value)[source]#

Network interface link state enumeration.

Member Type

str

Valid values are as follows:

UP = <LinkState.UP: 'up'>#
DOWN = <LinkState.DOWN: 'down'>#
UNKNOWN = <LinkState.UNKNOWN: 'unknown'>#
enum illumio.util.constants.VisibilityLevel(value)[source]#

Workload visibility level enumeration.

Member Type

str

Valid values are as follows:

FLOW_FULL_DETAIL = <VisibilityLevel.FLOW_FULL_DETAIL: 'flow_full_detail'>#
FLOW_SUMMARY = <VisibilityLevel.FLOW_SUMMARY: 'flow_summary'>#
FLOW_DROPS = <VisibilityLevel.FLOW_DROPS: 'flow_drops'>#
FLOW_OFF = <VisibilityLevel.FLOW_OFF: 'flow_off'>#
ENHANCED_DATA_COLLECTION = <VisibilityLevel.ENHANCED_DATA_COLLECTION: 'enhanced_data_collection'>#
enum illumio.util.constants.Transmission(value)[source]#

Traffic flow transmission enumeration.

Member Type

str

Valid values are as follows:

BROADCAST = <Transmission.BROADCAST: 'broadcast'>#
MULTICAST = <Transmission.MULTICAST: 'multicast'>#
UNICAST = <Transmission.UNICAST: 'unicast'>#
enum illumio.util.constants.FlowDirection(value)[source]#

Traffic flow direction enumeration.

Member Type

str

Valid values are as follows:

INBOUND = <FlowDirection.INBOUND: 'inbound'>#
OUTBOUND = <FlowDirection.OUTBOUND: 'outbound'>#
enum illumio.util.constants.TrafficState(value)[source]#

Traffic flow state enumeration.

Member Type

str

Valid values are as follows:

ACTIVE = <TrafficState.ACTIVE: 'active'>#
CLOSED = <TrafficState.CLOSED: 'closed'>#
TIMED_OUT = <TrafficState.TIMED_OUT: 'timed out'>#
SNAPSHOT = <TrafficState.SNAPSHOT: 'snapshot'>#
NEW = <TrafficState.NEW: 'new'>#
UNKNOWN = <TrafficState.UNKNOWN: 'unknown'>#
INCOMPLETE = <TrafficState.INCOMPLETE: 'incomplete'>#
enum illumio.util.constants.ApplyTo(value)[source]#

Virtual service apply to value enumeration.

Member Type

str

Valid values are as follows:

HOST_ONLY = <ApplyTo.HOST_ONLY: 'host_only'>#
INTERNAL_BRIDGE_NETWORK = <ApplyTo.INTERNAL_BRIDGE_NETWORK: 'internal_bridge_network'>#
enum illumio.util.constants.VENType(value)[source]#

VEN type enumeration.

Member Type

str

Valid values are as follows:

SERVER = <VENType.SERVER: 'server'>#
ENDPOINT = <VENType.ENDPOINT: 'endpoint'>#
CONTAINERIZED = <VENType.CONTAINERIZED: 'containerized'>#
enum illumio.util.constants.ChangeType(value)[source]#

Resource event change type enumeration.

Member Type

str

Valid values are as follows:

CREATE = <ChangeType.CREATE: 'create'>#
UPDATE = <ChangeType.UPDATE: 'update'>#
DELETE = <ChangeType.DELETE: 'delete'>#
enum illumio.util.constants.EventSeverity(value)[source]#

Event severity enumeration.

Member Type

str

Valid values are as follows:

EMERGENCY = <EventSeverity.EMERGENCY: 'emerg'>#
ALERT = <EventSeverity.ALERT: 'alert'>#
CRITICAL = <EventSeverity.CRITICAL: 'crit'>#
ERROR = <EventSeverity.ERROR: 'err'>#
WARNING = <EventSeverity.WARNING: 'warning'>#
NOTICE = <EventSeverity.NOTICE: 'notice'>#
INFO = <EventSeverity.INFO: 'info'>#
DEBUG = <EventSeverity.DEBUG: 'debug'>#
enum illumio.util.constants.EventStatus(value)[source]#

Event status enumeration.

Member Type

str

Valid values are as follows:

SUCCESS = <EventStatus.SUCCESS: 'success'>#
FAILURE = <EventStatus.FAILURE: 'failure'>#

Helper Functions#

illumio.util.functions.convert_active_href_to_draft(href)[source]#

Given an HREF string, converts policy version to draft.

If a draft HREF is provided, this function has no effect.

Parameters

href (str) – PCE object HREF.

Returns

draft policy version HREF.

Return type

str

illumio.util.functions.convert_draft_href_to_active(href)[source]#

Given an HREF string, converts policy version to active.

If an active HREF is provided, this function has no effect.

Parameters

href (str) – PCE object HREF.

Returns

active policy version HREF.

Return type

str

illumio.util.functions.convert_protocol(protocol)[source]#

Given a protocol name, returns the integer ID of that protocol.

Usage:
>>> convert_protocol('tcp')
6
Parameters

protocol (str) – case-insensitive protocol string, e.g. ‘tcp’, ‘UDP’

Raises

IllumioException – if an invalid protocol name is provided.

Returns

the integer ID of the given protocol, e.g. 17 for ‘udp’

Return type

int

illumio.util.functions.deprecated(deprecated_in, message=None)[source]#

Deprecation decorator, adapted from https://stackoverflow.com/a/30253848 Will emit a warning when the decorated function is called.

illumio.util.functions.ignore_empty_keys(o)[source]#

Removes keys with None-type values from the provided dict.

Used for JSON encoding to avoid schema errors due to empty or invalid parameters.

Parameters

o (dict) –

illumio.util.functions.parse_url(url)[source]#

Parses given URL into its scheme and hostname, stripping port and path.

Parameters

url (str) – URL to parse.

Returns

parsed (scheme, hostname)

Return type

tuple

illumio.util.functions.pce_api(name, endpoint=None, is_sec_policy=False, is_global=False)[source]#

Decorates an IllumioObject subclass to denote it as a PCE API object type.

This registers the type in the PCE_APIS mapping used to determine whether a given name corresponds to an API-accessible type.

We can then leverage __getattr__ to instantiate a generic API interface for any registered type (with some caveats, see the _PCEObjectAPI documentation).

By default, registers the endpoint as /{name}, but the endpoint can also be specified in the decorator function call.

For example:

>>> @pce_api('labels', endpoint='/labels')
>>> class Label(IllumioObject):
...     ...
>>> pce = PolicyComputeEngine(...)
>>> # the 'labels' name is registered, and so we can
>>> # call /labels endpoints through the _PCEObjectAPI interface
>>> labels = pce.labels.get()
>>> labels
[
    Label(
        href='/orgs/1/labels/1',
        key='role',
        value='R-DB',
        ...
    ),
    ...
]
Parameters
  • name (str) – the name of the API. used as a PolicyComputeEngine attribute name to generate the API interface.

  • endpoint (str, optional) – _description_. Defaults to None.

  • is_sec_policy (bool, optional) – whether or not the object reflects a security policy API with the sec_policy/{pversion} prefix. Defaults to False.

  • is_global (bool, optional) – whether or not the object reflects a global API, such as /health or /users. These APIs operate on the entire PCE rather than a single tenant, and don’t need the /orgs/{org_id} prefix.

illumio.util.jsonutils.href_from(reference)[source]#

Attempts to parse HREF value from a provided source.

Parameters

reference (Union[Reference, dict, str]) – source reference. If a string value is passed, it is returned unchanged. The href field is returned from a Reference object or subclass instance, and the 'href' key is returned from a provided dictionary.

Raises

IllumioException – if an invalid reference type is provided, or if the href value is null or falsy.

Base Classes#

class illumio.util.jsonutils.JsonObject[source]#

Base dataclass for all derived PCE objects.

Provides custom encoding, decoding, and type validation to and from JSON.

class illumio.util.jsonutils.IllumioObject[source]#

Base class for most PCE objects.

Parameters
  • name (str, optional) – object name.

  • description (str, optional) – object description.

  • external_data_set (str, optional) – unique namespace identifier for an external source creating PCE objects. If set, external_data_reference must also be provided.

  • external_data_reference (str, optional) – unique identifier within the external_data_set. If set, external_data_set must also be provided. external_data_set + external_data_reference must be globally unique.

  • caps (List[str], optional) – defines the requesting user’s capabilities/permissions on the object.

class illumio.util.jsonutils.Reference[source]#

Simplest PCE object type, containing only an HREF.

Used in most API schema to refer to other PCE objects.

Parameters

href (str, optional) – PCE object HREF.