diff --git a/apps/filebrowser/src/filebrowser/views_test.py b/apps/filebrowser/src/filebrowser/views_test.py
index 514248637c5..46340d9db25 100644
--- a/apps/filebrowser/src/filebrowser/views_test.py
+++ b/apps/filebrowser/src/filebrowser/views_test.py
@@ -53,7 +53,7 @@
 from desktop.lib.django_test_util import make_logged_in_client
 from desktop.lib.test_utils import grant_access, add_to_group, add_permission, remove_from_group
 from desktop.lib.view_util import location_to_url
-from desktop.conf import is_oozie_enabled, RAZ
+from desktop.conf import is_oozie_enabled, RAZ, is_ofs_enabled, OZONE
 
 from hadoop import pseudo_hdfs4
 from hadoop.conf import UPLOAD_CHUNK_SIZE
@@ -1530,6 +1530,45 @@ def test_has_default_permissions(self):
     finally:
       remove_from_group(self.user.username, 'has_adls')
 
+class TestOFSAccessPermissions(object):
+  def setUp(self):
+    self.client = make_logged_in_client(username="test", groupname="default", recreate=True, is_superuser=False)
+    grant_access('test', 'test', 'filebrowser')
+    add_to_group('test')
+
+    self.user = User.objects.get(username="test")
+
+  def test_no_default_permissions(self):
+    response = self.client.get('/filebrowser/view=ofs://')
+    assert_equal(500, response.status_code)
+
+    response = self.client.get('/filebrowser/view=ofs://volume')
+    assert_equal(500, response.status_code)
+
+    response = self.client.get('/filebrowser/view=ofs://volume/bucket')
+    assert_equal(500, response.status_code)
+
+    response = self.client.get('/filebrowser/view=ofs://volume/bucket/hue')
+    assert_equal(500, response.status_code)
+
+    response = self.client.post('/filebrowser/rmtree', dict(path=['ofs://volume/bucket/hue']))
+    assert_equal(500, response.status_code)
+
+    # 500 for real currently
+    assert_raises(IOError, self.client.get, '/filebrowser/edit=ofs://volume/bucket/hue')
+
+  def test_has_default_permissions(self):
+    if not is_ofs_enabled():
+      raise SkipTest
+
+    add_permission(self.user.username, 'has_ofs', permname='ofs_access', appname='filebrowser')
+
+    try:
+      response = self.client.get('/filebrowser/view=ofs://')
+      assert_equal(200, response.status_code)
+    finally:
+      remove_from_group(self.user.username, 'has_ofs')
+
 
 class TestFileChooserRedirect(object):
 
@@ -1540,7 +1579,7 @@ def setUp(self):
 
     self.user = User.objects.get(username="test")
 
-  def test_hdfs_redirect(self):
+  def test_fs_redirect(self):
     with patch('desktop.lib.fs.proxyfs.ProxyFS.isdir') as is_dir:
       is_dir.return_value = True
 
@@ -1550,11 +1589,34 @@ def test_hdfs_redirect(self):
       assert_equal(302, response.status_code)
       assert_equal('/filebrowser/view=%2Fuser%2Ftest', response.url)
 
+      # OFS - default_ofs_home
+      reset = OZONE['default'].WEBHDFS_URL.set_for_testing(None)
+      try:
+        response = self.client.get('/filebrowser/view=%2F?default_ofs_home')
+
+        assert_equal(302, response.status_code)
+        assert_equal('/filebrowser/view=ofs%3A%2F%2F', response.url)
+      finally:
+        reset()
+
+      reset = OZONE['default'].WEBHDFS_URL.set_for_testing('http://localhost:9778/webhdfs/v1')
+      try:
+        response = self.client.get('/filebrowser/view=%2F?default_ofs_home')
+
+        assert_equal(302, response.status_code)
+        assert_equal('/filebrowser/view=ofs%3A%2F%2F', response.url)
+      finally:
+        reset()
+
       # ABFS - default_abfs_home
-      response = self.client.get('/filebrowser/view=%2F?default_abfs_home')
+      reset = ABFS_CLUSTERS['default'].FS_DEFAULTFS.set_for_testing(None)
+      try:
+        response = self.client.get('/filebrowser/view=%2F?default_abfs_home')
 
