Coverage for src/notify.py: 100%

1"""Send an email if the 2FA is expired.""" 

2 

3import datetime 

4import smtplib 

5 

6import requests 

7 

8from src import config_parser, get_logger 

9from src.email_message import EmailMessage as Message 

10 

11LOGGER = get_logger() 

12 

13 

14def notify_telegram(config, message, last_send=None, dry_run=False): 

15 """Send telegram notification.""" 

16 sent_on = None 

17 bot_token = config_parser.get_telegram_bot_token(config=config) 

18 chat_id = config_parser.get_telegram_chat_id(config=config) 

19 

20 if last_send and last_send > datetime.datetime.now() - datetime.timedelta(hours=24): 

21 LOGGER.info("Throttling telegram to once a day") 

22 sent_on = last_send 

23 elif bot_token and chat_id: 

24 sent_on = datetime.datetime.now() 

25 if not dry_run: 

26 # Post message to telegram bot using API 

27 if not post_message_to_telegram( 

28 bot_token, 

29 chat_id, 

30 message, 

31 ): 

32 sent_on = None 

33 else: 

34 LOGGER.warning("Not sending 2FA notification because Telegram is not configured.") 

35 return sent_on 

36 

37 

38def post_message_to_telegram(bot_token, chat_id, message): 

39 """Post message to telegram bot using API.""" 

40 url = f"https://api.telegram.org/bot{bot_token}/sendMessage" 

41 params = {"chat_id": chat_id, "text": message} 

42 response = requests.post(url, params=params, timeout=10) 

43 if response.status_code == 200: 

44 return True 

45 # Log error message 

46 LOGGER.error(f"Failed to send telegram notification. Response: {response.text}") 

47 return False 

48 

49 

50def post_message_to_discord(webhook_url, username, message): 

51 """Post message to discord webhook.""" 

52 data = {"username": username, "content": message} 

53 response = requests.post(webhook_url, data=data, timeout=10) 

54 if response.status_code == 204: 

55 return True 

56 # Log error message 

57 LOGGER.error(f"Failed to send Discord notification. Response: {response.text}") 

58 return False 

59 

60 

61def notify_discord(config, message, last_send=None, dry_run=False): 

62 """Send discord notification.""" 

63 sent_on = None 

64 webhook_url = config_parser.get_discord_webhook_url(config=config) 

65 username = config_parser.get_discord_username(config=config) 

66 

67 if last_send and last_send > datetime.datetime.now() - datetime.timedelta(hours=24): 

68 LOGGER.info("Throttling discord to once a day") 

69 sent_on = last_send 

70 elif webhook_url and username: 

71 sent_on = datetime.datetime.now() 

72 if not dry_run: 

73 # Post message to discord webhook using API 

74 if not post_message_to_discord(webhook_url, username, message): 

75 sent_on = None 

76 else: 

77 LOGGER.warning("Not sending 2FA notification because Discord is not configured.") 

78 return sent_on 

79 

80def notify_pushover(config, message, last_send=None, dry_run=False): 

81 """Send Pushover notification.""" 

82 sent_on = None 

83 user_key = config_parser.get_pushover_user_key(config=config) 

84 api_token = config_parser.get_pushover_api_token(config=config) 

85 

86 if last_send and last_send > datetime.datetime.now() - datetime.timedelta(hours=24): 

87 LOGGER.info("Throttling Pushover to once a day") 

88 sent_on = last_send 

89 elif user_key and api_token: 

90 sent_on = datetime.datetime.now() 

91 if not dry_run: 

92 if not post_message_to_pushover(api_token, user_key, message): 

93 sent_on = None 

94 else: 

95 LOGGER.warning("Not sending 2FA notification because Pushover is not configured.") 

96 return sent_on 

97 

98def post_message_to_pushover(api_token, user_key, message): 

99 """Post message to Pushover API.""" 

100 url = "https://api.pushover.net/1/messages.json" 

101 data = {"token": api_token, "user": user_key, "message": message} 

102 response = requests.post(url, data=data, timeout=10) 

103 if response.status_code == 200: 

104 return True 

105 LOGGER.error(f"Failed to send Pushover notification. Response: {response.text}") 

106 return False 

107 

108def send(config, username, last_send=None, dry_run=False, region="global"): 

109 """Send notifications.""" 

110 sent_on = None 

111 region_opt = "" 

112 if region != "global": 

113 region_opt = f"--region={region} " 

114 message = f"""Two-step authentication for iCloud Drive, Photos (Docker) is required. 

115 Please login to your server and authenticate. Please run - 

116 `docker exec -it --user=abc icloud /bin/sh -c 

117 "icloud --session-directory=/config/session_data {region_opt}--username={username}"`.""" 

118 subject = f"icloud-docker: Two step authentication is required for {username}" 

119 notify_telegram(config=config, message=message, last_send=last_send, dry_run=dry_run) 

120 notify_discord(config=config, message=message, last_send=last_send, dry_run=dry_run) 

121 notify_pushover(config=config, message=message, last_send=last_send, dry_run=dry_run) 

122 email = config_parser.get_smtp_email(config=config) 

123 to_email = config_parser.get_smtp_to_email(config=config) 

124 host = config_parser.get_smtp_host(config=config) 

125 port = config_parser.get_smtp_port(config=config) 

126 no_tls = config_parser.get_smtp_no_tls(config=config) 

127 username = config_parser.get_smtp_username(config=config) 

128 password = config_parser.get_smtp_password(config=config) 

129 

130 if last_send and last_send > datetime.datetime.now() - datetime.timedelta(hours=24): 

131 LOGGER.info("Throttling email to once a day") 

132 sent_on = last_send 

133 elif email and host and port: 

134 try: 

135 sent_on = datetime.datetime.now() 

136 if not dry_run: 

137 smtp = smtplib.SMTP(host, port) 

138 smtp.set_debuglevel(0) 

139 smtp.connect(host, port) 

140 if not no_tls: 

141 smtp.starttls() 

142 

143 if password: 

144 if username: 

145 smtp.login(username, password) 

146 else: 

147 smtp.login(email, password) 

148 

149 msg = build_message(email, to_email, message, subject) 

150 

151 smtp.sendmail(from_addr=email, to_addrs=to_email, msg=msg.as_string()) 

152 smtp.quit() 

153 except Exception as e: 

154 sent_on = None 

155 LOGGER.error(f"Failed to send email: {e!s}.") 

156 else: 

157 LOGGER.warning("Not sending 2FA notification because SMTP is not configured") 

158 

159 return sent_on 

160 

161 

162def build_message(email, to_email, message, subject): 

163 """Create email message.""" 

164 msg = Message(to=to_email) 

165 msg.sender = "icloud-docker <" + email + ">" 

166 msg.date = datetime.datetime.now().strftime("%d/%m/%Y %H:%M") 

167 msg.subject = subject 

168 msg.body = message 

169 

170 return msg