Merge branch 'main' into release-v1.0.15

aikido_zen/context/__init__.py

@@ -10,8 +10,9 @@
 from aikido_zen.helpers.logging import logger
 from .wsgi import set_wsgi_attributes_on_context
 from .asgi import set_asgi_attributes_on_context
+from .extract_route_params import extract_route_params
 
-UINPUT_SOURCES = ["body", "cookies", "query", "headers", "xml"]
+UINPUT_SOURCES = ["body", "cookies", "query", "headers", "xml", "route_params"]
 current_context = contextvars.ContextVar("current_context", default=None)
 
 WSGI_SOURCES = ["django", "flask"]
@@ -56,6 +57,7 @@ def __init__(self, context_obj=None, body=None, req=None, source=None):
 
         # Define variables using parsed request :
         self.route = build_route_from_url(self.url)
+        self.route_params = extract_route_params(self.url)
         self.subdomains = get_subdomains_from_url(self.url)
 
     def __reduce__(self):
@@ -76,6 +78,7 @@ def __reduce__(self):
                     "user": self.user,
                     "xml": self.xml,
                     "outgoing_req_redirects": self.outgoing_req_redirects,
+                    "route_params": self.route_params,
                 },
                 None,
                 None,

aikido_zen/context/extract_route_params.py

@@ -0,0 +1,31 @@
+"""Exports extract_route_params function"""
+
+from urllib.parse import quote, unquote
+from aikido_zen.helpers.try_parse_url_path import try_parse_url_path
+from aikido_zen.helpers.build_route_from_url import replace_url_segment_with_param
+
+
+def extract_route_params(url):
+    """Will try and build an array of user input based on the url"""
+    results = []
+    try:
+        path = try_parse_url_path(url)
+        segments = path.split("/")
+        for segment in segments:
+            segment = unquote(segment)
+            if segment.isalnum():
+                continue  # Ignore alphanumerical parts of the url
+
+            if segment is not quote(segment):
+                results.append(segment)  # This is not a standard piece of the URL
+            elif replace_url_segment_with_param(segment) is not segment:
+                results.append(segment)  # Might be a secret, a hash, ...
+
+        if len(results) > 0 or "." in unquote(path):
+            # There are already phishy parts of the url OR
+            # urldecoded path contains dots, which is uncommon and could point to path traversal.
+            results.append(path[1:])  # Add path after slash as user input
+
+    except Exception:
+        pass
+    return results

aikido_zen/context/extract_route_params_test.py

