adding dropbox auth manager class, test script
This commit is contained in:
adamhaley
@@ -0,0 +1,93 @@
|
||||
# dropbox_auth_manager.py
|
||||
import os
|
||||
import requests
|
||||
import keyring
|
||||
from keyring.errors import KeyringError
|
||||
|
||||
DROPBOX_KEYRING_SERVICE = "comfyui_dropbox"
|
||||
|
||||
class DropboxAuthManager:
|
||||
def __init__(self, app_key=None, app_secret=None):
|
||||
self.app_key = app_key or keyring.get_password(DROPBOX_KEYRING_SERVICE, "app_key")
|
||||
self.app_secret = app_secret or keyring.get_password(DROPBOX_KEYRING_SERVICE, "app_secret")
|
||||
self.refresh_token = keyring.get_password(DROPBOX_KEYRING_SERVICE, "refresh_token")
|
||||
|
||||
def is_connected(self):
|
||||
return bool(self.app_key and self.app_secret and self.refresh_token)
|
||||
|
||||
def store_tokens(self, app_key, app_secret, refresh_token):
|
||||
try:
|
||||
keyring.set_password(DROPBOX_KEYRING_SERVICE, "app_key", app_key)
|
||||
keyring.set_password(DROPBOX_KEYRING_SERVICE, "app_secret", app_secret)
|
||||
keyring.set_password(DROPBOX_KEYRING_SERVICE, "refresh_token", refresh_token)
|
||||
|
||||
# Update instance variables after successful storage
|
||||
self.app_key = app_key
|
||||
self.app_secret = app_secret
|
||||
self.refresh_token = refresh_token
|
||||
except KeyringError as e:
|
||||
raise RuntimeError(f"Failed to store Dropbox tokens securely: {e}")
|
||||
|
||||
def exchange_auth_code(self, auth_code):
|
||||
if not (self.app_key and self.app_secret):
|
||||
raise ValueError("App key and secret must be set before exchanging auth code.")
|
||||
|
||||
data = {
|
||||
"code": auth_code,
|
||||
"grant_type": "authorization_code",
|
||||
"client_id": self.app_key,
|
||||
"client_secret": self.app_secret
|
||||
}
|
||||
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
||||
response = requests.post("https://api.dropbox.com/oauth2/token", headers=headers, data=data)
|
||||
response.raise_for_status()
|
||||
|
||||
creds = response.json()
|
||||
refresh_token = creds.get("refresh_token")
|
||||
if not refresh_token:
|
||||
raise RuntimeError("Failed to obtain refresh token from Dropbox.")
|
||||
|
||||
self.store_tokens(self.app_key, self.app_secret, refresh_token)
|
||||
self.refresh_token = refresh_token
|
||||
return True
|
||||
|
||||
def get_access_token(self):
|
||||
if not self.is_connected():
|
||||
raise RuntimeError("Dropbox not connected. Please run authentication first.")
|
||||
|
||||
data = {
|
||||
"refresh_token": self.refresh_token,
|
||||
"grant_type": "refresh_token",
|
||||
"client_id": self.app_key,
|
||||
"client_secret": self.app_secret
|
||||
}
|
||||
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
||||
response = requests.post("https://api.dropbox.com/oauth2/token", headers=headers, data=data)
|
||||
response.raise_for_status()
|
||||
|
||||
access_token = response.json().get("access_token")
|
||||
if not access_token:
|
||||
raise RuntimeError("Failed to retrieve access token.")
|
||||
return access_token
|
||||
|
||||
def reset(self):
|
||||
"""Clear stored tokens"""
|
||||
try:
|
||||
keyring.delete_password(DROPBOX_KEYRING_SERVICE, "app_key")
|
||||
keyring.delete_password(DROPBOX_KEYRING_SERVICE, "app_secret")
|
||||
keyring.delete_password(DROPBOX_KEYRING_SERVICE, "refresh_token")
|
||||
|
||||
# Clear instance variables
|
||||
self.app_key = None
|
||||
self.app_secret = None
|
||||
self.refresh_token = None
|
||||
except KeyringError:
|
||||
# If passwords don't exist, that's fine
|
||||
pass
|
||||
|
||||
def get_oauth_url(self):
|
||||
"""Generate OAuth URL for initial authorization"""
|
||||
if not self.app_key:
|
||||
raise ValueError("App key must be set to generate OAuth URL.")
|
||||
return f"https://www.dropbox.com/oauth2/authorize?response_type=code&client_id={self.app_key}&token_access_type=offline"
|
||||
|
||||
@@ -2,3 +2,4 @@ requests
|
||||
python-dotenv
|
||||
watchdog
|
||||
dropbox
|
||||
keyring
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_dropbox_auth.py - Manual testing script for DropboxAuthManager
|
||||
|
||||
from dropbox_auth_manager import DropboxAuthManager
|
||||
|
||||
def test_first_time_setup():
|
||||
"""Test the initial auth code exchange flow"""
|
||||
print("🧪 STEP 1: First-time setup test")
|
||||
print("-" * 40)
|
||||
|
||||
auth_code = input("Paste your Dropbox auth code: ").strip()
|
||||
app_key = input("App Key: ").strip()
|
||||
app_secret = input("App Secret: ").strip()
|
||||
|
||||
auth = DropboxAuthManager(app_key, app_secret)
|
||||
|
||||
try:
|
||||
auth.exchange_auth_code(auth_code)
|
||||
print("✅ Refresh token stored securely in system keyring.")
|
||||
return auth
|
||||
except Exception as e:
|
||||
print(f"❌ Error exchanging auth code: {e}")
|
||||
return None
|
||||
|
||||
def test_stored_credentials():
|
||||
"""Test retrieving and using stored credentials"""
|
||||
print("\n🧪 STEP 2: Stored credentials test")
|
||||
print("-" * 40)
|
||||
|
||||
try:
|
||||
# Test loading from keyring (no app_key/secret needed)
|
||||
auth = DropboxAuthManager()
|
||||
|
||||
if not auth.is_connected():
|
||||
print("❌ No stored credentials found. Run first-time setup first.")
|
||||
return False
|
||||
|
||||
print("✅ Stored credentials found.")
|
||||
|
||||
# Test getting access token
|
||||
access_token = auth.get_access_token()
|
||||
print(f"✅ Got access token: {access_token[:10]}...")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error getting access token: {e}")
|
||||
return False
|
||||
|
||||
def test_oauth_url_generation():
|
||||
"""Test OAuth URL generation"""
|
||||
print("\n🧪 STEP 3: OAuth URL generation test")
|
||||
print("-" * 40)
|
||||
|
||||
app_key = input("App Key (for URL generation): ").strip()
|
||||
auth = DropboxAuthManager(app_key=app_key)
|
||||
|
||||
try:
|
||||
oauth_url = auth.get_oauth_url()
|
||||
print(f"✅ Generated OAuth URL:")
|
||||
print(f" {oauth_url}")
|
||||
print(f" Copy this URL to authorize your app")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"❌ Error generating OAuth URL: {e}")
|
||||
return False
|
||||
|
||||
def test_reset():
|
||||
"""Test credential reset functionality"""
|
||||
print("\n🧪 STEP 4: Reset credentials test")
|
||||
print("-" * 40)
|
||||
|
||||
confirm = input("Are you sure you want to clear stored credentials? (y/N): ").strip().lower()
|
||||
if confirm != 'y':
|
||||
print("⚠️ Skipping reset test.")
|
||||
return
|
||||
|
||||
try:
|
||||
auth = DropboxAuthManager()
|
||||
auth.reset()
|
||||
print("✅ Credentials cleared from keyring.")
|
||||
|
||||
# Verify they're actually gone
|
||||
auth_check = DropboxAuthManager()
|
||||
if not auth_check.is_connected():
|
||||
print("✅ Verified: No stored credentials found after reset.")
|
||||
else:
|
||||
print("❌ Warning: Credentials still found after reset.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error during reset: {e}")
|
||||
|
||||
def main():
|
||||
print("🚀 Dropbox Auth Manager Test Suite")
|
||||
print("=" * 50)
|
||||
|
||||
while True:
|
||||
print("\nSelect a test:")
|
||||
print("1. First-time setup (exchange auth code)")
|
||||
print("2. Test stored credentials")
|
||||
print("3. Generate OAuth URL")
|
||||
print("4. Reset stored credentials")
|
||||
print("5. Exit")
|
||||
|
||||
choice = input("\nEnter choice (1-5): ").strip()
|
||||
|
||||
if choice == "1":
|
||||
test_first_time_setup()
|
||||
elif choice == "2":
|
||||
test_stored_credentials()
|
||||
elif choice == "3":
|
||||
test_oauth_url_generation()
|
||||
elif choice == "4":
|
||||
test_reset()
|
||||
elif choice == "5":
|
||||
print("👋 Exiting test suite.")
|
||||
break
|
||||
else:
|
||||
print("❌ Invalid choice. Please select 1-5.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user