Coverage for modules\email\gmail_oauth.py: 14%

1""" 

2Gmail OAuth2 Service - Alternative to IMAP/SMTP for Gmail users 

3""" 

4import os 

5import json 

6from typing import Optional, Dict 

7 

8try: 

9 from google.oauth2.credentials import Credentials 

10 from google_auth_oauthlib.flow import InstalledAppFlow 

11 from google.auth.transport.requests import Request 

12 from googleapiclient.discovery import build 

13 GOOGLE_AVAILABLE = True 

14except ImportError: 

15 GOOGLE_AVAILABLE = False 

16 Credentials = None 

17 InstalledAppFlow = None 

18 

19 

20# If modifying these scopes, delete the file token.json. 

21SCOPES = [ 

22 'https://www.googleapis.com/auth/gmail.readonly', 

23 'https://www.googleapis.com/auth/gmail.send', 

24 'https://www.googleapis.com/auth/gmail.modify', 

25 'https://www.googleapis.com/auth/gmail.labels', 

26] 

27 

28 

29class GmailOAuthService: 

30 """ 

31 Service for Gmail OAuth2 authentication. 

32 Provides an alternative to IMAP/SMTP for users who prefer OAuth. 

33 """ 

34 

35 def __init__(self, credentials_path: str = None, token_path: str = None): 

36 """ 

37 Initialize Gmail OAuth service. 

38 

39 Args: 

40 credentials_path: Path to credentials.json (OAuth client config) 

41 token_path: Path to store user's access/refresh tokens 

42 """ 

43 if not GOOGLE_AVAILABLE: 

44 raise ImportError( 

45 "Google OAuth libraries not installed. " 

46 "Install with: pip install google-api-python-client google-auth-oauthlib" 

47 ) 

48 self.credentials_path = credentials_path or os.environ.get( 

49 'GOOGLE_OAUTH_CREDENTIALS', 

50 'credentials.json' 

51 ) 

52 self.token_path = token_path or os.environ.get( 

53 'GOOGLE_OAUTH_TOKEN', 

54 'token.json' 

55 ) 

56 self.creds: Optional[Credentials] = None 

57 self.service = None 

58 

59 def get_authorization_url(self, redirect_uri: str = 'http://localhost:8000/api/email/oauth/callback') -> str: 

60 """ 

61 Get the URL for user authorization. 

62 Returns the URL to redirect the user to. 

63 """ 

64 flow = InstalledAppFlow.from_client_secrets_file( 

65 self.credentials_path, SCOPES) 

66 flow.redirect_uri = redirect_uri 

67 

68 # This returns the URL to which we need to redirect the user 

69 auth_url, _ = flow.authorization_url(prompt='consent', access_type='offline') 

70 return auth_url 

71 

72 def exchange_code_for_tokens(self, code: str, redirect_uri: str = 'http://localhost:8000/api/email/oauth/callback') -> bool: 

73 """ 

74 Exchange authorization code for access/refresh tokens. 

75  

76 Returns: 

77 True if successful, False otherwise 

78 """ 

79 try: 

80 flow = InstalledAppFlow.from_client_secrets_file( 

81 self.credentials_path, SCOPES) 

82 flow.redirect_uri = redirect_uri 

83 

84 self.creds = flow.fetch_token(code=code) 

85 

86 # Save tokens 

87 with open(self.token_path, 'w') as token_file: 

88 token_file.write(json.dumps(self.creds)) 

89 

90 return True 

91 except Exception as e: 

92 print(f"Error exchanging code for tokens: {e}") 

93 return False 

94 

95 def load_tokens(self) -> bool: 

96 """Load tokens from file.""" 

97 try: 

98 if os.path.exists(self.token_path): 

99 with open(self.token_path, 'r') as f: 

100 self.creds = Credentials.from_authorized_user_info( 

101 json.load(f), SCOPES 

102 ) 

103 return True 

104 except Exception as e: 

105 print(f"Error loading tokens: {e}") 

106 return False 

107 

108 def save_tokens(self): 

109 """Save tokens to file.""" 

110 if self.creds and self.creds.token: 

111 with open(self.token_path, 'w') as f: 

