docs: added documentation to all API endpoint routes so far

src/pwncore/models/statusCode_models.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+from typing import List, Optional
+
+class ErrorResponse(BaseModel):
+    """
+    Generic error response
+    msg_code: 0 (db_error), 2 (ctf_not_found)
+    """
+    msg_code: int
+
+class ContainerStartResponse(BaseModel):
+    """
+    Response for container start operation
+    msg_code: 3 (success), 7 (already running), 8 (limit reached), 0 (db_error)
+    """
+    msg_code: int
+    ports: Optional[List[int]] = None
+    ctf_id: Optional[int] = None
+
+class ContainerStopResponse(BaseModel):
+    """
+    Response for container stop operations
+    msg_code: 4 (stop success), 5 (stopall success), 6 (not found), 0 (db_error)
+    """
+    msg_code: int

src/pwncore/routes/admin.py

@@ -4,6 +4,8 @@
 from fastapi import APIRouter, Request, Response
 from passlib.hash import bcrypt
 from tortoise.transactions import atomic, in_transaction
+from pydantic import BaseModel
+from typing import Optional
 
 import pwncore.containerASD as containerASD
 from pwncore.config import config
@@ -52,14 +54,26 @@ async def _del_cont(id: str):
     await container.delete()
 
 
+class AdminResponse(BaseModel):
+    success: bool
+    message: Optional[str] = None
+
+# shorten response_description
 @atomic()
-@router.get("/union")
+@router.get("/union",
+    response_model=AdminResponse,
+    response_description="""Successful calculation of team points and coin updates.
+
+    Response parameters: boolean `success`, `message`
+
+    Note: Returns 401 if authentication fails.
+    """)
 async def calculate_team_coins(
     response: Response, req: Request
 ):  # Inefficient, anyways will be used only once
     if not bcrypt.verify((await req.body()).strip(), ADMIN_HASH):
         response.status_code = 401
-        return
+        return AdminResponse(success=False, message="Authentication failed")
     async with in_transaction():
         logging.info("Calculating team points form pre-event CTFs:")
         team_ids = await Team.filter().values_list("id", flat=True)
