security: remediate 7 appsec findings in Flask wine app

templates/login.html

@@ -93,12 +93,11 @@ <h1 style="text-align:left">Please Login:</h1>
       </div>
       <div class="form-group">
         <label for="password">Password: </label>
-        <input type="text" id="password" name="password" required><br>
+        <input type="password" id="password" name="password" required><br>
       </div>
     </div>
     <button type="submit" class="btn btn-primary btn-block btn-large">Login</button>
   </form>
-  <br>
   <a href="{{ url_for('register') }}">Not registered? Click here</a>
   <br><br>
   <b>{{login_text}}</b>

templates/profile.html

@@ -93,7 +93,7 @@ <h2>Change Password</h2>
   <form method="POST" action="/change_password">
     <div class="form-group">
       <label for="current_password">Current Password:</label>
-      <input type="password" id="current_password" name="current_password" value="{{ current_password or ''}}" required>
+      <input type="password" id="current_password" name="current_password" required>
     </div>
     <div class="form-group">
       <label for="new_password">New Password:</label>

templates/register.html

@@ -101,11 +101,11 @@ <h1 style="text-align:left">Please Register As A User</h1>
       </div>
       <div class="form-group">
         <label for="password">Password: </label>
-        <input type="text" id="password" name="password" required><br>
+        <input type="password" id="password" name="password" required><br>
       </div>
       <div class="form-group">
         <label for="confirmpw">Confirm Password: </label>
-        <input type="text" id="confirmpw" name="confirmpw" required><br>
+        <input type="password" id="confirmpw" name="confirmpw" required><br>
       </div>
       <div class="form-group">
         <label for="email">Email Address: </label>

tests/test_security.py

