Add smbclient.liststreams to enumerate ADS streams

src/smbclient/__init__.py

@@ -14,6 +14,7 @@
     getxattr,
     link,
     listdir,
+    liststreams,
     listxattr,
     lstat,
     makedirs,

src/smbclient/_os.py

@@ -37,6 +37,7 @@
     FileLinkInformation,
     FileRenameInformation,
     FileStandardInformation,
+    FileStreamInformation,
 )
 from smbprotocol.header import NtStatus
 from smbprotocol.ioctl import (
@@ -1019,6 +1020,36 @@ def setxattr(path, attribute, value, flags=0, follow_symlinks=True, **kwargs):
         set_info(transaction, ea_info)
 
 
+def liststreams(path, follow_symlinks=True, **kwargs) -> list[str]:
+    """
+    Return a list of the alternative data streams on a path. Listed streams can
+    be opened by appending their name to the original path. An example call for
+    a file with a single extra stream may return:
+
+    ```
+    [":extra_stream:$DATA", "::$DATA"]
+    ```
+
+    :param path: The full UNC path to the file to get the list of streams for.
+    :param follow_symlinks: Whether to follow the symlink at path if encountered.
+    :param kwargs: Common SMB Session arguments for smbclient.
+    :return: List of streams on the file with each entry being a string.
+    """
+
+    raw = SMBRawIO(
+        path,
+        mode="r",
+        share_access="r",
+        create_options=0 if follow_symlinks else CreateOptions.FILE_OPEN_REPARSE_POINT,
+        **kwargs,
+    )
+
+    with SMBFileTransaction(raw) as transaction:
+        query_info(transaction, FileStreamInformation, output_buffer_length=MAX_PAYLOAD_SIZE)
+
+    return [s["stream_name"].get_value() for s in transaction.results[0]]
+
+
 def _delete(raw_type, path, **kwargs):
     # Ensures we delete the symlink (if present) and don't follow it down.
     co = CreateOptions.FILE_OPEN_REPARSE_POINT

tests/test_smbclient_os.py

@@ -834,10 +834,7 @@ def test_open_file_with_ads(smb_share):
 
     assert smbclient.listdir(smb_share) == ["file.txt"]
 
-    with smbclient.open_file(filename, buffering=0, mode="rb") as fd, SMBFileTransaction(fd) as trans:
-        query_info(trans, FileStreamInformation, output_buffer_length=1024)
-
-    actual = sorted([s["stream_name"].get_value() for s in trans.results[0]])
+    actual = sorted(smbclient.liststreams(filename))
     assert actual == ["::$DATA", ":ads:$DATA"]