Examples¶
Runnable example scripts live in the examples/ directory of the repository (and ship in the source distribution). Each is a small, standalone program you can read top-to-bottom and run against a PCE.
Running an example¶
Install the client, then provide connection details via environment variables or
a .env file (copy examples/.env.example):
pip install illumio-py-open
PCE_HOST=my.pce.com
PCE_PORT=443
PCE_ORG_ID=1
API_KEY=api_xxxxxxxx
API_SECRET=xxxxxxxx
Run an example from inside the examples/ directory (they share
_common.py for the connection):
cd examples
python 01_connect.py
Note
Write examples create clearly-named Example* objects and clean up after
themselves. Security-policy objects are created in draft and are not
enforced until provisioned. In Illumio rules, consumers are the sources
that initiate connections to providers (the destinations).
The scripts¶
Script |
What it shows |
Writes? |
|---|---|---|
Connect and read basic facts |
read-only |
|
Labels CRUD + label groups |
creates/deletes |
|
IP lists and services |
creates/deletes |
|
Query workloads; create one; enforcement mode |
creates/deletes |
|
Build an allow policy: labels → ruleset → rule |
creates/deletes |
|
Deny and override-deny rules |
creates/deletes |
|
Explorer async traffic query |
read-only |
|
draft → active provisioning workflow |
creates + provisions |
|
Inspect VENs, statistics, software releases |
read-only |
|
Bulk create/delete workloads |
creates/deletes |
|
Org settings, password policy, reports |
read-only |
|
Pairing profile + pairing key |
creates/deletes |
|
Bulk-onboard unmanaged workloads + labels, rate-limit-safe |
creates/deletes |
Selected examples¶
Connecting to the PCE (01_connect.py):
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Connect to the PCE and read a few basic facts. Read-only."""
from _common import connect
def main():
pce = connect()
print("Connected to the PCE (org {}).".format(pce.org_id))
# A cheap read to confirm the API works end to end.
labels = pce.labels.get(params={"max_results": 5})
print("First few labels:")
for label in labels:
print(" {}={} {}".format(label.key, label.value, label.href))
# Count workloads without pulling them all back.
workloads = pce.workloads.get(params={"max_results": 1})
print("Workloads reachable:", "yes" if workloads is not None else "no")
if __name__ == "__main__":
main()
Building an allow policy (05_allow_rules.py):
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Build an allow policy end to end: labels -> ruleset -> rule.
WRITES: creates draft labels, a ruleset, and an allow rule, then deletes them.
Direction reminder: consumers are the *sources* that initiate connections to
providers (the *destinations*).
"""
from illumio import Label, LabelSet, RuleSet, Rule
from _common import connect
def main():
pce = connect()
# Scope labels for the ruleset (app / env / loc define where it applies).
app = pce.labels.create(Label(key="app", value="A-Example-Shop"))
env = pce.labels.create(Label(key="env", value="E-Example-Prod"))
loc = pce.labels.create(Label(key="loc", value="L-Example-AWS"))
web = pce.labels.create(Label(key="role", value="R-Example-Web"))
db = pce.labels.create(Label(key="role", value="R-Example-DB"))
ruleset = pce.rule_sets.create(
RuleSet(name="RS-Example-Shop", scopes=[LabelSet(labels=[app, env, loc])])
)
print("Created ruleset:", ruleset.href)
# Allow the web tier (consumers/source) to reach the DB tier (providers/dest)
# on PostgreSQL.
rule = Rule.build(
providers=[db],
consumers=[web],
ingress_services=[{"port": 5432, "proto": "tcp"}],
)
rule = pce.rules.create(rule, parent=ruleset)
print("Created allow rule:", rule.href)
# (To enforce, provision the ruleset — see 10_provisioning.py.)
# Clean up the drafts.
pce.rule_sets.delete(ruleset)
for label in (web, db, app, env, loc):
pce.labels.delete(label)
print("Cleaned up example ruleset, rule, and labels.")
if __name__ == "__main__":
main()
Deny and override-deny rules (06_deny_rules.py):
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Create deny and override-deny rules within a ruleset.
WRITES: creates a draft ruleset with two deny rules, then deletes them.
There is a single deny-rule object with an ``override`` flag. Precedence is
override-deny > allow > deny. Override-deny rules are created through
``pce.deny_rules`` (there is no separate override_deny_rules collection).
Direction: consumers (sources) are blocked from reaching providers (destinations).
"""
from illumio import RuleSet, LabelSet, DenyRule, OverrideDenyRule, AMS
from _common import connect
def main():
pce = connect()
ruleset = pce.rule_sets.create(
RuleSet(name="RS-Example-Deny", scopes=[LabelSet(labels=[])]) # empty scope = all
)
any_ip = pce.get_default_ip_list()
print("Created ruleset:", ruleset.href)
# Ordinary deny rule: block SSH from anywhere to all workloads.
deny = DenyRule.build(
providers=[AMS], # all workloads (destinations)
consumers=[any_ip.href], # any source
ingress_services=[{"port": 22, "proto": "tcp"}],
)
deny = pce.deny_rules.create(deny, parent=ruleset)
print("Created deny rule (override={}): {}".format(deny.override, deny.href))
# Override-deny rule: highest precedence, block RDP even if an allow rule exists.
override = OverrideDenyRule.build(
providers=[AMS],
consumers=[any_ip.href],
ingress_services=[{"port": 3389, "proto": "tcp"}],
)
override = pce.deny_rules.create(override, parent=ruleset)
print("Created override-deny rule (override={}): {}".format(override.override, override.href))
# A ruleset exposes both kinds in a single deny_rules array.
fetched = pce.rule_sets.get_by_reference(ruleset.href)
for r in (fetched.deny_rules or []):
print(" deny_rules entry override={} {}".format(r.override, r.href))
# Clean up.
pce.rule_sets.delete(ruleset)
print("Cleaned up example ruleset and deny rules.")
if __name__ == "__main__":
main()
Onboarding unmanaged workloads at scale (13_onboard_unmanaged_workloads.py).
This is the common “sync” pattern — prefetch the label map once, get-or-create
labels from it, and bulk_create the workloads — which avoids the per-label
GET loop that causes HTTP 429 rate-limit errors:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Onboard many unmanaged workloads and label them — the rate-limit-safe way.
This is the common "sync" use case: you have a list of things (load balancers,
network devices, cloud instances) to represent in the PCE as unmanaged workloads,
each with a set of labels.
The efficient, rate-limit-safe pattern is:
1. Fetch all existing labels ONCE and build a {(key, value): Label} map.
2. Resolve/create each workload's labels from that map (only create the
missing ones) — no per-label GET in a loop.
3. Create the workloads with ``bulk_create`` (one call for the whole batch).
Doing a ``GET /labels?key=..&value=..`` for every label of every workload is what
triggers HTTP 429 rate-limit errors at scale — this avoids it entirely.
WRITES: creates labels (only missing ones) and a batch of unmanaged workloads,
then cleans up the workloads it created.
"""
from illumio import Label, Workload
from _common import connect
# The inventory you want to represent as unmanaged workloads. In a real job this
# comes from your CMDB / config files / cloud API.
INVENTORY = [
{"name": "lb-web-01", "hostname": "lb-web-01.example.com", "public_ip": "10.30.0.11",
"labels": {"role": "R-LoadBalancer", "app": "A-Shop", "env": "E-Prod", "loc": "L-AWS"}},
{"name": "lb-web-02", "hostname": "lb-web-02.example.com", "public_ip": "10.30.0.12",
"labels": {"role": "R-LoadBalancer", "app": "A-Shop", "env": "E-Prod", "loc": "L-AWS"}},
{"name": "lb-api-01", "hostname": "lb-api-01.example.com", "public_ip": "10.30.1.11",
"labels": {"role": "R-LoadBalancer", "app": "A-API", "env": "E-Prod", "loc": "L-AWS"}},
{"name": "cache-01", "hostname": "cache-01.example.com", "public_ip": "10.30.2.11",
"labels": {"role": "R-Cache", "app": "A-Shop", "env": "E-Prod", "loc": "L-AWS"}},
]
def build_label_map(pce):
"""Fetch every label once and index it by (key, value). One request instead
of one per label lookup. For very large orgs, ``pce.labels.get_async()`` runs
it as a single PCE job."""
all_labels = pce.labels.get(params={"max_results": 100000})
return {(label.key, label.value): label for label in all_labels}
def get_or_create_label(pce, label_map, key, value):
"""Resolve a label from the in-memory map, creating it only if missing."""
label = label_map.get((key, value))
if label is None:
label = pce.labels.create(Label(key=key, value=value))
label_map[(key, value)] = label # keep the map current for the rest of the run
return label
def main():
pce = connect()
# 1) Prefetch labels once.
label_map = build_label_map(pce)
print("Prefetched {} existing labels.".format(len(label_map)))
# 2) Build the workload objects, resolving labels from the map.
workloads = []
for item in INVENTORY:
label_refs = [
get_or_create_label(pce, label_map, key, value)
for key, value in item["labels"].items()
]
workloads.append(Workload(
name=item["name"],
hostname=item["hostname"],
public_ip=item["public_ip"],
labels=label_refs,
))
# 3) Create the whole batch in one call. bulk_create returns per-item results;
# check for errors rather than assuming success.
results = pce.workloads.bulk_create(workloads)
created = [r.get("href") for r in results if isinstance(r, dict) and r.get("href")]
errors = [r for r in results if isinstance(r, dict) and r.get("errors")]
print("Created {} unmanaged workloads; {} error(s).".format(len(created), len(errors)))
for href in created:
print(" ", href)
for err in errors:
print(" ERROR:", err.get("errors"))
# Clean up the workloads this example created (leaves labels in place).
if created:
pce.workloads.bulk_delete(created)
print("Cleaned up example workloads.")
if __name__ == "__main__":
main()