#!/usr/bin/env python3 """ Dexcom Sandbox OAuth Integration Complete implementation for Dexcom Sandbox authentication with user selection """ import os import requests import urllib.parse import json import secrets import webbrowser import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple, Any from dataclasses import dataclass # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Load environment variables from dotenv import load_dotenv load_dotenv() # Dexcom Sandbox Configuration CLIENT_ID = os.getenv("DEXCOM_CLIENT_ID", "mLElKHKRwRDVUrAOPBzktFGY7qkTc7Zm") CLIENT_SECRET = os.getenv("DEXCOM_CLIENT_SECRET", "HmFpgyVweuwKrQpf") REDIRECT_URI = "http://localhost:7860/callback" # Dexcom Sandbox Users (selection-based, no passwords) SANDBOX_USERS = { "user1": "SandboxUser1", "user2": "SandboxUser2", "user3": "SandboxUser3", "user4": "SandboxUser4", "user5": "SandboxUser5", "user6": "SandboxUser6 (Dexcom G6)", "user7": "SandboxUser7 (Dexcom G7)", "user8": "SandboxUser8" } # Dexcom API Endpoints - Following Official Documentation # Based on https://developer.dexcom.com/docs/dexcomv2/endpoint-overview/ and v3 # OAuth endpoints are shared between v2 and v3 (always v2 OAuth) OAUTH_ENDPOINTS = { "login": "https://api.dexcom.com/v2/oauth2/login", # Production OAuth login "token": "https://api.dexcom.com/v2/oauth2/token", # Production OAuth token "sandbox_login": "https://developer.dexcom.com/sandbox-login" # Special sandbox login } # API Base URLs per official documentation API_BASE_URLS = { "production": "https://api.dexcom.com", "sandbox": "https://sandbox-api.dexcom.com" } # Recommended configuration (Sandbox login + Production OAuth + Sandbox API) DEFAULT_ENDPOINTS = { "login": "https://developer.dexcom.com/sandbox-login", # Special sandbox login "token": "https://api.dexcom.com/v2/oauth2/token", # Production OAuth (works for sandbox too) "api_v2": "https://sandbox-api.dexcom.com/v2", # Sandbox API v2 "api_v3": "https://sandbox-api.dexcom.com/v3" # Sandbox API v3 } # Alternative configurations for troubleshooting ENDPOINT_CONFIGURATIONS = [ { "name": "Sandbox Login + Production OAuth + Sandbox API (Recommended)", "login": "https://developer.dexcom.com/sandbox-login", "token": "https://api.dexcom.com/v2/oauth2/token", "api_v2": "https://sandbox-api.dexcom.com/v2", "api_v3": "https://sandbox-api.dexcom.com/v3" }, { "name": "All Production OAuth + Production API", "login": "https://api.dexcom.com/v2/oauth2/login", "token": "https://api.dexcom.com/v2/oauth2/token", "api_v2": "https://api.dexcom.com/v2", "api_v3": "https://api.dexcom.com/v3" }, { "name": "All Sandbox (May not work)", "login": "https://sandbox-api.dexcom.com/v2/oauth2/login", "token": "https://sandbox-api.dexcom.com/v2/oauth2/token", "api_v2": "https://sandbox-api.dexcom.com/v2", "api_v3": "https://sandbox-api.dexcom.com/v3" } ] @dataclass class DexcomSandboxUser: """Profile for Dexcom Sandbox User""" name: str = "Dexcom Sandbox User" age: int = 35 device_type: str = "Dexcom G6 (Sandbox)" username: str = "sandbox_user" password: str = "selection_based" description: str = "Dexcom Sandbox User with OAuth authentication" diabetes_type: str = "Type 1" years_with_diabetes: int = 8 typical_glucose_pattern: str = "sandbox_data" auth_type: str = "dexcom_sandbox" class DexcomSandboxOAuth: """ Dexcom Sandbox OAuth implementation with user selection Flow: 1. Generate auth URL โ†’ developer.dexcom.com/sandbox-login 2. User selects sandbox user from dropdown (no password) 3. Get authorization code from callback 4. Exchange code for token โ†’ sandbox-api.dexcom.com/v2/oauth2/token 5. Use token for API calls โ†’ sandbox-api.dexcom.com/v2/... """ def __init__(self, api_version: str = "v3"): self.client_id = CLIENT_ID self.client_secret = CLIENT_SECRET self.redirect_uri = REDIRECT_URI self.api_version = api_version # "v2" or "v3" # OAuth state self.access_token = None self.refresh_token = None self.token_expires_at = None self.state = None # Use recommended configuration (sandbox login + production OAuth + sandbox API) self.working_endpoints = DEFAULT_ENDPOINTS.copy() logger.info(f"โœ… Dexcom Sandbox OAuth initialized (API {api_version})") logger.info(f" Client ID: {self.client_id[:8]}...") logger.info(f" Redirect URI: {self.redirect_uri}") if api_version == "v3": logger.info(" Using API v3 for G6, G7, ONE, ONE+ device support") else: logger.info(" Using API v2 for G5, G6 device support") def get_api_base_url(self) -> str: """Get the correct API base URL for the current version""" if self.api_version == "v3": return self.working_endpoints["api_v3"] else: return self.working_endpoints["api_v2"] def generate_auth_url(self) -> str: """Generate OAuth authorization URL for Dexcom Sandbox""" # Generate secure state parameter self.state = secrets.token_urlsafe(32) # OAuth parameters params = { 'client_id': self.client_id, 'redirect_uri': self.redirect_uri, 'response_type': 'code', 'scope': 'offline_access', 'state': self.state } # Build auth URL using current working endpoints query_string = urllib.parse.urlencode(params) auth_url = f"{self.working_endpoints['login']}?{query_string}" logger.info(f"๐Ÿ”— Generated auth URL: {auth_url}") return auth_url def start_oauth_flow(self) -> Dict[str, Any]: """Start the Dexcom Sandbox OAuth flow""" try: auth_url = self.generate_auth_url() # Try to open browser automatically try: webbrowser.open(auth_url) browser_opened = True logger.info("โœ… Browser opened automatically") except Exception as e: browser_opened = False logger.warning(f"โš ๏ธ Could not open browser: {e}") return { "success": True, "auth_url": auth_url, "browser_opened": browser_opened, "state": self.state, "instructions": self._get_oauth_instructions(auth_url, browser_opened) } except Exception as e: logger.error(f"Failed to start OAuth flow: {e}") return { "success": False, "error": f"Failed to start OAuth: {str(e)}" } def _get_oauth_instructions(self, auth_url: str, browser_opened: bool) -> str: """Generate user-friendly OAuth instructions""" browser_status = "โœ… Browser opened automatically" if browser_opened else "โš ๏ธ Please open URL manually" return f""" ๐Ÿš€ **Dexcom Sandbox OAuth Started** {browser_status} **๐Ÿ“‹ Step-by-Step Instructions:** 1. ๐ŸŒ Browser should open to: {auth_url} 2. ๐Ÿ‘ฅ **Select a Sandbox User** from the dropdown: โ€ข **SandboxUser6** - Dexcom G6 device (recommended) โ€ข **SandboxUser7** - Dexcom G7 device โ€ข **SandboxUser1-8** - Various test scenarios 3. โœ… Click "Authorize" to grant access 4. โŒ **You will get a 404 error - THIS IS EXPECTED!** 5. ๐Ÿ“‹ **Copy the COMPLETE URL** from your browser's address bar 6. ๐Ÿ“ฅ Paste the URL below and click "Complete OAuth" **๐Ÿ“ฑ Example callback URL:** `http://localhost:7860/callback?code=ABC123XYZ&state=your_state_here` **๐ŸŽฏ Important:** - No password needed - just select a user and authorize - Copy the **entire URL** (not just the code part) - SandboxUser6 = Dexcom G6 device data (most common) """ def complete_oauth(self, callback_url: str) -> Dict[str, Any]: """Complete OAuth by processing callback URL and exchanging code for token""" if not callback_url or not callback_url.strip(): return { "success": False, "error": "Please provide the callback URL from your browser" } try: # Extract authorization code from callback URL auth_code = self._extract_auth_code(callback_url) if not auth_code: return { "success": False, "error": "Could not extract authorization code from URL" } # Validate state parameter callback_state = self._extract_state(callback_url) if callback_state != self.state: logger.warning(f"State mismatch: expected {self.state}, got {callback_state}") # Continue anyway for sandbox testing # Exchange code for tokens token_result = self._exchange_code_for_tokens(auth_code) if token_result["success"]: logger.info("โœ… Dexcom Sandbox OAuth completed successfully") return { "success": True, "message": "โœ… Dexcom Sandbox authentication successful", "access_token": self.access_token, "token_expires_at": self.token_expires_at, "user_profile": DexcomSandboxUser() } else: return token_result except Exception as e: logger.error(f"OAuth completion failed: {e}") return { "success": False, "error": f"OAuth completion failed: {str(e)}" } def _extract_auth_code(self, callback_url: str) -> Optional[str]: """Extract authorization code from callback URL""" try: parsed_url = urllib.parse.urlparse(callback_url) query_params = urllib.parse.parse_qs(parsed_url.query) if 'code' in query_params: code = query_params['code'][0] logger.info(f"โœ… Extracted auth code: {code[:15]}...") return code else: logger.error("No 'code' parameter found in callback URL") return None except Exception as e: logger.error(f"Error extracting auth code: {e}") return None def _extract_state(self, callback_url: str) -> Optional[str]: """Extract state parameter from callback URL""" try: parsed_url = urllib.parse.urlparse(callback_url) query_params = urllib.parse.parse_qs(parsed_url.query) if 'state' in query_params: return query_params['state'][0] else: return None except Exception as e: logger.error(f"Error extracting state: {e}") return None def _exchange_code_for_tokens(self, auth_code: str) -> Dict[str, Any]: """Exchange authorization code for access token following Dexcom API v3 guidelines""" # Try sandbox configuration first, then alternatives configurations_to_try = [self.working_endpoints] + ENDPOINT_CONFIGURATIONS for i, endpoint_config in enumerate(configurations_to_try): endpoint_name = endpoint_config.get('name', f'Sandbox Config {i+1}') token_url = endpoint_config["token"] logger.info(f"๐Ÿ”„ Attempting token exchange #{i+1}: {endpoint_name}") logger.info(f" Token URL: {token_url}") # Token exchange data per Dexcom OAuth2 spec data = { 'client_id': self.client_id, 'client_secret': self.client_secret, 'code': auth_code, 'grant_type': 'authorization_code', 'redirect_uri': self.redirect_uri } # Headers per Dexcom API guidelines headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json', 'User-Agent': 'GlycoAI-DexcomSandbox/1.0', 'Cache-Control': 'no-cache' } try: logger.info(f"๐Ÿ“ค Auth code: {auth_code[:15]}...") response = requests.post(token_url, data=data, headers=headers, timeout=30) logger.info(f"๐Ÿ“จ Response status: {response.status_code}") logger.info(f"๐Ÿ“จ Response headers: {dict(response.headers)}") if response.status_code == 200: try: token_data = response.json() except json.JSONDecodeError: logger.error(f"โŒ Invalid JSON response: {response.text}") continue # Store tokens self.access_token = token_data.get('access_token') self.refresh_token = token_data.get('refresh_token') if not self.access_token: logger.error(f"โŒ No access_token in response: {token_data}") continue # Calculate expiration expires_in = token_data.get('expires_in', 3600) # Default 1 hour self.token_expires_at = datetime.now() + timedelta(seconds=expires_in) logger.info(f"โœ… Token exchange successful with {endpoint_name}!") logger.info(f" Access token: {self.access_token[:25]}...") logger.info(f" Token type: {token_data.get('token_type', 'Bearer')}") logger.info(f" Expires in: {expires_in} seconds ({expires_in/3600:.1f} hours)") logger.info(f" Scope: {token_data.get('scope', 'offline_access')}") # Update working endpoints to use the successful configuration self.working_endpoints = endpoint_config.copy() return { "success": True, "access_token": self.access_token, "refresh_token": self.refresh_token, "expires_in": expires_in, "working_endpoint": endpoint_config, "token_type": token_data.get('token_type', 'Bearer'), "scope": token_data.get('scope', 'offline_access') } else: # Log detailed error information error_text = response.text try: error_data = response.json() error_message = error_data.get('error_description', error_data.get('error', 'Unknown error')) error_code = error_data.get('error', 'unknown_error') except: error_message = error_text error_code = f"http_{response.status_code}" logger.warning(f"โŒ {endpoint_name} failed: {response.status_code} - {error_message}") # Specific error handling based on Dexcom API behavior if response.status_code == 404: logger.info(f" โ†’ Token endpoint not found, trying next configuration...") continue elif response.status_code == 400 and 'invalid_grant' in error_code: logger.info(f" โ†’ Invalid/expired authorization code, trying next configuration...") continue elif response.status_code == 401 and 'invalid_client' in error_code: logger.info(f" โ†’ Invalid client credentials, trying next configuration...") continue else: logger.info(f" โ†’ HTTP {response.status_code} error, trying next configuration...") continue except requests.exceptions.Timeout: logger.warning(f"โฑ๏ธ {endpoint_name} timed out, trying next configuration...") continue except requests.exceptions.RequestException as e: logger.warning(f"๐ŸŒ {endpoint_name} network error: {e}, trying next configuration...") continue except Exception as e: logger.warning(f"โ“ {endpoint_name} unexpected error: {e}, trying next configuration...") continue # All endpoints failed logger.error("โŒ All token endpoint configurations failed") return { "success": False, "error": "All token endpoints failed. This may be due to invalid authorization code, expired code, or Dexcom API issues.", "suggestion": "Please try getting a fresh authorization code (they expire in 60 seconds) or check your Dexcom developer credentials.", "endpoints_tried": configurations_to_try, "troubleshooting": [ "1. Make sure you copied the COMPLETE callback URL", "2. Authorization codes expire in 60 seconds - get a fresh one", "3. Verify your CLIENT_ID and CLIENT_SECRET are correct", "4. Check if your app is properly registered in Dexcom developer portal", "5. Ensure redirect URI matches exactly: http://localhost:7860/callback" ] } def _ensure_valid_token(self): """Ensure we have a valid access token""" if not self.access_token: raise Exception("No access token available. Please authenticate first.") if self.token_expires_at and datetime.now() >= self.token_expires_at - timedelta(minutes=5): logger.info("Token expiring soon, need to refresh...") if self.refresh_token: if not self._refresh_token(): raise Exception("Token expired and refresh failed") else: raise Exception("Token expired and no refresh token available") def _refresh_token(self) -> bool: """Refresh the access token using refresh token""" if not self.refresh_token: return False token_url = self.working_endpoints["token"] data = { 'client_id': self.client_id, 'client_secret': self.client_secret, 'refresh_token': self.refresh_token, 'grant_type': 'refresh_token' } headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } try: response = requests.post(token_url, data=data, headers=headers) if response.status_code == 200: token_data = response.json() self.access_token = token_data.get('access_token') if 'refresh_token' in token_data: self.refresh_token = token_data.get('refresh_token') expires_in = token_data.get('expires_in', 7200) self.token_expires_at = datetime.now() + timedelta(seconds=expires_in) logger.info("โœ… Token refreshed successfully") return True else: logger.error(f"Token refresh failed: {response.status_code}") return False except Exception as e: logger.error(f"Token refresh error: {e}") return False def get_auth_headers(self) -> Dict[str, str]: """Get authorization headers for API calls""" self._ensure_valid_token() return { 'Authorization': f'Bearer {self.access_token}', 'Accept': 'application/json', 'User-Agent': 'GlycoAI-DexcomSandbox/1.0' } def get_data_range(self) -> Dict[str, Any]: """Get available data range from Dexcom API""" api_base = self.get_api_base_url() url = f"{api_base}/users/self/dataRange" headers = self.get_auth_headers() try: response = requests.get(url, headers=headers, timeout=30) if response.status_code == 200: data = response.json() logger.info(f"โœ… Data range retrieved from API {self.api_version}: {data}") return data else: logger.error(f"Data range API error: {response.status_code} - {response.text}") raise Exception(f"Data range API error: {response.status_code}") except Exception as e: logger.error(f"Error getting data range: {e}") raise def get_glucose_data(self, start_date: str = None, end_date: str = None) -> List[Dict]: """Get glucose (EGV) data from Dexcom API""" api_base = self.get_api_base_url() url = f"{api_base}/users/self/egvs" headers = self.get_auth_headers() # Per Dexcom documentation: all endpoints except dataRange require startDate and endDate params = {} if start_date: params['startDate'] = start_date if end_date: params['endDate'] = end_date # Validate date range (max 90 days per Dexcom spec) if start_date and end_date: from datetime import datetime start_dt = datetime.fromisoformat(start_date.replace('Z', '+00:00')) end_dt = datetime.fromisoformat(end_date.replace('Z', '+00:00')) delta = (end_dt - start_dt).days if delta > 90: logger.warning(f"Date range {delta} days exceeds Dexcom 90-day limit") # Adjust to last 90 days start_dt = end_dt - timedelta(days=90) params['startDate'] = start_dt.isoformat() + 'Z' logger.info(f"Adjusted to 90-day window: {params['startDate']} to {end_date}") try: logger.info(f"๐Ÿ” Fetching glucose data from API {self.api_version}") logger.info(f" URL: {url}") logger.info(f" Params: {params}") response = requests.get(url, headers=headers, params=params, timeout=30) if response.status_code == 200: data = response.json() egvs = data.get('egvs', []) logger.info(f"โœ… Retrieved {len(egvs)} glucose readings from API {self.api_version}") return egvs else: logger.error(f"EGV API error: {response.status_code} - {response.text}") raise Exception(f"EGV API error: {response.status_code}") except Exception as e: logger.error(f"Error getting glucose data: {e}") raise def get_events_data(self, start_date: str = None, end_date: str = None) -> List[Dict]: """Get events data from Dexcom API""" api_base = self.get_api_base_url() url = f"{api_base}/users/self/events" headers = self.get_auth_headers() params = {} if start_date: params['startDate'] = start_date if end_date: params['endDate'] = end_date try: response = requests.get(url, headers=headers, params=params, timeout=30) if response.status_code == 200: data = response.json() events = data.get('events', []) logger.info(f"โœ… Retrieved {len(events)} events from API {self.api_version}") return events else: logger.warning(f"Events API returned {response.status_code}, continuing without events") return [] except Exception as e: logger.warning(f"Error getting events data: {e}, continuing without events") return [] class DexcomSandboxIntegration: """Integration wrapper for Gradio app with API version selection""" def __init__(self, api_version: str = "v3"): self.oauth = DexcomSandboxOAuth(api_version=api_version) self.api_version = api_version self.authenticated = False self.user_profile = None self.glucose_data = None self.events_data = None self.data_loaded_at = None def start_oauth(self) -> Tuple[str, bool, bool]: """Start OAuth flow for Gradio interface""" result = self.oauth.start_oauth_flow() if result["success"]: return ( result["instructions"], True, # Show callback input True # Show complete button ) else: return ( f"โŒ Failed to start OAuth: {result['error']}", False, False ) def complete_oauth(self, callback_url: str) -> Tuple[str, bool]: """Complete OAuth flow for Gradio interface""" result = self.oauth.complete_oauth(callback_url) if result["success"]: self.authenticated = True self.user_profile = result["user_profile"] return ( f"โœ… Dexcom Sandbox authenticated successfully! Click 'Load Data' to begin.", True # Show main interface ) else: return ( f"โŒ OAuth failed: {result['error']}", False ) def load_glucose_data(self, days: int = 14) -> Dict[str, Any]: """Load glucose data for the specified number of days""" if not self.authenticated: return { "success": False, "error": "Not authenticated. Please complete OAuth first." } try: # Calculate date range end_time = datetime.now() start_time = end_time - timedelta(days=days) # Fetch glucose data self.glucose_data = self.oauth.get_glucose_data( start_date=start_time.isoformat(), end_date=end_time.isoformat() ) # Fetch events data self.events_data = self.oauth.get_events_data( start_date=start_time.isoformat(), end_date=end_time.isoformat() ) self.data_loaded_at = datetime.now() return { "success": True, "glucose_readings": len(self.glucose_data), "events": len(self.events_data), "date_range": f"{start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}", "user": self.user_profile.name } except Exception as e: logger.error(f"Failed to load glucose data: {e}") return { "success": False, "error": f"Failed to load data: {str(e)}" } def get_user_profile(self) -> Optional[DexcomSandboxUser]: """Get authenticated user profile""" return self.user_profile if self.authenticated else None def get_glucose_data_for_ui(self) -> Optional[List[Dict]]: """Get glucose data formatted for UI display""" return self.glucose_data if self.authenticated else None def get_status(self) -> Dict[str, Any]: """Get current authentication and data status""" return { "authenticated": self.authenticated, "user": self.user_profile.name if self.user_profile else None, "glucose_readings": len(self.glucose_data) if self.glucose_data else 0, "events": len(self.events_data) if self.events_data else 0, "data_loaded_at": self.data_loaded_at.isoformat() if self.data_loaded_at else None, "token_expires_at": self.oauth.token_expires_at.isoformat() if self.oauth.token_expires_at else None } def debug_oauth_endpoints(): """Debug function to test all OAuth endpoints""" print("๐Ÿ” DEBUGGING DEXCOM OAUTH ENDPOINTS") print("=" * 60) # Test all endpoint configurations all_configs = [DEFAULT_ENDPOINTS] + ENDPOINT_CONFIGURATIONS for i, config in enumerate(all_configs): name = config.get('name', f'Configuration {i+1}') print(f"\n๐Ÿงช Testing {name}:") print(f" Login: {config['login']}") print(f" Token: {config['token']}") print(f" API v2: {config['api_v2']}") print(f" API v3: {config['api_v3']}") # Test if endpoints are reachable for endpoint_type, url in config.items(): if endpoint_type == 'name': continue try: # Just check if the endpoint exists (don't send real requests) response = requests.head(url, timeout=5) status = "โœ… Reachable" if response.status_code != 404 else "โŒ 404 Not Found" print(f" {endpoint_type.upper()}: {status} ({response.status_code})") except requests.exceptions.RequestException as e: print(f" {endpoint_type.upper()}: โŒ Error - {e}") print(f"\n๐Ÿ’ก Key Points from Official Documentation:") print("1. OAuth endpoints are always v2 (even for v3 API calls)") print("2. Sandbox login uses special developer.dexcom.com/sandbox-login") print("3. Token exchange uses production OAuth endpoint for sandbox too") print("4. API v3 supports G6, G7, ONE, ONE+ devices") print("5. API v2 supports G5, G6 devices (legacy)") print("6. Time window max 90 days for all endpoints except dataRange") def test_dexcom_sandbox(): """Test Dexcom Sandbox OAuth implementation with both API versions""" print("๐Ÿงช Testing Dexcom Sandbox OAuth Implementation") print("=" * 60) # Run endpoint debug first debug_oauth_endpoints() print(f"\n๐Ÿš€ OAuth Flow Test:") # Test both API versions for api_version in ["v3", "v2"]: print(f"\n--- Testing API {api_version} ---") # Initialize OAuth oauth = DexcomSandboxOAuth(api_version=api_version) # Test auth URL generation auth_url = oauth.generate_auth_url() print(f"โœ… Auth URL generated: {auth_url}") # Test integration wrapper integration = DexcomSandboxIntegration(api_version=api_version) instructions, show_input, show_button = integration.start_oauth() print(f"โœ… OAuth flow started for API {api_version}") print(f" Show callback input: {show_input}") print(f" Show complete button: {show_button}") print(f"\n๐Ÿ“‹ Next steps for testing:") print(f"1. Choose API version (v3 recommended for newer devices)") print(f"2. Open auth URL and select sandbox user") print(f"3. SandboxUser6 (G6) works with both v2 and v3") print(f"4. SandboxUser7 (G7) requires API v3") print(f"5. Copy callback URL after 404 error") print(f"6. System will automatically try multiple token endpoints") print(f"\n๐Ÿ‘ฅ Available Sandbox Users:") for key, name in SANDBOX_USERS.items(): print(f" โ€ข {name}") if "G6" in name: print(f" โ†ณ ๐ŸŽฏ Works with both API v2 and v3") elif "G7" in name: print(f" โ†ณ ๐ŸŽฏ Requires API v3") return oauth, integration if __name__ == "__main__": test_dexcom_sandbox()