Coverage for paperless_mail/oauth.py: 39%

1import asyncio 

2import logging 

3import secrets 

4from datetime import timedelta 

5 

6from django.conf import settings 

7from django.utils import timezone 

8from httpx_oauth.clients.google import GoogleOAuth2 

9from httpx_oauth.clients.microsoft import MicrosoftGraphOAuth2 

10from httpx_oauth.oauth2 import OAuth2Token 

11from httpx_oauth.oauth2 import RefreshTokenError 

12 

13from paperless_mail.models import MailAccount 

14 

15 

16class PaperlessMailOAuth2Manager: 

17 def __init__(self, state: str | None = None): 

18 self._gmail_client = None 

19 self._outlook_client = None 

20 self.state = state if state is not None else secrets.token_urlsafe(32) 

21 

22 @property 

23 def gmail_client(self) -> GoogleOAuth2: 

24 if self._gmail_client is None: 

25 self._gmail_client = GoogleOAuth2( 

26 settings.GMAIL_OAUTH_CLIENT_ID, 

27 settings.GMAIL_OAUTH_CLIENT_SECRET, 

28 ) 

29 return self._gmail_client 

30 

31 @property 

32 def outlook_client(self) -> MicrosoftGraphOAuth2: 

33 if self._outlook_client is None: 

34 self._outlook_client = MicrosoftGraphOAuth2( 

35 settings.OUTLOOK_OAUTH_CLIENT_ID, 

36 settings.OUTLOOK_OAUTH_CLIENT_SECRET, 

37 ) 

38 return self._outlook_client 

39 

40 @property 

41 def oauth_callback_url(self) -> str: 

42 return f"{settings.OAUTH_CALLBACK_BASE_URL if settings.OAUTH_CALLBACK_BASE_URL is not None else settings.PAPERLESS_URL}{settings.BASE_URL}api/oauth/callback/" 

43 

44 @property 

45 def oauth_redirect_url(self) -> str: 

46 return f"{'http://localhost:4200/' if settings.DEBUG else settings.BASE_URL}mail" # e.g. "http://localhost:4200/mail" or "/mail" 

47 

48 def get_gmail_authorization_url(self) -> str: 

49 return asyncio.run( 

50 self.gmail_client.get_authorization_url( 

51 redirect_uri=self.oauth_callback_url, 

52 scope=["https://mail.google.com/"], 

53 extras_params={"prompt": "consent", "access_type": "offline"}, 

54 state=self.state, 

55 ), 

56 ) 

57 

58 def get_outlook_authorization_url(self) -> str: 

59 return asyncio.run( 

60 self.outlook_client.get_authorization_url( 

61 redirect_uri=self.oauth_callback_url, 

62 scope=[ 

63 "offline_access", 

64 "https://outlook.office.com/IMAP.AccessAsUser.All", 

65 ], 

66 state=self.state, 

67 ), 

68 ) 

69 

70 def get_gmail_access_token(self, code: str) -> OAuth2Token: 

71 return asyncio.run( 

72 self.gmail_client.get_access_token( 

73 code=code, 

74 redirect_uri=self.oauth_callback_url, 

75 ), 

76 ) 

77 

78 def get_outlook_access_token(self, code: str) -> OAuth2Token: 

79 return asyncio.run( 

80 self.outlook_client.get_access_token( 

81 code=code, 

82 redirect_uri=self.oauth_callback_url, 

83 ), 

84 ) 

85 

86 def refresh_account_oauth_token(self, account: MailAccount) -> bool: 

87 """ 

88 Refreshes the oauth token for the given mail account. 

89 """ 

90 logger = logging.getLogger("paperless_mail") 

91 logger.debug(f"Attempting to refresh oauth token for account {account}") 

92 try: 

93 result: OAuth2Token 

94 if account.account_type == MailAccount.MailAccountType.GMAIL_OAUTH: 

95 result = asyncio.run( 

96 self.gmail_client.refresh_token( 

97 refresh_token=account.refresh_token, 

98 ), 

99 ) 

100 elif account.account_type == MailAccount.MailAccountType.OUTLOOK_OAUTH: 

101 result = asyncio.run( 

102 self.outlook_client.refresh_token( 

103 refresh_token=account.refresh_token, 

104 ), 

105 ) 

106 if "refresh_token" in result: 

107 # Outlook returns a new refresh token on refresh, Gmail does not 

108 account.refresh_token = result["refresh_token"] 

109 account.password = result["access_token"] 

110 account.expiration = timezone.now() + timedelta( 

111 seconds=result["expires_in"], 

112 ) 

113 account.save() 

114 logger.debug(f"Successfully refreshed oauth token for account {account}") 

115 return True 

116 except RefreshTokenError as e: 

117 logger.error(f"Failed to refresh oauth token for account {account}: {e}") 

118 return False 

119 

120 def validate_state(self, state: str) -> bool: 

121 return settings.DEBUG or (len(state) > 0 and state == self.state)