apiVersion: templates.gatekeeper.sh/v1beta1 kind: ConstraintTemplate metadata: name: k8ssvcclusteripexternalipsallow spec: crd: spec: names: kind: K8sSvcClusterIpExternalIpsAllow listKind: K8sSvcClusterIpExternalIpsAllowList plural: k8ssvcclusteripexternalipsallow singular: k8ssvcclusteripexternalipsallow validation: # Schema for the `parameters` field openAPIV3Schema: properties: allowedOps: type: array items: type: string allowedIps: type: array items: type: string errMsg: type: string targets: - target: admission.k8s.gatekeeper.sh libs: - | package lib.k8s.helpers allowed_operations = value { value := {"CREATE", "UPDATE"} } review_operation = value { value := input.review.operation } review_metadata_labels = value { value := input.review.object.metadata.labels } review_spec_template_metadata_labels = value { value := input.review.object.spec.template.metadata.labels } deployment_error = value { value := "DEPLOYMENT_INVALID" } deployment_containers = value { value := input.review.object.spec.template.spec.containers } deployment_role = value { value := input.review.object.spec.template.metadata.annotations["iam.amazonaws.com/role"] } ips_allowed(a,x) { allowedIPs := {ip | ip := a[_]} externalIPs := {ip | ip := x[_]} forbiddenIPs := externalIPs - allowedIPs count(forbiddenIPs) > 0 } review_object = o { o := input.review.object } review_id = value { value := sprintf("%v/%v/%v", [ review_namespace, review_name, review_kind ]) } review_name = value { value := input.review.object.metadata.name } else = value { value := "NOT_FOUND" } review_namespace = value { value := input.review.object.metadata.namespace } else = value { value := "NOT_FOUND" } review_kind = value { value := input.review.kind.kind } else = value { value := "NOT_FOUND" } rego: | package k8sdepregistry import data.lib.k8s.helpers as helpers violation[{"msg": msg, "details": {}}] { helpers.review_operation = input.parameters.allowedOps[_] helpers.review_kind = "Service" helpers.review_object.spec.type = "ClusterIP" helpers.review_object.spec.externalIPs aips := input.parameters.allowedIps ips := helpers.review_object.spec.externalIPs helpers.ips_allowed(aips,ips) msg = sprintf("%v: ClusterIP service external IPs are not found in the Allowed IPs list. Allowed IPs: %v, Submitted IPs: %v. Resource ID (ns/name/kind): %v", [input.parameters.errMsg,aips,ips,helpers.review_id]) }