Socket rewrite

.gitignore

@@ -1,3 +1,4 @@
 streets.txt
 __pycache__
-.idea
+.idea
+*.pem

game.py

@@ -28,11 +28,25 @@ class GameState(Enum):
     FINISHED = 2
 
 
+def int_to_hex_str(x: int):
+    hex_x = format(x, "x")
+    return hex_x if len(hex_x) > 1 else f"0{hex_x}"
+
+
 class Color:
     def __init__(self, red: int, green: int, blue: int):
         self.red = red
         self.green = green
         self.blue = blue
+        hex_red = int_to_hex_str(red)
+        hex_green = int_to_hex_str(green)
+        hex_blue = int_to_hex_str(blue)
+        self.hex = f'#{hex_red}{hex_green}{hex_blue}'
+
+    def get_dict(self):
+        return {
+            "color": self.hex
+        }
 
 
 class DefaultColors(Enum):
@@ -45,11 +59,23 @@ class DefaultColors(Enum):
 
 
 class User:
-    def __init__(self, username: str):
+    def __init__(self, username: str, password: str):
         self.username = username
+        self.password = password
         self.color = None
         self.initiative = None
         self.money = None
+        self.token = str(uuid4())
+        self.in_room = ""
+        self.room_password = ""
+
+    def get_dict(self):
+        return {
+            "username": self.username,
+            "color": get_color(self.color).get_dict(),
+            "initiative": self.initiative,
+            "money": self.money,
+        }
 
 
 class StreetType(Enum):
@@ -65,6 +91,14 @@ class StreetType(Enum):
     TO_PRISON = 9
 
 
+def get_color(color: Color | DefaultColors) -> Color:
+    if color is None:
+        return Color(0, 0, 0)
+    if color in list(DefaultColors):
+        return color.value
+    return color
+
+
 class Street:
     def __init__(self, street_type: StreetType, name: str, color: Color, price: int, home_price: int,
                  hotel_price: int, mortgage: int, rent: tuple):
@@ -78,6 +112,19 @@ class Street:
         self.rent = rent
         self.owner = None
 
+    def get_dict(self):
+        return {
+            "street_type": self.street_type.name,
+            "name": self.name,
+            "color": get_color(self.color).get_dict(),
+            "price": self.price,
+            "home_price": self.home_price,
+            "hotel_price": self.hotel_price,
+            "mortgage": self.mortgage,
+            "rent": self.rent,
+            "owner": self.owner
+        }
+
 
 def str_to_street_type(s: str):
     return StreetType[s.upper()]
@@ -120,15 +167,30 @@ def roll_dices(dice: int, n: int):
 roll_2d6 = roll_dices(6, 2)
 
 
+def get_dict_user_or_none(user: User | None):
+    if user:
+        return user.get_dict()
+    return user
+
+
 class Room:
     def __init__(self, password: str = ""):
-        self.room_id = uuid4()
+        self.room_id = str(uuid4())
         self.users = []
         self.state = GameState.WAITING
         self.password = password
         self.current_player = None
         self.streets = copy_streets()
 
+    def get_dict(self):
+        return {
+            "room_id": self.room_id,
+            "users": [user.get_dict()for user in self.users],
+            "state": self.state.name,
+            "current_player": self.current_player.get_dict() if self.current_player else self.current_player,
+            "streets": [street.get_dict() for street in self.streets]
+        }
+
     def check_color(self, color: Color):
         return color not in [user.color for user in self.users]
 

main.py

@@ -12,59 +12,129 @@
 #
 #     You should have received a copy of the GNU Affero General Public License
 #     along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import logging
 
 from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
 from fastapi_socketio import SocketManager
-from cachetools import TTLCache
 from game import Room, User, load_streets
+from objects import ErrCode, UserCheck
 
+logging.basicConfig(level=logging.DEBUG, format="%(levelname)s:\t%(asctime)s\t%(message)s")
 load_streets()
 
 app = FastAPI()
-sio = SocketManager(app=app)
-sio_sessions = TTLCache(maxsize=10000, ttl=24 * 60 * 60)
+origins = ["*"]
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+sio = SocketManager(app=app, mount_location="/ws", socketio_path="game", cors_allowed_origins=[])
+# sio_sessions = TTLCache(maxsize=10000, ttl=24 * 60 * 60)
 rooms = []
