Feat: SQS policy config (#334)

* Adding policy config types

* Generating policy from config

* SNS ignoring new prop

* release prepare

* Adding tests

* Lint fixes

* Adding doc

* Export fixes

Author: CarlosGamero

README.md

@@ -32,6 +32,7 @@ They implement the following public methods:
         * `messageTypeField` - which field in the message describes the type of a message. This field needs to be defined as `z.literal` in the schema and is used for resolving the correct schema for validation
         * `locatorConfig` - configuration for resolving existing queue and/or topic. Should not be specified together with the `creationConfig`.
         * `creationConfig` - configuration for queue and/or topic to create, if one does not exist. Should not be specified together with the `locatorConfig`;
+        * `policyConfig` - SQS only - configuration for queue access policies (see [SQS Policy Configuration](#sqs-policy-configuration) for more information);
         * `deletionConfig` - automatic cleanup of resources;
         * `handlerSpy` - allow awaiting certain messages to be published (see [Handler Spies](#handler-spies) for more information);
         * `logMessages` - add logs for processed messages.
@@ -99,6 +100,7 @@ Multi-schema consumers support multiple message types via handler configs. They
         * `locatorConfig` - configuration for resolving existing queue and/or topic. Should not be specified together with the `creationConfig`.
         * `creationConfig` - configuration for queue and/or topic to create, if one does not exist. Should not be specified together with the `locatorConfig`.
         * `subscriptionConfig` - SNS SQS consumer only - configuration for SNS -> SQS subscription to create, if one doesn't exist.
+        * `policyConfig` - SQS only - configuration for queue access policies (see [SQS Policy Configuration](#sqs-policy-configuration) for more information);
         * `deletionConfig` - automatic cleanup of resources;
         * `consumerOverrides` – available only for SQS consumers;
         * `deadLetterQueue` - available only for SQS and SNS consumers (see [Dead Letter Queue](#dead-letter-queue) for more information);
@@ -237,6 +239,143 @@ Both publishers and consumers accept a queue name and configuration as parameter
 
 If you do not want to create a new queue/topic, you can set `queueLocator` field for `queueConfiguration`. In that case `message-queue-toolkit` will not attempt to create a new queue or topic, and instead throw an error if they don't already exist.
 
+## SQS Policy Configuration
+
+SQS queues can be configured with access policies to control who can send messages to and receive messages from the queue. The `policyConfig` parameter allows you to define these policies when creating or updating SQS queues.
+
+### Policy Configuration Options
+
+The `policyConfig` parameter accepts the following structure:
+
+```typescript
+type SQSPolicyConfig = {
+  resource: string | typeof SQS_RESOURCE_ANY | typeof SQS_RESOURCE_CURRENT_QUEUE
+  statements?: SQSPolicyStatement | SQSPolicyStatement[]
+}
+
+type SQSPolicyStatement = {
+  Effect?: string        // "Allow" or "Deny" (default: "Allow")
+  Principal?: string    // AWS principal ARN (default: "*")
+  Action?: string[]     // SQS actions (default: ["sqs:SendMessage", "sqs:GetQueueAttributes", "sqs:GetQueueUrl"])
+}
+```
+
+### Resource Types
+
+The `resource` field supports three types of values:
+
+- **`SQS_RESOURCE_CURRENT_QUEUE`**: Uses the ARN of the current queue being created
+- **`SQS_RESOURCE_ANY`**: Uses the wildcard ARN `arn:aws:sqs:*:*:*`
+- **Custom ARN**: Any specific SQS queue ARN (e.g., `arn:aws:sqs:us-east-1:123456789012:my-queue`)
+
+### Usage Examples
+
+#### Basic Policy Configuration
+
+```typescript
+import { SQS_RESOURCE_ANY } from "@message-queue-toolkit/sqs"
+
+const publisher = new SqsPermissionPublisher(dependencies, {
+  creationConfig: {
+    queue: {
+      QueueName: "my-queue",
+      Attributes: {
+        VisibilityTimeout: "30",
+      },
+    },
+    policyConfig: {
+      resource: SQS_RESOURCE_ANY,
+      statements: {
+        Effect: "Allow",
+        Principal: "arn:aws:iam::123456789012:user/my-user",
+        Action: ["sqs:SendMessage", "sqs:ReceiveMessage"],
+      },
+    },
+  },
+})
+```
+
+#### Multiple Policy Statements
+
+```typescript
+const consumer = new SqsPermissionConsumer(dependencies, {
+  creationConfig: {
+    queue: {
+      QueueName: "my-queue",
+    },
+    policyConfig: {
+      resource: SQS_RESOURCE_CURRENT_QUEUE,
+      statements: [
+        {
+          Effect: "Allow",
+          Principal: "arn:aws:iam::123456789012:role/my-role",
+          Action: ["sqs:SendMessage", "sqs:ReceiveMessage"],
+        },
+        {
+          Effect: "Deny",
+          Principal: "arn:aws:iam::123456789012:user/blocked-user",
+          Action: ["sqs:SendMessage"],
+        },
+      ],
+    },
+  },
+})
+```
+
+#### Policy Updates
+
+To update an existing queue's policy, use `updateAttributesIfExists: true`:
+
+```typescript
+const updatedPublisher = new SqsPermissionPublisher(dependencies, {
+  creationConfig: {
+    queue: {
+      QueueName: "existing-queue",
+    },
+    policyConfig: {
+      resource: SQS_RESOURCE_ANY,
+      statements: {
+        Effect: "Allow",
+        Principal: "arn:aws:iam::123456789012:user/new-user",
+        Action: ["sqs:SendMessage"],
+      },
+    },
+    updateAttributesIfExists: true,
+  },
+  deletionConfig: {
+    deleteIfExists: false,
+  },
+})
+```
+
+### Default Behavior
+
+- If `policyConfig` is not provided or set to `undefined`, no policy is applied to the queue
+- If `statements` is not provided, default values are used:
+  - `Effect`: "Allow"
+  - `Principal`: "*" (allows all AWS principals)
+  - `Action`: ["sqs:SendMessage", "sqs:GetQueueAttributes", "sqs:GetQueueUrl"]
+
+### Policy Structure
+
+The generated policy follows the AWS IAM policy format:
+
+```json
+{
+  "Version": "2012-10-17",
+  "Resource": "arn:aws:sqs:*:*:*",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": {"AWS": "arn:aws:iam::123456789012:user/my-user"},
+      "Action": ["sqs:SendMessage", "sqs:ReceiveMessage"]
+    }
+  ]
+}
+```
+
+> **_NOTE:_**  See [SqsPermissionConsumer.spec.ts](./packages/sqs/test/consumers/SqsPermissionConsumer.spec.ts) and [SqsPermissionPublisher.spec.ts](./packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts) for practical examples of policy configuration tests.
+
 ## Handler Spies
 
 In certain cases you want to await until certain publisher publishes a message, or a certain handler consumes a message. For that you can use handler spy functionality, built into `message-queue-toolkit` directly.

packages/sns/lib/sns/AbstractSnsSqsConsumer.ts

@@ -17,7 +17,7 @@ export type SNSSQSConsumerDependencies = SQSConsumerDependencies & {
   snsClient: SNSClient
   stsClient: STSClient
 }
-export type SNSSQSCreationConfig = SQSCreationConfig & SNSCreationConfig
+export type SNSSQSCreationConfig = Omit<SQSCreationConfig, 'policyConfig'> & SNSCreationConfig
 
 export type SNSSQSQueueLocatorType = Partial<SQSQueueLocatorType> &
   SNSTopicLocatorType & {

packages/sns/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@message-queue-toolkit/sns",
-    "version": "23.1.0",
+    "version": "23.1.2",
     "private": false,
     "license": "MIT",
     "description": "SNS adapter for message-queue-toolkit",

packages/sqs/lib/index.ts

@@ -8,13 +8,16 @@ export {
   AbstractSqsPublisher,
   OFFLOADED_PAYLOAD_SIZE_ATTRIBUTE,
 } from './sqs/AbstractSqsPublisher.ts'
-export type {
-  ExtraSQSCreationParams,
-  SQSCreationConfig,
-  SQSDependencies,
-  SQSQueueLocatorType,
+export {
+  type ExtraSQSCreationParams,
+  SQS_MESSAGE_MAX_SIZE,
+  SQS_RESOURCE_ANY,
+  SQS_RESOURCE_CURRENT_QUEUE,
+  type SQSCreationConfig,
+  type SQSDependencies,
+  type SQSPolicyConfig,
+  type SQSQueueLocatorType,
 } from './sqs/AbstractSqsService.ts'
-export { SQS_MESSAGE_MAX_SIZE } from './sqs/AbstractSqsService.ts'
 export type { CommonMessage, SQSMessage } from './types/MessageTypes.ts'
 export { resolveOutgoingMessageAttributes } from './utils/messageUtils.ts'
 export {

packages/sqs/lib/sqs/AbstractSqsConsumer.ts

@@ -590,6 +590,6 @@ export abstract class AbstractSqsConsumer<
     }
 
     // parseInt is safe because if the value is not a number process should have failed on init
-    return visibilityTimeoutString ? Number.parseInt(visibilityTimeoutString) : undefined
+    return visibilityTimeoutString ? Number.parseInt(visibilityTimeoutString, 10) : undefined
   }
 }

packages/sqs/lib/sqs/AbstractSqsPublisher.ts

@@ -77,7 +77,6 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
         this.logMessage(resolvedLogMessage)
       }
 
-      // biome-ignore lint/style/noParameterAssign: This is expected
       message = this.updateInternalProperties(message)
       const maybeOffloadedPayloadMessage = await this.offloadMessagePayloadIfNeeded(message, () =>
         calculateOutgoingMessageSize(message),

packages/sqs/lib/sqs/AbstractSqsService.ts

@@ -6,6 +6,8 @@ import { deleteSqs, initSqs } from '../utils/sqsInitter.ts'
 
 // https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/quotas-messages.html
 export const SQS_MESSAGE_MAX_SIZE = 256 * 1024 // 256KB
+export const SQS_RESOURCE_ANY = Symbol('any')
+export const SQS_RESOURCE_CURRENT_QUEUE = Symbol('current_queue')
 
 export type SQSDependencies = QueueDependencies & {
   sqsClient: SQSClient
@@ -15,12 +17,25 @@ export type ExtraSQSCreationParams = {
   topicArnsWithPublishPermissionsPrefix?: string
   updateAttributesIfExists?: boolean
   forceTagUpdate?: boolean
+  policyConfig?: SQSPolicyConfig
+}
+
+type SQSPolicyStatement = {
+  Effect?: string
+  Principal?: string
+  Action?: string[]
+}
+
+export type SQSPolicyConfig = {
+  resource: string | typeof SQS_RESOURCE_ANY | typeof SQS_RESOURCE_CURRENT_QUEUE
+  statements?: SQSPolicyStatement | SQSPolicyStatement[]
 }
 
 export type SQSCreationConfig = {
   queue: CreateQueueRequest
   updateAttributesIfExists?: boolean
   forceTagUpdate?: boolean
+  policyConfig?: SQSPolicyConfig
 } & ExtraSQSCreationParams
 
 export type SQSQueueLocatorType =