policies/opa/gatekeeper/constraint-templates/8-svc-clusterip-ext-ips-allowed.yaml (98 lines of code) (raw):
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])
}