Python clients for accessing Cerbos.
Cerbos is the open core, language-agnostic, scalable authorization solution that makes user permissions and authorization simple to implement and manage by writing context-aware access control policies for your application resources.
This library is available from PyPI as cerbos. It supports both async and non-async modes.
pip install cerbosThere are two clients available; gRPC and HTTP. New projects should use the gRPC client.
(Available from v0.8.0 onwards)
Making a request
from cerbos.sdk.grpc.client import CerbosClient
from cerbos.engine.v1 import engine_pb2
from cerbos.request.v1 import request_pb2
from google.protobuf.struct_pb2 import Value
principal = engine_pb2.Principal(
id="john",
roles={"employee"},
policy_version="20210210",
attr={
"department": Value(string_value="marketing"),
"geography": Value(string_value="GB"),
"team": Value(string_value="design"),
},
)
resource = engine_pb2.Resource(
id="XX125",
kind="leave_request",
attr={
"id": Value(string_value="XX125"),
"department": Value(string_value="marketing"),
"geography": Value(string_value="GB"),
"team": Value(string_value="design"),
"owner": Value(string_value="john"),
}
)
plan_resource = engine_pb2.PlanResourcesInput.Resource(
kind="leave_request",
policy_version="20210210"
)
with CerbosClient("localhost:3593", tls_verify=False) as c:
# Check a single action on a single resource
if c.is_allowed("view", principal, resource):
# perform some action
pass
# Get the query plan for "view" action
plan = c.plan_resources(action="view", principal=principal, resource=plan_resource)Async usage
from cerbos.sdk.grpc.client import AsyncCerbosClient
async with AsyncCerbosClient("localhost:3593", tls_verify=False) as c:
...
allowed = await c.is_allowed("view:public", p, r)
print(allowed)
# Get the query plan for "view" action
...
plan = await c.plan_resources("view", p, rd)
print(plan.filter.to_json())Admin API
There is also a client available for interacting with the Admin API. See the docs for information on how to configure your PDP to enable this.
from cerbos.policy.v1 import policy_pb2
from cerbos.sdk.grpc.client import AdminCredentials, AsyncCerbosAdminClient
admin_credentials = AdminCredentials(username="admin", password="some_password")
async with AsyncCerbosAdminClient("localhost:3593", admin_credentials=admin_credentials) as c:
await c.add_or_update_policy(
[
policy_pb2.Policy(
api_version="api.cerbos.dev/v1",
principal_policy=policy_pb2.PrincipalPolicy(
principal="terry", version="default"
),
)
]
)Connecting to a Unix domain socket
with CerbosClient("unix:/var/cerbos.sock", tls_verify=False) as c:
...Enabling TLS
tls_verify can either be the certificate location (string) or a boolean. If True, it'll look for the file at the location specified by the environment variable SSL_CERT_FILE, else the default OS location.
with CerbosClient("localhost:3593", tls_verify=True) as c:
...with CerbosClient("localhost:3593", tls_verify="path/to/tls.crt") as c:
...Optional channel arguments
You can pass additional options in the channel_options dict.
Available options are described here.
The argument is of type dict[str, Any] where the Any value must match the expected type defined in the previous link.
IMPORTANT: We use the config key grpc.service_config to set service-specific configuration (retry policies, backoffs etc) within the nested JSON field. Passing this as a channel_options key will override that configuration entirely. We recommend leaving this untouched, however, if you need to pass custom config, ensure you pass the entire existing dict along with the desired updates (this can be found within the AsyncClientBase.__init__ method).
NOTE: We provide this as a generic method to set arbitrary options for particular use cases.
For purely demonstrative purposes, our example below overrides grpc.ssl_target_name_override, which is certainly not recommended practice for production applications.
opts = {
"grpc.ssl_target_name_override": "localhost"
}
with CerbosClient("localhost:3593", tls_verify=True, channel_options=opts) as c:
...We maintain this for backwards compatibility. It is recommended to use the gRPC client.
Making a request
from cerbos.sdk.model import *
from cerbos.sdk.client import CerbosClient
with CerbosClient("https://localhost:3592", debug=True, tls_verify=False) as c:
p = Principal(
"john",
roles={"employee"},
policy_version="20210210",
attr={"department": "marketing", "geography": "GB", "team": "design"},
)
# Check a single action on a single resource
r = Resource(
"XX125",
"leave_request",
policy_version="20210210",
attr={
"id": "XX125",
"department": "marketing",
"geography": "GB",
"team": "design",
"owner": "john",
},
)
allowed = c.is_allowed("view:public", p, r)
print(allowed)
# Get the query plan for "view" action
rd = ResourceDesc("leave_request", policy_version="20210210")
plan = c.plan_resources("view", p, rd)
print(plan.filter.to_json())Async usage
from cerbos.sdk.model import *
from cerbos.sdk.client import AsyncCerbosClient
async with AsyncCerbosClient("https://localhost:3592", debug=True, tls_verify=False) as c:
...
# Check a single action on a single resource
...
allowed = await c.is_allowed("view:public", p, r)
print(allowed)
# Get the query plan for "view" action
...
plan = await c.plan_resources("view", p, rd)
print(plan.filter.to_json())Connecting to a Unix domain socket
Use unix+http:///path/to/sock for HTTP over UDS or unix+https:///path/to/sock for HTTPS over UDS.
with CerbosClient("unix+https:///var/cerbos.sock", debug=True, tls_verify=False) as c:
...Testing with TestContainers
NOTE: Requires cerbos[testcontainers] dependency to be installed.
from cerbos.sdk.client import CerbosClient
from cerbos.sdk.container import CerbosContainer
container = CerbosContainer()
policy_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "store")
container.with_volume_mapping(policy_dir, "/policies")
with container:
container.wait_until_ready()
host = container.http_host()
with CerbosClient(host) as c:
...See the tests available in the tests directory for more examples.
The gRPC client uses protoc generated python classes from definitions retrieved from our buf registry.
When making changes to this library, be sure to run the ./proto/generate_protos.sh to update definitions and generate python classes.
- Visit the Cerbos website
- Read the documentation
- Join the Cerbos community on Slack
- Email us at [email protected]