AWS Transfer Family Custom Identity Provider Lambda — ServerResponse Schema and SSH Key Auth

Creating a server with Lambda identity provider

Transfer Family added native Lambda support (AWS_LAMBDA identity provider type) as a simpler alternative to the legacy API_GATEWAY type that required an API Gateway proxy in front of the Lambda. With AWS_LAMBDA, Transfer Family invokes the Lambda directly. The Lambda function ARN is the only configuration needed — no API Gateway setup required.

# Create Transfer Family server with Lambda identity provider
aws transfer create-server \
  --protocols SFTP \
  --endpoint-type PUBLIC \
  --identity-provider-type AWS_LAMBDA \
  --identity-provider-details '{
    "Function": "arn:aws:lambda:us-east-1:123456789012:function:sftp-auth"
  }' \
  --logging-role arn:aws:iam::123456789012:role/TransferLoggingRole

# Grant Transfer Family permission to invoke the Lambda
aws lambda add-permission \
  --function-name sftp-auth \
  --statement-id transfer-family-invoke \
  --action lambda:InvokeFunction \
  --principal transfer.amazonaws.com \
  --source-arn arn:aws:transfer:us-east-1:123456789012:server/s-0abc

# Test the authentication Lambda directly
aws transfer test-identity-provider \
  --server-id s-0abc \
  --user-name alice \
  --user-password secretpassword123 \
  --source-ip 203.0.113.1
# Returns: { "Response": "{...ServerResponse JSON...}", "StatusCode": 200, "Url": "" }

The test-identity-provider API call is invaluable for debugging — it shows the exact Lambda response that Transfer Family received, the HTTP status code, and whether the response was treated as success or failure. Use it to verify your Lambda returns the correct ServerResponse schema before testing with a real SFTP client. Note that test-identity-provider does not create a real SFTP session — it only exercises the Lambda invocation path.

Lambda event input structure

Transfer Family invokes the Lambda with a JSON event containing the authentication context. The event structure differs slightly between password authentication and SSH key authentication. For SSH key auth, Transfer Family calls the Lambda first with the username (no password) to retrieve the user's authorized public keys, then verifies the client's private key against those public keys locally — the Lambda never sees the private key.

# Lambda event for password authentication
{
  "username": "alice",
  "password": "secretpassword123",
  "protocol": "SFTP",          # SFTP | FTPS | FTP
  "serverId": "s-0abc123",
  "sourceIp": "203.0.113.1"   # client IP address
}

# Lambda event for SSH public key authentication
# Password field is absent or empty string
{
  "username": "alice",
  "password": "",
  "protocol": "SFTP",
  "serverId": "s-0abc123",
  "sourceIp": "203.0.113.1"
}

# Python handler — detecting auth type
def handler(event, context):
    username = event['username']
    password = event.get('password', '')
    source_ip = event.get('sourceIp', '')
    server_id = event['serverId']

    # Determine authentication type
    is_key_auth = not password

    if is_key_auth:
        # Return user config including PublicKeys
        # Transfer Family will verify the client key against these
        return get_user_config_with_keys(username, source_ip)
    else:
        # Verify password against your backend
        if verify_password(username, password):
            return get_user_config(username, source_ip)
        else:
            return {}  # Empty dict = authentication denied

For SSH key authentication, the Lambda is called once at connection time with an empty password field. The Lambda must return the user's authorized public keys in the PublicKeys array — Transfer Family then challenges the client with the SSH key exchange and verifies the client's signature against one of the returned public keys. This means your Lambda needs to look up the user's SSH public keys from a database or Secrets Manager on every connection — cache aggressively to reduce latency.

ServerResponse schema

The ServerResponse is a JSON object that tells Transfer Family the authenticated user's permissions and home directory. All fields are optional except Role — but in practice you always need Role plus either HomeDirectory (for ABSOLUTE mode) or HomeDirectoryDetails + HomeDirectoryType: "LOGICAL". An empty object {} denies authentication. A 200 response with any non-empty Role allows authentication.

# Full ServerResponse schema (Python dict, serialized to JSON string)
def get_user_config(username, source_ip):
    return {
        # Required: IAM role ARN the user assumes for S3 access
        "Role": "arn:aws:iam::123456789012:role/TransferUserRole",

        # Optional: session policy (JSON string, not dict)
        # Restricts the role's effective permissions for this session
        "Policy": json.dumps({
            "Version": "2012-10-17",
            "Statement": [{
                "Effect": "Allow",
                "Action": ["s3:*"],
                "Resource": [
                    "arn:aws:s3:::mcp-file-intake",
                    f"arn:aws:s3:::mcp-file-intake/{username}/*"
                ]
            }]
        }),

        # LOGICAL home directory: user sees "/" as their root
        "HomeDirectoryType": "LOGICAL",
        "HomeDirectoryDetails": json.dumps([
            {
                "Entry": "/",
                "Target": f"/mcp-file-intake/{username}"
            },
            {
                "Entry": "/shared",
                "Target": "/mcp-file-intake/shared-read"
            }
        ]),

        # Optional: SSH public keys (for key-based auth lookup)
        # Each entry is a full public key string including type and comment
        "PublicKeys": [
            "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC... alice@laptop",
            "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI... alice@workstation"
        ]
    }

