Hash emails (#30)

* Add development container configuration with Dockerfile and VSCode settings

* Update Super Linter configuration to use latest version and exclude specific files from validation

* Update Super Linter workflow to trigger on pushes to the main branch

* Add email encryption and hashing functions; update authentication routes

* Restore login endpoint (oopsie)

* Add email masking functionality and update person routes to handle encrypted emails

* Implement email masking in person routes by decrypting and masking email addresses

* Fix typo in utility.py: correct function name from 'decrpyt_email' to 'decrypt_email'

* Refactor AES_KEY initialization to use bytes.fromhex for better security

* Enhance error handling in app and improve email masking logic in person routes

* Improve error message extraction to handle SQL errors more gracefully

* Add command-line argument for debug mode in Flask app

* Refactor error handling in email decryption and improve SQL error message extraction

* Refactor import statements in authentication module and clean up utility exports

Author: Vianpyro

app.py

@@ -1,4 +1,7 @@
-from flask import Flask, jsonify
+import argparse
+import traceback
+
+from flask import Flask, jsonify, request
 from flask_cors import CORS
 
 from config import Config, limiter
@@ -56,11 +59,29 @@ def ratelimit_error(e):
 
 @app.errorhandler(Exception)
 def handle_exception(e):
-    return (
-        jsonify(error="Internal Server Error", message=extract_error_message(str(e))),
-        500,
-    )
+    # If the app is in debug mode, return the full traceback
+    if app.debug:
+        return (
+            jsonify(
+                error="Internal Server Error",
+                message=str(e),
+                type=type(e).__name__,
+                url=request.url,
+                traceback=traceback.format_exc().splitlines(),
+            ),
+            500,
+        )
+
+    # Otherwise, return a more user-friendly error message
+    error_message = extract_error_message(str(e))
+    return jsonify(error="Internal Server Error", message=error_message), 500
 
 
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=5000)
+    parser = argparse.ArgumentParser(description="Run the Flask application.")
+    parser.add_argument(
+        "--debug", action="store_true", help="Run the app in debug mode."
+    )
+    args = parser.parse_args()
+
+    app.run(host="0.0.0.0", port=5000, debug=args.debug)

requirements.txt

@@ -1,4 +1,5 @@
 argon2-cffi>=23.1.0
+cryptography>=44.0.2
 Flask>=3.0.3
 Flask-JWT-Extended>=2.8.0
 Flask-Limiter>=3.7.0

routes/authentication.py

@@ -3,7 +3,6 @@
 from pymysql import MySQLError
 
 from config import limiter
-
 from jwt_helper import (
     TokenError,
     extract_token_from_header,
@@ -13,6 +12,8 @@
 )
 from utility import (
     database_cursor,
+    encrypt_email,
+    hash_email,
     hash_password,
     validate_password,
     verify_password,
@@ -22,8 +23,10 @@
 
 
 def login_person_by_email(email):
+    email_hash = hash_email(email)
+
     with database_cursor() as cursor:
-        cursor.callproc("login_person_by_email", (email,))
+        cursor.callproc("login_person_by_email", (email_hash,))
         return cursor.fetchone()
 
 
@@ -48,11 +51,12 @@ def register():
         return jsonify(message="Password does not meet security requirements"), 400
 
     hashed_password = hash_password(password)
+    email = hash_email(email), encrypt_email(email)
 
     try:
         with database_cursor() as cursor:
             cursor.callproc(
-                "register_person", (name, email, hashed_password, language_code)
+                "register_person", (name, *email, hashed_password, language_code)
             )
     except MySQLError as e:
         if "User name already exists" in str(e):

routes/person.py

@@ -3,25 +3,49 @@
 
 from utility import (
     database_cursor,
+    decrypt_email,
+    encrypt_email,
+    hash_email,
     hash_password,
+    mask_email,
     validate_password,
     verify_password,
 )
 
 person_blueprint = Blueprint("person", __name__)
 
 
-def login_person_by_id(person_id):
+def login_person_by_id(person_id: int) -> dict:
     with database_cursor() as cursor:
         cursor.callproc("login_person_by_id", (person_id,))
         return cursor.fetchone()
 
 
+def mask_person_email(person: dict) -> None:
+    """Mask the email address of a person safely."""
+    encrypted_email = person.get("encrypted_email")
+
+    if not encrypted_email:
+        person["email"] = "Unknown"
+        return
+
+    try:
+        person["email"] = mask_email(decrypt_email(encrypted_email))
+    except Exception:
+        person["email"] = "Decryption Error"
+
+    # Remove unreadable fields
+    person.pop("encrypted_email", None)
+    person.pop("hashed_password", None)
+
+
 def update_person_in_db(person_id, name, email, hashed_password, locale_code):
+    email = hash_email(email), encrypt_email(email)
+
     with database_cursor() as cursor:
         cursor.callproc(
             "update_person",
-            (person_id, name, email, hashed_password, locale_code),
+            (person_id, name, *email, hashed_password, locale_code),
         )
 
 
@@ -30,6 +54,10 @@ def get_all_persons():
     with database_cursor() as cursor:
         cursor.callproc("get_all_persons")
         persons = cursor.fetchall()
+
+    for person in persons:
+        mask_person_email(person)
+
     return jsonify(persons)
 
 
@@ -38,6 +66,8 @@ def get_person_by_id(person_id):
     with database_cursor() as cursor:
         cursor.callproc("get_person_by_id", (person_id,))
         person = cursor.fetchone()
+
+    mask_person_email(person)
     return jsonify(person)
 
 

utility.py

@@ -1,14 +1,20 @@
+import base64
+import hashlib
 import os
+import re
 from contextlib import contextmanager
 from re import match
 
 from argon2 import PasswordHasher
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from dotenv import load_dotenv
 
 from db import get_db_connection
 
 load_dotenv()
 ph = PasswordHasher()
+AES_KEY = bytes.fromhex(os.getenv("AES_SECRET_KEY", os.urandom(32).hex()))
 PEPPER = os.getenv("PEPPER", "SuperSecretPepper").encode("utf-8")
 
 
@@ -27,18 +33,69 @@ def database_cursor():
         db.close()
 
 
+def decrypt_email(encrypted_email: str) -> str:
+    """Decrypts an AES-256 encrypted email."""
+    encrypted_data = base64.b64decode(encrypted_email)
+    iv, ciphertext = encrypted_data[:16], encrypted_data[16:]
+
+    cipher = Cipher(algorithms.AES(AES_KEY), modes.CBC(iv), backend=default_backend())
+    decryptor = cipher.decryptor()
+    decrypted_email = decryptor.update(ciphertext) + decryptor.finalize()
+
+    return decrypted_email.strip().decode()
+
+
+def encrypt_email(email: str) -> str:
+    """Encrypts an email using AES-256."""
+    iv = os.urandom(16)  # Generate a random IV
+    cipher = Cipher(algorithms.AES(AES_KEY), modes.CBC(iv), backend=default_backend())
+    encryptor = cipher.encryptor()
+
+    # Pad email to 16-byte blocks
+    padded_email = email + (16 - len(email) % 16) * " "
+    ciphertext = encryptor.update(padded_email.encode()) + encryptor.finalize()
+
+    # Store IV + ciphertext (Base64 encoded)
+    return base64.b64encode(iv + ciphertext).decode()
+
+
 def extract_error_message(message):
     try:
-        return message.split(", ")[1].strip("()'")
+        cleaner_message = message.split(", ")[1].strip("()'")
+        return cleaner_message if "SQL" not in cleaner_message else "Database error"
     except IndexError:
         return "An unknown error occurred"
 
 
+def hash_email(email: str) -> str:
+    """Generate a SHA-256 hash of the email (used for fast lookup)."""
+    return hashlib.sha256(email.encode()).hexdigest()
+
+
 def hash_password(password: str) -> tuple[str, bytes]:
     peppered_password = password.encode("utf-8") + PEPPER
     return ph.hash(peppered_password)  # Argon2 applies salt automatically
 
 
+def mask_email(email: str) -> str:
+    """Masks the email address to protect user privacy."""
+    match = re.match(r"^([\w.+-]+)@([\w-]+)\.([a-zA-Z]{2,})$", email)
+    if not match:
+        raise ValueError("Invalid email format")
+
+    local_part, domain_name, domain_extension = match.groups()
+
+    # Mask the local part
+    if len(local_part) > 2:
+        local_part = local_part[0] + "*" * (len(local_part) - 2) + local_part[-1]
+
+    # Mask the domain name
+    if len(domain_name) > 2:
+        domain_name = domain_name[0] + "*" * (len(domain_name) - 2) + domain_name[-1]
+
+    return f"{local_part}@{domain_name}.{domain_extension}"
+
+
 def validate_password(password):
     """
     Validates a password based on the following criteria: