add auto send email function
This commit is contained in:
parent
80da8c2b4e
commit
48520621e5
11
.env.example
11
.env.example
@ -20,4 +20,13 @@ MAX_SESSION_LIMIT=86400
|
||||
TIMEZONE=UTC
|
||||
|
||||
# Session cleanup (for maintenance scripts)
|
||||
SESSION_CLEANUP_DAYS=90
|
||||
SESSION_CLEANUP_DAYS=90
|
||||
|
||||
# Email Configuration
|
||||
SMTP_HOST=outbound.daouoffice.com
|
||||
SMTP_PORT=465
|
||||
SMTP_USERNAME=dev@dnkinno.com
|
||||
SMTP_PASSWORD=Dki@2025
|
||||
SMTP_USE_TLS=False
|
||||
SMTP_FROM_EMAIL=noreply@dnkinno.com
|
||||
SMTP_FROM_NAME=VPN
|
||||
61
.serena/memories/code_style_conventions.md
Normal file
61
.serena/memories/code_style_conventions.md
Normal file
@ -0,0 +1,61 @@
|
||||
# VPN Access Server - Code Style and Conventions
|
||||
|
||||
## Python Version and Requirements
|
||||
- **Python version**: 3.13+ (minimum requirement)
|
||||
- **Type hints**: Required for all function parameters and return values
|
||||
- **Docstrings**: Mandatory for all functions, classes, and modules
|
||||
- **Encoding**: UTF-8 for all files
|
||||
|
||||
## Naming Conventions
|
||||
- **Functions and variables**: snake_case (e.g., `validate_mac_address`, `user_id`)
|
||||
- **Classes**: PascalCase (e.g., `ClientRequest`, `DatabaseConnection`)
|
||||
- **Constants**: UPPER_CASE (e.g., `DEFAULT_SESSION_LIMIT`)
|
||||
- **Modules**: snake_case (e.g., `auth.py`, `session.py`)
|
||||
|
||||
## Import Organization
|
||||
```python
|
||||
# Standard library imports
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
# Third-party imports
|
||||
import mysql.connector
|
||||
from fastapi import FastAPI
|
||||
|
||||
# Local imports
|
||||
from config import config
|
||||
from utils import setup_logging
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
- Use try/except blocks for all database operations and external API calls
|
||||
- Custom exit codes: 0=success, 1=authentication failure, 2=configuration error
|
||||
- Log all errors with appropriate log levels
|
||||
- Use `safe_exit()` utility for graceful error termination
|
||||
|
||||
## Logging
|
||||
- Use `setup_logging()` from `access.utils` for consistent logging
|
||||
- Structured log format with timestamps, logger name, level, and message
|
||||
- Log levels: DEBUG, INFO, WARNING, ERROR, CRITICAL
|
||||
- File and console handlers supported
|
||||
|
||||
## Security Practices
|
||||
- **SQL**: Always use parameterized queries to prevent injection
|
||||
- **Credentials**: Store in environment variables, never in code
|
||||
- **Passwords**: Hash using secure algorithms (bcrypt/scrypt)
|
||||
- **Configuration**: Validate all configuration before use
|
||||
- **Access**: Principle of least privilege for database accounts
|
||||
|
||||
## Code Structure
|
||||
- **Functions**: Small, focused, single responsibility
|
||||
- **Classes**: When needed for related functionality
|
||||
- **Modules**: Logical grouping of related functions
|
||||
- **Error messages**: Clear, actionable, and secure (no sensitive data)
|
||||
|
||||
## Database Design
|
||||
- Use foreign keys and constraints
|
||||
- Index frequently queried columns
|
||||
- Use appropriate data types
|
||||
- Handle connection pooling for performance
|
||||
34
.serena/memories/post_task_actions.md
Normal file
34
.serena/memories/post_task_actions.md
Normal file
@ -0,0 +1,34 @@
|
||||
# VPN Access Server - Post-Task Completion Actions
|
||||
|
||||
## After Code Changes
|
||||
1. **Run tests**: `uv run pytest tests/` - Ensure all tests pass
|
||||
2. **Health check**: `uv run vpn-access-server health-check` - Verify system integrity
|
||||
3. **Type checking**: Consider running mypy if configured (not currently set up)
|
||||
4. **Code review**: Check for adherence to style guidelines
|
||||
|
||||
## After Adding Dependencies
|
||||
1. **Update lockfile**: `uv lock` - Ensure reproducible builds
|
||||
2. **Test installation**: `uv install` - Verify dependencies install correctly
|
||||
3. **Update pyproject.toml**: Add version constraints if needed
|
||||
|
||||
## After Database Schema Changes
|
||||
1. **Update scripts**: Modify `scripts/init_db.py` for new schema
|
||||
2. **Test migration**: Run `uv run vpn-access-server init-db` on test environment
|
||||
3. **Update seed data**: Modify `scripts/seed_data.py` if needed
|
||||
4. **Verify queries**: Test all database operations in affected modules
|
||||
|
||||
## After Adding New Modules (e.g., Email Module)
|
||||
1. **Update configuration**: Add new config classes and validation in `access/config.py`
|
||||
2. **Update .env.example**: Add required environment variables with defaults
|
||||
3. **Update imports**: Ensure proper import paths in dependent modules
|
||||
4. **Add tests**: Create corresponding test files in `tests/` directory following existing patterns
|
||||
5. **Install test dependencies**: Run `uv sync --group test` if needed
|
||||
6. **Update CLI**: Add new commands to `cli.py` if needed
|
||||
7. **Update documentation**: Update README.md and any relevant docs
|
||||
|
||||
## General Best Practices
|
||||
- **Commit frequently**: Small, focused commits with clear messages
|
||||
- **Test thoroughly**: Unit tests for new functions, integration tests for workflows
|
||||
- **Security review**: Check for credential leaks, SQL injection risks, etc.
|
||||
- **Performance check**: Monitor database queries and memory usage
|
||||
- **Documentation**: Update docstrings and comments for complex logic
|
||||
28
.serena/memories/project_overview.md
Normal file
28
.serena/memories/project_overview.md
Normal file
@ -0,0 +1,28 @@
|
||||
# VPN Access Server Project Overview
|
||||
|
||||
## Purpose
|
||||
This is a VPN Access Server project that extends OpenVPN functionality with custom MAC address validation and session time management. The system integrates with MySQL for data persistence and is designed for deployment on Ubuntu VPN gateways.
|
||||
|
||||
## Tech Stack
|
||||
- **Python 3.13+** with type hints and docstrings
|
||||
- **MySQL** database for data persistence
|
||||
- **FastAPI** for API endpoints and client configuration generation
|
||||
- **OpenVPN** integration via client-connect/client-disconnect hooks
|
||||
- **uv** package manager for dependency management
|
||||
- **pytest** for testing
|
||||
|
||||
## Architecture
|
||||
The system follows a modular layered architecture:
|
||||
- **access/**: Core authentication and session management logic
|
||||
- **util/**: Utility modules (client generation, email sending)
|
||||
- **tests/**: Unit and integration tests
|
||||
- **scripts/**: Database setup and helper scripts
|
||||
- **main.py**: FastAPI server entry point
|
||||
- **cli.py**: Command-line interface for all operations
|
||||
|
||||
## Key Features
|
||||
- User authentication and MAC address validation
|
||||
- Session time management and limits
|
||||
- Client configuration generation (.ovpn files)
|
||||
- Database-backed user and session management
|
||||
- SMTP email notifications (planned)
|
||||
40
.serena/memories/suggested_commands.md
Normal file
40
.serena/memories/suggested_commands.md
Normal file
@ -0,0 +1,40 @@
|
||||
# VPN Access Server - Suggested Commands
|
||||
|
||||
## Package Management
|
||||
- **Install dependencies**: `uv install`
|
||||
- **Add dependency**: `uv add package-name`
|
||||
- **Add test dependency**: `uv add --group test package-name`
|
||||
- **Update lockfile**: `uv lock`
|
||||
|
||||
## Running the Application
|
||||
- **Show help**: `uv run vpn-access-server --help`
|
||||
- **Authentication**: `uv run vpn-access-server auth`
|
||||
- **Session management**: `uv run vpn-access-server session`
|
||||
- **Health check**: `uv run vpn-access-server health-check`
|
||||
- **Show status**: `uv run vpn-access-server status`
|
||||
- **Initialize database**: `uv run vpn-access-server init-db`
|
||||
- **Seed test data**: `uv run vpn-access-server seed-data`
|
||||
- **Generate client**: `uv run vpn-access-server gen-client <username>`
|
||||
|
||||
## Testing
|
||||
- **Run all tests**: `uv run pytest tests/`
|
||||
- **Run specific test**: `uv run pytest tests/test_specific.py::TestClass::test_method`
|
||||
- **Run with coverage**: `uv run pytest --cov=access tests/`
|
||||
- **Run specific module tests**: `uv run pytest tests/test_utils.py`
|
||||
|
||||
## Development
|
||||
- **Direct Python execution**: `uv run python main.py`
|
||||
- **Direct module execution**: `uv run python -m access.auth`
|
||||
- **Format code**: Use black if configured (not currently in project)
|
||||
- **Lint code**: Use flake8 if configured (not currently in project)
|
||||
|
||||
## Database Operations
|
||||
- **Setup database**: `uv run vpn-access-server init-db && uv run vpn-access-server seed-data`
|
||||
- **Check database health**: `uv run vpn-access-server health-check`
|
||||
|
||||
## System Commands (Darwin/macOS)
|
||||
- **List files**: `ls -la`
|
||||
- **Change directory**: `cd path`
|
||||
- **Find files**: `find . -name "*.py"`
|
||||
- **Search text**: `grep -r "pattern" .`
|
||||
- **Git operations**: `git status`, `git add .`, `git commit -m "message"`
|
||||
23
AGENTS.md
Normal file
23
AGENTS.md
Normal file
@ -0,0 +1,23 @@
|
||||
# VPN Access Server - Agent Guidelines
|
||||
|
||||
## Commands
|
||||
- **Install deps**: `uv install`
|
||||
- **Run CLI**: `uv run vpn-access-server [command]` (auth, session, init-db, seed-data, test, health-check, status, gen-client)
|
||||
- **Run all tests**: `uv run pytest tests/`
|
||||
- **Run single test**: `uv run pytest tests/test_specific.py::TestClass::test_method`
|
||||
- **Database setup**: `uv run vpn-access-server init-db && uv run vpn-access-server seed-data`
|
||||
|
||||
## Architecture
|
||||
- **Python 3.13+** VPN server extending OpenVPN with MAC validation and session management
|
||||
- **MySQL database** for users, MAC addresses, and session tracking
|
||||
- **FastAPI** for client config generation and management
|
||||
- **Modular structure**: `access/` (core auth/session logic), `tests/` (unit tests), `scripts/` (DB setup), `util/` (client tools)
|
||||
|
||||
## Code Style
|
||||
- **Type hints** required for all function parameters and return values
|
||||
- **Docstrings** mandatory for all functions, classes, and modules
|
||||
- **Snake_case** for functions/variables, PascalCase for classes
|
||||
- **Logging** via `setup_logging()` with structured format
|
||||
- **Error handling** with try/except, custom exit codes (0=success, 1=auth fail, 2=config error)
|
||||
- **Imports** grouped: stdlib, third-party, local modules
|
||||
- **Security** paramount: parameterized SQL, secure logging, environment variables for secrets
|
||||
@ -34,12 +34,37 @@ class ServerConfig:
|
||||
timezone: str = 'UTC'
|
||||
|
||||
|
||||
@dataclass
|
||||
class EmailConfig:
|
||||
"""Email service configuration."""
|
||||
smtp_host: str
|
||||
smtp_port: int
|
||||
smtp_username: str
|
||||
smtp_password: str
|
||||
smtp_use_tls: bool
|
||||
from_email: str
|
||||
from_name: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class EmailConfig:
|
||||
"""Email service configuration."""
|
||||
smtp_host: str
|
||||
smtp_port: int
|
||||
smtp_username: str
|
||||
smtp_password: str
|
||||
smtp_use_tls: bool
|
||||
from_email: str
|
||||
from_name: str
|
||||
|
||||
|
||||
class Config:
|
||||
"""Main configuration loader for VPN Access Server."""
|
||||
|
||||
def __init__(self):
|
||||
self.database = self._load_database_config()
|
||||
self.server = self._load_server_config()
|
||||
self.email = self._load_email_config()
|
||||
|
||||
def _load_database_config(self) -> DatabaseConfig:
|
||||
"""Load database configuration from environment variables."""
|
||||
@ -65,6 +90,18 @@ class Config:
|
||||
timezone=os.getenv('TIMEZONE', 'UTC')
|
||||
)
|
||||
|
||||
def _load_email_config(self) -> EmailConfig:
|
||||
"""Load email configuration from environment variables."""
|
||||
return EmailConfig(
|
||||
smtp_host=os.getenv('SMTP_HOST', 'smtp.gmail.com'),
|
||||
smtp_port=int(os.getenv('SMTP_PORT', '587')),
|
||||
smtp_username=os.getenv('SMTP_USERNAME', ''),
|
||||
smtp_password=os.getenv('SMTP_PASSWORD', ''),
|
||||
smtp_use_tls=os.getenv('SMTP_USE_TLS', 'true').lower() == 'true',
|
||||
from_email=os.getenv('SMTP_FROM_EMAIL', 'noreply@yourcompany.com'),
|
||||
from_name=os.getenv('SMTP_FROM_NAME', 'VPN Access Server')
|
||||
)
|
||||
|
||||
def validate(self) -> bool:
|
||||
"""Validate configuration values."""
|
||||
errors = []
|
||||
@ -83,6 +120,16 @@ class Config:
|
||||
if self.server.default_session_limit > self.server.max_session_limit:
|
||||
errors.append("DEFAULT_SESSION_LIMIT cannot exceed MAX_SESSION_LIMIT")
|
||||
|
||||
# Validate email config
|
||||
if not self.email.smtp_username:
|
||||
errors.append("SMTP_USERNAME is required for email functionality")
|
||||
|
||||
if not self.email.smtp_password:
|
||||
errors.append("SMTP_PASSWORD is required for email functionality")
|
||||
|
||||
if self.email.smtp_port < 1 or self.email.smtp_port > 65535:
|
||||
errors.append("SMTP_PORT must be between 1 and 65535")
|
||||
|
||||
if errors:
|
||||
for error in errors:
|
||||
print(f"Configuration Error: {error}")
|
||||
|
||||
29
main.py
29
main.py
@ -9,6 +9,7 @@ from fastapi.responses import FileResponse
|
||||
from pydantic import BaseModel
|
||||
|
||||
from util.client.generate_client import generate_client_config
|
||||
from util.email import send_vpn_config
|
||||
|
||||
app = FastAPI(
|
||||
title="VPN Access Server API",
|
||||
@ -23,17 +24,41 @@ class ClientRequest(BaseModel):
|
||||
@app.post("/api/client/generate", summary="Generate a new VPN client configuration")
|
||||
def generate_client(request: ClientRequest):
|
||||
"""
|
||||
Generates a new OpenVPN client configuration (.ovpn) file.
|
||||
Generates a new OpenVPN client configuration (.ovpn) file and sends it via email.
|
||||
|
||||
This endpoint will:
|
||||
1. Trigger `easyrsa` to generate a new client certificate and key.
|
||||
2. Assemble the `.ovpn` file with the new certificate/key and the server's CA and TA keys.
|
||||
3. Save the file to the server's client configuration directory.
|
||||
4. Send the .ovpn file and user guide via email to the provided email address.
|
||||
"""
|
||||
success, message = generate_client_config(request.username, request.email)
|
||||
if not success:
|
||||
raise HTTPException(status_code=500, detail=message)
|
||||
return {"message": message}
|
||||
|
||||
# Send email with VPN config and user guide
|
||||
vpn_config_path = os.path.join("generated-clients", f"{request.username}.ovpn")
|
||||
user_guide_path = "material/sample-ppt.pptx"
|
||||
|
||||
email_success = send_vpn_config(
|
||||
to_email=request.email,
|
||||
username=request.username,
|
||||
vpn_config_path=vpn_config_path,
|
||||
user_guide_path=user_guide_path
|
||||
)
|
||||
|
||||
if not email_success:
|
||||
# Configuration was generated successfully, but email failed
|
||||
# Return success with warning about email
|
||||
return {
|
||||
"message": f"{message} Warning: Failed to send email with configuration files.",
|
||||
"email_sent": False
|
||||
}
|
||||
|
||||
return {
|
||||
"message": f"{message} Configuration files sent to {request.email}",
|
||||
"email_sent": True
|
||||
}
|
||||
|
||||
@app.get("/api/client/get-config/{username}", summary="Download a client configuration file")
|
||||
def get_client_config(username: str, email: str):
|
||||
|
||||
0
material/sample-ppt.pptx
Normal file
0
material/sample-ppt.pptx
Normal file
@ -82,7 +82,11 @@ class TestConfiguration(unittest.TestCase):
|
||||
self.assertEqual(config.server.default_session_limit, 14400)
|
||||
self.assertEqual(config.server.max_session_limit, 43200)
|
||||
|
||||
@patch.dict(os.environ, {'DB_PASSWORD': 'valid_password'})
|
||||
@patch.dict(os.environ, {
|
||||
'DB_PASSWORD': 'valid_password',
|
||||
'SMTP_USERNAME': 'test@example.com',
|
||||
'SMTP_PASSWORD': 'test_password'
|
||||
})
|
||||
def test_valid_configuration(self):
|
||||
"""Test configuration validation with valid values."""
|
||||
from config import Config
|
||||
|
||||
237
tests/test_email.py
Normal file
237
tests/test_email.py
Normal file
@ -0,0 +1,237 @@
|
||||
"""
|
||||
Unit tests for email service functionality.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import patch, MagicMock, mock_open
|
||||
|
||||
# Add the util module to the Python path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'util'))
|
||||
|
||||
from util.email.email_service import (
|
||||
EmailService,
|
||||
send_credential_notification,
|
||||
send_user_guide,
|
||||
send_log_files,
|
||||
send_vpn_config
|
||||
)
|
||||
|
||||
|
||||
class TestEmailService(unittest.TestCase):
|
||||
"""Test email service functionality."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
self.email_service = EmailService()
|
||||
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_email_success(self, mock_smtp_class):
|
||||
"""Test successful email sending."""
|
||||
# Mock SMTP connection
|
||||
mock_server = MagicMock()
|
||||
mock_smtp_class.return_value = mock_server
|
||||
|
||||
# Test sending email
|
||||
result = self.email_service.send_email(
|
||||
to_email='test@example.com',
|
||||
subject='Test Subject',
|
||||
body='<p>Test body</p>'
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_smtp_class.assert_called_once_with('smtp.gmail.com', 587)
|
||||
mock_server.starttls.assert_called_once()
|
||||
mock_server.login.assert_called_once()
|
||||
mock_server.sendmail.assert_called_once()
|
||||
mock_server.quit.assert_called_once()
|
||||
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_email_with_attachments(self, mock_smtp_class):
|
||||
"""Test email sending with file attachments."""
|
||||
# Mock SMTP connection
|
||||
mock_server = MagicMock()
|
||||
mock_smtp_class.return_value = mock_server
|
||||
|
||||
# Create a temporary test file
|
||||
test_file = '/tmp/test_attachment.txt'
|
||||
with open(test_file, 'w') as f:
|
||||
f.write('test content')
|
||||
|
||||
try:
|
||||
# Test sending email with attachment
|
||||
result = self.email_service.send_email(
|
||||
to_email='test@example.com',
|
||||
subject='Test Subject',
|
||||
body='<p>Test body</p>',
|
||||
attachments=[test_file]
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_server.sendmail.assert_called_once()
|
||||
finally:
|
||||
# Clean up
|
||||
if os.path.exists(test_file):
|
||||
os.remove(test_file)
|
||||
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_email_failure(self, mock_smtp_class):
|
||||
"""Test email sending failure."""
|
||||
# Mock SMTP to raise exception
|
||||
mock_smtp_class.side_effect = Exception("SMTP error")
|
||||
|
||||
# Test sending email
|
||||
result = self.email_service.send_email(
|
||||
to_email='test@example.com',
|
||||
subject='Test Subject',
|
||||
body='<p>Test body</p>'
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
@patch('tests.test_email.EmailService.send_email')
|
||||
def test_send_credential_notification(self, mock_send_email):
|
||||
"""Test sending credential notification."""
|
||||
mock_send_email.return_value = True
|
||||
|
||||
result = send_credential_notification(
|
||||
to_email='user@example.com',
|
||||
username='testuser',
|
||||
temp_password='temppass123',
|
||||
vpn_config_url='http://example.com/config'
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_send_email.assert_called_once()
|
||||
args, kwargs = mock_send_email.call_args
|
||||
self.assertEqual(args[0], 'user@example.com') # to_email
|
||||
self.assertEqual(args[1], 'Your VPN Access Credentials') # subject
|
||||
self.assertIn('testuser', args[2]) # body
|
||||
self.assertIn('temppass123', args[2]) # body
|
||||
self.assertIn('http://example.com/config', args[2]) # body
|
||||
|
||||
@patch('tests.test_email.EmailService.send_email')
|
||||
def test_send_user_guide(self, mock_send_email):
|
||||
"""Test sending user guide with attachments."""
|
||||
mock_send_email.return_value = True
|
||||
|
||||
# Create a temporary test file
|
||||
test_file = '/tmp/test_guide.pdf'
|
||||
with open(test_file, 'w') as f:
|
||||
f.write('test guide content')
|
||||
|
||||
try:
|
||||
result = send_user_guide(
|
||||
to_email='user@example.com',
|
||||
username='testuser',
|
||||
guide_attachments=[test_file]
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_send_email.assert_called_once()
|
||||
args, kwargs = mock_send_email.call_args
|
||||
self.assertEqual(args[0], 'user@example.com') # to_email
|
||||
self.assertEqual(args[1], 'VPN Access User Guide and Materials') # subject
|
||||
self.assertEqual(args[3], [test_file]) # attachments
|
||||
finally:
|
||||
# Clean up
|
||||
if os.path.exists(test_file):
|
||||
os.remove(test_file)
|
||||
|
||||
@patch('tests.test_email.EmailService.send_email')
|
||||
def test_send_log_files(self, mock_send_email):
|
||||
"""Test sending log files."""
|
||||
mock_send_email.return_value = True
|
||||
|
||||
# Create a temporary test log file
|
||||
test_log = '/tmp/test_log.txt'
|
||||
with open(test_log, 'w') as f:
|
||||
f.write('test log content')
|
||||
|
||||
try:
|
||||
result = send_log_files(
|
||||
to_email='admin@example.com',
|
||||
username='admin',
|
||||
log_files=[test_log],
|
||||
date_range='2024-01-01 to 2024-01-31'
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_send_email.assert_called_once()
|
||||
args, kwargs = mock_send_email.call_args
|
||||
self.assertEqual(args[0], 'admin@example.com') # to_email
|
||||
self.assertEqual(args[1], 'VPN Access Server Log Files') # subject
|
||||
self.assertEqual(args[3], [test_log]) # attachments
|
||||
self.assertIn('2024-01-01 to 2024-01-31', args[2]) # body
|
||||
finally:
|
||||
# Clean up
|
||||
if os.path.exists(test_log):
|
||||
os.remove(test_log)
|
||||
|
||||
@patch('tests.test_email.EmailService.send_email')
|
||||
def test_send_vpn_config(self, mock_send_email):
|
||||
"""Test sending VPN configuration and user guide."""
|
||||
mock_send_email.return_value = True
|
||||
|
||||
# Create temporary files
|
||||
vpn_config = '/tmp/test_config.ovpn'
|
||||
user_guide = '/tmp/test_guide.pptx'
|
||||
|
||||
with open(vpn_config, 'w') as f:
|
||||
f.write('client config content')
|
||||
with open(user_guide, 'w') as f:
|
||||
f.write('guide content')
|
||||
|
||||
try:
|
||||
result = send_vpn_config(
|
||||
to_email='user@example.com',
|
||||
username='testuser',
|
||||
vpn_config_path=vpn_config,
|
||||
user_guide_path=user_guide
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_send_email.assert_called_once()
|
||||
args, kwargs = mock_send_email.call_args
|
||||
self.assertEqual(args[0], 'user@example.com') # to_email
|
||||
self.assertEqual(args[1], 'Your VPN Client Configuration and User Guide') # subject
|
||||
self.assertEqual(args[3], [vpn_config, user_guide]) # attachments
|
||||
self.assertIn('testuser', args[2]) # body contains username
|
||||
finally:
|
||||
# Clean up
|
||||
for file_path in [vpn_config, user_guide]:
|
||||
if os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
|
||||
@patch('tests.test_email.EmailService.send_email')
|
||||
def test_send_vpn_config_without_user_guide(self, mock_send_email):
|
||||
"""Test sending VPN configuration without user guide."""
|
||||
mock_send_email.return_value = True
|
||||
|
||||
# Create temporary VPN config file
|
||||
vpn_config = '/tmp/test_config.ovpn'
|
||||
|
||||
with open(vpn_config, 'w') as f:
|
||||
f.write('client config content')
|
||||
|
||||
try:
|
||||
result = send_vpn_config(
|
||||
to_email='user@example.com',
|
||||
username='testuser',
|
||||
vpn_config_path=vpn_config,
|
||||
user_guide_path=None # No user guide
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_send_email.assert_called_once()
|
||||
args, kwargs = mock_send_email.call_args
|
||||
self.assertEqual(args[3], [vpn_config]) # Only VPN config attached
|
||||
finally:
|
||||
# Clean up
|
||||
if os.path.exists(vpn_config):
|
||||
os.remove(vpn_config)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
10
util/email/__init__.py
Normal file
10
util/email/__init__.py
Normal file
@ -0,0 +1,10 @@
|
||||
"""
|
||||
Email service module for VPN Access Server.
|
||||
|
||||
Provides functionality for sending various types of email notifications
|
||||
including credential notifications, user guides with attachments, and log files.
|
||||
"""
|
||||
|
||||
from .email_service import EmailService, send_credential_notification, send_user_guide, send_log_files, send_vpn_config
|
||||
|
||||
__all__ = ['EmailService', 'send_credential_notification', 'send_user_guide', 'send_log_files', 'send_vpn_config']
|
||||
419
util/email/email_service.py
Normal file
419
util/email/email_service.py
Normal file
@ -0,0 +1,419 @@
|
||||
#!/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'<p><a href="{vpn_config_url}">Download VPN Configuration</a></p>'
|
||||
|
||||
body = f"""
|
||||
<html>
|
||||
<body>
|
||||
<h2>Welcome to VPN Access Server</h2>
|
||||
<p>Your VPN account has been created successfully.</p>
|
||||
|
||||
<div style="background-color: #f0f0f0; padding: 15px; border-radius: 5px; margin: 20px 0;">
|
||||
<h3>Your Credentials:</h3>
|
||||
<p><strong>Username:</strong> {username}</p>
|
||||
<p><strong>Temporary Password:</strong> {temp_password}</p>
|
||||
</div>
|
||||
|
||||
{config_link}
|
||||
|
||||
<p><strong>Important:</strong> Please change your password after first login.</p>
|
||||
<p>For support, contact your system administrator.</p>
|
||||
|
||||
<hr>
|
||||
<p style="color: #666; font-size: 12px;">
|
||||
This is an automated message from VPN Access Server.
|
||||
</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
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"""
|
||||
<html>
|
||||
<body>
|
||||
<h2>VPN Access User Guide</h2>
|
||||
<p>Hello {username},</p>
|
||||
|
||||
<p>Please find attached the user guide and materials for VPN access.</p>
|
||||
|
||||
<div style="background-color: #e8f4fd; padding: 15px; border-radius: 5px; margin: 20px 0;">
|
||||
<h3>Attachments:</h3>
|
||||
<ul>
|
||||
"""
|
||||
|
||||
# Add attachment list to body
|
||||
for attachment in guide_attachments:
|
||||
filename = os.path.basename(attachment)
|
||||
body += f"<li>{filename}</li>"
|
||||
|
||||
body += """
|
||||
</ul>
|
||||
</div>
|
||||
|
||||
<p>If you have any questions, please contact your system administrator.</p>
|
||||
|
||||
<hr>
|
||||
<p style="color: #666; font-size: 12px;">
|
||||
This is an automated message from VPN Access Server.
|
||||
</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
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"""
|
||||
<html>
|
||||
<body>
|
||||
<h2>VPN Access Server Log Files</h2>
|
||||
<p>Hello {username},</p>
|
||||
|
||||
<p>Please find attached the VPN access log files{date_info}.</p>
|
||||
|
||||
<div style="background-color: #fff3cd; padding: 15px; border-radius: 5px; margin: 20px 0;">
|
||||
<h3>Attached Log Files:</h3>
|
||||
<ul>
|
||||
"""
|
||||
|
||||
# Add log file list to body
|
||||
for log_file in log_files:
|
||||
filename = os.path.basename(log_file)
|
||||
body += f"<li>{filename}</li>"
|
||||
|
||||
body += """
|
||||
</ul>
|
||||
</div>
|
||||
|
||||
<p>These logs contain connection and authentication information.</p>
|
||||
<p>For security reasons, please handle these files appropriately.</p>
|
||||
|
||||
<hr>
|
||||
<p style="color: #666; font-size: 12px;">
|
||||
This is an automated message from VPN Access Server.
|
||||
</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
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"""
|
||||
<html>
|
||||
<body>
|
||||
<h2>VPN Client Configuration</h2>
|
||||
<p>Hello {username},</p>
|
||||
|
||||
<p>Your VPN client configuration has been successfully generated.</p>
|
||||
|
||||
<div style="background-color: #e8f4fd; padding: 15px; border-radius: 5px; margin: 20px 0;">
|
||||
<h3>Attachments:</h3>
|
||||
<ul>
|
||||
"""
|
||||
|
||||
# Add attachment list to body
|
||||
for attachment in attachments:
|
||||
filename = os.path.basename(attachment)
|
||||
body += f"<li>{filename}</li>"
|
||||
|
||||
body += """
|
||||
</ul>
|
||||
</div>
|
||||
|
||||
<p><strong>Instructions:</strong></p>
|
||||
<ol>
|
||||
<li>Download and install an OpenVPN client for your operating system</li>
|
||||
<li>Import the .ovpn configuration file</li>
|
||||
<li>Connect to the VPN using your credentials</li>
|
||||
</ol>
|
||||
|
||||
<p>Please review the user guide for detailed setup instructions.</p>
|
||||
|
||||
<p>If you encounter any issues, please contact your system administrator.</p>
|
||||
|
||||
<hr>
|
||||
<p style="color: #666; font-size: 12px;">
|
||||
This is an automated message from VPN Access Server.
|
||||
</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
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)
|
||||
Loading…
Reference in New Issue
Block a user