feat(event_source): add AWS Transfer Family classes (aws-powertools#5912)

* Adding TransferFamilyAuthorizer and TransferFamilyAuthorizerResponse class

* Modify tests

* Modify tests

* Modify tests

* Merging from develop

Author: leandrodamascena

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -45,6 +45,7 @@
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
 from .sqs_event import SQSEvent
+from .transfer_family_event import TransferFamilyAuthorizer, TransferFamilyAuthorizerResponse
 from .vpc_lattice import VPCLatticeEvent, VPCLatticeEventV2
 
 __all__ = [
@@ -87,4 +88,6 @@
     "VPCLatticeEvent",
     "VPCLatticeEventV2",
     "CloudFormationCustomResourceEvent",
+    "TransferFamilyAuthorizerResponse",
+    "TransferFamilyAuthorizer",
 ]

aws_lambda_powertools/utilities/data_classes/transfer_family_event.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from aws_lambda_powertools.utilities.data_classes.common import (
+    DictWrapper,
+)
+
+
+class TransferFamilyAuthorizer(DictWrapper):
+    @property
+    def username(self) -> str:
+        """The username used for authentication"""
+        return self["username"]
+
+    @property
+    def password(self) -> str | None:
+        """
+        The password used for authentication.
+        None in case customer authenticating with certificates
+        """
+        return self["password"]
+
+    @property
+    def protocol(self) -> str:
+        """The protocol can be SFTP, FTP or FTPS"""
+        return self["protocol"]
+
+    @property
+    def server_id(self) -> str:
+        """The AWS Transfer Family ServerID"""
+        return self["serverId"]
+
+    @property
+    def source_ip(self) -> str:
+        """The customer IP used for connection"""
+        return self["sourceIp"]
+
+
+class TransferFamilyAuthorizerResponse:
+
+    def _build_authentication_response(
+        self,
+        role_arn: str,
+        policy: str | None = None,
+        home_directory: str | None = None,
+        home_directory_details: dict | None = None,
+        home_directory_type: Literal["LOGICAL", "PATH"] = "PATH",
+        user_gid: int | None = None,
+        user_uid: int | None = None,
+        public_keys: str | None = None,
+    ) -> dict[str, Any]:
+
+        response: dict[str, Any] = {}
+
+        if home_directory_type == "PATH":
+            if not home_directory:
+                raise ValueError("home_directory must be set when home_directory_type is PATH")
+
+            response["HomeDirectory"] = home_directory
+        elif home_directory_type == "LOGICAL":
+            if not home_directory_details:
+                raise ValueError("home_directory_details must be set when home_directory_type is LOGICAL")
+
+            response["HomeDirectoryDetails"] = [home_directory_details]
+
+        else:
+            raise ValueError(f"Invalid home_directory_type: {home_directory_type}")
+
+        if user_uid is not None:
+            response["PosixProfile"] = {"Gid": user_gid, "Uid": user_gid}
+
+        if policy:
+            response["Policy"] = policy
+
+        if public_keys:
+            response["PublicKeys"] = public_keys
+
+        response["Role"] = role_arn
+        response["HomeDirectoryType"] = home_directory_type
+
+        return response
+
+    def build_authentication_response_efs(
+        self,
+        role_arn: str,
+        user_gid: int,
+        user_uid: int,
+        policy: str | None = None,
+        home_directory: str | None = None,
+        home_directory_details: dict | None = None,
+        home_directory_type: Literal["LOGICAL", "PATH"] = "PATH",
+        public_keys: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Build an authentication response for AWS Transfer Family using EFS (Elastic File System).
+
+        Parameters:
+        -----------
+        role_arn : str
+            The Amazon Resource Name (ARN) of the IAM role.
+        user_gid : int
+            The group ID of the user.
+        user_uid : int
+            The user ID.
+        policy : str | None, optional
+            The IAM policy document. Defaults to None.
+        home_directory : str | None, optional
+            The home directory path. Required if home_directory_type is "PATH". Defaults to None.
+        home_directory_details : dict | None, optional
+            Details of the home directory. Required if home_directory_type is "LOGICAL". Defaults to None.
+        home_directory_type : Literal["LOGICAL", "PATH"], optional
+            The type of home directory. Must be either "LOGICAL" or "PATH". Defaults to "PATH".
+        public_keys : str | None, optional
+            The public keys associated with the user. Defaults to None.
+
+        Returns:
+        --------
+        dict[str, Any]
+            A dictionary containing the authentication response with various details such as
+            role ARN, policy, home directory information, and user details.
+
+        Raises:
+        -------
+        ValueError
+            If an invalid home_directory_type is provided or if required parameters are missing
+            for the specified home_directory_type.
+        """
+
+        return self._build_authentication_response(
+            role_arn=role_arn,
+            policy=policy,
+            home_directory=home_directory,
+            home_directory_details=home_directory_details,
+            home_directory_type=home_directory_type,
+            public_keys=public_keys,
+            user_gid=user_gid,
+            user_uid=user_uid,
+        )
+
+    def build_authentication_response_s3(
+        self,
+        role_arn: str,
+        policy: str | None = None,
+        home_directory: str | None = None,
+        home_directory_details: dict | None = None,
+        home_directory_type: Literal["LOGICAL", "PATH"] = "PATH",
+        public_keys: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Build an authentication response for Amazon S3.
+
+        This method constructs an authentication response tailored for S3 access,
+        likely by calling an internal method with the provided parameters.
+
+        Parameters:
+        -----------
+        role_arn : str
+            The Amazon Resource Name (ARN) of the IAM role for S3 access.
+        policy : str | None, optional
+            The IAM policy document for S3 access. Defaults to None.
+        home_directory : str | None, optional
+            The home directory path in S3. Required if home_directory_type is "PATH". Defaults to None.
+        home_directory_details : dict | None, optional
+            Details of the home directory in S3. Required if home_directory_type is "LOGICAL". Defaults to None.
+        home_directory_type : Literal["LOGICAL", "PATH"], optional
+            The type of home directory in S3. Must be either "LOGICAL" or "PATH". Defaults to "PATH".
+        public_keys : str | None, optional
+            The public keys associated with the user for S3 access. Defaults to None.
+
+        Returns:
+        --------
+        dict[str, Any]
+            A dictionary containing the authentication response with various details such as
+            role ARN, policy, home directory information, and potentially other S3-specific attributes.
+
+        Raises:
+        -------
+        ValueError
+            If an invalid home_directory_type is provided or if required parameters are missing
+            for the specified home_directory_type.
+        """
+        return self._build_authentication_response(
+            role_arn=role_arn,
+            policy=policy,
+            home_directory=home_directory,
+            home_directory_details=home_directory_details,
+            home_directory_type=home_directory_type,
+            public_keys=public_keys,
+        )

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -109,7 +109,7 @@
 )
 from .sns import SnsModel, SnsNotificationModel, SnsRecordModel
 from .sqs import SqsAttributesModel, SqsModel, SqsMsgAttributeModel, SqsRecordModel
-from .transfer_family import TransferFamily
+from .transfer_family import TransferFamilyAuthorizer
 from .vpc_lattice import VpcLatticeModel
 from .vpc_latticev2 import VpcLatticeV2Model
 
@@ -180,7 +180,7 @@
     "SqsAttributesModel",
     "S3SqsEventNotificationModel",
     "S3SqsEventNotificationRecordModel",
-    "TransferFamily",
+    "TransferFamilyAuthorizer",
     "APIGatewayProxyEventModel",
     "APIGatewayEventRequestContext",
     "APIGatewayEventAuthorizer",

aws_lambda_powertools/utilities/parser/models/transfer_family.py

@@ -4,7 +4,7 @@
 from pydantic.networks import IPvAnyAddress
 
 
-class TransferFamily(BaseModel):
+class TransferFamilyAuthorizer(BaseModel):
     username: str
     password: Optional[str] = None
     protocol: Literal["SFTP", "FTP", "FTPS"]

docs/utilities/data_classes.md

@@ -98,6 +98,8 @@ Each event source is linked to its corresponding GitHub file with the full set o
 | [SES](#ses) | `SESEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/ses_event.py) |
 | [SNS](#sns) | `SNSEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/sns_event.py) |
 | [SQS](#sqs) | `SQSEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/sqs_event.py) |
+| [TransferFamilyAuthorizer]                                                    | `TransferFamilyAuthorizer`                         | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/transfer_family_event.py) |
+| [TransferFamilyAuthorizerResponse]                                            | `TransferFamilyAuthorizerResponse`                 | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/transfer_family_event.py) |
 | [VPC Lattice V2](#vpc-lattice-v2) | `VPCLatticeV2Event` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/vpc_lattice.py) |
 | [VPC Lattice V1](#vpc-lattice-v1) | `VPCLatticeEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/vpc_lattice.py) |
 

docs/utilities/parser.md

@@ -132,7 +132,7 @@ The example above uses `SqsModel`. Other built-in models can be found below.
 | **SesModel**                                | Lambda Event Source payload for Amazon Simple Email Service                           |
 | **SnsModel**                                | Lambda Event Source payload for Amazon Simple Notification Service                    |
 | **SqsModel**                                | Lambda Event Source payload for Amazon SQS                                            |
-| **TransferFamily**                          | Lambda Event Source payload for AWS Transfer Family custom identity provider          |
+| **TransferFamilyAuthorizer**                | Lambda Event Source payload for AWS Transfer Family Lambda authorizer                 |
 | **VpcLatticeModel**                         | Lambda Event Source payload for Amazon VPC Lattice                                    |
 | **VpcLatticeV2Model**                       | Lambda Event Source payload for Amazon VPC Lattice v2 payload                         |
 

tests/events/TransferFamily.json

@@ -0,0 +1,7 @@
+{
+  "username": "value",
+  "password": "value",
+  "protocol": "SFTP",
+  "serverId": "s-abcd123456",
+  "sourceIp": "192.168.0.100"
+}