Advanced Custom Rules

Note

• For an overview on custom rules, see Writing Rules.

• For details on simple custom rules, see Simple Custom Rules.

An advanced custom rule tests one or more resource types. Because it evaluates resources in the full context of an environment, it can also filter out nonapplicable resources of a given type and check whether required resources are missing from an environment.

Here’s an advanced rule written for AWS runtime and Terraform repository environments. It checks whether an S3 bucket with a stage:prod tag has a private ACL, and ignores buckets that do not have the tag:

package rules.filterbuckets
import data.fugue

__rego__metadoc__ := {
  "title": "AWS S3 buckets tagged 'stage:prod' must have private ACLs",
  "description": "S3 buckets with the tag key 'stage' and value 'prod' must have private ACLs",
  "custom": {
    "providers": ["AWS", "REPOSITORY"],
    "severity": "High"
  }
}

input_type = "tf"

resource_type = "MULTIPLE"

buckets = fugue.resources("aws_s3_bucket")

policy[r] {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  bucket.acl == "private"
  r = fugue.allow_resource(bucket)
} {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  not bucket.acl == "private"
  r = fugue.deny_resource(bucket)
}

To create an advanced custom rule, you write Rego code defining pass and fail conditions, using the fugue.allow_resource(resource) and fugue.deny_resource(resource) functions. Then, you validate and add the rule to Fugue from the Custom Rules page (UI), CLI, or API.

To write a custom rule, you must:

1. Determine the environment provider(s) the rule should apply to

2. Determine the resource type(s)

3. Determine the input type

4. Determine the resource attribute(s) you want Fugue to check

5. Determine whether to write a simple or advanced rule

6. Define pass/fail conditions for the resource

7. Write metadata

You can write custom rules using Rego, an open source query language for Open Policy Agent (OPA).

We won’t detail how to write Rego here, but we encourage you to consult the Rego documentation for the basics.

Optional but recommended step

To ensure compatibility with Regula, start your advanced rule with a package declaration and an import declaration:

package rules.filterbuckets
import data.fugue

The package name should start with rules, and the rule should import the data.fugue library.

This is an optional step if you’re only using Fugue, but it’s a good practice. See Compatibility with Regula for details.

Step 1: Determine provider(s)

Once you have a policy in mind that your resources should comply with, you can start writing a custom rule.

To translate policy to code, you’ll first need to determine the provider(s) involved.

Which provider is your rule for? (Reference)

• Runtime environments:

  • AWS

  • AWS_GOVCLOUD

  • AZURE (includes Azure Government)

  • GOOGLE

• Infrastructure as code (IaC) environments:

  • REPOSITORY, containing one of the following IaC input types:

    • Terraform for AWS/AWS GovCloud, Azure, or Google

    • CloudFormation for AWS and AWS GovCloud

    • Kubernetes

You can also write a rule for multiple providers. For instance:

• Terraform for AWS/AWS Govcloud (REPOSITORY) + AWS

• Terraform for Google (REPOSITORY) + GOOGLE

• Terraform for Azure/Azure Government (REPOSITORY) + AZURE

• AWS + AZURE + GOOGLE

Note

CloudFormation cannot be combined with AWS/AWS GovCloud runtime.

In this example, our rule is for AWS runtime and Terraform (HCL and plan) repository environments.

Declaring the provider

There are different ways you can declare the provider(s) for a rule, depending on the interface chosen:

• UI: In the “Providers” field

• CLI fugue create rule command: In the --provider flag (single provider only)

• CLI fugue sync rules command: In the providers metadata property of the Rego code, which accepts a list of strings

• API: In the providers request body parameter

Our example rule is meant for AWS and repository environments, and we intend to upload it using the fugue sync rules CLI command, so we’ll include it in metadata when we get to step 7:

"providers": ["AWS", "REPOSITORY"]

Step 2: Determine the resource type(s)

Advanced rules differ from simple rules in that the resource type provided to Fugue is always "MULTIPLE". This is true even if there’s only one resource type in the rule.

That’s because the input for an advanced rule is an entire environment, rather than a single resource at a time.

Elsewhere in the rule, you need to specify the actual resource types using the fugue.resources(resource_type) function, like we’ve done in our example rule, which looks at Amazon S3 buckets:

buckets = fugue.resources("aws_s3_bucket")

