NSFS | Versioning | Delete of partial directory of nested key results in AccessDeniedError

Signed-off-by: naveenpaul1 <napaul@redhat.com>

Author: naveenpaul1

src/sdk/namespace_fs.js

@@ -1926,7 +1926,10 @@ class NamespaceFS {
             await this._check_path_in_bucket_boundaries(fs_context, file_path);
             dbg.log0('NamespaceFS: delete_object', file_path);
             let res;
-            if (this._is_versioning_disabled()) {
+            const is_key_dir_path = await this._is_key_dir_path(fs_context, params.key);
+            if (this._is_versioning_disabled() || is_key_dir_path) {
+                // TODO- Directory object (key/) is currently can't co-exist while key (without slash) exists. see -https://github.com/noobaa/noobaa-core/issues/8320
+                // Also, Directory object (key/) is currently not supported combined with versioning - see - https://github.com/noobaa/noobaa-core/issues/8531
                 await this._delete_single_object(fs_context, file_path, params);
             } else {
                 res = params.version_id ?
@@ -2784,6 +2787,23 @@ class NamespaceFS {
         const versioned_path = this._get_version_path(key, version_id);
         return versioned_path;
     }
+    /** 
+    * _is_key_dir_path will check if key is pointing to a directory or a file
+    * @param {nb.NativeFSContext} fs_context
+    * @param {string} key
+    * @returns {Promise<boolean>}
+    */
+    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._is_key_dir_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) {

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,31 @@ 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);
+            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 });
+            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);
+            await fs_utils.file_must_not_exist(path.join(nested_keys_full_path, dir_path_complete));
+        });
     });
 
     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';
@@ -460,35 +461,82 @@ mocha.describe('namespace_fs', function() {
             const source = buffer_utils.buffer_to_read_stream(data);
             await upload_object(ns_tmp, upload_bkt, upload_key_3, dummy_object_sdk, source);
             await delete_object(ns_tmp, upload_bkt, upload_key_1, dummy_object_sdk);
-
             let entries;
             try {
                 entries = await nb_native().fs.readdir(DEFAULT_FS_CONFIG, ns_tmp_bucket_path + dir_2);
             } catch (e) {
                 assert.ifError(e);
             }
-            console.log('stop when not empty - entries', entries);
             assert.strictEqual(entries.length, 1);
 
         });
 
+        mocha.it('delete partial dir object without last slash 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 partial_dir_3 = dir_3.slice(0, -1); // the path without the last slash
+            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));
+            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));
+            await fs_utils.file_must_not_exist(path.join(ns_tmp_bucket_path, '/x/y/z/'));
+            // 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, 0);
+        });
+
+        mocha.it('delete dir object, version enabled - /x/y/z/ - multiple files', async function() {
+            const source = buffer_utils.buffer_to_read_stream(data);
+            const source1 = buffer_utils.buffer_to_read_stream(data);
+            ns_tmp.set_bucket_versioning('ENABLED', dummy_object_sdk);
+            const dir_3_object = path.join(dir_3, 'obj1');
+            await upload_object(ns_tmp, upload_bkt, dir_3, dummy_object_sdk, source);
+            await upload_object(ns_tmp, upload_bkt, dir_3_object, dummy_object_sdk, source1);
+            await fs_utils.file_must_exist(path.join(ns_tmp_bucket_path, dir_3, config.NSFS_FOLDER_OBJECT_NAME));
+            await fs_utils.file_must_exist(path.join(ns_tmp_bucket_path, dir_3_object));
+            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, dir_3, config.NSFS_FOLDER_OBJECT_NAME));
+            await fs_utils.file_must_exist(path.join(ns_tmp_bucket_path, dir_3));
+            await fs_utils.file_must_exist(path.join(ns_tmp_bucket_path, dir_3_object));
+            // 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);
         });
     });
 

src/util/native_fs_utils.js

@@ -254,9 +254,9 @@ async function unlink_ignore_enoent(fs_context, to_delete_path) {
     try {
         await nb_native().fs.unlink(fs_context, to_delete_path);
     } catch (err) {
-        dbg.warn(`native_fs_utils.unlink_ignore_enoent unlink error: file path ${to_delete_path} error`, err);
-        if (err.code !== 'ENOENT') throw err;
-        dbg.warn(`native_fs_utils.unlink_ignore_enoent unlink: file ${to_delete_path} already deleted, ignoring..`);
+        dbg.warn(`native_fs_utils.unlink_ignore_enoent unlink error: file path ${to_delete_path} error`, err, err.code, err.code !== 'EISDIR');
+        if (err.code !== 'ENOENT' && err.code !== 'EISDIR') throw err;
+        dbg.warn(`native_fs_utils.unlink_ignore_enoent unlink: file ${to_delete_path} already deleted or key is pointing to dir, ignoring..`);
     }
 }