@@ -0,0 +1,112 @@
+import pytest
+from .extract_route_params import extract_route_params
+
+
+def test_with_urlencoded_urls():
+    url1 = "http://localhost:8080/app/shell/ls%20-la"
+    assert extract_route_params(url1) == ["ls -la", "app/shell/ls%20-la"]
+
+    url2 = "http://localhost:8080/app/shell/ls -la"
+    assert extract_route_params(url2) == ["ls -la", "app/shell/ls -la"]
+
+
+def test_uses_keys():
+    url = "http://localhost:8080/app/shell/[email protected]/017shell/127.0.0.1/"
+    assert extract_route_params(url) == [
+        "[email protected]",
+        "127.0.0.1",
+        "app/shell/[email protected]/017shell/127.0.0.1/",
+    ]
+
+
+def test_normal_urls():
+    assert extract_route_params("http://localhost:8080/a/b/abc2393027def/def") == []
+
+
+def test_with_empty_route():
+    url1 = "http://localhost:8080"
+    assert extract_route_params(url1) == []
+
+    url2 = "http://localhost:8080"
+    assert extract_route_params(url2) == []
+
+
+def test_special_characters():
+    url1 = "http://localhost:8080/app/shell/!@#$%^&*()"  # Everything past hashtag is not url anymore
+    assert extract_route_params(url1) == ["!@", "app/shell/!@"]
+
+    url2 = "http://localhost:8080/app/shell/space test"
+    assert extract_route_params(url2) == ["space test", "app/shell/space test"]
+
+    url3 = "http://localhost:8080/app/shell/hello%20world"
+    assert extract_route_params(url3) == ["hello world", "app/shell/hello%20world"]
+
+
+def test_numeric_segments():
+    # Alphanum is ignored:
+    url1 = "http://localhost:8080/app/shell/12345"
+    assert extract_route_params(url1) == []
+
+    url2 = "http://localhost:8080/app/shell/67890/abc"
+    assert extract_route_params(url2) == []
+
+
+def test_mixed_segments():
+    url1 = "http://localhost:8080/app/shell/abc123/!@#"
+    assert extract_route_params(url1) == ["!@", "app/shell/abc123/!@"]
+
+    url2 = "http://localhost:8080/app/shell/abc/123/!@#"
+    assert extract_route_params(url2) == ["!@", "app/shell/abc/123/!@"]
+
+
+def test_encoded_and_unencoded():
+    url1 = "http://localhost:8080/app/shell/%E2%9C%93"
+    assert extract_route_params(url1) == ["✓", "app/shell/%E2%9C%93"]
+
+    url2 = "http://localhost:8080/app/shell/%E2%9C%93/normal"
+    assert extract_route_params(url2) == ["✓", "app/shell/%E2%9C%93/normal"]
+
+
+def test_no_params():
+    url1 = "http://localhost:8080/app/shell/"
+    assert extract_route_params(url1) == []
+
+    url2 = "http://localhost:8080/app/"
+    assert extract_route_params(url2) == []
+
+
+def test_edge_cases():
+    url1 = "http://localhost:8080/app/shell/.."
+    assert extract_route_params(url1) == ["..", "app/shell/.."]
+
+    url2 = "http://localhost:8080/app/shell/./"
+    assert extract_route_params(url2) == ["app/shell/./"]
+
+
+def test_long_urls():
+    url1 = "http://localhost:8080/app./shell/" + "a" * 1000
+    assert extract_route_params(url1) == ["app.", "app./shell/" + "a" * 1000]
+
+    url2 = "http://localhost:8080/app./shell/" + "b" * 1000 + "/c" * 1000
+    assert extract_route_params(url2) == [
+        "app.",
+        "app./shell/" + "b" * 1000 + "/c" * 1000,
+    ]
+
+
+def test_query_parameters():
+    # Test query parameters are ignored:
+    url1 = "http://localhost:8080/app/./shell/?param=value"
+    assert extract_route_params(url1) == ["app/./shell/"]
+
+    url2 = "http://localhost:8080/app/./shell/?key1=value1&key2=value2"
+    assert extract_route_params(url2) == ["app/./shell/"]
+
+
+def test_fragment_identifiers():
+    # Fragments should be ignored:
+    url1 = "http://localhost:8080/app/./shell/#section1"
+    assert extract_route_params(url1) == ["app/./shell/"]
+
+    url2 = "http://localhost:8080/app/shell/#/path/to/resource"
+    assert extract_route_params(url2) == []

aikido_zen/context/init_test.py

@@ -53,6 +53,7 @@ def test_wsgi_context_1():
         "parsed_userinput": {},
         "xml": {},
         "outgoing_req_redirects": [],
+        "route_params": [],
     }
 
 
@@ -92,6 +93,7 @@ def test_wsgi_context_2():
         "parsed_userinput": {},
         "xml": {},
         "outgoing_req_redirects": [],
+        "route_params": [],
     }
 
 

aikido_zen/sources/django/run_init_stage.py

@@ -18,6 +18,12 @@ def run_init_stage(request):
         if len(body) == 0:
             # E.g. XML Data
             body = request.body
+        if len(body) == 0:
+            # During a GET request, django leaves the body as an empty byte string (e.g. `b''`).
+            # When an attack is detected, this body needs to be serialized which would fail.
+            # So a byte string gets converted into a string to stop that from happening.
+            body = ""
+            # Set body to an empty string.
 
         context = None
         if hasattr(request, "scope"):  # This request is an ASGI request