You specify the resource types in the fugue.resources(resource_type) function according to the provider. (Reference)

• Runtime-only rules use Fugue names for resources:

  • AWS and AWS GovCloud (e.g., AWS.S3.Bucket)

  • Azure and Azure Government (e.g., Azure.Storage.Account)

  • Google (e.g., Google.Storage.Bucket)

• Repository (or repository + runtime) rules use the resource name according to the input type:

  • Terraform HCL and JSON plans

    • AWS and AWS GovCloud (e.g., aws_s3_bucket)

    • Azure (e.g., azurerm_storage_account)

    • Google (e.g., google_storage_bucket)

  • CloudFormation (e.g., AWS::S3::Bucket)

  • Kubernetes (see the KIND column) (e.g., Pod)

If your rule involves multiple resource types, include a fugue.resources(resource_type) function for each resource type:

trails = fugue.resources("aws_cloudtrail")
buckets = fugue.resources("aws_s3_bucket")

Our example rule is for AWS runtime and Terraform environments, so we’re using aws_s3_bucket (the Terraform name) rather than AWS.S3.Bucket (the Fugue name).

Declaring the resource type

There are different ways you can let Fugue know the resource type is “MULTIPLE”:

• Define in code. Required for:

  • Rules for REPOSITORY environments

  • Rules involving multiple providers of any type

  • Rules uploaded to Fugue via fugue sync rules

• Define externally. Required for:

  • Rules for single, runtime-only providers (for example, just AWS, or just GOOGLE) not uploaded with fugue sync rules

Rules that must define the resource type in code require the resource_type declaration. For example:

resource_type = "MULTIPLE"

If your rule is for a single, runtime-only provider, and you’re not uploading it to Fugue with fugue sync rules, you can declare the resource type according to the interface you use:

• UI: In the “Resource Type” field

• CLI fugue create rule command: In the --resource-type flag

• API: In the resource_type request body parameter

Our example rule is meant for AWS and AWS repository (Terraform) environments, so we’ll include the resource_type declaration in the Rego code:

resource_type = "MULTIPLE"

Step 3: Determine the input type

Repository (and repository + runtime) rules require an input_type declaration, unless you’re writing a rule for Terraform, which is the default. For example:

input_type = "k8s"

Runtime-only rules do not require an input_type declaration.

What input type does your rule involve?

• Terraform HCL and JSON plans: tf (default value)

• Terraform JSON plans only: tf_plan

• CloudFormation: cfn

• Kubernetes: k8s

Input types cannot be combined; you cannot write a rule for both CloudFormation and Terraform, for instance.

Our example rule is for Terraform HCL and plans as well as AWS runtime resources, so setting the input type to tf is optional since it’s the default value. For completeness, we’ve included the declaration anyway:

input_type = "tf"

Step 4: Determine resource attribute(s) to check

If you have an environment (or create a new one) that contains the resource type you’re evaluating, you can determine which attributes you need to specify in the rule by using the CLI or API to retrieve the results of a recent scan, known as the “test input”.

Once you have the JSON document containing the test input, you’ll see the configuration for every resource in the scan. You can then find the attribute in question.

For instance, in this example we want to check S3 buckets that have stage:prod tags, and the test input for a scan of an AWS runtime environment has this information:

{
    "resources": {
        "aws_s3_bucket.ZnVndWUtbGFy": {
            "_provider": "provider.aws.us-east-1",
            "_skeleton": {
                "depends_on": null,
                "deposed": null,
                "primary": {
                    "id": "example-bucket",
                    "meta": null,
                    "tainted": false
                },
                "provider": "provider.aws.us-east-1",
                "type": "aws_s3_bucket"
            },
            "_type": "aws_s3_bucket",
            "acl": "private",
            "arn": "arn:aws:s3:::example-bucket",
            "bucket": "example-bucket",
            "bucket_domain_name": "example-bucket.s3.amazonaws.com",
            "bucket_regional_domain_name": "example-bucket.s3.amazonaws.com",
            "cors_rule": [],
            "force_destroy": false,
            "grant": [],
            "hosted_zone_id": "Z3AQBSABCDEFGH",
            "id": "example-bucket",
            "lifecycle_rule": [],
            "logging": [],
            "object_lock_configuration": [],
            "region": "us-east-1",
            "replication_configuration": [],
            "request_payer": "BucketOwner",
            "server_side_encryption_configuration": [],
            "tags": {
                "stage": "prod"
            },
               },
            "versioning": [
                {
                    "enabled": false,
                    "mfa_delete": false
                }
            ],
            "website": []
        }
    }
}

