Admission Controllers is what you are looking for.
Admission controllers intercept operations to validate what should happen before the operation is committed by the api-server.
An example is the ImagePolicyWebhook, an admission controller that intercept Image operations to validate if it should be allowed or rejected.
It will make a call to an REST endpoint with a payload like:
{
"apiVersion":"imagepolicy.k8s.io/v1alpha1",
"kind":"ImageReview",
"spec":{
"containers":[
{
"image":"myrepo/myimage:v1"
},
{
"image":"myrepo/myimage@sha256:beb6bd6a68f114c1dc2ea4b28db81bdf91de202a9014972bec5e4d9171d90ed"
}
],
"annotations":[
"mycluster.image-policy.k8s.io/ticket-1234": "break-glass"
],
"namespace":"mynamespace"
}
}
and the API answer with Allowed:
{
"apiVersion": "imagepolicy.k8s.io/v1alpha1",
"kind": "ImageReview",
"status": {
"allowed": true
}
}
or Rejected:
{
"apiVersion": "imagepolicy.k8s.io/v1alpha1",
"kind": "ImageReview",
"status": {
"allowed": false,
"reason": "image currently blacklisted"
}
}
The endpoint could be a Lambda function or a container running in the cluster.
This github repo github.com/flavio/kube-image-bouncer implements a sample using ImagePolicyWebhook to reject containers using the tag "Latest".
There is also the option to use the flag registry-whitelist
on startup to a pass a comma separated list of allowed registries, this will be used by the ValidatingAdmissionWebhook to validate if the registry is whitelisted.
.
The other alternative is the project Open Policy Agent[OPA].
OPA is a flexible engine used to create policies based on rules to match resources and take decisions according to the result of these expressions. It is a mutating and a validating webhook that gets called for matching Kubernetes API server requests by the admission controller mentioned above. In summary, the operation would work similarly as described above, the only difference is that the rules are written as configuration instead of code. The same example above rewritter to use OPA would be similar to this:
package admission
import data.k8s.matches
deny[{
"id": "container-image-whitelist", # identifies type of violation
"resource": {
"kind": "pods", # identifies kind of resource
"namespace": namespace, # identifies namespace of resource
"name": name # identifies name of resource
},
"resolution": {"message": msg}, # provides human-readable message to display
}] {
matches[["pods", namespace, name, matched_pod]]
container = matched_pod.spec.containers[_]
not re_match("^registry.acmecorp.com/.+$", container.image) # The actual validation
msg := sprintf("invalid container registry image %q", [container.image])
}
The above translates to: deny any pod where the container image does not match the following registry registry.acmecorp.com