112 f.write(json.dumps({ 

113 'token': self.creds.token, 

114 'refresh_token': self.creds.refresh_token, 

115 'token_uri': self.creds.token_uri, 

116 'client_id': self.creds.client_id, 

117 'client_secret': self.creds.client_secret, 

118 'scopes': self.creds.scopes, 

119 })) 

120 

121 def refresh_if_needed(self) -> bool: 

122 """Refresh access token if expired.""" 

123 if not self.creds or not self.creds.valid: 

124 if self.creds and self.creds.expired and self.creds.refresh_token: 

125 try: 

126 self.creds.refresh(Request()) 

127 self.save_tokens() 

128 return True 

129 except Exception as e: 

130 print(f"Error refreshing token: {e}") 

131 return False 

132 return self.creds and self.creds.valid 

133 

134 def connect(self) -> bool: 

135 """Connect to Gmail API.""" 

136 if not self.load_tokens(): 

137 return False 

138 

139 if not self.refresh_if_needed(): 

140 return False 

141 

142 try: 

143 self.service = build('gmail', 'v1', credentials=self.creds) 

144 return True 

145 except Exception as e: 

146 print(f"Error connecting to Gmail API: {e}") 

147 return False 

148 

149 def list_messages(self, label: str = 'INBOX', max_results: int = 50) -> list: 

150 """List messages from a label.""" 

151 if not self.service: 

152 if not self.connect(): 

153 return [] 

154 

155 try: 

156 results = self.service.users().messages().list( 

157 userId='me', 

158 labelIds=[label], 

159 maxResults=max_results 

160 ).execute() 

161 

162 messages = results.get('messages', []) 

163 

164 # Get details for each message 

165 detailed_messages = [] 

166 for msg_ref in messages: 

167 msg = self.service.users().messages().get( 

168 userId='me', 

169 id=msg_ref['id'], 

170 format='metadata' 

171 ).execute() 

172 

173 detailed_messages.append(self._parse_gmail_message(msg)) 

174 

175 return detailed_messages 

176 

177 except Exception as e: 

178 print(f"Error listing messages: {e}") 

179 return [] 

180 

181 def get_message(self, message_id: str) -> Optional[dict]: 

182 """Get a specific message by ID.""" 

183 if not self.service: 

184 if not self.connect(): 

185 return None 

186 

187 try: 

188 msg = self.service.users().messages().get( 

189 userId='me', 

190 id=message_id, 

191 format='full' 

192 ).execute() 

193 

194 return self._parse_gmail_message(msg, full=True) 

195 

196 except Exception as e: 

197 print(f"Error getting message: {e}") 

198 return None 

199 

200 def send_message(self, to: str, subject: str, body: str, is_html: bool = False) -> bool: 

201 """Send an email.""" 

202 if not self.service: 

203 if not self.connect(): 

204 return False 

205 

206 try: 

207 from email.mime.text import MIMEText 

208 import base64 

209 

210 msg = MIMEText(body, 'html' if is_html else 'plain') 

211 msg['to'] = to 

212 msg['subject'] = subject 

213 

214 # Encode and send 

215 raw = base64.urlsafe_b64encode(msg.as_bytes()).decode() 

216 self.service.users().messages().send( 

217 userId='me', 

218 body={'raw': raw} 

219 ).execute() 

220 

221 return True 

222 

223 except Exception as e: 

224 print(f"Error sending message: {e}") 

225 return False 

226 

227 def mark_as_read(self, message_id: str) -> bool: 

228 """Mark message as read.""" 

229 return self._modify_labels(message_id, remove_labels=['UNREAD']) 

230 

231 def mark_as_unread(self, message_id: str) -> bool: 

232 """Mark message as unread.""" 

233 return self._modify_labels(message_id, add_labels=['UNREAD']) 

234 

235 def trash_message(self, message_id: str) -> bool: 

236 """Move message to trash.""" 

237 if not self.service: 

238 if not self.connect(): 

239 return False 

240 

241 try: 

242 self.service.users().messages().trash( 

243 userId='me', 

244 id=message_id 

245 ).execute() 

246 return True 

