first commit

Author: zdszero

.gitignore

@@ -0,0 +1,7 @@
+.idea
+venv
+logs
+__pycache__
+*.pyc
+docker_deploy
+migrations

account/__init__.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class AccountConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'account'

account/apps.py

@@ -0,0 +1,163 @@
+import functools
+import hashlib
+import time
+
+from .models import User
+from problem.models import Problem
+from contest.models import Contest, ContestType, ContestStatus, ContestRuleType
+from utils.api import JSONResponse, APIError
+from utils.constants import CONTEST_PASSWORD_SESSION_KEY
+from .models import ProblemPermission
+
+
+class BasePermissionDecorator(object):
+    def __init__(self, func):
+        self.func = func
+
+    def __get__(self, obj, obj_type):
+        return functools.partial(self.__call__, obj)
+
+    def error(self, data):
+        return JSONResponse.response({"error": "permission-denied", "data": data})
+
+    def __call__(self, *args, **kwargs):
+        self.request = args[1]
+
+        if self.check_permission():
+            if self.request.user.is_disabled:
+                return self.error("Your account is disabled")
+            return self.func(*args, **kwargs)
+        else:
+            return self.error("Please login first")
+
+    def check_permission(self):
+        raise NotImplementedError()
+
+
+class login_required(BasePermissionDecorator):
+    def check_permission(self):
+        return self.request.user.is_authenticated
+
+
+class super_admin_required(BasePermissionDecorator):
+    def check_permission(self):
+        user = self.request.user
+        return user.is_authenticated and user.is_super_admin()
+
+
+class admin_role_required(BasePermissionDecorator):
+    def check_permission(self):
+        user = self.request.user
+        return user.is_authenticated and user.is_admin_role()
+
+
+class problem_permission_required(admin_role_required):
+    def check_permission(self):
+        if not super(problem_permission_required, self).check_permission():
+            return False
+        if self.request.user.problem_permission == ProblemPermission.NONE:
+            return False
+        return True
+
+
+def check_contest_password(password, contest_password):
+    if not (password and contest_password):
+        return False
+    if password == contest_password:
+        return True
+    else:
+        # sig#timestamp 这种形式的密码也可以，但是在界面上没提供支持
+        # sig = sha256(contest_password + timestamp)[:8]
+        if "#" in password:
+            s = password.split("#")
+            if len(s) != 2:
+                return False
+            sig, ts = s[0], s[1]
+
+            if sig == hashlib.sha256((contest_password + ts).encode("utf-8")).hexdigest()[:8]:
+                try:
+                    ts = int(ts)
+                except Exception:
+                    return False
+                return int(time.time()) < ts
+            else:
+                return False
+        else:
+            return False
+
+
+def check_contest_permission(check_type="details"):
+    """
+    只供Class based view 使用，检查用户是否有权进入该contest, check_type 可选 details, problems, ranks, submissions
+    若通过验证，在view中可通过self.contest获得该contest
+    """
+
+    def decorator(func):
+        def _check_permission(*args, **kwargs):
+            self = args[0]
+            request = args[1]
+            user = request.user
+            if request.data.get("contest_id"):
+                contest_id = request.data["contest_id"]
+            else:
+                contest_id = request.GET.get("contest_id")
+            if not contest_id:
+                return self.error("Parameter error, contest_id is required")
+
+            try:
+                # use self.contest to avoid query contest again in view.
+                self.contest = Contest.objects.select_related("created_by").get(id=contest_id, visible=True)
+            except Contest.DoesNotExist:
+                return self.error("Contest %s doesn't exist" % contest_id)
+
+            # Anonymous
+            if not user.is_authenticated:
+                return self.error("Please login first.")
+
+            # creator or owner
+            if user.is_contest_admin(self.contest):
+                return func(*args, **kwargs)
+
+            if self.contest.contest_type == ContestType.PASSWORD_PROTECTED_CONTEST:
+                # password error
+                if not check_contest_password(
+                        request.session.get(CONTEST_PASSWORD_SESSION_KEY, {}).get(self.contest.id),
+                        self.contest.password):
+                    return self.error("Wrong password or password expired")
+
+            # regular user get contest problems, ranks etc. before contest started
+            if self.contest.status == ContestStatus.CONTEST_NOT_START:
+                return self.error("Contest has not started yet.")
+
+
+            return func(*args, **kwargs)
+
+        return _check_permission
+
+    return decorator
+
+
+def ensure_created_by(obj, user):
+    e = APIError(msg=f"{obj.__class__.__name__} does not exist")
+    if not user.is_admin_role():
+        raise e
+    if user.is_super_admin():
+        return
+    if isinstance(obj, Problem):
+        if not user.can_mgmt_all_problem() and obj.created_by != user:
+            raise e
+    elif obj.created_by != user:
+        raise e
+
+
+# 确定对Lab Contest有管理权限
+def ensure_managed_by(contest: Contest, user: User):
+    e = APIError("Illegal Manager")
+    if not user.is_admin_role():
+        raise e
+    if user.is_super_admin():
+        return
+    if contest.created_by == user or str(user.id) in contest.contest_admin:
+        return
+    else:
+        raise e

account/decorators.py

