Ankur Bhaiya Project
### Project Overview: Simulated Online Banking Platform
This project aims to create a simulated online banking system in Python, focusing on key features like account management, fund transfers, transaction history, and secure authentication with encryption. We'll use Flask for the web framework (as it's lightweight and suitable for a demo), SQLite for the database (easy for simulation), and libraries like `cryptography` for encryption, `pyotp` for OTP generation, and `threading` for simulating concurrency. This is a local, simulated environment—no real money or external services involved.
**Prerequisites:**
- Install required libraries: `flask`, `sqlite3` (built-in), `cryptography`, `pyotp`, `werkzeug` (for password hashing).
- Run the app locally with `python app.py`.
I'll address each task step-by-step, including code snippets. The full code can be assembled into a single `app.py` file for simplicity.
#### 1. Design Customer Account Database
Use SQLite to store user accounts, including fields for secure storage. Tables:
- `users`: id, username, hashed_password, email, balance (encrypted), otp_secret (for OTP).
- `transactions`: id, user_id, type (deposit/withdraw/transfer), amount, to_user (if transfer), timestamp.
Code to initialize the database:
```python
import sqlite3
from cryptography.fernet import Fernet
# Generate encryption key (store securely in production; here, hardcoded for demo)
key = Fernet.generate_key()
cipher = Fernet(key)
def init_db():
conn = sqlite3.connect('bank.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY, username TEXT UNIQUE, hashed_password TEXT, email TEXT,
encrypted_balance BLOB, otp_secret TEXT)''')
c.execute('''CREATE TABLE IF NOT EXISTS transactions
(id INTEGER PRIMARY KEY, user_id INTEGER, type TEXT, amount REAL, to_user TEXT, timestamp DATETIME)''')
conn.commit()
conn.close()
# Example: Encrypt balance
def encrypt_balance(balance):
return cipher.encrypt(str(balance).encode())
# Call init_db() once at app start
```
Insert sample users during setup (hash passwords with `werkzeug.security.generate_password_hash`).
#### 2. Implement Secure Login with OTP
Use two-factor authentication: username/password + OTP (via `pyotp` for TOTP). Sessions managed with Flask's session.
Code snippet for login route:
```python
from flask import Flask, request, session, redirect, url_for
import pyotp
from werkzeug.security import check_password_hash, generate_password_hash
import time
app = Flask(__name__)
app.secret_key = 'secret_key' # Change in production
@app.route('/login', methods=['POST'])
def login():
username = request.form['username']
password = request.form['password']
otp = request.form['otp']
conn = sqlite3.connect('bank.db')
c = conn.cursor()
c.execute("SELECT hashed_password, otp_secret FROM users WHERE username=?", (username,))
user = c.fetchone()
conn.close()
if user and check_password_hash(user[0], password):
totp = pyotp.TOTP(user[1])
if totp.verify(otp):
session['user'] = username
session['login_time'] = time.time()
return redirect(url_for('dashboard'))
return 'Invalid credentials'
# User registration (for demo): Generate OTP secret
def register_user(username, password, email, initial_balance=1000.0):
hashed_pw = generate_password_hash(password)
otp_secret = pyotp.random_base32()
encrypted_bal = encrypt_balance(initial_balance)
conn = sqlite3.connect('bank.db')
c = conn.cursor()
c.execute("INSERT INTO users (username, hashed_password, email, encrypted_balance, otp_secret) VALUES (?, ?, ?, ?, ?)",
(username, hashed_pw, email, encrypted_bal, otp_secret))
conn.commit()
conn.close()
return otp_secret # Show to user for OTP app setup
```
Users scan the OTP secret into an authenticator app (e.g., Google Authenticator).
#### 3. Develop Fund Transfer Module
Allow transfers between users, checking balance and updating both accounts atomically (use transactions in SQLite).
Code:
```python
def decrypt_balance(encrypted_bal):
return float(cipher.decrypt(encrypted_bal).decode())
@app.route('/transfer', methods=['POST'])
def transfer():
if 'user' not in session:
return 'Unauthorized'
from_user = session['user']
to_user = request.form['to_user']
amount = float(request.form['amount'])
conn = sqlite3.connect('bank.db')
conn.execute('BEGIN TRANSACTION')
try:
c = conn.cursor()
# Get balances
c.execute("SELECT encrypted_balance FROM users WHERE username=?", (from_user,))
from_bal_enc = c.fetchone()[0]
from_bal = decrypt_balance(from_bal_enc)
if from_bal < amount:
raise ValueError('Insufficient funds')
c.execute("SELECT encrypted_balance FROM users WHERE username=?", (to_user,))
to_bal_enc = c.fetchone()[0]
to_bal = decrypt_balance(to_bal_enc)
# Update balances
new_from_bal = encrypt_balance(from_bal - amount)
new_to_bal = encrypt_balance(to_bal + amount)
c.execute("UPDATE users SET encrypted_balance=? WHERE username=?", (new_from_bal, from_user))
c.execute("UPDATE users SET encrypted_balance=? WHERE username=?", (new_to_bal, to_user))
# Log transaction
c.execute("INSERT INTO transactions (user_id, type, amount, to_user, timestamp) VALUES ((SELECT id FROM users WHERE username=?), 'transfer', ?, ?, DATETIME('now'))",
(from_user, -amount, to_user))
c.execute("INSERT INTO transactions (user_id, type, amount, to_user, timestamp) VALUES ((SELECT id FROM users WHERE username=?), 'transfer', ?, ?, DATETIME('now'))",
(to_user, amount, from_user))
conn.commit()
return 'Transfer successful'
except Exception as e:
conn.rollback()
return str(e)
finally:
conn.close()
```
#### 4. Add Transaction History Tracking
Query the `transactions` table for a user's history.
Code:
```python
@app.route('/history')
def history():
if 'user' not in session:
return 'Unauthorized'
user_id = get_user_id(session['user']) # Helper function to fetch id
conn = sqlite3.connect('bank.db')
c = conn.cursor()
c.execute("SELECT type, amount, to_user, timestamp FROM transactions WHERE user_id=? ORDER BY timestamp DESC", (user_id,))
history = c.fetchall()
conn.close()
return str(history) # Render as HTML table in full app
```
#### 5. Integrate Encryption for Sensitive Data
We're already using Fernet (symmetric encryption from `cryptography`) for balances. Extend to other sensitive fields if needed (e.g., emails). Passwords are hashed, not encrypted.
#### 6. Implement Session Timeout Feature
Check session age on each request; logout after 5 minutes inactivity.
Code (use Flask's before_request):
```python
@app.before_request
def check_session_timeout():
if 'user' in session and 'login_time' in session:
if time.time() - session['login_time'] > 300: # 5 minutes
session.pop('user', None)
session.pop('login_time', None)
return 'Session timed out'
session['login_time'] = time.time() # Refresh on activity
```
#### 7. Develop Admin Monitoring Dashboard
Add an admin user role. Dashboard shows all users' balances (decrypted) and recent transactions.
First, add role to users table (e.g., role='admin' or 'user').
Code:
```python
@app.route('/admin_dashboard')
def admin_dashboard():
if 'user' not in session or get_user_role(session['user']) != 'admin':
return 'Unauthorized'
conn = sqlite3.connect('bank.db')
c = conn.cursor()
c.execute("SELECT username, encrypted_balance FROM users")
users = [(row[0], decrypt_balance(row[1])) for row in c.fetchall()]
c.execute("SELECT * FROM transactions ORDER BY timestamp DESC LIMIT 50")
recent_tx = c.fetchall()
conn.close()
return str(users + recent_tx) # Render as HTML in full app
```
#### 8. Simulate Fraud Detection Alerts
Simple rule-based: Alert if transfer > $1000 or multiple transfers in short time.
Code (integrate into transfer function):
```python
# In transfer function, before commit:
c.execute("SELECT COUNT(*) FROM transactions WHERE user_id=(SELECT id FROM users WHERE username=?) AND timestamp > DATETIME('now', '-1 hour')", (from_user,))
if c.fetchone()[0] > 5 or amount > 1000:
# Simulate alert: Log or email (print for demo)
print(f"Fraud alert for user {from_user}: High activity/amount")
# Optionally rollback for demo
```
#### 9. Conduct Security Testing
- **Manual Tests:** SQL injection (use prepared statements—already done), XSS (sanitize inputs), brute-force login (add rate limiting).
- **Tools:** Use `bandit` for static analysis, `pytest` for unit tests (e.g., test encryption/decryption).
- **Example Test Code (using pytest):**
```python
import pytest
def test_encryption():
bal = 1000.0
enc = encrypt_balance(bal)
assert decrypt_balance(enc) == bal
```
Run `pytest` locally.
#### 10. Deploy on Secure Local Server
Use Flask's built-in server for demo: `python app.py` (runs on http://127.0.0.1:5000). For "secure," use HTTPS with self-signed cert via `flask run --cert=adhoc`. Add firewall rules if needed. No cloud deployment for simulation.
#### 11. Test Concurrent Transactions
Use `threading` to simulate multiple transfers.
Code:
```python
import threading
def simulate_transfer(from_user, to_user, amount):
# Call transfer logic here (mock request)
pass # Implement as per transfer function
threads = []
for i in range(10):
t = threading.Thread(target=simulate_transfer, args=('user1', 'user2', 10.0))
threads.append(t)
t.start()
for t in threads:
t.join()
# Check database integrity after (no race conditions due to SQLite transactions)
```
#### 12. Document System Security Measures
- **Authentication:** Password hashing with werkzeug, TOTP with pyotp.
- **Encryption:** Fernet for data at rest (balances).
- **Authorization:** Session-based, with timeouts.
- **Input Validation:** Use prepared SQL statements to prevent injection.
- **Fraud Detection:** Rule-based alerts.
- **Best Practices:** Never store plaintext passwords/OTPs; use HTTPS; log activities.
- **Limitations:** This is a simulation—real systems need audits, compliance (e.g., PCI DSS), and more robust scaling.
To run the full app, combine these snippets into `app.py`, call `init_db()`, register sample users, and access routes like `/login`. For enhancements, add HTML templates with Flask. If you need specific code expansions or debugging, provide more details!
Comments
Post a Comment