247 except Exception as e: 

248 print(f"Error trashing message: {e}") 

249 return False 

250 

251 def _modify_labels(self, message_id: str, add_labels: list = None, remove_labels: list = None) -> bool: 

252 """Modify message labels.""" 

253 if not self.service: 

254 if not self.connect(): 

255 return False 

256 

257 try: 

258 body = {} 

259 if add_labels: 

260 body['addLabelIds'] = add_labels 

261 if remove_labels: 

262 body['removeLabelIds'] = remove_labels 

263 

264 if body: 

265 self.service.users().messages().modify( 

266 userId='me', 

267 id=message_id, 

268 body=body 

269 ).execute() 

270 return True 

271 except Exception as e: 

272 print(f"Error modifying labels: {e}") 

273 return False 

274 

275 def _parse_gmail_message(self, msg: dict, full: bool = False) -> dict: 

276 """Parse Gmail API message to our format.""" 

277 headers = msg['payload'].get('headers', []) 

278 

279 # Extract headers 

280 header_dict = {} 

281 for h in headers: 

282 header_dict[h['name'].lower()] = h['value'] 

283 

284 # Parse addresses 

285 from_header = header_dict.get('from', '') 

286 to_header = header_dict.get('to', '') 

287 

288 from_name, from_email = self._parse_address(from_header) 

289 

290 # Parse recipients 

291 to_addresses = [] 

292 for addr in to_header.split(','): 

293 name, email = self._parse_address(addr.strip()) 

294 to_addresses.append({'name': name, 'email': email}) 

295 

296 # Get body 

297 body_text, body_html = self._get_message_body(msg['payload']) 

298 

299 # Parse date 

300 sent_at = None 

301 date_str = header_dict.get('date', '') 

302 if date_str: 

303 try: 

304 from email.utils import parsedate_to_datetime 

305 sent_at = parsedate_to_datetime(date_str) 

306 except: 

307 pass 

308 

309 result = { 

310 'message_id': msg.get('id'), 

311 'thread_id': msg.get('threadId'), 

312 'subject': header_dict.get('subject', ''), 

313 'from_address': from_email, 

314 'from_name': from_name, 

315 'to_addresses': to_addresses, 

316 'sent_at': sent_at, 

317 'snippet': msg.get('snippet', ''), 

318 'is_read': 'UNREAD' not in msg.get('labelIds', []), 

319 'labels': msg.get('labelIds', []), 

320 } 

321 

322 if full: 

323 result['body_text'] = body_text 

324 result['body_html'] = body_html 

325 

326 return result 

327 

328 def _parse_address(self, address: str) -> tuple: 

329 """Parse email address to (name, email).""" 

330 if not address: 

331 return None, None 

332 

333 try: 

334 from email.utils import parseaddr 

335 name, email = parseaddr(address) 

336 return (name if name else None), (email if email else address) 

337 except: 

338 return None, address 

339 

340 def _get_message_body(self, payload: dict) -> tuple: 

341 """Extract body from message payload.""" 

342 text_body = '' 

343 html_body = '' 

344 

345 if 'parts' in payload: 

346 for part in payload['parts']: 

347 if part['mimeType'] == 'text/plain': 

348 data = part.get('body', {}).get('data', '') 

349 if data: 

350 text_body = base64.urlsafe_b64decode(data).decode('utf-8', errors='replace') 

351 elif part['mimeType'] == 'text/html': 

352 data = part.get('body', {}).get('data', '') 

353 if data: 

354 html_body = base64.urlsafe_b64decode(data).decode('utf-8', errors='replace') 

355 elif 'parts' in part: 

356 # Recurse into nested parts 

357 sub_text, sub_html = self._get_message_body(part) 

358 text_body = text_body or sub_text 

359 html_body = html_body or sub_html 

360 else: 

361 data = payload.get('body', {}).get('data', '') 

362 if data: 

363 decoded = base64.urlsafe_b64decode(data).decode('utf-8', errors='replace') 

364 if payload.get('mimeType') == 'text/html': 

365 html_body = decoded 

366 else: 

367 text_body = decoded 

368 

369 return text_body, html_body