Skip DFS referrals without referral_entries (#149)

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 1.8.1 - TBD
+
+* Raises `ObjectPathNotFound` if a DFS referral is required but not referrals are available (https://github.com/jborean93/smbprotocol/pull/149)
+
+
 ## 1.8.0 - 2021-10-21
 
 * Added support for 256bit keyed encryption ciphers

smbclient/_io.py

@@ -156,6 +156,9 @@ def _resolve_dfs(raw_io):
     referral = dfs_request(raw_io.fd.tree_connect, raw_path[1:])
     client_config.cache_referral(referral)
     info = client_config.lookup_referral([p for p in raw_path.split("\\") if p])
+    if not info:
+        raise ObjectPathNotFound()
+
     connection_kwargs = getattr(raw_io, '_%s__kwargs' % type(raw_io).__name__, {})
 
     for target in info:

smbclient/_pool.py

@@ -28,6 +28,7 @@
 from smbprotocol.exceptions import (
     BadNetworkName,
     InvalidParameter,
+    ObjectPathNotFound,
 )
 
 from smbprotocol.ioctl import (
@@ -142,7 +143,8 @@ def domain_controller(self, value):
             self._domain_cache.append(DomainEntry(domain_referral))
 
     def cache_referral(self, referral):
-        self._referral_cache.append(ReferralEntry(referral))
+        if referral['number_of_referrals'].get_value() > 0:
+            self._referral_cache.append(ReferralEntry(referral))
 
     def lookup_domain(self, domain_name):  # type: (str) -> Optional[DomainEntry]
         # TODO: Check domain referral expiry and resend request if expired
@@ -298,6 +300,9 @@ def get_smb_tree(path, username=None, password=None, port=445, encrypt=None, con
             referral_response = dfs_request(ipc_tree, "\\%s\\%s" % (path_split[0], path_split[1]))
             client_config.cache_referral(referral_response)
             referral = client_config.lookup_referral(path_split)
+            if not referral:
+                raise ObjectPathNotFound()
+
             path = path.replace(referral.dfs_path, referral.target_hint.target_path, 1)
             path_split = [p for p in path.split("\\") if p]
 
@@ -321,6 +326,11 @@ def get_smb_tree(path, username=None, password=None, port=445, encrypt=None, con
             ipc_tree = get_smb_tree(ipc_path, **get_kwargs)[0]
             referral = dfs_request(ipc_tree, "\\%s\\%s" % (path_split[0], path_split[1]))
             client_config.cache_referral(referral)
+
+            # Sometimes a DFS referral may return 0 referrals, this needs to be checked here to avoid repeats.
+            if not client_config.lookup_referral(path_split):
+                raise ObjectPathNotFound()
+
             return get_smb_tree(path, **get_kwargs)
 
     file_path = ""

tests/test_smbclient_pool.py

@@ -4,12 +4,18 @@
 
 import pytest
 import smbclient._pool as pool
+import smbclient._io as io
+
+from smbprotocol.dfs import DFSReferralResponse
 
 from smbprotocol.exceptions import (
+    BadNetworkName,
     InvalidParameter,
+    ObjectPathNotFound,
 )
 
 from .conftest import (
+    DC_REFERRAL,
     DOMAIN_NAME,
     DOMAIN_REFERRAL,
 )
@@ -97,3 +103,55 @@ def test_reset_connection_error_warning(monkeypatch, mocker):
 
     assert warning_mock.call_count == 1
     assert warning_mock.call_args[0][0] == 'Failed to close connection conn: exception'
+
+
+def test_dfs_referral_no_links_no_domain(reset_config, monkeypatch, mocker):
+    no_referral = DFSReferralResponse()
+    no_referral["path_consumed"] = 0
+    no_referral["number_of_referrals"] = 0
+    dfs_mock = mocker.MagicMock()
+    dfs_mock.side_effect = [no_referral]
+
+    tree_mock = mocker.MagicMock()
+    tree_mock.side_effect = (BadNetworkName(), None)
+
+    monkeypatch.setattr(pool, "dfs_request", dfs_mock)
+    monkeypatch.setattr(pool.TreeConnect, "connect", tree_mock)
+    monkeypatch.setattr(pool, "register_session", mocker.MagicMock())
+
+    with pytest.raises(ObjectPathNotFound):
+        pool.get_smb_tree(r"\\server\dfs")
+
+
+def test_dfs_referral_no_links_from_domain(reset_config, monkeypatch, mocker):
+    actual_get_smb_tree = pool.get_smb_tree
+
+    no_referral = DFSReferralResponse()
+    no_referral["path_consumed"] = 0
+    no_referral["number_of_referrals"] = 0
+    dfs_mock = mocker.MagicMock()
+    dfs_mock.side_effect = [DOMAIN_REFERRAL, DC_REFERRAL, no_referral]
+
+    monkeypatch.setattr(pool, "dfs_request", dfs_mock)
+    monkeypatch.setattr(pool, 'get_smb_tree', mocker.MagicMock())
+    config = pool.ClientConfig()
+    config.domain_controller = DOMAIN_NAME
+
+    with pytest.raises(ObjectPathNotFound):
+        actual_get_smb_tree(f"\\\\{DOMAIN_NAME}\\dfs")
+
+
+def test_resolve_dfs_referral_no_links(reset_config, monkeypatch, mocker):
+    no_referral = DFSReferralResponse()
+    no_referral["path_consumed"] = 0
+    no_referral["number_of_referrals"] = 0
+    dfs_mock = mocker.MagicMock()
+    dfs_mock.side_effect = [no_referral]
+    monkeypatch.setattr(io, "dfs_request", dfs_mock)
+
+    raw_io = mocker.MagicMock()
+    raw_io.name = r"\\server\dfs"
+    raw_io.fd.tree_connect.is_dfs_share = True
+
+    with pytest.raises(ObjectPathNotFound):
+        list(io._resolve_dfs(raw_io))