Adding CA certificates logic for verification (#92)

Author: tmikula-dev

README.md

@@ -82,9 +82,22 @@ Example (sanitized):
   "token_provider_url": "https://<token-ui.example>",
   "token_public_keys_url": "https://<token-api.example>/token/public-keys",
   "kafka_bootstrap_server": "broker1:9092,broker2:9092",
-  "event_bus_arn": "arn:aws:events:region:acct:event-bus/your-bus"
+  "event_bus_arn": "arn:aws:events:region:acct:event-bus/your-bus",
+  "ssl_ca_bundle": "/path/to/ca-bundle.pem"
 }
 ```
+
+Configuration keys:
+- `access_config` – local file path or S3 URI for access control map.
+- `token_provider_url` – external URL for obtaining JWT tokens.
+- `token_public_keys_url` – endpoint serving JWT verification public keys (RS256).
+- `kafka_bootstrap_server` – comma-separated Kafka broker addresses.
+- `event_bus_arn` – AWS EventBridge event bus ARN for EventBridge writer.
+- `ssl_ca_bundle` (optional) – SSL certificate verification for S3 access and token public key requests.
+  - `true` - default, uses system CA bundle
+  - `false` - disables verification, not recommended for production
+  - `"/path/to/ca-bundle.pem"` - custom CA bundle
+
 Supporting configs:
 - `access.json` – map: topicName -> array of authorized subjects (JWT `sub`). May reside locally or at S3 path referenced by `access_config`.
 - `topic_*.json` – each file contains a JSON Schema for a topic. In the current code these are explicitly loaded inside `event_gate_lambda.py`. (Future enhancement: auto-discover or index file.)

conf/config.json

@@ -3,5 +3,6 @@
     "token_provider_url": "https://<redacted>",
     "token_public_keys_url": "https://<redacted>",
     "kafka_bootstrap_server": "localhost:9092",
-    "event_bus_arn": "arn:aws:events:<redacted>"
+    "event_bus_arn": "arn:aws:events:<redacted>",
+    "ssl_ca_bundle": "/path/to/ca-bundle.pem"
 }

scripts/notebook.ipynb

@@ -23,19 +23,19 @@
 
 import boto3
 import jwt
-import urllib3
+from botocore.exceptions import BotoCoreError, NoCredentialsError
 from jsonschema import validate
 from jsonschema.exceptions import ValidationError
 
 from src.handlers.handler_token import HandlerToken
+from src.utils.constants import SSL_CA_BUNDLE_KEY
 from src.writers import writer_eventbridge, writer_kafka, writer_postgres
 from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
 
 # Internal aliases used by rest of module
 _CONF_DIR = CONF_DIR
 _INVALID_CONF_ENV = INVALID_CONF_ENV
 
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
 logger = logging.getLogger(__name__)
 log_level = os.environ.get("LOG_LEVEL", "INFO")
@@ -64,8 +64,14 @@
     config = json.load(file)
 logger.debug("Loaded main CONFIG")
 
-aws_s3 = boto3.Session().resource("s3", verify=False)  # nosec Boto verify disabled intentionally
-logger.debug("Initialized AWS S3 Client")
+# Initialize S3 client with SSL verification
+try:
+    ssl_verify = config.get(SSL_CA_BUNDLE_KEY, True)
+    aws_s3 = boto3.Session().resource("s3", verify=ssl_verify)
+    logger.debug("Initialized AWS S3 Client")
+except (BotoCoreError, NoCredentialsError) as exc:
+    logger.exception("Failed to initialize AWS S3 client")
+    raise RuntimeError("AWS S3 client initialization failed") from exc
 
 if config["access_config"].startswith("s3://"):
     name_parts = config["access_config"].split("/")

scripts/prepare.deplyoment.sh

@@ -30,7 +30,12 @@
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 
-from src.utils.constants import TOKEN_PROVIDER_URL_KEY, TOKEN_PUBLIC_KEYS_URL_KEY, TOKEN_PUBLIC_KEY_URL_KEY
+from src.utils.constants import (
+    TOKEN_PROVIDER_URL_KEY,
+    TOKEN_PUBLIC_KEYS_URL_KEY,
+    TOKEN_PUBLIC_KEY_URL_KEY,
+    SSL_CA_BUNDLE_KEY,
+)
 
 logger = logging.getLogger(__name__)
 log_level = os.environ.get("LOG_LEVEL", "INFO")
@@ -49,6 +54,7 @@ def __init__(self, config):
         self.public_keys_url: str = config.get(TOKEN_PUBLIC_KEYS_URL_KEY) or config.get(TOKEN_PUBLIC_KEY_URL_KEY)
         self.public_keys: list[RSAPublicKey] = []
         self._last_loaded_at: datetime | None = None
+        self.ssl_ca_bundle: str | bool = config.get(SSL_CA_BUNDLE_KEY, True)
 
     def _refresh_keys_if_needed(self) -> None:
         """
@@ -79,7 +85,7 @@ def load_public_keys(self) -> "HandlerToken":
         logger.debug("Loading token public keys from %s", self.public_keys_url)
 
         try:
-            response_json = requests.get(self.public_keys_url, verify=False, timeout=5).json()
+            response_json = requests.get(self.public_keys_url, verify=self.ssl_ca_bundle, timeout=5).json()
             raw_keys: list[str] = []
 
             if isinstance(response_json, dict):

src/event_gate_lambda.py

@@ -18,7 +18,8 @@
 This module contains all constants and enums used across the project.
 """
 
-# Token related configuration keys
+# Configuration keys
 TOKEN_PROVIDER_URL_KEY = "token_provider_url"
 TOKEN_PUBLIC_KEY_URL_KEY = "token_public_key_url"
 TOKEN_PUBLIC_KEYS_URL_KEY = "token_public_keys_url"
+SSL_CA_BUNDLE_KEY = "ssl_ca_bundle"

src/handlers/handler_token.py

@@ -142,3 +142,17 @@ def test_decode_jwt_triggers_refresh_check(token_handler):
         with patch("jwt.decode", return_value={"sub": "TestUser"}):
             token_handler.decode_jwt("dummy-token")
             mock_refresh.assert_called_once()
+
+
+def test_handler_token_default_ssl_ca_bundle():
+    """HandlerToken should default to True for ssl_ca_bundle when not specified."""
+    config = {"token_public_keys_url": "https://example.com/keys"}
+    handler = HandlerToken(config)
+    assert handler.ssl_ca_bundle is True
+
+
+def test_handler_token_custom_ssl_ca_bundle_path():
+    """HandlerToken should accept custom CA bundle path."""
+    config = {"token_public_keys_url": "https://example.com/keys", "ssl_ca_bundle": "/path/to/custom/ca-bundle.pem"}
+    handler = HandlerToken(config)
+    assert handler.ssl_ca_bundle == "/path/to/custom/ca-bundle.pem"

src/utils/constants.py

@@ -15,7 +15,7 @@
 #
 
 import json
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 
 
 # --- GET flows ---
@@ -230,3 +230,31 @@ def test_post_invalid_json_body(event_gate_module, make_event):
     assert resp["statusCode"] == 500
     body = json.loads(resp["body"])
     assert any(e["type"] == "internal" for e in body["errors"])  # internal error path
+
+
+def test_boto3_s3_client_default_ssl_verification():
+    """Test that boto3 S3 client uses default SSL verification when ssl_ca_bundle not specified."""
+    config = {}
+
+    with patch("boto3.Session") as mock_session:
+        mock_session_instance = MagicMock()
+        mock_session.return_value = mock_session_instance
+
+        ssl_verify = config.get("ssl_ca_bundle", True)
+        mock_session_instance.resource("s3", verify=ssl_verify)
+
+        mock_session_instance.resource.assert_called_once_with("s3", verify=True)
+
+
+def test_boto3_s3_client_custom_ca_bundle():
+    """Test that boto3 S3 client uses custom CA bundle when ssl_ca_bundle is specified."""
+    config = {"ssl_ca_bundle": "/path/to/custom-ca-bundle.pem"}
+
+    with patch("boto3.Session") as mock_session:
+        mock_session_instance = MagicMock()
+        mock_session.return_value = mock_session_instance
+
+        ssl_verify = config.get("ssl_ca_bundle", True)
+        mock_session_instance.resource("s3", verify=ssl_verify)
+
+        mock_session_instance.resource.assert_called_once_with("s3", verify="/path/to/custom-ca-bundle.pem")