AWS Lambda handlers

lambda_handler(
    _event: dict[str, Any], _context: object
) -> None

Apply all configured suppression rules to every currently matching Security Hub finding.

Raises:

• RuntimeError –

  If not all matched findings could be successfully processed. Check the logged warnings above the raised error for which finding(s)/rule(s) were affected.

lambda_handler(
    event: dict[str, Any], _context: object
) -> dict[str, str]

Suppress the findings carried by an EventBridge "Security Hub Findings - Imported" event.

Parameters:

• event

  (dict[str, Any]) –

  The EventBridge event. event["detail"]["findings"] is expected to be a list of Security Hub finding dicts.

• _context

  (object) –

  The Lambda context object (unused).

Returns:

• dict[str, str] –

  {"finding_state": "suppressed"} if at least one finding matched a rule and all matching updates were processed successfully, {"finding_state": "skipped"} otherwise (no rule matched, the manager could not be initialized, or an update failed).

lambda_handler(
    _event: dict[str, Any], _context: object
) -> None

Load the configured rules and place each one on SQS for the worker Lambda to apply.

Reads the destination queue URL from the SQS_QUEUE_URL environment variable.

Raises:

• Exception –

  Re-raises any failure to load the rules or send a message to SQS, after logging it.

lambda_handler(
    event: dict[str, Any], _context: object
) -> dict[str, list[dict[str, str]]]

Apply the suppression rule carried by each SQS record to matching Security Hub findings.

Parameters:

• event

  (dict[str, Any]) –

  The SQS event. Each event["Records"] entry's body is expected to be a JSON-encoded rules document with exactly one rule in its Rules list.

• _context

  (object) –

  The Lambda context object (unused).

Returns:

• dict[str, list[dict[str, str]]] –

  {"batchItemFailures": [{"itemIdentifier": <messageId>}, ...]}, naming only the records that failed to process. Requires function_response_types = ["ReportBatchItemFailures"] on the SQS event source mapping to take effect.

load_rules(
    s3_client: BaseClient | None = None,
) -> dict[str, Any]

Load the rules document, from S3 if configured, otherwise from a local file.

If both the S3_BUCKET_NAME and S3_OBJECT_NAME environment variables are set, the rules document is loaded from that S3 object. Otherwise, it is loaded from the local file at RULES_PATH (defaults to rules.yaml).

Parameters:

• s3_client

  (BaseClient, default: None ) –

  The boto3 S3 client to use when loading rules from S3. Only used (and required) when S3_BUCKET_NAME and S3_OBJECT_NAME are both set. Accepting it as a parameter, rather than creating one internally, keeps this function free of AWS calls in unit tests.

Returns:

• dict[str, Any] –

  The parsed rules document, expected to contain a top-level Rules key and optionally a ManagerConfig key. See awsfindingsmanagerlib.Manager.from_rules_document for turning this into a ready-to-use Manager.

Raises:

• ValueError –

  If S3 loading is configured but no s3_client was provided.