Coverage for chalicelib/utils/smtp.py: 39%

1import logging 

2import smtplib 

3from smtplib import SMTPAuthenticationError 

4 

5from decouple import config 

6from fastapi import HTTPException 

7 

8logger = logging.getLogger(__name__) 

9 

10class EmptySMTP: 

11 def sendmail(self, from_addr, to_addrs, msg, mail_options=(), rcpt_options=()): 

12 logger.error("!! CANNOT SEND EMAIL, NO VALID SMTP CONFIGURATION FOUND") 

13 

14 def send_message(self, msg): 

15 self.sendmail( msg["FROM"], msg["TO"], msg.as_string() ) 

16 

17class SMTPClient: 

18 server = None 

19 

20 def __init__(self): 

21 if config("EMAIL_HOST") is None or len(config("EMAIL_HOST")) == 0: 21 ↛ 23line 21 didn't jump to line 23 because the condition on line 21 was always true

22 return 

23 elif not config("EMAIL_USE_SSL", cast=bool): 

24 self.server = smtplib.SMTP(host=config("EMAIL_HOST"), port=config("EMAIL_PORT", cast=int)) 

25 else: 

26 if len(config("EMAIL_SSL_KEY")) == 0 or len(config("EMAIL_SSL_CERT")) == 0: 

27 self.server = smtplib.SMTP_SSL(host=config("EMAIL_HOST"), port=config("EMAIL_PORT", cast=int)) 

28 else: 

29 self.server = smtplib.SMTP_SSL(host=config("EMAIL_HOST"), port=config("EMAIL_PORT", cast=int), 

30 keyfile=config("EMAIL_SSL_KEY"), certfile=config("EMAIL_SSL_CERT")) 

31 

32 def __enter__(self): 

33 if self.server is None: 33 ↛ 35line 33 didn't jump to line 35 because the condition on line 33 was always true

34 return EmptySMTP() 

35 self.server.ehlo() 

36 if not config("EMAIL_USE_SSL", cast=bool) and config("EMAIL_USE_TLS", cast=bool): 

37 self.server.starttls() 

38 # stmplib docs recommend calling ehlo() before & after starttls() 

39 self.server.ehlo() 

40 if len(config("EMAIL_USER", default="")) > 0 and len(config("EMAIL_PASSWORD", default="")) > 0: 

41 try: 

42 self.server.login(user=config("EMAIL_USER"), password=config("EMAIL_PASSWORD")) 

43 except SMTPAuthenticationError: 

44 raise HTTPException(401, "SMTP Authentication Error") 

45 return self.server 

46 

47 def __exit__(self, *args): 

48 if self.server is None: 48 ↛ 50line 48 didn't jump to line 50 because the condition on line 48 was always true

49 return 

50 self.server.quit() 

51 

52 def test_configuration(self): 

53 # check server connexion 

54 try: 

55 status = self.server.noop()[0] 

56 if not (status == 250): 

57 raise Exception(f"SMTP connexion error, status:{status}") 

58 except Exception as e: # smtplib.SMTPServerDisconnected 

59 logger.error( 

60 f'!! SMTP connexion error to {config("EMAIL_HOST")}:{config("EMAIL_PORT", cast=int)}') 

61 logger.error(e) 

62 return False, e 

63 

64 # check authentication 

65 try: 

66 self.__enter__() 

67 self.__exit__() 

68 except Exception as e: 

69 logger.error(f'!! SMTP authentication error to {config("EMAIL_HOST")}:{config("EMAIL_PORT", cast=int)}') 

70 logger.error(e) 

71 return False, e 

72 

73 return True, None 

74 

75 

76VALID_SMTP = None 

77SMTP_ERROR = None 

78SMTP_NOTIFIED = False 

79 

80 

81def has_smtp(): 

82 global VALID_SMTP, SMTP_ERROR, SMTP_NOTIFIED 

83 if SMTP_ERROR is not None: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true

84 logger.error("!!! SMTP error found, disabling SMTP configuration:") 

85 logger.error(SMTP_ERROR) 

86 

87 if VALID_SMTP is not None: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 return VALID_SMTP 

89 

90 if config("EMAIL_HOST") is not None and len(config("EMAIL_HOST")) > 0: 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true

91 VALID_SMTP, SMTP_ERROR = check_connexion() 

92 return VALID_SMTP 

93 elif not SMTP_NOTIFIED: 

94 SMTP_NOTIFIED = True 

95 logger.info("no SMTP configuration found") 

96 return False 

97 

98 

99def check_connexion(): 

100 # check SMTP host&port 

101 import socket 

102 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

103 sock.settimeout(config("EMAIL_CHECK_TIMEOUT", cast=int, default=5)) 

104 result = sock.connect_ex((config("EMAIL_HOST"), config("EMAIL_PORT", cast=int))) 

105 sock.close() 

106 if not (result == 0): 

107 error = f"""!! SMTP {config("EMAIL_HOST")}:{config("EMAIL_PORT", cast=int)} is unreachable 

108f'please make sure the host&port are correct, and the SMTP protocol is authorized on your server.""" 

109 logger.error(error) 

110 sock.close() 

111 return False, error 

112 

113 return SMTPClient().test_configuration()