@smbcloud / smbcloud-mail-imap
Code Commits Issues Pulls Cloud Agent CI/CD
copilot/add-pop3-protocol-examples
copilot/add-pop3-protocol-examples copilot/add-pop3-protocol-implementation copilot/implement-mail-handling-separation development main
py 413 lines 12.6 KB
Raw
1 #!/usr/bin/env python3
2 """
3 POP3 Client Example in Python
4
5 This script demonstrates how to use the Python poplib library to interact
6 with a POP3 server. It shows all major POP3 operations including:
7 - Connecting to a POP3 server
8 - Authentication
9 - Listing messages
10 - Retrieving messages
11 - Deleting messages
12 - Error handling
13
14 Usage:
15 python3 pop3-client.py
16
17 Requirements:
18 Python 3.6+
19 """
20
21 import poplib
22 from email.parser import Parser
23 from email.policy import default
24 import sys
25
26
27 def example_basic_connection():
28 """Example 1: Basic connection and authentication"""
29 print("\n" + "="*50)
30 print("EXAMPLE 1: Basic Connection and Authentication")
31 print("="*50 + "\n")
32
33 try:
34 # Connect to POP3 server
35 print("Connecting to localhost:110...")
36 pop3_server = poplib.POP3("localhost", 110)
37
38 # Read the server greeting
39 greeting = pop3_server.getwelcome()
40 print(f"Server greeting: {greeting.decode('utf-8')}")
41
42 # Authenticate
43 print("\nAuthenticating...")
44 pop3_server.user("testuser")
45 pop3_server.pass_("testpass")
46 print("Authentication successful!")
47
48 # Close connection
49 pop3_server.quit()
50 print("Connection closed")
51
52 return True
53
54 except poplib.error_proto as e:
55 print(f"POP3 protocol error: {e}")
56 return False
57 except Exception as e:
58 print(f"Error: {e}")
59 return False
60
61
62 def example_list_messages():
63 """Example 2: List all messages in mailbox"""
64 print("\n" + "="*50)
65 print("EXAMPLE 2: List All Messages")
66 print("="*50 + "\n")
67
68 try:
69 pop3_server = poplib.POP3("localhost", 110)
70 pop3_server.user("testuser")
71 pop3_server.pass_("testpass")
72
73 # Get mailbox statistics
74 message_count, mailbox_size = pop3_server.stat()
75 print(f"Mailbox statistics:")
76 print(f" Messages: {message_count}")
77 print(f" Total size: {mailbox_size} bytes")
78
79 # List all messages
80 print(f"\nListing all messages:")
81 response, message_list, octets = pop3_server.list()
82 print(f"Response: {response.decode('utf-8')}")
83
84 for msg in message_list:
85 msg_info = msg.decode('utf-8')
86 parts = msg_info.split()
87 if len(parts) >= 2:
88 msg_num, msg_size = parts[0], parts[1]
89 print(f" Message {msg_num}: {msg_size} bytes")
90
91 pop3_server.quit()
92 return True
93
94 except Exception as e:
95 print(f"Error: {e}")
96 return False
97
98
99 def example_retrieve_message():
100 """Example 3: Retrieve and parse a message"""
101 print("\n" + "="*50)
102 print("EXAMPLE 3: Retrieve and Parse Message")
103 print("="*50 + "\n")
104
105 try:
106 pop3_server = poplib.POP3("localhost", 110)
107 pop3_server.user("testuser")
108 pop3_server.pass_("testpass")
109
110 message_count, _ = pop3_server.stat()
111
112 if message_count > 0:
113 print(f"Retrieving message 1...")
114
115 # Retrieve message (returns response, lines, octets)
116 response, lines, octets = pop3_server.retr(1)
117 print(f"Response: {response.decode('utf-8')}")
118 print(f"Size: {octets} bytes")
119
120 # Parse the message
121 msg_content = b'\n'.join(lines)
122 msg = Parser(policy=default).parsestr(msg_content.decode('utf-8'))
123
124 print(f"\n--- Message Details ---")
125 print(f"From: {msg.get('From', 'N/A')}")
126 print(f"To: {msg.get('To', 'N/A')}")
127 print(f"Subject: {msg.get('Subject', 'N/A')}")
128 print(f"Date: {msg.get('Date', 'N/A')}")
129
130 # Get message body
131 if msg.is_multipart():
132 print("\n--- Message Body (multipart) ---")
133 for part in msg.walk():
134 if part.get_content_type() == "text/plain":
135 body = part.get_payload(decode=True)
136 if body:
137 print(body.decode('utf-8', errors='ignore')[:500])
138 break
139 else:
140 print("\n--- Message Body ---")
141 body = msg.get_payload(decode=True)
142 if body:
143 print(body.decode('utf-8', errors='ignore')[:500])
144 else:
145 print("No messages in mailbox")
146
147 pop3_server.quit()
148 return True
149
150 except Exception as e:
151 print(f"Error: {e}")
152 return False
153
154
155 def example_retrieve_all_messages():
156 """Example 4: Retrieve all messages"""
157 print("\n" + "="*50)
158 print("EXAMPLE 4: Retrieve All Messages")
159 print("="*50 + "\n")
160
161 try:
162 pop3_server = poplib.POP3("localhost", 110)
163 pop3_server.user("testuser")
164 pop3_server.pass_("testpass")
165
166 message_count, _ = pop3_server.stat()
167 print(f"Retrieving {message_count} message(s)...\n")
168
169 for i in range(1, message_count + 1):
170 print(f"--- Message {i} ---")
171 response, lines, octets = pop3_server.retr(i)
172
173 msg_content = b'\n'.join(lines)
174 msg = Parser(policy=default).parsestr(msg_content.decode('utf-8'))
175
176 print(f"Subject: {msg.get('Subject', 'N/A')}")
177 print(f"From: {msg.get('From', 'N/A')}")
178 print(f"Size: {octets} bytes")
179 print()
180
181 pop3_server.quit()
182 return True
183
184 except Exception as e:
185 print(f"Error: {e}")
186 return False
187
188
189 def example_delete_messages():
190 """Example 5: Delete messages"""
191 print("\n" + "="*50)
192 print("EXAMPLE 5: Delete Messages")
193 print("="*50 + "\n")
194
195 try:
196 pop3_server = poplib.POP3("localhost", 110)
197 pop3_server.user("testuser")
198 pop3_server.pass_("testpass")
199
200 message_count, _ = pop3_server.stat()
201
202 if message_count > 0:
203 print(f"Marking message 1 for deletion...")
204 response = pop3_server.dele(1)
205 print(f"Response: {response.decode('utf-8')}")
206
207 print("\nResetting (unmark deletion)...")
208 response = pop3_server.rset()
209 print(f"Response: {response.decode('utf-8')}")
210
211 # Verify message still exists
212 message_count_after_reset, _ = pop3_server.stat()
213 print(f"Messages after reset: {message_count_after_reset}")
214
215 # Mark for deletion again
216 print("\nMarking message 1 for deletion again...")
217 response = pop3_server.dele(1)
218 print(f"Response: {response.decode('utf-8')}")
219
220 print("\nNote: Deletion will be committed when we QUIT")
221 else:
222 print("No messages to delete")
223
224 pop3_server.quit()
225 print("Connection closed - deletion committed")
226 return True
227
228 except Exception as e:
229 print(f"Error: {e}")
230 return False
231
232
233 def example_top_command():
234 """Example 6: Use TOP command to get message headers"""
235 print("\n" + "="*50)
236 print("EXAMPLE 6: TOP Command (Headers Only)")
237 print("="*50 + "\n")
238
239 try:
240 pop3_server = poplib.POP3("localhost", 110)
241 pop3_server.user("testuser")
242 pop3_server.pass_("testpass")
243
244 message_count, _ = pop3_server.stat()
245
246 if message_count > 0:
247 # Get headers and first 0 lines of body
248 print("Retrieving headers of message 1...")
249 response, lines, octets = pop3_server.top(1, 0)
250 print(f"Response: {response.decode('utf-8')}")
251
252 print("\n--- Headers ---")
253 for line in lines[:10]: # Show first 10 lines
254 print(line.decode('utf-8', errors='ignore'))
255 else:
256 print("No messages in mailbox")
257
258 pop3_server.quit()
259 return True
260
261 except poplib.error_proto as e:
262 print(f"TOP command not supported by server: {e}")
263 return False
264 except Exception as e:
265 print(f"Error: {e}")
266 return False
267
268
269 def example_uidl_command():
270 """Example 7: Use UIDL command for unique message IDs"""
271 print("\n" + "="*50)
272 print("EXAMPLE 7: UIDL Command (Unique IDs)")
273 print("="*50 + "\n")
274
275 try:
276 pop3_server = poplib.POP3("localhost", 110)
277 pop3_server.user("testuser")
278 pop3_server.pass_("testpass")
279
280 message_count, _ = pop3_server.stat()
281
282 if message_count > 0:
283 print("Getting unique IDs for all messages...")
284 response, uidl_list, octets = pop3_server.uidl()
285 print(f"Response: {response.decode('utf-8')}")
286
287 print("\n--- Unique IDs ---")
288 for uidl in uidl_list:
289 print(uidl.decode('utf-8'))
290 else:
291 print("No messages in mailbox")
292
293 pop3_server.quit()
294 return True
295
296 except poplib.error_proto as e:
297 print(f"UIDL command not supported by server: {e}")
298 return False
299 except Exception as e:
300 print(f"Error: {e}")
301 return False
302
303
304 def example_error_handling():
305 """Example 8: Error handling scenarios"""
306 print("\n" + "="*50)
307 print("EXAMPLE 8: Error Handling")
308 print("="*50 + "\n")
309
310 # Test 1: Wrong credentials
311 print("--- Test 1: Authentication with wrong credentials ---")
312 try:
313 pop3_server = poplib.POP3("localhost", 110)
314 pop3_server.user("wronguser")
315 pop3_server.pass_("wrongpass")
316 pop3_server.quit()
317 print("Unexpected: Authentication succeeded")
318 except poplib.error_proto as e:
319 print(f"Expected error: {e}")
320 except Exception as e:
321 print(f"Connection error (server not running?): {e}")
322
323 # Test 2: Retrieve non-existent message
324 print("\n--- Test 2: Retrieve non-existent message ---")
325 try:
326 pop3_server = poplib.POP3("localhost", 110)
327 pop3_server.user("testuser")
328 pop3_server.pass_("testpass")
329
330 # Try to retrieve message 99999 (should not exist)
331 response, lines, octets = pop3_server.retr(99999)
332 print("Unexpected: Retrieved non-existent message")
333 pop3_server.quit()
334 except poplib.error_proto as e:
335 print(f"Expected error: {e}")
336 except Exception as e:
337 print(f"Connection error: {e}")
338
339 # Test 3: Connection to wrong port
340 print("\n--- Test 3: Connection to wrong port ---")
341 try:
342 pop3_server = poplib.POP3("localhost", 9999, timeout=2)
343 print("Unexpected: Connection succeeded")
344 except Exception as e:
345 print(f"Expected error: {e}")
346
347
348 def example_secure_connection():
349 """Example 9: Secure POP3 connection (POP3S)"""
350 print("\n" + "="*50)
351 print("EXAMPLE 9: Secure POP3 Connection (POP3S)")
352 print("="*50 + "\n")
353
354 try:
355 # POP3S typically uses port 995
356 print("Connecting to localhost:995 with SSL/TLS...")
357 pop3_server = poplib.POP3_SSL("localhost", 995)
358
359 greeting = pop3_server.getwelcome()
360 print(f"Server greeting: {greeting.decode('utf-8')}")
361
362 pop3_server.user("testuser")
363 pop3_server.pass_("testpass")
364 print("Secure authentication successful!")
365
366 pop3_server.quit()
367 return True
368
369 except Exception as e:
370 print(f"POP3S connection failed (expected if not configured): {e}")
371 return False
372
373
374 def main():
375 """Run all examples"""
376 print("="*60)
377 print("POP3 Client Examples in Python")
378 print("="*60)
379 print("\nThese examples demonstrate POP3 operations using Python's")
380 print("poplib library.")
381 print("\nNote: Examples assume a POP3 server is running on localhost:110")
382 print(" with user 'testuser' and password 'testpass'.")
383 print("\nIf the server is not running, you'll see connection errors.")
384
385 # Run examples
386 examples = [
387 example_basic_connection,
388 example_list_messages,
389 example_retrieve_message,
390 example_retrieve_all_messages,
391 example_delete_messages,
392 example_top_command,
393 example_uidl_command,
394 example_error_handling,
395 example_secure_connection,
396 ]
397
398 for example_func in examples:
399 try:
400 example_func()
401 except KeyboardInterrupt:
402 print("\n\nInterrupted by user")
403 sys.exit(0)
404 except Exception as e:
405 print(f"Unexpected error in {example_func.__name__}: {e}")
406
407 print("\n" + "="*60)
408 print("All examples completed!")
409 print("="*60)
410
411
412 if __name__ == "__main__":
413 main()