first

app/utils/social_utils.py

@@ -2,69 +2,192 @@ from google.oauth2 import id_token as google_id_token
 from google.auth.transport import requests as google_requests
 from jose import jwt as jose_jwt, JWTError, jwk
 from jose.utils import base64url_decode
-from typing import Dict
+from typing import Dict, Optional
 import httpx
-import json
-
-# Async wrapper to run Google’s sync function in a thread pool
 import asyncio
-from functools import partial
+from functools import partial, lru_cache
+import logging
+from datetime import datetime, timedelta
+
+logger = logging.getLogger(__name__)
+
+class TokenVerificationError(Exception):
+    """Custom exception for token verification errors"""
+    pass
+
+class GoogleTokenVerifier:
+    def __init__(self, client_id: str):
+        self.client_id = client_id
+    
+    async def verify_token(self, token: str) -> Dict:
+        """
+        Asynchronously verifies a Google ID token and returns the payload if valid.
+        """
+        try:
+            loop = asyncio.get_event_loop()
+            # Run the sync method in a thread to avoid blocking
+            idinfo = await loop.run_in_executor(
+                None,
+                partial(google_id_token.verify_oauth2_token, token, google_requests.Request(), self.client_id)
+            )
+            
+            # Validate issuer
+            if idinfo.get('iss') not in ['accounts.google.com', 'https://accounts.google.com']:
+                raise TokenVerificationError('Invalid issuer')
+            
+            # Additional validation
+            if idinfo.get('aud') != self.client_id:
+                raise TokenVerificationError('Invalid audience')
+            
+            # Check token expiration (extra safety)
+            exp = idinfo.get('exp')
+            if exp and datetime.fromtimestamp(exp) < datetime.utcnow():
+                raise TokenVerificationError('Token has expired')
+            
+            logger.info(f"Successfully verified Google token for user: {idinfo.get('email')}")
+            return idinfo
+            
+        except Exception as e:
+            logger.error(f"Google token verification failed: {str(e)}")
+            raise TokenVerificationError(f"Invalid Google token: {str(e)}")
 
-# ✅ Async Google token verification
+class AppleTokenVerifier:
+    def __init__(self, audience: str, cache_duration: int = 3600):
+        self.audience = audience
+        self.cache_duration = cache_duration
+        self._keys_cache = None
+        self._cache_timestamp = None
+    
+    async def _get_apple_keys(self) -> list:
+        """
+        Fetch Apple's public keys with caching to reduce API calls.
+        """
+        now = datetime.utcnow()
+        
+        # Check if cache is still valid
+        if (self._keys_cache and self._cache_timestamp and 
+            now - self._cache_timestamp < timedelta(seconds=self.cache_duration)):
+            return self._keys_cache
+        
+        try:
+            async with httpx.AsyncClient(timeout=10.0) as client:
+                response = await client.get('https://appleid.apple.com/auth/keys')
+                response.raise_for_status()
+                keys = response.json().get('keys', [])
+                
+                # Update cache
+                self._keys_cache = keys
+                self._cache_timestamp = now
+                
+                logger.info("Successfully fetched Apple public keys")
+                return keys
+                
+        except httpx.RequestError as e:
+            logger.error(f"Failed to fetch Apple keys: {str(e)}")
+            raise TokenVerificationError(f"Could not fetch Apple public keys: {str(e)}")
+    
+    async def verify_token(self, token: str) -> Dict:
+        """
+        Asynchronously verifies an Apple identity token and returns the decoded payload.
+        """
+        try:
+            # Fetch Apple's public keys
+            keys = await self._get_apple_keys()
+            
+            # Decode header to get kid and alg
+            header = jose_jwt.get_unverified_header(token)
+            kid = header.get('kid')
+            alg = header.get('alg')
+            
+            if not kid or not alg:
+                raise TokenVerificationError("Token header missing required fields")
+            
+            # Find matching key
+            key = next((k for k in keys if k['kid'] == kid and k['alg'] == alg), None)
+            if not key:
+                raise TokenVerificationError("Public key not found for Apple token")
+            
+            # Verify signature manually (additional safety)
+            public_key = jwk.construct(key)
+            message, encoded_sig = token.rsplit('.', 1)
+            decoded_sig = base64url_decode(encoded_sig.encode())
+            
+            if not public_key.verify(message.encode(), decoded_sig):
+                raise TokenVerificationError("Invalid Apple token signature")
+            
+            # Decode and validate claims
+            claims = jose_jwt.decode(
+                token,
+                key,
+                algorithms=['RS256'],
+                audience=self.audience,
+                issuer='https://appleid.apple.com'
+            )
+            
+            # Additional validation
+            if claims.get('aud') != self.audience:
+                raise TokenVerificationError('Invalid audience')
+            
+            logger.info(f"Successfully verified Apple token for user: {claims.get('sub')}")
+            return claims
+            
+        except JWTError as e:
+            logger.error(f"JWT error during Apple token verification: {str(e)}")
+            raise TokenVerificationError(f"Invalid Apple token: {str(e)}")
+        except Exception as e:
+            logger.error(f"Apple token verification failed: {str(e)}")
+            raise TokenVerificationError(f"Invalid Apple token: {str(e)}")
+
+# Factory class for easier usage
+class OAuthVerifier:
+    def __init__(self, google_client_id: Optional[str] = None, apple_audience: Optional[str] = None):
+        self.google_verifier = GoogleTokenVerifier(google_client_id) if google_client_id else None
+        self.apple_verifier = AppleTokenVerifier(apple_audience) if apple_audience else None
+    
+    async def verify_google_token(self, token: str) -> Dict:
+        if not self.google_verifier:
+            raise TokenVerificationError("Google verifier not configured")
+        return await self.google_verifier.verify_token(token)
+    
+    async def verify_apple_token(self, token: str) -> Dict:
+        if not self.apple_verifier:
+            raise TokenVerificationError("Apple verifier not configured")
+        return await self.apple_verifier.verify_token(token)
+
+# Convenience functions (backward compatibility)
 async def verify_google_token(token: str, client_id: str) -> Dict:
     """
     Asynchronously verifies a Google ID token and returns the payload if valid.
     """
