forked from spacebar/SplatNet-Backend
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c4fe4f1bf1 | |||
| ca8090d20d | |||
| 0b5e56b69b | |||
| c89806fe87 | |||
| 13afb709c5 | |||
| 2d703c78b3 |
8 changed files with 36 additions and 26 deletions
|
|
@ -11,9 +11,12 @@ jobs:
|
|||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v6
|
||||
|
||||
- name: Set up Docker Engine
|
||||
uses: docker/setup-docker-action@v4
|
||||
|
||||
- name: install docker???
|
||||
run: |
|
||||
apt update && apt install wget cmake -y && \
|
||||
wget "https://download.docker.com/linux/debian/dists/bookworm/pool/stable/amd64/docker-ce-cli_29.2.1-1~debian.12~bookworm_amd64.deb" && \
|
||||
dpkg --force-confold -i "docker-ce-cli_29.2.1-1~debian.12~bookworm_amd64.deb"
|
||||
|
||||
- name: set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v4
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ class Settings(BaseSettings):
|
|||
twitter_client_id: str
|
||||
twitter_client_secret: str
|
||||
twitter_redirect_uri: str
|
||||
account_client_secret: str
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_file=[
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ async def get_equipment_composite(request: Request, db: DBSession = Depends(get_
|
|||
try:
|
||||
decrypted_pass = cipher.decrypt(user.spfn_pass_enc.encode()).decode()
|
||||
token_data = auth.get_token(user.username, decrypted_pass)
|
||||
profile = json.loads(auth.get_profile(token_data["token"]))
|
||||
profile = auth.get_profile(token_data["access_token"])
|
||||
pid_val = int(profile.get("pid"))
|
||||
mii_name = profile.get("name", user.username)
|
||||
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ async def get_history(request: Request, db: DBSession = Depends(get_db)):
|
|||
try:
|
||||
decrypted_pass = cipher.decrypt(user.spfn_pass_enc.encode()).decode()
|
||||
token_data = auth.get_token(user.username, decrypted_pass)
|
||||
profile = json.loads(auth.get_profile(token_data["token"]))
|
||||
profile = auth.get_profile(token_data["access_token"])
|
||||
pid_val = int(profile.get("pid"))
|
||||
print(f"Fetching History for PID: {pid_val}")
|
||||
|
||||
|
|
|
|||
|
|
@ -33,10 +33,10 @@ async def get_my_profile(request: Request, db: DBSession = Depends(get_db)):
|
|||
decrypted_pass = cipher.decrypt(user.spfn_pass_enc.encode()).decode()
|
||||
|
||||
token_data = auth.get_token(user.username, decrypted_pass)
|
||||
if not token_data or "token" not in token_data:
|
||||
if not token_data or "access_token" not in token_data:
|
||||
return RedirectResponse(url="/sign_in/")
|
||||
|
||||
profile_data = auth.get_profile(token_data["token"])
|
||||
profile_data = auth.get_profile(token_data["access_token"])
|
||||
if not profile_data:
|
||||
raise HTTPException(status_code=404, detail="Profile not found")
|
||||
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ async def login(
|
|||
try:
|
||||
print("fetching spfn token")
|
||||
data = auth.get_token(username, password)
|
||||
if not data or "token" not in data:
|
||||
if not data or "access_token" not in data:
|
||||
print("auth failed")
|
||||
return RedirectResponse(f"{auth_path}{sep}error=auth&username={username}", 303)
|
||||
|
||||
|
|
|
|||
|
|
@ -93,7 +93,7 @@ async def confirm_twitter(state: str, code: str, db: DBSession = Depends(get_db)
|
|||
decrypted_pass = cipher.decrypt(user.spfn_pass_enc.encode()).decode()
|
||||
profile_auth = auth.get_token(user.username, decrypted_pass)
|
||||
|
||||
profile_response = auth.get_profile(profile_auth["token"])
|
||||
profile_response = auth.get_profile(profile_auth["access_token"])
|
||||
|
||||
if isinstance(profile_response, str):
|
||||
try:
|
||||
|
|
@ -149,7 +149,7 @@ async def get_twitter_status(request: Request, db: DBSession = Depends(get_db)):
|
|||
try:
|
||||
decrypted_pass = cipher.decrypt(user.spfn_pass_enc.encode()).decode()
|
||||
profile_auth = auth.get_token(user.username, decrypted_pass)
|
||||
profile_res = auth.get_profile(profile_auth["token"])
|
||||
profile_res = auth.get_profile(profile_auth["access_token"])
|
||||
|
||||
if isinstance(profile_res, str):
|
||||
profile_data = json.loads(profile_res)
|
||||
|
|
@ -183,7 +183,7 @@ async def unlink_twitter(request: Request, db: DBSession = Depends(get_db)):
|
|||
user = db.query(User).filter(User.username == db_session.username).first()
|
||||
decrypted_pass = cipher.decrypt(user.spfn_pass_enc.encode()).decode()
|
||||
profile_auth = auth.get_token(user.username, decrypted_pass)
|
||||
profile_data = auth.get_profile(profile_auth["token"])
|
||||
profile_data = auth.get_profile(profile_auth["access_token"])
|
||||
|
||||
if isinstance(profile_data, str):
|
||||
profile_data = json.loads(profile_data)
|
||||
|
|
|
|||
|
|
@ -1,35 +1,41 @@
|
|||
import requests
|
||||
import base64
|
||||
from config import settings
|
||||
from argon2 import PasswordHasher
|
||||
|
||||
API_URL = "https://account.spfn.net/api/v2"
|
||||
CLIENT_ID = "splatnet"
|
||||
CLIENT_SECRET = settings.account_client_secret
|
||||
ph = PasswordHasher()
|
||||
|
||||
def hash_password(password: str):
|
||||
return ph.hash(password)
|
||||
|
||||
def verify_password(hashed: str, password: str):
|
||||
try:
|
||||
return ph.verify(hashed, password)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def get_token(username, password):
|
||||
creds = f"{username} {password}"
|
||||
encoded = base64.b64encode(creds.encode()).decode()
|
||||
url = f"{API_URL}/oauth2/generate_token"
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Basic {encoded}",
|
||||
"User-Agent": "saturday & allison should lwk kiss..."
|
||||
payload = {
|
||||
"grant_type": "password",
|
||||
"username": username,
|
||||
"password": password,
|
||||
"client_id": CLIENT_ID,
|
||||
"client_secret": CLIENT_SECRET
|
||||
}
|
||||
|
||||
response = requests.get(f"{API_URL}/oauth2/generate_token", headers=headers, timeout=10)
|
||||
headers = {
|
||||
"User-Agent": "chiyo & eri should lwk kiss..."
|
||||
}
|
||||
|
||||
response = requests.post(url, data=payload, headers=headers, timeout=10)
|
||||
return response.json() if response.ok else None
|
||||
|
||||
def get_profile(token):
|
||||
url = f"{API_URL}/users/@me/profile"
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"User-Agent": "every update im going to change this user agent because im bored!!"
|
||||
"User-Agent": "lets make the most of the night like were gonna die young!!"
|
||||
}
|
||||
response = requests.get(f"{API_URL}/users/@me/profile", headers=headers, timeout=10)
|
||||
return response.text if response.ok else None
|
||||
|
||||
response = requests.get(url, headers=headers, timeout=10)
|
||||
|
||||
return response.json() if response.ok else None
|
||||
Loading…
Reference in a new issue