@@ -83,14 +97,24 @@ async def calculate_team_coins(
             logging.info(f"{team.id}) {team.name}: {team.coins}")
             await team.save()
 
+    return AdminResponse(success=True, message="Team coins updated successfully")
+
+# shorten response_description
+@router.get("/create",
+    response_model=AdminResponse,
+    response_description="""Database initialization with sample data.
+
+    Response parameters: boolean `success`, `message`
 
-@router.get("/create")
+    Note: Returns 401 if authentication fails.
+    This endpoint should only be used in development environment.
+    """)
 async def init_db(
     response: Response, req: Request
 ):  # Inefficient, anyways will be used only once
     if not bcrypt.verify((await req.body()).strip(), ADMIN_HASH):
         response.status_code = 401
-        return
+        return AdminResponse(success=False, message="Authentication failed")
     await Problem.create(
         name="Invisible-Incursion",
         description="Chod de tujhe se na ho paye",
@@ -208,3 +232,5 @@ async def init_db(
     await SolvedProblem.create(team_id=2, problem_id=1)
     await SolvedProblem.create(team_id=2, problem_id=2)
     await SolvedProblem.create(team_id=1, problem_id=2)
+
+    return AdminResponse(success=True, message="Database initialized with sample data")

src/pwncore/routes/auth.py

@@ -23,6 +23,7 @@
 logger = getLogger(__name__)
 
 
+# defining Pydantic response models
 class AuthBody(BaseModel):
     name: str
     password: str
@@ -34,12 +35,72 @@ class SignupBody(BaseModel):
     tags: set[str]
 
 
+# Response Models
+class SignupResponse(BaseModel):
+    """
+    msg_code: 13 (signup_success)
+    """
+    msg_code: t.Literal[13]
+
+class SignupErrorUsersNotFound(BaseModel):
+    """
+    msg_code: 24 (users_not_found)
+    """
+    msg_code: t.Literal[24]
+    tags: list[str]
+
+class SignupErrorUsersInTeam(BaseModel):
+    """
+    msg_code: 20 (user_already_in_team)
+    """
+    msg_code: t.Literal[20]
+    tags: list[str]
+
+class LoginResponse(BaseModel):
+    """
+    msg_code: 15 (login_success)
+    """
+    msg_code: t.Literal[15]
+    access_token: str
+    token_type: str
+
+class ErrorResponse(BaseModel):
+    """
+    msg_code can be:
+    0 (db_error)
+    17 (team_exists)
+    10 (team_not_found)
+    14 (wrong_password)
+    """
+    msg_code: t.Literal[0, 17, 10, 14]
+
+
 def normalise_tag(tag: str):
     return tag.strip().casefold()
 
-
+# shorten response_description
 @atomic()
-@router.post("/signup")
+@router.post("/signup",
+    response_model=SignupResponse,
+    responses={
+        406: {"model": ErrorResponse},
+        404: {"model": SignupErrorUsersNotFound},
+        401: {"model": SignupErrorUsersInTeam},
+        500: {"model": ErrorResponse}
+    },
+    response_description="""Create a new team with associated members.
+
+    Parameters:
+        - in request: `name`, `password`, `tags`
+        - in response: `msg_code`, `tags` _only in error responses_
+
+    msg_codes for Responses:
+    - 200: Successful signup : 13
+    - 406: Team already exists: 17
+    - 404: Users not found: 24
+    - 401: Users already in team: 20
+    - 500: Database error: 0
+    """)
 async def signup_team(team: SignupBody, response: Response):
     team.name = team.name.strip()
     members = set(map(normalise_tag, team.tags))
@@ -82,8 +143,25 @@ async def signup_team(team: SignupBody, response: Response):
         return {"msg_code": config.msg_codes["db_error"]}
     return {"msg_code": config.msg_codes["signup_success"]}
 
-
-@router.post("/login")
+# shorten response_description
+@router.post("/login",
+    response_model=LoginResponse,
+    responses={
+        404: {"model": ErrorResponse},
+        401: {"model": ErrorResponse}
+    },
+    response_description="""Authenticate a team and receive a JWT token.
+
+    Parameters:
+        - in request: `name`, `password`
+        - in successful response: `msg_code`, `access_token`,`token_type`
+        - in error responses: `msg_code`
+
+    msg_codes for Responses:
+    - 200: Successful login: 15
+    - 404: Team not found: 10
+    - 401: Wrong password: 14
+    """)
 async def team_login(team_data: AuthBody, response: Response):
     # TODO: Simplified logic since we're not doing refresh tokens.
 

src/pwncore/routes/ctf/__init__.py

@@ -52,8 +52,14 @@ def _invalid_order():
 class Flag(BaseModel):
     flag: str
 
-
-@router.get("/completed")
+# shorten response_description
+@router.get(
+    "/completed",
+    response_model=list[Problem_Pydantic],
+    response_description="""Returns all problems solved by the authenticated team.
+
+    Response Parameters: (6) `id`, `name`, `description`, `points`, `category`, `visible`
+    """)
 async def completed_problem_get(jwt: RequireJwt):
     team_id = jwt["team_id"]
     ViewedHint.filter(team_id=team_id).annotate()
@@ -62,8 +68,16 @@ async def completed_problem_get(jwt: RequireJwt):
     )
     return problems
 
-
-@router.get("/list")
+# shorten response_description
+@router.get(
+    "/list",
+    response_model=list[Problem_Pydantic],
+    response_description="""Returns all visible CTF problems with adjusted points based on hints used.
+
+    Response Parameters: (6) `id`, `name`, `description`, `points`, `category`, `visible`
+
+    Note: Points are adjusted based on hints viewed by the team.
+    """)
 async def ctf_list(jwt: RequireJwt):
     team_id = jwt["team_id"]
     problems = await Problem_Pydantic.from_queryset(Problem.filter(visible=True))
@@ -88,9 +102,23 @@ async def update_points(req: Request, ctf_id: int):
     except Exception:
         logger.exception("An error occured while updating points")
 
-
+# shorten response_description
 @atomic()
-@router.post("/{ctf_id}/flag")
+@router.post(
+    "/{ctf_id}/flag",
+    response_model=dict[str, bool | str],
+    response_description="""Submit a flag for a specific CTF problem.
+
+    Parameters:
+        - in request: string `flag`
+        - in response: boolean `status`
+
+    msg_codes for Error responses:
+    - 404: ctf_not_found : 2 
+    - 401: ctf_solved : 12 
+    - 500: db_error : 0
+    - 400: container_not_found : 6 
+    """)
 async def flag_post(
     req: Request, ctf_id: int, flag: Flag, response: Response, jwt: RequireJwt
 ):
@@ -136,9 +164,20 @@ async def flag_post(
         return {"status": True}
     return {"status": False}
 
-
+# shorten response_description
 @atomic()
-@router.get("/{ctf_id}/hint")
+@router.get(
+    "/{ctf_id}/hint",
+    response_model=dict[str, str | int],
+    response_description="""Retrieve the next available hint for a problem.
+
+    Response Parameters: `text`, `order`
+
+    msg_code for Error responses:
+    - 404: {"msg_code": 2} - ctf_not_found
+    - 403: {"msg_code": 9} - hint_limit_reached
+    - 400: {"msg_code": 22} - Insufficient coins
+    """)
 async def hint_get(ctf_id: int, response: Response, jwt: RequireJwt):
     team_id = jwt["team_id"]
     problem = await Problem.exists(id=ctf_id, visible=True)
@@ -176,17 +215,32 @@ async def hint_get(ctf_id: int, response: Response, jwt: RequireJwt):
         "order": hint.order,
     }
 
-
-@router.get("/{ctf_id}/viewed_hints")
+# shorten response_description
+@router.get(
+    "/{ctf_id}/viewed_hints",
+    response_model=list[Hint_Pydantic],
+    response_description="""Get all hints viewed by the team for a specific problem.
+
+    Response Parameters: `id`, `text`, `order`, `problem_id`
+    """)
 async def viewed_problem_hints_get(ctf_id: int, jwt: RequireJwt):
     team_id = jwt["team_id"]
     viewed_hints = await Hint_Pydantic.from_queryset(
         Hint.filter(problem_id=ctf_id, viewedhints__team_id=team_id)
     )
     return viewed_hints
 
-
-@router.get("/{ctf_id}")
+# shorten response_description
+@router.get(
+    "/{ctf_id}",
+    response_model=Problem_Pydantic,
+    response_description="""Get details of a specific CTF problem.
+
+    Response Parameters: (6) `id`, `name`, `description`, `points`, `category`, `visible`
+
+    msg_code:
+    - if ctf_not_found or not visible : 2
+    """)
 async def ctf_get(ctf_id: int, response: Response):
     problem = await Problem_Pydantic.from_queryset(
         Problem.filter(id=ctf_id, visible=True)