-    try:
-        loop = asyncio.get_event_loop()
-        # Run the sync method in a thread to avoid blocking
-        idinfo = await loop.run_in_executor(
-            None,
-            partial(google_id_token.verify_oauth2_token, token, google_requests.Request(), client_id)
-        )
-        if idinfo.get('iss') not in ['accounts.google.com', 'https://accounts.google.com']:
-            raise ValueError('Wrong issuer.')
-        return idinfo
-    except Exception as e:
-        raise ValueError(f"Invalid Google token: {str(e)}")
+    verifier = GoogleTokenVerifier(client_id)
+    return await verifier.verify_token(token)
 
-
-# ✅ Async Apple token verification
 async def verify_apple_token(token: str, audience: str) -> Dict:
     """
     Asynchronously verifies an Apple identity token and returns the decoded payload.
     """
-    try:
-        # Fetch Apple's public keys (JWKS)
-        async with httpx.AsyncClient() as client:
-            response = await client.get('https://appleid.apple.com/auth/keys')
-            response.raise_for_status()
-            keys = response.json().get('keys', [])
-
-        # Decode header to get kid and alg
-        header = jose_jwt.get_unverified_header(token)
-        kid = header.get('kid')
-        alg = header.get('alg')
-
-        key = next((k for k in keys if k['kid'] == kid and k['alg'] == alg), None)
-        if not key:
-            raise ValueError("Public key not found for Apple token")
+    verifier = AppleTokenVerifier(audience)
+    return await verifier.verify_token(token)
 
-        public_key = jwk.construct(key)
-        message, encoded_sig = token.rsplit('.', 1)
-        decoded_sig = base64url_decode(encoded_sig.encode())
-
-        if not public_key.verify(message.encode(), decoded_sig):
-            raise ValueError("Invalid Apple token signature")
-
-        claims = jose_jwt.decode(
-            token,
-            key,
-            algorithms=['RS256'],
-            audience=audience,
-            issuer='https://appleid.apple.com'
-        )
-        return claims
+# Example usage
+async def example_usage():
+    # Initialize verifier
+    oauth_verifier = OAuthVerifier(
+        google_client_id="your-google-client-id.googleusercontent.com",
+        apple_audience="your.app.bundle.id"
+    )
+    
+    try:
+        # Verify Google token
+        google_claims = await oauth_verifier.verify_google_token("google_id_token_here")
+        print(f"Google user: {google_claims.get('email')}")
+        
+        # Verify Apple token
+        apple_claims = await oauth_verifier.verify_apple_token("apple_id_token_here")
+        print(f"Apple user: {apple_claims.get('sub')}")
+        
+    except TokenVerificationError as e:
+        print(f"Verification failed: {e}")
 
-    except Exception as e:
-        raise ValueError(f"Invalid Apple token: {str(e)}")
+if __name__ == "__main__":
+    asyncio.run(example_usage())