Add missing wget binary

.github/workflows/tests.yml

@@ -25,6 +25,7 @@ jobs:
           --health-timeout 5s
           --health-retries 5
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     strategy:
       matrix:
         python-version: ["3.9", "3.10", "3.11", "3.12"]
@@ -53,6 +54,7 @@ jobs:
 
   test-docker:
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     steps:
       # BUILD PHASE
       - name: Checkout

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v5.0.0
     hooks:
       - id: check-yaml
       - id: end-of-file-fixer

docker/Dockerfile

@@ -38,8 +38,8 @@ COPY scripts/wait-for.sh .
 RUN chmod +x ./wait-for.sh
 RUN ln -s /opal/wait-for.sh /usr/wait-for.sh
 
-# netcat (nc) is used by the wait-for.sh script
-RUN apt-get update && apt-get install -y netcat-traditional jq && apt-get clean
+# netcat (nc) and wget are used by the wait-for.sh script
+RUN apt-get update && apt-get install -y netcat-traditional jq wget && apt-get clean
 
 # copy startup script (create link at old path to maintain backward compatibility)
 COPY ./scripts/start.sh .

documentation/docs/getting-started/configuration.mdx

@@ -24,7 +24,8 @@ Please use this table as a reference.
 | LOG_FILE_RETENTION                      |                                                                                                                                                                                 |         |
 | LOG_FILE_COMPRESSION                    |                                                                                                                                                                                 |         |
 | LOG_FILE_SERIALIZE                      | Serialize log messages in file into json format (useful for log aggregation platforms)                                                                                          |         |
-| LOG_FILE_LEVEL                          |                                                                                                                                                                                 |         |
+| LOG_FILE_LEVEL                          |
+| LOG_DIAGNOSE                            | Include diagnosis in log messages                                                                                                                                                                               |         |
 | STATISTICS_ENABLED                      | Collect statistics about OPAL clients.                                                                                                                                          |         |
 | STATISTICS_ADD_CLIENT_CHANNEL           | The topic to update about the new OPAL clients connection.                                                                                                                      |         |
 | STATISTICS_REMOVE_CLIENT_CHANNEL        | The topic to update about the OPAL clients disconnection.                                                                                                                       |         |

documentation/docs/getting-started/running-opal/run-opal-server/security-parameters.mdx

@@ -105,3 +105,5 @@ You must then configure the master token like so
 | Env Var Name           | Function                                                             |
 | :--------------------- | :------------------------------------------------------------------- |
 | OPAL_AUTH_MASTER_TOKEN | the master token generated by the cli (or any other secret you pick) |
+
+Ensure LOG_DIAGNOSE is set to False to disable diagnostic logging that may expose sensitive information.

documentation/docs/tutorials/run_opal_with_pulsar.mdx

