-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path02-sql-injection-prevention.py
More file actions
810 lines (652 loc) · 31 KB
/
02-sql-injection-prevention.py
File metadata and controls
810 lines (652 loc) · 31 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
"""Question: Implement SQL injection prevention techniques to secure database operations.
Create a secure database interface that demonstrates various SQL injection prevention methods
including parameterized queries, input validation, and ORM usage.
Requirements:
1. Create a vulnerable database class to show SQL injection risks
2. Create a secure database class with parameterized queries
3. Implement input validation and sanitization
4. Demonstrate ORM-based approaches
5. Show different attack vectors and their prevention
Example usage:
secure_db = SecureDatabase()
user = secure_db.get_user_by_credentials("john", "password123")
secure_db.create_user("jane", "jane@example.com", "securepass")
"""
# LEARNING CHALLENGE
#
# Before looking at any solution below, please try to solve this yourself first!
#
# Tips for success:
# - Read the question carefully
# - Think about what makes SQL queries vulnerable
# - Consider different types of SQL injection attacks
# - Start with a simple implementation
# - Test your code step by step
# - Don't worry if it's not perfect - learning is a process!
#
# Remember: The best way to learn programming is by doing, not by reading solutions!
#
# Take your time, experiment, and enjoy the learning process!
# Try to implement your solution here:
# (Write your code below this line)
# HINT SECTION (Only look if you're really stuck!)
#
# Think about:
# - What makes SQL queries vulnerable to injection?
# - How can parameterized queries prevent injection?
# - What input validation techniques can you use?
# - How do ORMs help prevent SQL injection?
#
# Remember: Start simple and build up complexity gradually!
# ===============================================================================
# STEP-BY-STEP SOLUTION
# ===============================================================================
#
# CLASSROOM-STYLE WALKTHROUGH
#
# Let's solve this problem step by step, just like in a programming class!
# Each step builds upon the previous one, so you can follow along and understand
# the complete thought process.
#
# ===============================================================================
# Step 1: Import modules and create a vulnerable database class
# ===============================================================================
# Explanation:
# First, we'll create a vulnerable database class to demonstrate SQL injection risks.
# This shows what NOT to do and helps understand the security issues.
import sqlite3
import re
import hashlib
import secrets
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
@dataclass
class User:
"""User data class."""
id: Optional[int] = None
username: str = ""
email: str = ""
password_hash: str = ""
class VulnerableDatabase:
"""VULNERABLE database class - DO NOT USE IN PRODUCTION!
This class demonstrates common SQL injection vulnerabilities.
It's included for educational purposes only.
"""
def __init__(self, db_path: str = ":memory:"):
self.connection = sqlite3.connect(db_path)
self.connection.row_factory = sqlite3.Row
self._create_tables()
def _create_tables(self):
"""Create user table."""
cursor = self.connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL
)
""")
self.connection.commit()
def get_user_by_credentials_vulnerable(self, username: str, password: str) -> Optional[User]:
"""VULNERABLE: Uses string concatenation - susceptible to SQL injection."""
cursor = self.connection.cursor()
# VULNERABLE CODE - DO NOT USE!
query = f"SELECT * FROM users WHERE username = '{username}' AND password_hash = '{password}'"
try:
cursor.execute(query)
row = cursor.fetchone()
if row:
return User(id=row['id'], username=row['username'],
email=row['email'], password_hash=row['password_hash'])
return None
except sqlite3.Error as e:
print(f"Database error: {e}")
return None
# Step 2: Add more vulnerable methods to show different attack vectors
# ===============================================================================
# Explanation:
# Let's add more vulnerable methods to demonstrate different types of SQL injection
# attacks including search functionality and user creation.
def search_users_vulnerable(self, search_term: str) -> List[User]:
"""VULNERABLE: Search users by username - susceptible to SQL injection."""
cursor = self.connection.cursor()
# VULNERABLE CODE - DO NOT USE!
query = f"SELECT * FROM users WHERE username LIKE '%{search_term}%'"
try:
cursor.execute(query)
rows = cursor.fetchall()
return [User(id=row['id'], username=row['username'],
email=row['email'], password_hash=row['password_hash'])
for row in rows]
except sqlite3.Error as e:
print(f"Database error: {e}")
return []
def create_user_vulnerable(self, username: str, email: str, password: str) -> bool:
"""VULNERABLE: Create user - susceptible to SQL injection."""
cursor = self.connection.cursor()
password_hash = hashlib.sha256(password.encode()).hexdigest()
# VULNERABLE CODE - DO NOT USE!
query = f"INSERT INTO users (username, email, password_hash) VALUES ('{username}', '{email}', '{password_hash}')"
try:
cursor.execute(query)
self.connection.commit()
return True
except sqlite3.Error as e:
print(f"Database error: {e}")
return False
# Step 3: Create input validation utilities
# ===============================================================================
# Explanation:
# Before creating secure database methods, we need input validation utilities
# to sanitize and validate user input.
class InputValidator:
"""Utility class for input validation and sanitization."""
@staticmethod
def validate_username(username: str) -> bool:
"""Validate username format."""
if not username or len(username) < 3 or len(username) > 50:
return False
# Allow only alphanumeric characters and underscores
return re.match(r'^[a-zA-Z0-9_]+$', username) is not None
@staticmethod
def validate_email(email: str) -> bool:
"""Validate email format."""
if not email or len(email) > 254:
return False
# Basic email validation
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
@staticmethod
def validate_password(password: str) -> bool:
"""Validate password strength."""
if not password or len(password) < 8 or len(password) > 128:
return False
# Require at least one uppercase, lowercase, digit, and special character
has_upper = any(c.isupper() for c in password)
has_lower = any(c.islower() for c in password)
has_digit = any(c.isdigit() for c in password)
has_special = any(c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in password)
return has_upper and has_lower and has_digit and has_special
@staticmethod
def sanitize_search_term(search_term: str) -> str:
"""Sanitize search term by removing potentially dangerous characters."""
if not search_term:
return ""
# Remove SQL special characters
sanitized = re.sub(r'[\'";\\%_]', '', search_term)
return sanitized[:100] # Limit length
# Step 4: Create secure database class with parameterized queries
# ===============================================================================
# Explanation:
# Now we'll create a secure database class that uses parameterized queries,
# input validation, and proper error handling to prevent SQL injection.
class SecureDatabase:
"""Secure database class with SQL injection prevention."""
def __init__(self, db_path: str = ":memory:"):
self.connection = sqlite3.connect(db_path)
self.connection.row_factory = sqlite3.Row
self._create_tables()
self.validator = InputValidator()
def _create_tables(self):
"""Create user table."""
cursor = self.connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
salt TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.connection.commit()
def _hash_password(self, password: str, salt: str = None) -> tuple[str, str]:
"""Hash password with salt."""
if salt is None:
salt = secrets.token_hex(32)
password_hash = hashlib.pbkdf2_hmac('sha256', password.encode(),
salt.encode(), 100000)
return password_hash.hex(), salt
def create_user(self, username: str, email: str, password: str) -> bool:
"""SECURE: Create user with input validation and parameterized queries."""
# Input validation
if not self.validator.validate_username(username):
raise ValueError("Invalid username format")
if not self.validator.validate_email(email):
raise ValueError("Invalid email format")
if not self.validator.validate_password(password):
raise ValueError("Password does not meet security requirements")
cursor = self.connection.cursor()
password_hash, salt = self._hash_password(password)
try:
# SECURE: Using parameterized query
cursor.execute("""
INSERT INTO users (username, email, password_hash, salt)
VALUES (?, ?, ?, ?)
""", (username, email, password_hash, salt))
self.connection.commit()
return True
except sqlite3.IntegrityError:
raise ValueError("Username or email already exists")
except sqlite3.Error as e:
raise RuntimeError(f"Database error: {e}")
def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
"""SECURE: Authenticate user with parameterized queries."""
# Input validation
if not username or not password:
return None
if not self.validator.validate_username(username):
return None
cursor = self.connection.cursor()
try:
# SECURE: Using parameterized query
cursor.execute("""
SELECT id, username, email, password_hash, salt
FROM users WHERE username = ?
""", (username,))
row = cursor.fetchone()
if not row:
return None
# Verify password
stored_hash = row['password_hash']
salt = row['salt']
password_hash, _ = self._hash_password(password, salt)
if password_hash == stored_hash:
return User(id=row['id'], username=row['username'],
email=row['email'], password_hash=stored_hash)
return None
except sqlite3.Error as e:
raise RuntimeError(f"Database error: {e}")
def search_users(self, search_term: str) -> List[User]:
"""SECURE: Search users with input sanitization and parameterized queries."""
# Input sanitization
sanitized_term = self.validator.sanitize_search_term(search_term)
if not sanitized_term:
return []
cursor = self.connection.cursor()
try:
# SECURE: Using parameterized query with LIKE
search_pattern = f"%{sanitized_term}%"
cursor.execute("""
SELECT id, username, email, password_hash
FROM users WHERE username LIKE ?
""", (search_pattern,))
rows = cursor.fetchall()
return [User(id=row['id'], username=row['username'],
email=row['email'], password_hash=row['password_hash'])
for row in rows]
except sqlite3.Error as e:
raise RuntimeError(f"Database error: {e}")
def get_user_by_id(self, user_id: int) -> Optional[User]:
"""SECURE: Get user by ID with parameterized query."""
if not isinstance(user_id, int) or user_id <= 0:
return None
cursor = self.connection.cursor()
try:
# SECURE: Using parameterized query
cursor.execute("""
SELECT id, username, email, password_hash
FROM users WHERE id = ?
""", (user_id,))
row = cursor.fetchone()
if row:
return User(id=row['id'], username=row['username'],
email=row['email'], password_hash=row['password_hash'])
return None
except sqlite3.Error as e:
raise RuntimeError(f"Database error: {e}")
# Step 5: Create ORM-based secure database class
# ===============================================================================
# Explanation:
# ORMs (Object-Relational Mappers) provide another layer of protection against
# SQL injection by abstracting SQL query construction.
try:
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
Base = declarative_base()
class UserORM(Base):
"""SQLAlchemy ORM model for users."""
__tablename__ = 'users_orm'
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(254), unique=True, nullable=False)
password_hash = Column(String(128), nullable=False)
salt = Column(String(64), nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
class ORMDatabase:
"""ORM-based secure database class."""
def __init__(self, db_url: str = "sqlite:///:memory:"):
self.engine = create_engine(db_url)
Base.metadata.create_all(self.engine)
Session = sessionmaker(bind=self.engine)
self.session = Session()
self.validator = InputValidator()
def _hash_password(self, password: str, salt: str = None) -> tuple[str, str]:
"""Hash password with salt."""
if salt is None:
salt = secrets.token_hex(32)
password_hash = hashlib.pbkdf2_hmac('sha256', password.encode(),
salt.encode(), 100000)
return password_hash.hex(), salt
def create_user(self, username: str, email: str, password: str) -> bool:
"""SECURE: Create user using ORM."""
# Input validation
if not self.validator.validate_username(username):
raise ValueError("Invalid username format")
if not self.validator.validate_email(email):
raise ValueError("Invalid email format")
if not self.validator.validate_password(password):
raise ValueError("Password does not meet security requirements")
password_hash, salt = self._hash_password(password)
try:
# ORM automatically handles parameterization
user = UserORM(username=username, email=email,
password_hash=password_hash, salt=salt)
self.session.add(user)
self.session.commit()
return True
except Exception as e:
self.session.rollback()
if "UNIQUE constraint failed" in str(e):
raise ValueError("Username or email already exists")
raise RuntimeError(f"Database error: {e}")
def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
"""SECURE: Authenticate user using ORM."""
if not username or not password:
return None
if not self.validator.validate_username(username):
return None
try:
# ORM automatically handles parameterization
user_orm = self.session.query(UserORM).filter(
UserORM.username == username
).first()
if not user_orm:
return None
# Verify password
password_hash, _ = self._hash_password(password, user_orm.salt)
if password_hash == user_orm.password_hash:
return User(id=user_orm.id, username=user_orm.username,
email=user_orm.email, password_hash=user_orm.password_hash)
return None
except Exception as e:
raise RuntimeError(f"Database error: {e}")
except ImportError:
# SQLAlchemy not available
class ORMDatabase:
def __init__(self, *args, **kwargs):
raise ImportError("SQLAlchemy is required for ORM functionality")
# Step 6: Demonstration and testing
# ===============================================================================
# Explanation:
# Let's demonstrate the differences between vulnerable and secure implementations
# and show how SQL injection attacks work and how to prevent them.
def demonstrate_sql_injection_attack():
"""Demonstrate SQL injection attack on vulnerable database."""
print("=== SQL Injection Attack Demonstration ===")
# Create vulnerable database and add test user
vuln_db = VulnerableDatabase()
vuln_db.create_user_vulnerable("admin", "admin@example.com", "password123")
vuln_db.create_user_vulnerable("user1", "user1@example.com", "userpass")
print("\n1. Normal login attempt:")
user = vuln_db.get_user_by_credentials_vulnerable("admin", "password123")
print(f"Login result: {user.username if user else 'Failed'}")
print("\n2. SQL Injection attack (bypassing authentication):")
# This malicious input bypasses authentication
malicious_username = "admin' OR '1'='1' --"
malicious_password = "anything"
user = vuln_db.get_user_by_credentials_vulnerable(malicious_username, malicious_password)
print(f"Attack result: {user.username if user else 'Failed'}")
print("⚠️ ATTACK SUCCESSFUL! Authentication bypassed!")
print("\n3. SQL Injection attack (data extraction):")
# This attack attempts to extract all usernames
malicious_search = "' UNION SELECT username, email, password_hash, id FROM users --"
try:
users = vuln_db.search_users_vulnerable(malicious_search)
print(f"Extracted {len(users)} users through injection")
for user in users:
print(f" - {user.username}: {user.email}")
except Exception as e:
print(f"Attack failed: {e}")
def demonstrate_secure_implementation():
"""Demonstrate secure database implementation."""
print("\n=== Secure Implementation Demonstration ===")
# Create secure database
secure_db = SecureDatabase()
print("\n1. Creating users with validation:")
try:
secure_db.create_user("john_doe", "john@example.com", "SecurePass123!")
print("✅ User created successfully")
except ValueError as e:
print(f"❌ Validation error: {e}")
try:
secure_db.create_user("invalid user", "invalid-email", "weak")
print("✅ User created successfully")
except ValueError as e:
print(f"❌ Validation error: {e}")
print("\n2. Attempting SQL injection on secure database:")
malicious_username = "admin' OR '1'='1' --"
malicious_password = "anything"
user = secure_db.get_user_by_credentials(malicious_username, malicious_password)
print(f"Attack result: {user.username if user else 'Failed'}")
print("✅ Attack prevented by input validation and parameterized queries!")
print("\n3. Secure search functionality:")
malicious_search = "' UNION SELECT username, email, password_hash, id FROM users --"
try:
users = secure_db.search_users(malicious_search)
print(f"Search returned {len(users)} users")
print("✅ Malicious SQL removed by input sanitization!")
except Exception as e:
print(f"Error: {e}")
# Step 7: Best practices and additional security measures
# ===============================================================================
# Explanation:
# Let's document best practices and additional security measures for preventing
# SQL injection and other database security issues.
class SecurityBestPractices:
"""Collection of security best practices for database operations."""
@staticmethod
def get_sql_injection_prevention_tips() -> List[str]:
"""Get list of SQL injection prevention tips."""
return [
"1. Always use parameterized queries or prepared statements",
"2. Validate and sanitize all user input",
"3. Use whitelist validation for expected input formats",
"4. Implement proper error handling without exposing database details",
"5. Use least privilege principle for database accounts",
"6. Regularly update database software and dependencies",
"7. Use stored procedures when appropriate",
"8. Implement input length limits",
"9. Use ORMs that handle parameterization automatically",
"10. Never concatenate user input directly into SQL queries",
"11. Use database-specific escaping functions as a last resort",
"12. Implement logging and monitoring for suspicious activities"
]
@staticmethod
def get_password_security_tips() -> List[str]:
"""Get list of password security tips."""
return [
"1. Use strong hashing algorithms (bcrypt, scrypt, or Argon2)",
"2. Always use salt for password hashing",
"3. Use sufficient iteration counts for key derivation",
"4. Implement password complexity requirements",
"5. Use secure random number generators for salts",
"6. Never store passwords in plain text",
"7. Implement account lockout mechanisms",
"8. Use multi-factor authentication when possible",
"9. Regularly audit password policies",
"10. Implement secure password reset mechanisms"
]
@staticmethod
def get_general_database_security_tips() -> List[str]:
"""Get list of general database security tips."""
return [
"1. Use connection pooling and connection limits",
"2. Implement database firewalls and network security",
"3. Encrypt sensitive data at rest and in transit",
"4. Regular security audits and penetration testing",
"5. Implement proper backup and recovery procedures",
"6. Use database activity monitoring",
"7. Implement role-based access control",
"8. Regular security updates and patches",
"9. Use secure configuration settings",
"10. Implement data masking for non-production environments"
]
# Step 8: Complete example with error handling and logging
# ===============================================================================
# Explanation:
# A complete, production-ready example that includes proper error handling,
# logging, and security measures.
import logging
from contextlib import contextmanager
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ProductionSecureDatabase:
"""Production-ready secure database class with comprehensive security measures."""
def __init__(self, db_path: str = ":memory:"):
self.connection = sqlite3.connect(db_path, check_same_thread=False)
self.connection.row_factory = sqlite3.Row
self._create_tables()
self.validator = InputValidator()
self._setup_security()
def _setup_security(self):
"""Setup additional security measures."""
# Enable foreign key constraints
self.connection.execute("PRAGMA foreign_keys = ON")
# Set secure temp store
self.connection.execute("PRAGMA secure_delete = ON")
self.connection.commit()
def _create_tables(self):
"""Create user table with additional security fields."""
cursor = self.connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
salt TEXT NOT NULL,
failed_login_attempts INTEGER DEFAULT 0,
account_locked_until TIMESTAMP NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_login TIMESTAMP NULL
)
""")
# Create audit log table
cursor.execute("""
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
action TEXT NOT NULL,
details TEXT,
ip_address TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users (id)
)
""")
self.connection.commit()
@contextmanager
def _get_cursor(self):
"""Context manager for database operations."""
cursor = self.connection.cursor()
try:
yield cursor
self.connection.commit()
except Exception:
self.connection.rollback()
raise
finally:
cursor.close()
def _log_security_event(self, action: str, details: str, user_id: int = None, ip_address: str = None):
"""Log security-related events."""
logger.info(f"Security event: {action} - {details}")
with self._get_cursor() as cursor:
cursor.execute("""
INSERT INTO audit_log (user_id, action, details, ip_address)
VALUES (?, ?, ?, ?)
""", (user_id, action, details, ip_address))
def create_user_secure(self, username: str, email: str, password: str, ip_address: str = None) -> bool:
"""Create user with comprehensive security measures."""
try:
# Input validation
if not self.validator.validate_username(username):
self._log_security_event("USER_CREATION_FAILED", f"Invalid username: {username}", ip_address=ip_address)
raise ValueError("Invalid username format")
if not self.validator.validate_email(email):
self._log_security_event("USER_CREATION_FAILED", f"Invalid email: {email}", ip_address=ip_address)
raise ValueError("Invalid email format")
if not self.validator.validate_password(password):
self._log_security_event("USER_CREATION_FAILED", f"Weak password for user: {username}", ip_address=ip_address)
raise ValueError("Password does not meet security requirements")
password_hash, salt = self._hash_password(password)
with self._get_cursor() as cursor:
cursor.execute("""
INSERT INTO users (username, email, password_hash, salt)
VALUES (?, ?, ?, ?)
""", (username, email, password_hash, salt))
self._log_security_event("USER_CREATED", f"User created: {username}", ip_address=ip_address)
return True
except sqlite3.IntegrityError:
self._log_security_event("USER_CREATION_FAILED", f"Duplicate user: {username}", ip_address=ip_address)
raise ValueError("Username or email already exists")
except sqlite3.Error as e:
self._log_security_event("DATABASE_ERROR", f"Error creating user: {str(e)}", ip_address=ip_address)
raise RuntimeError(f"Database error: {e}")
def _hash_password(self, password: str, salt: str = None) -> tuple[str, str]:
"""Hash password with salt using secure method."""
if salt is None:
salt = secrets.token_hex(32)
password_hash = hashlib.pbkdf2_hmac('sha256', password.encode(),
salt.encode(), 100000)
return password_hash.hex(), salt
def main():
"""Main function to demonstrate all concepts."""
print("SQL Injection Prevention Demonstration")
print("=" * 50)
# Demonstrate vulnerable implementation
demonstrate_sql_injection_attack()
# Demonstrate secure implementation
demonstrate_secure_implementation()
# Show best practices
print("\n=== Security Best Practices ===")
practices = SecurityBestPractices()
print("\nSQL Injection Prevention:")
for tip in practices.get_sql_injection_prevention_tips()[:5]:
print(f" {tip}")
print("\nPassword Security:")
for tip in practices.get_password_security_tips()[:5]:
print(f" {tip}")
print("\n✅ Remember: Security is a continuous process, not a one-time implementation!")
if __name__ == "__main__":
main()
# ===============================================================================
# SUMMARY
# ===============================================================================
#
# Key Concepts Covered:
# 1. SQL Injection vulnerabilities and attack vectors
# 2. Parameterized queries for prevention
# 3. Input validation and sanitization
# 4. Secure password hashing with salt
# 5. ORM-based approaches for additional security
# 6. Error handling and logging for security events
# 7. Best practices for database security
#
# Security Principles Applied:
# - Defense in depth (multiple layers of security)
# - Least privilege principle
# - Input validation and sanitization
# - Secure coding practices
# - Proper error handling
# - Audit logging
#
# Remember: Security is not just about preventing SQL injection - it's about
# implementing a comprehensive security strategy that includes proper
# authentication, authorization, encryption, and monitoring.
#
# ===============================================================================