-      assert_equal(302, response.status_code)
-      assert_equal('/filebrowser/view=abfs%3A%2F%2F', response.url)
+        assert_equal(302, response.status_code)
+        assert_equal('/filebrowser/view=abfs%3A%2F%2F', response.url)
+      finally:
+        reset()
 
       reset = ABFS_CLUSTERS['default'].FS_DEFAULTFS.set_for_testing('abfs://data-container@mystorage.dfs.core.windows.net')
       try:
@@ -1592,10 +1654,14 @@ def test_hdfs_redirect(self):
           reset()
 
       # S3A - default_s3_home
-      response = self.client.get('/filebrowser/view=%2F?default_s3_home')
+      reset = REMOTE_STORAGE_HOME.set_for_testing(None)
+      try:
+        response = self.client.get('/filebrowser/view=%2F?default_s3_home')
 
-      assert_equal(302, response.status_code)
-      assert_equal('/filebrowser/view=s3a%3A%2F%2F', response.url)
+        assert_equal(302, response.status_code)
+        assert_equal('/filebrowser/view=s3a%3A%2F%2F', response.url)
+      finally:
+        reset()
 
       reset = REMOTE_STORAGE_HOME.set_for_testing('s3a://my_bucket')
       try:
@@ -1683,3 +1749,19 @@ def test_s3a_correction_already_correct(self):
 
     normalized = _normalize_path(path)
     assert_equal(path, normalized)
