Coverage for youversion/core/authenticator.py: 100%

1"""Authentication handler for YouVersion API.""" 

2 

3import base64 

4import os 

5 

6import httpx 

7import jwt 

8from dotenv import load_dotenv 

9 

10from ..config import Config 

11from .interfaces import IAuthenticator 

12 

13 

14class Authenticator(IAuthenticator): 

15 """Handles authentication for YouVersion API using OAuth2.""" 

16 

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

18 """Initialize authenticator with credentials. 

19 

20 Args: 

21 username: Username for authentication 

22 password: Password for authentication 

23 """ 

24 load_dotenv() 

25 

26 self.username = username or os.getenv("YOUVERSION_USERNAME") 

27 self.password = password or os.getenv("YOUVERSION_PASSWORD") 

28 

29 if not self.username or not self.password: 

30 raise ValueError( 

31 "Username and password must be provided either as arguments " 

32 "or as YOUVERSION_USERNAME and YOUVERSION_PASSWORD environment variables" 

33 ) 

34 self.user_id = None 

35 self.access_token = None 

36 

37 async def authenticate(self, username: str, password: str) -> httpx.AsyncClient: 

38 """Authenticate using OAuth2 and return an HTTP client with Bearer token. 

39 

40 Args: 

41 username: Username for authentication 

42 password: Password for authentication 

43 

44 Returns: 

45 Authenticated httpx.AsyncClient with Bearer token 

46 """ 

47 # Get OAuth2 token 

48 token = await self._get_oauth2_token(username, password) 

49 

50 # Create HTTP client with Bearer token and default headers 

51 headers = {**Config.DEFAULT_HEADERS, "Authorization": f"Bearer {token}"} 

52 client = httpx.AsyncClient(headers=headers, timeout=Config.HTTP_TIMEOUT) 

53 

54 return client 

55 

56 async def _get_oauth2_token(self, username: str, password: str) -> str: 

57 """Get OAuth2 access token from YouVersion API. 

58 

59 Args: 

60 username: Username for authentication 

61 password: Password for authentication 

62 

63 Returns: 

64 OAuth2 access token 

65 

66 Raises: 

67 httpx.HTTPStatusError: If authentication fails 

68 """ 

69 async with httpx.AsyncClient(timeout=Config.HTTP_TIMEOUT) as client: 

70 response = await client.post( 

71 Config.AUTH_URL, 

72 data={ 

73 "client_id": base64.b64decode(Config.CLIENT_ID).decode(), 

74 "client_secret": base64.b64decode(Config.CLIENT_SECRET).decode(), 

75 "grant_type": "password", 

76 "username": username, 

77 "password": password, 

78 }, 

79 ) 

80 response.raise_for_status() 

81 token_data = response.json() 

82 if hasattr(token_data, "__await__"): 

83 token_data = await token_data 

84 

85 # Decode JWT token to extract user information 

86 try: 

87 decoded_token = jwt.decode( 

88 token_data["access_token"], 

89 Config.CLIENT_SECRET, 

90 algorithms=["HS256"], 

91 ) 

92 # Try user_id first, then sub (OAuth2 standard) 

93 self.user_id = decoded_token.get("user_id") or decoded_token.get("sub") 

94 except jwt.DecodeError: 

95 # If token cannot be decoded/verified, 

96 # try decoding without verification 

97 decoded_token = jwt.decode( 

98 token_data["access_token"], 

99 options={"verify_signature": False}, 

100 ) 

101 # Try user_id first, then sub (OAuth2 standard) 

102 self.user_id = decoded_token.get("user_id") or decoded_token.get("sub") 

103 except jwt.InvalidTokenError: 

104 # If decoding fails entirely, continue without user_id 

105 self.user_id = None 

106 

107 self.access_token = token_data["access_token"] 

108 return token_data["access_token"]