diff --git a/src/pwncore/config.py b/src/pwncore/config.py index d514afe..f24f47e 100644 --- a/src/pwncore/config.py +++ b/src/pwncore/config.py @@ -1,22 +1,48 @@ -from __future__ import annotations - -import typing as t - -__all__ = ( - "Config", - "BaseConfig", -) - - -class Config(t.Protocol): +from dataclasses import dataclass + +""" +Sample messages: +"db_error": "An error occurred, please try again.", +"port_limit_reached": "Server ran out of ports 💀", +"ctf_not_found": "CTF does not exist.", +"container_start": "Container started.", +"container_stop": "Container stopped.", +"containers_team_stop": "All team containers stopped.", +"container_not_found": "You have no running containers for this CTF.", +"container_already_running": "Your team already has a running container for this CTF.", +"container_limit_reached": "Your team already has reached the maximum number" + " of containers limit, please stop other unused containers." +""" + +msg_codes = { + "db_error": 0, + "port_limit_reached": 1, + "ctf_not_found": 2, + "container_start": 3, + "container_stop": 4, + "containers_team_stop": 5, + "container_not_found": 6, + "container_already_running": 7, + "container_limit_reached": 8, + "hint_limit_reached": 9, +} + + +@dataclass +class Config: development: bool - - -class BaseConfig(Config): - __slots__ = ("development",) - - def __init__(self, development: bool) -> None: - self.development = development - - -DEV_CONFIG: t.Final[BaseConfig] = BaseConfig(True) + msg_codes: dict + db_url: str + docker_url: str | None + flag: str + max_containers_per_team: int + + +config = Config( + development=True, + db_url="sqlite://:memory:", + docker_url=None, # None for default system docker + flag="C0D", + max_containers_per_team=3, + msg_codes=msg_codes, +) diff --git a/src/pwncore/models/ctf.py b/src/pwncore/models/ctf.py index 5fd103a..4c787bd 100644 --- a/src/pwncore/models/ctf.py +++ b/src/pwncore/models/ctf.py @@ -34,27 +34,6 @@ class Hint(Model): class Meta: ordering = ("order",) - async def return_hint(problem_id: int, team_id: int): - viewed = await ViewedHint.filter(team_id=team_id).values_list( - "hint_id", flat=True - ) - if viewed: - viewed_hint = await Hint.filter( - problem_id=problem_id, id__in=viewed - ).values_list("order", flat=True) - if viewed_hint: - hint = await Hint.get_or_none( - problem_id=problem_id, order=max(viewed_hint) + 1 - ).values() - else: - hint = await Hint.get(problem_id=problem_id, order=0).values() - else: - hint = await Hint.get(problem_id=problem_id, order=0).values() - - if hint: - await ViewedHint.create(hint_id=hint["id"], team_id=team_id) - return hint - class SolvedProblem(Model): team: fields.ForeignKeyRelation[Team] = fields.ForeignKeyField("models.Team") diff --git a/src/pwncore/routes/ctf/__init__.py b/src/pwncore/routes/ctf/__init__.py index c84b05b..173dc16 100644 --- a/src/pwncore/routes/ctf/__init__.py +++ b/src/pwncore/routes/ctf/__init__.py @@ -1,9 +1,10 @@ from __future__ import annotations from fastapi import APIRouter +from fastapi import Response -from pwncore.models import * -import pwncore.config as config +from pwncore.models import Problem, SolvedProblem, Container, Hint, ViewedHint, Team +from pwncore.config import config import uuid # Metadata at the top for instant accessibility @@ -25,49 +26,67 @@ def get_team_id(): @router.get("/create") async def init_db(): - await Problem.create( - name="Invisible-Incursion", - description="Chod de tujhe se na ho paye", - author="Meetesh Saini", - points=300, - image_name="key:latest", - image_config={"PortBindings": {"22/tcp": [{}]}}, - ) - await Problem.create( - name="In-Plain-Sight", - description="A curious image with hidden secrets?", - author="KreativeThinker", - points=300, - image_name="key:latest", - image_config={"PortBindings": {"22/tcp": [{}]}}, - ) - await Team.create(name="CID Squad" + uuid.uuid4().hex, secret_hash="veryverysecret") - await Container.create( - docker_id="letsjustsay1", - flag="pwncore{this_is_a_test_flag}", - problem_id=1, - team_id=1, - ) - await Hint.create(order=0, problem_id=1, text="This is the first hint") - await Hint.create(order=1, problem_id=1, text="This is the second hint") - await Hint.create(order=2, problem_id=1, text="This is the third hint") - await Hint.create(order=0, problem_id=2, text="This is the third hint") - await Hint.create(order=1, problem_id=2, text="This is the third hint") + if config.development: + await Problem.create( + name="Invisible-Incursion", + description="Chod de tujhe se na ho paye", + author="Meetesh Saini", + points=300, + image_name="key:latest", + image_config={"PortBindings": {"22/tcp": [{}]}}, + ) + await Problem.create( + name="In-Plain-Sight", + description="A curious image with hidden secrets?", + author="KreativeThinker", + points=300, + image_name="key:latest", + image_config={"PortBindings": {"22/tcp": [{}]}}, + ) + await Team.create( + name="CID Squad" + uuid.uuid4().hex, secret_hash="veryverysecret" + ) + await Team.create( + name="Triple A battery" + uuid.uuid4().hex, secret_hash="chotiwali" + ) + await Container.create( + docker_id="letsjustsay1", + flag="pwncore{this_is_a_test_flag}", + problem_id=1, + team_id=1, + ) + await Container.create( + docker_id="letsjustsay2", + flag="pwncore{this_is_a_test_flag}", + problem_id=2, + team_id=1, + ) + await Container.create( + docker_id="letsjustsay3", + flag="pwncore{farmers}", + problem_id=2, + team_id=2, + ) + await Hint.create(order=0, problem_id=1, text="This is the first hint") + await Hint.create(order=1, problem_id=1, text="This is the second hint") + await Hint.create(order=2, problem_id=1, text="This is the third hint") + await Hint.create(order=0, problem_id=2, text="This is the first hint") + await Hint.create(order=1, problem_id=2, text="This is the second hint") + + +response = Response() @router.get("/list") async def ctf_list(): # Get list of ctfs problems = await Problem.all().values() + if not problems: + response.status_code = 404 + return {"msg_code": config.msg_codes["ctf_not_found"]} return problems -@router.get("/{ctf_id}") -async def ctf_get(ctf_id: int): - problem = await Problem.get_or_none(id=ctf_id) - return problem - - # @router.get("/flag/{ctf_id}") # async def flag_get(ctf_id: int): # flag = await Container.filter( @@ -78,13 +97,75 @@ async def ctf_get(ctf_id: int): @router.get("/flag/{ctf_id}/{flag}") async def flag_get(ctf_id: int, flag: str): - flag = await Container.check_flag( - flag=flag, team_id=get_team_id(), problem_id=ctf_id + problem = await Problem.get_or_none(id=ctf_id) + if not problem: + response.status_code = 404 + return {"msg_code": config.msg_codes["ctf_not_found"]} + + status = await Container.get_or_none( + team_id=get_team_id(), flag=flag, problem_id=ctf_id ) - return flag + if status: + await SolvedProblem.get_or_create(team_id=get_team_id(), problem_id=ctf_id) + return {"status": True} + return {"status": False} @router.get("/hint/{ctf_id}") async def hint_get(ctf_id: int): - hint = await Hint.return_hint(ctf_id, get_team_id()) + problem = await Problem.get_or_none(id=ctf_id) + if not problem: + response.status_code = 404 + return {"msg_code": config.msg_codes["ctf_not_found"]} + + viewed = await ViewedHint.filter(team_id=get_team_id()).values_list( + "hint_id", flat=True + ) + if viewed: + viewed_hint = await Hint.filter(problem_id=ctf_id, id__in=viewed).values_list( + "order", flat=True + ) + if viewed_hint: + hint = await Hint.get_or_none( + problem_id=ctf_id, order=max(viewed_hint) + 1 + ).values() + else: + hint = await Hint.get(problem_id=ctf_id, order=0).values() + else: + hint = await Hint.get(problem_id=ctf_id, order=0).values() + + if not hint: + response.status_code = 403 + return {"msg_code": config.msg_codes["hint_limit_reached"]} + + await ViewedHint.create(hint_id=hint["id"], team_id=get_team_id()) return hint + + +# @router.get("/completed/hint") +@router.get("/viewed_hints") +async def completed_hints_get(): + hints = await ViewedHint.filter(team_id=get_team_id()).values_list( + "hint_id", flat=True + ) + viewed_hints = [await Hint.get(id=x) for x in hints] + return viewed_hints + + +@router.get("/completed") +async def completed_problem_get(): + problems = await SolvedProblem.filter(team_id=get_team_id()).values_list( + "problem_id", flat=True + ) + solved_problems = [await Problem.get(id=x) for x in problems] + return solved_problems + + +@router.get("/{ctf_id}") +async def ctf_get(ctf_id: int): + problem = await Problem.get_or_none(id=ctf_id) + if not problem: + response.status_code = 404 + return {"msg_code": config.msg_codes["ctf_not_found"]} + + return problem