@@ -0,0 +1,151 @@
+import pytest
+from unittest.mock import patch, MagicMock
+from werkzeug.security import generate_password_hash
+from wine import app
+
+
+@pytest.fixture
+def client():
+    app.config['TESTING'] = True
+    with app.test_client() as client:
+        yield client
+
+
+class TestDebugMode:
+    def test_debug_mode_disabled(self):
+        # Debug mode must be off to prevent Werkzeug interactive debugger exposure
+        assert app.debug is False
+
+
+class TestSecretKey:
+    def test_secret_key_is_not_hardcoded_default(self):
+        # Secret key must not be the known-bad default value
+        assert app.secret_key != b'your_secret_key'
+        assert app.secret_key != 'your_secret_key'
+
+    def test_secret_key_has_sufficient_length(self):
+        # Secret key should be at least 16 bytes
+        key = app.secret_key
+        key_bytes = key if isinstance(key, bytes) else key.encode()
+        assert len(key_bytes) >= 16
+
+
+class TestLoginHashedPassword:
+    def _make_mock_account(self, username='testuser', password='TestPass1',
+                           email='test@example.com'):
+        hashed = generate_password_hash(password)
+        return (1, username, hashed, email)
+
+    def test_login_succeeds_with_correct_hashed_password(self, client):
+        # Login must succeed when the submitted password matches the stored hash
+        account = self._make_mock_account()
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = account
+
+        with patch('sqlite3.connect') as mock_connect:
+            mock_connect.return_value.cursor.return_value = mock_cursor
+            response = client.post('/login', data={
+                'username': 'testuser',
+                'password': 'TestPass1'
+            })
+
+        assert b'Logged in successfully!' in response.data
+
+    def test_login_fails_with_wrong_password(self, client):
+        # Login must fail when password does not match the stored hash
+        account = self._make_mock_account()
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = account
+
+        with patch('sqlite3.connect') as mock_connect:
+            mock_connect.return_value.cursor.return_value = mock_cursor
+            response = client.post('/login', data={
+                'username': 'testuser',
+                'password': 'WrongPassword'
+            })
+
+        assert b'Incorrect username/password!' in response.data
+
+    def test_login_fails_with_plaintext_password_in_db(self, client):
+        # If the DB somehow contains plaintext, login must still fail (not bypass hash check)
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = (1, 'testuser', 'TestPass1', 'test@example.com')
+
+        with patch('sqlite3.connect') as mock_connect:
+            mock_connect.return_value.cursor.return_value = mock_cursor
+            response = client.post('/login', data={
+                'username': 'testuser',
+                'password': 'TestPass1'
+            })
+
+        assert b'Logged in successfully!' not in response.data
+
+
+class TestRegisterPasswordHash:
+    def test_register_stores_hashed_password(self, client):
+        # Registration must hash the password before storing it
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = None  # user does not exist
+
+        inserted_values = {}
+
+        def capture_execute(sql, params=None):
+            if params and 'INSERT' in sql.upper():
+                inserted_values['password'] = params[1]
+
+        mock_cursor.execute.side_effect = capture_execute
+        mock_cursor.connection.commit.return_value = None
+
+        with patch('sqlite3.connect') as mock_connect:
+            mock_connect.return_value.cursor.return_value = mock_cursor
+            client.post('/register', data={
+                'username': 'newuser',
+                'password': 'MySecret1',
+                'confirmpw': 'MySecret1',
+                'email': 'new@example.com'
+            })
+
+        assert 'password' in inserted_values
+        stored_pw = inserted_values['password']
+        # Stored value must not be the plaintext password
+        assert stored_pw != 'MySecret1'
+        # Stored value must be a valid Werkzeug hash
+        from werkzeug.security import check_password_hash
+        assert check_password_hash(stored_pw, 'MySecret1')
+
+    def test_register_rejects_mismatched_passwords(self, client):
+        # Server must reject registration when password and confirmation differ
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = None
+
+        with patch('sqlite3.connect') as mock_connect:
+            mock_connect.return_value.cursor.return_value = mock_cursor
+            response = client.post('/register', data={
+                'username': 'newuser',
+                'password': 'MySecret1',
+                'confirmpw': 'DifferentPassword',
+                'email': 'new@example.com'
+            })
+
+        assert b'Passwords do not match!' in response.data
+
+
+class TestPredictAuthGuard:
+    def test_predict_get_unauthenticated_redirects_to_login(self, client):
+        # Unauthenticated GET /predict must redirect to login
+        response = client.get('/predict')
+        assert response.status_code == 302
+        assert '/login' in response.headers['Location']
+
+    def test_predict_post_unauthenticated_redirects_to_login(self, client):
+        # Unauthenticated POST /predict must redirect to login
+        response = client.post('/predict', data={
+            'alcohol': '13.0', 'malic_acid': '2.0', 'ash': '2.3',
+            'alcalinity_of_ash': '15.0', 'magnesium': '100.0',
+            'total_phenols': '2.5', 'flavanoids': '2.5',
+            'nonflavanoid_phenols': '0.3', 'proanthocyanins': '1.5',
+            'color_intensity': '5.0', 'hue': '1.0',
+            'od280_od315_of_diluted_wines': '3.0', 'proline': '700.0'
+        })
+        assert response.status_code == 302
+        assert '/login' in response.headers['Location']

wine.py

@@ -15,6 +15,8 @@
 import pickle
 import signal
 import re
+import os
+from werkzeug.security import generate_password_hash, check_password_hash
 
 # Create a signal handler to shut down the server gracefully on shutdown
 def shutdown(signal_number, frame):
@@ -26,7 +28,9 @@ def shutdown(signal_number, frame):
 # Load the model from the pickle file
 app = Flask(__name__)
 model = pickle.load(open('model.pkl', 'rb'))
-app.secret_key = 'your_secret_key'
+# Set FLASK_SECRET_KEY env var in production. The os.urandom fallback generates a new
+# key on every restart (invalidating all sessions) and is only suitable for development.
+app.secret_key = os.environ.get('FLASK_SECRET_KEY', os.urandom(32))
 
 # # Configure MySQL service
 # app.config['MYSQL_HOST'] = 'localhost'
@@ -37,7 +41,7 @@ def shutdown(signal_number, frame):
 # mysql = MySQL(app)
 
 # Turn debugging mode off for production
-app.debug = True 
+app.debug = False
 
 # create a routine to notify test scipts that the server is up and running
 @app.route("/health")
@@ -60,10 +64,12 @@ def login():
         password = request.form['password']
         cursor = sqlite3.connect('wineusers.db').cursor()
         if cursor:
-            cursor.execute('SELECT * FROM accounts WHERE username = ? AND password = ?', (username, password))
+            cursor.execute('SELECT * FROM accounts WHERE username = ?', (username,))
             account = cursor.fetchone()
