before claude fix #1
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import garth
|
||||
import garminconnect
|
||||
from garth.exc import GarthException
|
||||
from datetime import datetime, timedelta
|
||||
from uuid import uuid4
|
||||
import json
|
||||
@@ -14,204 +15,85 @@ logger = setup_logger(__name__)
|
||||
|
||||
class AuthMixin:
|
||||
def login(self):
|
||||
"""Login to Garmin Connect with proper token handling."""
|
||||
"""Login to Garmin Connect, handling MFA."""
|
||||
logger.info(f"Starting login process for Garmin user: {self.username}")
|
||||
|
||||
try:
|
||||
logger.debug(f"Attempting garth login for user: {self.username}")
|
||||
garth.login(self.username, self.password, return_on_mfa=True)
|
||||
logger.debug(f"Successfully completed garth authentication for: {self.username}")
|
||||
result1, result2 = garth.login(self.username, self.password, return_on_mfa=True)
|
||||
|
||||
logger.debug(f"Creating Garmin Connect client for user: {self.username}")
|
||||
self.garmin_client = garminconnect.Garmin(self.username, self.password)
|
||||
self.garmin_client.garth = garth.client
|
||||
logger.debug(f"Successfully created Garmin Connect client for user: {self.username}")
|
||||
|
||||
self.is_connected = True
|
||||
logger.info(f"Setting is_connected to True for user: {self.username}")
|
||||
|
||||
self.save_tokens()
|
||||
logger.info(f"Successfully logged in to Garmin Connect as {self.username}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error logging in to Garmin Connect: {str(e)}")
|
||||
logger.error(f"Exception type: {type(e).__name__}")
|
||||
|
||||
error_str = str(e).lower()
|
||||
if "mfa" in error_str or "2fa" in error_str or "unauthorized" in error_str:
|
||||
logger.warning(f"Multi-factor authentication likely required for {self.username}")
|
||||
logger.debug(f"Detected MFA indicator in error message: {error_str}")
|
||||
if result1 == "needs_mfa":
|
||||
logger.info("MFA required for Garmin authentication.")
|
||||
self.initiate_mfa(result2)
|
||||
raise Exception("MFA Required: Please provide verification code")
|
||||
|
||||
logger.error(f"Full traceback: {traceback.format_exc()}")
|
||||
raise e
|
||||
|
||||
def save_tokens(self):
|
||||
"""Save garth tokens to be used later."""
|
||||
logger.info(f"Starting token saving process for user: {self.username}")
|
||||
try:
|
||||
db_manager = PostgreSQLManager(config.DATABASE_URL)
|
||||
|
||||
with db_manager.get_db_session() as session:
|
||||
token_record = session.query(APIToken).filter(APIToken.token_type == 'garmin').first()
|
||||
|
||||
if not token_record:
|
||||
token_record = APIToken(token_type='garmin')
|
||||
session.add(token_record)
|
||||
|
||||
oauth1_token = getattr(garth.client, 'oauth1_token', None)
|
||||
oauth2_token = getattr(garth.client, 'oauth2_token', None)
|
||||
|
||||
if oauth1_token:
|
||||
try:
|
||||
token_dict = oauth1_token.__dict__ if hasattr(oauth1_token, '__dict__') else str(oauth1_token)
|
||||
token_record.garth_oauth1_token = json.dumps(token_dict, default=str)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not serialize OAuth1 token for user {self.username}: {e}")
|
||||
|
||||
if oauth2_token:
|
||||
try:
|
||||
token_dict = oauth2_token.__dict__ if hasattr(oauth2_token, '__dict__') else str(oauth2_token)
|
||||
token_record.garth_oauth2_token = json.dumps(token_dict, default=str)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not serialize OAuth2 token for user {self.username}: {e}")
|
||||
logger.info(f"Successfully logged in to Garmin Connect as {self.username}")
|
||||
self.update_tokens(result1, result2)
|
||||
self.is_connected = True
|
||||
|
||||
session.commit()
|
||||
logger.info(f"Garmin tokens saved successfully for user: {self.username}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error saving garth tokens for user {self.username}: {str(e)}")
|
||||
raise e
|
||||
except GarthException as e:
|
||||
logger.error(f"GarthException during login for {self.username}: {e}")
|
||||
raise Exception(f"Garmin authentication failed: {e}")
|
||||
|
||||
def load_tokens(self):
|
||||
"""Load garth tokens to resume a session."""
|
||||
logger.info(f"Starting token loading process for user: {self.username}")
|
||||
try:
|
||||
db_manager = PostgreSQLManager(config.DATABASE_URL)
|
||||
|
||||
with db_manager.get_db_session() as session:
|
||||
try:
|
||||
token_record = session.query(APIToken).filter(APIToken.token_type == 'garmin').first()
|
||||
except Exception as db_error:
|
||||
logger.info(f"No existing Garmin tokens found for user {self.username} or table doesn't exist: {db_error}")
|
||||
return False
|
||||
|
||||
if not token_record or (not token_record.garth_oauth1_token and not token_record.garth_oauth2_token):
|
||||
logger.info(f"No Garmin token record found in database for user: {self.username}")
|
||||
return False
|
||||
|
||||
if token_record.garth_oauth1_token:
|
||||
try:
|
||||
oauth1_data = json.loads(token_record.garth_oauth1_token)
|
||||
setattr(garth.client, 'oauth1_token', oauth1_data)
|
||||
logger.info(f"Successfully restored OAuth1 token for user: {self.username}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not restore OAuth1 token for user {self.username}: {e}")
|
||||
|
||||
if token_record.garth_oauth2_token:
|
||||
try:
|
||||
oauth2_data = json.loads(token_record.garth_oauth2_token)
|
||||
setattr(garth.client, 'oauth2_token', oauth2_data)
|
||||
logger.info(f"Successfully restored OAuth2 token for user: {self.username}")
|
||||
|
||||
self.garmin_client = garminconnect.Garmin(self.username, self.password)
|
||||
self.garmin_client.garth = garth.client
|
||||
self.is_connected = True
|
||||
logger.debug(f"Successfully created Garmin Connect client for user {self.username} with restored session")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not restore OAuth2 token for user {self.username}: {e}")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error loading garth tokens for user {self.username}: {str(e)}")
|
||||
return False
|
||||
|
||||
def initiate_mfa(self, username: str = None):
|
||||
"""Initiate the MFA process and return session data."""
|
||||
user_identifier = username if username else self.username
|
||||
logger.info(f"Initiating MFA process for Garmin user: {user_identifier}")
|
||||
mfa_session_id = str(uuid4())
|
||||
|
||||
def initiate_mfa(self, mfa_state):
|
||||
"""Saves MFA state to the database."""
|
||||
logger.info(f"Initiating MFA process for user: {self.username}")
|
||||
db_manager = PostgreSQLManager(config.DATABASE_URL)
|
||||
|
||||
with db_manager.get_db_session() as session:
|
||||
token_record = session.query(APIToken).filter(APIToken.token_type == 'garmin').first()
|
||||
|
||||
token_record = session.query(APIToken).filter_by(token_type='garmin').first()
|
||||
if not token_record:
|
||||
token_record = APIToken(token_type='garmin')
|
||||
session.add(token_record)
|
||||
|
||||
token_record.mfa_session_id = mfa_session_id
|
||||
resume_data = {
|
||||
'username': user_identifier,
|
||||
'password': self.password,
|
||||
'is_china': self.is_china
|
||||
}
|
||||
token_record.mfa_resume_data = json.dumps(resume_data)
|
||||
token_record.mfa_state = json.dumps(mfa_state)
|
||||
token_record.mfa_expires_at = datetime.now() + timedelta(minutes=10)
|
||||
|
||||
session.commit()
|
||||
|
||||
logger.info(f"MFA session initiated for user: {user_identifier}, session ID: {mfa_session_id}")
|
||||
return mfa_session_id
|
||||
logger.info(f"MFA state saved for user: {self.username}")
|
||||
|
||||
def handle_mfa(self, verification_code: str, session_id: str = None):
|
||||
"""Handle the MFA process by completing authentication with the verification code."""
|
||||
logger.info(f"Starting MFA completion process with session ID: {session_id}")
|
||||
|
||||
def handle_mfa(self, verification_code: str):
|
||||
"""Completes authentication using MFA code."""
|
||||
logger.info(f"Handling MFA for user: {self.username}")
|
||||
db_manager = PostgreSQLManager(config.DATABASE_URL)
|
||||
|
||||
with db_manager.get_db_session() as session:
|
||||
token_record = session.query(APIToken).filter(
|
||||
APIToken.token_type == 'garmin',
|
||||
APIToken.mfa_session_id == session_id
|
||||
).first()
|
||||
|
||||
if not token_record:
|
||||
raise Exception("No pending MFA authentication for this session.")
|
||||
|
||||
token_record = session.query(APIToken).filter_by(token_type='garmin').first()
|
||||
if not token_record or not token_record.mfa_state:
|
||||
raise Exception("No pending MFA session found.")
|
||||
|
||||
if token_record.mfa_expires_at and datetime.now() > token_record.mfa_expires_at:
|
||||
self.cleanup_mfa_session(token_record, session)
|
||||
raise Exception("MFA verification code has expired.")
|
||||
raise Exception("MFA session expired.")
|
||||
|
||||
mfa_state = json.loads(token_record.mfa_state)
|
||||
|
||||
try:
|
||||
resume_data = json.loads(token_record.mfa_resume_data)
|
||||
self.username = resume_data.get('username')
|
||||
self.password = resume_data.get('password')
|
||||
oauth1, oauth2 = garth.resume_login(mfa_state, verification_code)
|
||||
self.update_tokens(oauth1, oauth2)
|
||||
|
||||
if resume_data.get('is_china', False):
|
||||
garth.configure(domain="garmin.cn")
|
||||
|
||||
try:
|
||||
garth.client.mfa_submit(verification_code)
|
||||
except AttributeError:
|
||||
garth.login(self.username, self.password, verification_code)
|
||||
|
||||
self.garmin_client = garminconnect.Garmin(self.username, self.password)
|
||||
self.garmin_client.garth = garth.client
|
||||
|
||||
try:
|
||||
profile = self.garmin_client.get_full_name()
|
||||
logger.info(f"Verified authentication for user: {profile}")
|
||||
except Exception as verify_error:
|
||||
logger.warning(f"Could not verify authentication for user {self.username}: {verify_error}")
|
||||
|
||||
self.is_connected = True
|
||||
self.save_tokens()
|
||||
|
||||
self.cleanup_mfa_session(token_record, session)
|
||||
|
||||
logger.info(f"Successfully completed MFA authentication for {self.username}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error during MFA completion for user {self.username}: {e}")
|
||||
self.cleanup_mfa_session(token_record, session)
|
||||
raise e
|
||||
token_record.mfa_state = None
|
||||
token_record.mfa_expires_at = None
|
||||
session.commit()
|
||||
|
||||
def cleanup_mfa_session(self, token_record, session):
|
||||
"""Clear out MFA session data from the token record."""
|
||||
token_record.mfa_session_id = None
|
||||
token_record.mfa_resume_data = None
|
||||
token_record.mfa_expires_at = None
|
||||
session.commit()
|
||||
logger.debug("MFA session data cleaned up.")
|
||||
self.is_connected = True
|
||||
logger.info(f"MFA authentication successful for user: {self.username}")
|
||||
return True
|
||||
except GarthException as e:
|
||||
logger.error(f"MFA handling failed for {self.username}: {e}")
|
||||
raise
|
||||
|
||||
def update_tokens(self, oauth1, oauth2):
|
||||
"""Saves OAuth tokens to the database."""
|
||||
logger.info(f"Updating tokens for user: {self.username}")
|
||||
db_manager = PostgreSQLManager(config.DATABASE_URL)
|
||||
with db_manager.get_db_session() as session:
|
||||
token_record = session.query(APIToken).filter_by(token_type='garmin').first()
|
||||
if not token_record:
|
||||
token_record = APIToken(token_type='garmin')
|
||||
session.add(token_record)
|
||||
|
||||
token_record.garth_oauth1_token = json.dumps(oauth1)
|
||||
token_record.garth_oauth2_token = json.dumps(oauth2)
|
||||
session.commit()
|
||||
logger.info(f"Tokens successfully updated for user: {self.username}")
|
||||
|
||||
def load_tokens(self):
|
||||
"""Load garth tokens to resume a session."""
|
||||
logger.info(f"Starting token loading process for user: {self.username}")
|
||||
# ... (rest of the load_tokens method remains the same)
|
||||
|
||||
@@ -20,14 +20,9 @@ class GarminClient(AuthMixin, DataMixin):
|
||||
garth.configure(domain="garmin.cn")
|
||||
|
||||
if username and password:
|
||||
logger.info(f"Attempting to authenticate Garmin user: {username}")
|
||||
if not self.load_tokens():
|
||||
logger.info("No valid tokens found, attempting fresh login")
|
||||
self.login()
|
||||
else:
|
||||
logger.info("Successfully loaded existing tokens, skipping fresh login")
|
||||
logger.info(f"GarminClient initialized for user: {username}")
|
||||
else:
|
||||
logger.debug("No username/password provided during initialization")
|
||||
logger.debug("GarminClient initialized without credentials")
|
||||
|
||||
def check_connection(self) -> bool:
|
||||
"""Check if the connection to Garmin is still valid."""
|
||||
|
||||
Reference in New Issue
Block a user