NSFS | Versioning | Delete of partial directory of nested key results…

… in AccessDeniedError

Signed-off-by: naveenpaul1 <[email protected]>

src/sdk/namespace_fs.js

@@ -2784,6 +2784,22 @@ class NamespaceFS {
         const versioned_path = this._get_version_path(key, version_id);
         return versioned_path;
     }
+    /** 
+    * Methid will check the the key is pointing to dir or file
+    * @param {import('./nb').NativeFSContext} fs_context
+    * @param {string} key
+    */
+    async _is_key_dir_path(fs_context, key) {
+        try {
+            const key_path = path.normalize(path.join(this.bucket_path, key));
+            const key_stat = await nb_native().fs.stat(fs_context, key_path, { skip_user_xattr: true });
+            const is_dir = native_fs_utils.isDirectory(key_stat);
+            return is_dir;
+        } catch (err) {
+            dbg.warn('NamespaceFS._correct_key_path : error while getting state for key ', key, err);
+        }
+        return false;
+    }
 
     _throw_if_delete_marker(stat, params) {
         if (this.versioning === versioning_status_enum.VER_ENABLED || this.versioning === versioning_status_enum.VER_SUSPENDED) {
@@ -3017,8 +3033,12 @@ class NamespaceFS {
         const is_gpfs = native_fs_utils._is_gpfs(fs_context);
         let retries = config.NSFS_RENAME_RETRIES;
         let latest_ver_info;
+        const is_key_dir_path = await this._is_key_dir_path(fs_context, params.key);
         for (;;) {
             try {
+                if (is_key_dir_path && !this._is_directory_content(latest_ver_path, params.key)) {
+                    throw new S3Error(S3Error.AccessDenied);
+                }
                 // TODO get latest version from file in POSIX like in GPFS path
                 latest_ver_info = await this._get_version_info(fs_context, latest_ver_path);
                 dbg.log1('Namespace_fs._delete_latest_version:', latest_ver_info);
@@ -3035,7 +3055,7 @@ class NamespaceFS {
                     const suspended_and_latest_is_not_null = this._is_versioning_suspended() &&
                         latest_ver_info.version_id_str !== NULL_VERSION_ID;
                     const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
-                    if (this._is_versioning_enabled() || suspended_and_latest_is_not_null) {
+                    if ((this._is_versioning_enabled() || suspended_and_latest_is_not_null) && !is_key_dir_path) {
                         await native_fs_utils._make_path_dirs(versioned_path, fs_context);
                          await native_fs_utils.safe_move_posix(fs_context, latest_ver_path, versioned_path, latest_ver_info,
                             bucket_tmp_dir_path);
@@ -3061,11 +3081,13 @@ class NamespaceFS {
             }
         }
         // create delete marker and move it to .versions/key_{delete_marker_version_id}
-        const created_version_id = await this._create_delete_marker(fs_context, params, latest_ver_info);
-        return {
-            created_delete_marker: true,
-            created_version_id
-        };
+        if (!is_key_dir_path) {
+            const created_version_id = await this._create_delete_marker(fs_context, params, latest_ver_info);
+            return {
+                created_delete_marker: true,
+                created_version_id
+            };
+        }
     }
 
     // We can have only one versioned object with null version ID per key.

src/test/unit_tests/test_bucketspace_versioning.js

@@ -26,6 +26,7 @@ const XATTR_VERSION_ID = XATTR_INTERNAL_NOOBAA_PREFIX + 'version_id';
 const XATTR_DELETE_MARKER = XATTR_INTERNAL_NOOBAA_PREFIX + 'delete_marker';
 const NULL_VERSION_ID = 'null';
 const HIDDEN_VERSIONS_PATH = '.versions';
+const NSFS_FOLDER_OBJECT_NAME = '.folder';
 
 const DEFAULT_FS_CONFIG = get_process_fs_context(IS_GPFS ? 'GPFS' : '');
 let CORETEST_ENDPOINT;
@@ -75,6 +76,7 @@ mocha.describe('bucketspace namespace_fs - versioning', function() {
     const suspended_dir1_versions_path = path.join(suspended_full_path, dir1, '.versions/');
 
     const dir_path_nested = 'photos/animals/January/';
+    const dir_path_complete = 'animal/mammals/dog/';
     const nested_key_level3 = path.join(dir_path_nested, 'cat.jpeg');
 
     mocha.before(async function() {
@@ -764,11 +766,34 @@ mocha.describe('bucketspace namespace_fs - versioning', function() {
 
         mocha.it('delete object - versioning enabled - nested key (more than 1 level) - delete inside directory', async function() {
             const res = await s3_uid6.deleteObject({ Bucket: nested_keys_bucket_name, Key: dir_path_nested });
-            assert.equal(res.DeleteMarker, true);
+            // object versioning is not enabled for dir, because of this no delete_marker.
+            assert.equal(res.DeleteMarker, undefined);
             const version_path_nested = path.join(nested_keys_full_path, dir_path_nested, HIDDEN_VERSIONS_PATH);
             const exist2 = await fs_utils.file_exists(version_path_nested);
             assert.ok(exist2);
         });
+
+        mocha.it('delete object - versioning enabled - nested key (more than 1 level)- delete partial directory', async function() {
+            const parital_nested_directory = dir_path_complete.slice(0, -1); // the directory without the last slash
+            const folder_path_nested = path.join(nested_keys_full_path, dir_path_complete, NSFS_FOLDER_OBJECT_NAME);
+            try {
+                const body_of_copied_key = 'make the lemon lemonade';
+                await s3_uid6.putObject({ Bucket: nested_keys_bucket_name, Key: dir_path_complete, Body: body_of_copied_key });
+                await fs_utils.file_must_exist(folder_path_nested);
+                await s3_uid6.deleteObject({ Bucket: nested_keys_bucket_name, Key: parital_nested_directory });
+            } catch (err) {
+                assert.equal(err.Code, 'AccessDenied');
+            }
+            await fs_utils.file_must_exist(folder_path_nested);
+        });
+
+        mocha.it('delete object - versioning enabled - nested key (more than 1 level)- delete complete directory', async function() {
+            const res = await s3_uid6.deleteObject({ Bucket: nested_keys_bucket_name, Key: dir_path_complete });
+            // object versioning is not enabled for dir, because of this no delete_marker.
+            assert.equal(res.DeleteMarker, undefined);
+            const folder_path_nested = path.join(nested_keys_full_path, dir_path_complete, NSFS_FOLDER_OBJECT_NAME);
+            await fs_utils.file_must_not_exist(folder_path_nested);
+        });
     });
 
     mocha.describe('copy object (latest version) - versioning suspended - nsfs copy fallback flow', function() {

src/test/unit_tests/test_namespace_fs.js

@@ -1,6 +1,6 @@
 /* Copyright (C) 2020 NooBaa */
 /*eslint max-lines-per-function: ["error", 900]*/
-/*eslint max-lines: ["error", 2100]*/
+/*eslint max-lines: ["error", 2200]*/
 'use strict';
 
 const _ = require('lodash');
@@ -426,6 +426,7 @@ mocha.describe('namespace_fs', function() {
 
         const dir_1 = '/a/b/c/';
         const dir_2 = '/a/b/';
+        const dir_3 = '/x/y/z/';
         const upload_key_1 = dir_1 + 'upload_key_1';
         const upload_key_2 = dir_1 + 'upload_key_2';
         const upload_key_3 = dir_2 + 'upload_key_3';
@@ -472,23 +473,54 @@ mocha.describe('namespace_fs', function() {
 
         });
 
+        mocha.it('delete partial dir object without last slash version enabled - /x/y/z', async function() {
+            try {
+                const source = buffer_utils.buffer_to_read_stream(data);
+                ns_tmp.set_bucket_versioning('ENABLED', dummy_object_sdk);
+                const partial_dir_3 = dir_3.slice(0, -1); // the path without the last slash
+                await upload_object(ns_tmp, upload_bkt, dir_3, dummy_object_sdk, source);
+                await fs_utils.file_must_exist(path.join(ns_tmp_bucket_path, '/x/y/z/', config.NSFS_FOLDER_OBJECT_NAME));
+                await delete_object(ns_tmp, upload_bkt, partial_dir_3, dummy_object_sdk);
+                await fs_utils.file_must_exist(path.join(ns_tmp_bucket_path, '/x/y/z/', config.NSFS_FOLDER_OBJECT_NAME));
+                assert.fail('head should have failed with access denaied error');
+            } catch (err) {
+                assert.equal(err.code, 'AccessDenied');
+                await delete_object(ns_tmp, upload_bkt, dir_3, dummy_object_sdk);
+            }
+        });
+
+        mocha.it('delete dir object, version enabled - /x/y/z/', async function() {
+            const source = buffer_utils.buffer_to_read_stream(data);
+            ns_tmp.set_bucket_versioning('ENABLED', dummy_object_sdk);
+            await upload_object(ns_tmp, upload_bkt, dir_3, dummy_object_sdk, source);
+            await fs_utils.file_must_exist(path.join(ns_tmp_bucket_path, '/x/y/z/', config.NSFS_FOLDER_OBJECT_NAME));
+            const resp = await delete_object(ns_tmp, upload_bkt, dir_3, dummy_object_sdk);
+            await fs_utils.file_must_not_exist(path.join(ns_tmp_bucket_path, '/x/y/z/', config.NSFS_FOLDER_OBJECT_NAME));
+            // object versioning is not enabled for dir, because of this no delete_marker.
+            assert.deepEqual(resp, {});
+            const res = await ns_tmp.list_object_versions({
+                bucket: upload_bkt,
+                prefix: '/x/y/'
+            }, dummy_object_sdk);
+            assert.equal(res.objects.length, 1);
+        });
+
         mocha.after(async function() {
             let entries_before;
             let entries_after;
             try {
                 entries_before = await nb_native().fs.readdir(DEFAULT_FS_CONFIG, ns_tmp_bucket_path);
-
                 const delete_res = await ns_tmp.delete_object({
                     bucket: upload_bkt,
                     key: upload_key_3,
                 }, dummy_object_sdk);
                 console.log('delete_object response', inspect(delete_res));
-
                 entries_after = await nb_native().fs.readdir(DEFAULT_FS_CONFIG, ns_tmp_bucket_path);
             } catch (e) {
                 assert.ifError(e);
             }
-            assert.strictEqual(entries_after.length, entries_before.length - 1);
+            ns_tmp.set_bucket_versioning('DISABLED', dummy_object_sdk);
+            assert.strictEqual(entries_after.length, entries_before.length);
         });
     });