bug: CFN Resources go unchecked

[m3newc]

What is the problem?

Some internal CDK Features use CfnResource instead of the normal L1 Constructs.
(see https://github.com/aws/aws-cdk/blob/4ff3565a9d7b0298bf884822fecabdd3cff643aa/packages/aws-cdk-lib/aws-s3/lib/notifications-resource/notifications-resource-handler.ts#L26 as an example).

This causes some type checks in CDK Nag to fail, and resources go unchecked.

Reproduction Steps

from aws_cdk import Aspects, Stack
from aws_cdk import aws_lambda, aws_s3, aws_s3_notifications;
from cdk_nag import AwsSolutionsChecks, NagSuppressions, NIST80053R5Checks
from constructs import Construct

class TestStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # L2 Case
        L2_lambda = aws_lambda.Function(
            self, "L2Lambda", environment = {'Something': "oof"},
            code = aws_lambda.InlineCode('oof'),
            handler = 'handler',
            runtime = aws_lambda.Runtime.PYTHON_3_9
        )

        # Check for CDk Custom Resource Built-ins, such as that for S3 Bucket Notifications
        bucket = aws_s3.Bucket.from_bucket_name(self, "ImportedBucket", "hello-bucket");
        bucket.add_event_notification(aws_s3.EventType.OBJECT_CREATED_PUT, aws_s3_notifications.LambdaDestination(L2_lambda));
        
        #Aspects
        aspects = Aspects.of(self);
        aspects.add(AwsSolutionsChecks());
        aspects.add(NIST80053R5Checks());

import os, aws_cdk;
app = aws_cdk.App()
env = aws_cdk.Environment();
TestStack(app, "TestStack", env = env, synthesizer=aws_cdk.DefaultStackSynthesizer(generate_bootstrap_version_rule=False));
app.synth()

{
 "Resources": {
  "L2LambdaServiceRole1E9E5A2E": {
   "Type": "AWS::IAM::Role",
   "Properties": {
    "AssumeRolePolicyDocument": {
     "Statement": [
      {
       "Action": "sts:AssumeRole",
       "Effect": "Allow",
       "Principal": {
        "Service": "lambda.amazonaws.com"
       }
      }
     ],
     "Version": "2012-10-17"
    },
    "ManagedPolicyArns": [
     {
      "Fn::Join": [
       "",
       [
        "arn:",
        {
         "Ref": "AWS::Partition"
        },
        ":iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
       ]
      ]
     }
    ]
   }
  },
  "L2LambdaC3DD76DF": {
   "Type": "AWS::Lambda::Function",
   "Properties": {
    "Code": {
     "ZipFile": "oof"
    },
    "Environment": {
     "Variables": {
      "Something": "oof"
     }
    },
    "Handler": "handler",
    "Role": {
     "Fn::GetAtt": [
      "L2LambdaServiceRole1E9E5A2E",
      "Arn"
     ]
    },
    "Runtime": "python3.9"
   },
   "DependsOn": [
    "L2LambdaServiceRole1E9E5A2E"
   ]
  },
  "ImportedBucketNotifications9C01C2F8": {
   "Type": "Custom::S3BucketNotifications",
   "Properties": {
    "ServiceToken": {
     "Fn::GetAtt": [
      "BucketNotificationsHandler050a0587b7544547bf325f094a3db8347ECC3691",
      "Arn"
     ]
    },
    "BucketName": "hello-bucket",
    "NotificationConfiguration": {
     "LambdaFunctionConfigurations": [
      {
       "Events": [
        "s3:ObjectCreated:Put"
       ],
       "LambdaFunctionArn": {
        "Fn::GetAtt": [
         "L2LambdaC3DD76DF",
         "Arn"
        ]
       }
      }
     ]
    },
    "Managed": false
   },
   "DependsOn": [
    "ImportedBucketAllowBucketNotificationsToTestStackL2LambdaDE99D47484A7D298"
   ]
  },
  "ImportedBucketAllowBucketNotificationsToTestStackL2LambdaDE99D47484A7D298": {
   "Type": "AWS::Lambda::Permission",
   "Properties": {
    "Action": "lambda:InvokeFunction",
    "FunctionName": {
     "Fn::GetAtt": [
      "L2LambdaC3DD76DF",
      "Arn"
     ]
    },
    "Principal": "s3.amazonaws.com",
    "SourceAccount": {
     "Ref": "AWS::AccountId"
    },
    "SourceArn": {
     "Fn::Join": [
      "",
      [
       "arn:",
       {
        "Ref": "AWS::Partition"
       },
       ":s3:::hello-bucket"
      ]
     ]
    }
   }
  },
  "BucketNotificationsHandler050a0587b7544547bf325f094a3db834RoleB6FB88EC": {
   "Type": "AWS::IAM::Role",
   "Properties": {
    "AssumeRolePolicyDocument": {
     "Statement": [
      {
       "Action": "sts:AssumeRole",
       "Effect": "Allow",
       "Principal": {
        "Service": "lambda.amazonaws.com"
       }
      }
     ],
     "Version": "2012-10-17"
    },
    "ManagedPolicyArns": [
     {
      "Fn::Join": [
       "",
       [
        "arn:",
        {
         "Ref": "AWS::Partition"
        },
        ":iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
       ]
      ]
     }
    ]
   }
  },
  "BucketNotificationsHandler050a0587b7544547bf325f094a3db834RoleDefaultPolicy2CF63D36": {
   "Type": "AWS::IAM::Policy",
   "Properties": {
    "PolicyDocument": {
     "Statement": [
      {
       "Action": [
        "s3:GetBucketNotification",
        "s3:PutBucketNotification"
       ],
       "Effect": "Allow",
       "Resource": "*"
      }
     ],
     "Version": "2012-10-17"
    },
    "PolicyName": "BucketNotificationsHandler050a0587b7544547bf325f094a3db834RoleDefaultPolicy2CF63D36",
    "Roles": [
     {
      "Ref": "BucketNotificationsHandler050a0587b7544547bf325f094a3db834RoleB6FB88EC"
     }
    ]
   }
  },
  "BucketNotificationsHandler050a0587b7544547bf325f094a3db8347ECC3691": {
   "Type": "AWS::Lambda::Function",
   "Properties": {
    "Description": "AWS CloudFormation handler for \"Custom::S3BucketNotifications\" resources (@aws-cdk/aws-s3)",
    "Code": {
     "ZipFile": "import boto3  # type: ignore\nimport json\nimport logging\nimport urllib.request\n\ns3 = boto3.client(\"s3\")\n\nEVENTBRIDGE_CONFIGURATION = 'EventBridgeConfiguration'\n\nCONFIGURATION_TYPES = [\"TopicConfigurations\", \"QueueConfigurations\", \"LambdaFunctionConfigurations\"]\n\ndef handler(event: dict, context):\n  response_status = \"SUCCESS\"\n  error_message = \"\"\n  try:\n    props = event[\"ResourceProperties\"]\n    bucket = props[\"BucketName\"]\n    notification_configuration = props[\"NotificationConfiguration\"]\n    request_type = event[\"RequestType\"]\n    managed = props.get('Managed', 'true').lower() == 'true'\n    stack_id = event['StackId']\n\n    if managed:\n      config = handle_managed(request_type, notification_configuration)\n    else:\n      config = handle_unmanaged(bucket, stack_id, request_type, notification_configuration)\n\n    put_bucket_notification_configuration(bucket, config)\n  except Exception as e:\n    logging.exception(\"Failed to put bucket notification configuration\")\n    response_status = \"FAILED\"\n    error_message = f\"Error: {str(e)}. \"\n  finally:\n    submit_response(event, context, response_status, error_message)\n\ndef handle_managed(request_type, notification_configuration):\n  if request_type == 'Delete':\n    return {}\n  return notification_configuration\n\ndef handle_unmanaged(bucket, stack_id, request_type, notification_configuration):\n  external_notifications = find_external_notifications(bucket, stack_id)\n\n  if request_type == 'Delete':\n    return external_notifications\n\n  def with_id(notification):\n    notification['Id'] = f\"{stack_id}-{hash(json.dumps(notification, sort_keys=True))}\"\n    return notification\n\n  notifications = {}\n  for t in CONFIGURATION_TYPES:\n    external = external_notifications.get(t, [])\n    incoming = [with_id(n) for n in notification_configuration.get(t, [])]\n    notifications[t] = external + incoming\n\n  if EVENTBRIDGE_CONFIGURATION in notification_configuration:\n    notifications[EVENTBRIDGE_CONFIGURATION] = notification_configuration[EVENTBRIDGE_CONFIGURATION]\n  elif EVENTBRIDGE_CONFIGURATION in external_notifications:\n    notifications[EVENTBRIDGE_CONFIGURATION] = external_notifications[EVENTBRIDGE_CONFIGURATION]\n\n  return notifications\n\ndef find_external_notifications(bucket, stack_id):\n  existing_notifications = get_bucket_notification_configuration(bucket)\n  external_notifications = {}\n  for t in CONFIGURATION_TYPES:\n    external_notifications[t] = [n for n in existing_notifications.get(t, []) if not n['Id'].startswith(f\"{stack_id}-\")]\n\n  if EVENTBRIDGE_CONFIGURATION in existing_notifications:\n    external_notifications[EVENTBRIDGE_CONFIGURATION] = existing_notifications[EVENTBRIDGE_CONFIGURATION]\n\n  return external_notifications\n\ndef get_bucket_notification_configuration(bucket):\n  return s3.get_bucket_notification_configuration(Bucket=bucket)\n\ndef put_bucket_notification_configuration(bucket, notification_configuration):\n  s3.put_bucket_notification_configuration(Bucket=bucket, NotificationConfiguration=notification_configuration)\n\ndef submit_response(event: dict, context, response_status: str, error_message: str):\n  response_body = json.dumps(\n    {\n      \"Status\": response_status,\n      \"Reason\": f\"{error_message}See the details in CloudWatch Log Stream: {context.log_stream_name}\",\n      \"PhysicalResourceId\": event.get(\"PhysicalResourceId\") or event[\"LogicalResourceId\"],\n      \"StackId\": event[\"StackId\"],\n      \"RequestId\": event[\"RequestId\"],\n      \"LogicalResourceId\": event[\"LogicalResourceId\"],\n      \"NoEcho\": False,\n    }\n  ).encode(\"utf-8\")\n  headers = {\"content-type\": \"\", \"content-length\": str(len(response_body))}\n  try:\n    req = urllib.request.Request(url=event[\"ResponseURL\"], headers=headers, data=response_body, method=\"PUT\")\n    with urllib.request.urlopen(req) as response:\n      print(response.read().decode(\"utf-8\"))\n    print(\"Status code: \" + response.reason)\n  except Exception as e:\n      print(\"send(..) failed executing request.urlopen(..): \" + str(e))\n"
    },
    "Handler": "index.handler",
    "Role": {
     "Fn::GetAtt": [
      "BucketNotificationsHandler050a0587b7544547bf325f094a3db834RoleB6FB88EC",
      "Arn"
     ]
    },
    "Runtime": "python3.9",
    "Timeout": 300
   },
   "DependsOn": [
    "BucketNotificationsHandler050a0587b7544547bf325f094a3db834RoleDefaultPolicy2CF63D36",
    "BucketNotificationsHandler050a0587b7544547bf325f094a3db834RoleB6FB88EC"
   ]
  }
 }
}

What did you expect to happen?

The resultant CFN Output had two AWS Lambda Functions, neither are in a VPC, have a DLQ, etc. I expected those to both be flagged for the Lambda CDK Nag Checks

What actually happened?

[Error at /TestStack/L2Lambda/ServiceRole/Resource] AwsSolutions-IAM4[Policy::arn:<AWS::Partition>:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole]: The IAM user, role, or group uses AWS managed policies.

[Error at /TestStack/L2Lambda/Resource] AwsSolutions-L1: The non-container Lambda function is not configured to use the latest runtime version.
[Error at /TestStack/L2Lambda/Resource] NIST.800.53.R5-LambdaConcurrency: The Lambda function is not configured with function-level concurrent execution limits - (Control IDs: AU-12(3), AU-14a, AU-14b, CA-7, CA-7b, PM-14a.1, PM-14b, PM-31, SC-6).

[Error at /TestStack/L2Lambda/Resource] NIST.800.53.R5-LambdaDLQ: The Lambda function is not configured with a dead-letter configuration - (Control IDs: AU-12(3), AU-14a, AU-14b, CA-2(2), CA-7, CA-7b, PM-14a.1, PM-14b, PM-31, SC-36(1)(a), SI-2a).

[Error at /TestStack/L2Lambda/Resource] NIST.800.53.R5-LambdaInsideVPC: The Lambda function is not VPC enabled - (Control IDs: AC-2(6), AC-3, AC-3(7), AC-4(21), AC-6, AC-17b, AC-17(1), AC-17(1), AC-17(4)(a), AC-17(9), AC-17(10), MP-2, SC-7a, SC-7b, SC-7c, SC-7(2), SC-7(3), SC-7(9)(a), SC-7(11), SC-7(12), SC-7(16), SC-7(20), SC-7(21), SC-7(24)(b), SC-25).

[Error at /TestStack/BucketNotificationsHandler050a0587b7544547bf325f094a3db834/Role/Resource] AwsSolutions-IAM4[Policy::arn:<AWS::Partition>:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole]: The IAM user, role, or group uses AWS managed policies.

[Error at /TestStack/BucketNotificationsHandler050a0587b7544547bf325f094a3db834/Role/DefaultPolicy/Resource] AwsSolutions-IAM5[Resource::*]: The IAM entity contains wildcard permissions and does not have a cdk-nag rule suppression with evidence for those permission.

[Error at /TestStack/BucketNotificationsHandler050a0587b7544547bf325f094a3db834/Role/DefaultPolicy/Resource] NIST.800.53.R5-IAMNoInlinePolicy: The IAM Group, User, or Role contains an inline policy - (Control IDs: AC-2i.2, AC-2(1), AC-2(6), AC-3, AC-3(3)(a), AC-3(3)(b)(1), AC-3(3)(b)(2), AC-3(3)(b)(3), AC-3(3)(b)(4), AC-3(3)(b)(5), AC-3(3)(c), AC-3(3), AC-3(4)(a), AC-3(4)(b), AC-3(4)(c), AC-3(4)(d), AC-3(4)(e), AC-3(4), AC-3(7), AC-3(8), AC-3(12)(a), AC-3(13), AC-3(15)(a), AC-3(15)(b), AC-4(28), AC-6, AC-6(3), AC-24, CM-5(1)(a), CM-6a, CM-9b, MP-2, SC-23(3)).

cdk-nag version

2.27.51

Language

Python

Other information

This appears to be caused just by the type check being done on the L1 construct (for Lambda:

cdk-nag/src/rules/lambda/LambdaInsideVPC.ts

Line 16 in 75acc38

if (node instanceof CfnFunction) {

).

We had a separate aspect (Python3) created, and were able to use node.cfn_resource_type == "AWS::Lambda::Function" as our type check.

[dontirun]

Thanks for bringing this issue to attention!