template.yaml

apiVersion: templates.gatekeeper.sh/v1beta1
kind: ConstraintTemplate
metadata:
  name: linkerdmutualtls
  labels:
    policy.linkerd.io: mtls
spec:
  crd:
    spec:
      names:
        kind: LinkerdMutualTLS
        plural: mtls
        singular: mtls
        shortNames: ["mtls"]
      validation:
        openAPIV3Schema:
          properties:
            labels:
              type: array
              items: string
  targets:
  - target: admission.k8s.gatekeeper.sh
    rego: |
      package linkerd.io

      meta = input.review.object.metadata
      spec = input.review.object.spec
      parameters = input.parameters


      violation[{"msg":msg, "details":{}}] {
        containers := spec.containers
        not spec_violation
        msg := "Make sure you run linkerd inject to enable mTLS!"
      }

      spec_violation {
        containers := spec.containers
        proxy(containers, parameters)
        # msg := sprintf("Missing proxy sidecar container: %v", [parameters.proxy.name])
      }

      spec_violation {
        containers := spec.containers
        parameters.metadata.annotations[key] == meta.annotations[key]
        #msg := sprintf("Missing required annotation %v=%v", [key, parameters.metadata.annotations[key]])
      }

      spec_violation {
        containers := spec.containers
        env := { variable |
          proxy := find_proxy(containers, parameters)
          parameters.proxy.env[j].name == proxy.env[k].name
          variable := parameters.proxy.env[j].name
        }
        count(env) > 0
        #msg := sprintf("Missing required environment variable: %v", [parameters.proxy.env[j].name])
      }

      spec_violation {
        volumes := spec.volumes
        find_identity_volume(volumes, parameters)
        #msg := sprintf("Missing required volumes: %v", [parameters.volumes[_].name])
      }

      spec_violation {
        containers := spec.containers
        volumeMount := { name: mountPath |
          proxy := find_proxy(containers, parameters)
          parameters.proxy.volumeMounts[i].name == proxy.volumeMounts[j].name
          parameters.proxy.volumeMounts[i].mountPath == proxy.volumeMounts[j].mountPath
          name :=  parameters.proxy.volumeMounts[i].name
          mountPath := parameters.proxy.volumeMounts[i].mountPath
        }
        count(volumeMount) > 0
        #msg := sprintf("Missing required volume mount: name=%v, mountPath=%v", [parameters.proxy.volumeMounts[i].name, parameters.proxy.volumeMounts[i].mountPath])
      }

      spec_violation {
        spec.automountServiceAccountToken == true
        #msg := "automountServiceAccountToken must be set to 'true' to enable mTLS identity"
      }

      proxy(containers, parameters) {
        spec.containers[_].name == parameters.proxy.name
      }

      find_proxy(containers, parameters) = proxy {
        spec.containers[i].name == parameters.proxy.name
        proxy := spec.containers[i]
      }

      find_identity_volume(volumes, parameters) {
        spec.volumes[_].name == parameters.volumes[_].name
      }