inspec/test/unit/resources/aws_s3_bucket_test.rb

require "helper"
require "inspec/resource"
require "resources/aws/aws_s3_bucket"

require "resource_support/aws"

# MSBSB = MockS3BucketSingleBackend
# Abbreviation not used outside this file

#=============================================================================#
#                            Constructor Tests
#=============================================================================#
class AwsS3BucketConstructor < Minitest::Test
  def setup
    AwsS3Bucket::BackendFactory.select(AwsMSBSB::Basic)
  end

  def test_constructor_no_args_raises
    assert_raises(ArgumentError) { AwsS3Bucket.new }
  end

  def test_constructor_accept_scalar_param
    AwsS3Bucket.new("some-bucket")
  end

  def test_constructor_accept_hash
    AwsS3Bucket.new(bucket_name: "some-bucket")
  end

  def test_constructor_reject_unknown_resource_params
    assert_raises(ArgumentError) { AwsS3Bucket.new(bla: "blabla") }
  end
end

class AwsS3BucketPropertiesTest < Minitest::Test
  def setup
    AwsS3Bucket::BackendFactory.select(AwsMSBSB::Basic)
  end

  #===========================================================================#
  #                               Search / Recall
  #===========================================================================#
  def test_recall_no_match_is_no_exception
    refute AwsS3Bucket.new("NonExistentBucket").exists?
  end

  def test_recall_match_single_result_works
    assert AwsS3Bucket.new("public").exists?
  end

  # No need to handle multiple hits; S3 bucket names are globally unique.

  #===========================================================================#
  #                               Properties
  #===========================================================================#
  #---------------------Bucket Name----------------------------#
  def test_property_bucket_name
    assert_equal("public", AwsS3Bucket.new("public").bucket_name)
  end

  #--------------------- Region ----------------------------#
  def test_property_region
    assert_equal("us-east-2", AwsS3Bucket.new("public").region)
    assert_equal("EU", AwsS3Bucket.new("private").region)
  end

  #---------------------- bucket_acl -------------------------------#
  def test_property_bucket_acl_structure
    bucket_acl = AwsS3Bucket.new("public").bucket_acl

    assert_kind_of(Array, bucket_acl)
    assert(bucket_acl.size > 0)
    assert(bucket_acl.all? { |g| g.respond_to?(:permission) })
    assert(bucket_acl.all? { |g| g.respond_to?(:grantee) })
    assert(bucket_acl.all? { |g| g.grantee.respond_to?(:type) })
  end

  def test_property_bucket_acl_public
    bucket_acl = AwsS3Bucket.new("public").bucket_acl

    public_grants = bucket_acl.select do |g|
      g.grantee.type == "Group" && g.grantee.uri =~ /AllUsers/
    end
    refute_empty(public_grants)
  end

  def test_property_bucket_acl_private
    bucket_acl = AwsS3Bucket.new("private").bucket_acl

    public_grants = bucket_acl.select do |g|
      g.grantee.type == "Group" && g.grantee.uri =~ /AllUsers/
    end
    assert_empty(public_grants)

    auth_users_grants = bucket_acl.select do |g|
      g.grantee.type == "Group" && g.grantee.uri =~ /AuthenticatedUsers/
    end
    assert_empty(auth_users_grants)
  end

  def test_property_bucket_acl_auth_users
    bucket_acl = AwsS3Bucket.new("auth-users").bucket_acl

    public_grants = bucket_acl.select do |g|
      g.grantee.type == "Group" && g.grantee.uri =~ /AllUsers/
    end
    assert_empty(public_grants)

    auth_users_grants = bucket_acl.select do |g|
      g.grantee.type == "Group" && g.grantee.uri =~ /AuthenticatedUsers/
    end
    refute_empty(auth_users_grants)
  end

  #---------------------- bucket_policy -------------------------------#
  def test_property_bucket_policy_structure
    bucket_policy = AwsS3Bucket.new("public").bucket_policy
    assert_kind_of(Array, bucket_policy)
    assert_kind_of(OpenStruct, bucket_policy.first)
    %i{effect principal action resource}.each do |field|
      assert_respond_to(bucket_policy.first, field)
    end
  end

  def test_property_bucket_policy_public
    bucket_policy = AwsS3Bucket.new("public").bucket_policy
    allow_all = bucket_policy.select { |s| s.effect == "Allow" && s.principal == "*" }
    assert_equal(1, allow_all.count)
  end

  def test_property_bucket_policy_private
    bucket_policy = AwsS3Bucket.new("private").bucket_policy
    allow_all = bucket_policy.select { |s| s.effect == "Allow" && s.principal == "*" }
    assert_equal(0, allow_all.count)
  end

  def test_property_bucket_policy_auth
    bucket_policy = AwsS3Bucket.new("auth").bucket_policy
    assert_empty(bucket_policy)
  end

  #===========================================================================#
  #                               Test Matchers
  #===========================================================================#
  def test_be_public_public_acl
    assert(AwsS3Bucket.new("public").public?)
  end

  def test_be_public_auth_acl
    assert(AwsS3Bucket.new("auth-users").public?)
  end

  def test_be_public_private_acl
    refute(AwsS3Bucket.new("private").public?)
  end

  def test_has_access_logging_enabled_positive
    assert(AwsS3Bucket.new("public").has_access_logging_enabled?)
  end

  def test_has_access_logging_enabled_negative
    refute(AwsS3Bucket.new("private").has_access_logging_enabled?)
  end

  def test_has_default_encryption_enabled_positive
    assert(AwsS3Bucket.new("public").has_default_encryption_enabled?)
  end

  def test_has_default_encryption_enabled_negative
    refute(AwsS3Bucket.new("private").has_default_encryption_enabled?)
  end