Here, we’d look at the input to see that the stage property is nested underneath tags, a “top-level” property, so the attribute we need to check for is tags.stage.

(Note that scan data for repository environments will look a little different because it has some additional information apart from resource attributes, such as _source_location, which indicates the line, column, and path of the resource in an infrastructure as code file.)

Alternate method for finding resource attributes

Alternatively, you can reference the Terraform documentation to determine the resource types and syntax required for AWS and AWS GovCloud, Azure, and Google. You’ll find the information you need in the Resource page of each provider resource. For an example, see Resource: aws_s3_bucket.

You can reference the CloudFormation documentation the same way. For an example, see AWS::S3::Bucket.

Step 5: Determine whether to write a simple or advanced rule

Once you determine the provider(s), resource type(s), and attribute(s) you need to check, you decide whether to write a simple or advanced rule.

If any of these conditions apply, the rule must be advanced:

• It involves multiple types (example rule)

• It filters out certain resources of the selected type; for instance, checking only some S3 buckets (see our example rule above)

• It evaluates whether a required resource type is missing (example rule)

In this example, our rule must be written as an advanced rule, because it filters out (ignores) S3 buckets that do not have a stage:prod tag. It cannot be written as a simple rule.

Step 6: Define pass/fail conditions for the resource

After determining the provider(s), resource type(s), input type (if needed), and attribute(s), you can start writing the rule logic. In advanced rules, that requires both the fugue.allow_resource(resource) and fugue.deny_resource(resource) functions.

The high-level workflow for advanced rules is usually:

1. Query for resources by using the fugue.resources(resource_type) function.

2. Declare a policy set rule that iterates over queried resources.

3. Extend the policy set using a combination of the following functions (explained in the next section):

• fugue.allow_resource(resource)

• fugue.deny_resource(resource)

• fugue.missing_resource(resource_type)

Advanced Rule Functions

fugue.resources(resource_type)

Returns a collection of resources of the given resource_type string. Fugue also keeps track of which resource types are requested. We use this information to decide whether there are simply no resources of a type, or if we do not have permission to scan that resource type.

fugue.allow_resource(resource)

Makes a PASS judgement about a resource retrieved using fugue.resources.

fugue.deny_resource(resource)

Makes a DENY judgement about a resource retrieved using fugue.resources.

fugue.missing_resource(resource_type)

Makes a DENY judgement about a resource type string without actually having access to a resource. This can be used to write a rule that checks for the presence of a specific resource.

Why do we need more than allow and deny? For example, if we require an AWS CloudTrail alarm for failed logins, it does not mean that other CloudTrail alarms are invalid — they are simply irrelevant. Hence, we should not mark these other CloudTrail alarms as DENY using fugue.deny_resource, but rather use fugue.missing_resource to indicate that a resource is missing.

fugue.input_resource_types

Is a set of all the resource types in an environment. This set is helpful to avoid the missing resource type errors for rules that check, but don’t require, lots of different resource types. Because it’s a set, not a function, it doesn’t take an argument. fugue.input_resource_types replaces the deprecated function fugue.resource_types().

Though the example rule below is not practical in a real-world scenario, it demonstrates how this set works. It collects all of the resource types in an environment and denies all resources of each type:

all_resources[id] = resource {
  resource_types = fugue.input_resource_types
  resource_types[ty]
  resource := fugue.resources(ty)[id]
}
policy[pol] {
  resource := all_resources[_]
  pol := fugue.deny_resource(resource)
}

fugue.allow_resource and fugue.deny_resource

In Fugue advanced rules, a policy rule collects all of the pass/fail judgments. In this policy rule, you define a “pass” state by using the fugue.allow_resource(resource) function. You define a “fail” state by using the fugue.deny_resource(resource) function.

If a resource is not explicitly allowed or denied, it is ignored.

In this example, we’ve determined that S3 buckets with a tag key of stage and the tag value prod should have a private ACL.

• If a bucket tagged stage:prod has a private ACL, it should be considered compliant (i.e., produce a PASS rule result).

