From 93407bcecbd6aa825d8ac3ac642a3cd91dd66c68 Mon Sep 17 00:00:00 2001 From: Flavio Castelli Date: Mon, 23 Oct 2023 16:46:11 +0200 Subject: [PATCH] feat: allow Rego policies to be context aware Allow Rego policies, both OPA and Gatekeeper ones, to leverage Kubernetes data at evaluation time. This change doesn't break the API of policy-evaluator, nor requires any special action by the consumers of this crate. Signed-off-by: Flavio Castelli --- src/policy_evaluator.rs | 13 +- src/policy_evaluator_builder.rs | 4 +- src/runtimes/rego/context_aware.rs | 255 ++++++++++++++++++++ src/runtimes/rego/gatekeeper_inventory.rs | 281 ++++++++++++++++++++++ src/runtimes/rego/mod.rs | 6 +- src/runtimes/rego/opa_inventory.rs | 274 +++++++++++++++++++++ src/runtimes/rego/runtime.rs | 41 ++-- src/runtimes/rego/stack.rs | 55 +++++ 8 files changed, 909 insertions(+), 20 deletions(-) create mode 100644 src/runtimes/rego/context_aware.rs create mode 100644 src/runtimes/rego/gatekeeper_inventory.rs create mode 100644 src/runtimes/rego/opa_inventory.rs create mode 100644 src/runtimes/rego/stack.rs diff --git a/src/policy_evaluator.rs b/src/policy_evaluator.rs index 393f4dce..e616895c 100644 --- a/src/policy_evaluator.rs +++ b/src/policy_evaluator.rs @@ -105,7 +105,18 @@ impl Evaluator for PolicyEvaluator { WapcRuntime(wapc_host).validate(&self.settings, &request) } Runtime::Burrego(ref mut burrego_evaluator) => { - BurregoRuntime(burrego_evaluator).validate(&self.settings, &request) + let kube_ctx = burrego_evaluator.build_kubernetes_context( + self.policy.callback_channel.as_ref(), + &self.policy.ctx_aware_resources_allow_list, + ); + match kube_ctx { + Ok(ctx) => { + BurregoRuntime(burrego_evaluator).validate(&self.settings, &request, &ctx) + } + Err(e) => { + AdmissionResponse::reject(request.uid().to_string(), e.to_string(), 500) + } + } } Runtime::Cli(ref mut cli_stack) => { WasiRuntime(cli_stack).validate(&self.settings, &request) diff --git a/src/policy_evaluator_builder.rs b/src/policy_evaluator_builder.rs index f66a1253..cf461b1d 100644 --- a/src/policy_evaluator_builder.rs +++ b/src/policy_evaluator_builder.rs @@ -279,8 +279,8 @@ impl PolicyEvaluatorBuilder { PolicyExecutionMode::Opa | PolicyExecutionMode::OpaGatekeeper => { let policy = Self::from_contents_internal( self.policy_id.clone(), - None, // callback_channel is not used by Rego policies - None, + self.callback_channel.clone(), + Some(self.ctx_aware_resources_allow_list.clone()), || None, Policy::new, execution_mode, diff --git a/src/runtimes/rego/context_aware.rs b/src/runtimes/rego/context_aware.rs new file mode 100644 index 00000000..ade88888 --- /dev/null +++ b/src/runtimes/rego/context_aware.rs @@ -0,0 +1,255 @@ +use std::collections::{HashMap, HashSet}; + +use crate::{ + callback_requests::{CallbackRequest, CallbackRequestType, CallbackResponse}, + policy_metadata::ContextAwareResource, + runtimes::rego::{gatekeeper_inventory::GatekeeperInventory, opa_inventory::OpaInventory}, +}; +use anyhow::{anyhow, Result}; +use kube::api::ObjectList; +use tokio::sync::{mpsc, oneshot}; + +#[derive(serde::Serialize)] +pub(crate) struct EmptyContext(serde_json::Value); + +impl Default for EmptyContext { + fn default() -> Self { + EmptyContext(serde_json::Value::Null) + } +} + +#[derive(serde::Serialize)] +pub(crate) enum KubernetesContext { + Empty(EmptyContext), + Opa(OpaInventory), + Gatekeeper(GatekeeperInventory), +} + +/// Uses the callback channel to get all the Kubernetes resources defined inside of +/// the cluster whose type is mentioned inside of `allowed_resources`. +/// +/// The resources are returned based on the actual RBAC privileges of the client +/// used by the runtime. +pub(crate) fn get_allowed_resources( + callback_channel: &mpsc::Sender, + allowed_resources: &HashSet, +) -> Result>> { + let mut kube_resources: HashMap> = + HashMap::new(); + + for resource in allowed_resources { + let resource_list = get_all_resources_by_type(callback_channel, resource)?; + kube_resources.insert(resource.to_owned(), resource_list); + } + + Ok(kube_resources) +} + +fn get_all_resources_by_type( + callback_channel: &mpsc::Sender, + resource_type: &ContextAwareResource, +) -> Result> { + let req_type = CallbackRequestType::KubernetesListResourceAll { + api_version: resource_type.api_version.to_owned(), + kind: resource_type.kind.to_owned(), + label_selector: None, + field_selector: None, + }; + + let response = make_request_via_callback_channel(req_type, callback_channel)?; + serde_json::from_slice::>(&response.payload).map_err( + |e| anyhow!("cannot convert callback response into a list of kubernetes objects: {e}"), + ) +} + +/// Creates a map that has ContextAwareResource as key, and its plural name as value. +/// For example, the key for {`apps/v1`, `Deployment`} will have `deployments` as value. +/// The map is built by making request via the given callback channel. +pub(crate) fn get_plural_names( + callback_channel: &mpsc::Sender, + allowed_resources: &HashSet, +) -> Result> { + let mut plural_names_by_resource: HashMap = HashMap::new(); + + for resource in allowed_resources { + let req_type = CallbackRequestType::KubernetesGetResourcePluralName { + api_version: resource.api_version.to_owned(), + kind: resource.kind.to_owned(), + }; + + let response = make_request_via_callback_channel(req_type, callback_channel)?; + let plural_name = serde_json::from_slice::(&response.payload).map_err(|e| { + anyhow!("get plural name failure, cannot convert callback response: {e}") + })?; + + plural_names_by_resource.insert(resource.to_owned(), plural_name); + } + + Ok(plural_names_by_resource) +} + +/// Internal helper function that sends a request over the callback channel and returns the +/// response +fn make_request_via_callback_channel( + request_type: CallbackRequestType, + callback_channel: &mpsc::Sender, +) -> Result { + let (tx, mut rx) = oneshot::channel::>(); + let req = CallbackRequest { + request: request_type, + response_channel: tx, + }; + callback_channel + .try_send(req) + .map_err(|e| anyhow!("error sending request over callback channel: {e}"))?; + + loop { + // Note: we cannot use `rx.blocking_recv`. The code would compile, but at runtime we would + // have a panic because this function is used inside of an async block. The `blocking_recv` + // method causes the tokio reactor to stop, which leads to a panic + match rx.try_recv() { + Ok(msg) => return msg, + Err(oneshot::error::TryRecvError::Empty) => { + // do nothing, keep waiting for a reply + } + Err(e) => { + return Err(anyhow!( + "error obtaining response from callback channel: {e}" + )); + } + } + } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + use assert_json_diff::assert_json_eq; + use serde_json::json; + use std::path::Path; + + pub fn dynamic_object_from_fixture( + resource_type: &str, + namespace: Option<&str>, + name: &str, + ) -> Result { + let path = Path::new("test_data/fixtures/kube_context") + .join(resource_type) + .join(namespace.unwrap_or_default()) + .join(format!("{name}.json")); + let contents = std::fs::read(path.clone()) + .map_err(|e| anyhow!("canont read fixture from path: {path:?}: {e}"))?; + serde_json::from_slice::(&contents) + .map_err(|e| anyhow!("json conversion error: {e}")) + } + + pub fn object_list_from_dynamic_objects( + objs: &[kube::core::DynamicObject], + ) -> Result> { + let raw_json = json!( + { + "items": objs, + "metadata": { + "resourceVersion": "" + } + } + ); + + let res: ObjectList = serde_json::from_value(raw_json) + .map_err(|e| anyhow!("cannot create ObjectList because of json error: {e}"))?; + Ok(res) + } + + #[tokio::test(flavor = "multi_thread")] + async fn get_all_resources_success() { + let (callback_tx, mut callback_rx) = mpsc::channel::(10); + let resource = ContextAwareResource { + api_version: "v1".to_string(), + kind: "Service".to_string(), + }; + let expected_resource = resource.clone(); + let services = [ + dynamic_object_from_fixture("services", Some("kube-system"), "kube-dns").unwrap(), + dynamic_object_from_fixture("services", Some("kube-system"), "metrics-server").unwrap(), + ]; + let services_list = object_list_from_dynamic_objects(&services).unwrap(); + + tokio::spawn(async move { + let req = match callback_rx.recv().await { + Some(r) => r, + None => return, + }; + match req.request { + CallbackRequestType::KubernetesListResourceAll { + api_version, + kind, + label_selector, + field_selector, + } => { + assert_eq!(api_version, expected_resource.api_version); + assert_eq!(kind, expected_resource.kind); + assert!(label_selector.is_none()); + assert!(field_selector.is_none()); + } + _ => { + panic!("not the expected request type"); + } + }; + + let services_list = object_list_from_dynamic_objects(&services).unwrap(); + let callback_response = CallbackResponse { + payload: serde_json::to_vec(&services_list).unwrap(), + }; + + req.response_channel.send(Ok(callback_response)).unwrap(); + }); + + let actual = get_all_resources_by_type(&callback_tx, &resource).unwrap(); + let actual_json = serde_json::to_value(&actual).unwrap(); + let expected_json = serde_json::to_value(&services_list).unwrap(); + assert_json_eq!(actual_json, expected_json); + } + + #[tokio::test(flavor = "multi_thread")] + async fn get_resource_plural_name_success() { + let (callback_tx, mut callback_rx) = mpsc::channel::(10); + let resource = ContextAwareResource { + api_version: "v1".to_string(), + kind: "Service".to_string(), + }; + let plural_name = "services"; + + let mut resources: HashSet = HashSet::new(); + resources.insert(resource.clone()); + + let mut expected_names: HashMap = HashMap::new(); + expected_names.insert(resource.clone(), plural_name.to_string()); + + let expected_resource = resource.clone(); + + tokio::spawn(async move { + let req = match callback_rx.recv().await { + Some(r) => r, + None => return, + }; + match req.request { + CallbackRequestType::KubernetesGetResourcePluralName { api_version, kind } => { + assert_eq!(api_version, expected_resource.api_version); + assert_eq!(kind, expected_resource.kind); + } + _ => { + panic!("not the expected request type"); + } + }; + + let callback_response = CallbackResponse { + payload: serde_json::to_vec(&plural_name).unwrap(), + }; + + req.response_channel.send(Ok(callback_response)).unwrap(); + }); + + let actual = get_plural_names(&callback_tx, &resources).unwrap(); + assert_eq!(actual, expected_names); + } +} diff --git a/src/runtimes/rego/gatekeeper_inventory.rs b/src/runtimes/rego/gatekeeper_inventory.rs new file mode 100644 index 00000000..f82b9304 --- /dev/null +++ b/src/runtimes/rego/gatekeeper_inventory.rs @@ -0,0 +1,281 @@ +/// This file defins structs that contain the kubernetes context aware data in a format that is +/// compatible with what Gatekeeper expects. +/// +/// If you don't care about the process, jump straight to the section about the inventory. +/// +/// ## The `inventory` object +/// +/// As documented [here](https://open-policy-agent.github.io/gatekeeper/website/docs/sync/), the Kubernetes details are made available to all the policies via the `data.inventory` object. +/// +/// Inventory is a JSON dictionary built using the rules mentioned by the doc: +/// - For cluster-scoped objects: `data.inventory.cluster[][][]` +/// - For namespace-scoped objects: `data.inventory.namespace[][groupVersion][][]` +/// +/// For example, all the `Namespace` objects are exposed this way: +/// +/// ```hcl +/// "cluster": { # that's because Namespace is a cluster wide resournce +/// "v1": { # this is the group version of the `Namespace` resource +/// "Namespace": { # this is the kind used by `Namespace` +/// "default": { # this is the name of the Namespace resource being "dumped" +/// # contents of `kubectl get ns default -o json` +/// }, +/// "kube-system": { # name of the namespace +/// # contents of the namespace object +/// } +/// # more entries... +/// } +/// } +/// } +/// ``` +/// +/// While all the Pods are exposed in this way: +/// +/// ```hcl +/// "namespace": { # this is used for namespaced resources, like `cluster` was used before for cluster-wide ones +/// "gatekeeper-system": { # name of the namespace that contains the Pod +/// "v1": { # the group version of the Pod resource +/// "Pod": { # the kind of the Pod resource +/// "gatekeeper-audit-fd9c6d89d-lrr9d": { # the name of the Pod +/// # contents of `kubectl get pod -n gatekeeper-system -o json gatekeeper-audit-fd9c6d89d-lrr9d` +/// } +/// # the other pods defined inside of the `gatekeeper-system` namespace are shown here +/// } +/// }, +/// "default": { # all the pods defined under the `default` namespace +/// "v1": { +/// "Pod": { +/// "foo": { +/// # definition of the `foo` pod, defined under the `default` namespace +/// } +/// } +/// } +/// } +/// } +/// ``` +/// +use crate::policy_metadata::ContextAwareResource; +use anyhow::{anyhow, Result}; +use kube::api::ObjectList; +use serde::Serialize; +use std::collections::HashMap; + +/// A wrapper around a dictionary that has the resource Name as key, +/// and a DynamicObject as value +#[derive(Serialize, Default)] +pub(crate) struct ResourcesByName(HashMap); + +impl ResourcesByName { + fn register(&mut self, obj: &kube::core::DynamicObject) -> Result<()> { + let name = obj + .metadata + .name + .clone() + .ok_or(anyhow!("DynamicObject does not have name"))?; + self.0.insert(name, obj.to_owned()); + Ok(()) + } +} + +/// A wrapper around a dictionary that has a Kubernetes Kind (e.g. `Pod`) +/// as key, and a ResourcesByName as value +#[derive(Serialize, Default)] +pub(crate) struct ResourcesByKind(HashMap); + +impl ResourcesByKind { + fn register( + &mut self, + obj: &kube::core::DynamicObject, + resource: &ContextAwareResource, + ) -> Result<()> { + self.0 + .entry(resource.kind.clone()) + .or_default() + .register(obj) + } +} + +/// A wrapper around a dictionary that has a Kubernetes GroupVersion (e.g. `apps/v1`) +/// as key, and a ResourcesByKind as value +#[derive(Serialize, Default)] +pub(crate) struct ResourcesByGroupVersion(HashMap); + +impl ResourcesByGroupVersion { + fn register( + &mut self, + obj: &kube::core::DynamicObject, + resource: &ContextAwareResource, + ) -> Result<()> { + self.0 + .entry(resource.api_version.clone()) + .or_default() + .register(obj, resource) + } +} + +/// A wrapper around a dictionary that has +/// the name of a Kubernetes Namespace (e.g. `kube-system`) as key, +/// and a ResourcesByGroupVersion as value +#[derive(Serialize, Default)] +pub(crate) struct ResourcesByNamespace(HashMap); + +impl ResourcesByNamespace { + fn register( + &mut self, + obj: &kube::core::DynamicObject, + resource: &ContextAwareResource, + ) -> Result<()> { + let namespace = obj + .metadata + .namespace + .clone() + .ok_or(anyhow!("DynamicObject does not have a Namespace"))?; + self.0.entry(namespace).or_default().register(obj, resource) + } +} + +/// A struct holding the Kubernetes context aware data in a format that is compabible with what +/// Gatekeeper expects +#[derive(Serialize, Default)] +pub(crate) struct GatekeeperInventory { + #[serde(rename = "cluster")] + cluster_resources: ResourcesByGroupVersion, + #[serde(rename = "namespace")] + namespaced_resources: ResourcesByNamespace, +} + +impl GatekeeperInventory { + /// Creates a GatekeeperInventory by querying a Kubernetes cluster + /// for all the resources specified + pub(crate) fn new( + kube_resources: &HashMap>, + ) -> Result { + let mut inventory = GatekeeperInventory::default(); + + for (resource, resources_list) in kube_resources { + for obj in resources_list { + inventory.register(obj, resource)? + } + } + + Ok(inventory) + } + + fn register( + &mut self, + obj: &kube::core::DynamicObject, + resource: &ContextAwareResource, + ) -> Result<()> { + match &obj.metadata.namespace { + Some(_) => { + // namespaced resource + self.namespaced_resources.register(obj, resource) + } + None => { + // cluster-wide resource + self.cluster_resources.register(obj, resource) + } + } + } +} + +#[cfg(test)] +mod tests { + use crate::runtimes::rego::context_aware::tests::{ + dynamic_object_from_fixture, object_list_from_dynamic_objects, + }; + + use super::*; + use assert_json_diff::assert_json_eq; + + #[test] + fn create() { + let mut kube_resources: HashMap< + ContextAwareResource, + ObjectList, + > = HashMap::new(); + + let services = [ + dynamic_object_from_fixture("services", Some("kube-system"), "kube-dns").unwrap(), + dynamic_object_from_fixture("services", Some("kube-system"), "metrics-server").unwrap(), + ]; + let services_list = object_list_from_dynamic_objects(&services).unwrap(); + kube_resources.insert( + ContextAwareResource { + api_version: "v1".to_string(), + kind: "Service".to_string(), + }, + services_list, + ); + + let deployments = [ + dynamic_object_from_fixture("deployments", Some("ingress"), "ingress-nginx").unwrap(), + dynamic_object_from_fixture("deployments", Some("kube-system"), "coredns").unwrap(), + dynamic_object_from_fixture( + "deployments", + Some("kube-system"), + "local-path-provisioner", + ) + .unwrap(), + ]; + let deployments_list = object_list_from_dynamic_objects(&deployments).unwrap(); + kube_resources.insert( + ContextAwareResource { + api_version: "apps/v1".to_string(), + kind: "Deployment".to_string(), + }, + deployments_list, + ); + + let namespaces = [ + dynamic_object_from_fixture("namespaces", None, "cert-manager").unwrap(), + dynamic_object_from_fixture("namespaces", None, "kube-system").unwrap(), + ]; + let namespaces_list = object_list_from_dynamic_objects(&namespaces).unwrap(); + kube_resources.insert( + ContextAwareResource { + api_version: "v1".to_string(), + kind: "Namespace".to_string(), + }, + namespaces_list, + ); + + let expected = serde_json::json!({ + "cluster": { + "v1": { + "Namespace": { + "kube-system": dynamic_object_from_fixture("namespaces", None, "kube-system").unwrap(), + "cert-manager": dynamic_object_from_fixture("namespaces", None, "cert-manager").unwrap(), + } + } + }, + "namespace": { + "kube-system": { + "v1": { + "Service": { + "kube-dns": dynamic_object_from_fixture("services", Some("kube-system"), "kube-dns").unwrap(), + "metrics-server": dynamic_object_from_fixture("services", Some("kube-system"), "metrics-server").unwrap(), + } + }, + "apps/v1": { + "Deployment": { + "coredns": dynamic_object_from_fixture("deployments", Some("kube-system"), "coredns").unwrap(), + "local-path-provisioner": dynamic_object_from_fixture("deployments", Some("kube-system"), "local-path-provisioner").unwrap(), + } + } + }, + "ingress": { + "apps/v1": { + "Deployment": { + "ingress-nginx": dynamic_object_from_fixture("deployments", Some("ingress"), "ingress-nginx").unwrap(), + } + } + } + } + }); + + let inventory = GatekeeperInventory::new(&kube_resources).unwrap(); + let inventory_json = serde_json::to_value(&inventory).unwrap(); + assert_json_eq!(inventory_json, expected); + } +} diff --git a/src/runtimes/rego/mod.rs b/src/runtimes/rego/mod.rs index deef1a8f..a33e086e 100644 --- a/src/runtimes/rego/mod.rs +++ b/src/runtimes/rego/mod.rs @@ -1,8 +1,12 @@ +mod context_aware; +mod gatekeeper_inventory; +mod opa_inventory; mod runtime; +mod stack; use burrego::host_callbacks::HostCallbacks; -pub(crate) use runtime::BurregoStack; pub(crate) use runtime::Runtime; +pub(crate) use stack::BurregoStack; #[tracing::instrument(level = "error")] fn opa_abort(msg: &str) {} diff --git a/src/runtimes/rego/opa_inventory.rs b/src/runtimes/rego/opa_inventory.rs new file mode 100644 index 00000000..b388ce3f --- /dev/null +++ b/src/runtimes/rego/opa_inventory.rs @@ -0,0 +1,274 @@ +/// This file builds a the context data required by OPA polices. +/// ## Docs references +/// +/// We define OPA policies the ones deployed via [kube-mgmt](https://github.com/open-policy-agent/kube-mgmt), which is an alternative to gatekeeper. +/// +/// kube-mgmt can be configured to expose Kubernetes resources to the policies. This is described in detail [here](https://github.com/open-policy-agent/kube-mgmt#caching). +/// +/// By default, the Kubernetes information are made available to the policies inside of `data.kubernetes`. The `kubernetes` key can be changed by the user via a configuration flag, but I think we can ignore this detail and conform to the default behavior. +/// +/// ## Incoming JSON structure +/// +/// Kubernetes resources are exposed using this format: +/// +/// * namespaced resources: `kubernetes...` +/// * cluster wide resources: `kubernetes..` +/// +/// It's important to point out that `` is the Kubernetes name of the resource obtained when doing: +/// +/// ```console +/// kubectl api-resources +/// ``` +/// +/// For example, the name of `v1/Service` is `services`. +/// +/// This is problematic, because the name of a resource isn't unique. For example: +/// +/// ```console +/// kubectl api-resources | grep events +/// events ev v1 true Event +/// events ev events.k8s.io/v1 true Event +/// ``` +/// +/// The problem might become even more evident when multiple CRDs are installed. +/// +/// However, this is is not our problem... Moreover, the admin decides what has to be shared with the policies. Hence he can pick, among the duplicates, which resource to share with the policies. +/// +/// ### Examples +/// +/// This is how the `data` payload would look like when Kubernetes Service and Namespace resources are shared with policies: +/// +/// ```hcl +/// { +/// "kubernetes": { # the default key +/// "services": { # the name of the resource +/// "default": { # the namespace inside of which the resources are defined +/// "example-service": { # the name of the Service +/// # the contents of `kubectl get svc -n default -o json example-service` +/// }, +/// "another-service": { +/// # the contents of `kubectl get svc -n default -o json another-service` +/// } +/// } +/// }, +/// "namespaces": { # the name of the resource - note: this is a cluster-wide resource +/// "default": { +/// # contents of `kubectl get ns default -o json` +/// }, +/// "kube-system": { +/// # contents of `kubectl get ns kube-system -o json` +/// } +/// } +/// } +/// } +/// ``` +/// +use crate::policy_metadata::ContextAwareResource; +use anyhow::{anyhow, Result}; +use kube::api::ObjectList; +use serde::Serialize; +use std::collections::HashMap; + +/// A wrapper around a dictionary that has the resource Name as key, +/// and a DynamicObject as value +#[derive(Serialize, Default)] +pub(crate) struct ResourcesByName(HashMap); + +impl ResourcesByName { + fn register(&mut self, obj: &kube::core::DynamicObject) -> Result<()> { + let name = obj + .metadata + .name + .clone() + .ok_or(anyhow!("DynamicObject does not have name"))?; + self.0.insert(name, obj.to_owned()); + Ok(()) + } +} + +#[derive(Serialize, Default)] +pub(crate) struct ResourcesByNamespace(HashMap); + +impl ResourcesByNamespace { + fn register(&mut self, obj: &kube::core::DynamicObject) -> Result<()> { + let namespace = obj + .metadata + .namespace + .clone() + .ok_or(anyhow!("DynamicObject does not have a Namespace"))?; + self.0.entry(namespace).or_default().register(obj) + } +} + +#[derive(Serialize)] +#[serde(untagged)] +pub(crate) enum ResourcesByScope { + Cluster(ResourcesByName), + Namespace(ResourcesByNamespace), +} + +impl ResourcesByScope { + fn register(&mut self, obj: &kube::core::DynamicObject) -> Result<()> { + match self { + ResourcesByScope::Cluster(cluster_resources) => cluster_resources.register(obj), + ResourcesByScope::Namespace(namespace_resources) => namespace_resources.register(obj), + } + } +} + +/// A wrapper around a dictionary that has +/// the plural name of a Kubernetes resource (e.g. `services`) as key, +/// and a ResourcesByScope as value +#[derive(Serialize, Default)] +pub(crate) struct ResourcesByPluralName(HashMap); + +impl ResourcesByPluralName { + fn register(&mut self, obj: &kube::core::DynamicObject, plural_name: &str) -> Result<()> { + let obj_namespaced = obj.metadata.namespace.is_some(); + + match self.0.get_mut(plural_name) { + Some(ref mut resources_by_scope) => { + match resources_by_scope { + ResourcesByScope::Cluster(_) => { + if obj_namespaced { + return Err(anyhow!("trying to add a namespaced resource to a list of clusterwide resources")); + } + } + ResourcesByScope::Namespace(_) => { + if !obj_namespaced { + return Err(anyhow!("trying to add a clusterwide resource to a list of namespaced resources")); + } + } + } + resources_by_scope.register(obj) + } + None => { + let mut resources_by_scope = if obj_namespaced { + ResourcesByScope::Namespace(ResourcesByNamespace::default()) + } else { + ResourcesByScope::Cluster(ResourcesByName::default()) + }; + resources_by_scope.register(obj)?; + self.0.insert(plural_name.to_owned(), resources_by_scope); + Ok(()) + } + } + } +} + +#[derive(Serialize, Default)] +pub(crate) struct OpaInventory(ResourcesByPluralName); + +impl OpaInventory { + /// Creates a GatekeeperInventory by querying a Kubernetes cluster + /// for all the resources specified + pub(crate) fn new( + kube_resources: &HashMap>, + plural_names: &HashMap, + ) -> Result { + let mut inventory = OpaInventory::default(); + + for (resource, resources_list) in kube_resources { + let plural_name = plural_names + .get(resource) + .ok_or_else(|| anyhow!("cannot find plural name for resource {resource:?}"))?; + + for obj in resources_list { + inventory.register(obj, plural_name)? + } + } + + Ok(inventory) + } + + fn register(&mut self, obj: &kube::core::DynamicObject, plural_name: &str) -> Result<()> { + self.0.register(obj, plural_name) + } +} + +#[cfg(test)] +mod tests { + use crate::runtimes::rego::context_aware::tests::{ + dynamic_object_from_fixture, object_list_from_dynamic_objects, + }; + + use super::*; + use assert_json_diff::assert_json_eq; + + #[test] + fn create() { + let mut kube_resources: HashMap< + ContextAwareResource, + ObjectList, + > = HashMap::new(); + let mut plural_names: HashMap = HashMap::new(); + + let services = [ + dynamic_object_from_fixture("services", Some("kube-system"), "kube-dns").unwrap(), + dynamic_object_from_fixture("services", Some("kube-system"), "metrics-server").unwrap(), + ]; + let services_list = object_list_from_dynamic_objects(&services).unwrap(); + let ctx_aware_resource = ContextAwareResource { + api_version: "v1".to_string(), + kind: "Service".to_string(), + }; + plural_names.insert(ctx_aware_resource.clone(), "services".to_string()); + kube_resources.insert(ctx_aware_resource, services_list); + + let deployments = [ + dynamic_object_from_fixture("deployments", Some("ingress"), "ingress-nginx").unwrap(), + dynamic_object_from_fixture("deployments", Some("kube-system"), "coredns").unwrap(), + dynamic_object_from_fixture( + "deployments", + Some("kube-system"), + "local-path-provisioner", + ) + .unwrap(), + ]; + let deployments_list = object_list_from_dynamic_objects(&deployments).unwrap(); + let ctx_aware_resource = ContextAwareResource { + api_version: "apps/v1".to_string(), + kind: "Deployment".to_string(), + }; + plural_names.insert(ctx_aware_resource.clone(), "deployments".to_string()); + kube_resources.insert(ctx_aware_resource, deployments_list); + + let namespaces = [ + dynamic_object_from_fixture("namespaces", None, "cert-manager").unwrap(), + dynamic_object_from_fixture("namespaces", None, "kube-system").unwrap(), + ]; + let namespaces_list = object_list_from_dynamic_objects(&namespaces).unwrap(); + let ctx_aware_resource = ContextAwareResource { + api_version: "v1".to_string(), + kind: "Namespace".to_string(), + }; + plural_names.insert(ctx_aware_resource.clone(), "namespaces".to_string()); + kube_resources.insert(ctx_aware_resource, namespaces_list); + + let expected = serde_json::json!({ + "namespaces": { + "kube-system": dynamic_object_from_fixture("namespaces", None, "kube-system").unwrap(), + "cert-manager": dynamic_object_from_fixture("namespaces", None, "cert-manager").unwrap(), + }, + "services": { + "kube-system": { + "kube-dns": dynamic_object_from_fixture("services", Some("kube-system"), "kube-dns").unwrap(), + "metrics-server": dynamic_object_from_fixture("services", Some("kube-system"), "metrics-server").unwrap(), + }, + }, + "deployments": { + "kube-system": { + "coredns": dynamic_object_from_fixture("deployments", Some("kube-system"), "coredns").unwrap(), + "local-path-provisioner": dynamic_object_from_fixture("deployments", Some("kube-system"), "local-path-provisioner").unwrap(), + }, + "ingress": { + "ingress-nginx": dynamic_object_from_fixture("deployments", Some("ingress"), "ingress-nginx").unwrap(), + } + } + }); + + let inventory = OpaInventory::new(&kube_resources, &plural_names).unwrap(); + let inventory_json = serde_json::to_value(&inventory).unwrap(); + assert_json_eq!(inventory_json, expected); + } +} diff --git a/src/runtimes/rego/runtime.rs b/src/runtimes/rego/runtime.rs index d7e7c832..13e3ee44 100644 --- a/src/runtimes/rego/runtime.rs +++ b/src/runtimes/rego/runtime.rs @@ -2,17 +2,12 @@ use anyhow::anyhow; use kubewarden_policy_sdk::settings::SettingsValidationResponse; use serde::Deserialize; use serde_json::json; -use tracing::error; +use tracing::{error, warn}; use crate::admission_response::{AdmissionResponse, AdmissionResponseStatus}; use crate::policy_evaluator::RegoPolicyExecutionMode; use crate::policy_evaluator::{PolicySettings, ValidateRequest}; - -pub(crate) struct BurregoStack { - pub evaluator: burrego::Evaluator, - pub entrypoint_id: i32, - pub policy_execution_mode: RegoPolicyExecutionMode, -} +use crate::runtimes::rego::{context_aware, BurregoStack}; pub(crate) struct Runtime<'a>(pub(crate) &'a mut BurregoStack); @@ -21,6 +16,7 @@ impl<'a> Runtime<'a> { &mut self, settings: &PolicySettings, request: &ValidateRequest, + ctx_data: &context_aware::KubernetesContext, ) -> AdmissionResponse { let uid = request.uid(); @@ -30,14 +26,27 @@ impl<'a> Runtime<'a> { // Policies for OPA expect the whole `AdmissionReview` // object: produce a synthetic external one so // existing OPA policies are compatible. - ( - json!({ - "apiVersion": "admission.k8s.io/v1", - "kind": "AdmissionReview", - "request": &request.0, - }), - json!(settings), - ) + let input = json!({ + "apiVersion": "admission.k8s.io/v1", + "kind": "AdmissionReview", + "request": &request.0, + }); + + // OPA data seems to be free-form, except for the + // Kubernetes context aware data that must be under the + // `kubernetes` key + // We don't know the data that is provided by the users via + // their settings, hence set the context aware data last, to + // ensure we overwrite what a user might have set. + let mut data = settings.clone(); + if data + .insert("kubernetes".to_string(), json!(ctx_data)) + .is_some() + { + warn!("OPA policy had user provided setting with key `kubernnetes`. This value has been overwritten with the actual kubernetes context data"); + } + + (input, json!(data)) } RegoPolicyExecutionMode::Gatekeeper => { // Gatekeeper policies include a toplevel `review` @@ -50,7 +59,7 @@ impl<'a> Runtime<'a> { "parameters": settings, "review": &request.0, }), - json!({"kubernetes": ""}), // TODO (ereslibre): Kubernetes context goes here + json!({"inventory": ctx_data}), ) } }; diff --git a/src/runtimes/rego/stack.rs b/src/runtimes/rego/stack.rs new file mode 100644 index 00000000..861b1484 --- /dev/null +++ b/src/runtimes/rego/stack.rs @@ -0,0 +1,55 @@ +use crate::{ + callback_requests::CallbackRequest, + policy_evaluator::RegoPolicyExecutionMode, + policy_metadata::ContextAwareResource, + runtimes::rego::{ + context_aware, gatekeeper_inventory::GatekeeperInventory, opa_inventory::OpaInventory, + }, +}; +use anyhow::{anyhow, Result}; +use std::collections::HashSet; +use tokio::sync::mpsc; + +pub(crate) struct BurregoStack { + pub evaluator: burrego::Evaluator, + pub entrypoint_id: i32, + pub policy_execution_mode: RegoPolicyExecutionMode, +} + +impl BurregoStack { + pub fn build_kubernetes_context( + &self, + callback_channel: Option<&mpsc::Sender>, + ctx_aware_resources_allow_list: &HashSet, + ) -> Result { + if ctx_aware_resources_allow_list.is_empty() { + return Ok(context_aware::KubernetesContext::Empty( + context_aware::EmptyContext::default(), + )); + } + + match callback_channel { + None => Err(anyhow!( + "cannot build Rego context aware data: callback channel is not set" + )), + Some(chan) => { + let cluster_resources = + context_aware::get_allowed_resources(chan, ctx_aware_resources_allow_list)?; + + match self.policy_execution_mode { + RegoPolicyExecutionMode::Opa => { + let plural_names_by_resource = + context_aware::get_plural_names(chan, ctx_aware_resources_allow_list)?; + let inventory = + OpaInventory::new(&cluster_resources, &plural_names_by_resource)?; + Ok(context_aware::KubernetesContext::Opa(inventory)) + } + RegoPolicyExecutionMode::Gatekeeper => { + let inventory = GatekeeperInventory::new(&cluster_resources)?; + Ok(context_aware::KubernetesContext::Gatekeeper(inventory)) + } + } + } + } + } +}