Spaces:
Sleeping
Sleeping
File size: 1,863 Bytes
f84a02d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | import os
import secrets
from typing import Dict, Optional
import urllib.parse
import requests
class OAuthIntegration:
def __init__(self):
self.oauth_server_url = "http://localhost:5058"
self.services = {
'github': {
'client_id': os.getenv('GITHUB_CLIENT_ID'),
'client_secret': os.getenv('GITHUB_CLIENT_SECRET'),
'auth_url': 'https://github.com/login/oauth/authorize'
},
'google': {
'client_id': os.getenv('GOOGLE_CLIENT_ID'),
'client_secret': os.getenv('GOOGLE_CLIENT_SECRET'),
'auth_url': 'https://accounts.google.com/o/oauth2/v2/auth'
},
'slack': {
'client_id': os.getenv('SLACK_CLIENT_ID'),
'client_secret': os.getenv('SLACK_CLIENT_SECRET'),
'auth_url': 'https://slack.com/oauth/v2/authorize'
}
}
async def initialize(self):
pass
async def close(self):
pass
def check_status(self) -> Dict:
return {"oauth_server": "connected"}
async def get_authorization_url(self, service: str) -> str:
if service not in self.services:
raise ValueError(f"Service {service} not supported")
service_config = self.services[service]
state = secrets.token_urlsafe(32)
redirect_uri = f"{self.oauth_server_url}/api/auth/{service}/callback"
auth_params = {
'client_id': service_config['client_id'],
'redirect_uri': redirect_uri,
'response_type': 'code',
'state': state
}
auth_url = f"{service_config['auth_url']}?{urllib.parse.urlencode(auth_params)}"
return auth_url
# Global instance
oauth_integration = OAuthIntegration()
|