(Feat) Add Basic API on RTB Objects

.gitignore

@@ -78,3 +78,5 @@ __pycache__
 
 # Readme
 !*README.md
+
+venv/*

handlers/ApiV2/BoxApi.py

@@ -0,0 +1,101 @@
+import json
+from ..BaseHandlers import BaseHandler
+from libs.SecurityDecorators import apikey, restrict_ip_address
+from models.Box import Box
+from models.Corporation import Corporation
+from models.GameLevel import GameLevel
+import logging
+from models import dbsession
+
+logger = logging.getLogger()
+
+
+class BoxApiHandler(BaseHandler):
+
+    @apikey
+    @restrict_ip_address
+    def get(self, id=None):
+        if id is None or id == "":
+            data = {"data": [box.to_dict() for box in Box.all()]}
+        else:
+            box = Box.by_id(id)
+            if box is not None:
+                data = {"data": box.to_dict()}
+            else:
+                data = {"message": "Box not found"}
+        self.write(json.dumps(data))
+
+    @apikey
+    @restrict_ip_address
+    def post(self, *args, **kwargs):
+        data = json.loads(self.request.body)
+        logger.info(f"Posted data : {data}")
+        if "corporation" not in data:
+            data = {
+                "data": {"corporation": None},
+                "message": "Corporation is required",
+            }
+            self.write(json.dumps(data))
+            return
+        if "name" not in data:
+            data = {"data": {"box": None}, "message": "Name is required"}
+            self.write(json.dumps(data))
+            return
+
+        if Box.by_name(data["name"]) is not None:
+            data = {
+                "data": {"box": None},
+                "message": "This box already exists",
+            }
+            self.write(json.dumps(data))
+            return
+
+        if Corporation.by_name(data["corporation"]) is None:
+            data = {
+                "data": {"corporation": data["corporation"]},
+                "message": "This corporation does not exist",
+            }
+            self.write(json.dumps(data))
+            return
+
+        new_box = Box()
+        new_box.name = data["name"]
+        new_box.description = data["description"] if "description" in data else ""
+        new_box.corporation_id = (
+            Corporation.by_name(data["corporation"]).id
+            if "corporation" in data
+            else None
+        )
+
+        flag_submission_type = (
+            data["flag_submission_type"]
+            if "flag_submission_type" in data
+            else "CLASSIC"
+        )
+        if flag_submission_type not in ["CLASSIC", "TOKEN"]:
+            new_box.flag_submission_type = "CLASSIC"
+        else:
+            new_box.flag_submission_type = flag_submission_type
+
+        new_box.game_level_id = GameLevel.by_number(0).id
+
+        dbsession.add(new_box)
+        dbsession.commit()
+        data = {
+            "data": {"box": new_box.to_dict()},
+            "message": "This box has been created",
+        }
+        self.write(json.dumps(data))
+
+    @apikey
+    @restrict_ip_address
+    def delete(self, id: str):
+        raise NotImplementedError()
+
+    @apikey
+    @restrict_ip_address
+    def put(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    def check_xsrf_cookie(self):
+        pass

handlers/ApiV2/CorporationApi.py

@@ -0,0 +1,90 @@
+import json
+from ..BaseHandlers import BaseHandler
+from libs.SecurityDecorators import apikey, restrict_ip_address
+from models.Corporation import Corporation
+import logging
+from models import Team, dbsession
+
+logger = logging.getLogger()
+
+
+class CorporationApiHandler(BaseHandler):
+
+    @apikey
+    @restrict_ip_address
+    def get(self, id: str = None):
+        if id is None or id == "":
+            data = {
+                "data": [corporation.to_dict() for corporation in Corporation.all()]
+            }
+        else:
+            corporation = Corporation.by_id(id)
+            if corporation is not None:
+                data = {"data": corporation.to_dict()}
+            else:
+                data = {"message": "Corporation not found"}
+        self.write(json.dumps(data))
+
+    @apikey
+    @restrict_ip_address
+    def post(self, *args, **kwargs):
+        data = json.loads(self.request.body)
+        logger.info(f"Post data : {data}")
+
+        if Corporation.by_name(data["name"]) is not None:
+            data = {
+                "data": {"corporation": data["name"]},
+                "message": "This corporation already exists",
+            }
+            self.write(json.dumps(data))
+            return
+
+        new_corporation = Corporation()
+        new_corporation.name = data["name"]
+        new_corporation.locked = data["locked"] if "locked" in data else False
+        new_corporation.description = (
+            data["description"] if "description" in data else ""
+        )
+
+        dbsession.add(new_corporation)
+        dbsession.commit()
+        data = {
+            "data": {
+                "corporation": new_corporation.to_dict(),
+            },
+            "message": "This corporation has been created",
+        }
+        self.write(json.dumps(data))
+
+    @apikey
+    @restrict_ip_address
+    def delete(self, id: str):
+        corporation = Corporation.by_id(id)
+        if corporation is not None:
+            dbsession.delete(corporation)
+            dbsession.commit()
+            self.write(
+                json.dumps(
+                    {
+                        "data": {"corporation": corporation.to_dict()},
+                        "message": "Corporation deleted",
+                    }
+                )
+            )
+        else:
+            self.write(
+                json.dumps(
+                    {
+                        "data": {"corporation": None},
+                        "message": "Corporation not found",
+                    }
+                )
+            )
+
+    @apikey
+    @restrict_ip_address
+    def put(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    def check_xsrf_cookie(self):
+        pass

handlers/ApiV2/FlagApi.py

@@ -0,0 +1,86 @@
+import json
+from ..BaseHandlers import BaseHandler
+from libs.SecurityDecorators import apikey, restrict_ip_address
+from models.Box import Box
+from models.Flag import Flag
+import logging
+from models import dbsession
+
+
+logger = logging.getLogger()
+
+
+class FlagApiHandler(BaseHandler):
+
+    @apikey
+    @restrict_ip_address
+    def get(self, id=None):
+        if id is None or id == "":
+            data = {"data": [flag.to_dict() for flag in Flag.all()]}
+        else:
+            flag = Flag.by_id(id)
+            if flag is not None:
+                data = {"data": flag.to_dict()}
+            else:
+                data = {"message": "Flag not found"}
+        self.write(json.dumps(data))
+
+    @apikey
+    @restrict_ip_address
+    def post(self, *args, **kwargs):
+        data = json.loads(self.request.body)
+        logger.info(f"Post data : {data}")
+
+        if "box" not in data:
+            data = {
+                "data": {"box": None},
+                "message": "Box is required",
+            }
+            self.write(json.dumps(data))
+            return
+
+        if Box.by_name(data["box"]) is None:
+            data = {
+                "data": {"box": data["box"]},
+                "message": "This box does not exist",
+            }
+            self.write(json.dumps(data))
+            return
+        box = Box.by_name(data["box"])
+
+        if Flag.by_token_and_box_id(data["token"], box.id) is not None:
+            data = {
+                "data": {
+                    "flag": data["token"],
+                    "box": data["box"],
+                },
+                "message": "This flag already exists",
+            }
+            self.write(json.dumps(data))
+            return
+
+        new_flag = Flag()
+        new_flag.name = data["name"]
+        new_flag.token = data["token"]
+        new_flag.value = int(data["value"]) if "value" in data else 1
+        new_flag.box_id = box.id
+        new_flag.type = "static"
+        new_flag.description = data["description"] if "description" in data else ""
+
+        dbsession.add(new_flag)
+        dbsession.commit()
+        data = {"data": new_flag.to_dict(), "message": "This flag has been created"}
+        self.write(json.dumps(data))
+
+    @apikey
+    @restrict_ip_address
+    def delete(self, id: str):
+        raise NotImplementedError()
+
+    @apikey
+    @restrict_ip_address
+    def put(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    def check_xsrf_cookie(self):
+        pass

handlers/ApiV2/TeamApi.py

@@ -0,0 +1,43 @@
+import json
+from ..BaseHandlers import BaseHandler
+from libs.SecurityDecorators import apikey, restrict_ip_address
+from models.Corporation import Corporation
+import logging
+from models import Team, dbsession
+
+logger = logging.getLogger()
+
+
+class TeamApiHandler(BaseHandler):
+
+    @apikey
+    @restrict_ip_address
+    def get(self, id: str = None):
+        with_flags = self.get_argument("with_flags", "false").lower() == "true"
+        if id is None or id == "":
+            data = {"data": [team.to_dict(with_flags) for team in Team.all()]}
+        else:
+            team = Team.by_uuid(id)
+            if team is not None:
+                data = {"data": team.to_dict(with_flags)}
+            else:
+                data = {"message": "Team not found"}
+        self.write(json.dumps(data))
+
+    @apikey
+    @restrict_ip_address
+    def post(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    @apikey
+    @restrict_ip_address
+    def delete(self, id: str):
+        raise NotImplementedError()
+
+    @apikey
+    @restrict_ip_address
+    def put(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    def check_xsrf_cookie(self):
+        pass

handlers/__init__.py

@@ -56,6 +56,7 @@
 from handlers.StaticFileHandler import StaticFileHandler
 from handlers.UpgradeHandlers import *
 from handlers.UserHandlers import *
+from handlers.ApiV2 import FlagApi, TeamApi, CorporationApi, BoxApi
 from libs.ConsoleColors import *
 from libs.DatabaseConnection import DatabaseConnection
 from libs.Scoreboard import Scoreboard, score_bots
@@ -195,6 +196,10 @@ def get_cookie_secret():
     (r"/admin/resetdelete", AdminResetDeleteHandler),
     # API handlers - APIHandlers.py
     (r"/api/actions", APIActionHandler),
+    (r"/api/v2/corporation/?(.*)", CorporationApi.CorporationApiHandler),
+    (r"/api/v2/box/?(.*)", BoxApi.BoxApiHandler),
+    (r"/api/v2/flag/?(.*)", FlagApi.FlagApiHandler),
+    (r"/api/v2/team/?(.*)", TeamApi.TeamApiHandler),
     # Error handlers - ErrorHandlers.py
     (r"/403", UnauthorizedHandler),
     (r"/gamestatus", StopHandler),

models/Corporation.py

@@ -119,6 +119,8 @@ def to_dict(self):
             "uuid": self.uuid,
             "name": self.name,
             "description": self.description,
+            "id": self.id,
+            "locked": self.locked,
             # "boxes": [box.uuid for box in self.boxes],
         }
 

models/Team.py

@@ -292,7 +292,7 @@ def file_by_file_name(self, file_name):
         ls = self.files.filter_by(file_name=file_name)
         return ls[0] if 0 < len(ls) else None
 
-    def to_dict(self):
+    def to_dict(self, with_flags: bool = False):
         """Use for JSON related tasks; return public data only"""
         return {
             "uuid": self.uuid,
@@ -301,6 +301,13 @@ def to_dict(self):
             "money": self.money,
             "avatar": self.avatar,
             "notes": self.notes,
+            "score": {
+                "money": self.get_score("money"),
+                "flags": self.get_score("flag"),
+                "hints": self.get_score("hint"),
+                "bots": self.get_score("bot"),
+            },
+            "flags": [flag.to_dict() for flag in self.flags] if with_flags else [],
         }
 
     def to_xml(self, parent):