Title / Description
Code """ Educational intentionally vulnerable Blood Donation System. Use only in a local lab environment. """ import os import subprocess from datetime import datetime import psycopg2 from psycopg2.extras import RealDictCursor from flask import Flask, render_template, request, redirect, url_for, session, flash, Response from markupsafe import Markup app = Flask(__name__) # INTENTIONAL LOW/MEDIUM ISSUE: hardcoded weak secret key for lab demonstration. app.secret_key = "blood-donation-lab-secret" app.config["SESSION_COOKIE_HTTPONLY"] = False app.config["SESSION_COOKIE_SECURE"] = False DB_HOST = os.getenv("DB_HOST", "localhost") DB_NAME = os.getenv("DB_NAME", "blood_donation") DB_USER = os.getenv("DB_USER", "postgres") DB_PASS = os.getenv("DB_PASS", "postgres") DB_PORT = os.getenv("DB_PORT", "5432") def get_conn(): return psycopg2.connect( host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS, port=DB_PORT, cursor_factory=RealDictCursor, ) def staff_required(): return "staff_id" in session @app.route("/") def index(): return render_template("index.html") @app.route("/donor/register", methods=["GET", "POST"]) def donor_register(): if request.method == "POST": full_name = request.form.get("full_name", "") email = request.form.get("email", "") phone = request.form.get("phone", "") ic_number = request.form.get("ic_number", "") address = request.form.get("address", "") gender = request.form.get("gender", "") date_of_birth = request.form.get("date_of_birth") or None # INTENTIONAL MEDIUM ISSUE: weak validation; stores personal info directly. conn = get_conn() cur = conn.cursor() cur.execute( """ INSERT INTO donors(full_name, email, phone, ic_number, address, gender, date_of_birth) VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id """, (full_name, email, phone, ic_number, address, gender, date_of_birth), ) donor_id = cur.fetchone()["id"] conn.commit() cur.close() conn.close() flash(f"Registration successful. Your Donor ID is {donor_id}.", "success") return redirect(url_for("index")) return render_template("donor_register.html") @app.route("/staff/login", methods=["GET", "POST"]) def staff_login(): if request.method == "POST": username = request.form.get("username", "") password = request.form.get("password", "") # INTENTIONAL CRITICAL VULNERABILITY: SQL Injection in login query. query = f"SELECT id, username, full_name, role FROM medical_staff WHERE username = '{username}' AND password = '{password}'" conn = get_conn() cur = conn.cursor() try: cur.execute(query) staff = cur.fetchone() except Exception as e: # INTENTIONAL LOW VULNERABILITY: verbose error leakage. return render_template("error.html", error=str(e), query=query), 500 finally: cur.close() conn.close() if staff: session["staff_id"] = staff["id"] session["username"] = staff["username"] session["role"] = staff["role"] flash("Login successful.", "success") return redirect(url_for("staff_dashboard")) flash("Invalid username or password.", "danger") return render_template("staff_login.html") @app.route("/staff/logout") def staff_logout(): session.clear() flash("Logged out successfully.", "info") return redirect(url_for("index")) @app.route("/staff/dashboard") def staff_dashboard(): if not staff_required(): return redirect(url_for("staff_login")) conn = get_conn() cur = conn.cursor() cur.execute("SELECT COUNT(*) AS total FROM donors") donors_count = cur.fetchone()["total"] cur.execute("SELECT COUNT(*) AS total FROM donations") donations_count = cur.fetchone()["total"] cur.execute("SELECT COUNT(*) AS total FROM blood_bank_queue") queue_count = cur.fetchone()["total"] cur.close() conn.close() return render_template( "staff_dashboard.html", donors_count=donors_count, donations_count=donations_count, queue_count=queue_count, ) @app.route("/staff/donors") def staff_donors(): if not staff_required(): return redirect(url_for("staff_login")) keyword = request.args.get("q", "") conn = get_conn() cur = conn.cursor() # INTENTIONAL HIGH/MEDIUM ISSUE: SQL Injection in donor search. if keyword: query = f"SELECT * FROM donors WHERE full_name ILIKE '%{keyword}%' OR ic_number ILIKE '%{keyword}%' ORDER BY id DESC" else: query = "SELECT * FROM donors ORDER BY id DESC" try: cur.execute(query) donors = cur.fetchall() except Exception as e: return render_template("error.html", error=str(e), query=query), 500 finally: cur.close() conn.close() return render_template("staff_donors.html", donors=donors, keyword=keyword) @app.route("/staff/donor/<int:donor_id>") def donor_detail(donor_id): if not staff_required(): return redirect(url_for("staff_login")) # INTENTIONAL HIGH VULNERABILITY: broken access control/IDOR-like direct record access. conn = get_conn() cur = conn.cursor() cur.execute("SELECT * FROM donors WHERE id = %s", (donor_id,)) donor = cur.fetchone() cur.execute("SELECT * FROM health_screenings WHERE donor_id = %s ORDER BY screening_date DESC", (donor_id,)) screenings = cur.fetchall() cur.execute("SELECT * FROM donations WHERE donor_id = %s ORDER BY donation_date DESC", (donor_id,)) donations = cur.fetchall() cur.execute("SELECT * FROM blood_bank_queue WHERE donor_id = %s ORDER BY collected_date DESC", (donor_id,)) blood_records = cur.fetchall() cur.close() conn.close() if not donor: flash("Donor not found.", "warning") return redirect(url_for("staff_donors")) return render_template( "donor_detail.html", donor=donor, screenings=screenings, donations=donations, blood_records=blood_records, ) @app.route("/staff/donor/<int:donor_id>/screening", methods=["POST"]) def update_screening(donor_id): if not staff_required(): return redirect(url_for("staff_login")) blood_type = request.form.get("blood_type", "") weight = request.form.get("weight") or None height = request.form.get("height") or None disease = request.form.get("disease", "") result = request.form.get("result", "Pending") remarks = request.form.get("remarks", "") # INTENTIONAL MEDIUM ISSUE: no CSRF protection on state-changing action. # INTENTIONAL MEDIUM ISSUE: disease/remarks can store script content and later render unsafely. conn = get_conn() cur = conn.cursor() cur.execute( """ INSERT INTO health_screenings(donor_id, staff_id, screening_date, blood_type, weight, height, disease, result, remarks) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, (donor_id, session.get("staff_id"), datetime.now(), blood_type, weight, height, disease, result, remarks), ) conn.commit() cur.close() conn.close() flash("Health screening updated.", "success") return redirect(url_for("donor_detail", donor_id=donor_id)) @app.route("/staff/donor/<int:donor_id>/donation", methods=["POST"]) def update_donation(donor_id): if not staff_required(): return redirect(url_for("staff_login")) donation_status = request.form.get("donation_status", "Pending") notes = request.form.get("notes", "") conn = get_conn() cur = conn.cursor() cur.execute( """ INSERT INTO donations(donor_id, staff_id, donation_date, status, notes) VALUES (%s, %s, %s, %s, %s) """, (donor_id, session.get("staff_id"), datetime.now(), donation_status, notes), ) conn.commit() cur.close() conn.close() flash("Donation status updated.", "success") return redirect(url_for("donor_detail", donor_id=donor_id)) @app.route("/staff/donor/<int:donor_id>/blood-record", methods=["POST"]) def create_blood_record(donor_id): if not staff_required(): return redirect(url_for("staff_login")) blood_record_id = request.form.get("blood_record_id", "") blood_type = request.form.get("blood_type", "") collected_date = request.form.get("collected_date") or datetime.now().date() queue_status = request.form.get("queue_status", "Queued") conn = get_conn() cur = conn.cursor() cur.execute( """ INSERT INTO blood_bank_queue(blood_record_id, donor_id, staff_id, collected_date, blood_type, queue_status) VALUES (%s, %s, %s, %s, %s, %s) """, (blood_record_id, donor_id, session.get("staff_id"), collected_date, blood_type, queue_status), ) conn.commit() cur.close() conn.close() flash("Blood record added to blood bank queue.", "success") return redirect(url_for("donor_detail", donor_id=donor_id)) @app.route("/staff/reports/operational") def operational_report(): if not staff_required(): return redirect(url_for("staff_login")) # Normal system feature: consolidated operational reporting for medical staff. # INTENTIONAL HIGH VULNERABILITY: excessive data exposure and missing role-based restriction. # Any authenticated staff account can view/export donor PII, screening data, blood bank queue data, # and plaintext staff credentials. This demonstrates what an attacker may do after initial access. conn = get_conn() cur = conn.cursor() cur.execute("SELECT id, username, password, full_name, role, created_at FROM medical_staff ORDER BY id") staff_accounts = cur.fetchall() cur.execute( """ SELECT d.id AS donor_id, d.full_name, d.ic_number, d.email, d.phone, d.address, hs.blood_type, hs.weight, hs.height, hs.disease, hs.result AS screening_result, hs.remarks, hs.screening_date FROM donors d LEFT JOIN LATERAL ( SELECT * FROM health_screenings h WHERE h.donor_id = d.id ORDER BY h.screening_date DESC LIMIT 1 ) hs ON true ORDER BY d.id DESC """ ) donor_records = cur.fetchall() cur.execute( """ SELECT bbq.blood_record_id, d.full_name, d.ic_number, bbq.blood_type, bbq.collected_date, bbq.queue_status, ms.full_name AS handled_by FROM blood_bank_queue bbq LEFT JOIN donors d ON d.id = bbq.donor_id LEFT JOIN medical_staff ms ON ms.id = bbq.staff_id ORDER BY bbq.created_at DESC """ ) blood_queue = cur.fetchall() cur.close() conn.close() if request.args.get("format") == "csv": # INTENTIONAL MEDIUM/HIGH ISSUE: bulk export of sensitive data without stronger authorization. lines = ["section,id_or_record,name_or_username,identifier,extra,status"] for staff in staff_accounts: lines.append(f"staff,{staff['id']},{staff['username']},{staff['password']},{staff['role']},active") for donor in donor_records: lines.append(f"donor,{donor['donor_id']},{donor['full_name']},{donor['ic_number']},{donor.get('blood_type')},{donor.get('screening_result')}") csv_data = "\n".join(lines) return Response( csv_data, mimetype="text/csv", headers={"Content-Disposition": "attachment; filename=operational_report.csv"}, ) return render_template( "operational_report.html", staff_accounts=staff_accounts, donor_records=donor_records, blood_queue=blood_queue, ) @app.route("/donor/donation-pass") def donation_pass(): # Normal system feature: donor can show a donation pass/reference during future visits. # INTENTIONAL CHAIN-RELATED VULNERABILITY: predictable reference + direct object access. # If an attacker discovers donor IDs through another weakness, they can access these records directly. donor_id = request.args.get("donor_id", "") ref = request.args.get("ref", "") expected_ref = f"BDP-{donor_id}" if not donor_id or ref != expected_ref: flash("Invalid donation pass reference.", "danger") return redirect(url_for("index")) conn = get_conn() cur = conn.cursor() try: cur.execute("SELECT * FROM donors WHERE id = %s", (donor_id,)) donor = cur.fetchone() cur.execute( "SELECT * FROM health_screenings WHERE donor_id = %s ORDER BY screening_date DESC LIMIT 1", (donor_id,), ) latest_screening = cur.fetchone() cur.execute( "SELECT * FROM donations WHERE donor_id = %s ORDER BY donation_date DESC LIMIT 5", (donor_id,), ) donations = cur.fetchall() finally: cur.close() conn.close() if not donor: flash("Donation pass not found.", "warning") return redirect(url_for("index")) return render_template( "donation_pass.html", donor=donor, latest_screening=latest_screening, donations=donations, ref=ref, ) @app.route("/staff/library-updates") def library_updates(): if not staff_required(): return redirect(url_for("staff_login")) # Requirement feature: staff can check whether Python libraries are outdated. # It is intentionally simple for teaching dependency maintenance. try: result = subprocess.run( ["python", "-m", "pip", "list", "--outdated"], capture_output=True, text=True, timeout=15, ) output = result.stdout or result.stderr or "No output returned." except Exception as e: output = f"Update check failed: {e}" return render_template("library_updates.html", output=output) @app.template_filter("unsafe") def unsafe_filter(value): # INTENTIONAL MEDIUM VULNERABILITY: disables output escaping for stored XSS demonstration. return Markup(value or "") if __name__ == "__main__": # INTENTIONAL MEDIUM VULNERABILITY: debug mode enabled. app.run(host="0.0.0.0", port=5000, debug=True)
Author
Highlight as C C++ CSS Clojure Delphi ERb Groovy (beta) HAML HTML JSON Java JavaScript PHP Plain text Python Ruby SQL XML YAML diff code
1.1.2
1.1.2