aikido_zen/sources/flask.py

@@ -50,8 +50,12 @@ def extract_and_save_data_from_flask_request(req):
                 context.body = req.form
             else:
                 context.body = req.data.decode("utf-8")
+
+            if getattr(req, "view_args"):
+                context.route_params = dict(req.view_args)
             context.cookies = req.cookies.to_dict()
             context.set_as_current_context()
+
     except Exception as e:
         logger.debug("Exception occured whilst extracting flask body data: %s", e)
 

aikido_zen/vulnerabilities/init_test.py

@@ -42,6 +42,7 @@ def get_context():
         body={"test_input_sql": "doggoss2', TRUE"},
         source="flask",
     )
+    context.route_params = {"test_input2": "cattss2', TRUE"}
     return context
 
 
@@ -101,6 +102,20 @@ def test_sql_injection(caplog, get_context, monkeypatch):
         )
 
 
+def test_sql_injection_with_route_params(caplog, get_context, monkeypatch):
+    from aikido_zen.vulnerabilities.sql_injection.dialects import MySQL
+
+    get_context.set_as_current_context()
+    cache = ThreadCache()
+    monkeypatch.setenv("AIKIDO_BLOCKING", "1")
+    with pytest.raises(AikidoSQLInjection):
+        run_vulnerability_scan(
+            kind="sql_injection",
+            op="test_op",
+            args=("INSERT * INTO VALUES ('cattss2', TRUE);", MySQL()),
+        )
+
+
 def test_sql_injection_with_comms(caplog, get_context, monkeypatch):
     from aikido_zen.vulnerabilities.sql_injection.dialects import MySQL
 

end2end/django_mysql_test.py

@@ -4,8 +4,8 @@
 from .server.check_events_from_mock import fetch_events_from_mock, validate_started_event, filter_on_event_type, validate_heartbeat
 
 # e2e tests for django_mysql sample app
-post_url_fw = "http://localhost:8080/app/create"
-post_url_nofw = "http://localhost:8081/app/create"
+base_url_fw = "http://localhost:8080/app"
+base_url_nofw = "http://localhost:8081/app"
 
 def test_firewall_started_okay():
     events = fetch_events_from_mock("http://localhost:5000")
@@ -15,18 +15,18 @@ def test_firewall_started_okay():
 
 def test_safe_response_with_firewall():
     dog_name = "Bobby Tables"
-    res = requests.post(post_url_fw, data={'dog_name': dog_name})
+    res = requests.post(base_url_fw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 200
 
 def test_safe_response_without_firewall():
     dog_name = "Bobby Tables"
-    res = requests.post(post_url_nofw, data={'dog_name': dog_name})
+    res = requests.post(base_url_nofw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 200
 
 
 def test_dangerous_response_with_firewall():
     dog_name = 'Dangerous bobby", 1); -- '
-    res = requests.post(post_url_fw, data={'dog_name': dog_name})
+    res = requests.post(base_url_fw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 500
     time.sleep(5) # Wait for attack to be reported
     events = fetch_events_from_mock("http://localhost:5000")
@@ -45,9 +45,31 @@ def test_dangerous_response_with_firewall():
         'user': None
     }
 
+def test_dangerous_response_with_firewall_shell():
+    dog_name = 'Dangerous bobby", 1); -- '
+    res = requests.get(base_url_fw + "/shell/ls -la")
+    assert res.status_code == 500
+    time.sleep(5) # Wait for attack to be reported
+    events = fetch_events_from_mock("http://localhost:5000")
+    attacks = filter_on_event_type(events, "detected_attack")
+
+    assert len(attacks) == 2
+    del attacks[0] # Previous attack
+    del attacks[0]["attack"]["stack"]
+    assert attacks[0]["attack"] == {
+        "blocked": True,
+        "kind": "shell_injection",
+        'metadata': {'command': 'ls -la'},
+        'operation': 'subprocess.Popen',
+        'pathToPayload': '.[0]',
+        'payload': '"ls -la"',
+        'source': "route_params",
+        'user': None
+    }
+
 def test_dangerous_response_without_firewall():
     dog_name = 'Dangerous bobby", 1); -- '
-    res = requests.post(post_url_nofw, data={'dog_name': dog_name})
+    res = requests.post(base_url_nofw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 200
 
 def test_initial_heartbeat():
@@ -63,5 +85,5 @@ def test_initial_heartbeat():
             "method": "POST",
             "path": "/app/create"
         }], 
-        {"aborted":0,"attacksDetected":{"blocked":1,"total":1},"total":0}
+        {"aborted":0,"attacksDetected":{"blocked":2,"total":2},"total":0}
     )

