"""Lambda Authorizer for Jira Webhook Signature Validation. This Lambda authorizer validates incoming Jira webhook requests by verifying the HMAC signature in the X-Hub-Signature header according to Atlassian's webhook security documentation. Reference: https://developer.atlassian.com/cloud/jira/platform/webhooks/#secure-admin-webhooks """ import hashlib import hmac import json from typing import Any, Dict, Optional import boto3 import sentry_sdk from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration from config import ENVIRONMENT, JIRA_SHARED_SECRET_NAME, LAMBDA_NAME, SENTRY_DSN, logger # Initialize Sentry for error tracking if SENTRY_DSN: sentry_sdk.init( dsn=SENTRY_DSN, environment=ENVIRONMENT, integrations=[AwsLambdaIntegration(timeout_warning=True)] ) # Initialize Secrets Manager client secrets_client = boto3.client("secretsmanager") # Cache the secret value to avoid repeated API calls _cached_secret: Optional[str] = None def get_jira_shared_secret() -> str: """Retrieve the Jira webhook shared secret from AWS Secrets Manager. The secret is cached after the first retrieval to improve performance across subsequent invocations in the same Lambda execution context. Returns: str: The Jira webhook shared secret Raises: Exception: If the secret cannot be retrieved """ global _cached_secret if _cached_secret: logger.debug("Using cached Jira shared secret") return _cached_secret try: logger.info(f"Retrieving secret: {JIRA_SHARED_SECRET_NAME}") response = secrets_client.get_secret_value(SecretId=JIRA_SHARED_SECRET_NAME) _cached_secret = response["SecretString"] logger.info("Successfully retrieved Jira shared secret") return _cached_secret except Exception as e: logger.error(f"Failed to retrieve secret {JIRA_SHARED_SECRET_NAME}: {str(e)}") raise def validate_jira_signature( payload: str, signature_header: Optional[str], secret: str ) -> bool: """Validate the HMAC signature from Jira webhook. Jira webhooks include an X-Hub-Signature header with format "method=signature" where method is the hash algorithm (e.g., sha256) and signature is the HMAC hex digest of the payload signed with the webhook's secret token. Reference: https://developer.atlassian.com/cloud/jira/platform/webhooks/#secure-admin-webhooks Example from Atlassian documentation: Secret: "It's a Secret to Everybody" Payload: "Hello World!" Expected Signature: "sha256=a4771c39fbe90f317c7824e83ddef3caae9cb3d976c214ace1f2937e133263c9" Args: payload: The raw request body as a string signature_header: The X-Hub-Signature header value (e.g., "sha256=abc123...") secret: The webhook secret token configured in Jira Returns: bool: True if the signature is valid, False otherwise """ if not signature_header: logger.warning("No X-Hub-Signature header present in request") return False try: # Parse the signature header format: "method=signature" if "=" not in signature_header: logger.warning(f"Invalid signature header format: {signature_header}") return False method, given_signature = signature_header.split("=", 1) # Jira currently uses SHA-256 if method.lower() != "sha256": logger.warning(f"Unsupported hash method: {method}") return False # Calculate the expected HMAC signature # Per Atlassian docs: HMAC-SHA256(secret, payload) hash_object = hmac.new( secret.encode("utf-8"), msg=payload.encode("utf-8"), digestmod=hashlib.sha256, ) calculated_signature = hash_object.hexdigest() # Use constant-time comparison to prevent timing attacks is_valid = hmac.compare_digest(calculated_signature, given_signature) if is_valid: logger.info("Webhook signature validation successful") else: logger.warning( "Webhook signature validation failed. " f"Expected: sha256={calculated_signature}, " f"Received: {signature_header}" ) return is_valid except Exception as e: logger.error(f"Error validating webhook signature: {str(e)}") return False def generate_policy( principal_id: str, effect: str, resource: str, context: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """Generate an IAM policy document for API Gateway authorizer response. Args: principal_id: The principal user identification effect: "Allow" or "Deny" resource: The ARN of the API Gateway method context: Optional context to pass to the API Gateway integration Returns: dict: IAM policy document in the format expected by API Gateway """ policy = { "principalId": principal_id, "policyDocument": { "Version": "2012-10-17", "Statement": [ { "Action": "execute-api:Invoke", "Effect": effect, "Resource": resource } ] } } if context: policy["context"] = context return policy def handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: """Lambda authorizer entry point for validating Jira webhook signatures. This function is invoked by API Gateway as a REQUEST authorizer. It extracts the X-Hub-Signature header, validates the webhook signature against the request body, and returns an IAM policy allowing or denying the request. Expected event structure from API Gateway REQUEST authorizer: { "type": "REQUEST", "methodArn": "arn:aws:execute-api:region:account:api-id/...", "resource": "/jira-webhook", "path": "/dev/jira-webhook", "httpMethod": "POST", "headers": { "X-Hub-Signature": "sha256=...", "Content-Type": "application/json" }, "queryStringParameters": {}, "pathParameters": {}, "stageVariables": {}, "requestContext": {...}, "body": "{...}" } Args: event: API Gateway REQUEST authorizer event containing the full request context: Lambda context object Returns: dict: IAM policy document allowing or denying the request """ logger.info(f"Starting {LAMBDA_NAME}") logger.debug(f"Received authorizer event: {json.dumps(event, default=str)}") try: # Get the method ARN method_arn = event.get("methodArn", "") # Verify this is a REQUEST authorizer (not TOKEN) event_type = event.get("type", "") if event_type == "TOKEN": logger.error( "TOKEN authorizer type detected. This authorizer requires " "REQUEST type to access the request body for signature " "validation." ) return generate_policy("user", "Deny", method_arn) # Extract headers and body from REQUEST authorizer event headers = event.get("headers", {}) body = event.get("body", "") if not headers: logger.warning("No headers found in request") return generate_policy("user", "Deny", method_arn) # Extract X-Hub-Signature from headers (case-insensitive) signature_header = next( (v for k, v in headers.items() if k.lower() == "x-hub-signature"), None ) if not signature_header: logger.warning("No X-Hub-Signature header found in request") return generate_policy("user", "Deny", method_arn) # Verify we have a request body to validate if not body: logger.warning("Request body is empty - cannot validate signature") return generate_policy("user", "Deny", method_arn) # Retrieve the shared secret from Secrets Manager try: secret = get_jira_shared_secret() except Exception as e: logger.error(f"Failed to retrieve secret: {str(e)}") return generate_policy("user", "Deny", method_arn) # Validate the signature is_valid = validate_jira_signature(body, signature_header, secret) if is_valid: logger.info("Request authorized - valid Jira webhook signature") return generate_policy( principal_id="jira-webhook", effect="Allow", resource=method_arn, context={ "validated": "true", "source": "jira-webhook" } ) else: logger.warning("Request denied - invalid Jira webhook signature") return generate_policy("user", "Deny", method_arn) except Exception as e: logger.error(f"Unexpected error in authorizer: {str(e)}") return generate_policy("user", "Deny", event.get("methodArn", "*"))