feat(aws): add support for Lambda Function URLs

configs/example.json

@@ -46,7 +46,11 @@
     "name": "aws",
     "aws": {
       "region": "us-east-1",
-      "lambda-role": ""
+      "lambda-role": "",
+      "resources": {
+        "use-function-url": false,
+        "function-url-auth-type": "NONE"
+      }
     },
     "azure": {
       "region": "westeurope"

sebs/aws/aws.py

@@ -639,6 +639,13 @@ def delete_function(self, func_name: str) -> None:
         except Exception:
             self.logging.error("Function {} does not exist!".format(func_name))
 
+    def delete_function_url(self, func_name: str) -> bool:
+        """
+        Delete the Function URL associated with a Lambda function.
+        Returns True if deleted successfully, False if it didn't exist.
+        """
+        return self.config.resources.delete_function_url(func_name, self.session)
+
     @staticmethod
     def parse_aws_report(
         log: str, requests: Union[ExecutionResult, Dict[str, ExecutionResult]]
@@ -874,28 +881,41 @@ def create_trigger(self, func: Function, trigger_type: Trigger.TriggerType) -> T
         Raises:
             RuntimeError: If trigger type is not supported
         """
-        from sebs.aws.triggers import HTTPTrigger
+        from sebs.aws.triggers import HTTPTrigger, FunctionURLTrigger
 
         function = cast(LambdaFunction, func)
 
         if trigger_type == Trigger.TriggerType.HTTP:
-            api_name = "{}-http-api".format(function.name)
-            http_api = self.config.resources.http_api(api_name, function, self.session)
-            # https://aws.amazon.com/blogs/compute/announcing-http-apis-for-amazon-api-gateway/
-            # but this is wrong - source arn must be {api-arn}/*/*
-            self.get_lambda_client().add_permission(
-                FunctionName=function.name,
-                StatementId=str(uuid.uuid1()),
-                Action="lambda:InvokeFunction",
-                Principal="apigateway.amazonaws.com",
-                SourceArn=f"{http_api.arn}/*/*",
-            )
-            trigger = HTTPTrigger(http_api.endpoint, api_name)
-            self.logging.info(
-                f"Created HTTP trigger for {func.name} function. "
-                "Sleep 5 seconds to avoid cloud errors."
-            )
-            time.sleep(5)
+
+            if self.config.resources.use_function_url:
+                # Use Lambda Function URL (no 29-second timeout limit)
+                func_url = self.config.resources.function_url(function, self.session)
+                trigger = FunctionURLTrigger(
+                    func_url.url, func_url.function_name, func_url.auth_type
+                )
+                self.logging.info(
+                    f"Created Function URL trigger for {func.name} function."
+                )
+            else:
+                # Use API Gateway (default, for backward compatibility)
+                api_name = "{}-http-api".format(function.name)
+                http_api = self.config.resources.http_api(api_name, function, self.session)
+                # https://aws.amazon.com/blogs/compute/announcing-http-apis-for-amazon-api-gateway/
+                # but this is wrong - source arn must be {api-arn}/*/*
+                self.get_lambda_client().add_permission(
+                    FunctionName=function.name,
+                    StatementId=str(uuid.uuid1()),
+                    Action="lambda:InvokeFunction",
+                    Principal="apigateway.amazonaws.com",
+                    SourceArn=f"{http_api.arn}/*/*",
+                )
+                trigger = HTTPTrigger(http_api.endpoint, api_name)
+                self.logging.info(
+                    f"Created HTTP API Gateway trigger for {func.name} function. "
+                    "Sleep 5 seconds to avoid cloud errors."
+                )
+                time.sleep(5)
+
             trigger.logging_handlers = self.logging_handlers
         elif trigger_type == Trigger.TriggerType.LIBRARY:
             # should already exist

sebs/aws/config.py

@@ -17,6 +17,7 @@
 import json
 import os
 import time
+from enum import Enum
 from typing import TYPE_CHECKING, cast, Dict, List, Optional, Tuple
 
 import boto3
@@ -30,6 +31,28 @@
 from sebs.utils import LoggingHandlers
 
 
+class FunctionURLAuthType(Enum):
+    """
+    Authentication types for AWS Lambda Function URLs.
+    - NONE: Public access, no authentication required
+    - AWS_IAM: Requires IAM authentication with SigV4 signing
+    """
+
+    NONE = "NONE"
+    AWS_IAM = "AWS_IAM"
+
+    @staticmethod
+    def from_string(value: str) -> "FunctionURLAuthType":
+        """Convert string to FunctionURLAuthType enum."""
+        try:
+            return FunctionURLAuthType(value)
+        except ValueError:
+            raise ValueError(
+                f"Invalid auth type '{value}'. Must be one of: "
+                f"{[e.value for e in FunctionURLAuthType]}"
+            )
+
+
 class AWSCredentials(Credentials):
     """AWS authentication credentials for SeBS.
 
@@ -264,6 +287,45 @@ def serialize(self) -> dict:
             out = {"arn": self.arn, "endpoint": self.endpoint}
             return out
 
+    class FunctionURL:
+        def __init__(
+            self,
+            url: str,
+            function_name: str,
+            auth_type: FunctionURLAuthType = FunctionURLAuthType.NONE,
+        ):
+            self._url = url
+            self._function_name = function_name
+            self._auth_type = auth_type
+
+        @property
+        def url(self) -> str:
+            return self._url
+
+        @property
+        def function_name(self) -> str:
+            return self._function_name
+
+        @property
+        def auth_type(self) -> FunctionURLAuthType:
+            return self._auth_type
+
+        @staticmethod
+        def deserialize(dct: dict) -> "AWSResources.FunctionURL":
+            auth_type_str = dct.get("auth_type", "NONE")
+            return AWSResources.FunctionURL(
+                dct["url"],
+                dct["function_name"],
+                FunctionURLAuthType.from_string(auth_type_str),
+            )
+
+        def serialize(self) -> dict:
+            return {
+                "url": self.url,
+                "function_name": self.function_name,
+                "auth_type": self.auth_type.value,
+            }
+
     def __init__(
         self,
         registry: Optional[str] = None,
@@ -284,6 +346,9 @@ def __init__(
         self._container_repository: Optional[str] = None
         self._lambda_role = ""
         self._http_apis: Dict[str, AWSResources.HTTPApi] = {}
+        self._function_urls: Dict[str, AWSResources.FunctionURL] = {}
+        self._use_function_url: bool = False
+        self._function_url_auth_type: FunctionURLAuthType = FunctionURLAuthType.NONE
 
     @staticmethod
     def typename() -> str:
@@ -330,6 +395,26 @@ def container_repository(self) -> Optional[str]:
         """
         return self._container_repository
 
+    @property
+    def use_function_url(self) -> bool:
+        return self._use_function_url
+
+    @use_function_url.setter
+    def use_function_url(self, value: bool):
+        self._use_function_url = value
+
+    @property
+    def function_url_auth_type(self) -> FunctionURLAuthType:
+        return self._function_url_auth_type
+
+    @function_url_auth_type.setter
+    def function_url_auth_type(self, value: FunctionURLAuthType):
+        if not isinstance(value, FunctionURLAuthType):
+            raise TypeError(
+                f"function_url_auth_type must be a FunctionURLAuthType enum, got {type(value)}"
+            )
+        self._function_url_auth_type = value
+
     def lambda_role(self, boto3_session: boto3.session.Session) -> str:
         """Get or create IAM role for Lambda execution.
 
@@ -495,6 +580,139 @@ def cleanup_http_apis(
 
         return deleted
 
+    def function_url(
+        self, func: LambdaFunction, boto3_session: boto3.session.Session
+    ) -> "AWSResources.FunctionURL":
+        """
+        Create or retrieve a Lambda Function URL for the given function.
+        Function URLs provide a simpler alternative to API Gateway without the
+        29-second timeout limit.
+        """
+        cached_url = self._function_urls.get(func.name)
+        if cached_url:
+            self.logging.info(f"Using cached Function URL for {func.name}")
+            return cached_url
+
+        # Check for unsupported auth type before attempting to create
+        if self._function_url_auth_type == FunctionURLAuthType.AWS_IAM:
+            raise NotImplementedError(
+                "AWS_IAM authentication for Function URLs is not yet supported. "
+                "SigV4 request signing is required for AWS_IAM auth type. "
+                "Please use auth_type='NONE' or implement SigV4 signing."
+            )
+
+        lambda_client = boto3_session.client(
+            service_name="lambda", region_name=cast(str, self._region)
+        )
+
+        try:
+            response = lambda_client.get_function_url_config(FunctionName=func.name)
+            self.logging.info(f"Using existing Function URL for {func.name}")
+            url = response["FunctionUrl"]
+            auth_type = FunctionURLAuthType.from_string(response["AuthType"])
+        except lambda_client.exceptions.ResourceNotFoundException:
+            self.logging.info(f"Creating Function URL for {func.name}")
+
+            auth_type = self._function_url_auth_type
+
+            if auth_type == FunctionURLAuthType.NONE:
+                self.logging.warning(
+                    f"Creating Function URL with auth_type=NONE for {func.name}. "
+                    "WARNING: This function will have unrestricted public access. "
+                    "Anyone with the URL can invoke this function."
+                )
+                try:
+                    lambda_client.add_permission(
+                        FunctionName=func.name,
+                        StatementId="FunctionURLAllowPublicAccess",
+                        Action="lambda:InvokeFunctionUrl",
+                        Principal="*",
+                        FunctionUrlAuthType="NONE",
+                    )
+                except lambda_client.exceptions.ResourceConflictException:
+                    # Permission with this StatementId already exists on the function.
+                    # This can happen if the function was previously configured with
+                    # a Function URL that was deleted but the permission remained,
+                    # or if there's a concurrent creation attempt. Safe to ignore.
+                    pass
+
+            retries = 0
+            while retries < 5:
+                try:
+                    response = lambda_client.create_function_url_config(
+                        FunctionName=func.name,
+                        AuthType=auth_type.value,
+                    )
+                    break
+                except lambda_client.exceptions.ResourceConflictException:
+                    # Function URL already exists - can happen if a concurrent process
+                    # created it between our check and create, or if there was a race
+                    # condition. Retrieve the existing configuration instead.
+                    response = lambda_client.get_function_url_config(
+                        FunctionName=func.name
+                    )
+                    break
+                except lambda_client.exceptions.TooManyRequestsException as e:
+                    # AWS is throttling requests - apply exponential backoff
+                    retries += 1
+                    if retries == 5:
+                        self.logging.error("Failed to create Function URL after 5 retries!")
+                        self.logging.exception(e)
+                        raise RuntimeError("Failed to create Function URL!") from e
+                    else:
+                        backoff_seconds = retries
+                        self.logging.info(
+                            f"Function URL creation rate limited, "
+                            f"retrying in {backoff_seconds}s (attempt {retries}/5)..."
+                        )
+                        time.sleep(backoff_seconds)
+
+            url = response["FunctionUrl"]
+
+        function_url_obj = AWSResources.FunctionURL(url, func.name, auth_type)
+        self._function_urls[func.name] = function_url_obj
+        return function_url_obj
+
+    def delete_function_url(
+        self, function_name: str, boto3_session: boto3.session.Session
+    ) -> bool:
+        """
+        Delete a Lambda Function URL for the given function.
+        Returns True if deleted successfully, False if it didn't exist.
+        """
+        lambda_client = boto3_session.client(
+            service_name="lambda", region_name=cast(str, self._region)
+        )
+
+        # Check if we have cached info about the auth type
+        cached_url = self._function_urls.get(function_name)
+        cached_auth_type = cached_url.auth_type if cached_url else None
+
+        try:
+            lambda_client.delete_function_url_config(FunctionName=function_name)
+            self.logging.info(f"Deleted Function URL for {function_name}")
+
+            # Only remove the public access permission if auth_type was NONE
+            # (AWS_IAM auth type doesn't create this permission)
+            if cached_auth_type is None or cached_auth_type == FunctionURLAuthType.NONE:
+                try:
+                    lambda_client.remove_permission(
+                        FunctionName=function_name,
+                        StatementId="FunctionURLAllowPublicAccess",
+                    )
+                except lambda_client.exceptions.ResourceNotFoundException:
+                    # Permission doesn't exist - either it was already removed,
+                    # or the function was using AWS_IAM auth type
+                    pass
+        except lambda_client.exceptions.ResourceNotFoundException:
+            self.logging.info(f"No Function URL found for {function_name}")
+            return False
+        else:
+            # Only runs if no exception was raised - cleanup cache
+            if function_name in self._function_urls:
+                del self._function_urls[function_name]
+            return True
+
     def check_ecr_repository_exists(
         self, ecr_client: ECRClient, repository_name: str
     ) -> Optional[str]:
@@ -715,6 +933,14 @@ def initialize(res: Resources, dct: dict) -> None:
             for key, value in dct["http-apis"].items():
                 ret._http_apis[key] = AWSResources.HTTPApi.deserialize(value)
 
+        if "function-urls" in dct:
+            for key, value in dct["function-urls"].items():
+                ret._function_urls[key] = AWSResources.FunctionURL.deserialize(value)
+
+        ret._use_function_url = dct.get("use-function-url", False)
+        auth_type_str = dct.get("function-url-auth-type", "NONE")
+        ret.function_url_auth_type = FunctionURLAuthType.from_string(auth_type_str)
+
     def serialize(self) -> dict:
         """Serialize AWS resources to dictionary.
 
@@ -725,6 +951,11 @@ def serialize(self) -> dict:
             **super().serialize(),
             "lambda-role": self._lambda_role,
             "http-apis": {key: value.serialize() for (key, value) in self._http_apis.items()},
+            "function-urls": {
+                key: value.serialize() for (key, value) in self._function_urls.items()
+            },
+            "use-function-url": self._use_function_url,
+            "function-url-auth-type": self._function_url_auth_type.value,
             "docker": {
                 "registry": self.docker_registry,
                 "username": self.docker_username,
@@ -753,6 +984,17 @@ def update_cache(self, cache: Cache) -> None:
         cache.update_config(val=self._lambda_role, keys=["aws", "resources", "lambda-role"])
         for name, api in self._http_apis.items():
             cache.update_config(val=api.serialize(), keys=["aws", "resources", "http-apis", name])
+        for name, func_url in self._function_urls.items():
+            cache.update_config(
+                val=func_url.serialize(), keys=["aws", "resources", "function-urls", name]
+            )
+        cache.update_config(
+            val=self._use_function_url, keys=["aws", "resources", "use-function-url"]
+        )
+        cache.update_config(
+            val=self._function_url_auth_type.value,
+            keys=["aws", "resources", "function-url-auth-type"],
+        )
 
     @staticmethod
     def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Resources: