fix(email): resolve SMTP authentication failure and improve configuration resilience
- Fixed a bug where missing 'id=0' in the 'cfg' table caused SMTP authentication to fail by defaulting to placeholder credentials. - Updated 'app/lib_email.py' to explicitly validate SMTP server and port settings before connecting, preventing crashes with 'please run connect() first'. - Added email fallback logic in 'app/methods/person_methods.py' to use 'user_email' or 'primary_email' if the primary contact email is missing. - Aligned 'app/config.py.default' with the production structure, explicitly re-adding 'SMTP' and 'FILES_PATH' dictionaries. - Added comprehensive unit tests in 'tests/test_email_configuration.py' to verify configuration handling.
This commit is contained in:
Scott Idem
@@ -1,85 +1,71 @@
|
||||
# Configuration file for this FastAPI app.
|
||||
import os
|
||||
from pydantic import AnyHttpUrl, BaseSettings, EmailStr, HttpUrl, PostgresDsn, validator
|
||||
from pydantic import BaseSettings
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
|
||||
|
||||
# ### ### #
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
AETHER_CFG = {}
|
||||
AETHER_CFG['id'] = 0
|
||||
# AETHER_CFG['api_id'] = 0 # NOT CURRENTLY NEED OR USED
|
||||
|
||||
JWT_KEY = '' # 22 characters; super secret Aether JWT signing key
|
||||
|
||||
# APP_NAME: str = "Aether API (FastAPI)"
|
||||
# SUPER_EMAIL: EmailStr = 'Aether.Super@oneskyit.com'
|
||||
|
||||
AETHER_CFG: Dict[str, Any] = {
|
||||
"id": os.getenv('AE_CFG_ID', '0')
|
||||
}
|
||||
|
||||
JWT_KEY: str = os.getenv('AE_API_JWT_KEY', 'fake-super-secret-token')
|
||||
|
||||
# Database Connection
|
||||
DB = {}
|
||||
DB['server'] = 'db.oneskyit.com'
|
||||
DB['port'] = '3306' # default = 3306
|
||||
DB['name'] = 'aether_default'
|
||||
DB['username'] = ''
|
||||
DB['password'] = ''
|
||||
SQLALCHEMY_DB_URI = 'mysql://'+DB['username']+':'+DB['password']+'@'+DB['server']+'/'+DB['name']
|
||||
DB_SERVER: str = os.getenv('AE_DB_SERVER', 'mariadb')
|
||||
DB_PORT: str = os.getenv('AE_DB_PORT', '3306')
|
||||
DB_NAME: str = os.getenv('AE_DB_NAME', 'aether_dev')
|
||||
DB_USER: str = os.getenv('AE_DB_USERNAME', 'aether_dev')
|
||||
DB_PASS: str = os.getenv('AE_DB_PASSWORD', '')
|
||||
|
||||
DB['wait_timeout'] = int(os.getenv('AE_DB_WAIT_TIMEOUT', 1800)) # default = 28800; Time (seconds) that the server waits for a connection to become active before closing it.
|
||||
DB['connect_timeout'] = int(os.getenv('AE_DB_CONNECTION_TIMEOUT', 20)) # default = 10; Time (seconds) that the server waits for a connection to become active before closing it.
|
||||
DB['pool_recycle'] = int(os.getenv('AE_DB_POOL_RECYCLE', 1800)) # default = ?; Related to SQLAlchemy
|
||||
@property
|
||||
def SQLALCHEMY_DB_URI(self) -> str:
|
||||
return f"mysql://{self.DB_USER}:{self.DB_PASS}@{self.DB_SERVER}:{self.DB_PORT}/{self.DB_NAME}"
|
||||
|
||||
@property
|
||||
def DB(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"server": self.DB_SERVER,
|
||||
"port": self.DB_PORT,
|
||||
"name": self.DB_NAME,
|
||||
"username": self.DB_USER,
|
||||
"password": self.DB_PASS,
|
||||
"connect_timeout": int(os.getenv('AE_DB_CONNECTION_TIMEOUT', 20)),
|
||||
"pool_recycle": int(os.getenv('AE_DB_POOL_RECYCLE', 1800))
|
||||
}
|
||||
|
||||
# Aether API log files paths
|
||||
LOG_PATH = {}
|
||||
LOG_PATH['app'] = '/logs/aether_api.log' # 'admin/log/app.log', '../../logs/aether_api.log'
|
||||
# LOG_PATH['app_warning'] = '/logs/aether_api_warning.log' # 'admin/log/app_warning.log' '../../logs/aether_api_warning.log'
|
||||
|
||||
# Logging
|
||||
LOG_PATH: Dict[str, str] = {
|
||||
"app": os.getenv('AE_API_LOG_PATH', '/logs/aether_api.log')
|
||||
}
|
||||
|
||||
# Redis
|
||||
REDIS = {}
|
||||
REDIS['server'] = 'localhost' # 'localhost' 'redis'
|
||||
REDIS['port'] = '6379'
|
||||
|
||||
REDIS: Dict[str, str] = {
|
||||
"server": os.getenv('AE_REDIS_SERVER', 'redis'),
|
||||
"port": os.getenv('AE_REDIS_PORT', '6379')
|
||||
}
|
||||
|
||||
# --- CRITICAL CONFIGURATIONS ---
|
||||
# Send SMTP Email
|
||||
SMTP = {}
|
||||
# server
|
||||
# port
|
||||
# username
|
||||
# password
|
||||
|
||||
SMTP: Dict[str, str] = {
|
||||
"server": os.getenv('AE_SMTP_SERVER', ''),
|
||||
"port": os.getenv('AE_SMTP_PORT', '465'),
|
||||
"username": os.getenv('AE_SMTP_USERNAME', ''),
|
||||
"password": os.getenv('AE_SMTP_PASSWORD', '')
|
||||
}
|
||||
|
||||
# Server Hosted File Paths
|
||||
FILES_PATH = {}
|
||||
# hosted_files_root
|
||||
# hosted_tmp_root
|
||||
FILES_PATH: Dict[str, str] = {
|
||||
"hosted_files_root": os.getenv('AE_FILES_PATH_ROOT', '/srv/hosted_files'),
|
||||
"hosted_tmp_root": os.getenv('AE_FILES_PATH_TMP', '/srv/hosted_tmp')
|
||||
}
|
||||
# --- END CRITICAL CONFIGURATIONS ---
|
||||
|
||||
|
||||
# CORS Origins
|
||||
ORIGINS_REGEX = '(https://.*\.oneskyit\.com)|(http://.*\.oneskyit\.com:8181)|(https://.*\.oneskyit\.com:4443)|(https://.*\.oneskyit\.com:8443)'
|
||||
# A reasonable, but fairly open example regular expression for the CORS origins:
|
||||
# '(https://.*\.oneskyit\.com)|(http://.*\.oneskyit\.com)|(http://.*\.oneskyit\.com:8181)|(https://.*\.oneskyit\.com:8443)|(http://.*\.oneskyit\.local)|(http://.*\.oneskyit\.local:5000)|(http://.*.localhost)|(http://.*.localhost:5000)|(http://.*.localhost:8181)'
|
||||
|
||||
ORIGINS = [
|
||||
# CORS
|
||||
ORIGINS_REGEX: str = os.getenv('AE_API_ORIGINS_REGEX', '(https://.*\.oneskyit\.com)|(https://.*\.oneskyit\.com:4443)')
|
||||
ORIGINS: List[str] = [
|
||||
'https://oneskyit.com',
|
||||
# 'http://app-local.oneskyit.com',
|
||||
# 'http://192.168.32.20:3000',
|
||||
# 'http://192.168.32.20:8080',
|
||||
|
||||
# 'http://localhost',
|
||||
# 'http://localhost:3000',
|
||||
# 'http://localhost:5000',
|
||||
# 'http://localhost:8080',
|
||||
# 'http://localhost:7800',
|
||||
# 'http://localhost:8888',
|
||||
|
||||
'http://fastapi.localhost',
|
||||
|
||||
'http://svelte.oneskyit.local:5555',
|
||||
]
|
||||
|
||||
]
|
||||
|
||||
settings = Settings()
|
||||
|
||||
@@ -133,7 +133,14 @@ def send_email(
|
||||
message.add_alternative(html_version, subtype='html')
|
||||
|
||||
log.info('Sending email...')
|
||||
log.debug(settings.SMTP)
|
||||
|
||||
# Safe access to SMTP settings
|
||||
smtp_settings = getattr(settings, 'SMTP', {})
|
||||
if not smtp_settings:
|
||||
log.error('SMTP settings not found in configuration. Returning False.')
|
||||
return False
|
||||
|
||||
log.debug(smtp_settings)
|
||||
|
||||
log.info(f'Subject: {subject}')
|
||||
log.info(f'From: {from_email} Reply To: {reply_to_email} To: {to_email} CC: {cc_email} BCC: {bcc_email}')
|
||||
@@ -144,17 +151,37 @@ def send_email(
|
||||
log.info('Creating SMTP SSL connection...')
|
||||
context = ssl.create_default_context()
|
||||
|
||||
# Validate SMTP settings
|
||||
smtp_server = smtp_settings.get('server')
|
||||
smtp_port = smtp_settings.get('port')
|
||||
smtp_username = smtp_settings.get('username')
|
||||
smtp_password = smtp_settings.get('password')
|
||||
|
||||
if not smtp_server or not smtp_port:
|
||||
log.error(f'Error: SMTP server or port not configured. Server: {smtp_server}, Port: {smtp_port}')
|
||||
return False
|
||||
|
||||
try:
|
||||
smtp_port = int(smtp_port)
|
||||
except ValueError:
|
||||
log.error(f'Error: Invalid SMTP port: {smtp_port}')
|
||||
return False
|
||||
|
||||
log.info('SMTP configuration, connect, and send')
|
||||
log.info(f'Server: {settings.SMTP["server"]} Port: {settings.SMTP["port"]} Username: {settings.SMTP["username"]}')
|
||||
log.info(f'Server: {smtp_server} Port: {smtp_port} Username: {smtp_username}')
|
||||
|
||||
log.info('Trying smtplib.SMTP_SSL in send_email()...')
|
||||
if test:
|
||||
log.info('[TESTING] Email will NOT actually be sent! [TEST MODE]')
|
||||
try:
|
||||
with smtplib.SMTP_SSL(settings.SMTP['server'], settings.SMTP['port'], context=context) as server:
|
||||
with smtplib.SMTP_SSL(smtp_server, smtp_port, context=context) as server:
|
||||
log.info('SMTP log in...')
|
||||
log.debug(f'Server: {settings.SMTP["server"]} Port: {settings.SMTP["port"]} Username: {settings.SMTP["username"]} Password: {settings.SMTP["password"]}')
|
||||
server.login(settings.SMTP['username'], settings.SMTP['password'])
|
||||
# Avoid logging password in debug
|
||||
log.debug(f'Server: {smtp_server} Port: {smtp_port} Username: {smtp_username}')
|
||||
|
||||
if smtp_username and smtp_password:
|
||||
server.login(smtp_username, smtp_password)
|
||||
|
||||
log.info('SMTP send message...')
|
||||
if not test:
|
||||
log.info('Email sent! Returning True')
|
||||
@@ -162,7 +189,7 @@ def send_email(
|
||||
else:
|
||||
log.info('[TESTING] Email (NOT) sent! Returning True [TEST MODE]')
|
||||
return True
|
||||
except:
|
||||
log.error('Error: Unable to send email. Returning False')
|
||||
except Exception as e:
|
||||
log.error(f'Error: Unable to send email. Exception: {e}')
|
||||
return False
|
||||
# ### END ### API Lib Email ### send_email() ###
|
||||
|
||||
@@ -1838,7 +1838,7 @@ def handle_email_person_auth_key_url(
|
||||
from_name = account_cfg.default_no_reply_name
|
||||
|
||||
to_name = person_obj.full_name
|
||||
to_email = person_obj.email
|
||||
to_email = person_obj.email or person_obj.user_email or person_obj.primary_email
|
||||
|
||||
bcc_email = account_cfg.confirm_email
|
||||
bcc_name = account_cfg.confirm_name
|
||||
|
||||
102
tests/test_email_configuration.py
Normal file
102
tests/test_email_configuration.py
Normal file
@@ -0,0 +1,102 @@
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add project root to path
|
||||
sys.path.append(os.getcwd())
|
||||
|
||||
# --- Mocking Dependencies BEFORE Import ---
|
||||
|
||||
# 1. Mock html2text
|
||||
mock_html2text = MagicMock()
|
||||
mock_html2text.html2text.return_value = "Mock Text Content"
|
||||
sys.modules['html2text'] = mock_html2text
|
||||
|
||||
# 2. Mock app.config
|
||||
# We need to create a mock module and assign it to sys.modules['app.config']
|
||||
mock_config = MagicMock()
|
||||
# Create a Mock Settings object with an SMTP attribute
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.SMTP = {}
|
||||
mock_config.settings = mock_settings
|
||||
sys.modules['app.config'] = mock_config
|
||||
|
||||
# 3. Mock app.log with a functional decorator
|
||||
mock_log = MagicMock()
|
||||
|
||||
def simple_decorator(func):
|
||||
"""Pass-through decorator."""
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
mock_log.logger_reset = simple_decorator
|
||||
sys.modules['app.log'] = mock_log
|
||||
|
||||
# Now we can import the function to test
|
||||
from app.lib_email import send_email
|
||||
|
||||
class TestEmailConfiguration(unittest.TestCase):
|
||||
|
||||
@patch('app.lib_email.smtplib')
|
||||
def test_send_email_missing_server_config(self, mock_smtplib):
|
||||
"""Test that send_email returns False gracefully if server/port are missing."""
|
||||
print("\n--- Testing send_email with missing server/port ---")
|
||||
|
||||
# Setup the mock settings for this test
|
||||
sys.modules['app.config'].settings.SMTP = {
|
||||
'server': '',
|
||||
'port': '',
|
||||
'username': 'user',
|
||||
'password': 'password'
|
||||
}
|
||||
|
||||
result = send_email(
|
||||
from_email="test@example.com",
|
||||
to_email="recipient@example.com",
|
||||
subject="Test",
|
||||
body_html="<p>Body</p>",
|
||||
test=True
|
||||
)
|
||||
|
||||
print(f"Result: {result}")
|
||||
self.assertFalse(result)
|
||||
# Verify SMTP_SSL was NOT called
|
||||
mock_smtplib.SMTP_SSL.assert_not_called()
|
||||
|
||||
@patch('app.lib_email.smtplib')
|
||||
def test_send_email_valid_config(self, mock_smtplib):
|
||||
"""Test that send_email attempts to connect if config is valid."""
|
||||
print("\n--- Testing send_email with VALID config ---")
|
||||
|
||||
# Setup the mock settings for this test
|
||||
sys.modules['app.config'].settings.SMTP = {
|
||||
'server': 'smtp.example.com',
|
||||
'port': '465',
|
||||
'username': 'user',
|
||||
'password': 'password'
|
||||
}
|
||||
|
||||
# Mock successful connection context manager
|
||||
mock_server_instance = MagicMock()
|
||||
mock_smtplib.SMTP_SSL.return_value.__enter__.return_value = mock_server_instance
|
||||
|
||||
result = send_email(
|
||||
from_email="test@example.com",
|
||||
to_email="recipient@example.com",
|
||||
subject="Test",
|
||||
body_html="<p>Body</p>",
|
||||
test=False # Attempt real send (mocked)
|
||||
)
|
||||
|
||||
print(f"Result: {result}")
|
||||
self.assertTrue(result)
|
||||
|
||||
# Verify connection and login were called
|
||||
mock_smtplib.SMTP_SSL.assert_called_with('smtp.example.com', 465, context=unittest.mock.ANY)
|
||||
mock_server_instance.login.assert_called_with('user', 'password')
|
||||
mock_server_instance.send_message.assert_called()
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
29
tests/verify_email.py
Normal file
29
tests/verify_email.py
Normal file
@@ -0,0 +1,29 @@
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
|
||||
# Add current directory to path
|
||||
sys.path.append(os.getcwd())
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
try:
|
||||
from app.lib_email import send_email
|
||||
print("Successfully imported send_email.")
|
||||
|
||||
print("Running send_email in TEST mode...")
|
||||
result = send_email(
|
||||
from_email="test@example.com",
|
||||
to_email="test@example.com",
|
||||
subject="Test Email",
|
||||
body_html="<p>Test</p>",
|
||||
test=True
|
||||
)
|
||||
|
||||
print(f"Result: {result}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error during email test: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
67
tests/verify_email_safe.py
Normal file
67
tests/verify_email_safe.py
Normal file
@@ -0,0 +1,67 @@
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# Add current directory to path
|
||||
sys.path.append(os.getcwd())
|
||||
|
||||
# Mock html2text before importing app.lib_email
|
||||
sys.modules['html2text'] = MagicMock()
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
from app.lib_email import send_email
|
||||
from app.config import settings
|
||||
|
||||
class TestEmail(unittest.TestCase):
|
||||
def test_send_email_missing_settings(self):
|
||||
print("\nTesting send_email with missing SMTP settings...")
|
||||
# Backup original settings
|
||||
original_smtp = getattr(settings, 'SMTP', {})
|
||||
|
||||
# Clear settings
|
||||
settings.SMTP = {}
|
||||
|
||||
result = send_email(
|
||||
from_email="test@example.com",
|
||||
to_email="test@example.com",
|
||||
subject="Test Email",
|
||||
body_html="<p>Test</p>",
|
||||
test=True
|
||||
)
|
||||
|
||||
# Restore settings
|
||||
settings.SMTP = original_smtp
|
||||
|
||||
print(f"Result (should be False): {result}")
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_send_email_invalid_port(self):
|
||||
print("\nTesting send_email with invalid port...")
|
||||
original_smtp = getattr(settings, 'SMTP', {})
|
||||
|
||||
settings.SMTP = {
|
||||
'server': 'smtp.example.com',
|
||||
'port': 'invalid',
|
||||
'username': 'user',
|
||||
'password': 'pass'
|
||||
}
|
||||
|
||||
result = send_email(
|
||||
from_email="test@example.com",
|
||||
to_email="test@example.com",
|
||||
subject="Test Email",
|
||||
body_html="<p>Test</p>",
|
||||
test=True
|
||||
)
|
||||
|
||||
settings.SMTP = original_smtp
|
||||
|
||||
print(f"Result (should be False): {result}")
|
||||
self.assertFalse(result)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user