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 

80 

81def send(config, username, last_send=None, dry_run=False): 

82 """Send notifications.""" 

83 sent_on = None 

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

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

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

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

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

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

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

91 email = config_parser.get_smtp_email(config=config) 

92 to_email = config_parser.get_smtp_to_email(config=config) 

93 host = config_parser.get_smtp_host(config=config) 

94 port = config_parser.get_smtp_port(config=config) 

95 no_tls = config_parser.get_smtp_no_tls(config=config) 

96 username = config_parser.get_smtp_username(config=config) 

97 password = config_parser.get_smtp_password(config=config) 

98 

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

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

101 sent_on = last_send 

102 elif email and host and port: 

103 try: 

104 sent_on = datetime.datetime.now() 

105 if not dry_run: 

106 smtp = smtplib.SMTP(host, port) 

107 smtp.set_debuglevel(0) 

108 smtp.connect(host, port) 

109 if not no_tls: 

110 smtp.starttls() 

111 

112 if password: 

113 if username: 

114 smtp.login(username, password) 

115 else: 

116 smtp.login(email, password) 

117 

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

119 

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

121 smtp.quit() 

122 except Exception as e: 

123 sent_on = None 

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

125 else: 

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

127 

128 return sent_on 

129 

130 

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

132 """Create email message.""" 

133 msg = Message(to=to_email) 

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

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

136 msg.subject = subject 

137 msg.body = message 

138 

139 return msg