@@ -0,0 +1,53 @@
+from django.conf import settings
+from django.db import connection
+from django.utils.timezone import now
+from django.utils.deprecation import MiddlewareMixin
+
+from utils.api import JSONResponse
+from account.models import User
+
+
+class APITokenAuthMiddleware(MiddlewareMixin):
+    def process_request(self, request):
+        appkey = request.META.get("HTTP_APPKEY")
+        if appkey:
+            try:
+                request.user = User.objects.get(open_api_appkey=appkey, open_api=True, is_disabled=False)
+                request.csrf_processing_done = True
+                request.auth_method = "api_key"
+            except User.DoesNotExist:
+                pass
+
+
+class SessionRecordMiddleware(MiddlewareMixin):
+    def process_request(self, request):
+        request.ip = request.META.get(settings.IP_HEADER, request.META.get("REMOTE_ADDR"))
+        if request.user.is_authenticated:
+            session = request.session
+            session["user_agent"] = request.META.get("HTTP_USER_AGENT", "")
+            session["ip"] = request.ip
+            session["last_activity"] = now()
+            user_sessions = request.user.session_keys
+            if session.session_key not in user_sessions:
+                user_sessions.append(session.session_key)
+                request.user.save()
+
+
+class AdminRoleRequiredMiddleware(MiddlewareMixin):
+    def process_request(self, request):
+        path = request.path_info
+        if path.startswith("/admin/") or path.startswith("/api/admin/"):
+            if not (request.user.is_authenticated and request.user.is_admin_role()):
+                return JSONResponse.response({"error": "login-required", "data": "Please login in first"})
+
+
+class LogSqlMiddleware(MiddlewareMixin):
+    def process_response(self, request, response):
+        print("\033[94m", "#" * 30, "\033[0m")
+        time_threshold = 0.03
+        for query in connection.queries:
+            if float(query["time"]) > time_threshold:
+                print("\033[93m", query, "\n", "-" * 30, "\033[0m")
+            else:
+                print(query, "\n", "-" * 30)
+        return response

account/middleware.py

@@ -0,0 +1,112 @@
+import uuid
+
+from django.db import models
+from django.contrib.auth.models import AbstractBaseUser
+
+# Create your models here.
+from utils.models import JSONField, ListFeild
+
+
+class AdminType(object):
+    REGULAR_USER = "Regular User" #student
+    ADMIN = "Admin" # tutor
+    SUPER_ADMIN = "Super Admin" #teacher
+
+class ProblemPermission(object):
+    NONE = "None"
+    OWN = "Own"
+    ALL = "All"
+
+class UserManager(models.Manager):
+    use_in_migrations = True
+
+    def get_by_natural_key(self, username):
+        return self.get(**{f"{self.model.USERNAME_FIELD}__iexact": username})
+
+class User(AbstractBaseUser):
+    #每次添加都会自动生成
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    username = models.TextField(unique=True)
+    email = models.TextField(null=True, unique=True)
+    create_time = models.DateTimeField(auto_now_add=True, null=True)
+    #UserType
+    admin_type = models.TextField(default=AdminType.REGULAR_USER)
+    problem_permission = models.TextField(default=ProblemPermission.NONE)
+    reset_password_token = models.TextField(null=True)
+    reset_password_token_expire_time = models.DateTimeField(null=True)
+    #SSO auth token
+    auth_token = models.TextField(null=True)
+    two_factor_auth = models.BooleanField(default=False)
+    tfa_token = models.TextField(null=True)
+    session_keys = JSONField(default=list)
+    # open api key
+    open_api = models.BooleanField(default=False)
+    open_api_appkey = models.TextField(null=True)
+    is_disabled = models.BooleanField(default=False)
+
+
+    USERNAME_FIELD = "username"
+    REQUIRED_FIELDS = []
+
+    objects = UserManager() #
+
+    def is_admin(self):
+        return self.admin_type == AdminType.ADMIN
+
+    def is_super_admin(self):
+        return self.admin_type == AdminType.SUPER_ADMIN
+
+    def is_admin_role(self):
+        return self.admin_type in [AdminType.ADMIN, AdminType.SUPER_ADMIN]
+
+    def can_mgmt_all_problem(self):
+        return self.problem_permission == ProblemPermission.ALL
+
+    def is_contest_admin(self, contest):
+        return self.is_authenticated and (str(self.id) in contest.contest_admin or self.admin_type == AdminType.SUPER_ADMIN)
+
+    class Meta:
+        db_table = "user"
+
+class UserProfile(models.Model):
+    user = models.OneToOneField(User, on_delete=models.CASCADE)
+    # Task_problems_status examples:
+    # {
+    #     "problems": {
+    #         "1": {
+    #             "status": JudgeStatus.ACCEPTED,
+    #             "Score:" 80,
+    #             "passed test case": [1, 2],
+    #             "_id": "1000"
+    #         }
+    #     },
+    #     "contest_problems": {
+    #         "1": {
+    #             "status": JudgeStatus.Error,
+    #             "Score:" 80,
+    #             "passed test case": [1, 2],
+    #             "_id": "1000"
+    #         }
+    #     }
+    # }
+    problems_status = JSONField(default=dict)
+    real_name = models.TextField(null=True)
+    blog = models.URLField(null=True)
+    github = models.TextField(null=True)
+    school = models.TextField(null=True)
+    major = models.TextField(null=True)
+    language = models.TextField(null=True)
+    #for Contest
+    accepted_number = models.IntegerField(default=0)
+    total_score = models.IntegerField(default=0)
+
+    def add_accepted_problem_number(self):
+        self.accepted_number = models.F("accepted_number") + 1
+        self.save()
+
+    def add_score(self, this_time_score, last_time_score=None):
+        last_time_score = last_time_score or 0
+        self.total_score = models.F("total_score") - last_time_score + this_time_score
+        self.save()
+    class Meta:
+        db_table = "user_profile"