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
59
60
61
62
63
64
65
66
67
68
69
|
from fastapi import FastAPI, Request, Query
from fastapi.responses import RedirectResponse
import httpx
import os
from .db import add_user
from dotenv import load_dotenv
load_dotenv()
app = FastAPI()
CLIENT_ID = os.environ.get("SLAKC_CLIENT_ID")
CLIENT_SECRET = os.environ.get("SLAKC_CLIENT_SECRET")
REDIRECT_URI = os.environ.get("SLAKC_REDIRECT_URI")
def validate_user_scopes(granted_scopes: str, required_scopes: list) -> bool:
if not granted_scopes:
return False
granted_set = set(granted_scopes.split(','))
required_set = set(required_scopes)
return required_set.issubset(granted_set)
@app.get("/slack/oauth/install")
async def install():
slack_auth_url = (
"https://slack.com/oauth/v2/authorize"
f"?client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URI}"
"&scope="
"&user_scope=channels:write,channels:history,channels:read,chat:write,im:history,users:read"
)
return RedirectResponse(slack_auth_url)
@app.get("/slack/oauth/redirect")
async def oauth_redirect(request: Request, code: str = Query(...), state: str = Query(None)):
async with httpx.AsyncClient() as client:
response = await client.post("https://slack.com/api/oauth.v2.access", data={
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"code": code,
"redirect_uri": REDIRECT_URI
})
data = response.json()
if not data.get("ok"):
return {"error": data.get("error", "Unknown error")}
user_id = data.get("authed_user", {}).get("id")
access_token = data.get("authed_user", {}).get("access_token")
granted_scopes = data.get("authed_user", {}).get("scope", "")
if not user_id or not access_token:
return {"error": "Could not get user info from OAuth response"}
required_scopes = [
"channels:write",
"channels:history",
"channels:read",
"chat:write",
"im:history",
"users:read"
]
if not validate_user_scopes(granted_scopes, required_scopes):
return {"error": "Missing required OAuth scopes. Please re-install the app without modifying the URL."}
await add_user(user_id, access_token)
return {"message": "Successfully joined Botnet!", "user_id": user_id}
|
