feat: add api token authentication (4624215d) · Commits · INJECT / backend

aai/access.py

+3 −3

Original line number	Diff line number	Diff line
		@@ -64,7 +64,7 @@ def get_single_access(context: Request) -> Optional[ExerciseAccess]:

		def team_access(context: Request, team_id: int) -> ExerciseAccess:
		user = user_from_context(context)
		if user.group == User.AuthGroup.ADMIN:
		if user.group == User.AuthGroup.ADMIN or user.group == User.AuthGroup.BOT:
		exercise = ensure_exists(
		Exercise.objects.prefetch_related("teams").filter(
		teams__in=[team_id]
		@@ -109,7 +109,7 @@ def check_teams_access(context: Request, team_ids: List[str]) -> ExerciseAccess:

		def exercise_access(context: Request, exercise_id: int) -> ExerciseAccess:
		user = user_from_context(context)
		if user.group == User.AuthGroup.ADMIN:
		if user.group == User.AuthGroup.ADMIN or user.group == User.AuthGroup.BOT:
		exercise = ensure_exists(
		Exercise.objects.prefetch_related("teams").filter(id=exercise_id)
		)
		@@ -139,7 +139,7 @@ def exercise_access(context: Request, exercise_id: int) -> ExerciseAccess:

		def definition_access(context: Request, definition_id: int):
		user = user_from_context(context)
		if user.group == User.AuthGroup.ADMIN:
		if user.group == User.AuthGroup.ADMIN or user.group == User.AuthGroup.BOT:
		return

		condition = False

aai/middleware.py→aai/authentication_middleware.py

+64 −0

Original line number	Diff line number	Diff line
		from importlib import import_module

		from django.conf import settings
		from django.contrib.auth.hashers import check_password
		from django.contrib.auth.middleware import get_user
		from django.contrib.auth.models import AnonymousUser
		from django.contrib.sessions.backends.base import UpdateError
		from django.contrib.sessions.exceptions import SessionInterrupted
		from django.utils.deprecation import MiddlewareMixin
		from django.utils.functional import SimpleLazyObject

		from user.models import User

		class SessionMiddleware(MiddlewareMixin):

		class AuthenticationMiddleware(MiddlewareMixin):
		def __init__(self, get_response=None):
		super().__init__(get_response)
		engine = import_module(settings.SESSION_ENGINE)
		self.SessionStore = engine.SessionStore

		def token_auth(self, request, token: str):
		bot = AnonymousUser()
		for possible_bot in User.objects.filter(
		group=User.AuthGroup.BOT, is_active=True
		):
		if check_password(token, possible_bot.api_token):
		bot = possible_bot
		break

		request.user = SimpleLazyObject(lambda: bot)

		def session_auth(self, request, session_id: str):
		request.session = self.SessionStore(session_id)
		request.user = SimpleLazyObject(lambda: get_user(request))

		def process_request(self, request):
		setattr(request, "_dont_enforce_csrf_checks", True)
		sessionid = request.META.get("HTTP_SESSION_ID")
		request.session = self.SessionStore(sessionid)

		session_id = request.META.get("HTTP_SESSION_ID")
		if session_id is not None:
		self.session_auth(request, session_id)
		return

		request.session = self.SessionStore(None)
		api_token = request.META.get("HTTP_API_TOKEN")
		if api_token is not None:
		self.token_auth(request, api_token)
		return

		request.user = SimpleLazyObject(lambda: AnonymousUser())

		def process_response(self, request, response):
		modified = request.session.modified

aai/decorators.py

+16 −4

Original line number	Diff line number	Diff line
		@@ -90,13 +90,18 @@ def _get_user(args, *kwargs) -> Optional[User]:
		return user_from_context(context) if context is not None else None


		def protected(required_group: User.AuthGroup):
		def protected(
		required_group: User.AuthGroup,
		required_bot_permission: Optional[str] = None,
		):
		"""
		Decorator that checks whether user of request has required authorization group.
		If not, PermissionDenied exception is raised.

		Args:
		required_group: minimal required authorization group to access endpoint
		required_bot_permission: a TokenEndpoints value required to access this endpoint with a bot account,
		if unspecified, bot access is not allowed
		"""

		def decorator(func):
		@@ -114,10 +119,17 @@ def protected(required_group: User.AuthGroup):
		if action_name != "Query.resolve_exercise_time_left":
		logger.info(log_text)

		if user.group >= required_group:
		return func(args, *kwargs)
		if user.group == User.AuthGroup.BOT:
		if (
		required_bot_permission is None
		or required_bot_permission not in user.allowed_endpoints
		):
		raise PermissionDenied("Permission denied")
		elif user.group < required_group:
		raise PermissionDenied("Permission denied")

		return func(args, *kwargs)

		return wrapper

		return decorator

common_lib/schema/enums.py

+2 −1

Original line number	Diff line number	Diff line
		@@ -8,7 +8,7 @@ from running_exercise.models import (
		QuestionnaireAnswerStatus,
		MilestoneModificationDetails,
		)
		from user.models import User
		from user.models import User, EndpointPermissions


		class EventType(models.TextChoices):
		@@ -46,3 +46,4 @@ QuestionnaireAnswerStateEnum = graphene.Enum.from_enum(
		QuestionnaireAnswerStatus
		)
		CauseTypeEnum = graphene.Enum.from_enum(MilestoneModificationDetails.CauseType)
		EndpointPermissionsEnum = graphene.Enum.from_enum(EndpointPermissions)

common_lib/schema/types.py

+7 −1

Original line number	Diff line number	Diff line
		@@ -23,6 +23,7 @@ from common_lib.schema.enums import (
		LogTypeEnum,
		InjectTypeEnum,
		CauseTypeEnum,
		EndpointPermissionsEnum,
		)
		from common_lib.schema.inputs import TimeIntervalType
		from common_lib.utils import ensure_all_exist, ensure_exists
		@@ -1220,6 +1221,8 @@ class TUserType(DjangoObjectType):
		"definitions",
		"uploaded_definitions",
		"created_exercises",
		"allowed_endpoints",
		"api_token",
		]


		@@ -1239,11 +1242,14 @@ class IUserType(DjangoObjectType):
		)
		teams = graphene.List(graphene.NonNull(RestrictedTeam), required=True)
		created_by = graphene.Field(UserInterface)
		allowed_endpoints = graphene.List(
		graphene.NonNull(EndpointPermissionsEnum), required=True
		)

		class Meta:
		model = User
		interfaces = (UserInterface,)
		exclude = ["password", "key_values"]
		exclude = ["password", "key_values", "api_token"]

		def resolve_definitions(self, info):
		return self.definitions.all()