inspec/test/unit/resources/aws_s3_bucket_test.rb

# encoding: utf-8
require 'helper'

# 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

#=============================================================================#
#                               Search / Recall
#=============================================================================#
class AwsS3BucketPropertiesTest < Minitest::Test
  def setup
    AwsS3Bucket::BackendFactory.select(AwsMSBSB::Basic)
  end

  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.
end

#=============================================================================#
#                               Properties
#=============================================================================#

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

  #---------------------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)
    [: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

end

#=============================================================================#
#                               Test Matchers
#=============================================================================#

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

  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_be_public_public_acl
    assert(AwsS3Bucket.new('public').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
  
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' }),
      }
      unless buckets.key?(query[:bucket])
        raise Aws::S3::Errors::NoSuchBucket.new(nil, nil)
      end
      buckets[query[:bucket]]
    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
      }
      unless buckets.key?(query[:bucket])
        raise Aws::S3::Errors::NoSuchBucketPolicy.new(nil, nil)
      end
      buckets[query[:bucket]]
    end

    def get_bucket_logging(query)
      buckets = {
        'public' => OpenStruct.new({ logging_enabled: OpenStruct.new({ target_bucket: 'log-bucket' }) }),
        'private' => OpenStruct.new({ logging_enabled: nil }),
      }
      unless buckets.key?(query[:bucket])
        raise Aws::S3::Errors::NoSuchBucket.new(nil, nil)
      end
      buckets[query[:bucket]]
    end
  end
end