@@ -0,0 +1,115 @@
+---
+sidebar_position: 12
+title: Run OPAL with Apache Pulsar
+---
+
+# Running OPAL-server with Apache Pulsar
+
+## Introduction
+
+OPAL-server supports multiple backbone pub/sub solutions for connecting distributed server instances. This guide explains how to set up and use Apache Pulsar as the backbone pub/sub (broadcast channel) for OPAL-server.
+
+## Apache Pulsar as the Backbone Pub/Sub
+
+### What is a backbone pub/sub?
+
+OPAL-server can scale out both in number of worker processes per server and across multiple servers. While OPAL provides a lightweight websocket pub/sub for OPAL-clients, multiple servers are linked together by a more robust messaging solution like Apache Pulsar, Kafka, Redis, or Postgres Listen/Notify.
+
+### Broadcaster Module
+
+Support for multiple backbone solutions is provided by the Permit's port of the [Python Broadcaster package](https://pypi.org/project/permit-broadcaster/). To use it with Apache Pulsar, install the `permit-broadcaster[pulsar]` module:
+
+```bash
+pip install permit-broadcaster[pulsar]
+```
+
+## Setting Up OPAL-server with Apache Pulsar
+
+### Configuration
+
+To use Apache Pulsar as the backbone, set the `OPAL_BROADCAST_URI` environment variable:
+
+```bash
+OPAL_BROADCAST_URI=pulsar://pulsar-host-name:6650
+```
+
+The "pulsar://" prefix tells OPAL-server to use Apache Pulsar.
+
+### Pulsar Topic
+
+OPAL-server uses a single Pulsar topic named 'broadcast' for all communication. This topic is automatically created when the producer and consumer are initialized.
+
+## Docker Compose Example
+
+Here's an example `docker-compose.yml` configuration that includes Apache Pulsar:
+
+```yaml
+version: '3'
+services:
+  pulsar:
+    image: apachepulsar/pulsar:3.3.1
+    command: bin/pulsar standalone
+    ports:
+      - 6650:6650
+      - 8080:8080
+    volumes:
+      - pulsardata:/pulsar/data
+      - pulsarconf:/pulsar/conf
+
+  opal-server:
+    image: permitio/opal-server:latest
+    environment:
+      - OPAL_BROADCAST_URI=pulsar://pulsar:6650
+    depends_on:
+      - pulsar
+
+volumes:
+  pulsardata:
+  pulsarconf:
+```
+
+Run this configuration with:
+
+```bash
+docker-compose up --force-recreate
+```
+
+Allow a few seconds for Apache Pulsar and OPAL to start up before testing connectivity.
+
+## Triggering Events
+
+You can trigger events using the OPAL CLI:
+
+```bash
+opal-client publish-data-update --src-url https://api.country.is/23.54.6.78 -t policy_data --dst-path /users/bob/location
+```
+
+You should see the effect in:
+- OPAL-server logs: "Broadcasting incoming event"
+- OPAL-client: Receiving and acting on the event
+- Pulsar: Event data in the 'broadcast' topic
+
+## Supported Backends
+
+| Backend  | Environment Variable                                   | Docker Compose Service |
+|----------|---------------------------------------------------------|------------------------|
+| Kafka    | `BROADCAST_URL=kafka://localhost:9092`                  | `docker-compose up kafka` |
+| Redis    | `BROADCAST_URL=redis://localhost:6379`                  | `docker-compose up redis` |
+| Postgres | `BROADCAST_URL=postgres://localhost:5432/broadcaster`   | `docker-compose up postgres` |
+| Pulsar   | `BROADCAST_URL=pulsar://localhost:6650`                 | `docker-compose up pulsar` |
+
+## Advanced: Publishing Events Directly to Pulsar
+
+You can trigger events by publishing messages directly to the 'broadcast' topic in Pulsar. Ensure the message format follows the OPAL-server schema for backbone events.
+
+## Conclusion
+
+This guide covered setting up and using Apache Pulsar as the backbone pub/sub for OPAL-server. By following these instructions, you can effectively scale your OPAL deployment across multiple servers.
+
+## Further Resources
+
+- [OPAL Documentation](https://www.opal.ac/docs/)
+- [Apache Pulsar Documentation](https://pulsar.apache.org/docs/en/standalone/)
+- [Python Broadcaster Package](https://pypi.org/project/broadcaster/)
+
+For more information or support, please refer to the OPAL community forums or contact the maintainers.

packages/opal-client/opal_client/client.py

@@ -13,6 +13,8 @@
 import websockets
 from fastapi import FastAPI, status
 from fastapi.responses import JSONResponse
+from fastapi_websocket_pubsub.pub_sub_client import PubSubOnConnectCallback
+from fastapi_websocket_rpc.rpc_channel import OnDisconnectCallback
 from opal_client.callbacks.api import init_callbacks_api
 from opal_client.callbacks.register import CallbacksRegister
 from opal_client.config import PolicyStoreTypes, opal_client_config
@@ -54,6 +56,10 @@ def __init__(
         store_backup_interval: Optional[int] = None,
         offline_mode_enabled: bool = False,
         shard_id: Optional[str] = None,
+        on_data_updater_connect: List[PubSubOnConnectCallback] = None,
+        on_data_updater_disconnect: List[OnDisconnectCallback] = None,
+        on_policy_updater_connect: List[PubSubOnConnectCallback] = None,
+        on_policy_updater_disconnect: List[OnDisconnectCallback] = None,
     ) -> None:
         """
         Args:
@@ -119,6 +125,8 @@ def __init__(
                     policy_store=self.policy_store,
                     callbacks_register=self._callbacks_register,
                     opal_client_id=opal_client_identifier,
+                    on_connect=on_policy_updater_connect,
+                    on_disconnect=on_policy_updater_disconnect,
                 )
         else:
             self.policy_updater = None
@@ -140,6 +148,8 @@ def __init__(
                     callbacks_register=self._callbacks_register,
                     opal_client_id=opal_client_identifier,
                     shard_id=self._shard_id,
+                    on_connect=on_data_updater_connect,
+                    on_disconnect=on_data_updater_disconnect,
                 )
         else:
             self.data_updater = None

packages/opal-client/opal_client/data/updater.py

@@ -9,7 +9,8 @@
 import aiohttp
 from aiohttp.client import ClientError, ClientSession
 from fastapi_websocket_pubsub import PubSubClient
-from fastapi_websocket_rpc.rpc_channel import RpcChannel
+from fastapi_websocket_pubsub.pub_sub_client import PubSubOnConnectCallback
+from fastapi_websocket_rpc.rpc_channel import OnDisconnectCallback, RpcChannel
 from opal_client.callbacks.register import CallbacksRegister
 from opal_client.callbacks.reporter import CallbacksReporter
 from opal_client.config import opal_client_config
@@ -54,6 +55,8 @@ def __init__(
         callbacks_register: Optional[CallbacksRegister] = None,
         opal_client_id: str = None,
         shard_id: Optional[str] = None,
+        on_connect: List[PubSubOnConnectCallback] = None,
+        on_disconnect: List[OnDisconnectCallback] = None,
     ):
         """Keeps policy-stores (e.g. OPA) up to date with relevant data Obtains
         data configuration on startup from OPAL-server Uses Pub/Sub to
@@ -132,6 +135,8 @@ def __init__(
         self._updates_storing_queue = TakeANumberQueue(logger)
         self._tasks = TasksPool()
         self._polling_update_tasks = []
+        self._on_connect_callbacks = on_connect or []
+        self._on_disconnect_callbacks = on_disconnect or []
 
     async def __aenter__(self):
         await self.start()
@@ -278,7 +283,8 @@ async def _subscriber(self):
             self._data_topics,
             self._update_policy_data_callback,
             methods_class=TenantAwareRpcEventClientMethods,
-            on_connect=[self.on_connect],
+            on_connect=[self.on_connect, *self._on_connect_callbacks],
+            on_disconnect=[self.on_disconnect, *self._on_disconnect_callbacks],
             extra_headers=self._extra_headers,
             keep_alive=opal_client_config.KEEP_ALIVE_INTERVAL,
             server_uri=self._server_url,

packages/opal-client/opal_client/engine/runner.py

@@ -136,8 +136,8 @@ async def _pipe_logs_stream(stream: asyncio.StreamReader):
 
                 line = b""
 
-        await asyncio.wait(
-            [
+        await asyncio.gather(
+            *[
                 _pipe_logs_stream(self._process.stdout),
                 _pipe_logs_stream(self._process.stderr),
             ]

packages/opal-client/opal_client/policy/updater.py

@@ -3,7 +3,8 @@
 
 import pydantic
 from fastapi_websocket_pubsub import PubSubClient
-from fastapi_websocket_rpc.rpc_channel import RpcChannel
+from fastapi_websocket_pubsub.pub_sub_client import PubSubOnConnectCallback
+from fastapi_websocket_rpc.rpc_channel import OnDisconnectCallback, RpcChannel
 from opal_client.callbacks.register import CallbacksRegister
 from opal_client.callbacks.reporter import CallbacksReporter
 from opal_client.config import opal_client_config
@@ -43,6 +44,8 @@ def __init__(
         data_fetcher: Optional[DataFetcher] = None,
         callbacks_register: Optional[CallbacksRegister] = None,
         opal_client_id: str = None,
+        on_connect: List[PubSubOnConnectCallback] = None,
+        on_disconnect: List[OnDisconnectCallback] = None,
     ):
         """inits the policy updater.
 
@@ -104,6 +107,8 @@ def __init__(
         )
         self._policy_update_queue = asyncio.Queue()
         self._tasks = TasksPool()
+        self._on_connect_callbacks = on_connect or []
+        self._on_disconnect_callbacks = on_disconnect or []
 
     async def __aenter__(self):
         await self.start()
@@ -243,8 +248,8 @@ async def _subscriber(self):
         self._client = PubSubClient(
             topics=self._topics,
             callback=self._update_policy_callback,
-            on_connect=[self._on_connect],
-            on_disconnect=[self._on_disconnect],
+            on_connect=[self._on_connect, *self._on_connect_callbacks],
+            on_disconnect=[self._on_disconnect, *self._on_disconnect_callbacks],
             extra_headers=self._extra_headers,
             keep_alive=opal_client_config.KEEP_ALIVE_INTERVAL,
             server_uri=self._server_url,

packages/opal-common/requires.txt

@@ -10,5 +10,5 @@ datadog>=0.44.0, <1
 ddtrace>=2.8.1,<3
 certifi>=2023.7.22 # not directly required, pinned by Snyk to avoid a vulnerability
 requests>=2.32.0 # not directly required, pinned by Snyk to avoid a vulnerability
-httpx==0.27.0
+httpx>=0.27.0
 urllib3>=2.2.2 # not directly required, pinned by Snyk to avoid a vulnerability