+
+  def test_ofs_correction(self):
+    path = 'ofs:%2Fsome%2Fpath'
+    expected_corrected_path = 'ofs://some/path'
+
+    normalized_once = _normalize_path(path)
+    assert_equal(expected_corrected_path, normalized_once)
+
+    normalized_twice = _normalize_path(normalized_once)
+    assert_equal(expected_corrected_path, normalized_twice)
+
+  def test_ofs_correction_already_correct(self):
+    path = 'ofs://some/path'
+
+    normalized = _normalize_path(path)
+    assert_equal(path, normalized)
diff --git a/desktop/core/src/desktop/lib/fs/ozone/ofs_test.py b/desktop/core/src/desktop/lib/fs/ozone/ofs_test.py
new file mode 100644
index 00000000000..35af487a116
--- /dev/null
+++ b/desktop/core/src/desktop/lib/fs/ozone/ofs_test.py
@@ -0,0 +1,116 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from nose.tools import assert_equal
+
+from desktop import conf
+from desktop.lib.fs.ozone.client import _make_ofs_client
+
+
+class TestOFSClient(object):
+
+  @classmethod
+  def setUpClass(cls):
+    cls._resets = [
+      conf.OZONE['default'].FS_DEFAULTFS.set_for_testing('ofs://ozone1'),
+      conf.OZONE['default'].LOGICAL_NAME.set_for_testing('test-logical-name'),
+      conf.OZONE['default'].WEBHDFS_URL.set_for_testing('http://gethue-ozone:9778/webhdfs/v1'),
+      conf.OZONE['default'].SECURITY_ENABLED.set_for_testing(True),
+      conf.OZONE['default'].SSL_CERT_CA_VERIFY.set_for_testing(True),
+      conf.OZONE['default'].TEMP_DIR.set_for_testing('/tmp')
+    ]
+
+    cls.ofs_client = _make_ofs_client('default')
+
+
+  def test_client_attributes(self):
+    assert_equal(self.ofs_client._url, 'http://gethue-ozone:9778/webhdfs/v1')
+    assert_equal(self.ofs_client._superuser, None)
+    assert_equal(self.ofs_client._security_enabled, True)
+    assert_equal(self.ofs_client._ssl_cert_ca_verify, True)
+    assert_equal(self.ofs_client._temp_dir, '/tmp')
+    assert_equal(self.ofs_client._umask, 530)
+    assert_equal(self.ofs_client._fs_defaultfs, 'ofs://ozone1')
+    assert_equal(self.ofs_client._logical_name, 'test-logical-name')
+    assert_equal(self.ofs_client._supergroup, None)
+    assert_equal(self.ofs_client._scheme, 'ofs')
+    assert_equal(self.ofs_client._netloc, 'ozone1')
+    assert_equal(self.ofs_client._is_remote, True)
+    assert_equal(self.ofs_client._has_trash_support, False)
+    assert_equal(self.ofs_client.expiration, None)
+    assert_equal(self.ofs_client._filebrowser_action, 'ofs_access')
+
+
+  def test_strip_normpath(self):
+    test_path = self.ofs_client.strip_normpath('ofs://vol1/buk1/key')
+    assert_equal(test_path, '/vol1/buk1/key')
+
+    test_path = self.ofs_client.strip_normpath('ofs:/vol1/buk1/key')
+    assert_equal(test_path, '/vol1/buk1/key')
+
+    test_path = self.ofs_client.strip_normpath('/vol1/buk1/key')
+    assert_equal(test_path, '/vol1/buk1/key')
+
+
+  def test_normpath(self):
+    test_path = self.ofs_client.normpath('ofs://')
+    assert_equal(test_path, 'ofs://')
+
+    test_path = self.ofs_client.normpath('ofs://vol1/buk1/key')
+    assert_equal(test_path, 'ofs://vol1/buk1/key')
+
+    test_path = self.ofs_client.normpath('ofs://vol1/buk1/key/')
+    assert_equal(test_path, 'ofs://vol1/buk1/key')
+
+    test_path = self.ofs_client.normpath('ofs://vol1/buk1/key//')
+    assert_equal(test_path, 'ofs://vol1/buk1/key')
+
+    test_path = self.ofs_client.normpath('ofs://vol1/buk1//key//')
+    assert_equal(test_path, 'ofs://vol1/buk1/key')
+
+
+  def test_isroot(self):
+    is_root = self.ofs_client.isroot('ofs://vol1/buk1/key')
+    assert_equal(is_root, False)
+
+    is_root = self.ofs_client.isroot('ofs://')
+    assert_equal(is_root, True)
+
+
+  def test_parent_path(self):
+    parent_path = self.ofs_client.parent_path('ofs://')
+    assert_equal(parent_path, 'ofs://')
+
+    parent_path = self.ofs_client.parent_path('ofs://vol1/buk1/dir1/file1.csv')
+    assert_equal(parent_path, 'ofs://vol1/buk1/dir1')
+
+    parent_path = self.ofs_client.parent_path('ofs://vol1/buk1/key')
+    assert_equal(parent_path, 'ofs://vol1/buk1')
+
+    parent_path = self.ofs_client.parent_path('ofs://vol1/buk1')
+    assert_equal(parent_path, 'ofs://vol1/')
+
+    parent_path = self.ofs_client.parent_path('ofs://vol1')
+    assert_equal(parent_path, 'ofs://')
+
+
+  @classmethod
+  def tearDownClass(cls):
+    for reset in cls._resets:
+      reset()
+
+
diff --git a/desktop/core/src/desktop/lib/fs/ozone/ofsstat_test.py b/desktop/core/src/desktop/lib/fs/ozone/ofsstat_test.py
new file mode 100644
index 00000000000..69d32f31c82
--- /dev/null
+++ b/desktop/core/src/desktop/lib/fs/ozone/ofsstat_test.py
@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from nose.tools import assert_equal
+from desktop.lib.fs.ozone.ofsstat import OzoneFSStat
+
+
+class TestOzoneFSStat(object):
+  def setUp(self):
+    test_file_status = {
+      'pathSuffix': 'testfile.csv', 'type': 'FILE', 'length': 32, 'owner': 'hueadmin', 'group': 'huegroup',
+      'permission': '666', 'accessTime': 1677914460588, 'modificationTime': 1677914460588, 'blockSize': 268435456, 'replication': 3}
+
+    test_parent_path = '/gethue/'
+
+    self.stat = OzoneFSStat(test_file_status, test_parent_path)
+
+
+  def test_stat_attributes(self):
+    assert_equal(self.stat.name, 'testfile.csv')
+    assert_equal(self.stat.path, 'ofs://gethue/testfile.csv')
+    assert_equal(self.stat.isDir, False)
+    assert_equal(self.stat.type, 'FILE')
+    assert_equal(self.stat.atime, 1677914460)
+    assert_equal(self.stat.mtime, 1677914460)
+    assert_equal(self.stat.user, 'hueadmin')
+    assert_equal(self.stat.group, 'huegroup')
+    assert_equal(self.stat.size, 32)
+    assert_equal(self.stat.blockSize, 268435456)
+    assert_equal(self.stat.replication, 3)
+    assert_equal(self.stat.aclBit, None)
+    assert_equal(self.stat.fileId, None)
+    assert_equal(self.stat.mode, 33206)
+
+
+  def test_to_json_dict(self):
+    expected_json_dict = {
+      'path': 'ofs://gethue/testfile.csv', 'size': 32, 'atime': 1677914460, 'mtime': 1677914460, 'mode': 33206, 'user': 'hueadmin',
+      'group': 'huegroup', 'blockSize': 268435456, 'replication': 3}
+
+    assert_equal(self.stat.to_json_dict(), expected_json_dict)
diff --git a/desktop/core/src/desktop/lib/fs/proxyfs_test.py b/desktop/core/src/desktop/lib/fs/proxyfs_test.py
index c1acbc10453..ce41b7af31e 100644
--- a/desktop/core/src/desktop/lib/fs/proxyfs_test.py
+++ b/desktop/core/src/desktop/lib/fs/proxyfs_test.py
@@ -42,8 +42,10 @@ def test_fs_selection():
   with patch('desktop.lib.fs.ProxyFS._has_access') as _has_access:
     _has_access.return_value = True
 
-    s3fs, adls, hdfs, abfs, gs = MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()
-    proxy_fs = ProxyFS({'s3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs)}, 'hdfs')
+    s3fs, adls, hdfs, abfs, gs, ofs = MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()
+    proxy_fs = ProxyFS({
+      's3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs), 'ofs': wrapper(ofs)},
+      'hdfs')
     proxy_fs.setuser(user)
 
     proxy_fs.isdir('s3a://bucket/key')
@@ -66,6 +68,10 @@ def test_fs_selection():
     gs.isdir.assert_called_once_with('gs://net/key')
     assert_false(hdfs.isdir.called)
 
+    proxy_fs.isdir('ofs://volume/bucket/key')
+    ofs.isdir.assert_called_once_with('ofs://volume/bucket/key')
+    assert_false(hdfs.isdir.called)
+
     assert_raises(IOError, proxy_fs.stats, 'ftp://host')
 
 def wrapper(mock):
@@ -81,8 +87,10 @@ def test_multi_fs_selection():
   with patch('desktop.lib.fs.ProxyFS._has_access') as _has_access:
     _has_access.return_value = True
 
-    s3fs, adls, hdfs, abfs, gs = MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()
-    proxy_fs = ProxyFS({'s3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs)}, 'hdfs')
+    s3fs, adls, hdfs, abfs, gs, ofs = MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()
+    proxy_fs = ProxyFS({
+      's3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs), 'ofs': wrapper(ofs)},
+      'hdfs')
     proxy_fs.setuser(user)
 
     proxy_fs.copy('s3a://bucket1/key', 's3a://bucket2/key')
@@ -109,8 +117,12 @@ def test_multi_fs_selection():
     gs.copyfile.assert_called_once_with('gs://bucket/key', 'key2')
     assert_false(hdfs.copyfile.called)
 
-    # Will be addressed in HUE-2934
-    assert_raises(NotImplementedError, proxy_fs.copy_remote_dir, 's3a://bucket/key', 'adl://tmp/dir') # Exception can only be thrown if scheme is specified, else default to 1st scheme
+    proxy_fs.copyfile('ofs://volume/bucket/key', 'key2')
+    ofs.copyfile.assert_called_once_with('ofs://volume/bucket/key', 'key2')
+    assert_false(hdfs.copyfile.called)
+
+    # Exception can only be thrown if scheme is specified, else default to 1st scheme
+    assert_raises(NotImplementedError, proxy_fs.copy_remote_dir, 's3a://bucket/key', 'adl://tmp/dir')
 
 
 def test_constructor_given_invalid_arguments():
@@ -136,8 +148,12 @@ def test_fs_permissions_regular_user(self):
     user_client = make_logged_in_client(username='test', groupname='default', recreate=True, is_superuser=False)
     user = User.objects.get(username='test')
 
-    s3fs, adls, hdfs, abfs, gs = MockFs("s3_access"), MockFs("adls_access"), MockFs(), MockFs("abfs_access"), MockFs("gs_access")
-    proxy_fs = ProxyFS({'s3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs)}, 'hdfs')
+    s3fs, adls, hdfs = MockFs("s3_access"), MockFs("adls_access"), MockFs()
+    abfs, gs, ofs = MockFs("abfs_access"), MockFs("gs_access"), MockFs("ofs_access")
+
+    proxy_fs = ProxyFS({
+      's3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs), 'ofs': wrapper(ofs)},
+      'hdfs')
     proxy_fs.setuser(user)
 
     f = proxy_fs._get_fs
@@ -146,6 +162,7 @@ def test_fs_permissions_regular_user(self):
     remove_from_group(user.username, 'has_adls')
     remove_from_group(user.username, 'has_abfs')
     remove_from_group(user.username, 'has_gs')
+    remove_from_group(user.username, 'has_ofs')
 
     # No perms by default
     assert_raises(Exception, f, 's3a://bucket')
@@ -154,6 +171,7 @@ def test_fs_permissions_regular_user(self):
     assert_raises(Exception, f, 'adl:/key')
     assert_raises(Exception, f, 'abfs:/key')
     assert_raises(Exception, f, 'gs://bucket/key')
+    assert_raises(Exception, f, 'ofs://volume/bucket/key')
     f('hdfs://path')
     f('/tmp')
 
@@ -163,6 +181,7 @@ def test_fs_permissions_regular_user(self):
       add_permission('test', 'has_adls', permname='adls_access', appname='filebrowser')
       add_permission('test', 'has_abfs', permname='abfs_access', appname='filebrowser')
       add_permission('test', 'has_gs', permname='gs_access', appname='filebrowser')
+      add_permission('test', 'has_ofs', permname='ofs_access', appname='filebrowser')
 
       f('s3a://bucket')
       f('S3A://bucket/key')
@@ -172,18 +191,24 @@ def test_fs_permissions_regular_user(self):
       f('hdfs://path')
       f('/tmp')
       f('gs://bucket')
+      f('ofs://volume/bucket/key')
     finally:
       remove_from_group(user.username, 'has_s3')
       remove_from_group(user.username, 'has_adls')
       remove_from_group(user.username, 'has_abfs')
       remove_from_group(user.username, 'has_gs')
+      remove_from_group(user.username, 'has_ofs')
 
   def test_fs_permissions_admin_user(self):
     user_client = make_logged_in_client(username='admin', groupname='default', recreate=True, is_superuser=True)
     user = User.objects.get(username='admin')
 
-    s3fs, adls, hdfs, abfs, gs = MockFs("s3_access"), MockFs("adls_access"), MockFs(), MockFs("abfs_access"), MockFs("gs_access")
-    proxy_fs = ProxyFS({'s3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs)}, 'hdfs')
+    s3fs, adls, hdfs = MockFs("s3_access"), MockFs("adls_access"), MockFs()
+    abfs, gs, ofs = MockFs("abfs_access"), MockFs("gs_access"), MockFs("ofs_access")
+
+    proxy_fs = ProxyFS({
+      's3a': wrapper(s3fs), 'hdfs': wrapper(hdfs), 'adl': wrapper(adls), 'abfs': wrapper(abfs), 'gs': wrapper(gs), 'ofs': wrapper(ofs)},
+      'hdfs')
     proxy_fs.setuser(user)
 
     f = proxy_fs._get_fs
@@ -196,3 +221,5 @@ def test_fs_permissions_admin_user(self):
     f('hdfs://path')
     f('/tmp')
     f('gs://bucket/key')
+    f('ofs://volume/bucket/key')
+