Examples - Formal Documentation

This page provides examples of common policy scenarios you might want to implement in Formal. Each example includes the policy code and an explanation of how it works.

Basic Policy Examples

Block All Connections by Default

A foundational security pattern: deny all access by default.

package formal.v2

import future.keywords.if

default session := {
  "action": "block",
  "type": "block_with_formal_message",
  "reason": "Default deny policy"
}

Allow Only Admin Group

Combine default deny with explicit allow for specific groups.

package formal.v2

import future.keywords.if
import future.keywords.in

default session := {
  "action": "block",
  "type": "block_with_formal_message"
}

session := {
  "action": "allow",
  "reason": "User is in admin group"
} if {
  "admin" in input.user.groups
}

Allow Machine User with End-User in Admin Group

Control access based on both the machine user and the actual person using it.

package formal.v2

import future.keywords.if
import future.keywords.in

default session := {
  "action": "block",
  "type": "block_with_formal_message"
}

session := {
  "action": "allow",
  "reason": "End-user is admin, machine user is authorized"
} if {
  "admin" in input.end_user.groups
  input.user.username == "idp:formal:machine:bi_tool"
  not input.end_user.email == "blocked@example.com"
}

Block Writes to Production

Prevent destructive operations in production environments.

package formal.v2

import future.keywords.if

request := {
  "action": "block",
  "type": "block_with_formal_message",
  "reason": "Write operations not allowed in production"
} if {
  input.resource.environment == "production"
  input.query.statement_type in ["INSERT", "UPDATE", "DELETE", "DROP"]
}

Block an HTTP Request by Hostname and Path

Use input.http.hostname to narrow a policy to a single upstream. Without it, a path-based rule would match that path on every HTTP resource.

package formal.v2

import future.keywords.if

request := {
  "action": "block",
  "type": "block_with_formal_message",
  "reason": "Access to this package is restricted"
} if {
  input.http.hostname == "registry.npmjs.org"
  input.http.method == "GET"
  input.http.path == "/express"
}

For the full list of HTTP policy inputs, see HTTP object.

Advanced Scenarios

Redact emails

The two policies below redact email columns coming back from a database. The first masks them, and the second blocks queries that could try to bypass the masking.

Policy 1 – Mask emails

This policy masks every column whose data_label is email_address, based on each column lineage: returned columns derived from an email source column (wildcards, joins, unions) are also masked.

package formal.v2

import future.keywords.if

# Mask emails
response := {
  "action": "mask",
  "type": "redact.partial",
  "sub_type": "email_mask_username",
  "columns": columns,
  "typesafe": "fallback_to_default"
} if {
  columns := [column |
    column := input.row[_]
    column.data_label == "email_address"
  ]
  columns != []
}

Policy 2 – Block queries that try to bypass masking

Attackers can leverage advanced engine features to escape masking. This policy blocks queries containing such constructs.

package formal.v2

import future.keywords.if
import future.keywords.in

# Block queries that may try to escape our masking
request := {
  "action": "block",
  "type": "block_with_formal_message",
} if {
  forbidden_substrings = [
    "CREATE", "SYSTEM$", "IDENTIFIER", "RESULT_SCAN", "PROCEDURE",
    "CLONE", "LATERAL", "REPLACE", "RENAME", "PIVOT", "RECURSIVE"
  ]
  some substring in forbidden_substrings
  contains(input.sql_query.query, substring)
}

Note that this policy uses request instead of response to block the query before execution.

Alternatively, CREATE queries can be blocked via input.sql_query.statement_type in another policy.

Block Access to Specific Native Users

You can control which users can access specific native users when they attempt to connect using the @<native_user> syntax. This is useful for enforcing least-privilege access and preventing users from accessing highly privileged accounts. The following policy blocks the user john@joinformal.com from using the native user devops:

package formal.v2

import future.keywords.if
import future.keywords.in

session := {
  "action": "block",
  "type": "block_with_formal_message"
} if {
  input.native_user == "devops"
  input.user.email == "john@joinformal.com"
}

You can extend this pattern to:

Block multiple users from specific native users
Allow only certain users to access privileged native users
Require additional authentication for certain native users

Block Access to Objects in AWS S3

This policy prevents access to S3 objects that were last modified more than 10 minutes ago. This is useful for ensuring data freshness and preventing access to outdated information.

package formal.v2

import future.keywords.if

# Helper function to check if a timestamp is older than 10 minutes
is_stale(timestamp) if {
  parsed_time = time.parse_rfc3339_ns(timestamp)
  current_time := time.now_ns()

  # 10 minutes = 10 * 60 * 1000000000 nanoseconds
  cutoff_time := current_time - (10 * 60 * 1000000000)

  parsed_time < cutoff_time
}

# Block access to stale objects in the local-bucket
response := {
  "action": "block",
  "type": "block_with_formal_message"
} if {
  input.bucket.name == "local-bucket"
  input.bucket.action == "GetObject"
  is_stale(input.file.last_modified)
}