# Note: HomeDirectoryDetails must be a JSON STRING, not a Python list
# Transfer Family parses it as a nested JSON string, not a direct JSON field

The most common mistake with ServerResponse is passing HomeDirectoryDetails as a Python list or dict instead of a JSON-serialized string. Transfer Family expects HomeDirectoryDetails to be a string value containing JSON — not a nested JSON object. Similarly, Policy must be a string, not a dict. Always call json.dumps() on both fields before returning them in the response.

Looking up user credentials from Secrets Manager

Storing per-user SFTP credentials in Secrets Manager is the recommended pattern for managed identity providers — each user has a secret at a predictable path like SFTP/alice, containing their hashed password and SSH public keys. The Lambda looks up the secret, verifies the password, and returns the user's configuration. This approach avoids a separate database and integrates with AWS Secrets Manager rotation for password cycling.

import boto3
import json
import hashlib
import hmac
import os

sm = boto3.client('secretsmanager')

def handler(event, context):
    username = event['username']
    password = event.get('password', '')
    server_id = event['serverId']

    # Look up user secret: "SFTP/{username}"
    secret_path = f"SFTP/{username}"
    try:
        secret = sm.get_secret_value(SecretId=secret_path)
        user_config = json.loads(secret['SecretString'])
    except sm.exceptions.ResourceNotFoundException:
        print(f"User not found: {username}")
        return {}  # Deny

    is_key_auth = not password

    if is_key_auth:
        # Return config with PublicKeys for SSH key verification
        return build_response(username, user_config)
    else:
        # Verify password using PBKDF2 hash stored in secret
        stored_hash = user_config.get('password_hash', '')
        salt = user_config.get('salt', '').encode()
        input_hash = hashlib.pbkdf2_hmac(
            'sha256', password.encode(), salt, 100000
        ).hex()

        if not hmac.compare_digest(input_hash, stored_hash):
            print(f"Password mismatch for user: {username}")
            return {}  # Deny

        return build_response(username, user_config)

def build_response(username, user_config):
    return {
        "Role": user_config['iam_role'],
        "Policy": json.dumps({
            "Version": "2012-10-17",
            "Statement": [{"Effect": "Allow", "Action": ["s3:*"],
              "Resource": [f"arn:aws:s3:::mcp-file-intake",
                           f"arn:aws:s3:::mcp-file-intake/{username}/*"]}]
        }),
        "HomeDirectoryType": "LOGICAL",
        "HomeDirectoryDetails": json.dumps([
            {"Entry": "/", "Target": f"/mcp-file-intake/{username}"}
        ]),
        "PublicKeys": user_config.get('ssh_public_keys', [])
    }

Secrets Manager adds ~50–100ms of latency per Lambda invocation. Use Lambda environment caching — store the secrets client outside the handler function to reuse the connection across warm invocations. For high-volume SFTP endpoints, consider adding a DynamoDB table as a first-level cache with a 1-hour TTL, falling back to Secrets Manager on miss. Secrets Manager GetSecretValue has a quota of 10,000 requests per second per region — unlikely to be a bottleneck for SFTP authentication traffic.

IP-based access control at authentication time

The sourceIp field in the Lambda event is the client's IP address. You can use it to enforce IP-based allow/deny lists at authentication time — returning an empty object for blocked IPs. This is an application-level control that complements security group rules. Use it when different users should be allowed from different IP ranges, or when you want to log blocked connection attempts with user context before denying.

import ipaddress

# Per-user IP allowlist stored in Secrets Manager user config
def is_ip_allowed(source_ip, allowed_cidrs):
    """Return True if source_ip is in any of the allowed CIDR ranges."""
    try:
        client_ip = ipaddress.ip_address(source_ip)
        return any(
            client_ip in ipaddress.ip_network(cidr, strict=False)
            for cidr in allowed_cidrs
        )
    except ValueError:
        return False  # Invalid IP format — deny

def handler(event, context):
    username = event['username']
    source_ip = event.get('sourceIp', '')

    user_config = get_user_config_from_secrets(username)
    if not user_config:
        return {}

    # Check IP allowlist if configured for this user
    allowed_cidrs = user_config.get('allowed_cidrs', [])
    if allowed_cidrs and not is_ip_allowed(source_ip, allowed_cidrs):
        print(f"IP {source_ip} not in allowlist for user {username}")
        return {}  # Deny

    # Proceed with password/key verification...
    return build_response(username, user_config)

IP-based controls in the Lambda are not a substitute for VPC security groups — a security group blocks the TCP connection before the Lambda is invoked, while the Lambda check happens after the TCP handshake. Use security groups for coarse network-level blocking (entire CIDR ranges) and the Lambda check for fine-grained per-user IP policies. Both controls complement each other without overlap.

Failure modes reference