Replace urlsplit with a new parser

Author: irene-sheen-reef

b2/_internal/_utils/uri.py

@@ -11,7 +11,6 @@
 
 import dataclasses
 import pathlib
-import urllib.parse
 from functools import singledispatchmethod
 from pathlib import Path
 from typing import Sequence
@@ -24,6 +23,7 @@
 )
 from b2sdk.v3.exception import B2Error
 
+from b2._internal._utils import uriparse
 from b2._internal._utils.python_compat import removeprefix
 
 
@@ -92,7 +92,7 @@ def parse_uri(uri: str, *, allow_all_buckets: bool = False) -> Path | B2URI | B2
     """
     if not uri:
         raise ValueError('URI cannot be empty')
-    parsed = urllib.parse.urlsplit(uri)
+    parsed = uriparse.b2_urlsplit(uri)
     if parsed.scheme == '':
         return pathlib.Path(uri)
     return _parse_b2_uri(uri, parsed, allow_all_buckets=allow_all_buckets)
@@ -110,34 +110,29 @@ def parse_b2_uri(
     :return: B2 URI
     :raises ValueError: if the URI is invalid
     """
-    parsed = urllib.parse.urlsplit(uri)
+    parsed = uriparse.b2_urlsplit(uri)
     return _parse_b2_uri(uri, parsed, allow_all_buckets=allow_all_buckets, allow_b2id=allow_b2id)
 
 
 def _parse_b2_uri(
     uri,
-    parsed: urllib.parse.SplitResult,
+    parsed: uriparse.SplitB2Result,
     *,
     allow_all_buckets: bool = False,
     allow_b2id: bool = True,
 ) -> B2URI | B2FileIdURI:
     if parsed.scheme in ('b2', 'b2id'):
-        path = urllib.parse.urlunsplit(parsed._replace(scheme='', netloc=''))
         if not parsed.netloc:
             if allow_all_buckets:
-                if path:
+                if parsed.path:
                     raise ValueError(
-                        f"Invalid B2 URI: all buckets URI doesn't allow non-empty path, but {path!r} was provided"
+                        f"Invalid B2 URI: all buckets URI doesn't allow non-empty path, but {parsed.path!r} was provided"
                     )
                 return B2URI(bucket_name='')
             raise ValueError(f'Invalid B2 URI: {uri!r}')
-        elif parsed.password or parsed.username:
-            raise ValueError(
-                'Invalid B2 URI: credentials passed using `user@password:` syntax is not supported in URI'
-            )
 
         if parsed.scheme == 'b2':
-            return B2URI(bucket_name=parsed.netloc, path=removeprefix(path, '/'))
+            return B2URI(bucket_name=parsed.netloc, path=removeprefix(parsed.path, '/'))
         elif parsed.scheme == 'b2id' and allow_b2id:
             return B2FileIdURI(file_id=parsed.netloc)
     else:

b2/_internal/_utils/uriparse.py

@@ -0,0 +1,76 @@
+######################################################################
+#
+# File: b2/_internal/_utils/uriparse.py
+#
+# Copyright 2025 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################
+import re
+
+_CONTROL_CHARACTERS_AND_SPACE = '\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r\x0e\x0f\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f '
+_B2_URL_RE = re.compile(
+    r"""
+    (
+        (?P<scheme>[a-z0-9]+)
+        ://
+    )?                       # the delimiter is optional if there is no scheme defined
+    (?P<netloc>[-a-z0-9]*)   # scheme and
+    (?P<path>\.{0,2}(/.*)?)            # everything else from the first / is part of the path
+    """,
+    re.VERBOSE | re.IGNORECASE,
+)
+
+
+class SplitB2Result:
+    def __init__(self, scheme, netloc, path):
+        self._parts = (scheme, netloc, path)
+
+    def __iter__(self):
+        return iter(self._parts)
+
+    def __getitem__(self, key):
+        return self._parts[key]
+
+    def __len__(self):
+        return len(self._parts)
+
+    def replace(self, scheme=None, netloc=None, path=None):
+        return SplitB2Result(
+            scheme=self.scheme if scheme is None else scheme,
+            netloc=self.netloc if netloc is None else netloc,
+            path=self.path if path is None else path,
+        )
+
+    @property
+    def scheme(self):
+        return self._parts[0]
+
+    @property
+    def netloc(self):
+        return self._parts[1]
+
+    @property
+    def path(self):
+        return self._parts[2]
+
+    def __repr__(self):
+        return f'SplitB2Result(scheme={self.scheme!r}, netloc={self.netloc!r}, path={self.path!r})'
+
+
+def b2_urlsplit(url: str) -> SplitB2Result:
+    # clean the url
+    url = url.lstrip(_CONTROL_CHARACTERS_AND_SPACE)
+    for i in ['\n', '\r', '\t']:
+        url.replace(i, '')
+
+    match = _B2_URL_RE.fullmatch(url)
+    if not match:
+        raise ValueError(f'Invalid B2 URI: {url!r}')
+
+    scheme = (match.group('scheme') or '').lower()
+    netloc = match.group('netloc') or ''
+    path = match.group('path') or ''
+
+    return SplitB2Result(scheme, netloc, path)

changelog.d/1090.fixed.md

@@ -0,0 +1 @@
+handle filenames starting with / or ending with # or ?

test/unit/_utils/test_uri.py

@@ -61,13 +61,17 @@ def test_b2fileuri_str():
     [
         ('some/local/path', Path('some/local/path')),
         ('./some/local/path', Path('some/local/path')),
+        ('.', Path('')),
         ('b2://bucket', B2URI(bucket_name='bucket')),
         ('b2://bucket/', B2URI(bucket_name='bucket')),
         ('b2://bucket/path/to/dir/', B2URI(bucket_name='bucket', path='path/to/dir/')),
         ('b2id://file123', B2FileIdURI(file_id='file123')),
         ('b2://bucket/wild[card]', B2URI(bucket_name='bucket', path='wild[card]')),
         ('b2://bucket/wild?card', B2URI(bucket_name='bucket', path='wild?card')),
         ('b2://bucket/special#char', B2URI(bucket_name='bucket', path='special#char')),
+        ('b2://bucket/special#', B2URI(bucket_name='bucket', path='special#')),
+        ('b2://bucket/special?', B2URI(bucket_name='bucket', path='special?')),
+        ('b2://bucket//special', B2URI(bucket_name='bucket', path='/special')),
     ],
 )
 def test_parse_uri(uri, expected):
@@ -94,11 +98,11 @@ def test_parse_uri__allow_all_buckets():
         # Test cases for B2 URIs with credentials
         (
             'b2://user@password:bucket/path',
-            'Invalid B2 URI: credentials passed using `user@password:` syntax is not supported in URI',
+            "Invalid B2 URI: 'b2://user@password:bucket/path'",
         ),
         (
             'b2id://user@password:file123',
-            'Invalid B2 URI: credentials passed using `user@password:` syntax is not supported in URI',
+            "Invalid B2 URI: 'b2id://user@password:file123'",
         ),
         # Test cases for unsupported URI schemes
         ('unknown://bucket/path', "Unsupported URI scheme: 'unknown'"),