sync

Author: thatstoasty

lightbug_http/strings.mojo

@@ -150,3 +150,12 @@ fn to_string(owned bytes: Bytes) -> String:
     if bytes[-1] != 0:
         bytes.append(0)
     return String(bytes^)
+
+
+fn find_all(s: String, sub_str: String) -> List[Int]:
+    match_idxs = List[Int]()
+    var current_idx: Int = s.find(sub_str)
+    while current_idx > -1:
+        match_idxs.append(current_idx)
+        current_idx = s.find(sub_str, start=current_idx + 1)
+    return match_idxs^

lightbug_http/uri.mojo

@@ -3,6 +3,7 @@ from memory import Span
 from collections import Optional, Dict
 from lightbug_http.io.bytes import Bytes, bytes, ByteReader, Constant
 from lightbug_http.strings import (
+    find_all,
     strSlash,
     strHttp11,
     strHttp10,
@@ -12,19 +13,81 @@ from lightbug_http.strings import (
     https,
 )
 
+
+fn unquote[expand_plus: Bool = False](input_str: String, disallowed_escapes: List[String] = List[String]()) -> String:
+    var encoded_str = input_str.replace(QueryDelimiters.PLUS_ESCAPED_SPACE, " ") if expand_plus else input_str
+
+    var percent_idxs: List[Int] = find_all(encoded_str, URIDelimiters.CHAR_ESCAPE)
+
+    if len(percent_idxs) < 1:
+        return encoded_str
+
+    var sub_strings = List[String]()
+
+    var current_idx = 0
+    var slice_start = 0
+    var slice_end = 0
+
+    var str_bytes = List[UInt8]()
+    while current_idx < len(percent_idxs):
+        slice_end = percent_idxs[current_idx]
+        sub_strings.append(encoded_str[slice_start:slice_end])
+
+        var current_offset = slice_end
+        while current_idx < len(percent_idxs):
+            var char_byte = -1
+            if (current_offset + 3) <= len(encoded_str):
+                try:
+                    char_byte = atol(
+                        encoded_str[current_offset + 1 : current_offset + 3],
+                        base=16,
+                    )
+                except:
+                    pass
+
+            if char_byte < 0:
+                break
+
+            str_bytes.append(char_byte)
+
+            if percent_idxs[current_idx + 1] != (current_offset + 3):
+                current_offset += 3
+                break
+
+            current_idx += 1
+            current_offset = percent_idxs[current_idx]
+
+        if len(str_bytes) > 0:
+            str_bytes.append(0x00)
+            var sub_str_from_bytes = String(str_bytes)
+            for disallowed in disallowed_escapes:
+                sub_str_from_bytes = sub_str_from_bytes.replace(disallowed[], "")
+            sub_strings.append(sub_str_from_bytes)
+            str_bytes.clear()
+
+        slice_start = current_offset
+        current_idx += 1
+
+    sub_strings.append(encoded_str[slice_start:])
+
+    return str("").join(sub_strings)
+
+
 alias QueryMap = Dict[String, String]
 
 
 struct QueryDelimiters:
     alias STRING_START = "?"
     alias ITEM = "&"
     alias ITEM_ASSIGN = "="
+    alias PLUS_ESCAPED_SPACE = "+"
 
 
 struct URIDelimiters:
     alias SCHEMA = "://"
     alias PATH = strSlash
     alias ROOT_PATH = strSlash
+    alias CHAR_ESCAPE = "%"
 
 
 @value
@@ -152,13 +215,12 @@ struct URI(Writable, Stringable, Representable):
 
             for item in query_items:
                 var key_val = item[].split(QueryDelimiters.ITEM_ASSIGN, 1)
+                var key = unquote[expand_plus=True](key_val[0])
 
-                if key_val[0]:
-                    queries[key_val[0]] = ""
+                if key:
+                    queries[key] = ""
                     if len(key_val) == 2:
-                        # TODO: Query values are going to be URI encoded strings and should be decoded as part of the
-                        # query processing
-                        queries[key_val[0]] = key_val[1]
+                        queries[key] = unquote[expand_plus=True](key_val[1])
 
         return URI(
             _original_path=path,

tests/lightbug_http/test_uri.mojo

@@ -62,6 +62,16 @@ def test_uri_parse_https_with_path():
     testing.assert_equal(uri.query_string, empty_string)
 
 
+def test_uri_parse_path_with_encoding():
+    var uri = URI.parse("https://example.com/test%20test/index.html")
+    testing.assert_equal(uri.path, "/test test/index.html")
+
+
+def test_uri_parse_path_with_encoding_ignore_slashes():
+    var uri = URI.parse("https://example.com/trying_to%2F_be_clever/42.html")
+    testing.assert_equal(uri.path, "/trying_to_be_clever/42.html")
+
+
 def test_uri_parse_http_basic():
     var uri = URI.parse("http://example.com")
     testing.assert_equal(uri.scheme, "http")
@@ -106,13 +116,14 @@ def test_uri_parse_multiple_query_parameters():
 
 
 def test_uri_parse_query_with_special_characters():
-    var uri = URI.parse("https://example.com/path?name=John+Doe&email=john%40example.com")
+    var uri = URI.parse("https://example.com/path?name=John+Doe&email=john%40example.com&escaped%40%20name=42")
     testing.assert_equal(uri.scheme, "https")
     testing.assert_equal(uri.host, "example.com")
     testing.assert_equal(uri.path, "/path")
-    testing.assert_equal(uri.query_string, "name=John+Doe&email=john%40example.com")
-    # testing.assert_equal(uri.queries["name"], "John Doe") - fails, contains John+Doe
-    # testing.assert_equal(uri.queries["email"], "john@example.com") - fails, contains john%40example.com
+    testing.assert_equal(uri.query_string, "name=John+Doe&email=john%40example.com&escaped%40%20name=42")
+    testing.assert_equal(uri.queries["name"], "John Doe")
+    testing.assert_equal(uri.queries["email"], "john@example.com")
+    testing.assert_equal(uri.queries["escaped@ name"], "42")
 
 
 def test_uri_parse_empty_query_values():
@@ -139,8 +150,8 @@ def test_uri_parse_complex_query():
 def test_uri_parse_query_with_unicode():
     var uri = URI.parse("http://example.com/search?q=%E2%82%AC&lang=%F0%9F%87%A9%F0%9F%87%AA")
     testing.assert_equal(uri.query_string, "q=%E2%82%AC&lang=%F0%9F%87%A9%F0%9F%87%AA")
-    # testing.assert_equal(uri.queries["q"], "€") - fails, contains %E2%82%AC
-    # testing.assert_equal(uri.queries["lang"], "🇩🇪") - fails, contains %F0%9F%87%A9%F0%9F%87%AA
+    testing.assert_equal(uri.queries["q"], "€")
+    testing.assert_equal(uri.queries["lang"], "🇩🇪")
 
 
 # def test_uri_parse_query_with_fragments():