magmaus3/CustomIWMServer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Update server to work with the latest version :3

Browse source
This commit is contained in:
magmaus3 2023-07-06 17:36:55 +02:00
parent d73c7e44d5
commit a7b9b5b87f
Signed by: magmaus3
GPG key ID: 966755D3F4A9B251
6 changed files with 186 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
Dockerfile
View file

@ -1,4 +1,7 @@
FROM python:3.11
ADD https://github.com/krallin/tini/releases/download/v0.19.0/tini /tini
RUN chmod +x /tini
ENTRYPOINT ["/tini", "--"]
WORKDIR /app
@ -6,4 +9,4 @@ COPY . .
RUN --mount=type=cache,target=/tmp/pip pip install . --cache-dir /tmp/pip
CMD python3 -m customiwmserver
CMD python3 -m uvicorn customiwmserver.main:app --host 0.0.0.0 --port 8001 --reload

26
customiwmserver/data_types.py
View file

@ -2,10 +2,19 @@ from pydantic import BaseModel
from typing import List, Optional, Union
from datetime import datetime
class MapPlay(BaseModel):
"""Used to store map playthroughs for players"""
Played: bool = True # Obviously
Clear: bool = False
FullClear: bool = False
class User(BaseModel):
"""Pydantic model for user data.
Created from the user response (`/api/v1/user/x`),
some values might be unused."""
some values might be unused.
Color values are decimal versions of joined hexadecimal codes
(for example (0xFF, 0x00, 0x00) is joined to 0xFF0000 and converted to 16711680)
"""
Admin: bool = True
Banned: bool = False
@ -55,7 +64,12 @@ class User(BaseModel):
# Ratings have to be stored in the following format: {mapID: rating},
# where rating is either 1 or 5, or None.
Ratings: dict = {}
Ratings: dict[str, Optional[int]] = {}
# Plays are stored in the following format:
# {mapID: {}}
# (str type is used instead of an int because MongoDB)
Plays: dict[str, MapPlay] = {}
class Notification(BaseModel):
@ -160,7 +174,8 @@ class Map(BaseModel):
MapCode: str = "AAABBAAA"
Listed: bool = True
HiddenInChallenges: bool = False
DragonCoins: bool = False
ReplayVisibility: int = 0
DragonCoins: bool = False # Gems
# Might be duplicates from the User data
@ -216,6 +231,10 @@ class Map(BaseModel):
BestTimePlaytime: int = 0
MyBestPlaytime: int = 0
BestFullTimeUserID: int = 0
BestFullTimeUsername: str = ""
BestFullTimePlaytime: int = 0
FirstClearUserID: Union[int, None] = 0
# Propably not needed, because you can simply read the
@ -240,6 +259,7 @@ class MapLeaderboard(BaseModel):
BestPlaytime: int
BestPlaytimeTime: str = datetime.strftime(datetime.now(), "%Y-%m-%dT%H:%M:%SZ")
BestReplay: str
FullClear: bool = False
CreatorName: str
UserID: int

16
customiwmserver/database.py
View file

@ -1,4 +1,5 @@
from datetime import datetime
from typing import Literal
from pymongo import MongoClient
@ -28,25 +29,26 @@ def LogAdminAction(
)
def auth_check(Authorization):
def auth_check(Authorization) -> (tuple[Literal[False], Literal["noauth"]] | tuple[Literal[True], dict]):
"""Checks credentials.
Returns a tuple with result (for example False, "nouser").
Results:
- nouser = user not found
- wrongpass = wrong password
- [dictionary] = query
- False if wrong username or password
- True, [dict] if correct
"""
# FIXME: This function currently DOES NOT perform any authentication.
# This means that ANYONE knowing the username could perform actions as the user.
if Authorization is None:
return False, "noauth"
username, password = Authorization.split(":")
query = user_collection.find_one({"Username": username})
if not query:
return False, "nouser"
return False, "noauth"
if query["Password"] != password:
return False, "wrongpass"
# if query["Password"] != password:
# return False, "wrongpass"
return True, query

1
customiwmserver/hooks/__init__.py
View file

@ -1,5 +1,4 @@
# from . import *
import pkgutil
__path__ = pkgutil.extend_path(__path__, __name__)

177
customiwmserver/main.py
View file

@ -38,10 +38,7 @@ async def login(username: str = Form(), password: str = Form(), version: str = F
hook.execute_hooks("player_login", username=username)
auth = db.auth_check(username + ":" + password)
if not auth[0]:
if auth[1] == "nouser":
raise HTTPException(404, detail="User not found.")
elif auth[1] == "wrongpass":
raise HTTPException(403, detail="Password is incorrect.")
raise HTTPException(403, detail="Wrong username or password.")
else:
return {"token": username + ":" + password, "userId": auth[1]["ID"]}
@ -72,9 +69,43 @@ async def create_user(
@app.get("/api/v1/notifunread")
async def notifunread():
"""Returns the number of unread notifications."""
# FIXME Add notifications
# FIXME: Add notifications
return 0
@app.post("/api/v1/user/{userID}")
async def update_user_profile(
userID: int,
Authorization: Union[str, None] = Header(default=None),
ID: int = Body(),
Country: int = Body(),
PantsColor: int = Body(),
HatSpr: int = Body(),
HairColor: int = Body(),
CapeColor: int = Body(),
ShoesColor: int = Body(),
ShirtColor: int = Body(),
SkinColor: int = Body(),
):
authcheck = db.auth_check(Authorization)
if not authcheck[0]:
raise HTTPException(403, detail="Wrong username or password")
elif authcheck[0]:
userData = types.User(**authcheck[1])
userData.Country = Country
userData.PantsColor = PantsColor
userData.HatSpr = HatSpr
userData.HairColor = HairColor
userData.CapeColor = CapeColor
userData.ShoesColor = ShoesColor
userData.ShirtColor = ShirtColor
userData.SkinColor = SkinColor
userData.UpdatedAt = datetime.strftime(datetime.now(), "%Y-%m-%dT%H:%M:%SZ")
db.user_collection.update_one({"ID":userID}, {"$set": {**userData.dict()} })
hook.execute_hooks("player_update", userID=userID)
return userData.dict()
return HTTPException(500, detail="Server failed to handle login (somehow)")
@app.get("/api/v1/refresh")
async def refresh_login(Authorization: Union[str, None] = Header(default=None)):
@ -106,6 +137,7 @@ async def useruploadcooldown():
@app.get("/api/v1/map")
async def search_for_maps(
Authorization: Union[str, None] = Header(default=None),
start: int = 0,
limit: int = 5,
min_diff: float = 0.0,
@ -121,6 +153,18 @@ async def search_for_maps(
admin_show_unlisted: Optional[int] = 0,
):
"""Search for maps."""
if Authorization is not None:
authcheck = db.auth_check(Authorization)
if not authcheck[0]:
raise HTTPException(403,
detail="Wrong username or password."
+ "\nIf you're using the API directly, "
+ "then you can just skip the Authorization")
userData = types.User(**authcheck[1])
plays = userData.Plays
else:
userData = None
plays = {}
# query = db.maps_collection.find({ "CreatorId": author_id }).limit(limit)
if code:
@ -138,6 +182,9 @@ async def search_for_maps(
for i in query:
del i["_id"]
del i["MapData"]
del i["Played"]
del i["Clear"]
del i["FullClear"]
if not i["Listed"] and admin_show_unlisted != 1:
continue
@ -145,6 +192,29 @@ async def search_for_maps(
level_tags = i['TagIDs'].split(",")
CreatedAt = datetime.strptime(i["CreatedAt"], "%Y-%m-%dT%H:%M:%SZ")
Played = False
Clear = False
FullClear = False
if plays and str(i["ID"]) in plays:
play = plays[str(i["ID"])]
Played = play.Played
Clear = play.Clear
FullClear = play.FullClear
# if userData:
# # Old system that checked if a user has a record, remove
# leaderboard = i["Leaderboard"]
# for record in leaderboard:
# record = types.MapLeaderboard(**record)
# if record.UserID == userData.ID:
# Played = True
# Clear = True
# if record.FullClear:
# FullClear = True
# break
# print(f"DBG: {i}, {leaderboard[i]}\t{leaderboard[i]['UserID'] == int(userID):}")
# if leaderboard[i]["UserID"] == int(userID) and (BestTime is None or leaderboard[i]["BestPlaytime"] < BestTime):
# replayIndex = i
# TODO: Improve tag filtering
required_tags_included = False if len(required_tag_list) != 0 else True
disallowed_tags_included = False
@ -154,7 +224,7 @@ async def search_for_maps(
else:
if required_tags_included and not disallowed_tags_included:
if not last_x_hours or CreatedAt.hour < last_x_hours:
entries.append(i)
entries.append(types.Map(Played=Played, Clear=Clear, FullClear=FullClear, **i))
return entries
@ -162,7 +232,7 @@ async def search_for_maps(
async def getMap(mapID: int):
query = db.maps_collection.find_one({"ID": mapID})
del query["_id"]
return query
return types.Map(**query)
@app.post("/api/v1/map/{mapID}/start")
@ -171,20 +241,35 @@ async def startMap(
Authorization: Union[str, None] = Header(default=None)
):
authcheck = db.auth_check(Authorization)
if not authcheck[0] and authcheck[1] == "nouser":
raise HTTPException(404, detail="User not found")
elif not authcheck[0] and authcheck[1] == "wrongpass":
raise HTTPException(403, detail="Wrong password")
if not authcheck[0] and authcheck[1]:
raise HTTPException(403, detail="Wrong username or password.")
elif authcheck[0]:
userData = types.User(**authcheck[1])
query = db.maps_collection.find_one({"ID": mapID})
del query["_id"]
Clear = False
Played = False
if str(mapID) not in userData.Plays :
updateQuery = db.user_collection.update_one(
{"ID": userData.ID},
{
"$set": {
f"Plays.{mapID}": dict(types.MapPlay())
}
}
)
else:
play = userData.Plays[str(mapID)]
Clear = play.Clear
Played = play.Played
FullClear = play.FullClear
returned_resp = {
"BestDeaths": 0,
"BestPlaytime": 0,
"Clear": False,
"CurMap": query,
"CurMap": types.Map(**query),
"Difficulty": 0,
"Followed": False,
"Played": True,
@ -203,21 +288,29 @@ async def stopMapPlay(
playtime: int = Form(),
totalDeaths: int = Form(),
totalTime: int = Form(),
fullClear: int = Form(),
replayData: str = Form(),
Authorization: Union[str, None] = Header(default=None),
):
"""Saves the map replay, and informs the user if their play is a record"""
authcheck = db.auth_check(Authorization)
if not authcheck[0] and authcheck[1] == "nouser":
raise HTTPException(404, detail="User not found")
elif not authcheck[0] and authcheck[1] == "wrongpass":
raise HTTPException(403, detail="Wrong password")
if not authcheck[0]:
raise HTTPException(403, detail="Wrong username or password")
elif authcheck[0]:
NewMapRecord = False
FirstClear = False
userData = types.User(**authcheck[1])
userUpdateQuery = db.user_collection.update_one(
{"ID": userData.ID},
{
"$set": {
f"Plays.{mapID}": dict(types.MapPlay(Clear=bool(clear), FullClear=bool(fullClear)))
}
}
)
query = db.maps_collection.find_one(
{"ID": mapID, "Leaderboard.UserID": userData.ID}
)
@ -226,6 +319,7 @@ async def stopMapPlay(
BestUserTime = None
BestTime = None
BestFullTime = None
if query is not None and "Leaderboard" in query:
for i in query["Leaderboard"]:
if i["UserID"] == userData.ID:
@ -233,9 +327,12 @@ async def stopMapPlay(
BestUserTime = i["BestPlaytime"]
if BestTime is None or BestTime > i["BestPlaytime"]:
BestTime = i["BestPlaytime"]
elif BestFullTime is None or BestFullTime > i["BestFullTimePlaytime"]:
BestFullTime = i["BestFullTimePlaytime"]
if len(query["Leaderboard"]) <= 1 and i["UserID"] != userData.ID:
FirstClear = True
if BestUserTime is None or playtime < BestUserTime:
if (BestTime is None or playtime < BestTime) or (BestFullTime is None or playtime < BestFullTime):
updateQuery = db.maps_collection.update_one(
{"ID": mapID},
{"$pull": {"Leaderboard": {"UserID": userData.ID}},
@ -271,11 +368,12 @@ async def stopMapPlay(
BestReplay=replayData,
CreatorName=userData.Username,
UserID=userData.ID,
FullClear=bool(fullClear)
).dict()
},
},
)
if BestTime is None or playtime < BestTime:
if (BestTime is None or playtime < BestTime) or (BestFullTime is None or playtime < BestFullTime):
NewMapRecord = True
hook.execute_hooks(
@ -285,6 +383,7 @@ async def stopMapPlay(
clear=clear,
deaths=deaths,
playtime=playtime,
fullClear=fullClear,
FirstClear=FirstClear,
NewMapRecord=NewMapRecord,
)
@ -308,15 +407,16 @@ async def upload_map(
listed: int = Form(),
requiresCancels: int = Form(),
hideInChallenges: int = Form(),
tags: str = Form(),
tags: str = Form(default=""),
rng: int = Form(),
numSubmaps: int = Form(),
dragonCoins: int = Form(),
replayVisibility: int = Form(),
clientVersion: float = Form(),
):
authcheck = db.auth_check(Authorization)
if not authcheck[0] and authcheck[1] == "nouser":
raise HTTPException(404, detail="User not found")
elif not authcheck[0] and authcheck[1] == "wrongpass":
raise HTTPException(403, detail="Wrong password")
if not authcheck[0]:
raise HTTPException(403, detail="Wrong username or password")
elif authcheck[0]:
print(authcheck)
userData = types.User(**authcheck[1])
@ -454,13 +554,21 @@ async def rateMap(mapID: int, Rating: Rating, Authorization: Union[str, None] =
}
@app.get("/api/v1/map/{mapID}/besttimes/{maxEntries}")
async def getMapLeaderboard(mapID: int, maxEntries: int = 5):
async def getMapLeaderboard(mapID: int, maxEntries: int = 5, full_clear: int = 0):
"""Returns maxEntries records for the specified level"""
query = db.maps_collection.find_one({"ID": int(mapID)})
if not query:
raise HTTPException(404, detail="Map not found")
del query["_id"]
leaderboard = query["Leaderboard"]
leaderboard = []
for record in query["Leaderboard"]:
record = types.MapLeaderboard(**record)
if record.FullClear and full_clear == 1:
leaderboard.append(dict(record))
elif not record.FullClear and full_clear == 0:
leaderboard.append(dict(record))
else:
leaderboard.append(dict(record))
return sorted(leaderboard, key=lambda sort: sort["BestPlaytime"])[0:maxEntries]
@ -513,9 +621,7 @@ async def invalidateAllTimes(
authcheck = db.auth_check(Authorization)
if not authcheck[0] and authcheck[1] == "nouser":
raise HTTPException(404, detail="User not found")
elif not authcheck[0] and authcheck[1] == "wrongpass":
raise HTTPException(403, detail="Wrong password")
raise HTTPException(403, detail="Wrong username or password")
elif authcheck[0]:
userData = types.User(**authcheck[1])
if userData.Admin:
@ -562,9 +668,7 @@ async def reportContent(
authcheck = db.auth_check(Authorization)
if not authcheck[0] and authcheck[1] == "nouser":
raise HTTPException(404, detail="User not found")
elif not authcheck[0] and authcheck[1] == "wrongpass":
raise HTTPException(403, detail="Wrong password")
raise HTTPException(403, detail="Wrong user or password")
elif authcheck[0]:
db.reports_collection.insert_one(
{
@ -590,6 +694,17 @@ async def followcheck():
# FIXME: Stub
return 0
@app.post("/api/v1/unlockachievement", response_class=PlainTextResponse)
async def unlockachievement(
Authorization: Union[str, None] = Header(default=None),
achievementId: int = Form()
):
"""
Adds an achivement to the user
STUB
"""
return "Success"
def start():
"""Launched with `poetry run start` at root level"""
from . import __main__

5
docker-compose.yml
View file

@ -1,6 +1,7 @@
services:
backend:
build: .
# restart: on-failure
ports:
- "8001:8001"
volumes:
@ -15,7 +16,7 @@ services:
mongo:
image: mongo
restart: always
# restart: on-failure
expose:
- "27017:27017"
environment:
@ -26,7 +27,7 @@ services:
mongo-express:
image: mongo-express
restart: always
# restart: on-failure
ports:
- 8081:8081
environment: