Add optional SQS SSE-SQS support at adapter level

[RishabhOrange]

User description

Summary

• Add optional sqs_managed_sse_enabled on AWSPubSubAdapter (defaults to False — fully backward compatible)
• When enabled, create_queue sets SqsManagedSseEnabled=true in CreateQueue attributes for standard and FIFO queues
• If the queue already exists, SSE is applied via SetQueueAttributes
• Add set_sqs_managed_sse_enabled() for runtime toggle
• Bump version to 0.1.22

Problem

Python services (e.g. health-api) use pubsublib create_queue, which did not set SSE on SQS queues. OMS/Citadel handle this in app-level sqs init; Python pubsub consumers had no equivalent.

Root Cause

create_queue only set VisibilityTimeout and MessageRetentionPeriod — no SqsManagedSseEnabled attribute.

Changes

• src/pubsublib/aws/main.py — adapter flag, attribute builder, enforce on existing queues
• pyproject.toml — version 0.1.22
• README.md — usage docs

Backward compatibility

Existing callers unchanged — omit sqs_managed_sse_enabled or pass False. create_queue signature is unchanged.

Usage (health-api / other consumers)

pubsub_adapter = AWSPubSubAdapter(
    ...,
    sqs_managed_sse_enabled=True,
)
pubsub_adapter.create_queue(name='my-consumer.fifo', is_fifo=True)

Test plan

• Verify existing adapter construction without new param behaves as before
• With sqs_managed_sse_enabled=True, new queue shows SSE-SQS in AWS console
• With sqs_managed_sse_enabled=True, existing unencrypted queue gets SSE after create_queue call
• Publish 0.1.22 to PyPI after merge

Made with Cursor

---

CodeAnt-AI Description

Add optional SQS server-side encryption when creating queues

What Changed

• Queue creation can now turn on SQS-managed server-side encryption from the adapter, without changing existing callers
• New queues created with this setting enabled include encryption automatically for both standard and FIFO queues
• If a queue already exists, the adapter now applies encryption to that queue instead of leaving it unchanged
• The adapter can also be switched on or off at runtime, and the README now shows how to use the setting

Impact

✅ Encrypted SQS queues from app setup
✅ Fewer manual steps for existing queues
✅ No change for current users who keep the setting off

💡 Usage Guide

Checking Your Pull Request

Every time you make a pull request, our system automatically looks through it. We check for security issues, mistakes in how you're setting up your infrastructure, and common code problems. We do this to make sure your changes are solid and won't cause any trouble later.

Talking to CodeAnt AI

Got a question or need a hand with something in your pull request? You can easily get in touch with CodeAnt AI right here. Just type the following in a comment on your pull request, and replace "Your question here" with whatever you want to ask:

@codeant-ai ask: Your question here

This lets you have a chat with CodeAnt AI about your pull request, making it easier to understand and improve your code.

Example

@codeant-ai ask: Can you suggest a safer alternative to storing this secret?

Preserve Org Learnings with CodeAnt

You can record team preferences so CodeAnt AI applies them in future reviews. Reply directly to the specific CodeAnt AI suggestion (in the same thread) and replace "Your feedback here" with your input:

@codeant-ai: Your feedback here

This helps CodeAnt AI learn and adapt to your team's coding style and standards.

Example

@codeant-ai: Do not flag unused imports.

Retrigger review

Ask CodeAnt AI to review the PR again, by typing:

@codeant-ai: review

Check Your Repository Health

To analyze the health of your code repository, visit our dashboard at https://app.codeant.ai. This tool helps you identify potential issues and areas for improvement in your codebase, ensuring your repository maintains high standards of code health.

[codeant-ai]

---

Thanks for using CodeAnt! 🎉

We're free for open-source projects. if you're enjoying it, help us grow by sharing.

Share on X ·
Reddit ·
LinkedIn

[codeant-ai]

Suggestion: SSE enforcement failures are swallowed and only logged, so callers can get a successful return path even when encryption was not actually applied (for example due to IAM denial or transient AWS failures). This creates a security gap because the method reports success while the queue may remain unencrypted. Propagate the exception (or return a failure signal checked by callers) when set_queue_attributes fails. [security]

Severity Level: Critical 🚨

- ❌ SSE failures return success, leaving SQS queues unencrypted.
- ⚠️ Violates encryption requirements for sensitive message payloads.
- ⚠️ Security audits may miss unencrypted queues in production.

Steps of Reproduction ✅

1. In a consumer service, create an `AWSPubSubAdapter` instance from
`src/pubsublib/aws/main.py:16-28` with `sqs_managed_sse_enabled=True` so that the
constructor logic at lines 53-56 enables SSE behavior.

2. In AWS, provision an SQS queue named `my-queue` without SSE; configure the IAM role
used by this adapter so that `sqs:GetQueueUrl` and `sqs:CreateQueue` succeed but
`sqs:SetQueueAttributes` (used for enabling SSE) fails with `AccessDenied`.

