Add code with more vulns (#3)

* Add call for sql sensitive data exposure * Add HTTP header misconfiguration * Tweak name of identifiers holding session info * Add code for XSS vuln * Add code for Open Redirect * Reformat code with black * Fix pylint errors

Author: ursachec

flask_webgoat/__init__.py

@@ -4,11 +4,12 @@
 
 from flask import Flask, g
 
-DB_FILENAME = 'database.db'
+DB_FILENAME = "database.db"
 
 
-def query_db(query, args = (), one=False, commit=False):
+def query_db(query, args=(), one=False, commit=False):
  with sqlite3.connect(DB_FILENAME) as conn:
+ conn.set_trace_callback(print)
  cur = conn.cursor().execute(query, args)
  if commit:
  conn.commit()
@@ -17,7 +18,7 @@ def query_db(query, args = (), one=False, commit=False):
 
 def create_app():
  app = Flask(__name__)
- app.secret_key = 'aeZ1iwoh2ree2mo0Eereireong4baitixaixu5Ee'
+ app.secret_key = "aeZ1iwoh2ree2mo0Eereireong4baitixaixu5Ee"
 
  db_path = Path(DB_FILENAME)
  if db_path.exists():
@@ -34,15 +35,16 @@ def create_app():
  conn.commit()
  conn.close()
 
-
  with app.app_context():
- from . import status
+ from . import actions
  from . import auth
+ from . import status
+ from . import ui
  from . import users
- from . import actions
- app.register_blueprint(status.bp)
+
+ app.register_blueprint(actions.bp)
  app.register_blueprint(auth.bp)
+ app.register_blueprint(status.bp)
+ app.register_blueprint(ui.bp)
  app.register_blueprint(users.bp)
- app.register_blueprint(actions.bp)
  return app
-

flask_webgoat/actions.py

@@ -1,52 +1,49 @@
 from pathlib import Path
 import subprocess
-import uuid
-from flask import request
 
-from flask import (
- Blueprint, jsonify, request, jsonify, session
-)
-from werkzeug.security import check_password_hash
-from flask import current_app as app
+from flask import Blueprint, request, jsonify, session
 
-bp = Blueprint('actions', __name__)
+bp = Blueprint("actions", __name__)
 
 
-@bp.route('/message', methods=['POST'])
+@bp.route("/message", methods=["POST"])
 def log_entry():
- auth = session.get('user_info', None)
- if auth is None:
- return jsonify({'error': 'no auth found in session'})
- access_level = auth[2]
+ user_info = session.get("user_info", None)
+ if user_info is None:
+ return jsonify({"error": "no user_info found in session"})
+ access_level = user_info[2]
  if access_level > 2:
- return jsonify({'error': 'access level < 2 is required for this action'})
- filename_param = request.form.get('filename')
+ return jsonify({"error": "access level < 2 is required for this action"})
+ filename_param = request.form.get("filename")
  if filename_param is None:
- return jsonify({'error': 'filename parameter is required'})
- text_param = request.form.get('text')
+ return jsonify({"error": "filename parameter is required"})
+ text_param = request.form.get("text")
  if text_param is None:
- return jsonify({'error': 'text parameter is required'})
+ return jsonify({"error": "text parameter is required"})
 
- user_id = auth[0]
+ user_id = user_info[0]
  user_dir = "data/" + str(user_id)
  user_dir_path = Path(user_dir)
  if not user_dir_path.exists():
  user_dir_path.mkdir()
 
  filename = filename_param + ".txt"
  path = Path(user_dir + "/" + filename)
- with path.open("w", encoding ="utf-8") as f:
- f.write(text_param)
- return jsonify({'success': True})
+ with path.open("w", encoding="utf-8") as open_file:
+ open_file.write(text_param)
+ return jsonify({"success": True})
 
 
-@bp.route('/grep_processes')
+@bp.route("/grep_processes")
 def grep_processes():
- name = request.args.get('name')
- res = subprocess.run(["ps aux | grep " + name + " | awk '{print $11}'"], shell=True, capture_output=True)
+ name = request.args.get("name")
+ res = subprocess.run(
+ ["ps aux | grep " + name + " | awk '{print $11}'"],
+ shell=True,
+ capture_output=True,
+ )
  if res.stdout is None:
- return jsonify({'error': 'no stdout returned'})
+ return jsonify({"error": "no stdout returned"})
  out = res.stdout.decode("utf-8")
- names = out.split('\n')
- return jsonify({'success': True, 'names': names})
-
+ names = out.split("\n")
+ return jsonify({"success": True, "names": names})

flask_webgoat/auth.py

@@ -1,24 +1,46 @@
-from flask import (
- Blueprint, jsonify, request, jsonify, session
-)
-from werkzeug.security import check_password_hash
-from flask import current_app as app
+from flask import Blueprint, request, jsonify, session, redirect
 from . import query_db
 
-bp = Blueprint('auth', __name__)
+bp = Blueprint("auth", __name__)
 
 
-@bp.route('/login', methods=['POST'])
+@bp.route("/login", methods=["POST"])
 def login():
- username = request.form.get('username')
- password = request.form.get('password')
+ username = request.form.get("username")
+ password = request.form.get("password")
  if username is None or password is None:
- return jsonify({'error': 'username and password parameter have to be provided'}), 400
+ return (
+ jsonify({"error": "username and password parameter have to be provided"}),
+ 400,
+ )
 
- query = "SELECT id, username, access_level FROM user WHERE username = '%s' AND password = '%s'" % (username, password)
+ query = (
+ "SELECT id, username, access_level FROM user WHERE username = '%s' AND password = '%s'"
+ % (username, password)
+ )
  result = query_db(query, [], True)
  if result is None:
- return jsonify({'bad_login': True}), 400
- session['user_info'] = (result[0], result[1], result[2])
- return jsonify({'success': True})
+ return jsonify({"bad_login": True}), 400
+ session["user_info"] = (result[0], result[1], result[2])
+ return jsonify({"success": True})
 
+
+@bp.route("/login_and_redirect")
+def login_and_redirect():
+ username = request.args.get("username")
+ password = request.args.get("password")
+ url = request.args.get("url")
+ if username is None or password is None or url is None:
+ return (
+ jsonify(
+ {"error": "username, password, and url parameters have to be provided"}
+ ),
+ 400,
+ )
+
+ query = "SELECT id, username, access_level FROM user WHERE username = ? AND password = ?"
+ result = query_db(query, (username, password), True)
+ if result is None:
+ return redirect(url)
+ session["user_info"] = (result[0], result[1], result[2])
+ return jsonify({"success": True})

flask_webgoat/status.py

@@ -1,14 +1,13 @@
-from flask import (
- Blueprint, jsonify
-)
+from flask import Blueprint, jsonify
 
-bp = Blueprint('status', __name__)
+bp = Blueprint("status", __name__)
 
-@bp.route('/status')
+
+@bp.route("/status")
 def status():
- return jsonify({'success': True})
+ return jsonify({"success": True})
 
 
-@bp.route('/ping')
+@bp.route("/ping")
 def ping():
- return jsonify({'success': True})
+ return jsonify({"success": True})

flask_webgoat/templates/base.html

@@ -0,0 +1,9 @@
+<!doctype html>
+<title>flask_webgoat</title>
+<nav>
+ <h1>Flask Webgoat</h1>
+</nav>
+<section class="content">
+ {% block content %}{% endblock %}
+</section>
+

flask_webgoat/templates/error.html

@@ -0,0 +1,11 @@
+{% extends 'base.html' %}
+
+{% block header %}
+ <h1>{% block title %}Register{% endblock %}</h1>
+{% endblock %}
+
+{% block content %}
+ <div class="error">
+ {{ message }}
+ </div>
+{% endblock %}

flask_webgoat/templates/search.html

@@ -0,0 +1,14 @@
+{% extends 'base.html' %}
+
+{% block header %}
+ <h1>{% block title %}Register{% endblock %}</h1>
+{% endblock %}
+
+{% block content %}
+ <div class="results">
+ Found {{ num_results }} results for query {{ query }}.
+ {% for result in results %}
+ <div class="result">{{ result }}</div>
+ {% endfor %}
+ </div>
+{% endblock %}

flask_webgoat/ui.py

@@ -0,0 +1,24 @@
+import sqlite3
+
+from flask import Blueprint, request, render_template
+from . import query_db
+
+bp = Blueprint("ui", __name__)
+
+
+@bp.route("/search")
+def search():
+ query_param = request.args.get("query")
+ if query_param is None:
+ message = "please provide the query parameter"
+ return render_template("error.html", message=message)
+
+ try:
+ query = "SELECT username, access_level FROM user WHERE username LIKE ?;"
+ results = query_db(query, (query_param,))
+ return render_template(
+ "search.html", results=results, num_results=len(results), query=query_param
+ )
+ except sqlite3.Error as err:
+ message = "Error while executing query " + query_param + ": " + err
+ return render_template("error.html", message=message)

flask_webgoat/users.py

@@ -1,35 +1,46 @@
-from flask import (
- Blueprint, jsonify, request, jsonify, session
-)
-from werkzeug.security import check_password_hash
-from flask import current_app as app
+import sqlite3
+
+from flask import Blueprint, jsonify, session, request
+
 from . import query_db
 
-bp = Blueprint('users', __name__)
+bp = Blueprint("users", __name__)
 
 
-@bp.route('/create_user', methods=['POST'])
+@bp.route("/create_user", methods=["POST"])
 def create_user():
- auth = session.get('user_info', None)
- if auth is None:
- return jsonify({'error': 'no auth found in session'})
+ user_info = session.get("user_info", None)
+ if user_info is None:
+ return jsonify({"error": "no user_info found in session"})
 
- access_level = auth[2]
+ access_level = user_info[2]
  if access_level != 0:
- return jsonify({'error': 'access level of 0 is required for this action'})
- username = request.form.get('username')
- password = request.form.get('password')
- access_level = request.form.get('access_level')
+ return jsonify({"error": "access level of 0 is required for this action"})
+ username = request.form.get("username")
+ password = request.form.get("password")
+ access_level = request.form.get("access_level")
  if username is None or password is None or access_level is None:
- return jsonify({'error': 'username, password and access_level parameters have to be provided'}), 400
+ return (
+ jsonify(
+ {
+ "error": "username, password and access_level parameters have to be provided"
+ }
+ ),
+ 400,
+ )
  if len(password) < 3:
- return jsonify({'error': 'the password needs to be at least 3 characters long'}), 402
+ return (
+ jsonify({"error": "the password needs to be at least 3 characters long"}),
+ 402,
+ )
 
- query = "INSERT INTO user (username, password, access_level) VALUES ('%s', '%s', %d)" % (username, password, int(access_level))
+ query = (
+ "INSERT INTO user (username, password, access_level) VALUES ('%s', '%s', %d)"
+ % (username, password, int(access_level))
+ )
 
  try:
  query_db(query, [], False, True)
- return jsonify({'success': True})
+ return jsonify({"success": True})
  except sqlite3.Error as err:
- return jsonify({'error': 'could not create user:' + err})
-
+ return jsonify({"error": "could not create user:" + err})

run.py

@@ -2,5 +2,11 @@
 
 app = create_app()
 
+@app.after_request
+def add_csp_headers(response):
+ response.headers['Access-Control-Allow-Origin'] = '*'
+ response.headers['Content-Security-Policy'] = "script-src 'self' 'unsafe-inline'"
+ return response
+
 if __name__ == '__main__':
  app.run()