• If a bucket tagged stage:prod does not have a private ACL, the bucket should raise a compliance violation (i.e., produce a FAIL rule result).

• If a bucket is not tagged stage:prod, it should be ignored (i.e., the rule will not be applied to the bucket at all, and it won’t have a PASS or FAIL rule result).

So, the custom rule should query each aws_s3_bucket in an environment where tags.stage is set to prod and verify that the acl property is set to private. If so, the resource passes the compliance check. If not, the resource fails. And again, if the bucket does not have tags.stage set to prod, it’s filtered out (ignored).

Let’s look at the example logic from the beginning of this document:

buckets = fugue.resources("aws_s3_bucket")

policy[r] {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  bucket.acl == "private"
  r = fugue.allow_resource(bucket)
} {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  not bucket.acl == "private"
  r = fugue.deny_resource(bucket)
}

We followed the high-level workflow to build a skeleton like so:

1. Query for S3 bucket resources:

buckets = fugue.resources("aws_s3_bucket")

2. Declare a policy[r] set rule and iterate over queried resources:

policy[r] {
  bucket = buckets[_]
}

3. Extend the policy[r] set by using both fugue.allow_resource(resource) and fugue.deny_resource(resource):

policy[r] {
  bucket = buckets[_]
  r = fugue.allow_resource(bucket)
} {
  bucket = buckets[_]
  r = fugue.deny_resource(bucket)
}

Now, we can fill in the conditions where a bucket should be allowed (pass) or denied (fail).

But first, we want to filter out buckets that don’t have tags.stage == "prod", so we only examine buckets that meet that condition. We can do this by adding the condition to both the fugue.allow_resource and fugue.deny_resource rules:

policy[r] {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  r = fugue.allow_resource(bucket)
} {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  r = fugue.deny_resource(bucket)
}

As a result, a bucket not meeting that condition is neither allowed nor denied (i.e., it’s ignored).

Finally, we specify the pass or fail conditions. In this case, a tagged bucket passes if bucket.acl == "private", and it fails if not bucket.acl == "private":

buckets = fugue.resources("aws_s3_bucket")

policy[r] {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  bucket.acl == "private"
  r = fugue.allow_resource(bucket)
} {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  not bucket.acl == "private"
  r = fugue.deny_resource(bucket)
}

Input in simple vs. advanced rules

Simple rules use input to refer to a single resource of the declared type, because simple rules look at each resource individually. So in the simple rule example, we used input.tags.stage to check a single bucket’s tags.

However, the input for an advanced rule is an entire environment, so to evaluate a single resource in this example, first we must collect the resources with buckets = fugue.resources("aws_s3_bucket"), then iterate through the queried resources with bucket = buckets[_]. After that, we evaluate that individual bucket with bucket.tags.stage, rather than input.tags.stage.

Multiple resource types

An advanced rule can examine multiple resources types at once. For instance, the example below checks an AWS runtime or AWS Terraform repository environment to see if Amazon S3 buckets storing CloudTrail logs have a private ACL:

package rules.trailbucketlogs
import data.fugue

__rego__metadoc__ := {
  "title": "S3 buckets containing CloudTrail logs must be private",
  "description": "Amazon S3 buckets containing CloudTrail logs must have a private ACL",
  "custom": {
    "providers": ["AWS", "REPOSITORY"],
    "severity": "High"
  }
}

input_type = "tf"

resource_type = "MULTIPLE"

trails = fugue.resources("aws_cloudtrail")
buckets = fugue.resources("aws_s3_bucket")

trail_bucket_ids[bucket] {
  bucket = trails[_].s3_bucket_name
}

policy[r] {
  bucket = buckets[_]
  trail_bucket_ids[bucket.bucket]
  bucket.acl == "private"
  r = fugue.allow_resource(bucket)
} {
  bucket = buckets[_]
  trail_bucket_ids[bucket.bucket]
  not bucket.acl == "private"
  r = fugue.deny_resource(bucket)
}

Missing resource types

An advanced rule can check for the existence of required resource types, using the fugue.missing_resource(resource_type) function. For instance, the rule below checks whether an AWS runtime or AWS Terraform repository environment has an iam_password_policy resource, and raises a compliance violation if it is missing or if it exists but the minimum_password_length property is set to less than 16 characters:

package rules.requirepasswordpolicy
import data.fugue

__rego__metadoc__ := {
  "title": "AWS accounts require a password policy",
  "description": "All AWS accounts must contain a password policy resource requiring a minimum password length of 16 characters",
  "custom": {
    "providers": ["AWS", "REPOSITORY"],
    "severity": "High"
  }
}

input_type = "tf"

resource_type = "MULTIPLE"

password_policies = fugue.resources("aws_iam_account_password_policy")

policy[r] {
  password_policy = password_policies[_]
  password_policy.minimum_password_length >= 16
  r = fugue.allow_resource(password_policy)
} {
  password_policy = password_policies[_]
  not password_policy.minimum_password_length >= 16
  r = fugue.deny_resource(password_policy)
} {
  count(password_policies) == 0
  r = fugue.missing_resource("aws_iam_account_password_policy")
}

Custom violation messages

You can provide a custom violation message when a rule fails by using one or both of the following functions:

• fugue.deny_resource_with_message(resource, message)

• fugue.missing_resource_with_message(resource_type, message)

These functions take the place of fugue.deny_resource(resource_type) and fugue.missing_resource(resource_type), respectively.

Here’s an example that uses both functions, based on the missing resource example rule above:

package rules.passwordpolicymessage
import data.fugue

__rego__metadoc__ := {
  "title": "AWS accounts require a password policy",
  "description": "All AWS accounts must contain a password policy resource requiring a minimum password length of 16 characters",
  "custom": {
    "providers": ["AWS", "REPOSITORY"],
    "severity": "High"
  }
}

input_type = "tf"

resource_type = "MULTIPLE"

password_policies = fugue.resources("aws_iam_account_password_policy")

policy[r] {
  password_policy = password_policies[_]
  password_policy.minimum_password_length >= 16
  r = fugue.allow_resource(password_policy)
} {
  password_policy = password_policies[_]
  not password_policy.minimum_password_length >= 16
  msg = "Password policy is too short. It must be at least 16 characters."
  r = fugue.deny_resource_with_message(password_policy, msg)
} {
  count(password_policies) == 0
  msg = "No password policy exists."
  r = fugue.missing_resource_with_message("aws_iam_account_password_policy", msg)
}

You’ll see this custom message in the Compliance by Control tab, under Rule Message:

Step 7: Write metadata

If you’re going to upload the rule to Fugue via the fugue sync rules CLI command, you must add metadata to tell Fugue additional information about the rule:

• Title

• Description

• Providers

• Severity

It’s also a good idea to include it when writing a rule that will be used with Regula.

If you’re not syncing the rule to Fugue with fugue sync rules, you define these fields externally – via the UI, CLI fugue create rule command, or API at the time of creation/update. In these cases, any metadata provided is simply ignored.

Declaring the metadata

Because we plan to sync our example rule to Fugue via fugue sync rules, we must include metadata in the __rego__metadoc__ (see metadata reference):

__rego__metadoc__ := {
  "title": "AWS S3 buckets tagged 'stage:prod' must have private ACLs",
  "description": "S3 buckets with the tag key 'stage' and value 'prod' must have private ACLs",
  "custom": {
    "providers": ["AWS", "REPOSITORY"],
    "severity": "High"
  }
}

Here’s the complete rule, with metadata:

package rules.filterbuckets
import data.fugue

__rego__metadoc__ := {
  "title": "AWS S3 buckets tagged 'stage:prod' must have private ACLs",
  "description": "S3 buckets with the tag key 'stage' and value 'prod' must have private ACLs",
  "custom": {
    "providers": ["AWS", "REPOSITORY"],
    "severity": "High"
  }
}

input_type = "tf"

resource_type = "MULTIPLE"

buckets = fugue.resources("aws_s3_bucket")

policy[r] {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  bucket.acl == "private"
  r = fugue.allow_resource(bucket)
} {
  bucket = buckets[_]
  bucket.tags.stage == "prod"
  not bucket.acl == "private"
  r = fugue.deny_resource(bucket)
}

What’s Next?

Once you’ve written a rule, the next step is to create it:

• Via the UI

• Via the CLI fugue create rule command (to create a single rule)

• Via the CLI fugue sync rules command (to upload a directory of one or more rules)

• Via the API

Note

Looking for some examples of custom rules? Check out our GitHub repo!