# STR policy evaluations =======================================================
# Controls: STR-001 through STR-007

package controls.aws.str

import data.common.exemptions as exemptions
import data.common.outputs as outputs
import data.common.policydoc as policydoc

# Helpers ----------------------------------------------------------------------

s3_buckets := {r.name: r |
	some r in input.resource_changes
	r.type == "aws_s3_bucket"
	r.change.actions[_] in ["create", "update"]
}

companion_exists(companion_type, bucket_name) if {
	some r in input.configuration.root_module.resources
	r.type == companion_type
	some ref in r.expressions.bucket.references
	ref == concat(".", ["aws_s3_bucket", bucket_name])
}

# STR-001 — Require S3 Block Public Access ------------------------------------

deny contains msg if {
	some policy_id in {"ICP-TF-AWS-STR-001", "ICP-TF-AWS-GAC-001"}

	some name, bucket in s3_buckets
	not companion_exists("aws_s3_bucket_public_access_block", name)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket '%s' is missing a public access block", [name]),
	)
}

deny contains msg if {
	some policy_id in {"ICP-TF-AWS-STR-001", "ICP-TF-AWS-GAC-001"}

	some r in input.resource_changes
	r.type == "aws_s3_bucket_public_access_block"
	r.change.actions[_] in ["create", "update"]

	flags := [
		r.change.after.block_public_acls,
		r.change.after.block_public_policy,
		r.change.after.ignore_public_acls,
		r.change.after.restrict_public_buckets,
	]
	false in flags

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 public access block '%s' has one or more flags disabled", [r.name]),
	)
}

# STR-002 — Prohibit Public Read Access on S3 Buckets -------------------------

deny contains msg if {
	policy_id := "ICP-TF-AWS-STR-002"

	some r in input.resource_changes
	r.type == "aws_s3_bucket_policy"
	r.change.actions[_] in ["create", "update"]

	some stmt in policydoc.statements(r.change.after.policy)
	stmt.Effect == "Allow"
	policydoc.public_principal(stmt)

	some action in policydoc.actions(stmt)
	_read_action(action)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket policy '%s' grants public read access", [r.name]),
	)
}

_read_action(a) if a == "*"
_read_action(a) if a == "s3:*"
_read_action(a) if a == "s3:GetObject"
_read_action(a) if a == "s3:ListBucket"
_read_action(a) if a == "s3:GetBucketAcl"

# STR-003 — Prohibit Public Write Access on S3 Buckets ------------------------

deny contains msg if {
	policy_id := "ICP-TF-AWS-STR-003"

	some r in input.resource_changes
	r.type == "aws_s3_bucket_policy"
	r.change.actions[_] in ["create", "update"]

	some stmt in policydoc.statements(r.change.after.policy)
	stmt.Effect == "Allow"
	policydoc.public_principal(stmt)

	some action in policydoc.actions(stmt)
	_write_action(action)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket policy '%s' grants public write access", [r.name]),
	)
}

_write_action(a) if a == "*"
_write_action(a) if a == "s3:*"
_write_action(a) if a == "s3:PutObject"
_write_action(a) if a == "s3:DeleteObject"
_write_action(a) if a == "s3:PutBucketAcl"

# STR-004 — Require Default S3 Server-Side Encryption -------------------------

deny contains msg if {
	some policy_id in {"ICP-TF-AWS-STR-004", "ICP-TF-AWS-GAC-002"}

	some name, bucket in s3_buckets
	not companion_exists("aws_s3_bucket_server_side_encryption_configuration", name)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket '%s' is missing server-side encryption configuration", [name]),
	)
}

# STR-005 — Require TLS-Only Access to S3 Buckets ----------------------------

deny contains msg if {
	some policy_id in {"ICP-TF-AWS-STR-005", "ICP-TF-AWS-GAC-004"}

	some name, _ in s3_buckets
	not companion_exists("aws_s3_bucket_policy", name)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket '%s' has no bucket policy — TLS-only deny statement required", [name]),
	)
}

deny contains msg if {
	policy_id := "ICP-TF-AWS-STR-005"

	some r in input.resource_changes
	r.type == "aws_s3_bucket_policy"
	r.change.actions[_] in ["create", "update"]

	not _has_secure_transport_deny(r.change.after.policy)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket policy '%s' is missing a SecureTransport deny statement", [r.name]),
	)
}

_has_secure_transport_deny(raw) if {
	some stmt in policydoc.statements(raw)
	stmt.Effect == "Deny"
	stmt.Condition.Bool["aws:SecureTransport"] == "false"
}

# STR-006 — Restrict Cross-Account Grants in S3 Bucket Policies ---------------

deny contains msg if {
	policy_id := "ICP-TF-AWS-STR-006"

	some r in input.resource_changes
	r.type == "aws_s3_bucket_policy"
	r.change.actions[_] in ["create", "update"]

	some stmt in policydoc.statements(r.change.after.policy)
	stmt.Effect == "Allow"

	_broad_principal(stmt)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket policy '%s' grants access to a broad or wildcard principal", [r.name]),
	)
}

_broad_principal(stmt) if stmt.Principal == "*"
_broad_principal(stmt) if stmt.Principal.AWS == "*"
_broad_principal(stmt) if endswith(stmt.Principal.AWS, ":root")

# STR-007 — Enforce S3 Bucket Owner Object Ownership --------------------------

deny contains msg if {
	policy_id := "ICP-TF-AWS-STR-007"

	some name, _ in s3_buckets
	not companion_exists("aws_s3_bucket_ownership_controls", name)

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 bucket '%s' is missing ownership controls", [name]),
	)
}

deny contains msg if {
	policy_id := "ICP-TF-AWS-STR-007"

	some r in input.resource_changes
	r.type == "aws_s3_bucket_ownership_controls"
	r.change.actions[_] in ["create", "update"]

	some rule in r.change.after.rule
	rule.object_ownership != "BucketOwnerEnforced"

	meta := data.policies[policy_id]
	meta.status == "active"

	not exemptions.policy_exemption(policy_id)
	not exemptions.severity_exemption(meta.severity)

	msg := outputs.violation(
		policy_id, meta,
		sprintf("S3 ownership controls '%s' object_ownership is not BucketOwnerEnforced", [r.name]),
	)
}