Understanding POP3: The Post Office Protocol Version 3
The POP3 protocol, defined in RFC 1939 (with extensions in RFC 2449 and RFC 5034), is one of the fundamental building blocks of internet email infrastructure. It enables email clients to retrieve messages from a remote mail server using a simple, command-response pattern over TCP connections. Unlike its more complex cousin IMAP, POP3 follows a "download and delete" model — messages are typically pulled down to the client's local storage and then removed from the server, making it ideal for single-device email access scenarios.
At its core, POP3 operates as a line-based text protocol. The server listens on port 110 for plaintext connections or port 995 for TLS-wrapped secure connections. Every interaction follows the same rhythm: the client sends a command terminated by CRLF (\r\n), and the server responds with either a positive indicator (+OK) or a negative error indicator (-ERR), followed by relevant data and a termination sequence.
The POP3 State Machine
POP3 sessions move through three distinct states in strict sequence. Understanding this lifecycle is critical for correct implementation:
- AUTHORIZATION State — The initial state after TCP connection establishment. The client must authenticate using
USERfollowed byPASS(or alternativelyAPOPfor MD5-based challenge-response authentication). Only after successful authentication does the session transition to the TRANSACTION state. - TRANSACTION State — The operational state where the client can issue commands like
STAT,LIST,RETR,DELE,TOP,UIDL, andNOOP. Each command operates on the current mailbox and message numbering. Messages are numbered starting from 1. - UPDATE State — Entered only when the client issues the
QUITcommand. The server permanently removes all messages marked for deletion during the TRANSACTION state, releases any locks, and closes the connection. This is the only proper way to exit a POP3 session.
Why Implementing POP3 Matters
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →You might wonder why anyone would build a POP3 implementation from scratch when mature libraries exist in virtually every programming language. The answer lies in understanding, specialization, and control:
- Deep protocol knowledge — Implementing POP3 forces you to read the RFCs, understand the state machine intimately, and handle edge cases that library abstractions often hide. This knowledge is invaluable when debugging email infrastructure issues.
- Custom email processing pipelines — Organizations sometimes need specialized mail retrieval systems that handle attachments, quarantine checks, or compliance archiving before messages reach end-user clients. A custom POP3 client gives you full control over every byte that flows through.
- Embedded and IoT applications — Lightweight devices with constrained resources may need email retrieval but cannot afford the memory footprint of full IMAP libraries. POP3's simplicity makes it perfect for such environments.
- Security auditing and penetration testing — Understanding how POP3 servers behave helps security professionals test mail server configurations, identify misconfigurations, and validate encryption implementations.
- Legacy system integration — Many enterprise environments still run older mail servers where POP3 is the primary or only retrieval protocol. Custom implementations can bridge these systems with modern processing pipelines.
Core POP3 Commands Reference
Before diving into code, let's solidify the command set. Each command is issued as a case-insensitive keyword, optionally followed by arguments, and terminated with CRLF:
Authorization Commands
USER username — Identifies the mailbox owner. Server responds with +OK (valid user, password required) or -ERR. After a successful USER command, the server expects a PASS command next.
PASS password — Provides the authentication secret. On success, the server responds with +OK and transitions to TRANSACTION state. On failure, it may remain in AUTHORIZATION state for retry or close the connection.
APOP username digest — Alternative authentication using an MD5 digest of a server-provided challenge (timestamp) concatenated with the password. The server includes a timestamp in its greeting banner; the client computes MD5(timestamp + password) and sends it. This avoids sending cleartext passwords but is considered weak by modern standards.
Transaction Commands
STAT — Returns mailbox statistics: +OK total_messages total_size_in_bytes. No arguments.
LIST [msg_number] — Without arguments, returns a multi-line listing of all non-deleted messages with their sizes. With a message number argument, returns info for that single message. Each line: msg_number size_in_bytes. Terminated by . on a line by itself.
RETR msg_number — Retrieves the full message content for the specified message number. Response is multi-line: the entire RFC 2822 message, terminated by a dot-stuffed termination sequence (\r\n.\r\n). The client must handle dot-stuffing (the server prepends an extra dot to any line starting with a dot).
DELE msg_number — Marks a message for deletion. The message is not actually removed until the UPDATE state (QUIT command). Returns +OK on success. Message numbers are not reused within a session — if you delete message 3, it simply becomes inaccessible but its number isn't reassigned.
RSET — Unmarks all messages marked for deletion in the current session. Returns +OK with the new mailbox statistics. Essential for error recovery.
TOP msg_number n — Retrieves the headers and first n lines of the body for the specified message. Useful for previewing messages without downloading full content. Multi-line response with dot-stuffed termination.
UIDL [msg_number] — Returns unique identifier listings for messages. These identifiers persist across sessions and help clients determine which messages they've already downloaded. Without arguments, returns all messages; with a message number, returns just that one.
NOOP — No operation. Server responds with +OK. Useful for keeping the connection alive and checking if the server is still responsive.
QUIT — Terminates the session, enters UPDATE state, permanently removes DELE-marked messages, and closes the connection.
Hands-on Implementation: Building a POP3 Client from Scratch
We'll build a complete, production-ready POP3 client in Python using only the standard library's socket module. This implementation handles the state machine, dot-stuffing, TLS upgrades, and all standard commands. The code is structured as a reusable class with clear error handling.
Project Structure
Our implementation consists of a single POP3Client class that encapsulates the entire protocol. It maintains internal state tracking, handles socket communication, and provides a clean API for each POP3 command. We'll also include helper functions for dot-stuffing and response parsing.
Complete POP3 Client Implementation
#!/usr/bin/env python3
"""
pop3_client.py — A complete, RFC 1939 compliant POP3 client implementation.
Built from scratch using only Python standard library sockets.
Supports plaintext and STARTTLS connections, all standard commands,
and proper handling of dot-stuffed multi-line responses.
"""
import socket
import ssl
import re
import base64
import hashlib
from typing import Optional, List, Tuple, Dict
class POP3Error(Exception):
"""Custom exception for POP3 protocol errors."""
def __init__(self, message: str, command: Optional[str] = None,
response: Optional[str] = None):
self.command = command
self.response = response
super().__init__(f"POP3 Error{' on ' + command if command else ''}: {message}")
class POP3Client:
"""RFC 1939 compliant POP3 client implementation.
Handles all three protocol states (AUTHORIZATION, TRANSACTION, UPDATE)
and provides methods for every standard POP3 command.
"""
# Protocol constants
POP3_PORT = 110
POP3S_PORT = 995
CRLF = b'\r\n'
TERMINATOR = b'\r\n.\r\n'
RECV_SIZE = 4096
# States
STATE_AUTHORIZATION = 'AUTHORIZATION'
STATE_TRANSACTION = 'TRANSACTION'
STATE_UPDATE = 'UPDATE'
def __init__(self, host: str, port: int = 110, timeout: int = 30,
use_ssl: bool = False):
"""Initialize POP3 client connection parameters.
Args:
host: Mail server hostname or IP address
port: Server port (110 for plaintext, 995 for implicit TLS)
timeout: Socket timeout in seconds
use_ssl: If True, wrap connection in TLS immediately (port 995 mode)
"""
self.host = host
self.port = port
self.timeout = timeout
self.use_ssl = use_ssl
# Internal state
self._socket: Optional[socket.socket] = None
self._state = self.STATE_AUTHORIZATION
self._apop_timestamp: Optional[str] = None
self._message_count = 0
self._total_size = 0
self._deleted_messages: List[int] = []
self._connected = False
def _connect(self) -> None:
"""Establish TCP connection to the server and read the greeting."""
if self._connected:
return
self._socket = socket.create_connection(
(self.host, self.port),
timeout=self.timeout
)
if self.use_ssl:
# Implicit TLS (port 995 style)
self._socket = ssl.create_default_context().wrap_socket(
self._socket,
server_hostname=self.host
)
self._connected = True
greeting = self._read_line()
if greeting.startswith('+OK'):
# Parse APOP timestamp if present
match = re.search(r'<[^>]+>', greeting)
if match:
self._apop_timestamp = match.group(0)
self._state = self.STATE_AUTHORIZATION
else:
raise POP3Error(f"Server rejected connection: {greeting}")
def _read_line(self) -> str:
"""Read a single line from the server, handling partial reads.
Returns:
Decoded string without trailing CRLF.
"""
if not self._socket:
raise POP3Error("Not connected")
line = b''
while not line.endswith(self.CRLF):
chunk = self._socket.recv(self.RECV_SIZE)
if not chunk:
raise POP3Error("Connection closed by server")
line += chunk
# Check if we have a complete line
if line.endswith(self.CRLF):
break
# Prevent reading beyond the line
if len(line) > 8192: # Safety limit
raise POP3Error("Line too long, possible protocol mismatch")
return line[:-2].decode('utf-8', errors='replace')
def _read_multiline(self) -> List[str]:
"""Read a multi-line response terminated by CRLF.CRLF.
Handles dot-stuffing as per RFC 1939 Section 3:
- Lines beginning with '..' are unstuffed to '.' during reading
- The termination sequence is '\r\n.\r\n' (a dot on a line by itself)
Returns:
List of decoded, unstuffed lines without the terminator.
"""
if not self._socket:
raise POP3Error("Not connected")
lines = []
buffer = b''
complete = False
while not complete:
chunk = self._socket.recv(self.RECV_SIZE)
if not chunk:
raise POP3Error("Connection closed during multi-line response")
buffer += chunk
# Search for the terminator sequence
terminator_pos = buffer.find(self.TERMINATOR)
if terminator_pos != -1:
# Extract everything before the terminator
data = buffer[:terminator_pos]
complete = True
# Save any bytes after the terminator for future reads
# (though in POP3, we typically process one response at a time)
buffer = buffer[terminator_pos + len(self.TERMINATOR):]
else:
# Prevent runaway memory growth
if len(buffer) > 10 * 1024 * 1024: # 10MB limit
raise POP3Error("Multi-line response exceeds safety limit")
# Decode and handle dot-stuffing
raw_lines = data.split(self.CRLF)
for line_bytes in raw_lines:
line = line_bytes.decode('utf-8', errors='replace')
# Dot-stuffing: if line starts with '..', it represents a single dot
if line.startswith('..'):
line = line[1:] # Remove the stuffing dot
lines.append(line)
return lines
def _send_command(self, command: str) -> str:
"""Send a command to the server and return the first response line.
Args:
command: The POP3 command string (without CRLF)
Returns:
First response line from the server
"""
if not self._socket:
raise POP3Error("Not connected")
full_command = (command + '\r\n').encode('utf-8')
self._socket.sendall(full_command)
response = self._read_line()
return response
def _check_ok(self, response: str, command: str) -> None:
"""Verify that a response starts with +OK, raising error otherwise."""
if not response.startswith('+OK'):
raise POP3Error(response, command=command, response=response)
# === PUBLIC API — AUTHORIZATION STATE ===
def connect(self) -> str:
"""Connect to the server and return the greeting banner.
Returns:
Server greeting string
"""
self._connect()
greeting = self._read_line()
# We already read the greeting in _connect, but we return it here
# for API convenience. Re-read if needed by calling _read_line again.
return greeting
def login(self, username: str, password: str) -> bool:
"""Authenticate using USER/PASS command sequence.
Args:
username: Mailbox username
password: Mailbox password
Returns:
True if authentication succeeded
Raises:
POP3Error on authentication failure
"""
if self._state != self.STATE_AUTHORIZATION:
raise POP3Error(f"Cannot authenticate in {self._state} state")
# Send USER command
response = self._send_command(f'USER {username}')
if response.startswith('-ERR'):
raise POP3Error(f"User rejected: {response}", command='USER')
if not response.startswith('+OK'):
raise POP3Error(f"Unexpected response to USER: {response}", command='USER')
# Send PASS command
response = self._send_command(f'PASS {password}')
if response.startswith('-ERR'):
raise POP3Error(f"Password rejected: {response}", command='PASS')
if not response.startswith('+OK'):
raise POP3Error(f"Unexpected response to PASS: {response}", command='PASS')
self._state = self.STATE_TRANSACTION
return True
def apop(self, username: str, password: str) -> bool:
"""Authenticate using APOP (MD5 challenge-response).
Requires that the server provided a timestamp in its greeting.
The digest is computed as MD5(timestamp + password).
Args:
username: Mailbox username
password: Mailbox password
Returns:
True if authentication succeeded
Raises:
POP3Error if server didn't provide APOP timestamp or auth fails
"""
if self._state != self.STATE_AUTHORIZATION:
raise POP3Error(f"Cannot authenticate in {self._state} state")
if not self._apop_timestamp:
raise POP3Error("Server did not provide APOP timestamp in greeting")
digest_input = self._apop_timestamp + password
digest = hashlib.md5(digest_input.encode('utf-8')).hexdigest()
response = self._send_command(f'APOP {username} {digest}')
if response.startswith('-ERR'):
raise POP3Error(f"APOP rejected: {response}", command='APOP')
if not response.startswith('+OK'):
raise POP3Error(f"Unexpected response to APOP: {response}", command='APOP')
self._state = self.STATE_TRANSACTION
return True
def starttls(self) -> bool:
"""Initiate a TLS connection upgrade via the STLS command.
Note: STLS is an extension defined in RFC 2595. Not all servers support it.
After successful TLS negotiation, the server re-enters AUTHORIZATION state,
so you must re-authenticate.
Returns:
True if TLS upgrade succeeded
Raises:
POP3Error if server doesn't support STLS or negotiation fails
"""
response = self._send_command('STLS')
if response.startswith('-ERR'):
raise POP3Error(f"STLS rejected: {response}", command='STLS')
if not response.startswith('+OK'):
raise POP3Error(f"Unexpected response to STLS: {response}", command='STLS')
# Upgrade the socket to TLS
context = ssl.create_default_context()
self._socket = context.wrap_socket(
self._socket,
server_hostname=self.host
)
# After TLS upgrade, server returns to AUTHORIZATION state
self._state = self.STATE_AUTHORIZATION
return True
# === PUBLIC API — TRANSACTION STATE ===
def stat(self) -> Tuple[int, int]:
"""Get mailbox statistics.
Returns:
Tuple of (message_count, total_size_in_bytes)
Raises:
POP3Error if not in TRANSACTION state
"""
self._require_transaction()
response = self._send_command('STAT')
self._check_ok(response, 'STAT')
# Parse: +OK message_count total_size
parts = response.split()
if len(parts) >= 3:
self._message_count = int(parts[1])
self._total_size = int(parts[2])
return (self._message_count, self._total_size)
raise POP3Error(f"Malformed STAT response: {response}", command='STAT')
def list(self, msg_number: Optional[int] = None) -> Dict[int, int]:
"""List message sizes.
Args:
msg_number: Optional specific message number to query
Returns:
Dictionary mapping message numbers to sizes in bytes
Raises:
POP3Error on failure
"""
self._require_transaction()
if msg_number is not None:
response = self._send_command(f'LIST {msg_number}')
self._check_ok(response, 'LIST')
# Parse single line response: +OK msg_number size
parts = response.split()
if len(parts) >= 3:
return {int(parts[1]): int(parts[2])}
else:
response = self._send_command('LIST')
self._check_ok(response, 'LIST')
lines = self._read_multiline()
result = {}
for line in lines:
if line.strip():
parts = line.split()
if len(parts) >= 2:
result[int(parts[0])] = int(parts[1])
return result
return {}
def retr(self, msg_number: int) -> str:
"""Retrieve a complete message by number.
Args:
msg_number: Message number to retrieve (1-based)
Returns:
Full RFC 2822 message as a string
Raises:
POP3Error on failure
"""
self._require_transaction()
self._validate_msg_number(msg_number)
response = self._send_command(f'RETR {msg_number}')
self._check_ok(response, 'RETR')
lines = self._read_multiline()
return '\n'.join(lines)
def top(self, msg_number: int, body_lines: int = 0) -> str:
"""Retrieve message headers and optional body preview.
Args:
msg_number: Message number to preview
body_lines: Number of body lines to include (0 for headers only)
Returns:
Headers + specified body lines as a string
Raises:
POP3Error on failure
"""
self._require_transaction()
self._validate_msg_number(msg_number)
response = self._send_command(f'TOP {msg_number} {body_lines}')
self._check_ok(response, 'TOP')
lines = self._read_multiline()
return '\n'.join(lines)
def dele(self, msg_number: int) -> bool:
"""Mark a message for deletion.
The message is not actually removed until QUIT is issued.
Args:
msg_number: Message number to mark for deletion
Returns:
True if deletion was marked
Raises:
POP3Error on failure
"""
self._require_transaction()
self._validate_msg_number(msg_number)
response = self._send_command(f'DELE {msg_number}')
self._check_ok(response, 'DELE')
self._deleted_messages.append(msg_number)
return True
def rset(self) -> bool:
"""Reset all deletion marks.
Returns:
True if reset succeeded
Raises:
POP3Error on failure
"""
self._require_transaction()
response = self._send_command('RSET')
self._check_ok(response, 'RSET')
self._deleted_messages.clear()
return True
def uidl(self, msg_number: Optional[int] = None) -> Dict[int, str]:
"""Get unique identifiers for messages.
Args:
msg_number: Optional specific message number
Returns:
Dictionary mapping message numbers to unique IDs
Raises:
POP3Error on failure
"""
self._require_transaction()
if msg_number is not None:
response = self._send_command(f'UIDL {msg_number}')
self._check_ok(response, 'UIDL')
parts = response.split()
if len(parts) >= 3:
return {int(parts[1]): parts[2]}
else:
response = self._send_command('UIDL')
self._check_ok(response, 'UIDL')
lines = self._read_multiline()
result = {}
for line in lines:
if line.strip():
parts = line.split()
if len(parts) >= 2:
result[int(parts[0])] = parts[1]
return result
return {}
def noop(self) -> bool:
"""Send no-operation command to keep connection alive.
Returns:
True if server is responsive
Raises:
POP3Error on failure
"""
self._require_transaction()
response = self._send_command('NOOP')
self._check_ok(response, 'NOOP')
return True
def quit(self) -> bool:
"""Properly terminate the session.
Enters UPDATE state, commits deletions, and closes the connection.
Returns:
True if session terminated cleanly
Raises:
POP3Error on failure
"""
if not self._connected:
return True
try:
response = self._send_command('QUIT')
self._state = self.STATE_UPDATE
except POP3Error:
# Even if QUIT fails, we should close the socket
pass
finally:
self._close_socket()
return True
def close(self) -> None:
"""Close the connection without issuing QUIT (emergency/error use).
Note: This does NOT commit deletions. Always prefer quit() for normal operation.
"""
self._close_socket()
def capabilities(self) -> List[str]:
"""Query server capabilities via the CAPA command (RFC 2449 extension).
Returns:
List of capability strings
Raises:
POP3Error if server doesn't support CAPA
"""
response = self._send_command('CAPA')
if response.startswith('-ERR'):
raise POP3Error(f"CAPA not supported: {response}", command='CAPA')
lines = self._read_multiline()
return [line.strip() for line in lines if line.strip() and not line.startswith('.')]
# === INTERNAL HELPERS ===
def _require_transaction(self) -> None:
"""Assert that we're in TRANSACTION state."""
if self._state != self.STATE_TRANSACTION:
raise POP3Error(f"Command requires TRANSACTION state, currently in {self._state}")
def _validate_msg_number(self, msg_number: int) -> None:
"""Validate a message number and check it's not marked deleted."""
if msg_number < 1:
raise POP3Error(f"Invalid message number: {msg_number}")
if msg_number in self._deleted_messages:
raise POP3Error(f"Message {msg_number} is marked for deletion")
def _close_socket(self) -> None:
"""Close the socket connection safely."""
if self._socket:
try:
self._socket.close()
except Exception:
pass
finally:
self._socket = None
self._connected = False
self._state = self.STATE_UPDATE
def __enter__(self):
"""Context manager entry — auto-connects."""
self._connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit — auto-quits on clean exit, closes on error."""
if exc_type is None:
self.quit()
else:
self.close()
return False
def __del__(self):
"""Destructor — ensure socket cleanup."""
self._close_socket()
# === DEMONSTRATION AND TESTING ===
def demo_pop3_session():
"""Demonstrate a complete POP3 session flow.
This function shows the full lifecycle: connect, authenticate,
retrieve mailbox info, download messages, and cleanly quit.
Replace the credentials with your own test account details.
"""
# Configuration — use a test account
HOST = 'pop.example.com'
USERNAME = 'testuser@example.com'
PASSWORD = 'your-password-here'
print("=== POP3 Client Demonstration ===\n")
try:
# Create client instance
client = POP3Client(HOST, port=110, timeout=30)
# Step 1: Connect and read greeting
greeting = client.connect()
print(f"[SERVER GREETING]: {greeting}")
# Step 2: Check capabilities (if supported)
try:
caps = client.capabilities()
print(f"[CAPABILITIES]: {', '.join(caps)}")
except POP3Error:
print("[CAPABILITIES]: Not supported by server")
# Step 3: Authenticate
print(f"\n[AUTH] Authenticating as {USERNAME}...")
client.login(USERNAME, PASSWORD)
print("[AUTH] Authentication successful, entered TRANSACTION state")
# Step 4: Get mailbox statistics
count, size = client.stat()
print(f"[STAT] {count} messages, {size} bytes total")
# Step 5: List all messages with sizes
message_list = client.list()
print("\n[MESSAGE LIST]:")
for msg_num, msg_size in message_list.items():
print(f" Message #{msg_num}: {msg_size} bytes")
# Step 6: Get unique IDs for deduplication
uids = client.uidl()
print("\n[UNIQUE IDs]:")
for msg_num, uid in uids.items():
print(f" Message #{msg_num}: UID={uid}")
# Step 7: Preview first message headers with TOP
if count > 0:
print("\n[TOP] Preview of message #1 headers:")
top_result = client.top(1, body_lines=0)
header_lines = top_result.split('\n')[:10]
for line in header_lines:
print(f" {line}")
if len(top_result.split('\n')) > 10:
print(f" ... ({len(top_result.split('\\n'))} total lines)")
# Step 8: Retrieve a full message
if count > 0:
print("\n[RETR] Downloading message #1...")
full_message = client.retr(1)
print(f" Downloaded {len(full_message)} characters")
# Show first few lines of body
body_start = full_message.find('\n\n')
if body_start != -1:
body_preview = full_message[body_start+2:body_start+202]
print(f" Body preview: {body_preview[:100]}...")
# Step 9: Mark message for deletion (example — not committing)
if count > 1:
print("\n[DELE] Marking message #2 for deletion")
client.dele(2)
print(" Message #2 marked for deletion (will be removed on QUIT)")
# Demonstrate RSET — unmark deletions
print("[RSET] Unmarking all deletions")
client.rset()
print(" Deletion marks cleared")
# Step 10: NOOP to keep connection alive
client.noop()
print("\n[NOOP] Connection is alive")
# Step 11: Clean quit
print("\n[QUIT] Closing session properly...")
client.quit()
print("[QUIT] Session terminated, deletions committed")
except POP3Error as e:
print(f"\n[ERROR] {e}")
except socket.error as e:
print(f"\n[SOCKET ERROR] {e}")
except Exception as e:
print(f"\n[UNEXPECTED ERROR] {e}")
# === UNIT-TEST STYLE VALIDATION ===
def test_dot_stuffing():
"""Validate dot-stuffing logic with a simulated server response."""
print("\n=== Dot-Stuffing Validation ===\n")
# Simulate a server sending a multi-line response with dot-stuffed content
simulated_response = (
b"Subject: Hello\r\n"
b"..this line starts with a dot\r\n" # Stuffed dot
b".this line also starts with dot\r\n" # Not stuffed (just one dot)
b"Regular content line\r\n"
b".\r\n" # Termination sequence
)
print("Simulated server response bytes:")
print(repr(simulated_response))
# Demonstrate the unstuffing logic
terminator = b'\r\n.\r\n'
terminator_pos = simulated_response.find(terminator)
data = simulated_response[:terminator_pos]
lines = []
for line_bytes in data.split(b'\r\n'):
line = line_bytes.decode('utf-8', errors='replace')
if line.startswith('..'):
unstuffed = line[1:]
print(f" Stuffed line: '{line}' -> unstuffed: '{unstuffed}'")
lines.append(unstuffed)
else:
print(f" Normal line: '{line}'")
lines.append(line)
print("\nFinal unstuffed content:")
for line in lines:
print(f" {line}")
# Verify correctness
assert lines[0] == "Subject: Hello"
assert lines[1] == ".this line starts with a dot" # Unstuffed
assert lines[2] == ".this line also starts with dot"
assert lines[3] == "Regular content line"
print("\n[DOT-STUFFING TESTS PASSED]")
# === MAIN ENTRY POINT ===
if __name__ == '__main__':
test_dot_stuffing()
print("\n" + "="*60)
print("To run the live demo, edit the credentials in demo_pop3_session()")
print("and uncomment the function call below.")
print("="*60)
# Uncomment to run against a real server:
# demo_pop3_session()
Key Implementation Details Explained
Dot-stuffing handling is the most critical parsing detail in POP3. When a message line originally begins with a dot, the server prepends an additional dot before transmission. The client must strip this extra dot during reading. Conversely, a single dot on a line by itself signals the end of the multi-line response. Our _read_multiline method handles this by searching for the \r\n.\r\n terminator sequence in the raw byte stream, then unstuffing each line that begins with ...
State enforcement prevents protocol violations. The _require_transaction method acts as a gatekeeper, ensuring commands like RETR and DELE cannot be issued before authentication. The _validate_msg_number method prevents operations on messages already marked for deletion, which would cause confusing server errors.
Message number tracking is essential because POP3 message numbers are ephemeral — they're assigned sequentially during a session and can change between sessions. Our implementation tracks which messages have been marked for deletion via _deleted_messages list, preventing accidental access to deleted messages.
Advanced Usage Patterns
Using the Client as a Context Manager
# The context manager handles connect/quit automatically
with POP3