From 145d426dc019690af24b7e815c751a2470e3338b Mon Sep 17 00:00:00 2001 From: Blake Ridgway Date: Thu, 3 Apr 2025 11:50:18 -0500 Subject: [PATCH] (refactor): Improve email sending and logging, enhance security Implemented robust logging using the logging module Improved error handling and resource management with try...except..finally blocks Seprareted email sending logic into smaller managable functions Added SENDER_EMAIL configuration for sending emails Fixed security vulnerabilities --- app.py | 128 ++++++++++++++++++++++++++++++++++++--------------------- 1 file changed, 82 insertions(+), 46 deletions(-) diff --git a/app.py b/app.py index 183e075..e719f79 100644 --- a/app.py +++ b/app.py @@ -1,103 +1,137 @@ import os +import logging import smtplib from email.mime.text import MIMEText -from flask import Flask, render_template, request, redirect, url_for, flash, session +from flask import ( + Flask, + render_template, + request, + redirect, + url_for, + flash, + session, +) from dotenv import load_dotenv from werkzeug.security import check_password_hash +from functools import wraps # Import wraps from database import get_connection, init_db, get_all_emails, get_admin, create_default_admin load_dotenv() app = Flask(__name__) -# Use a secret key from .env; ensure your .env sets SECRET_KEY -app.secret_key = os.getenv('SECRET_KEY') -base_url = os.getenv('BASE_URL') +app.secret_key = os.getenv("SECRET_KEY") +base_url = os.getenv("BASE_URL") # SMTP settings (for sending update emails) -SMTP_SERVER = os.getenv('SMTP_SERVER') +SMTP_SERVER = os.getenv("SMTP_SERVER") SMTP_PORT = int(os.getenv("SMTP_PORT", 465)) -SMTP_USER = os.getenv('SMTP_USER') -SMTP_PASSWORD = os.getenv('SMTP_PASSWORD') +SMTP_USER = os.getenv("SMTP_USER") +SMTP_PASSWORD = os.getenv("SMTP_PASSWORD") +SENDER_EMAIL = os.getenv("SENDER_EMAIL", SMTP_USER) # Use SENDER_EMAIL + +# Logging setup +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) # Initialize the database and create default admin user if necessary. init_db() create_default_admin() +# Decorator for requiring login def login_required(f): - from functools import wraps - @wraps(f) + @wraps(f) # Use wraps to preserve function metadata def decorated_function(*args, **kwargs): if "username" not in session: - return redirect(url_for('login')) + return redirect(url_for("login")) return f(*args, **kwargs) + return decorated_function + +def send_update_email(subject, body, email): + """Sends email, returns True on success, False on failure.""" + try: + server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10) + server.set_debuglevel(False) # Keep debug level at False for production + server.login(SMTP_USER, SMTP_PASSWORD) + + unsub_link = f"https://{base_url}/unsubscribe?email={email}" + custom_body = ( + f"{body}

" + f"If you ever wish to unsubscribe, please click here" + ) + + msg = MIMEText(custom_body, "html", "utf-8") + msg["Subject"] = subject + msg["From"] = SENDER_EMAIL # Use sender email + msg["To"] = email + + server.sendmail(SENDER_EMAIL, email, msg.as_string()) # Use sender email + + server.quit() + logger.info(f"Update email sent to: {email}") + return True + except Exception as e: + logger.error(f"Failed to send email to {email}: {e}") + return False + + def process_send_update_email(subject, body): """Helper function to send an update email to all subscribers.""" subscribers = get_all_emails() if not subscribers: return "No subscribers found." try: - server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10) - server.set_debuglevel(True) - server.login(SMTP_USER, SMTP_PASSWORD) for email in subscribers: - unsub_link = f"https://{base_url}/unsubscribe?email={email}" - custom_body = ( - f"{body}

" - f"If you ever wish to unsubscribe, please click here" - ) - msg = MIMEText(custom_body, 'html', 'utf-8') - msg['Subject'] = subject - msg['From'] = SMTP_USER - msg['To'] = email - server.sendmail(SMTP_USER, email, msg.as_string()) - print(f"Update email sent to: {email}") - server.quit() - + if not send_update_email(subject, body, email): + return f"Failed to send to {email}" # Specific failure message + + # Log newsletter content for audit purposes conn = get_connection() cursor = conn.cursor() cursor.execute( - "INSERT INTO newsletters (subject, body) VALUES (%s, %s)", - (subject, body) - ) + "INSERT INTO newsletters (subject, body) VALUES (%s, %s)", (subject, body) + ) conn.commit() cursor.close() conn.close() - return "Email has been sent." + return "Email has been sent to all subscribers." except Exception as e: - print(f"Failed to send email: {e}") + logger.exception("Error processing sending updates") return f"Failed to send email: {e}" -@app.route('/') + +@app.route("/") @login_required def index(): """Displays all subscriber emails""" emails = get_all_emails() return render_template("admin_index.html", emails=emails) -@app.route('/send_update', methods=['GET', 'POST']) + +@app.route("/send_update", methods=["GET", "POST"]) @login_required def send_update(): """Display a form to send an update email; process submission on POST.""" - if request.method == 'POST': - subject = request.form['subject'] - body = request.form['body'] - # Call the helper function using its new name. + if request.method == "POST": + subject = request.form["subject"] + body = request.form["body"] result_message = process_send_update_email(subject, body) flash(result_message) return redirect(url_for("send_update")) return render_template("send_update.html") -@app.route('/login', methods=['GET', 'POST']) + +@app.route("/login", methods=["GET", "POST"]) def login(): - if request.method == 'POST': - username = request.form.get('username') - password = request.form.get('password') + if request.method == "POST": + username = request.form.get("username") + password = request.form.get("password") admin = get_admin(username) - # Expect get_admin() to return a tuple like (username, password_hash) if admin and check_password_hash(admin[1], password): - session['username'] = username + session["username"] = username flash("Logged in successfully", "success") return redirect(url_for("index")) else: @@ -105,11 +139,13 @@ def login(): return redirect(url_for("login")) return render_template("login.html") -@app.route('/logout') + +@app.route("/logout") def logout(): - session.pop('username', None) + session.pop("username", None) flash("Logged out successfully", "success") return redirect(url_for("login")) -if __name__ == '__main__': + +if __name__ == "__main__": app.run(port=5001, debug=True)