How it works:

is_stale function: Compares the object’s last_modified timestamp with the current time minus 10 minutes
Time calculation: Uses nanoseconds for precision (10 minutes = 600,000,000,000 nanoseconds)
Blocking logic: Blocks GetObject requests to objects in local-bucket that are older than 10 minutes
Response stage: Uses response to evaluate after the request is processed but before returning results

Use cases:

Prevent access to outdated configuration files
Ensure users only see recent data exports
Block access to temporary files that should have been cleaned up

Rate Limiting Access

Formal automatically tracks access patterns across different time windows and makes this data available to policies. You can use access count data to implement rate limiting policies that block users who exceed defined thresholds.

Basic Rate Limiting by Minute

This policy blocks access to sensitive-bucket if the user has accessed it more than five times in the last minute:

package formal.v2

import future.keywords.if

request := {
  "action": "block",
  "type": "block_with_formal_message",
  "reason": "Rate limit exceeded: too many requests per minute"
} if {
  data.access_count_minute["sensitive-bucket"] > 5
}

Rate Limiting by Hour

You can also limit access over longer time windows. This policy blocks access if the user has made more than 100 requests in the last hour:

package formal.v2

import future.keywords.if

request := {
  "action": "block",
  "type": "block_with_formal_message",
  "reason": "Hourly rate limit exceeded: too many requests per hour"
} if {
  data.access_count_hour["sensitive-database"] > 5
}

How it works:

Automatic tracking: Formal automatically tracks access counts per user, per resource, per path (database/bucket name)
Time windows: Access counts are available for minute (data.access_count_minute) and hour (data.access_count_hour) windows
Cluster-wide: Access counts are synchronized across all Connector instances using distributed state
Policy evaluation: Access count data is fetched and made available during policy evaluation at the request, response, and session stages

Available data:

data.access_count_minute[<bucket_name>] - Number of accesses in the last minute
data.access_count_hour[<bucket_name>] - Number of accesses in the last hour

Use cases:

Prevent denial-of-service attacks by limiting request rates
Enforce fair usage policies across users
Protect against runaway scripts or applications
Different rate limits for different buckets or databases

Rate limiting also works for databases. Use data.access_count_minute[<database_name>] and data.access_count_hour[<database_name>] to limit database access patterns.

Clustering Required: If you’re running multiple Connector instances, you need to enable the instances to form a cluster. See the Clustering page for configuration details.

Rewrite HTTP Request Headers

The rewrite action adds, replaces, or removes HTTP headers before the request reaches the upstream service. The following policy adds a custom header and removes a debug header on all HTTP requests:

package formal.v2

import future.keywords.if

rewritten_headers := object.union(input.http.headers, {
  "x-custom-rewrite": ["formal-injected-value"],
  "x-debug": [],
})

request := {
  "action": "rewrite",
  "headers": rewritten_headers,
} if {
  input.resource.technology == "http"
}

How it works:

object.union merges the new headers into the existing input.http.headers. Keys in the second object overwrite keys in the first.
Setting a header to an empty array ([]) removes it from the request.
Header names are normalized to canonical HTTP casing before forwarding (e.g., x-custom-rewrite becomes X-Custom-Rewrite).
Formal logs both received and sent headers, visible in the Logs page.

For more details on header rewrites, see HTTP resources - Header Rewrites.

Best Practices

As you can see with this example, policies can quickly get quite extensive. When creating policies, keep these best practices in mind:

Use the correct package name (formal.v2 for data policies).
Import required keywords (future.keywords.if, future.keywords.in).
Use session, request, and response stages accordingly.
Test policies before deploying to production by using the Dry-run status.

For more information about policy evaluation and available input objects, see the Evaluation page.

We recommend that you write and version your policies in git, and deploy them with Terraform using our provider.

Documentation Index

​Basic Policy Examples

​Block All Connections by Default

​Allow Only Admin Group

​Allow Machine User with End-User in Admin Group

​Block Writes to Production

​Block an HTTP Request by Hostname and Path

​Advanced Scenarios

​Redact emails

​Policy 1 – Mask emails

​Policy 2 – Block queries that try to bypass masking

​Block Access to Specific Native Users

​Block Access to Objects in AWS S3

​Rate Limiting Access

​Basic Rate Limiting by Minute

​Rate Limiting by Hour

​Rewrite HTTP Request Headers

​Best Practices

Basic Policy Examples

Block All Connections by Default

Allow Only Admin Group

Allow Machine User with End-User in Admin Group

Block Writes to Production

Block an HTTP Request by Hostname and Path

Advanced Scenarios

Redact emails

Policy 1 – Mask emails

Policy 2 – Block queries that try to bypass masking

Block Access to Specific Native Users

Block Access to Objects in AWS S3

Rate Limiting Access

Basic Rate Limiting by Minute

Rate Limiting by Hour

Rewrite HTTP Request Headers

Best Practices