fix(asyncio): reject connection URLs missing the scheme separator

redis/asyncio/connection.py

@@ -1766,6 +1766,15 @@ class ConnectKwargs(TypedDict, total=False):
 
 
 def parse_url(url: str) -> ConnectKwargs:
+    # Scheme names are case-insensitive (RFC 3986), so normalize before the
+    # prefix check; the "://" is required so a URL like "redis:foo" (which
+    # urlparse would still report as the "redis" scheme) is rejected.
+    if not url.lower().startswith(("redis://", "rediss://", "unix://")):
+        raise ValueError(
+            "Redis URL must specify one of the following schemes "
+            "(redis://, rediss://, unix://)"
+        )
+
     parsed: ParseResult = urlparse(url)
     kwargs: ConnectKwargs = {}
 
@@ -1795,7 +1804,7 @@ def parse_url(url: str) -> ConnectKwargs:
             kwargs["path"] = unquote(parsed.path)
         kwargs["connection_class"] = UnixDomainSocketConnection
 
-    elif parsed.scheme in ("redis", "rediss"):
+    else:  # implied:  parsed.scheme in ("redis", "rediss")
         if parsed.hostname:
             kwargs["host"] = unquote(parsed.hostname)
         if parsed.port:
@@ -1812,12 +1821,6 @@ def parse_url(url: str) -> ConnectKwargs:
         if parsed.scheme == "rediss":
             kwargs["connection_class"] = SSLConnection
 
-    else:
-        valid_schemes = "redis://, rediss://, unix://"
-        raise ValueError(
-            f"Redis URL must specify one of the following schemes ({valid_schemes})"
-        )
-
     return kwargs
 
 

redis/connection.py

@@ -2328,11 +2328,10 @@ def parse_ssl_verify_flags(value):
 
 
 def parse_url(url):
-    if not (
-        url.startswith("redis://")
-        or url.startswith("rediss://")
-        or url.startswith("unix://")
-    ):
+    # Scheme names are case-insensitive (RFC 3986), so normalize before the
+    # prefix check; the "://" is required so a URL like "redis:foo" (which
+    # urlparse would still report as the "redis" scheme) is rejected.
+    if not url.lower().startswith(("redis://", "rediss://", "unix://")):
         raise ValueError(
             "Redis URL must specify one of the following "
             "schemes (redis://, rediss://, unix://)"

tests/test_asyncio/test_connection_pool.py

@@ -662,6 +662,28 @@ def test_invalid_scheme_raises_error(self):
             "(redis://, rediss://, unix://)"
         )
 
+    def test_invalid_scheme_raises_error_when_double_slash_missing(self):
+        with pytest.raises(ValueError) as cm:
+            redis.ConnectionPool.from_url("redis:foo.bar.com:12345")
+        assert str(cm.value) == (
+            "Redis URL must specify one of the following schemes "
+            "(redis://, rediss://, unix://)"
+        )
+
+    def test_uppercase_scheme_is_accepted(self):
+        # URL schemes are case-insensitive (RFC 3986)
+        pool = redis.ConnectionPool.from_url("REDIS://my.host")
+        assert pool.connection_class == redis.Connection
+        assert_kwargs_subset(pool.connection_kwargs, {"host": "my.host"})
+
+        ssl_pool = redis.ConnectionPool.from_url("REDISS://my.host")
+        assert ssl_pool.connection_class == redis.SSLConnection
+        assert_kwargs_subset(ssl_pool.connection_kwargs, {"host": "my.host"})
+
+        unix_pool = redis.ConnectionPool.from_url("UNIX:///tmp/redis.sock")
+        assert unix_pool.connection_class == redis.UnixDomainSocketConnection
+        assert_kwargs_subset(unix_pool.connection_kwargs, {"path": "/tmp/redis.sock"})
+
 
 @pytest.mark.fixed_client
 class TestBlockingConnectionPoolURLParsing:

tests/test_connection_pool.py

@@ -611,6 +611,20 @@ def test_invalid_scheme_raises_error_when_double_slash_missing(self):
             "(redis://, rediss://, unix://)"
         )
 
+    def test_uppercase_scheme_is_accepted(self):
+        # URL schemes are case-insensitive (RFC 3986)
+        pool = redis.ConnectionPool.from_url("REDIS://my.host")
+        assert pool.connection_class == redis.Connection
+        assert_kwargs_subset(pool.connection_kwargs, {"host": "my.host"})
+
+        ssl_pool = redis.ConnectionPool.from_url("REDISS://my.host")
+        assert ssl_pool.connection_class == redis.SSLConnection
+        assert_kwargs_subset(ssl_pool.connection_kwargs, {"host": "my.host"})
+
+        unix_pool = redis.ConnectionPool.from_url("UNIX:///tmp/redis.sock")
+        assert unix_pool.connection_class == redis.UnixDomainSocketConnection
+        assert_kwargs_subset(unix_pool.connection_kwargs, {"path": "/tmp/redis.sock"})
+
 
 @pytest.mark.fixed_client
 class TestBlockingConnectionPoolURLParsing: