Scan every request cookie, including ones SimpleCookie rejected

aikido_zen/context/asgi/__init__.py

@@ -17,12 +17,8 @@ def set_asgi_attributes_on_context(context, scope):
     context.method = scope["method"]
     context.headers = normalize_asgi_headers(scope["headers"])
 
-    if context.headers.get_header("COOKIE"):
-        # Right now just use the first Cookie header, will change later to use
-        # framework definition for cookies.
-        context.cookies = parse_cookies(context.headers.get_header("COOKIE"))
-    else:
-        context.cookies = {}
+    # Parse every Cookie header in both its decoded readings, so no value is missed.
+    context.cookies = parse_cookies(*context.headers.get("COOKIE", []))
 
     context.url = build_url_from_asgi(scope)
     context.query = parse_qs(scope["query_string"].decode("utf-8", errors="replace"))

aikido_zen/context/asgi/init_test.py

@@ -1,6 +1,10 @@
 import pytest
 from urllib.parse import parse_qs
+from aikido_zen.context import Context as ZenContext
 from aikido_zen.context.asgi import set_asgi_attributes_on_context
+from aikido_zen.vulnerabilities.sql_injection.context_contains_sql_injection import (
+    context_contains_sql_injection,
+)
 
 
 class Context:
@@ -180,6 +184,80 @@ def test_asgi_scope_invalid_utf8():
     assert context5.url == "http://192.168.0.5:80/search"
 
 
+def test_sql_injection_in_a_cookie_next_to_a_rejected_cookie_is_detected():
+    payload = "1' OR '1'='1"
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "headers": [(b"cookie", f"id={payload}; ,2=2".encode())],
+        "query_string": b"",
+        "client": ["6.6.6.6"],
+        "server": ["192.168.0.6", 80],
+        "scheme": "http",
+        "root_path": "",
+        "path": "/users",
+    }
+    context = ZenContext(req=scope, source="starlette")
+    query = f"SELECT * FROM users WHERE id = '{payload}'"
+    assert context.cookies == {"id": payload, ",2": "2"}
+    assert context_contains_sql_injection(query, "select", context, "sqlite")
+
+
+def test_sql_injection_in_an_earlier_cookie_header_is_detected():
+    payload = "1' OR '1'='1"
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "headers": [
+            (b"cookie", f"id={payload}".encode()),
+            (b"cookie", b"sid=abc"),
+        ],
+        "query_string": b"",
+        "client": ["6.6.6.6"],
+        "server": ["192.168.0.6", 80],
+        "scheme": "http",
+        "root_path": "",
+        "path": "/users",
+    }
+    context = ZenContext(req=scope, source="starlette")
+    query = f"SELECT * FROM users WHERE id = '{payload}'"
+    assert context.cookies == {"id": payload, "sid": "abc"}
+    assert context_contains_sql_injection(query, "select", context, "sqlite")
+
+
+@pytest.mark.parametrize(
+    "cookie_headers, app_value",
+    [
+        ([b"id=1' OR '1'='1; id=1"], "1' OR '1'='1"),
+        ([b"id=1; id=1' OR '1'='1"], "1' OR '1'='1"),
+        ([b"id=1' OR '1'='1", b"id=1"], "1' OR '1'='1"),
+        ([b"id=1", b"id=1' OR '1'='1"], "1' OR '1'='1"),
+        (["id=café' OR '1'='1".encode()], "café' OR '1'='1"),
+        (["id=café' OR '1'='1".encode()], "café' OR '1'='1"),
+    ],
+)
+def test_sql_injection_in_any_cookie_value_the_app_can_read_is_detected(
+    cookie_headers, app_value
+):
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "headers": [(b"cookie", value) for value in cookie_headers],
+        "query_string": b"",
+        "client": ["6.6.6.6"],
+        "server": ["192.168.0.6", 80],
+        "scheme": "http",
+        "root_path": "",
+        "path": "/users",
+    }
+    context = ZenContext(req=scope, source="starlette")
+    query = f"SELECT * FROM users WHERE id = '{app_value}'"
+    assert context_contains_sql_injection(query, "select", context, "sqlite")
+    assert not context_contains_sql_injection(
+        "SELECT * FROM users WHERE id = '1'", "select", context, "sqlite"
+    )
+
+
 def test_valid_utf8_query_matches_strict_decode():
     raw = "q=café' OR 1=1 --&b=d".encode("utf-8")
     scope = {

aikido_zen/context/parse_cookies.py

@@ -1,21 +1,29 @@
 """
-Exports `parse_cookies` function which parses WSGI cookies
+Exports `parse_cookies` function which parses request cookies
 """
 
-from http.cookies import SimpleCookie, CookieError
+from http.cookies import _unquote
 
 
-def parse_cookies(cookie_str):
-    """Parse cookie string from headers"""
-    cookie_dict = {}
-    cookies = SimpleCookie()
-    try:
-        cookies.load(cookie_str)
-    except CookieError:
-        # Invalid keys can cause errors, we want to ignore them
-        pass
-
-    for key, morsel in cookies.items():
-        cookie_dict[key] = morsel.value
-
-    return cookie_dict
+def parse_cookies(*cookie_strs):
+    """Parse cookie strings from headers"""
+    # Unlike SimpleCookie, this keeps every cookie, even one with a space, a comma or JSON,
+    # and every value of a repeated name, because the app may read any of them.
+    cookies = {}
+    for cookie_str in cookie_strs:
+        for chunk in cookie_str.split(";"):
+            if "=" in chunk:
+                key, val = chunk.split("=", 1)
+            else:
+                key, val = "", chunk
+            key, val = key.strip(), _unquote(val.strip())
+            if not key and not val:
+                continue
+            if key not in cookies:
+                cookies[key] = val
+            elif isinstance(cookies[key], list):
+                if val not in cookies[key]:
+                    cookies[key].append(val)
+            elif cookies[key] != val:
+                cookies[key] = [cookies[key], val]
+    return cookies

aikido_zen/context/parse_cookies_test.py

@@ -24,3 +24,43 @@ def test_parse_cookies_with_special_characters():
     cookie_str = "sessionid=abc123; user=John@Doe; theme=dark&light"
     expected = {"sessionid": "abc123", "user": "John@Doe", "theme": "dark&light"}
     assert parse_cookies(cookie_str) == expected
+
+
+@pytest.mark.parametrize(
+    "cookie_str, expected",
+    [
+        (
+            'user={"username":{"$regex":".*"}}; ,2=2',
+            {"user": '{"username":{"$regex":".*"}}', ",2": "2"},
+        ),
+        ('user={"a":1}', {"user": '{"a":1}'}),
+        ("id=1' OR '1'='1", {"id": "1' OR '1'='1"}),
+        (",2=2; id=abc", {",2": "2", "id": "abc"}),
+        ('user={"a":1}; sid=abc', {"user": '{"a":1}', "sid": "abc"}),
+        (
+            'sid=abc; user={"a":1}; theme=dark',
+            {"sid": "abc", "user": '{"a":1}', "theme": "dark"},
+        ),
+        ("novalue", {"": "novalue"}),
+    ],
+)
+def test_parse_cookies_keeps_cookies_simplecookie_rejects(cookie_str, expected):
+    assert parse_cookies(cookie_str) == expected
+
+
+@pytest.mark.parametrize(
+    "cookie_strs, expected",
+    [
+        (["id=a; id=b"], {"id": ["a", "b"]}),
+        (["id=a", "id=b"], {"id": ["a", "b"]}),
+        (["id=a; sid=1", "id=a"], {"id": "a", "sid": "1"}),
+        (['id="a b"'], {"id": "a b"}),
+        (['id="a\\"b"'], {"id": 'a"b'}),
+        (["name=café"], {"name": "café"}),
+        (["café=1"], {"café": "1"}),
+        (["a; b"], {"": ["a", "b"]}),
+        (["a=1; flag; b=2"], {"a": "1", "": "flag", "b": "2"}),
+    ],
+)
+def test_parse_cookies_keeps_every_value(cookie_strs, expected):
+    assert parse_cookies(*cookie_strs) == expected