Merge pull request #137 from microsoft/danmihai1/copy-file

Improve the SetPolicy and CopyFile policy behavior

src/agent/src/policy.rs

@@ -4,12 +4,20 @@
 //
 
 use anyhow::{bail, Result};
-use sha2::{Sha256, Digest};
+use nix::sys::stat;
+use protobuf::MessageDyn;
 use serde::{Deserialize, Serialize};
-use crate::slog::Drain;
+use sha2::{Digest, Sha256};
+use slog::Drain;
+use std::ffi::OsStr;
+use std::os::unix::ffi::OsStrExt;
+use std::path::PathBuf;
 use tokio::io::AsyncWriteExt;
 use tokio::time::{sleep, Duration};
 
+use crate::rpc::ttrpc_error;
+use crate::AGENT_POLICY;
+
 static EMPTY_JSON_INPUT: &str = "{\"input\":{}}";
 
 static OPA_DATA_PATH: &str = "/data";
@@ -24,6 +32,81 @@ macro_rules! sl {
     };
 }
 
+async fn allow_request(policy: &mut AgentPolicy, ep: &str, request: &str) -> ttrpc::Result<()> {
+    if !policy.allow_request(ep, request).await {
+        warn!(sl!(), "{ep} is blocked by policy");
+        Err(ttrpc_error(
+            ttrpc::Code::PERMISSION_DENIED,
+            format!("{ep} is blocked by policy"),
+        ))
+    } else {
+        Ok(())
+    }
+}
+
+pub async fn is_allowed(req: &(impl MessageDyn + serde::Serialize)) -> ttrpc::Result<()> {
+    let request = serde_json::to_string(req).unwrap();
+    let mut policy = AGENT_POLICY.lock().await;
+    allow_request(&mut policy, req.descriptor_dyn().name(), &request).await
+}
+
+/// PolicyCopyFileRequest is very similar to CopyFileRequest from src/libs/protocols, except:
+/// - When creating a symbolic link, the symlink_src field is a string representation of the
+///   data bytes vector from CopyFileRequest. It's easier to verify a string compared with
+///   a bytes vector in OPA.
+/// - When not creating a symbolic link, the data bytes field from CopyFileRequest is not
+///   present in PolicyCopyFileRequest, because it might be large and probably unused by OPA.
+#[derive(::serde::Serialize)]
+struct PolicyCopyFileRequest {
+    path: String,
+    file_size: i64,
+    file_mode: u32,
+    dir_mode: u32,
+    uid: i32,
+    gid: i32,
+    offset: i64,
+
+    symlink_src: PathBuf,
+}
+
+pub async fn is_allowed_copy_file(req: &protocols::agent::CopyFileRequest) -> ttrpc::Result<()> {
+    let sflag = stat::SFlag::from_bits_truncate(req.file_mode);
+    let symlink_src = if sflag.contains(stat::SFlag::S_IFLNK) {
+        // The symlink source path
+        PathBuf::from(OsStr::from_bytes(&req.data))
+    } else {
+        // If this CopyFile request is not creating a symlink, remove the incoming data bytes,
+        // to avoid sending large amounts of data to OPA, that is unlikely to be use this data anyway.
+        PathBuf::new()
+    };
+
+    let policy_req = PolicyCopyFileRequest {
+        path: req.path.clone(),
+        file_size: req.file_size,
+        file_mode: req.file_mode,
+        dir_mode: req.dir_mode,
+        uid: req.uid,
+        gid: req.gid,
+        offset: req.offset,
+
+        symlink_src,
+    };
+
+    let request = serde_json::to_string(&policy_req).unwrap();
+    let mut policy = AGENT_POLICY.lock().await;
+    allow_request(&mut policy, "CopyFileRequest", &request).await
+}
+
+pub async fn do_set_policy(req: &protocols::agent::SetPolicyRequest) -> ttrpc::Result<()> {
+    let request = serde_json::to_string(req).unwrap();
+    let mut policy = AGENT_POLICY.lock().await;
+    allow_request(&mut policy, "SetPolicyRequest", &request).await?;
+    policy
+        .set_policy(&req.policy)
+        .await
+        .map_err(|e| ttrpc_error(ttrpc::Code::INVALID_ARGUMENT, e))
+}
+
 /// Example of HTTP response from OPA: {"result":true}
 #[derive(Debug, Serialize, Deserialize)]
 struct AllowResponse {
@@ -128,7 +211,7 @@ impl AgentPolicy {
     }
 
     /// Ask OPA to check if an API call should be allowed or not.
-    pub async fn is_allowed_endpoint(&mut self, ep: &str, request: &str) -> bool {
+    async fn allow_request(&mut self, ep: &str, request: &str) -> bool {
         let post_input = format!("{{\"input\":{request}}}");
         self.log_opa_input(ep, &post_input).await;
         match self.post_query(ep, &post_input).await {

src/agent/src/rpc.rs

@@ -23,7 +23,7 @@ use ttrpc::{
 use anyhow::{anyhow, Context, Result};
 use cgroups::freezer::FreezerState;
 use oci::{LinuxNamespace, Root, Spec};
-use protobuf::{MessageDyn, MessageField};
+use protobuf::MessageField;
 use protocols::agent::{
     AddSwapRequest, AgentDetails, CopyFileRequest, GetIPTablesRequest, GetIPTablesResponse,
     GuestDetailsResponse, Interfaces, Metrics, OOMEvent, ReadStreamResponse, Routes,
@@ -69,7 +69,7 @@ use crate::trace_rpc_call;
 use crate::tracer::extract_carrier_from_ttrpc;
 
 #[cfg(feature = "agent-policy")]
-use crate::AGENT_POLICY;
+use crate::policy::{do_set_policy, is_allowed, is_allowed_copy_file};
 
 use opentelemetry::global;
 use tracing::span;
@@ -123,31 +123,17 @@ fn sl() -> slog::Logger {
 }
 
 // Convenience function to wrap an error and response to ttrpc client
-fn ttrpc_error(code: ttrpc::Code, err: impl Debug) -> ttrpc::Error {
+pub fn ttrpc_error(code: ttrpc::Code, err: impl Debug) -> ttrpc::Error {
     get_rpc_status(code, format!("{:?}", err))
 }
 
 #[cfg(not(feature = "agent-policy"))]
-async fn is_allowed(_req: &(impl MessageDyn + serde::Serialize)) -> ttrpc::Result<()> {
+async fn is_allowed(_req: &impl serde::Serialize) -> ttrpc::Result<()> {
     Ok(())
 }
-
-#[cfg(feature = "agent-policy")]
-async fn is_allowed(req: &(impl MessageDyn + serde::Serialize)) -> ttrpc::Result<()> {
-    let request = serde_json::to_string(req).unwrap();
-    let mut policy = AGENT_POLICY.lock().await;
-    if !policy
-        .is_allowed_endpoint(req.descriptor_dyn().name(), &request)
-        .await
-    {
-        warn!(sl(), "{} is blocked by policy", req.descriptor_dyn().name());
-        Err(ttrpc_error(
-            ttrpc::Code::PERMISSION_DENIED,
-            format!("{} is blocked by policy", req.descriptor_dyn().name()),
-        ))
-    } else {
-        Ok(())
-    }
+#[cfg(not(feature = "agent-policy"))]
+async fn is_allowed_copy_file(_req: &CopyFileRequest) -> ttrpc::Result<()> {
+    Ok(())
 }
 
 fn same<E>(e: E) -> E {
@@ -1340,7 +1326,7 @@ impl agent_ttrpc::AgentService for AgentService {
         req: protocols::agent::CopyFileRequest,
     ) -> ttrpc::Result<Empty> {
         trace_rpc_call!(ctx, "copy_file", req);
-        is_allowed(&req).await?;
+        is_allowed_copy_file(&req).await?;
 
         do_copy_file(&req).map_ttrpc_err(same)?;
 
@@ -1439,14 +1425,8 @@ impl agent_ttrpc::AgentService for AgentService {
         req: protocols::agent::SetPolicyRequest,
     ) -> ttrpc::Result<Empty> {
         trace_rpc_call!(ctx, "set_policy", req);
-        is_allowed(&req).await?;
 
-        AGENT_POLICY
-            .lock()
-            .await
-            .set_policy(&req.policy)
-            .await
-            .map_ttrpc_err(same)?;
+        do_set_policy(&req).await?;
 
         Ok(Empty::new())
     }

src/tools/genpolicy/rules.rego

@@ -1065,9 +1065,23 @@ check_directory_traversal(i_path) {
     endswith(i_path, "/..") == false
 }
 
+check_symlink_source {
+    # TODO: delete this rule once the symlink_src field gets implemented
+    # by all/most Guest VMs.
+    not input.symlink_src
+}
+check_symlink_source {
+    i_src := input.symlink_src
+    print("check_symlink_source: i_src =", i_src)
+
+    startswith(i_src, "/") == false
+    check_directory_traversal(i_src)
+}
+
 CopyFileRequest {
     print("CopyFileRequest: input.path =", input.path)
 
+    check_symlink_source
     check_directory_traversal(input.path)
 
     some regex1 in policy_data.request_defaults.CopyFileRequest