#!/usr/bin/env python3
"""
Email service for VPN Access Server.
Provides SMTP-based email functionality for sending notifications,
user guides with attachments, and log files from NAS.
"""
import os
import smtplib
import logging
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from email.mime.base import MIMEBase
from email import encoders
from typing import List, Optional, Dict, Any
from pathlib import Path
from access.config import config
from access.utils import setup_logging
logger = logging.getLogger(__name__)
class EmailService:
"""SMTP email service for sending various types of notifications."""
def __init__(self):
"""Initialize email service with configuration."""
self.smtp_host = config.email.smtp_host
self.smtp_port = config.email.smtp_port
self.smtp_username = config.email.smtp_username
self.smtp_password = config.email.smtp_password
self.use_tls = config.email.smtp_use_tls
self.from_email = config.email.from_email
self.from_name = config.email.from_name
# Set up logger if not already configured
if not logger.handlers:
self.logger = setup_logging()
else:
self.logger = logger
def _create_smtp_connection(self) -> smtplib.SMTP:
"""
Create and configure SMTP connection.
Returns:
Configured SMTP connection
Raises:
smtplib.SMTPException: If connection fails
"""
try:
if self.use_tls:
server = smtplib.SMTP(self.smtp_host, self.smtp_port)
server.starttls()
else:
server = smtplib.SMTP_SSL(self.smtp_host, self.smtp_port)
server.login(self.smtp_username, self.smtp_password)
return server
except smtplib.SMTPException as e:
self.logger.error(f"Failed to create SMTP connection: {e}")
raise
def _create_message(
self,
to_email: str,
subject: str,
body: str,
attachments: Optional[List[str]] = None
) -> MIMEMultipart:
"""
Create email message with optional attachments.
Args:
to_email: Recipient email address
subject: Email subject
body: Email body (HTML)
attachments: List of file paths to attach
Returns:
Configured email message
"""
msg = MIMEMultipart('alternative')
msg['From'] = f"{self.from_name} <{self.from_email}>"
msg['To'] = to_email
msg['Subject'] = subject
# Add HTML body
html_part = MIMEText(body, 'html')
msg.attach(html_part)
# Add attachments if provided
if attachments:
for attachment_path in attachments:
if os.path.exists(attachment_path):
self._add_attachment(msg, attachment_path)
else:
self.logger.warning(f"Attachment file not found: {attachment_path}")
return msg
def _add_attachment(self, msg: MIMEMultipart, file_path: str) -> None:
"""
Add file attachment to email message.
Args:
msg: Email message to attach to
file_path: Path to file to attach
"""
try:
filename = os.path.basename(file_path)
# Read file
with open(file_path, 'rb') as f:
file_data = f.read()
# Create attachment
attachment = MIMEApplication(file_data, Name=filename)
attachment['Content-Disposition'] = f'attachment; filename="{filename}"'
msg.attach(attachment)
self.logger.debug(f"Added attachment: {filename}")
except Exception as e:
self.logger.error(f"Failed to add attachment {file_path}: {e}")
def send_email(
self,
to_email: str,
subject: str,
body: str,
attachments: Optional[List[str]] = None
) -> bool:
"""
Send email with optional attachments.
Args:
to_email: Recipient email address
subject: Email subject
body: Email body (HTML)
attachments: List of file paths to attach
Returns:
True if email sent successfully, False otherwise
"""
try:
# Create message
msg = self._create_message(to_email, subject, body, attachments)
# Send email
server = self._create_smtp_connection()
server.sendmail(self.from_email, to_email, msg.as_string())
server.quit()
self.logger.info(f"Email sent successfully to {to_email}")
return True
except Exception as e:
self.logger.error(f"Failed to send email to {to_email}: {e}")
return False
def send_credential_notification(
to_email: str,
username: str,
temp_password: str,
vpn_config_url: Optional[str] = None
) -> bool:
"""
Send credential notification email to new user.
Args:
to_email: Recipient email address
username: VPN username
temp_password: Temporary password
vpn_config_url: Optional URL to download VPN config
Returns:
True if email sent successfully, False otherwise
"""
subject = "Your VPN Access Credentials"
config_link = ""
if vpn_config_url:
config_link = f'Download VPN Configuration
'
body = f"""
Welcome to VPN Access Server
Your VPN account has been created successfully.
Your Credentials:
Username: {username}
Temporary Password: {temp_password}
{config_link}
Important: Please change your password after first login.
For support, contact your system administrator.
This is an automated message from VPN Access Server.
"""
email_service = EmailService()
return email_service.send_email(to_email, subject, body)
def send_user_guide(
to_email: str,
username: str,
guide_attachments: List[str]
) -> bool:
"""
Send user guide with attachments.
Args:
to_email: Recipient email address
username: VPN username
guide_attachments: List of guide file paths to attach
Returns:
True if email sent successfully, False otherwise
"""
subject = "VPN Access User Guide and Materials"
body = f"""
VPN Access User Guide
Hello {username},
Please find attached the user guide and materials for VPN access.
Attachments:
"""
# Add attachment list to body
for attachment in guide_attachments:
filename = os.path.basename(attachment)
body += f"- {filename}
"
body += """
If you have any questions, please contact your system administrator.
This is an automated message from VPN Access Server.
"""
email_service = EmailService()
return email_service.send_email(to_email, subject, body, guide_attachments)
def send_log_files(
to_email: str,
username: str,
log_files: List[str],
date_range: Optional[str] = None
) -> bool:
"""
Send log files from NAS.
Args:
to_email: Recipient email address
username: VPN username
log_files: List of log file paths to attach
date_range: Optional date range description
Returns:
True if email sent successfully, False otherwise
"""
subject = "VPN Access Server Log Files"
date_info = ""
if date_range:
date_info = f" for {date_range}"
body = f"""
VPN Access Server Log Files
Hello {username},
Please find attached the VPN access log files{date_info}.
Attached Log Files:
"""
# Add log file list to body
for log_file in log_files:
filename = os.path.basename(log_file)
body += f"- {filename}
"
body += """
These logs contain connection and authentication information.
For security reasons, please handle these files appropriately.
This is an automated message from VPN Access Server.
"""
email_service = EmailService()
return email_service.send_email(to_email, subject, body, log_files)
def send_vpn_config(
to_email: str,
username: str,
vpn_config_path: str,
user_guide_path: Optional[str] = None
) -> bool:
"""
Send VPN client configuration and user guide via email.
Args:
to_email: Recipient email address
username: VPN username
vpn_config_path: Path to the .ovpn configuration file
user_guide_path: Optional path to user guide file
Returns:
True if email sent successfully, False otherwise
"""
subject = "Your VPN Client Configuration and User Guide"
attachments = [vpn_config_path]
if user_guide_path and os.path.exists(user_guide_path):
attachments.append(user_guide_path)
body = f"""
VPN Client Configuration
Hello {username},
Your VPN client configuration has been successfully generated.
Attachments:
"""
# Add attachment list to body
for attachment in attachments:
filename = os.path.basename(attachment)
body += f"- {filename}
"
body += """
Instructions:
- Download and install an OpenVPN client for your operating system
- Import the .ovpn configuration file
- Connect to the VPN using your credentials
Please review the user guide for detailed setup instructions.
If you encounter any issues, please contact your system administrator.
This is an automated message from VPN Access Server.
"""
email_service = EmailService()
return email_service.send_email(to_email, subject, body, attachments)
# Convenience functions for direct usage
def send_credentials(to_email: str, username: str, temp_password: str, vpn_config_url: Optional[str] = None) -> bool:
"""Send credential notification (alias for send_credential_notification)."""
return send_credential_notification(to_email, username, temp_password, vpn_config_url)
def send_guide(to_email: str, username: str, guide_attachments: List[str]) -> bool:
"""Send user guide (alias for send_user_guide)."""
return send_user_guide(to_email, username, guide_attachments)
def send_logs(to_email: str, username: str, log_files: List[str], date_range: Optional[str] = None) -> bool:
"""Send log files (alias for send_log_files)."""
return send_log_files(to_email, username, log_files, date_range)