implemented mtls for all outgoing https traffic (expect local opa)

horizon/facts/client.py

@@ -9,13 +9,15 @@
 from starlette.responses import Response as FastApiResponse, StreamingResponse
 
 from horizon.config import sidecar_config
+from horizon.ssl import get_mtls_httpx_kwargs
 from horizon.startup.remote_config import get_remote_config
 from horizon.startup.api_keys import get_env_api_key
 
 
 class FactsClient:
     def __init__(self):
         self._client: Optional[AsyncClient] = None
+        self._mtls_kwargs = get_mtls_httpx_kwargs()
 
     @property
     def client(self) -> AsyncClient:
@@ -24,6 +26,7 @@ def client(self) -> AsyncClient:
             self._client = AsyncClient(
                 base_url=sidecar_config.CONTROL_PLANE,
                 headers={"Authorization": f"Bearer {env_api_key}"},
+                **self._mtls_kwargs,
             )
         return self._client
 

horizon/pdp.py

@@ -30,7 +30,7 @@
 )
 from horizon.local.api import init_local_cache_api_router
 from horizon.opal_relay_api import OpalRelayAPIClient
-from horizon.proxy.api import router as proxy_router
+from horizon.proxy.api import init_cloud_proxy_router
 from horizon.startup.remote_config import get_remote_config
 from horizon.startup.exceptions import InvalidPDPTokenException
 from horizon.startup.api_keys import get_env_api_key
@@ -331,6 +331,8 @@ def _configure_api_routes(self, app: FastAPI):
         # Init system router
         system_router = init_system_api_router()
 
+        proxy_router = init_cloud_proxy_router()
+
         # include the api routes
         app.include_router(
             enforcer_router,

horizon/proxy/api.py

@@ -12,6 +12,7 @@
 from pydantic import BaseModel, Field, parse_obj_as
 
 from horizon.config import sidecar_config
+from horizon.ssl import get_mtls_aiohttp_kwargs
 
 HTTP_GET = "GET"
 HTTP_DELETE = "DELETE"
@@ -42,9 +43,6 @@ class JSONPatchAction(BaseModel):
     )
 
 
-router = APIRouter()
-
-
 async def patch_handler(response: Response) -> Response:
     """
     Handle write APIs (from the SDK) where OpalClient will have to be manually updated from sidecar.
@@ -82,101 +80,106 @@ async def patch_handler(response: Response) -> Response:
 }
 
 
-@router.api_route(
-    "/cloud/{path:path}",
-    methods=ALL_METHODS,
-    summary="Proxy Endpoint",
-    include_in_schema=False,
-)
-async def cloud_proxy(request: Request, path: str):
-    """
-    Proxies the request to the cloud API. Actual API docs are located here: https://api.permit.io/redoc
-    """
-    write_route = any(
-        request.method == route[0] and route[1].match(request.path_params["path"])
-        for route in write_routes
-    )
+def init_cloud_proxy_router() -> APIRouter:
+    mtls_kwargs = get_mtls_aiohttp_kwargs()
 
-    headers = {}
-    if write_route:
-        headers["X-Include-Patch"] = "true"
+    router = APIRouter()
 
-    response = await proxy_request_to_cloud_service(
-        request,
-        path,
-        cloud_service_url=sidecar_config.BACKEND_SERVICE_URL,
-        additional_headers=headers,
+    @router.api_route(
+        "/cloud/{path:path}",
+        methods=ALL_METHODS,
+        summary="Proxy Endpoint",
+        include_in_schema=False,
     )
+    async def cloud_proxy(request: Request, path: str):
+        """
+        Proxies the request to the cloud API. Actual API docs are located here: https://api.permit.io/redoc
+        """
+        write_route = any(
+            request.method == route[0] and route[1].match(request.path_params["path"])
+            for route in write_routes
+        )
 
-    if write_route:
-        return await patch_handler(response)
-
-    return response
+        headers = {}
+        if write_route:
+            headers["X-Include-Patch"] = "true"
 
+        response = await proxy_request_to_cloud_service(
+            request,
+            path,
+            cloud_service_url=sidecar_config.BACKEND_SERVICE_URL,
+            additional_headers=headers,
+            request_kwargs=mtls_kwargs,
+        )
 
-@router.api_route(
-    "/healthchecks/opa/ready",
-    methods=[HTTP_GET],
-    summary="Proxy ready healthcheck - OPAL_OPA_HEALTH_CHECK_POLICY_ENABLED must be set to True",
-)
-async def ready_opa_healthcheck(request: Request):
-    return await proxy_request_to_cloud_service(
-        request,
-        path="v1/data/system/opal/ready",
-        cloud_service_url=opal_client_config.POLICY_STORE_URL,
-        additional_headers={},
-    )
+        if write_route:
+            return await patch_handler(response)
 
+        return response
 
-@router.api_route(
-    "/healthchecks/opa/healthy",
-    methods=[HTTP_GET],
-    summary="Proxy healthy healthcheck -  OPAL_OPA_HEALTH_CHECK_POLICY_ENABLED must be set to True",
-)
-async def health_opa_healthcheck(request: Request):
-    return await proxy_request_to_cloud_service(
-        request,
-        path="v1/data/system/opal/healthy",
-        cloud_service_url=opal_client_config.POLICY_STORE_URL,
-        additional_headers={},
+    @router.api_route(
+        "/healthchecks/opa/ready",
+        methods=[HTTP_GET],
+        summary="Proxy ready healthcheck - OPAL_OPA_HEALTH_CHECK_POLICY_ENABLED must be set to True",
     )
+    async def ready_opa_healthcheck(request: Request):
+        return await proxy_request_to_cloud_service(
+            request,
+            path="v1/data/system/opal/ready",
+            cloud_service_url=opal_client_config.POLICY_STORE_URL,
+            additional_headers={},
+        )
 
-
-@router.api_route(
-    "/healthchecks/opa/system",
-    methods=[HTTP_GET],
-    summary="Proxy system data -  OPAL_OPA_HEALTH_CHECK_POLICY_ENABLED must be set to True",
-)
-async def system_opa_healthcheck(request: Request):
-    return await proxy_request_to_cloud_service(
-        request,
-        path="v1/data/system/opal",
-        cloud_service_url=opal_client_config.POLICY_STORE_URL,
-        additional_headers={},
+    @router.api_route(
+        "/healthchecks/opa/healthy",
+        methods=[HTTP_GET],
+        summary="Proxy healthy healthcheck -  OPAL_OPA_HEALTH_CHECK_POLICY_ENABLED must be set to True",
     )
+    async def health_opa_healthcheck(request: Request):
+        return await proxy_request_to_cloud_service(
+            request,
+            path="v1/data/system/opal/healthy",
+            cloud_service_url=opal_client_config.POLICY_STORE_URL,
+            additional_headers={},
+        )
 
+    @router.api_route(
+        "/healthchecks/opa/system",
+        methods=[HTTP_GET],
+        summary="Proxy system data -  OPAL_OPA_HEALTH_CHECK_POLICY_ENABLED must be set to True",
+    )
+    async def system_opa_healthcheck(request: Request):
+        return await proxy_request_to_cloud_service(
+            request,
+            path="v1/data/system/opal",
+            cloud_service_url=opal_client_config.POLICY_STORE_URL,
+            additional_headers={},
+        )
 
-# TODO: remove this once we migrate all clients
-@router.api_route(
-    "/sdk/{path:path}",
-    methods=ALL_METHODS,
-    summary="Old Proxy Endpoint",
-    include_in_schema=False,
-)
-async def old_proxy(request: Request, path: str):
-    return await proxy_request_to_cloud_service(
-        request,
-        path,
-        cloud_service_url=sidecar_config.BACKEND_LEGACY_URL,
-        additional_headers={},
+    # TODO: remove this once we migrate all clients
+    @router.api_route(
+        "/sdk/{path:path}",
+        methods=ALL_METHODS,
+        summary="Old Proxy Endpoint",
+        include_in_schema=False,
     )
+    async def old_proxy(request: Request, path: str):
+        return await proxy_request_to_cloud_service(
+            request,
+            path,
+            cloud_service_url=sidecar_config.BACKEND_LEGACY_URL,
+            additional_headers={},
+        )
+
+    return router
 
 
 async def proxy_request_to_cloud_service(
     request: Request,
     path: str,
     cloud_service_url: str,
     additional_headers: Dict[str, str],
+    request_kwargs: dict = {},
 ) -> Response:
     auth_header = request.headers.get("Authorization")
     if auth_header is None:
@@ -210,13 +213,13 @@ async def proxy_request_to_cloud_service(
     async with aiohttp.ClientSession() as session:
         if request.method == HTTP_GET:
             async with session.get(
-                path, headers=headers, params=params
+                path, headers=headers, params=params, **request_kwargs
             ) as backend_response:
                 return await proxy_response(backend_response)
 
         if request.method == HTTP_DELETE:
             async with session.delete(
-                path, headers=headers, params=params
+                path, headers=headers, params=params, **request_kwargs
             ) as backend_response:
                 return await proxy_response(backend_response)
 
@@ -225,19 +228,19 @@ async def proxy_request_to_cloud_service(
 
         if request.method == HTTP_POST:
             async with session.post(
-                path, headers=headers, data=data, params=params
+                path, headers=headers, data=data, params=params, **request_kwargs
             ) as backend_response:
                 return await proxy_response(backend_response)
 
         if request.method == HTTP_PUT:
             async with session.put(
-                path, headers=headers, data=data, params=params
+                path, headers=headers, data=data, params=params, **request_kwargs
             ) as backend_response:
                 return await proxy_response(backend_response)
 
         if request.method == HTTP_PATCH:
             async with session.patch(
-                path, headers=headers, data=data, params=params
+                path, headers=headers, data=data, params=params, **request_kwargs
             ) as backend_response:
                 return await proxy_response(backend_response)
 

horizon/ssl.py

@@ -0,0 +1,67 @@
+from typing import Optional
+
+from loguru import logger
+from opal_common.config import opal_common_config
+from opal_common.security.sslcontext import (
+    CustomSSLContext,
+    get_custom_ssl_context_for_mtls,
+)
+
+
+def get_mtls_context() -> Optional[CustomSSLContext]:
+    custom_ssl_context: Optional[CustomSSLContext] = None
+
+    if (
+        opal_common_config.MTLS_CLIENT_CERT is not None
+        and opal_common_config.MTLS_CLIENT_KEY is not None
+    ):
+        custom_ssl_context = get_custom_ssl_context_for_mtls(
+            client_cert_file=opal_common_config.MTLS_CLIENT_CERT,
+            client_key_file=opal_common_config.MTLS_CLIENT_KEY,
+            ca_file=opal_common_config.MTLS_CA_CERT,
+        )
+
+        if custom_ssl_context is not None:
+            logger.info(
+                "Using client mTLS SSL context, client_cert_file={}, client_key_file={}, ca_file={}",
+                custom_ssl_context.certfile,
+                custom_ssl_context.keyfile,
+                custom_ssl_context.cafile,
+            )
+            return custom_ssl_context
+
+    return None
+
+
+def get_mtls_requests_kwargs() -> dict:
+    custom_ssl_context: Optional[CustomSSLContext] = get_mtls_context()
+
+    if custom_ssl_context is not None:
+        return dict(
+            cert=(
+                custom_ssl_context.certfile,
+                custom_ssl_context.keyfile,
+            ),
+            verify=(
+                custom_ssl_context.cafile
+                if custom_ssl_context.cafile is not None
+                else True
+            ),
+        )
+
+    return dict()
+
+
+def get_mtls_aiohttp_kwargs() -> dict:
+    custom_ssl_context: Optional[CustomSSLContext] = get_mtls_context()
+
+    return (
+        dict(ssl=custom_ssl_context.ssl_context)
+        if custom_ssl_context is not None
+        else dict()
+    )
+
+
+def get_mtls_httpx_kwargs() -> dict:
+    # same params as in requests library
+    return get_mtls_requests_kwargs()