end

#=============================================================================#
#                               Test Fixtures
#=============================================================================#

module AwsMSBSB
  class Basic < AwsBackendBase
    def get_bucket_acl(query)
      owner_full_control = OpenStruct.new(
        grantee: OpenStruct.new(
          type: "CanonicalUser"
        ),
        permission: "FULL_CONTROL"
      )

      buckets = {
        "public" => OpenStruct.new(
          grants: [
            owner_full_control,
            OpenStruct.new(
              grantee: OpenStruct.new(
                type: "Group",
                uri: "http://acs.amazonaws.com/groups/global/AllUsers"
              ),
              permission: "READ"
            ),
          ]
        ),
        "auth-users" => OpenStruct.new(
          grants: [
            owner_full_control,
            OpenStruct.new(
              grantee: OpenStruct.new(
                type: "Group",
                uri: "http://acs.amazonaws.com/groups/global/AuthenticatedUsers"
              ),
              permission: "READ"
            ),
          ]
        ),
        "private" => OpenStruct.new(grants: [ owner_full_control ]),
        "private-acl-public-policy" => OpenStruct.new(grants: [ owner_full_control ]),
      }
      buckets[query[:bucket]]
    end

    def get_bucket_location(query)
      buckets = {
        "public" => OpenStruct.new(location_constraint: "us-east-2"),
        "private" => OpenStruct.new(location_constraint: "EU"),
        "auth-users" => OpenStruct.new(location_constraint: "ap-southeast-1"),
        "private-acl-public-policy" => OpenStruct.new(location_constraint: "ap-southeast-2"),
      }

      buckets.fetch(query[:bucket]) { raise Aws::S3::Errors::NoSuchBucket.new(nil, nil) }
    end

    def get_bucket_policy(query)
      buckets = {
        "public" => OpenStruct.new(
          policy: StringIO.new(<<~'EOP')
            {
              "Version": "2012-10-17",
              "Statement": [
                {
                  "Sid": "AllowGetObject",
                  "Effect": "Allow",
                  "Principal": "*",
                  "Action": "s3:GetObject",
                  "Resource": "arn:aws:s3:::public/*"
                }
              ]
            }
          EOP
        ),
        "private" => OpenStruct.new(
          policy: StringIO.new(<<~'EOP')
            {
              "Version": "2012-10-17",
              "Statement": [
                {
                  "Sid": "DenyGetObject",
                  "Effect": "Deny",
                  "Principal": "*",
                  "Action": "s3:GetObject",
                  "Resource": "arn:aws:s3:::private/*"
                }
              ]
            }
          EOP
        ),
        "private-acl-public-policy" => OpenStruct.new(
          policy: StringIO.new(<<~'EOP')
            {
              "Version": "2012-10-17",
              "Statement": [
                {
                  "Sid": "AllowGetObject",
                  "Effect": "Allow",
                  "Principal": "*",
                  "Action": "s3:GetObject",
                  "Resource": "arn:aws:s3:::private-acl-public-policy/*"
                }
              ]
            }
          EOP
        ),
        # No policies for auth bucket
      }

      buckets.fetch(query[:bucket]) { raise Aws::S3::Errors::NoSuchBucketPolicy.new(nil, nil) }
    end

    def get_bucket_logging(query)
      buckets = {
        "public" => OpenStruct.new(logging_enabled: OpenStruct.new(target_bucket: "log-bucket")),
        "private" => OpenStruct.new(logging_enabled: nil ),
      }

      buckets.fetch(query[:bucket]) { raise Aws::S3::Errors::NoSuchBucket.new(nil, nil) }
    end

    def get_bucket_encryption(query)
      buckets = {
          "public" => OpenStruct.new(server_side_encryption_configuration: OpenStruct.new(rules: [])),
      }
      if query[:bucket].eql? "private"
        raise Aws::S3::Errors::ServerSideEncryptionConfigurationNotFoundError.new(nil, nil)
      end

      buckets.fetch(query[:bucket]) { raise Aws::S3::Errors::NoSuchBucket.new(nil, nil) }
    end
  end
end