Support relative paths with home directory resolution (#282)

Author: Kimahriman

rust/src/client.rs

@@ -14,6 +14,7 @@ use crate::file::{FileReader, FileWriter};
 use crate::hdfs::protocol::NamenodeProtocol;
 use crate::hdfs::proxy::NameServiceProxy;
 use crate::proto::hdfs::hdfs_file_status_proto::FileType;
+use crate::security::user::User;
 
 use crate::glob::{GlobPattern, expand_glob, get_path_components, unescape_component};
 use crate::proto::hdfs::{ContentSummaryProto, HdfsFileStatusProto};
@@ -118,16 +119,46 @@ impl MountLink {
 struct MountTable {
     mounts: Vec<MountLink>,
     fallback: MountLink,
+    home_dir: String,
 }
 
 impl MountTable {
     fn resolve(&self, src: &str) -> (&MountLink, String) {
+        let path = if src.starts_with('/') {
+            src.to_string()
+        } else {
+            format!("{}/{}", self.home_dir, src)
+        };
+
         for link in self.mounts.iter() {
-            if let Some(resolved) = link.resolve(src) {
+            if let Some(resolved) = link.resolve(&path) {
                 return (link, resolved);
             }
         }
-        (&self.fallback, self.fallback.resolve(src).unwrap())
+        (&self.fallback, self.fallback.resolve(&path).unwrap())
+    }
+}
+
+fn build_home_dir(
+    scheme: &str,
+    host: Option<&str>,
+    config: &Configuration,
+    username: &str,
+) -> String {
+    let prefix = match scheme {
+        "hdfs" => config.get("dfs.user.home.dir.prefix"),
+        "viewfs" => {
+            host.and_then(|host| config.get(&format!("fs.viewfs.mounttable.{host}.homedir")))
+        }
+        _ => None,
+    }
+    .unwrap_or("/user");
+
+    let prefix = prefix.trim_end_matches('/');
+    if prefix.is_empty() {
+        format!("/{username}")
+    } else {
+        format!("{prefix}/{username}")
     }
 }
 
@@ -334,6 +365,19 @@ impl Client {
 
         let rt_holder = RuntimeHolder::new(rt);
 
+        let user_info = User::get_user_info();
+        let username = user_info
+            .effective_user
+            .as_deref()
+            .or(user_info.real_user.as_deref())
+            .expect("User info must include a username");
+        let home_dir = build_home_dir(
+            resolved_url.scheme(),
+            resolved_url.host_str(),
+            config.as_ref(),
+            username,
+        );
+
         let mount_table = match url.scheme() {
             "hdfs" => {
                 let proxy = NameServiceProxy::new(
@@ -346,13 +390,15 @@ impl Client {
                 MountTable {
                     mounts: Vec::new(),
                     fallback: MountLink::new("/", "/", protocol),
+                    home_dir,
                 }
             }
             "viewfs" => Self::build_mount_table(
                 // Host is guaranteed to be present.
                 resolved_url.host_str().expect("URL must have a host"),
                 Arc::clone(&config),
                 rt_holder.get_handle(),
+                home_dir,
             )?,
             _ => {
                 return Err(HdfsError::InvalidArgument(
@@ -372,6 +418,7 @@ impl Client {
         host: &str,
         config: Arc<Configuration>,
         handle: Handle,
+        home_dir: String,
     ) -> Result<MountTable> {
         let mut mounts: Vec<MountLink> = Vec::new();
         let mut fallback: Option<MountLink> = None;
@@ -408,7 +455,11 @@ impl Client {
             mounts.sort_by_key(|m| m.viewfs_path.chars().filter(|c| *c == '/').count());
             mounts.reverse();
 
-            Ok(MountTable { mounts, fallback })
+            Ok(MountTable {
+                mounts,
+                fallback,
+                home_dir,
+            })
         } else {
             Err(HdfsError::InvalidArgument(
                 "No viewfs fallback mount found".to_string(),
@@ -1180,6 +1231,7 @@ mod test {
         let mount_table = MountTable {
             mounts: vec![link1, link2, link3],
             fallback,
+            home_dir: "/user/test".to_string(),
         };
 
         // Exact mount path resolves to the exact HDFS path
@@ -1204,6 +1256,39 @@ mod test {
         let (link, resolved) = mount_table.resolve("/mount3/nested/file");
         assert_eq!(link.viewfs_path, "/mount3/nested");
         assert_eq!(resolved, "/path3/file");
+
+        let (link, resolved) = mount_table.resolve("file");
+        assert_eq!(link.viewfs_path, "");
+        assert_eq!(resolved, "/path4/user/test/file");
+
+        let (link, resolved) = mount_table.resolve("dir/subdir");
+        assert_eq!(link.viewfs_path, "");
+        assert_eq!(resolved, "/path4/user/test/dir/subdir");
+
+        let mount_table = MountTable {
+            mounts: vec![
+                MountLink::new(
+                    "/mount1",
+                    "/path1/nested",
+                    create_protocol("hdfs://127.0.0.1:9000"),
+                ),
+                MountLink::new(
+                    "/mount2",
+                    "/path2",
+                    create_protocol("hdfs://127.0.0.1:9001"),
+                ),
+            ],
+            fallback: MountLink::new("/", "/path4", create_protocol("hdfs://127.0.0.1:9003")),
+            home_dir: "/mount1/user".to_string(),
+        };
+
+        let (link, resolved) = mount_table.resolve("file");
+        assert_eq!(link.viewfs_path, "/mount1");
+        assert_eq!(resolved, "/path1/nested/user/file");
+
+        let (link, resolved) = mount_table.resolve("dir/subdir");
+        assert_eq!(link.viewfs_path, "/mount1");
+        assert_eq!(resolved, "/path1/nested/user/dir/subdir");
     }
 
     #[test]

rust/src/minidfs.rs

@@ -95,8 +95,7 @@ impl MiniDfs {
 
         if features.contains(&DfsFeatures::Security) {
             let krb_conf = output.next().unwrap().unwrap();
-            let kdestroy_exec = which("kdestroy").expect("Failed to find kdestroy executable");
-            Command::new(kdestroy_exec).spawn().unwrap().wait().unwrap();
+            Self::kdestroy();
 
             if !PathBuf::from("target/test/hdfs.keytab").exists() {
                 panic!("Failed to find keytab");
@@ -147,6 +146,17 @@ impl MiniDfs {
             url: url.to_string(),
         }
     }
+
+    fn kdestroy() {
+        let kdestroy_exec = which("kdestroy").expect("Failed to find kdestroy executable");
+        Command::new(kdestroy_exec)
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()
+            .unwrap()
+            .wait()
+            .unwrap();
+    }
 }
 
 impl Default for MiniDfs {
@@ -162,5 +172,7 @@ impl Drop for MiniDfs {
         stdin.write_all(b"\n").unwrap();
         self.process.kill().unwrap();
         self.process.wait().unwrap();
+
+        Self::kdestroy();
     }
 }

rust/src/security/gssapi.rs

@@ -209,6 +209,64 @@ impl Drop for GssName {
     }
 }
 
+struct GssCred {
+    cred: bindings::gss_cred_id_t,
+}
+
+impl GssCred {
+    fn acquire_default() -> crate::Result<Self> {
+        let mut minor = 0;
+        let mut cred = ptr::null_mut();
+        let major = unsafe {
+            libgssapi()?.gss_acquire_cred(
+                &mut minor,
+                ptr::null_mut(),
+                bindings::_GSS_C_INDEFINITE,
+                ptr::null_mut(),
+                bindings::GSS_C_INITIATE as bindings::gss_cred_usage_t,
+                &mut cred,
+                ptr::null_mut(),
+                ptr::null_mut(),
+            )
+        };
+        check_gss_ok(major, minor)?;
+        Ok(Self { cred })
+    }
+
+    fn name(&self) -> crate::Result<GssName> {
+        let mut minor = 0;
+        let mut name = ptr::null_mut::<bindings::gss_name_struct>();
+        let major = unsafe {
+            libgssapi()?.gss_inquire_cred(
+                &mut minor,
+                self.cred,
+                &mut name as *mut bindings::gss_name_t,
+                ptr::null_mut(),
+                ptr::null_mut(),
+                ptr::null_mut(),
+            )
+        };
+        check_gss_ok(major, minor)?;
+        Ok(GssName { name })
+    }
+}
+
+impl Drop for GssCred {
+    fn drop(&mut self) {
+        if !self.cred.is_null() {
+            let mut minor = bindings::GSS_S_COMPLETE;
+            let major = unsafe {
+                libgssapi()
+                    .unwrap()
+                    .gss_release_cred(&mut minor, &mut self.cred)
+            };
+            if let Err(e) = check_gss_ok(major, minor) {
+                warn!("Failed to release GSSAPI credential: {:?}", e);
+            }
+        }
+    }
+}
+
 struct GssClientCtx {
     ctx: bindings::gss_ctx_id_t,
     target: GssName,
@@ -440,6 +498,12 @@ impl GssapiSession {
         let state = GssapiState::Pending(GssClientCtx::new(target));
         Ok(Self { state })
     }
+
+    pub(crate) fn get_default_principal() -> crate::Result<String> {
+        let cred = GssCred::acquire_default()?;
+        let name = cred.name()?;
+        name.display_name()
+    }
 }
 
 impl SaslSession for GssapiSession {

rust/src/security/sasl.rs

@@ -198,7 +198,6 @@ fn select_method(
                     auth.server_id().to_string(),
                     token,
                 );
-                // let session = GSASLSession::new(auth.protocol(), auth.server_id(), token)?;
 
                 return Ok((auth.clone(), Some(Box::new(session))));
             }
@@ -261,7 +260,7 @@ impl SaslReader {
     }
 
     pub(crate) async fn read_exact(&mut self, buf: &mut [u8]) -> Result<usize> {
-        if self.session.is_some() {
+        if let Some(session) = self.session.clone() {
             let read_len = buf.len();
             let mut bytes_remaining = read_len;
             while bytes_remaining > 0 {
@@ -271,14 +270,7 @@ impl SaslReader {
                         todo!();
                     }
 
-                    // let mut writer = BytesMut::with_capacity(response.token().len()).writer();
-                    let decoded = self
-                        .session
-                        .as_ref()
-                        .unwrap()
-                        .lock()
-                        .unwrap()
-                        .decode(response.token())?;
+                    let decoded = session.lock().unwrap().decode(response.token())?;
                     self.buffer = Bytes::from(decoded)
                 }
                 let copy_len = usize::min(bytes_remaining, self.buffer.remaining());
@@ -340,17 +332,13 @@ impl SaslWriter {
     }
 
     pub(crate) async fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
-        if self.session.is_some() {
+        if let Some(session) = &self.session {
             let mut rpc_sasl = RpcSaslProto {
                 state: SaslState::Wrap as i32,
                 ..Default::default()
             };
 
-            // let mut writer = Vec::with_capacity(buf.len()).writer();
-            let encoded = self
-                .session
-                .as_ref()
-                .unwrap()
+            let encoded = session
                 .lock()
                 .unwrap()
                 .encode(buf)

rust/src/security/user.rs

@@ -17,6 +17,7 @@ use crate::proto::common::TokenProto;
 use crate::proto::hdfs::AccessModeProto;
 use crate::proto::hdfs::BlockTokenSecretProto;
 use crate::proto::hdfs::StorageTypeProto;
+use crate::security::gssapi::GssapiSession;
 
 const HADOOP_USER_NAME: &str = "HADOOP_USER_NAME";
 const HADOOP_PROXY_USER: &str = "HADOOP_PROXY_USER";
@@ -381,6 +382,14 @@ impl User {
         }
     }
 
+    pub(crate) fn get_user_info() -> UserInfo {
+        if let Ok(principal) = GssapiSession::get_default_principal() {
+            return User::get_user_info_from_principal(&principal);
+        }
+
+        User::get_simple_user()
+    }
+
     pub(crate) fn get_user_from_principal(principal: &str) -> String {
         // If there's a /, take the part before it.
         if let Some(index) = principal.find('/') {