add an e2e test case for django asgi uvicorn

Author: bitterpanda63

.github/workflows/end2end.yml

@@ -29,6 +29,7 @@ jobs:
           - { name: django-mysql, testfile: end2end/django_mysql_test.py }
           - { name: django-mysql-gunicorn, testfile: end2end/django_mysql_gunicorn_test.py }
           - { name: django-postgres-gunicorn, testfile: end2end/django_postgres_gunicorn_test.py }
+          - { name: django-asgi-uvicorn, testfile: end2end/django_asgi_uvicorn_test.py }
           - { name: flask-mongo, testfile: end2end/flask_mongo_test.py }
           - { name: flask-mysql, testfile: end2end/flask_mysql_test.py }
           - { name: flask-mysql-uwsgi, testfile: end2end/flask_mysql_uwsgi_test.py }

end2end/django_asgi_uvicorn_test.py

@@ -0,0 +1,59 @@
+import pytest
+import requests
+import time
+from .server.check_events_from_mock import fetch_events_from_mock, validate_started_event, filter_on_event_type, \
+    clear_events_from_mock
+
+# e2e tests for django_postgres_gunicorn sample app
+post_url_fw = "http://localhost:8114/create"
+post_url_nofw = "http://localhost:8115/create"
+
+def test_firewall_started_okay():
+    events = fetch_events_from_mock("http://localhost:5000")
+    started_events = filter_on_event_type(events, "started")
+    assert len(started_events) == 1
+    validate_started_event(started_events[0], ["gunicorn", "django", "psycopg2-binary"])
+
+def test_safe_response_with_firewall():
+    dog_name = "Bobby Tables"
+    res = requests.post(post_url_fw, 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})
+    assert res.status_code == 200
+
+
+def test_dangerous_response_with_firewall():
+    clear_events_from_mock("http://localhost:5000")
+    dog_name = "Dangerous bobby', TRUE); -- "
+    res = requests.post(post_url_fw, 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")
+    attacks = filter_on_event_type(events, "detected_attack")
+    
+    assert len(attacks) == 1
+    del attacks[0]["attack"]["stack"]
+    assert attacks[0]["attack"] == {
+        "blocked": True,
+        "kind": "sql_injection",
+        'metadata': {
+            'dialect': "postgres",
+            'sql': "INSERT INTO sample_app_Dogs (dog_name, is_admin) VALUES ('Dangerous bobby', TRUE); -- ', FALSE)"
+        },
+        'operation': "psycopg2.Connection.Cursor.execute",
+        'pathToPayload': '.dog_name.[0]',
+        'payload': "\"Dangerous bobby', TRUE); -- \"",
+        'source': "body",
+        'user': None
+    }
+
+def test_dangerous_response_without_firewall():
+    dog_name = "Dangerous bobby', TRUE); -- "
+    res = requests.post(post_url_nofw, data={'dog_name': dog_name})
+    assert res.status_code == 200
+