|
1 | 1 | import json |
2 | 2 | import logging |
3 | 3 | import re |
| 4 | +from typing import Any |
4 | 5 |
|
5 | 6 | import httpx |
6 | 7 | from django.conf import settings |
7 | 8 | from django.contrib.auth import get_user_model |
8 | 9 | from django.contrib.auth.models import AbstractUser |
9 | 10 | from django.contrib.auth.password_validation import validate_password |
10 | 11 | from django.core.exceptions import ValidationError |
| 12 | +from rest_framework.authentication import SessionAuthentication |
11 | 13 | from rest_framework.response import Response |
12 | | -from typing import Any |
13 | 14 |
|
14 | 15 | from apps.web.models import Student |
15 | 16 |
|
| 17 | +logger = logging.getLogger(__name__) |
| 18 | + |
16 | 19 | AUTH_SETTINGS = settings.AUTH |
17 | 20 | PASSWORD_LENGTH_MIN = AUTH_SETTINGS["PASSWORD_LENGTH_MIN"] |
18 | 21 | PASSWORD_LENGTH_MAX = AUTH_SETTINGS["PASSWORD_LENGTH_MAX"] |
|
23 | 26 | QUEST_BASE_URL = QUEST_SETTINGS["BASE_URL"] |
24 | 27 |
|
25 | 28 |
|
| 29 | +class CSRFCheckSessionAuthentication(SessionAuthentication): |
| 30 | + def authenticate(self, request): |
| 31 | + super().enforce_csrf(request) |
| 32 | + return super().authenticate(request) |
| 33 | + |
| 34 | + |
26 | 35 | def get_survey_details(action: str) -> dict[str, Any] | None: |
27 | 36 | """ |
28 | 37 | A single, clean function to get all survey details for a given action. |
29 | | - Valid actions: "signup", "login", "reset". |
| 38 | + Valid actions: "signup", "login", "reset_password". |
30 | 39 | """ |
31 | 40 |
|
32 | 41 | action_details = QUEST_SETTINGS.get(action.upper()) |
33 | 42 |
|
34 | 43 | if not action_details: |
35 | | - logging.error("Invalid quest action requested: %s", action) |
| 44 | + logger.error("Invalid quest action requested: %s", action) |
36 | 45 | return None |
37 | 46 |
|
38 | 47 | try: |
39 | 48 | question_id = int(action_details.get("QUESTIONID")) |
40 | 49 | except (ValueError, TypeError): |
41 | | - logging.error( |
| 50 | + logger.error( |
42 | 51 | "Could not parse 'QUESTIONID' for action '%s'. Check your settings.", action |
43 | 52 | ) |
44 | 53 | return None |
@@ -66,18 +75,18 @@ async def verify_turnstile_token( |
66 | 75 | }, |
67 | 76 | ) |
68 | 77 | if not response.json().get("success"): |
69 | | - logging.warning("Turnstile verification failed: %s", response.json()) |
| 78 | + logger.warning("Turnstile verification failed: %s", response.json()) |
70 | 79 | return False, Response( |
71 | 80 | {"error": "Turnstile verification failed"}, status=403 |
72 | 81 | ) |
73 | 82 | return True, None |
74 | 83 | except httpx.TimeoutException: |
75 | | - logging.error("Turnstile verification timed out") |
| 84 | + logger.error("Turnstile verification timed out") |
76 | 85 | return False, Response( |
77 | 86 | {"error": "Turnstile verification timed out"}, status=504 |
78 | 87 | ) |
79 | | - except Exception as e: |
80 | | - logging.error(f"Error verifying Turnstile token: {e}") |
| 88 | + except Exception: |
| 89 | + logger.error("Turnstile verification error") |
81 | 90 | return False, Response({"error": "Turnstile verification error"}, status=500) |
82 | 91 |
|
83 | 92 |
|
@@ -132,19 +141,19 @@ async def get_latest_answer( |
132 | 141 | response.raise_for_status() # Raise an exception for bad status codes |
133 | 142 | full_data = response.json() |
134 | 143 | except httpx.TimeoutException: |
135 | | - logging.exception("Questionnaire API query timed out") |
| 144 | + logger.error("Questionnaire API query timed out") |
136 | 145 | return None, Response( |
137 | 146 | {"error": "Questionnaire API query timed out"}, |
138 | 147 | status=504, |
139 | 148 | ) |
140 | | - except httpx.RequestError as e: |
141 | | - logging.exception(f"Error querying questionnaire API: {e}") |
| 149 | + except httpx.RequestError: |
| 150 | + logger.error("Error querying questionnaire API") |
142 | 151 | return None, Response( |
143 | 152 | {"error": "Failed to query questionnaire API"}, |
144 | 153 | status=500, |
145 | 154 | ) |
146 | | - except Exception as e: |
147 | | - logging.exception(f"An unexpected error occurred: {e}") |
| 155 | + except Exception: |
| 156 | + logger.error("An unexpected error occurred") |
148 | 157 | return None, Response({"error": "An unexpected error occurred"}, status=500) |
149 | 158 |
|
150 | 159 | # Filter and return only the required fields from the first row |
@@ -180,7 +189,7 @@ async def get_latest_answer( |
180 | 189 | key in filtered_data and filtered_data[key] is not None |
181 | 190 | for key in ["id", "submitted_at", "account", "otp"] |
182 | 191 | ): |
183 | | - logging.warning("Missing required field(s) in questionnaire response") |
| 192 | + logger.warning("Missing required field(s) in questionnaire response") |
184 | 193 | return None, Response( |
185 | 194 | {"error": "Missing required field(s) in questionnaire response"}, |
186 | 195 | status=400, |
@@ -211,7 +220,8 @@ def rate_password_strength(password: str) -> int: |
211 | 220 | if re.search(r"[^a-zA-Z0-9\s]", password): |
212 | 221 | score += 1 |
213 | 222 |
|
214 | | - length_step = (PASSWORD_LENGTH_MAX - PASSWORD_LENGTH_MIN) // 10 |
| 223 | + length_range = max(1, PASSWORD_LENGTH_MAX - PASSWORD_LENGTH_MIN) |
| 224 | + length_step = max(1, length_range // 10) |
215 | 225 |
|
216 | 226 | score += (len(password) - PASSWORD_LENGTH_MIN) // length_step |
217 | 227 |
|
|
0 commit comments