+            conn = cursor.connection
             cursor.close()
-            if account:
+            conn.close()
+            if account and check_password_hash(account[2], password):
                 session['loggedin'] = True
                 session['id'] = account[0]
                 session['username'] = account[1]
@@ -101,13 +107,11 @@ def register():
     if request.method == 'POST' and 'username' in request.form and 'password' in request.form and 'email' in request.form :
         username = request.form['username']
         password = request.form['password']
+        confirmpw = request.form.get('confirmpw', '')
         email = request.form['email']
         cursor = sqlite3.connect('wineusers.db').cursor()
         cursor.execute('SELECT * FROM accounts WHERE username = ?', (username, ))
         account = cursor.fetchone()
-        # cursor = mysql.connection.cursor()
-        # cursor.execute('SELECT * FROM accounts WHERE username = % s', (username, ))
-        # account = cursor.fetchone()
         if account:
             msg = 'Account already exists!'
         elif not re.match(r'[^@]+@[^@]+\.[^@]+', email):
@@ -116,12 +120,13 @@ def register():
             msg = 'Username must contain only characters and numbers!'
         elif not username or not password or not email:
             msg = 'Please fill out the form!'
+        elif password != confirmpw:
+            msg = 'Passwords do not match!'
         else:
-            cursor.execute('INSERT INTO accounts VALUES (NULL, ?, ?, ?)', (username, password, email, ))
+            hashed_password = generate_password_hash(password)
+            cursor.execute('INSERT INTO accounts VALUES (NULL, ?, ?, ?)', (username, hashed_password, email, ))
             cursor.connection.commit()
             cursor.close()
-            # cursor.execute('INSERT INTO accounts VALUES (NULL, % s, % s, % s)', (username, password, email, ))
-            # mysql.connection.commit()
             msg = 'You have successfully registered!'
     elif request.method == 'POST':
         msg = 'Please fill out the form!'
@@ -143,6 +148,8 @@ def contact():
 # predict uses the machine learning model to predict the wine type based on the user's input
 @app.route("/predict", methods=['GET', 'POST'])
 def predict():
+    if 'username' not in session:
+        return redirect(url_for('login'))
     if request.method == 'GET':
         return render_template('wine.html')
     else:
@@ -177,22 +184,24 @@ def change_password():
             return render_template('profile.html', username_text='Username: {}'.format(session['username']), email_text='Email: {}'.format(session['email']), profile_text='New passwords do not match!')
 
         cursor = sqlite3.connect('wineusers.db').cursor()
-        cursor.execute('SELECT * FROM accounts WHERE username = ? AND password = ?', (session['username'], current_password))
+        cursor.execute('SELECT * FROM accounts WHERE username = ?', (session['username'],))
         account = cursor.fetchone()
-        # cursor = mysql.connection.cursor()
-        # cursor.execute('SELECT * FROM accounts WHERE username = %s AND password = %s', (session['username'], current_password))
-        # account = cursor.fetchone()
 
-        if account:
-            cursor.execute('UPDATE accounts SET password = ? WHERE username = ?', (new_password, session['username']))
+        if account and check_password_hash(account[2], current_password):
+            hashed_new_password = generate_password_hash(new_password)
+            cursor.execute('UPDATE accounts SET password = ? WHERE username = ?', (hashed_new_password, session['username']))
             cursor.connection.commit()
+            conn = cursor.connection
             cursor.close()
+            conn.close()
             # cursor.execute('UPDATE accounts SET password = %s WHERE username = %s', (new_password, session['username']))
             # mysql.connection.commit()
             return render_template('profile.html', username_text='Username: {}'.format(session['username']), email_text='Email: {}'.format(session['email']), profile_text='Password changed successfully!')
         else:
+            conn = cursor.connection
             cursor.close()
-            return render_template('profile.html', username_text='Username: {}'.format(session['username']), email_text='Email: {}'.format(session['email']), current_password=current_password, profile_text='Current password is incorrect!')
+            conn.close()
+            return render_template('profile.html', username_text='Username: {}'.format(session['username']), email_text='Email: {}'.format(session['email']), profile_text='Current password is incorrect!')
     else:
         return redirect(url_for('login'))