93 lines
2.5 KiB
Python
93 lines
2.5 KiB
Python
"""Apple OAuth client with iCloud Drive API support.
|
|
|
|
Activated when OAUTH_APPLE_ID is set in .env.
|
|
Note: Apple requires Sign in with Apple capability and team ID configuration.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
|
|
from app.core.config import get_settings
|
|
|
|
APPLE_OAUTH_URL = "https://appleid.apple.com/auth/authorize"
|
|
APPLE_TOKEN_URL = "https://appleid.apple.com/auth/token"
|
|
|
|
SCOPES = "name email"
|
|
|
|
VOIDEA_DRIVE_FOLDER = "VoIdeaAI"
|
|
|
|
|
|
@dataclass
|
|
class AppleUserInfo:
|
|
id: str
|
|
email: str
|
|
display_name: str
|
|
|
|
|
|
def is_available() -> bool:
|
|
return get_settings().apple_oauth_enabled
|
|
|
|
|
|
async def get_authorize_url() -> str:
|
|
settings = get_settings()
|
|
params = {
|
|
"response_type": "code id_token",
|
|
"client_id": settings.oauth_apple_id,
|
|
"redirect_uri": settings.oauth_apple_redirect_uri,
|
|
"scope": SCOPES,
|
|
"response_mode": "form_post",
|
|
}
|
|
query = "&".join(f"{k}={v}" for k, v in params.items())
|
|
return f"{APPLE_OAUTH_URL}?{query}"
|
|
|
|
|
|
async def exchange_code(code: str) -> dict | None:
|
|
settings = get_settings()
|
|
if not settings.apple_oauth_enabled:
|
|
return None
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
APPLE_TOKEN_URL,
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"client_id": settings.oauth_apple_id,
|
|
"client_secret": settings.oauth_apple_secret,
|
|
"redirect_uri": settings.oauth_apple_redirect_uri,
|
|
},
|
|
)
|
|
if resp.status_code != 200:
|
|
return None
|
|
return resp.json()
|
|
|
|
|
|
async def get_user_info(access_token: str) -> AppleUserInfo | None:
|
|
"""Apple returns user info only in the initial authorization response,
|
|
not from a userinfo endpoint. This method decodes the id_token claims.
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
"https://appleid.apple.com/auth/keys",
|
|
)
|
|
if resp.status_code != 200:
|
|
return None
|
|
return None # Full implementation requires JWT id_token decoding
|
|
|
|
|
|
async def ensure_app_folder(access_token: str) -> str | None:
|
|
return None # iCloud Drive API requires additional entitlements
|
|
|
|
|
|
async def upload_file(
|
|
access_token: str,
|
|
file_name: str,
|
|
file_content: bytes,
|
|
parent_id: str | None = None,
|
|
) -> bool:
|
|
return False # Requires CloudKit API integration
|
|
|
|
|
|
async def get_disk_info(access_token: str) -> dict | None:
|
|
return None # Requires CloudKit API integration
|