3. Call `adapter.create_queue(name="my-queue", is_fifo=False)` which executes
`create_queue` at `src/pubsublib/aws/main.py:315-84` and then `__create_standard_queue` at
`src/pubsublib/aws/main.py:345-119`; because the queue already exists, the
`self.sqs_client.create_queue` call at line 360 raises a `ClientError` with `Error.Code ==
"QueueNameExists"`, triggering the `except` block at line 371.

4. Inside the `except` block, the condition on line 372 (`if self.sqs_managed_sse_enabled
and error_code in ("QueueNameExists", "QueueAlreadyExists"):`) is true, so
`_enforce_sqs_managed_sse(name)` is called; `_enforce_sqs_managed_sse` at
`src/pubsublib/aws/main.py:84-101` successfully gets the queue URL at line 86 but
`set_queue_attributes` at lines 94-97 raises a `ClientError` due to IAM denial, which is
caught by the inner `except ClientError` at lines 99-101 that only logs `\"Couldn't
enforce SSE on existing queue %s.\"` and returns, after which `__create_standard_queue`
continues at line 374 to return `self.sqs_client.get_queue_url(QueueName=name)`, causing
the caller to see a successful result even though SSE was never applied and the queue
remains unencrypted.

(Use Cmd/Ctrl + Click for best experience)

Prompt for AI Agent 🤖

This is a comment left during a code review.

**Path:** src/pubsublib/aws/main.py
**Line:** 93:100
**Comment:**
	*Security: SSE enforcement failures are swallowed and only logged, so callers can get a successful return path even when encryption was not actually applied (for example due to IAM denial or transient AWS failures). This creates a security gap because the method reports success while the queue may remain unencrypted. Propagate the exception (or return a failure signal checked by callers) when `set_queue_attributes` fails.

Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix

👍 | 👎

[codeant-ai]

Suggestion: The QueueNameExists/QueueAlreadyExists branch now treats every name-collision error as recoverable when SSE is enabled, but AWS uses this error code for any attribute mismatch (visibility timeout, retention, FIFO flags, etc.), not just SSE. This will silently return an existing queue URL even when the queue configuration is incompatible with the requested settings. Only suppress this error after explicitly validating that the existing queue's non-SSE attributes match the requested ones; otherwise re-raise the exception. [incorrect condition logic]

Severity Level: Major ⚠️

- ❌ Queue creation silently ignores mismatched visibility/retention/FIFO attributes.
- ⚠️ Downstream consumers misconfigured to expect different SQS queue behavior.
- ⚠️ Operational debugging harder; misconfigurable queues appear successfully provisioned.

Steps of Reproduction ✅

1. In a consumer service, instantiate `AWSPubSubAdapter` from
`src/pubsublib/aws/main.py:16-28` with `sqs_managed_sse_enabled=True` so the constructor
logic at lines 53-56 sets `self.sqs_managed_sse_enabled` to `True`.

2. From that service, call `adapter.create_queue(name="my-queue", is_fifo=False,
visiblity_timeout=60, message_retention_period=86400, tags={})`, which invokes
`create_queue` defined at `src/pubsublib/aws/main.py:315-84` and then routes to
`__create_standard_queue` defined at `src/pubsublib/aws/main.py:345-92`.

3. Ensure an SQS queue named `my-queue` already exists in AWS with non-SSE attributes that
differ from the requested ones (for example a different `VisibilityTimeout` or
`MessageRetentionPeriod`); the call to `self.sqs_client.create_queue` inside
`__create_standard_queue` at `src/pubsublib/aws/main.py:360-109` then raises a
`ClientError` with `Error.Code == "QueueNameExists"` according to SQS semantics for
attribute mismatches.

4. The `ClientError` is caught by the `except` block at
`src/pubsublib/aws/main.py:371-117`, where line 372 executes `if
self.sqs_managed_sse_enabled and error_code in ("QueueNameExists", "QueueAlreadyExists"):`
and line 373 calls `_enforce_sqs_managed_sse(name)` before line 374 returns
`self.sqs_client.get_queue_url(QueueName=name)`; no validation of visibility timeout,
retention period, FIFO flags, or deduplication settings is performed, so the method
reports success while the existing queue's non-SSE configuration remains incompatible with
the requested settings.

(Use Cmd/Ctrl + Click for best experience)

Prompt for AI Agent 🤖

This is a comment left during a code review.

**Path:** src/pubsublib/aws/main.py
**Line:** 372:374
**Comment:**
	*Incorrect Condition Logic: The `QueueNameExists`/`QueueAlreadyExists` branch now treats every name-collision error as recoverable when SSE is enabled, but AWS uses this error code for any attribute mismatch (visibility timeout, retention, FIFO flags, etc.), not just SSE. This will silently return an existing queue URL even when the queue configuration is incompatible with the requested settings. Only suppress this error after explicitly validating that the existing queue's non-SSE attributes match the requested ones; otherwise re-raise the exception.

Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix

👍 | 👎