Enable OAuth2 authentication.

Author: Ondrej Scecina

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

@@ -39,11 +39,27 @@
 from opal_common.schemas.store import TransactionType
 from opal_common.security.sslcontext import get_custom_ssl_context
 from opal_common.synchronization.hierarchical_lock import HierarchicalLock
-from opal_common.utils import get_authorization_header
+from opal_common.utils import (
+    get_authorization_header,
+    tuple_to_dict,
+)
 from pydantic.json import pydantic_encoder
 
 
 class DataUpdater:
+    async def get_base_policy_data(
+        self, config_url: str = None, data_fetch_reason="Initial load"
+    ):
+        pass
+
+    async def stop(self):
+        pass
+
+    async def wait_until_done(self):
+        pass
+
+
+class DefaultDataUpdater(DataUpdater):
     """The DataUpdater is responsible for synchronizing data sources with the
     policy store (e.g. OPA). It listens to Pub/Sub topics for data updates,
     fetches the updated data, and writes it into the policy store. The updater
@@ -142,11 +158,11 @@ def __init__(
         self._opal_client_id = opal_client_id
 
         # Prepare any extra headers (token, shard id, etc.)
-        self._extra_headers = []
-        if self._token is not None:
-            self._extra_headers.append(get_authorization_header(self._token))
+        self._extra_headers = {}
+        if self._token is not None and opal_common_config.AUTH_TYPE != "oauth2":
+            self._extra_headers = tuple_to_dict(get_authorization_header(self._token))
         if self._shard_id is not None:
-            self._extra_headers.append(("X-Shard-ID", self._shard_id))
+            self._extra_headers["X-Shard-ID"] = self._shard_id
         if len(self._extra_headers) == 0:
             self._extra_headers = None
 
@@ -236,11 +252,11 @@ async def get_policy_data_config(self, url: str = None) -> DataSourceConfig:
             url = self._data_sources_config_url
         logger.info("Getting data-sources configuration from '{source}'", source=url)
 
-
         headers = {}
         if self._extra_headers is not None:
             headers = self._extra_headers.copy()
-        headers['Accept'] = "application/json"
+        headers["Accept"] = "application/json"
+        await self._authenticator.authenticate(headers)
 
         try:
             response = await self._load_policy_data_config(url, headers)
@@ -256,7 +272,9 @@ async def get_policy_data_config(self, url: str = None) -> DataSourceConfig:
             logger.exception("Failed to load data sources config")
             raise
 
-    async def _load_policy_data_config(self, url: str, headers) -> aiohttp.ClientResponse:
+    async def _load_policy_data_config(
+        self, url: str, headers
+    ) -> aiohttp.ClientResponse:
         async with ClientSession(headers=headers, trust_env=True) as session:
             return await session.get(url, **self._ssl_context_kwargs)
 
@@ -370,8 +388,7 @@ async def _subscriber(self):
             methods_class=TenantAwareRpcEventClientMethods,
             on_connect=[self.on_connect, *self._on_connect_callbacks],
             on_disconnect=[self.on_disconnect, *self._on_disconnect_callbacks],
-            additional_headers=self._extra_headers,
-            extra_headers=headers,
+            additional_headers=headers,
             keep_alive=opal_client_config.KEEP_ALIVE_INTERVAL,
             server_uri=self._server_url,
             **self._ssl_context_kwargs,

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

@@ -6,6 +6,7 @@
 from opal_client.logger import logger
 from opal_common.authentication.authenticator import Authenticator
 from opal_common.authentication.authenticator_factory import AuthenticatorFactory
+from opal_common.config import opal_common_config
 from opal_common.schemas.policy import PolicyBundle
 from opal_common.security.sslcontext import get_custom_ssl_context
 from opal_common.utils import (
@@ -47,7 +48,7 @@ def __init__(
             self._authenticator = AuthenticatorFactory.create()
         self._token = token or opal_client_config.CLIENT_TOKEN
         self._backend_url = backend_url or opal_client_config.SERVER_URL
-        if self._token is not None:
+        if self._token is not None and opal_common_config.AUTH_TYPE != "oauth2":
             self._auth_headers = tuple_to_dict(get_authorization_header(self._token))
         else:
             self._auth_headers = dict()

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

@@ -17,13 +17,17 @@
     DEFAULT_POLICY_STORE_GETTER,
 )
 from opal_common.async_utils import TakeANumberQueue, TasksPool
+from opal_common.authentication.authenticator import Authenticator
 from opal_common.config import opal_common_config
 from opal_common.schemas.data import DataUpdateReport
 from opal_common.schemas.policy import PolicyBundle, PolicyUpdateMessage
 from opal_common.schemas.store import TransactionType
 from opal_common.security.sslcontext import get_custom_ssl_context
 from opal_common.topics.utils import pubsub_topics_from_directories
-from opal_common.utils import get_authorization_header
+from opal_common.utils import (
+    get_authorization_header,
+    tuple_to_dict,
+)
 
 
 class PolicyUpdater:
@@ -77,10 +81,11 @@ def __init__(
         # pub/sub server url and authentication data
         self._server_url = pubsub_url
         self._token = token
-        if self._token is None:
+        self._extra_headers = {}
+        if self._token is not None and opal_common_config.AUTH_TYPE != "oauth2":
+            self._extra_headers = tuple_to_dict(get_authorization_header(self._token))
+        if len(self._extra_headers) == 0:
             self._extra_headers = None
-        else:
-            self._extra_headers = [get_authorization_header(self._token)]
         # Pub/Sub topics we subscribe to for policy updates
         if self._scope_id == "default":
             self._topics = pubsub_topics_from_directories(
@@ -261,7 +266,7 @@ async def _subscriber(self):
             callback=self._update_policy_callback,
             on_connect=[self._on_connect, *self._on_connect_callbacks],
             on_disconnect=[self._on_disconnect, *self._on_disconnect_callbacks],
-            extra_headers=headers,
+            additional_headers=headers,
             keep_alive=opal_client_config.KEEP_ALIVE_INTERVAL,
             server_uri=self._server_url,
             **self._ssl_context_kwargs,

packages/opal-client/opal_client/policy_store/opa_client.py

@@ -430,7 +430,7 @@ async def _get_oauth_token(self):
                 ) as oauth_response:
                     response = await oauth_response.json()
                     logger.info(
-                        f"got access_token, expires in {response['expires_in']} seconds"
+                        f"Got access_token, expires in {response['expires_in']} seconds"
                     )
 
                     return {

packages/opal-common/opal_common/authentication/deps.py

@@ -78,10 +78,11 @@ def verifier(self) -> JWTVerifier:
         return self._verifier
 
     def signer(self) -> Optional[JWTSigner]:
+        logger.info("!!!!! signer == _verifier")
         return self._verifier
 
     @property
-    def enabled(self) -> JWTVerifier:
+    def enabled(self) -> bool:
         return self._verifier.enabled
 
     async def authenticate(self, headers):

packages/opal-common/opal_common/authentication/oauth2.py

@@ -155,7 +155,7 @@ async def token(self):
             claims = self._delegate.verify(token)
 
             self._token = token
-            self._exp = claims["exp"]
+            self._exp = claims.get("exp")
 
             return self._token
 

packages/requires.txt

@@ -13,3 +13,4 @@ fastapi-utils>=0.2.1,<1
 setuptools>=70.0.0 # not directly required, pinned by Snyk to avoid a vulnerability
 anyio>=4.4.0 # not directly required, pinned by Snyk to avoid a vulnerability
 starlette>=0.40.0 # not directly required, pinned by Snyk to avoid a vulnerability
+cachetools>=5.3.3