From a7ac597f77dff820ffc4108bd7c166594b0e438e Mon Sep 17 00:00:00 2001 From: tvdven Date: Mon, 10 Aug 2026 13:11:46 +0200 Subject: [PATCH] Add Cerbos policy store support Co-Authored-By: Claude Sonnet 5 --- .github/workflows/cerbos-app-test.yml | 42 +++ .gitignore | 1 + Makefile | 4 + app-tests/cerbos-test.py | 208 +++++++++++ app-tests/clean-cerbos-services.sh | 9 + app-tests/docker-compose-app-tests-cerbos.yml | 62 ++++ app-tests/run-cerbos-services.sh | 45 +++ app-tests/wait-for-policy-bundle.sh | 46 +++ docker/Dockerfile | 42 ++- docker/docker-compose-example-cerbos.yml | 74 ++++ packages/opal-client/opal_client/client.py | 48 ++- packages/opal-client/opal_client/config.py | 48 +++ .../opal-client/opal_client/engine/options.py | 42 +++ .../opal-client/opal_client/engine/runner.py | 82 +++++ .../opal_client/policy_store/cerbos_client.py | 325 ++++++++++++++++++ .../policy_store_client_factory.py | 10 +- .../opal_client/policy_store/schemas.py | 1 + 17 files changed, 1085 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/cerbos-app-test.yml create mode 100644 app-tests/cerbos-test.py create mode 100755 app-tests/clean-cerbos-services.sh create mode 100644 app-tests/docker-compose-app-tests-cerbos.yml create mode 100755 app-tests/run-cerbos-services.sh create mode 100755 app-tests/wait-for-policy-bundle.sh create mode 100644 docker/docker-compose-example-cerbos.yml create mode 100644 packages/opal-client/opal_client/policy_store/cerbos_client.py diff --git a/.github/workflows/cerbos-app-test.yml b/.github/workflows/cerbos-app-test.yml new file mode 100644 index 000000000..387d6560d --- /dev/null +++ b/.github/workflows/cerbos-app-test.yml @@ -0,0 +1,42 @@ +name: Cerbos Tests + +on: + push: + branches: [ master ] + pull_request: + branches: [ master ] + +jobs: + test: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v3 + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.x' + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install pytest pytest-asyncio aiohttp + + - name: Start services + run: | + ./app-tests/run-cerbos-services.sh + + - name: Run tests + run: | + pytest app-tests/cerbos-test.py -v + + - name: Show service logs on failure + if: failure() + run: | + docker compose -f app-tests/docker-compose-app-tests-cerbos.yml logs + + - name: Cleanup + if: always() # Run cleanup even if tests fail + run: | + ./app-tests/clean-cerbos-services.sh diff --git a/.gitignore b/.gitignore index c3224ec1c..f8de5844d 100644 --- a/.gitignore +++ b/.gitignore @@ -140,3 +140,4 @@ dmypy.json # Private Claude Code working artifacts (plans/specs) — never commit .claude/ +/.mcp.json diff --git a/Makefile b/Makefile index 7212170b2..f2119d953 100644 --- a/Makefile +++ b/Makefile @@ -16,6 +16,7 @@ help: @echo " docker-run-client - Run opal-client in Docker" @echo " docker-run-server - Run opal-server in Docker" @echo " docker-build-client-eopa - Build opal-client-eopa Docker image" + @echo " docker-build-client-cerbos - Build opal-client-cerbos Docker image" OPAL_SERVER_URL ?= http://host.docker.internal:7002 OPAL_AUTH_PRIVATE_KEY ?= /root/ssh/opal_rsa @@ -68,6 +69,9 @@ docker-build-client-eopa: docker-build-client-cedar: @docker build -t permitio/opal-client-cedar --target client-cedar -f docker/Dockerfile . +docker-build-client-cerbos: + @docker build -t permitio/opal-client-cerbos --target client-cerbos -f docker/Dockerfile . + docker-build-client-standalone: @docker build -t permitio/opal-client-standalone --target client-standalone -f docker/Dockerfile . diff --git a/app-tests/cerbos-test.py b/app-tests/cerbos-test.py new file mode 100644 index 000000000..400601e12 --- /dev/null +++ b/app-tests/cerbos-test.py @@ -0,0 +1,208 @@ +import json +from typing import Any, Dict, List + +import aiohttp +import pytest + +# Constants +CERBOS_URL = "http://localhost:3592" + +# Test cases, mirroring the "User Role" / "Admin Role" suites from +# cerbos/example-cerbos-policy-repository's basicResource_test.yaml. +# Format: (principal_id, roles, resource_id, attrs, action, expected_allow, description) +RESOURCES = { + "resource1": {"ownerId": "sally", "isPublished": True}, + "resource2": {"ownerId": "sally", "isPublished": True}, + "resource3": {"ownerId": "sally", "isPublished": False}, +} + +TEST_CASES = [ + # Admin can do everything + ( + "ian", + ["ADMIN"], + "resource1", + RESOURCES["resource1"], + "read", + True, + "Admin can read", + ), + ( + "ian", + ["ADMIN"], + "resource1", + RESOURCES["resource1"], + "update", + True, + "Admin can update", + ), + ( + "ian", + ["ADMIN"], + "resource3", + RESOURCES["resource3"], + "delete", + True, + "Admin can delete unpublished", + ), + # Owner (sally) can do everything to her own resources + ( + "sally", + ["USER"], + "resource1", + RESOURCES["resource1"], + "read", + True, + "Owner can read", + ), + ( + "sally", + ["USER"], + "resource1", + RESOURCES["resource1"], + "update", + True, + "Owner can update", + ), + ( + "sally", + ["USER"], + "resource3", + RESOURCES["resource3"], + "delete", + True, + "Owner can delete unpublished own resource", + ), + # Non-owner (frank) can read published resources but not modify them + ( + "frank", + ["USER"], + "resource1", + RESOURCES["resource1"], + "read", + True, + "Non-owner can read published resource", + ), + ( + "frank", + ["USER"], + "resource1", + RESOURCES["resource1"], + "update", + False, + "Non-owner cannot update", + ), + ( + "frank", + ["USER"], + "resource1", + RESOURCES["resource1"], + "delete", + False, + "Non-owner cannot delete", + ), + # Non-owner cannot even read unpublished resources + ( + "frank", + ["USER"], + "resource3", + RESOURCES["resource3"], + "read", + False, + "Non-owner cannot read unpublished resource", + ), +] + + +@pytest.fixture +async def http_client() -> aiohttp.ClientSession: + async with aiohttp.ClientSession() as client: + yield client + + +class CerbosApiClient: + """Helper class for Cerbos check API interactions.""" + + def __init__(self, client: aiohttp.ClientSession): + self.client = client + + async def check( + self, + principal_id: str, + roles: List[str], + resource_id: str, + resource_kind: str, + attrs: Dict[str, Any], + actions: List[str], + ) -> Dict[str, Any]: + url = f"{CERBOS_URL}/api/check/resources" + payload = { + "principal": {"id": principal_id, "roles": roles}, + "resources": [ + { + "resource": { + "id": resource_id, + "kind": resource_kind, + "attr": attrs, + }, + "actions": actions, + } + ], + } + async with self.client.post(url, json=payload) as response: + response.raise_for_status() + result = await response.json() + print(f"\nCheck: {principal_id} {actions} {resource_id}") + print(f"Response: {json.dumps(result, indent=2)}") + return result + + +class TestCerbosPermissions: + """Test suite for Cerbos permissions, against the real basicResource policy + from cerbos/example-cerbos-policy-repository.""" + + @pytest.fixture + async def cerbos_client( + self, http_client: aiohttp.ClientSession + ) -> CerbosApiClient: + return CerbosApiClient(http_client) + + @pytest.mark.parametrize( + "principal_id, roles, resource_id, attrs, action, expected_allow, description", + TEST_CASES, + ) + async def test_permissions( + self, + cerbos_client: CerbosApiClient, + principal_id: str, + roles: List[str], + resource_id: str, + attrs: Dict[str, Any], + action: str, + expected_allow: bool, + description: str, + ): + result = await cerbos_client.check( + principal_id, roles, resource_id, "basicResource", attrs, [action] + ) + actions = result["results"][0]["actions"] + expected_effect = "EFFECT_ALLOW" if expected_allow else "EFFECT_DENY" + assert actions.get(action) == expected_effect, ( + f"Test failed: {description}\n" + f"Principal: {principal_id}, Action: {action}, Resource: {resource_id}\n" + f"Expected: {expected_effect}, Got: {actions.get(action)}" + ) + + async def test_policy_synced(self, http_client: aiohttp.ClientSession): + """Confirms OPAL actually pushed the real policy (not just that the + Cerbos PDP is up) - fails loudly if the policy sync silently no-op'd.""" + async with http_client.get( + f"{CERBOS_URL}/admin/policies", + auth=aiohttp.BasicAuth("cerbos", "cerbosAdmin"), + ) as response: + response.raise_for_status() + result = await response.json() + print("\nSynced policies:", result) + assert result.get( + "policyIds" + ), "No policies found in Cerbos - sync did not happen" diff --git a/app-tests/clean-cerbos-services.sh b/app-tests/clean-cerbos-services.sh new file mode 100755 index 000000000..86687b2ec --- /dev/null +++ b/app-tests/clean-cerbos-services.sh @@ -0,0 +1,9 @@ +#!/bin/bash + +# Make paths below independent of the caller's cwd. +cd "$(dirname "$0")" + +echo "Cleaning up services..." +docker compose -f docker-compose-app-tests-cerbos.yml down -v + +echo "Cleanup complete" diff --git a/app-tests/docker-compose-app-tests-cerbos.yml b/app-tests/docker-compose-app-tests-cerbos.yml new file mode 100644 index 000000000..92189bb82 --- /dev/null +++ b/app-tests/docker-compose-app-tests-cerbos.yml @@ -0,0 +1,62 @@ +name: opal-cerbos-app-tests + +services: + broadcast_channel: + image: postgres:alpine + environment: + - POSTGRES_DB=postgres + - POSTGRES_USER=postgres + - POSTGRES_PASSWORD=postgres + networks: + - opal-network + + opal_server: + image: permitio/opal-server:latest + environment: + - OPAL_BROADCAST_URI=postgres://postgres:postgres@broadcast_channel:5432/postgres + - UVICORN_NUM_WORKERS=4 + - OPAL_POLICY_REPO_URL=https://github.com/cerbos/example-cerbos-policy-repository + - OPAL_POLICY_REPO_MAIN_BRANCH=main + # Two separate filters decide what lands in the bundle, and both must allow + # yaml. FILTER_FILE_EXTENSIONS picks which repo files are read at all; + # POLICY_REPO_POLICY_EXTENSIONS decides which of those count as policy + # modules. The published opal-server image defaults FILTER_FILE_EXTENSIONS + # to .rego/.json, so without this the Cerbos yaml policies are dropped + # before the second filter ever sees them. + - OPAL_FILTER_FILE_EXTENSIONS=.rego,.json,.yaml,.yml + - OPAL_POLICY_REPO_POLICY_EXTENSIONS=.yaml,.yml,.json + # Cerbos example repos ship test suites and engine config next to the real + # policies; neither is a policy document. + - OPAL_BUNDLE_IGNORE=*_test.yaml,.cerbos.yaml,.cerbos-hub.yaml + - OPAL_POLICY_REPO_POLLING_INTERVAL=30 + - OPAL_DATA_CONFIG_SOURCES={"config":{"entries":[{"url":"http://opal_server:7002/policy-data","topics":["policy_data"],"dst_path":"/static"}]}} + - OPAL_LOG_FORMAT_INCLUDE_PID=true + ports: + - "7002:7002" + depends_on: + - broadcast_channel + networks: + - opal-network + + opal_client_cerbos: + build: + context: .. + dockerfile: docker/Dockerfile + target: client-cerbos + environment: + - OPAL_SERVER_URL=http://opal_server:7002 + - OPAL_LOG_FORMAT_INCLUDE_PID=true + ports: + - "7766:7000" + - "3592:3592" + depends_on: + - opal_server + networks: + - opal-network + volumes: + - ./wait-for-policy-bundle.sh:/opal/wait-for-policy-bundle.sh:ro + command: sh -c "exec ./wait-for.sh opal_server:7002 --timeout=40 -- sh -c './wait-for-policy-bundle.sh http://opal_server:7002 && exec ./start.sh'" + +networks: + opal-network: + driver: bridge diff --git a/app-tests/run-cerbos-services.sh b/app-tests/run-cerbos-services.sh new file mode 100755 index 000000000..c7e66cc91 --- /dev/null +++ b/app-tests/run-cerbos-services.sh @@ -0,0 +1,45 @@ +#!/bin/bash +set -e + +# Make paths below independent of the caller's cwd. +cd "$(dirname "$0")" + +echo "Building client-cerbos image..." +docker compose -f docker-compose-app-tests-cerbos.yml build + +echo "Starting Cerbos and OPAL services..." +docker compose -f docker-compose-app-tests-cerbos.yml up -d + +echo "Waiting for opal-client to finish syncing policy..." +ready=false +for _ in $(seq 1 30); do + if curl -sf http://localhost:7766/ready > /dev/null 2>&1; then + ready=true + break + fi + sleep 2 +done + +if [ "${ready}" != "true" ]; then + echo "opal-client did not become ready" >&2 + docker compose -f docker-compose-app-tests-cerbos.yml logs + exit 1 +fi + +# /ready only confirms OPAL successfully PUT the policy to Cerbos's admin API - +# Cerbos's own storage layer (sqlite3 driver, file-watch reload) can briefly lag +# behind that write before the policy is actually queryable/enforceable. Poll +# Cerbos's own admin API directly so we don't start the tests during that gap. +echo "Waiting for Cerbos to index the synced policy..." +for _ in $(seq 1 30); do + if curl -sf -u cerbos:cerbosAdmin http://localhost:3592/admin/policies \ + | python3 -c "import json, sys; sys.exit(0 if json.load(sys.stdin).get('policyIds') else 1)" 2>/dev/null; then + echo "Services ready" + exit 0 + fi + sleep 1 +done + +echo "Cerbos did not index the synced policy" >&2 +docker compose -f docker-compose-app-tests-cerbos.yml logs +exit 1 diff --git a/app-tests/wait-for-policy-bundle.sh b/app-tests/wait-for-policy-bundle.sh new file mode 100755 index 000000000..b48b22186 --- /dev/null +++ b/app-tests/wait-for-policy-bundle.sh @@ -0,0 +1,46 @@ +#!/bin/sh +# Waits until opal-server's policy bundle actually contains basicResource.yaml. +# +# opal-server's git clone and initial bundle build happen asynchronously in +# the background after startup, and the client's wait-for.sh only checks that +# opal-server's TCP port is open - which is true before that clone finishes. +# Waiting for the bundle contents rather than the port keeps the client from +# syncing against a server that cannot serve policy yet. +# +# A file that never appears is a configuration problem, not a slow clone: the +# bundle is built through two independent filters, OPAL_FILTER_FILE_EXTENSIONS +# (which repo files are read at all) and OPAL_POLICY_REPO_POLICY_EXTENSIONS +# (which of those count as policy modules), and Cerbos yaml needs both to allow +# it. The bundle is printed on each attempt to make that case diagnosable. +set -e + +SERVER_URL="${1:-http://opal_server:7002}" +RESPONSE_FILE="/tmp/policy-bundle-check.json" + +echo "Waiting for opal-server to have a complete policy bundle..." +for i in $(seq 1 40); do + if ! wget -q -O "${RESPONSE_FILE}" "${SERVER_URL}/policy?path=."; then + echo " attempt ${i}: wget failed to reach ${SERVER_URL}/policy?path=." + sleep 1 + continue + fi + if python3 -c " +import json, sys +with open('${RESPONSE_FILE}') as f: + d = json.load(f) +paths = [m['path'] for m in d.get('policy_modules', [])] +sys.exit(0 if 'basicResource.yaml' in paths else 1) +"; then + echo "Policy bundle is complete" + exit 0 + fi + # Show what we actually got, so a repeat failure is diagnosable instead of silent. + echo " attempt ${i}: basicResource.yaml not yet in bundle. Response was:" + cat "${RESPONSE_FILE}" + echo + sleep 1 +done + +echo "opal-server's policy bundle never became complete" >&2 +echo "If the bundle above is stable and simply lacks the file, check the extension filters on opal_server." >&2 +exit 1 diff --git a/docker/Dockerfile b/docker/Dockerfile index c50aa1064..e69ab85f8 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -181,6 +181,47 @@ ENV OPAL_POLICY_STORE_URL=http://localhost:8180 EXPOSE 8180 USER opal +# CERBOS BINARY DOWNLOAD STAGE ----------------------- +# --------------------------------------------------- +FROM alpine:latest AS cerbos-extractor +USER root + +RUN apk add --no-cache wget + +WORKDIR /download + +# Download pre-built Cerbos binary based on architecture +ARG cerbos_tag=v0.54.0 +ARG TARGETARCH +RUN case "${TARGETARCH}" in \ + "amd64") CERBOS_ARCH="x86_64" ;; \ + "arm64") CERBOS_ARCH="arm64" ;; \ + *) echo "Unsupported architecture: ${TARGETARCH}" && exit 1 ;; \ + esac && \ + echo "Downloading Cerbos for Linux_${CERBOS_ARCH}" && \ + wget -O cerbos.tar.gz "https://github.com/cerbos/cerbos/releases/download/${cerbos_tag}/cerbos_${cerbos_tag#v}_Linux_${CERBOS_ARCH}.tar.gz" && \ + tar xzf cerbos.tar.gz && \ + chmod +x cerbos + +# CERBOS CLIENT IMAGE -------------------------------- +# Using standalone image as base -------------------- +# --------------------------------------------------- +FROM client-standalone AS client-cerbos + +# Temporarily move back to root for additional setup +USER root + +# copy cerbos from cerbos-extractor +COPY --from=cerbos-extractor /download/cerbos /bin/cerbos + +# enable inline Cerbos PDP +ENV OPAL_POLICY_STORE_TYPE=CERBOS +ENV OPAL_INLINE_CERBOS_ENABLED=true +ENV OPAL_INLINE_CERBOS_EXEC_PATH=/bin/cerbos +ENV OPAL_POLICY_STORE_URL=http://localhost:3592 +# expose cerbos HTTP port +EXPOSE 3592 +USER opal # OPENFGA CLIENT IMAGE -------------------------------- # Using standalone image as base -------------------- @@ -220,7 +261,6 @@ USER opal CMD ["./start-openfga.sh"] - # SERVER IMAGE -------------------------------------- # --------------------------------------------------- FROM common AS server diff --git a/docker/docker-compose-example-cerbos.yml b/docker/docker-compose-example-cerbos.yml new file mode 100644 index 000000000..702924b56 --- /dev/null +++ b/docker/docker-compose-example-cerbos.yml @@ -0,0 +1,74 @@ +name: opal-cerbos-example + +services: + # When scaling the opal-server to multiple nodes and/or multiple workers, we use + # a *broadcast* channel to sync between all the instances of opal-server. + # Under the hood, this channel is implemented by encode/broadcaster. + broadcast_channel: + image: postgres:alpine + environment: + - POSTGRES_DB=postgres + - POSTGRES_USER=postgres + - POSTGRES_PASSWORD=postgres + networks: + - opal-network + + # OPAL server configuration + # Handles policy updates and coordinates with the broadcast channel + opal_server: + image: permitio/opal-server:latest + environment: + - OPAL_BROADCAST_URI=postgres://postgres:postgres@broadcast_channel:5432/postgres + - UVICORN_NUM_WORKERS=4 + # cerbos/example-cerbos-policy-repository - a real, public example repo maintained by + # Cerbos, containing a resourcePolicy (basicResource.yaml) plus its own test suite + - OPAL_POLICY_REPO_URL=https://github.com/cerbos/example-cerbos-policy-repository + # this repo's default branch is "main", not OPAL's own default of "master" + - OPAL_POLICY_REPO_MAIN_BRANCH=main + # Cerbos policies are .yaml/.json, not .rego - this must be set explicitly per + # deployment rather than changed globally, since raising the shared default would + # affect every OPA/Cedar deployment too, picking up unrelated .yaml/.json files + # as if they were policy modules. + # Both filters have to allow yaml: FILTER_FILE_EXTENSIONS decides which repo + # files are read into the bundle at all, and only the survivors are classified + # by POLICY_REPO_POLICY_EXTENSIONS. The published opal-server image still + # defaults the former to .rego/.json. + - OPAL_FILTER_FILE_EXTENSIONS=.rego,.json,.yaml,.yml + - OPAL_POLICY_REPO_POLICY_EXTENSIONS=.yaml,.yml,.json + # test suites and engine config live next to the policies in this repo + - OPAL_BUNDLE_IGNORE=*_test.yaml,.cerbos.yaml,.cerbos-hub.yaml + - OPAL_POLICY_REPO_POLLING_INTERVAL=30 + - OPAL_LOG_FORMAT_INCLUDE_PID=true + # No OPAL_DATA_CONFIG_SOURCES: Cerbos has no data-document store to sync - + # see CerbosClient.set_policy_data. + ports: + - "7002:7002" + depends_on: + - broadcast_channel + networks: + - opal-network + + # OPAL client configured for Cerbos, running Cerbos inline + opal_client_cerbos: + build: + context: .. + dockerfile: docker/Dockerfile + target: client-cerbos + environment: + - OPAL_SERVER_URL=http://opal_server:7002 + - OPAL_LOG_FORMAT_INCLUDE_PID=true + ports: + # exposes opal client API at: http://localhost:7766 + - "7766:7000" + # exposes the Cerbos HTTP API (check/admin) at: http://localhost:3592 + - "3592:3592" + depends_on: + - opal_server + networks: + - opal-network + # Ensures opal-server is ready before starting the client + command: sh -c "exec ./wait-for.sh opal_server:7002 --timeout=20 -- ./start.sh" + +networks: + opal-network: + driver: bridge diff --git a/packages/opal-client/opal_client/client.py b/packages/opal-client/opal_client/client.py index fbee048c6..6a714abda 100644 --- a/packages/opal-client/opal_client/client.py +++ b/packages/opal-client/opal_client/client.py @@ -23,10 +23,16 @@ from opal_client.data.updater import DataUpdater from opal_client.engine.options import ( CedarServerOptions, + CerbosServerOptions, OpaServerOptions, OpenFGAServerOptions, ) -from opal_client.engine.runner import CedarRunner, OpaRunner, OpenFGARunner +from opal_client.engine.runner import ( + CedarRunner, + CerbosRunner, + OpaRunner, + OpenFGARunner, +) from opal_client.limiter import StartupLoadLimiter from opal_client.policy.api import init_policy_router from opal_client.policy.updater import PolicyUpdater @@ -57,6 +63,8 @@ def __init__( inline_openfga_options: OpenFGAServerOptions = None, inline_cedar_enabled: bool = None, inline_cedar_options: CedarServerOptions = None, + inline_cerbos_enabled: bool = None, + inline_cerbos_options: CerbosServerOptions = None, verifier: Optional[JWTVerifier] = None, store_backup_path: Optional[str] = None, store_backup_interval: Optional[int] = None, @@ -92,6 +100,9 @@ def __init__( inline_cedar_enabled: bool = ( inline_cedar_enabled or opal_client_config.INLINE_CEDAR_ENABLED ) + inline_cerbos_enabled: bool = ( + inline_cerbos_enabled or opal_client_config.INLINE_CERBOS_ENABLED + ) opal_client_identifier: str = ( opal_client_config.OPAL_CLIENT_STAT_ID or f"CLIENT_{uuid.uuid4().hex}" ) @@ -179,9 +190,11 @@ def __init__( self.engine_runner = self._init_engine_runner( inline_opa_enabled, inline_cedar_enabled, + inline_cerbos_enabled, inline_openfga_enabled, inline_opa_options, inline_cedar_options, + inline_cerbos_options, inline_openfga_options, ) @@ -222,11 +235,13 @@ def _init_engine_runner( self, inline_opa_enabled: bool, inline_cedar_enabled: bool, + inline_cerbos_enabled: bool, inline_openfga_enabled: bool, inline_opa_options: Optional[OpaServerOptions] = None, inline_cedar_options: Optional[CedarServerOptions] = None, + inline_cerbos_options: Optional[CerbosServerOptions] = None, inline_openfga_options: Optional[OpenFGAServerOptions] = None, - ) -> Union[OpaRunner, CedarRunner, OpenFGARunner, Literal[False]]: + ) -> Union[OpaRunner, CedarRunner, CerbosRunner, OpenFGARunner, Literal[False]]: """Initialize appropriate engine runner based on policy store type.""" # Setup rehydration callbacks for all policy store types @@ -296,6 +311,35 @@ async def _rehydrate_data(): piped_logs_format=opal_client_config.INLINE_CEDAR_LOG_FORMAT, ) + elif ( + inline_cerbos_enabled and self.policy_store_type == PolicyStoreTypes.CERBOS + ): + inline_cerbos_options = ( + inline_cerbos_options or opal_client_config.INLINE_CERBOS_CONFIG + ) + rehydration_callbacks = [] + + # Cerbos's disk store starts empty on every process restart (policies + # are pushed via the admin API, not persisted to disk), so a fresh + # policy sync is needed each time the engine comes back up - same + # reasoning as OPA's rehydration above. Data has no Cerbos equivalent + # (see CerbosClient.set_policy_data), so there's nothing to rehydrate there. + if self.policy_updater: + + async def _rehydrate_policy(): + if not self.opal_server_connectivity_disabled: + await self.policy_updater.trigger_update_policy( + force_full_update=True, + ) + + rehydration_callbacks.append(_rehydrate_policy) + + return CerbosRunner.setup_cerbos_runner( + options=inline_cerbos_options, + piped_logs_format=opal_client_config.INLINE_CERBOS_LOG_FORMAT, + rehydration_callbacks=rehydration_callbacks, + ) + # OpenFGA Runner elif ( inline_openfga_enabled diff --git a/packages/opal-client/opal_client/config.py b/packages/opal-client/opal_client/config.py index e365175c5..e25e3daf9 100644 --- a/packages/opal-client/opal_client/config.py +++ b/packages/opal-client/opal_client/config.py @@ -2,6 +2,7 @@ from opal_client.engine.options import ( CedarServerOptions, + CerbosServerOptions, OpaServerOptions, OpenFGAServerOptions, ) @@ -245,6 +246,53 @@ def load_policy_store(): description="The log format to use for inline Cedar logs", ) + # Cerbos runner configuration (Cerbos-engine can optionally be run by OPAL) -------- + + # whether or not OPAL should run the Cerbos PDP by itself in the same container + INLINE_CERBOS_ENABLED = confi.bool( + "INLINE_CERBOS_ENABLED", + True, + description="Whether or not OPAL should run the Cerbos PDP by itself in the same container", + ) + + INLINE_CERBOS_EXEC_PATH = confi.str( + "INLINE_CERBOS_EXEC_PATH", + None, + description="Path to the Cerbos executable. Defaults to searching for 'cerbos' binary in PATH if not specified.", + ) + + # if inline Cerbos is indeed enabled, user can pass cli options + # (configuration) that affects how the PDP will run + INLINE_CERBOS_CONFIG = confi.model( + "INLINE_CERBOS_CONFIG", + CerbosServerOptions, + {}, # defaults are being set according to CerbosServerOptions pydantic definitions (see class) + description="CLI options used when running the Cerbos PDP inline", + ) + + INLINE_CERBOS_LOG_FORMAT: EngineLogFormat = confi.enum( + "INLINE_CERBOS_LOG_FORMAT", + EngineLogFormat, + EngineLogFormat.NONE, + description="The log format to use for inline Cerbos logs", + ) + + # Credentials OPAL client uses to authenticate against the Cerbos admin API when + # pushing policies. Must match whatever the Cerbos PDP is actually configured with - + # for inline mode that's INLINE_CERBOS_CONFIG's admin_username/admin_password_hash + # (a bcrypt hash of this same password, not the plaintext value below). + CERBOS_ADMIN_USERNAME = confi.str( + "CERBOS_ADMIN_USERNAME", + "cerbos", + description="Username OPAL client uses to authenticate against the Cerbos admin API", + ) + CERBOS_ADMIN_PASSWORD = confi.str( + "CERBOS_ADMIN_PASSWORD", + "cerbosAdmin", + description="Password OPAL client uses to authenticate against the Cerbos admin API " + "(plaintext here - Cerbos itself is configured with a bcrypt hash of this value)", + ) + # OpenFGA runner configuration INLINE_OPENFGA_ENABLED = confi.bool( "INLINE_OPENFGA_ENABLED", diff --git a/packages/opal-client/opal_client/engine/options.py b/packages/opal-client/opal_client/engine/options.py index 40d1bd478..d5f17ee3c 100644 --- a/packages/opal-client/opal_client/engine/options.py +++ b/packages/opal-client/opal_client/engine/options.py @@ -238,3 +238,45 @@ def get_args(self) -> Iterable[str]: yield match.group("port") or "8180" # TODO: files + + +class CerbosServerOptions(BaseModel): + """Options to configure the Cerbos PDP (apply when choosing to run Cerbos + inline). + + Cerbos's admin API requires a bcrypt password hash, base64-encoded - + not a plaintext password. The default below is the hash of Cerbos's + own documented default password ("cerbosAdmin"); change it (and the + username) for anything beyond local/demo use. Generate a new one + with: echo "" | htpasswd -niBC 10 | cut -d + ':' -f 2 | base64 + """ + + addr: str = Field( + "0.0.0.0:3592", + description="listening address of the Cerbos HTTP API (e.g., [ip]: for TCP)", + ) + sqlite_dsn: str = Field( + "file::memory:?cache=shared", + description="DSN for Cerbos's sqlite3 storage driver, which OPAL uses because it's " + "the store admin API pushes actually accept - Cerbos's disk/git/blob drivers only " + "support triggering a reload of files already present, not accepting pushed content", + ) + admin_username: str = Field("cerbos", description="username for Cerbos's admin API") + admin_password_hash: str = Field( + "JDJ5JDEwJHdIc29ZSEFRNEdTVWE1YTcyQzhvWS5DcVlXOVFaNnhXWkdaNWFxSmlmRXBDckphS2tPVU9L", + description="base64-encoded bcrypt hash of the admin API password", + ) + + def get_args(self) -> Iterable[str]: + if not HOST_ADDR_PATTERN.match(self.addr): + raise ValueError( + f"Invalid addr format: {self.addr}. Expected [ip]:, e.g. '0.0.0.0:3592', ':3592'" + ) + + yield "--set=storage.driver=sqlite3" + yield f"--set=storage.sqlite3.dsn={self.sqlite_dsn}" + yield f"--set=server.httpListenAddr={self.addr}" + yield "--set=server.adminAPI.enabled=true" + yield f"--set=server.adminAPI.adminCredentials.username={self.admin_username}" + yield f"--set=server.adminAPI.adminCredentials.passwordHash={self.admin_password_hash}" diff --git a/packages/opal-client/opal_client/engine/runner.py b/packages/opal-client/opal_client/engine/runner.py index 532043678..350165e4e 100644 --- a/packages/opal-client/opal_client/engine/runner.py +++ b/packages/opal-client/opal_client/engine/runner.py @@ -10,6 +10,7 @@ from opal_client.engine.logger import log_engine_output_opa, log_engine_output_simple from opal_client.engine.options import ( CedarServerOptions, + CerbosServerOptions, OpaServerOptions, OpenFGAServerOptions, ) @@ -527,3 +528,84 @@ async def handle_log_line(self, line: bytes) -> bool: await log_engine_output_simple(line) return False + + +class CerbosRunner(PolicyEngineRunner): + def __init__( + self, + options: Optional[CerbosServerOptions] = None, + piped_logs_format: EngineLogFormat = EngineLogFormat.NONE, + ): + super().__init__(piped_logs_format) + self._options = options or CerbosServerOptions() + + def get_executable_path(self) -> str: + if opal_client_config.INLINE_CERBOS_EXEC_PATH: + return opal_client_config.INLINE_CERBOS_EXEC_PATH + else: + logger.warning( + "Cerbos executable path not set, looking for 'cerbos' binary in system PATH. " + "It is recommended to set the INLINE_CERBOS_EXEC_PATH configuration." + ) + path = shutil.which("cerbos") + if path is None: + raise FileNotFoundError("Cerbos executable not found in PATH") + return path + + def get_arguments(self) -> list[str]: + return ["server"] + list(self._options.get_args()) + + async def health_check(self) -> bool: + """Performs a health check on the Cerbos PDP by calling its health + endpoint.""" + try: + health_url = f"{opal_client_config.POLICY_STORE_URL}/_cerbos/health" + timeout_seconds = opal_client_config.POLICY_STORE_CONN_RETRY.wait_time + timeout = aiohttp.ClientTimeout(total=timeout_seconds) + async with aiohttp.ClientSession( + trust_env=True, timeout=timeout + ) as session: + response = await session.get(health_url) + return response.status == 200 + except Exception as e: + logger.debug(f"Cerbos health check failed: {e}") + return False + + @staticmethod + def setup_cerbos_runner( + options: Optional[CerbosServerOptions] = None, + piped_logs_format: EngineLogFormat = EngineLogFormat.NONE, + initial_start_callbacks: Optional[List[AsyncCallback]] = None, + rehydration_callbacks: Optional[List[AsyncCallback]] = None, + ): + """Factory for CerbosRunner, accept optional callbacks to run in + certain lifecycle events. + + Initial Start Callbacks: + The first time we start the engine, we might want to do certain actions (like launch tasks) + that are dependent on the policy store being up (such as PolicyUpdater, DataUpdater). + + Rehydration Callbacks: + when the engine restarts, its policies are gone (disk storage starts empty) + and it does not have the state necessary to handle authorization queries. + therefore it is necessary that we rehydrate the store with fresh policies + fetched from the server. + """ + cerbos_runner = CerbosRunner( + options=options, piped_logs_format=piped_logs_format + ) + + if initial_start_callbacks: + cerbos_runner.register_process_initial_start_callbacks( + initial_start_callbacks + ) + + if rehydration_callbacks: + cerbos_runner.register_process_restart_callbacks(rehydration_callbacks) + + return cerbos_runner + + async def handle_log_line(self, line: bytes) -> bool: + await log_engine_output_simple(line) + + return False diff --git a/packages/opal-client/opal_client/policy_store/cerbos_client.py b/packages/opal-client/opal_client/policy_store/cerbos_client.py new file mode 100644 index 000000000..1852ba872 --- /dev/null +++ b/packages/opal-client/opal_client/policy_store/cerbos_client.py @@ -0,0 +1,325 @@ +import asyncio +import json +from typing import Dict, List, Optional + +import aiohttp +import yaml +from aiofiles.threadpool.text import AsyncTextIOWrapper +from opal_client.config import opal_client_config +from opal_client.logger import logger +from opal_client.policy_store.base_policy_store_client import ( + BasePolicyStoreClient, + JsonableValue, +) +from opal_client.policy_store.liveness_probe import LivenessProbeMixin +from opal_client.policy_store.opa_client import ( + RETRY_CONFIG, + affects_transaction, + fail_silently, +) +from opal_client.policy_store.schemas import PolicyStoreAuth +from opal_common.schemas.policy import PolicyBundle +from opal_common.schemas.store import StoreTransaction, TransactionType +from tenacity import retry + +# Top-level keys that identify a parsed file as an actual Cerbos policy +# (as opposed to some other .json/.yaml file that happens to share the +# extension bundle-makers use to select "policy" files - see _parse_policy_file). +_CERBOS_POLICY_KIND_KEYS = ( + "resourcePolicy", + "principalPolicy", + "derivedRoles", + "exportVariables", + "exportConstants", +) + +# Cerbos's AddOrUpdatePolicy admin API request caps at 100 policies per call. +_MAX_POLICIES_PER_REQUEST = 100 + + +class CerbosClient(LivenessProbeMixin, BasePolicyStoreClient): + """Policy store client for Cerbos. + + Cerbos decisions are computed from principal/resource attributes passed + in each request rather than from server-stored data documents (unlike + OPA), so this client has no real equivalent of set_policy_data - see the + no-op implementations below. + """ + + def __init__( + self, + cerbos_server_url=None, + cerbos_auth_token: Optional[str] = None, + auth_type: PolicyStoreAuth = PolicyStoreAuth.NONE, + admin_username: Optional[str] = None, + admin_password: Optional[str] = None, + ): + base_url = cerbos_server_url or opal_client_config.POLICY_STORE_URL + self._cerbos_url = base_url.rstrip("/") + self._policy_version: Optional[str] = None + self._lock = asyncio.Lock() + self._token = cerbos_auth_token + self._auth_type: PolicyStoreAuth = auth_type + self._admin_auth = aiohttp.BasicAuth( + admin_username or opal_client_config.CERBOS_ADMIN_USERNAME, + admin_password or opal_client_config.CERBOS_ADMIN_PASSWORD, + ) + + self._had_successful_data_transaction = False + self._had_successful_policy_transaction = False + self._most_recent_data_transaction: Optional[StoreTransaction] = None + self._most_recent_policy_transaction: Optional[StoreTransaction] = None + + # `_engine_reachable` defaults to True so /healthy preserves historical + # behavior; `start_liveness_probe()` runs an initial sample synchronously + # and overwrites this before the probe loop begins. + self._engine_reachable: bool = True + self._init_liveness_probe() + + if auth_type == PolicyStoreAuth.OAUTH: + raise ValueError("Cerbos doesn't support OAuth.") + if auth_type == PolicyStoreAuth.TOKEN and self._token is None: + logger.error("POLICY_STORE_AUTH_TOKEN can not be empty") + raise TypeError("required variables for token auth are not set") + + logger.info(f"Authentication mode for policy store: {auth_type}") + + async def _get_auth_headers(self) -> Dict[str, str]: + headers: Dict[str, str] = {} + if self._auth_type == PolicyStoreAuth.TOKEN and self._token is not None: + headers["Authorization"] = f"Bearer {self._token}" + return headers + + @staticmethod + def _parse_policy_file(path: str, content: str) -> Optional[dict]: + """Parse a bundle file's content as a Cerbos policy. + + Returns None (and logs) if the file doesn't actually look like a + Cerbos policy - e.g. a data file that happens to share the .json/ + .yaml extension bundle-makers use to select "policy" files, or one + of the *_test.yaml test-suite files Cerbos example repos ship + alongside real policies. + """ + try: + if path.endswith((".yaml", ".yml")): + parsed = yaml.safe_load(content) + elif path.endswith(".json"): + parsed = json.loads(content) + else: + return None + except (yaml.YAMLError, json.JSONDecodeError) as e: + logger.warning(f"Skipping unparsable Cerbos policy file {path}: {e}") + return None + + if not isinstance(parsed, dict) or "apiVersion" not in parsed: + logger.debug(f"Skipping {path}: does not look like a Cerbos policy") + return None + if not any(key in parsed for key in _CERBOS_POLICY_KIND_KEYS): + logger.debug(f"Skipping {path}: no recognized Cerbos policy kind") + return None + return parsed + + async def _push_policies(self, policies: List[dict]) -> None: + # Cerbos's AddOrUpdatePolicy request caps at 100 policies per call. + for i in range(0, len(policies), _MAX_POLICIES_PER_REQUEST): + await self._push_policies_batch(policies[i : i + _MAX_POLICIES_PER_REQUEST]) + + async def _push_policies_batch(self, policies: List[dict]) -> None: + async with aiohttp.ClientSession(trust_env=True) as session: + try: + async with session.put( + f"{self._cerbos_url}/admin/policy", + json={"policies": policies}, + auth=self._admin_auth, + ) as response: + if response.status >= 400: + body = await response.text() + raise Exception( + f"Failed to push policies to Cerbos: HTTP {response.status} - {body}" + ) + except aiohttp.ClientError as e: + logger.warning("Cerbos connection error: {err}", err=repr(e)) + raise + + @affects_transaction + @retry(**RETRY_CONFIG) + async def set_policy( + self, + policy_id: str, + policy_code: str, + transaction_id: Optional[str] = None, + ): + """Push a single policy (JSON-encoded Cerbos policy document) to + Cerbos's admin API.""" + await self._push_policies([json.loads(policy_code)]) + + @fail_silently() + @retry(**RETRY_CONFIG) + async def get_policy(self, policy_id: str) -> Optional[str]: + async with aiohttp.ClientSession(trust_env=True) as session: + try: + async with session.get( + f"{self._cerbos_url}/admin/policy", + params={"id": policy_id}, + auth=self._admin_auth, + ) as response: + result = await response.json() + policies = result.get("policies", []) + return json.dumps(policies[0]) if policies else None + except aiohttp.ClientError as e: + logger.warning("Cerbos connection error: {err}", err=repr(e)) + raise + + @fail_silently() + @retry(**RETRY_CONFIG) + async def get_policy_module_ids(self) -> List[str]: + async with aiohttp.ClientSession(trust_env=True) as session: + try: + async with session.get( + f"{self._cerbos_url}/admin/policies", auth=self._admin_auth + ) as response: + result = await response.json() + return result.get("policyIds", []) + except aiohttp.ClientError as e: + logger.warning("Cerbos connection error: {err}", err=repr(e)) + raise + + async def get_policies(self) -> Optional[Dict[str, str]]: + ids = await self.get_policy_module_ids() or [] + policies: Dict[str, str] = {} + for policy_id in ids: + content = await self.get_policy(policy_id) + if content is not None: + policies[policy_id] = content + return policies + + @affects_transaction + @retry(**RETRY_CONFIG) + async def delete_policy(self, policy_id: str, transaction_id: Optional[str] = None): + async with aiohttp.ClientSession(trust_env=True) as session: + try: + async with session.post( + f"{self._cerbos_url}/admin/policy/delete", + params={"id": policy_id}, + auth=self._admin_auth, + ) as response: + if response.status >= 400: + body = await response.text() + raise Exception( + f"Failed to delete Cerbos policy {policy_id}: HTTP {response.status} - {body}" + ) + except aiohttp.ClientError as e: + logger.warning("Cerbos connection error: {err}", err=repr(e)) + raise + + @affects_transaction + async def set_policies( + self, bundle: PolicyBundle, transaction_id: Optional[str] = None + ): + policies = [] + for module in bundle.policy_modules: + parsed = self._parse_policy_file(module.path, module.rego) + if parsed is not None: + policies.append(parsed) + + if policies: + await self._push_policies(policies) + + # Deliberately no diff-and-delete step here (unlike the OPA/Cedar + # clients): Cerbos's policy id format depends on the backing store + # (a filename for disk/git/blob stores, a kind.name.version triple + # for SQL stores) and isn't reliably derivable here, so a wrong + # guess could delete the wrong policy. Removing a file from the + # repo currently leaves the old policy in Cerbos until deleted + # directly (e.g. via cerbosctl). + self._policy_version = bundle.hash + + # Cerbos has no server-stored data-document concept: decisions are + # computed from attributes passed in each check request, not from data + # OPAL pushes ahead of time. These are no-ops so /ready's data component + # (which requires at least one successful data transaction, same as + # every other backend) is satisfied without pretending to sync anything. + @affects_transaction + async def set_policy_data( + self, + policy_data: JsonableValue, + path: str = "", + transaction_id: Optional[str] = None, + ): + logger.debug( + "Ignoring data update for Cerbos - Cerbos has no data-document store, " + "decisions use attributes passed in each check request" + ) + + @affects_transaction + async def patch_policy_data( + self, + policy_data: JsonableValue, + path: str = "", + transaction_id: Optional[str] = None, + ): + await self.set_policy_data( + policy_data, path=path, transaction_id=transaction_id + ) + + @affects_transaction + async def delete_policy_data( + self, path: str = "", transaction_id: Optional[str] = None + ): + pass + + async def get_data(self, path: str) -> Dict: + return {} + + async def log_transaction(self, transaction: StoreTransaction): + if transaction.transaction_type == TransactionType.policy: + self._most_recent_policy_transaction = transaction + if transaction.success: + self._had_successful_policy_transaction = True + elif transaction.transaction_type == TransactionType.data: + self._most_recent_data_transaction = transaction + if transaction.success: + self._had_successful_data_transaction = True + + async def is_ready(self) -> bool: + return ( + self._had_successful_policy_transaction + and self._had_successful_data_transaction + ) + + async def is_healthy(self) -> bool: + transactions_healthy: bool = ( + self._most_recent_policy_transaction is not None + and self._most_recent_policy_transaction.success + ) and ( + self._most_recent_data_transaction is not None + and self._most_recent_data_transaction.success + ) + return transactions_healthy and self._engine_reachable + + @property + def _probe_log_label(self) -> str: + return "Cerbos" + + async def _probe_engine_reachable(self, session: aiohttp.ClientSession) -> bool: + health_url = f"{self._cerbos_url}/_cerbos/health" + async with session.get(health_url) as response: + return response.status == 200 + + def _set_engine_reachable(self, value: bool) -> None: + self._engine_reachable = value + + def _get_engine_reachable(self) -> bool: + return self._engine_reachable + + async def get_policy_version(self) -> Optional[str]: + return self._policy_version + + async def full_export(self, writer: AsyncTextIOWrapper) -> None: + policies = await self.get_policies() + await writer.write(json.dumps({"policies": policies, "data": {}}, default=str)) + + async def full_import(self, reader: AsyncTextIOWrapper) -> None: + import_data = json.loads(await reader.read()) + for policy_id, raw in import_data["policies"].items(): + await self.set_policy(policy_id=policy_id, policy_code=raw) diff --git a/packages/opal-client/opal_client/policy_store/policy_store_client_factory.py b/packages/opal-client/opal_client/policy_store/policy_store_client_factory.py index 2186aaaf0..303461987 100644 --- a/packages/opal-client/opal_client/policy_store/policy_store_client_factory.py +++ b/packages/opal-client/opal_client/policy_store/policy_store_client_factory.py @@ -152,7 +152,16 @@ def create( cedar_auth_token=store_token, auth_type=auth_type, ) + elif PolicyStoreTypes.CERBOS == store_type: + from opal_client.policy_store.cerbos_client import CerbosClient + res = CerbosClient( + url, + cerbos_auth_token=store_token, + auth_type=auth_type, + admin_username=opal_client_config.CERBOS_ADMIN_USERNAME, + admin_password=opal_client_config.CERBOS_ADMIN_PASSWORD, + ) # Openfga elif PolicyStoreTypes.OPENFGA == store_type: from opal_client.policy_store.openfga_client import OpenFGAClient @@ -165,7 +174,6 @@ def create( data_updater_enabled=data_updater_enabled, policy_updater_enabled=policy_updater_enabled, ) - # MOCK elif PolicyStoreTypes.MOCK == store_type: from opal_client.policy_store.mock_policy_store_client import ( diff --git a/packages/opal-client/opal_client/policy_store/schemas.py b/packages/opal-client/opal_client/policy_store/schemas.py index c88cd8010..370fccf50 100644 --- a/packages/opal-client/opal_client/policy_store/schemas.py +++ b/packages/opal-client/opal_client/policy_store/schemas.py @@ -7,6 +7,7 @@ class PolicyStoreTypes(Enum): OPA = "OPA" CEDAR = "CEDAR" + CERBOS = "CERBOS" OPENFGA = "OPENFGA" MOCK = "MOCK"