end2end/flask_mysql_test.py

@@ -3,8 +3,8 @@
 import time
 from .server.check_events_from_mock import fetch_events_from_mock, validate_started_event, filter_on_event_type
 # e2e tests for flask_mysql sample app
-post_url_fw = "http://localhost:8086/create"
-post_url_nofw = "http://localhost:8087/create"
+base_url_fw = "http://localhost:8086"
+base_url_nofw = "http://localhost:8087"
 
 def test_firewall_started_okay():
     events = fetch_events_from_mock("http://localhost:5000")
@@ -14,21 +14,21 @@ def test_firewall_started_okay():
 
 def test_safe_response_with_firewall():
     dog_name = "Bobby Tables"
-    res = requests.post(post_url_fw, data={'dog_name': dog_name})
+    res = requests.post(base_url_fw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 200
 
 
 def test_safe_response_without_firewall():
     dog_name = "Bobby Tables"
-    res = requests.post(post_url_nofw, data={'dog_name': dog_name})
+    res = requests.post(base_url_nofw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 200
 
 
 def test_dangerous_response_with_firewall():
     events = fetch_events_from_mock("http://localhost:5000")
     assert len(filter_on_event_type(events, "detected_attack")) == 0
     dog_name = 'Dangerous bobby", 1); -- '
-    res = requests.post(post_url_fw, data={'dog_name': dog_name})
+    res = requests.post(base_url_fw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 500
 
     time.sleep(5) # Wait for attack to be reported
@@ -48,9 +48,33 @@ def test_dangerous_response_with_firewall():
         'user': None
     }
 
+def test_dangerous_response_with_firewall_route_params():
+    events = fetch_events_from_mock("http://localhost:5000")
+    assert len(filter_on_event_type(events, "detected_attack")) == 1
+    res = requests.get(base_url_fw + "/shell/ls -la")
+    assert res.status_code == 500
+
+    time.sleep(5) # Wait for attack to be reported
+    events = fetch_events_from_mock("http://localhost:5000")
+    attacks = filter_on_event_type(events, "detected_attack")
+
+    assert len(attacks) == 2
+    del attacks[0]
+    del attacks[0]["attack"]["stack"]
+    assert attacks[0]["attack"] == {
+        "blocked": True,
+        "kind": "shell_injection",
+        'metadata': {'command': 'ls -la'},
+        'operation': 'subprocess.Popen',
+        'pathToPayload': '.command',
+        'payload': '"ls -la"',
+        'source': "route_params",
+        'user': None
+    }
+
 
 def test_dangerous_response_without_firewall():
     dog_name = 'Dangerous bobby", 1); -- '
-    res = requests.post(post_url_nofw, data={'dog_name': dog_name})
+    res = requests.post(base_url_nofw + "/create", data={'dog_name': dog_name})
     assert res.status_code == 200
 

sample-apps/django-mysql/sample_app/urls.py

@@ -5,5 +5,6 @@
 urlpatterns = [
     path("", views.index, name="index"),
     path("dogpage/<int:dog_id>", views.dog_page, name="dog_page"),
+    path("shell/<str:user_command>", views.shell_url, name="shell"),
     path("create", views.create_dogpage, name="create")
 ]