Collecting together Kubernetes rego examples, including porting the https://kubesec.io rules to rego

kubernetes.rego

package kubernetes

name = input.metadata.name

kind = input.kind

is_service {
	kind = "Service"
}

is_deployment {
	kind = "Deployment"
}

is_pod {
	kind = "Pod"
}

split_image(image) = [image, "latest"] {
	not contains(image, ":")
}

split_image(image) = [image_name, tag] {
	[image_name, tag] = split(image, ":")
}

pod_containers(pod) = all_containers {
	keys = {"containers", "initContainers"}
	all_containers = [c | keys[k]; c = pod.spec[k][_]]
}

containers[container] {
	pods[pod]
	all_containers = pod_containers(pod)
	container = all_containers[_]
}

containers[container] {
	all_containers = pod_containers(input)
	container = all_containers[_]
}

pods[pod] {
	is_deployment
	pod = input.spec.template
}

pods[pod] {
	is_pod
	pod = input
}

volumes[volume] {
	pods[pod]
	volume = pod.spec.volumes[_]
}

dropped_capability(container, cap) {
	container.securityContext.capabilities.drop[_] == cap
}

added_capability(container, cap) {
	container.securityContext.capabilities.add[_] == cap
}

labels.rego

package main

import data.kubernetes

labels {
	input.metadata.labels["app.kubernetes.io/name"]
	input.metadata.labels["app.kubernetes.io/instance"]
	input.metadata.labels["app.kubernetes.io/version"]
	input.metadata.labels["app.kubernetes.io/component"]
	input.metadata.labels["app.kubernetes.io/part-of"]
	input.metadata.labels["app.kubernetes.io/managed-by"]
}

deny[msg] {
	kubernetes.is_deployment
	not labels
	msg = sprintf("%s must include Kubernetes recommended labels: https://kubernetes.io/docs/concepts/overview/working-with-objects/common-labels/#labels ", [kubernetes.name])
}

security.rego

package main

import data.kubernetes

deny[msg] {
	kubernetes.containers[container]
	[image_name, "latest"] = kubernetes.split_image(container.image)
	msg = sprintf("%s in the %s %s has an image, %s, using the latest tag", [container.name, kubernetes.kind, image_name, kubernetes.name])
}

# https://kubesec.io/basics/containers-resources-limits-memory
deny[msg] {
	kubernetes.containers[container]
	not container.resources.limits.memory
	msg = sprintf("%s in the %s %s does not have a memory limit set", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/containers-resources-limits-cpu/
deny[msg] {
	kubernetes.containers[container]
	not container.resources.limits.cpu
	msg = sprintf("%s in the %s %s does not have a CPU limit set", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/containers-securitycontext-capabilities-add-index-sys-admin/
deny[msg] {
	kubernetes.containers[container]
	kubernetes.added_capability(container, "CAP_SYS_ADMIN")
	msg = sprintf("%s in the %s %s has SYS_ADMIN capabilties", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/containers-securitycontext-capabilities-drop-index-all/
deny[msg] {
	kubernetes.containers[container]
	not kubernetes.dropped_capability(container, "all")
	msg = sprintf("%s in the %s %s doesn't drop all capabilities", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/containers-securitycontext-privileged-true/
deny[msg] {
	kubernetes.containers[container]
	container.securityContext.privileged
	msg = sprintf("%s in the %s %s is privileged", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/containers-securitycontext-readonlyrootfilesystem-true/
deny[msg] {
	kubernetes.containers[container]
	not container.securityContext.readOnlyRootFilesystem = true
	msg = sprintf("%s in the %s %s is not using a read only root filesystem", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/containers-securitycontext-runasnonroot-true/
deny[msg] {
	kubernetes.containers[container]
	not container.securityContext.runAsNonRoot = true
	msg = sprintf("%s in the %s %s is running as root", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/containers-securitycontext-runasuser/
deny[msg] {
	kubernetes.containers[container]
	container.securityContext.runAsUser < 10000
	msg = sprintf("%s in the %s %s has a UID of less than 10000", [container.name, kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/spec-hostaliases/
deny[msg] {
	kubernetes.pods[pod]
	pod.spec.hostAliases
	msg = sprintf("The %s %s is managing host aliases", [kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/spec-hostipc/
deny[msg] {
	kubernetes.pods[pod]
	pod.spec.hostIPC
	msg = sprintf("%s %s is sharing the host IPC namespace", [kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/spec-hostnetwork/
deny[msg] {
	kubernetes.pods[pod]
	pod.spec.hostNetwork
	msg = sprintf("The %s %s is connected to the host network", [kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/spec-hostpid/
deny[msg] {
	kubernetes.pods[pod]
	pod.spec.hostPID
	msg = sprintf("The %s %s is sharing the host PID", [kubernetes.kind, kubernetes.name])
}

# https://kubesec.io/basics/spec-volumes-hostpath-path-var-run-docker-sock/
deny[msg] {
	kubernetes.volumes[volume]
	volume.hostpath.path = "/var/run/docker.sock"
	msg = sprintf("The %s %s is mounting the Docker socket", [kubernetes.kind, kubernetes.name])
}

kubernetes.rego:split_image(image) = [name, tag] { ... } does not work, because of clash with previously defined variable name = input.metadata.name.

@dewe thanks, appreciated. Should be fixed I think. I plan on writing tests for these rules when I get a moment and will probably catch a few other issues I'd wager.