+users = []  # TODO: implement as dict
+
+
+# TODO: Time-based tokens
+def check_user(username, password=None, token=None):
+    if not (password or token):
+        return False
+    else:
+        global users
+        selected_users = tuple(filter(lambda usr: usr.username == username, users))
+        if len(selected_users) == 0:
+            if token:
+                logging.debug(f"Invalid token for user {username}")
+                return UserCheck.INVALID_CREDENTIALS
+            logging.debug(f"Invalid user {username}")
+            return UserCheck.USER_DOESNT_EXISTS
+        else:
+            user = selected_users[0]
+            if user.token == token or user.password == password:
+                logging.debug(f"Valid credentials for user {username}")
+                return user
+            else:
+                logging.debug(f"Invalid token or password for user {username}:\n{user.token} --- {token}")
+                return UserCheck.INVALID_CREDENTIALS
+
+
+@sio.on("createRoom")
+async def sio_create_room(sid, username, token, password):
+    match check_user(username, token=token):
+        case User(username=username) as user:
+            new_room = Room(password)
+            logging.info(f"User {username} created room {new_room.room_id}")
+            new_room.add_user(user)
+            rooms.append(new_room)
+            user.in_room = new_room.room_id  # TODO: not add user on creation
+            user.room_password = new_room.password
+            await sio_send_user_info(sid, user)
+        case _:
+            return False  # TODO: some errors
+
+async def sio_send_user_info(sid, user):
+    userdata = user.get_dict()
+    userdata["token"] = user.token
+    if user.in_room:
+        userdata["in_room"] = user.in_room
+        userdata["password"] = user.password
+    await sio.emit("userInfo", userdata, room=sid)
+
+
+@sio.on("joinRoom")
+async def sio_join_room(sid, room_id, room_password):  # TODO: Check if user already in room
+    global rooms
+    selected_room = tuple(filter(lambda room: room.room_id == room_id, rooms))
+    if len(selected_room) == 0:
+        await sio_throw_error(sid, ErrCode.ROOM_DOESNT_EXIST)
+    else:
+        room = selected_room[0]
+        room_data = room.get_dict()
+        sio.enter_room(sid, room.room_id)
+        if room.password == room_password:
+            await sio.emit("roomInfo", room_data, room=sid)
 
 
 @sio.on("connect")
-async def sio_connect(sid):
-    sio_sessions[sid] = {
-        "sid": sid,
-        "user": None,
-        "room": None,
-    }
+async def sio_connect(sid, _, auth):
+    if not ("username" in auth and ("password" in auth or "token" in auth))\
+            or not auth["username"].isalnum()\
+            or len(auth["username"]) > 16:
+        await sio_throw_error(sid, ErrCode.INVALID_CREDENTIALS)
+        return False
+    username = auth["username"]
+    token = auth.get("token", None)
+    password = auth.get("password", None)
+    match check_user(username, password, token):
+        case User(username=username) as user:
+            await sio_send_user_info(sid, user)
+        case UserCheck.USER_DOESNT_EXISTS:
+            if not password:
+                return False
+            user = User(username, password)
+            users.append(user)
+            await sio_send_user_info(sid, user)
+        case UserCheck.INVALID_CREDENTIALS:
+            await sio_throw_error(sid, ErrCode.INVALID_CREDENTIALS)
+            return False
+        case _:
+            await sio_throw_error(sid, ErrCode.UNKNOWN_ERROR)
+            return False
+    logging.debug(f"{sid} connected as {username}")
+
+
+async def sio_throw_error(sid, error):
+    await sio.emit("error", error.name, room=sid)
 
 
 @sio.on("disconnect")
 async def sio_disconnect(sid):
-    if sid not in sio_sessions[sid]:
-        return
-    room = sio_sessions[sid].get("room", None)
-    user = sio_sessions[sid].get("user", None)
-    if room is not None:
-        room.remove_user(user)
-        if not room.users:
-            del room
-        del user
-    del sio_sessions[sid]
-
-
-@sio.on("joinRoom")
-async def sio_join_room(sid, username: str = "", password: str = ""):
-    pass
-
-
-@sio.on("createRoom")
-async def sio_create_room(sid, username: str = "", password: str = ""):
-    if sid not in sio_sessions[sid]:
-        return
-    new_room = Room(password)
-    new_user = User(username)
-    new_room.add_user(new_user)
-    sio_sessions[sid]["room"] = new_room
-    sio_sessions[sid]["user"] = new_user
+    logging.debug(f"{sid} disconnected")
 
 
 @sio.on("rollDices")
 async def sio_roll_dices(sid, token: str = ""):
-    pass
+    print("test")

objects.py

@@ -0,0 +1,15 @@
+from enum import Enum
+
+
+class UserCheck(Enum):
+    USER_DOESNT_EXISTS = 0
+    INVALID_CREDENTIALS = 1
+
+
+class ErrCode(Enum):
+    UNKNOWN_ERROR = 0
+    USER_ALREADY_EXISTS = 1
+    ALREADY_LOGGED_IN = 2
+    INVALID_CREDENTIALS = 3
+    INVALID_TOKEN = 4
+    ROOM_DOESNT_EXIST = 10