diff --git a/aikido_zen/context/asgi/__init__.py b/aikido_zen/context/asgi/__init__.py index fa59fe6d2..71f047a4b 100644 --- a/aikido_zen/context/asgi/__init__.py +++ b/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")) diff --git a/aikido_zen/context/asgi/init_test.py b/aikido_zen/context/asgi/init_test.py index 53b54bd85..d4af745ae 100644 --- a/aikido_zen/context/asgi/init_test.py +++ b/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 = { diff --git a/aikido_zen/context/parse_cookies.py b/aikido_zen/context/parse_cookies.py index ddc729ca1..bbc5a2d9c 100644 --- a/aikido_zen/context/parse_cookies.py +++ b/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 diff --git a/aikido_zen/context/parse_cookies_test.py b/aikido_zen/context/parse_cookies_test.py index 72008c48e..dd0533bbf 100644 --- a/aikido_zen/context/parse_cookies_test.py +++ b/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