beppause/smtp-ethical-hack
0
1"""
2EmailSender class for handling SMTP connections and email sending with detailed logging.
3"""
4import smtplib
5import ssl
6from email.mime.multipart import MIMEMultipart
7from email.mime.text import MIMEText
8from email.mime.base import MIMEBase
9from email import encoders
10import os
11from typing import List, Optional, Callable
12from dataclasses import dataclass
13
14
15@dataclass
16class EmailConfig:
17 """Configuration for email sending."""
18 smtp_server: str
19 smtp_port: int
20 username: str
21 password: str
22 use_tls: bool = True
23 ssl_verify: bool = True
24
25
26class EmailSender:
27 """Handles SMTP connections and email sending with detailed logging."""
28
29 def __init__(self, log_callback: Optional[Callable[[str], None]] = None):
30 """
31 Initialize EmailSender with optional logging callback.
32
33 Args:
34 log_callback: Function to call with log messages for real-time updates
35 """
36 self.smtp_connection = None
37 self.log_callback = log_callback
38 self.log_buffer = []
39
40 def _log(self, message: str):
41 """Log a message and send to callback if provided."""
42 self.log_buffer.append(message)
43 if self.log_callback:
44 self.log_callback(message)
45
46 def connect(self, config: EmailConfig) -> bool:
47 """
48 Connect to SMTP server with the given configuration.
49
50 Args:
51 config: EmailConfig object with connection details
52
53 Returns:
54 bool: True if connection successful, False otherwise
55 """
56 try:
57 self._log(f"CONNECTING TO {config.smtp_server}:{config.smtp_port}")
58
59 # Create SMTP connection
60 self.smtp_connection = smtplib.SMTP(config.smtp_server, config.smtp_port)
61 self._log(f"CLIENT: Connected to {config.smtp_server}:{config.smtp_port}")
62
63 # Enable debug to capture SMTP dialogue
64 self.smtp_connection.set_debuglevel(1)
65
66 # Send EHLO and handle response
67 ehlo_response = self.smtp_connection.ehlo()
68 self._log(f"SERVER: {ehlo_response}")
69
70 if config.use_tls:
71 self._log("CLIENT: Starting TLS encryption")
72 # Create SSL context with optional verification
73 ssl_context = ssl.create_default_context()
74 if not config.ssl_verify:
75 ssl_context.check_hostname = False
76 ssl_context.verify_mode = ssl.CERT_NONE
77 self._log("WARNING: SSL certificate verification disabled")
78
79 tls_response = self.smtp_connection.starttls(context=ssl_context)
80 self._log(f"SERVER: {tls_response}")
81
82 # Send EHLO again after TLS
83 ehlo_response = self.smtp_connection.ehlo()
84 self._log(f"SERVER: {ehlo_response}")
85
86 # Authenticate if credentials provided
87 if config.username and config.password:
88 self._log("CLIENT: Attempting authentication")
89 auth_response = self.smtp_connection.login(config.username, config.password)
90 self._log(f"SERVER: Authentication successful")
91
92 self._log("CONNECTION ESTABLISHED SUCCESSFULLY")
93 return True
94
95 except Exception as e:
96 self._log(f"ERROR: Connection failed - {str(e)}")
97 return False
98
99 def send_email(
100 self,
101 from_addr: str,
102 to_addrs: List[str],
103 subject: str,
104 body: str,
105 attachments: Optional[List[str]] = None
106 ) -> bool:
107 """
108 Send an email with optional attachments.
109
110 Args:
111 from_addr: Sender email address
112 to_addrs: List of recipient email addresses
113 subject: Email subject
114 body: Email body text
115 attachments: List of file paths to attach
116
117 Returns:
118 bool: True if email sent successfully, False otherwise
119 """
120 if not self.smtp_connection:
121 self._log("ERROR: Not connected to SMTP server")
122 return False
123
124 try:
125 # Create message
126 msg = MIMEMultipart()
127 msg['From'] = from_addr
128 msg['To'] = ', '.join(to_addrs)
129 msg['Subject'] = subject
130
131 # Add body
132 msg.attach(MIMEText(body, 'plain'))
133
134 # Add attachments
135 if attachments:
136 for attachment_path in attachments:
137 if os.path.exists(attachment_path):
138 self._log(f"ATTACHING: {os.path.basename(attachment_path)}")
139 self._add_attachment(msg, attachment_path)
140 else:
141 self._log(f"WARNING: Attachment not found - {attachment_path}")
142
143 # Send email
144 self._log(f"CLIENT: Sending email from {from_addr} to {to_addrs}")
145 self.smtp_connection.sendmail(from_addr, to_addrs, msg.as_string())
146 self._log("EMAIL SENT SUCCESSFULLY")
147 return True
148
149 except Exception as e:
150 self._log(f"ERROR: Failed to send email - {str(e)}")
151 return False
152
153 def _add_attachment(self, msg: MIMEMultipart, file_path: str):
154 """Add a file attachment to the email message."""
155 with open(file_path, "rb") as attachment:
156 part = MIMEBase('application', 'octet-stream')
157 part.set_payload(attachment.read())
158
159 encoders.encode_base64(part)
160 part.add_header(
161 'Content-Disposition',
162 f'attachment; filename={os.path.basename(file_path)}',
163 )
164 msg.attach(part)
165
166 def disconnect(self):
167 """Disconnect from SMTP server."""
168 if self.smtp_connection:
169 try:
170 self._log("CLIENT: Quitting SMTP connection")
171 self.smtp_connection.quit()
172 self._log("DISCONNECTED SUCCESSFULLY")
173 except Exception as e:
174 self._log(f"ERROR during disconnect: {str(e)}")
175 finally:
176 self.smtp_connection = None
177
178 def get_log(self) -> str:
179 """Get the complete log as a string."""
180 return "\n".join(self.log_buffer)
181
182 def clear_log(self):
183 """Clear the log buffer."""
184 self.log_buffer.clear()
185