Implemented inline Cedar support.

Author: Shaul Kremer

packages/opal-client/opal_client/client.py

@@ -4,7 +4,7 @@
 import signal
 import uuid
 from logging import disable
-from typing import List, Optional
+from typing import List, Optional, Literal, Union
 
 import aiofiles
 import aiofiles.os
@@ -19,8 +19,8 @@
 from opal_client.data.fetcher import DataFetcher
 from opal_client.data.updater import DataUpdater
 from opal_client.limiter import StartupLoadLimiter
-from opal_client.opa.options import OpaServerOptions
-from opal_client.opa.runner import OpaRunner
+from opal_client.opa.options import OpaServerOptions, CedarServerOptions
+from opal_client.opa.runner import OpaRunner, CedarRunner
 from opal_client.policy.api import init_policy_router
 from opal_client.policy.updater import PolicyUpdater
 from opal_client.policy_store.api import init_policy_store_router
@@ -46,6 +46,8 @@ def __init__(
         policy_updater: PolicyUpdater = None,
         inline_opa_enabled: bool = None,
         inline_opa_options: OpaServerOptions = None,
+        inline_cedar_enabled: bool = None,
+        inline_cedar_options: CedarServerOptions = None,
         verifier: Optional[JWTVerifier] = None,
         store_backup_path: Optional[str] = None,
         store_backup_interval: Optional[int] = None,
@@ -67,8 +69,8 @@ def __init__(
         inline_opa_enabled: bool = (
             inline_opa_enabled or opal_client_config.INLINE_OPA_ENABLED
         )
-        inline_opa_options: OpaServerOptions = (
-            inline_opa_options or opal_client_config.INLINE_OPA_CONFIG
+        inline_cedar_enabled: bool = (
+            inline_cedar_enabled or opal_client_config.INLINE_CEDAR_ENABLED
         )
         opal_client_identifier: str = (
             opal_client_config.OPAL_CLIENT_STAT_ID or f"CLIENT_{uuid.uuid4().hex}"
@@ -140,29 +142,7 @@ def __init__(
 
         # Internal services
         # Policy store
-        if self.policy_store_type == PolicyStoreTypes.OPA and inline_opa_enabled:
-            rehydration_callbacks = [
-                # refetches policy code (e.g: rego) and static data from server
-                functools.partial(
-                    self.policy_updater.update_policy, force_full_update=True
-                ),
-            ]
-
-            if self.data_updater:
-                rehydration_callbacks.append(
-                    functools.partial(
-                        self.data_updater.get_base_policy_data,
-                        data_fetch_reason="policy store rehydration",
-                    )
-                )
-
-            self.opa_runner = OpaRunner.setup_opa_runner(
-                options=inline_opa_options,
-                piped_logs_format=opal_client_config.INLINE_OPA_LOG_FORMAT,
-                rehydration_callbacks=rehydration_callbacks,
-            )
-        else:
-            self.opa_runner = False
+        self.opa_runner = self._init_engine_runner(inline_opa_enabled, inline_cedar_enabled, inline_opa_options, inline_cedar_options)
 
         custom_ssl_context = get_custom_ssl_context()
         if (
@@ -197,6 +177,47 @@ def __init__(
         # init fastapi app
         self.app: FastAPI = self._init_fast_api_app()
 
+    def _init_engine_runner(self,
+                            inline_opa_enabled: bool, inline_cedar_enabled: bool,
+        inline_opa_options: Optional[OpaServerOptions] = None,
+        inline_cedar_options: Optional[CedarServerOptions] = None,
+                            ) -> Union[OpaRunner, CedarRunner, Literal[False]]:
+        if inline_opa_enabled and self.policy_store_type == PolicyStoreTypes.OPA:
+            inline_opa_options = (
+                inline_opa_options or opal_client_config.INLINE_OPA_CONFIG
+            )
+            rehydration_callbacks = [
+                # refetches policy code (e.g: rego) and static data from server
+                functools.partial(
+                    self.policy_updater.update_policy, force_full_update=True
+                ),
+            ]
+
+            if self.data_updater:
+                rehydration_callbacks.append(
+                    functools.partial(
+                        self.data_updater.get_base_policy_data,
+                        data_fetch_reason="policy store rehydration",
+                    )
+                )
+
+            return OpaRunner.setup_opa_runner(
+                    options=inline_opa_options,
+                    piped_logs_format=opal_client_config.INLINE_OPA_LOG_FORMAT,
+                    rehydration_callbacks=rehydration_callbacks,
+                )
+
+        elif inline_cedar_enabled and self.policy_store_type == PolicyStoreTypes.CEDAR:
+            inline_cedar_options = (
+                inline_cedar_options or opal_client_config.INLINE_CEDAR_CONFIG
+            )
+            return CedarRunner.setup_cedar_runner(
+                    options=inline_cedar_options,
+                    piped_logs_format=opal_client_config.INLINE_OPA_LOG_FORMAT,
+                )
+
+        return False
+
     def _init_fast_api_app(self):
         """inits the fastapi app object."""
         app = FastAPI(

packages/opal-client/opal_client/config.py

@@ -1,6 +1,6 @@
 from enum import Enum
 
-from opal_client.opa.options import OpaServerOptions
+from opal_client.opa.options import OpaServerOptions, CedarServerOptions
 from opal_client.policy.options import PolicyConnRetryOptions
 from opal_client.policy_store.schemas import PolicyStoreAuth, PolicyStoreTypes
 from opal_common.confi import Confi, confi
@@ -96,6 +96,18 @@ def load_policy_store():
         description="cli options used when running `opa run --server` inline",
     )
 
+    # whether or not OPAL should run the Cedar agent by itself in the same container
+    INLINE_CEDAR_ENABLED = confi.bool("INLINE_CEDAR_ENABLED", True)
+
+    # if inline Cedar is indeed enabled, user can pass cli options
+    # (configuration) that affects how the agent will run
+    INLINE_CEDAR_CONFIG = confi.model(
+        "INLINE_CEDAR_CONFIG",
+        CedarServerOptions,
+        {},  # defaults are being set according to CedarServerOptions pydantic definitions (see class)
+        description="cli options used when running the Cedar agent inline",
+    )
+
     INLINE_OPA_LOG_FORMAT: OpaLogFormat = confi.enum(
         "INLINE_OPA_LOG_FORMAT", OpaLogFormat, OpaLogFormat.NONE
     )

packages/opal-client/opal_client/opa/options.py

@@ -1,7 +1,7 @@
 from enum import Enum
-from typing import List, Optional
+from typing import List, Optional, Any
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, validator
 
 
 class LogLevel(str, Enum):
@@ -81,3 +81,55 @@ def get_opa_startup_files(self) -> str:
         """returns a list of startup policies and data."""
         files = self.files if self.files is not None else []
         return " ".join(files)
+
+class CedarServerOptions(BaseModel):
+    """Options to configure the Cedar agent (apply when choosing to run Cedar inline).
+    """
+
+    addr: str = Field(
+        ":8181",
+        description="listening address of the Cedar agent (e.g., [ip]:<port> for TCP)",
+    )
+    authentication: AuthenticationScheme = Field(
+        AuthenticationScheme.off, description="Cedar agent authentication scheme (default off)"
+    )
+    authentication_token: Optional[str] = Field(
+        None, description="If authentication is 'token', this specifies the token to use."
+    )
+    files: Optional[List[str]] = Field(
+        None,
+        description="list of built-in policies files that must be loaded on startup.",
+    )
+
+    class Config:
+        use_enum_values = True
+        allow_population_by_field_name = True
+
+        @classmethod
+        def alias_generator(cls, string: str) -> str:
+            """converts field named tls_private_key_file to --tls-private-key-
+            file (to be used by opa cli)"""
+            return "--{}".format(string.replace("_", "-"))
+
+    @validator("authentication")
+    def validate_authentication(cls, v: AuthenticationScheme):
+        if v not in [AuthenticationScheme.off, AuthenticationScheme.token]:
+            raise ValueError("Invalid AuthenticationScheme for Cedar.")
+        return v
+
+    @validator("authentication_token")
+    def validate_authentication_token(cls, v: Optional[str], values: dict[str, Any]):
+        if values['authentication'] == AuthenticationScheme.token and v is None:
+            raise ValueError("A token must be speicified for AuthenticationScheme.token.")
+        return v
+
+    def get_cmdline(self) -> str:
+        result = [
+            "cedar-agent",
+        ]
+        if self.authentication == AuthenticationScheme.token and self.authentication_token is not None:
+            result += [
+                "-a", self.authentication_token,
+            ]
+        # TODO: files
+        return " ".join(result)

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

@@ -6,7 +6,7 @@
 from opal_client.config import OpaLogFormat
 from opal_client.logger import logger
 from opal_client.opa.logger import pipe_opa_logs
-from opal_client.opa.options import OpaServerOptions
+from opal_client.opa.options import OpaServerOptions, CedarServerOptions
 from tenacity import retry, wait_random_exponential
 
 AsyncCallback = Callable[[], Coroutine]
@@ -183,8 +183,24 @@ def _init_events(self):
         if self._should_stop is None:
             self._should_stop = asyncio.Event()
 
+class OpaRunner(PolicyEngineRunner):
+    def __init__(
+        self,
+        options: Optional[OpaServerOptions] = None,
+        piped_logs_format: OpaLogFormat = OpaLogFormat.NONE,
+    ):
+        super().__init__(piped_logs_format)
+        self._options = options or OpaServerOptions()
+
+    @property
+    def command(self) -> str:
+        opts = self._options.get_cli_options_dict()
+        opts_string = " ".join([f"{k}={v}" for k, v in opts.items()])
+        startup_files = self._options.get_opa_startup_files()
+        return f"opa run --server {opts_string} {startup_files}".strip()
+
     @staticmethod
-    def setup_process_runner(
+    def setup_opa_runner(
         options: Optional[OpaServerOptions] = None,
         piped_logs_format: OpaLogFormat = OpaLogFormat.NONE,
         initial_start_callbacks: Optional[List[AsyncCallback]] = None,
@@ -202,25 +218,50 @@ def setup_process_runner(
             to handle authorization queries. therefore it is necessary that we rehydrate the
             cache with fresh state fetched from the server.
         """
-        process_runner = OpaRunner(options=options, piped_logs_format=piped_logs_format)
+        opa_runner = OpaRunner(options=options, piped_logs_format=piped_logs_format)
         if initial_start_callbacks:
-            process_runner.register_process_initial_start_callbacks(initial_start_callbacks)
+            opa_runner.register_process_initial_start_callbacks(initial_start_callbacks)
         if rehydration_callbacks:
-            process_runner.register_process_restart_callbacks(rehydration_callbacks)
-        return process_runner
+            opa_runner.register_process_restart_callbacks(rehydration_callbacks)
+        return opa_runner
 
-class OpaRunner(PolicyEngineRunner):
+
+
+class CedarRunner(PolicyEngineRunner):
     def __init__(
         self,
-        options: Optional[OpaServerOptions] = None,
+        options: Optional[CedarServerOptions] = None,
         piped_logs_format: OpaLogFormat = OpaLogFormat.NONE,
     ):
         super().__init__(piped_logs_format)
-        self._options = options or OpaServerOptions()
+        self._options = options or CedarServerOptions()
 
     @property
     def command(self) -> str:
-        opts = self._options.get_cli_options_dict()
-        opts_string = " ".join([f"{k}={v}" for k, v in opts.items()])
-        startup_files = self._options.get_opa_startup_files()
-        return f"opa run --server {opts_string} {startup_files}".strip()
+        return self._options.get_cmdline()
+
+    @staticmethod
+    def setup_cedar_runner(
+        options: Optional[CedarServerOptions] = None,
+        piped_logs_format: OpaLogFormat = OpaLogFormat.NONE,
+        initial_start_callbacks: Optional[List[AsyncCallback]] = None,
+        rehydration_callbacks: Optional[List[AsyncCallback]] = None,
+    ):
+        """factory for CedarRunner, 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 cache is clean and it does not have the state necessary
+            to handle authorization queries. therefore it is necessary that we rehydrate the
+            cache with fresh state fetched from the server.
+        """
+        cedar_runner = CedarRunner(options=options, piped_logs_format=piped_logs_format)
+        if initial_start_callbacks:
+            cedar_runner.register_process_initial_start_callbacks(initial_start_callbacks)
+        if rehydration_callbacks:
+            cedar_runner.register_process_restart_callbacks(rehydration_callbacks)
+        return cedar_runner