Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_login_without_method_raises_error(mock_auth):
mock_auth.server_auth_methods = []
with pytest.raises(SMTPException):
await mock_auth.login("username", "bogus")
Only supports plain text emails with a single message body.
No attachments etc.
"""
if type(mail_to) == str:
mail_to = [mail_to]
smtp = aiosmtplib.SMTP(hostname=server, port=25, loop=loop)
try:
await smtp.connect()
for rcpt in mail_to:
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = mail_from
msg['To'] = rcpt
await smtp.send_message(msg)
await smtp.quit()
except aiosmtplib.errors.SMTPException as e:
log.msg('Error sending smtp notification: %s' % (str(e)), 'NOTIFICATIONS')
"SMTPConnectTimeoutError",
"SMTPReadTimeoutError",
)
class SMTPException(Exception):
"""
Base class for all SMTP exceptions.
"""
def __init__(self, message: str) -> None:
self.message = message
self.args = (message,)
class SMTPServerDisconnected(SMTPException, ConnectionError):
"""
The connection was lost unexpectedly, or a command was run that requires
a connection.
"""
class SMTPConnectError(SMTPException, ConnectionError):
"""
An error occurred while connecting to the SMTP server.
"""
class SMTPTimeoutError(SMTPException, TimeoutError):
"""
A timeout occurred while performing a network operation.
"""
self,
username: str,
password: str,
timeout: Optional[Union[float, Default]] = _default,
) -> SMTPResponse:
"""
Tries to login with supported auth methods.
Some servers advertise authentication methods they don't really
support, so if authentication fails, we continue until we've tried
all methods.
"""
await self._ehlo_or_helo_if_needed()
if not self.supports_extension("auth"):
raise SMTPException("SMTP AUTH extension not supported by server.")
response = None # type: Optional[SMTPResponse]
exception = None # type: Optional[SMTPAuthenticationError]
for auth_name in self.supported_auth_methods:
method_name = "auth_{}".format(auth_name.replace("-", ""))
try:
auth_method = getattr(self, method_name)
except AttributeError:
raise RuntimeError(
"Missing handler for auth method {}".format(auth_name)
)
try:
response = await auth_method(username, password, timeout=timeout)
except SMTPAuthenticationError as exc:
exception = exc
else:
"""
class SMTPConnectTimeoutError(SMTPTimeoutError, SMTPConnectError):
"""
A timeout occurred while connecting to the SMTP server.
"""
class SMTPReadTimeoutError(SMTPTimeoutError):
"""
A timeout occurred while waiting for a response from the SMTP server.
"""
class SMTPNotSupported(SMTPException):
"""
A command or argument sent to the SMTP server is not supported.
"""
class SMTPResponseException(SMTPException):
"""
Base class for all server responses with error codes.
"""
def __init__(self, code: int, message: str) -> None:
self.code = code
self.message = message
self.args = (code, message)
validate_certs=validate_certs,
client_cert=client_cert,
client_key=client_key,
cert_bundle=cert_bundle,
tls_context=tls_context,
timeout=timeout,
)
self._validate_config()
if server_hostname is None:
server_hostname = self.hostname
tls_context = self._get_tls_context()
if not self.supports_extension("starttls"):
raise SMTPException("SMTP STARTTLS extension not supported by server.")
response = await self.protocol.start_tls(
tls_context, server_hostname=server_hostname, timeout=self.timeout
)
if self.protocol is None:
raise SMTPServerDisconnected("Connection lost")
# Update our transport reference
self.transport = self.protocol.transport
# RFC 3207 part 4.2:
# The client MUST discard any knowledge obtained from the server, such
# as the list of SMTP service extensions, which was not obtained from
# the TLS negotiation itself.
self._reset_server_state()
return response
self.args = (code, message, sender)
class SMTPRecipientRefused(SMTPResponseException):
"""
SMTP server refused a message recipient.
"""
def __init__(self, code: int, message: str, recipient: str) -> None:
self.code = code
self.message = message
self.recipient = recipient
self.args = (code, message, recipient)
class SMTPRecipientsRefused(SMTPException):
"""
SMTP server refused multiple recipients.
"""
def __init__(self, recipients: List[SMTPRecipientRefused]) -> None:
self.recipients = recipients
self.args = (recipients,)
Base class for all SMTP exceptions.
"""
def __init__(self, message: str) -> None:
self.message = message
self.args = (message,)
class SMTPServerDisconnected(SMTPException, ConnectionError):
"""
The connection was lost unexpectedly, or a command was run that requires
a connection.
"""
class SMTPConnectError(SMTPException, ConnectionError):
"""
An error occurred while connecting to the SMTP server.
"""
class SMTPTimeoutError(SMTPException, TimeoutError):
"""
A timeout occurred while performing a network operation.
"""
class SMTPConnectTimeoutError(SMTPTimeoutError, SMTPConnectError):
"""
A timeout occurred while connecting to the SMTP server.
"""
"""
class SMTPReadTimeoutError(SMTPTimeoutError):
"""
A timeout occurred while waiting for a response from the SMTP server.
"""
class SMTPNotSupported(SMTPException):
"""
A command or argument sent to the SMTP server is not supported.
"""
class SMTPResponseException(SMTPException):
"""
Base class for all server responses with error codes.
"""
def __init__(self, code: int, message: str) -> None:
self.code = code
self.message = message
self.args = (code, message)
class SMTPHeloError(SMTPResponseException):
"""
Server refused HELO or EHLO.
"""
try:
auth_method = getattr(self, method_name)
except AttributeError:
raise RuntimeError(
"Missing handler for auth method {}".format(auth_name)
)
try:
response = await auth_method(username, password, timeout=timeout)
except SMTPAuthenticationError as exc:
exception = exc
else:
# No exception means we're good
break
if response is None:
raise exception or SMTPException("No suitable authentication method found.")
return response