From f9f65ae6a9c05bef7e78464e8aa06d9466b26889 Mon Sep 17 00:00:00 2001 From: Vikrant Kumar Date: Wed, 30 Sep 2026 12:04:54 +0530 Subject: [PATCH 1/4] Authorise on the role being acted in, not every role ever held MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit role_required checked the set of designations somebody holds, so a student who also holds an office kept that office's authority while acting as a student — and anything reaching their session reached the office's endpoints with it. It now checks the acting role, defaulting to the office when the switcher has never been used, so nobody is locked out. Keyed on HoldsDesignation.working rather than user, which is what the model's own docstring asks for and what the IAM already projects: reading the other one authorises somebody on leave and refuses their stand-in. update_last_selected_role accepted any string and stored it. That was harmless while it only drove the sidebar and a free promotion once authorisation reads it, so a post nobody holds is now refused. --- FusionIIIT/applications/globals/api/views.py | 29 +++++++-- FusionIIIT/applications/globals/decorators.py | 59 ++++++++++++++----- 2 files changed, 68 insertions(+), 20 deletions(-) diff --git a/FusionIIIT/applications/globals/api/views.py b/FusionIIIT/applications/globals/api/views.py index 160455d0c..a2bf22689 100644 --- a/FusionIIIT/applications/globals/api/views.py +++ b/FusionIIIT/applications/globals/api/views.py @@ -16,6 +16,7 @@ import logging import secrets import re +import threading from datetime import timedelta from django.conf import settings from django.core.mail import EmailMultiAlternatives @@ -36,7 +37,7 @@ from .utils import get_and_authenticate_user from notifications.models import Notification from notifications.signals import notify -from applications.globals.decorators import role_required +from applications.globals.decorators import held_designations, role_required from applications.globals import iam_bridge from applications.globals.programme_scope import ( ALL_ACAD_ROLES, @@ -154,7 +155,8 @@ def auth_view(request): 'accessible_modules': accessible_modules, 'last_selected_role': last_selected_role, 'must_complete_profile': must_complete_profile, - # The menus plugged modules contribute, so the sidebar can expand them. + # The menus plugged modules contribute, under the designation that grants + # each one, so a role never draws another role's screens. 'plugged_navigation': plugged_nav, } @@ -223,6 +225,12 @@ def update_last_selected_role(request): if new_role is None: return Response({'error': 'last_selected_role is required'}, status=status.HTTP_400_BAD_REQUEST) + # Only a post this person actually holds may be acted in; authorisation + # reads this column, so an unchecked write here is a free promotion. + if not held_designations(request.user) & {str(new_role).strip().lower()}: + return Response({'error': 'You do not hold that role.'}, + status=status.HTTP_403_FORBIDDEN) + extra_info = get_object_or_404(ExtraInfo, user=request.user) extra_info.last_selected_role = new_role @@ -718,6 +726,16 @@ def _otp_hash(otp: str) -> str: def _token_hash(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() + +def _deliver_otp_email(email, username): + """Run in a thread; a delivery failure must never reach the caller, who is + told the same thing whether or not the address exists.""" + try: + email.send() + except Exception: + _security_log.exception("OTP email delivery failed | user=%s", username) + + _safe_eq = hmac.compare_digest _SEND_OTP_OK = { @@ -837,8 +855,11 @@ def password_reset_send_otp(request): ) email.attach_alternative(html_content, "text/html") - email.send() - + # Off the request path: the OTP is already saved and the reply is the + # same either way, so the caller was only ever waiting on SMTP. + threading.Thread(target=_deliver_otp_email, args=(email, user.username), + daemon=True).start() + except Exception: _security_log.exception("OTP email delivery failed | user=%s", user.username) diff --git a/FusionIIIT/applications/globals/decorators.py b/FusionIIIT/applications/globals/decorators.py index 53e123fae..97230782b 100644 --- a/FusionIIIT/applications/globals/decorators.py +++ b/FusionIIIT/applications/globals/decorators.py @@ -5,30 +5,57 @@ from applications.globals.models import HoldsDesignation +#: Held by definition rather than assigned, so they carry no HoldsDesignation row. +BASIC_ROLES = {'student', 'faculty', 'staff'} -def role_required(allowed_roles): + +def held_designations(user): + """Every post this user may act in, lowercased. + + Keyed on `working`, not `user`: HoldsDesignation's own docstring says + "Use 'working' to handle permissions in code", because `working` is whoever + currently occupies the post and `user` is its substantive holder. Reading + the wrong one authorises the person on leave and refuses their stand-in, + and disagrees with the IAM, which projects on `working`. + """ + names = (HoldsDesignation.objects + .select_related('designation') + .filter(working=user) + .values_list('designation__name', flat=True)) + return {name.lower() for name in names if name} + + +def active_designation(user): + """The post this user is currently acting in. + + Their own choice when they have made one and still hold it. Otherwise an + office outranks the basic role, which is the rule the IAM applies too — so + somebody who has never touched the role switcher is not locked out of the + post they hold. """ - Decorator factory that accepts a list of allowed role names. - Accepts multiple HoldsDesignation records per user. + held = held_designations(user) + chosen = getattr(getattr(user, 'extrainfo', None), 'last_selected_role', None) + if chosen and chosen.lower() in held: + return chosen.lower() + return next(iter(sorted(held - BASIC_ROLES)), None) + + +def role_required(allowed_roles): + """Allow the request only if the role being acted in is one of these. + + The set of posts somebody holds is not the same question. Checking the set + means a student who also holds an office keeps that office's authority while + acting as a student — so anything that reaches their session, an XSS above + all, reaches the office's endpoints too. """ allowed_lower = {role.lower() for role in allowed_roles} def decorator(view_func): @wraps(view_func) def _wrapped_view(request, *args, **kwargs): - # Fetch all designations for this user - user_roles = ( - HoldsDesignation.objects - .select_related('designation') - .filter(user=request.user) - .values_list('designation__name', flat=True) # or whichever field holds the string - ) - - # Normalize to lowercase for comparison - user_roles_lower = {r.lower() for r in user_roles} - - # Check intersection - if not (user_roles_lower & allowed_lower): + acting = active_designation(request.user) + + if acting is None or acting not in allowed_lower: return Response( {"error": "Permission denied: one of %s required" % allowed_roles}, status=status.HTTP_403_FORBIDDEN From c39bc7e30e134b8ac9353c621d7e08f994e617db Mon Sep 17 00:00:00 2001 From: Vikrant Kumar Date: Wed, 30 Sep 2026 12:05:06 +0530 Subject: [PATCH 2/4] Hand an uploaded file over as a download, never as a page Uploads are served from the same origin as the portal and every module framed inside it, so a file served inline runs as the application: an uploaded .html or .svg would execute in this origin and could read the API token the client keeps in localStorage. Forcing attachment, nosniff and a sandbox policy takes that away whatever the file claims to be, and holds whether or not the media route is reachable. --- FusionIIIT/Fusion/urls.py | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/FusionIIIT/Fusion/urls.py b/FusionIIIT/Fusion/urls.py index 07490f8b4..4b1139036 100755 --- a/FusionIIIT/Fusion/urls.py +++ b/FusionIIIT/Fusion/urls.py @@ -26,6 +26,24 @@ from applications.globals.views import RateLimitedPasswordResetView +def serve_media(request, path, document_root=None): + """Hand an uploaded file over as a download, never as a page. + + Uploads share this origin with the portal and with every module framed + inside it, so anything served inline here runs as the application: an + uploaded .html or .svg would execute in the app's own origin and could read + the API token the client keeps. Forcing a download and refusing content-type + sniffing takes that away whatever the file claims to be. + """ + from django.views.static import serve + + response = serve(request, path, document_root=document_root) + response['Content-Disposition'] = 'attachment' + response['X-Content-Type-Options'] = 'nosniff' + response['Content-Security-Policy'] = "default-src 'none'; sandbox" + return response + + urlpatterns = [ url(r'^', include('applications.globals.urls')), url(r'^feeds/', include('applications.feeds.urls')), @@ -94,4 +112,5 @@ ), name='password_reset_complete', ), -] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT) +] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT, + view=serve_media) From 181fc7b64c139fc41b7f118dc5ee4b0b1e5be45d Mon Sep 17 00:00:00 2001 From: Vikrant Kumar Date: Wed, 30 Sep 2026 12:05:06 +0530 Subject: [PATCH 3/4] Expire API tokens, and make the session cookie secure by default A DRF token never expires, and this one lives in localStorage where any script on the origin can read it, so a copy taken once works for good. The class is here; the settings switch is deployment's to make. The IAM session cookie was secure only when an environment variable said so. It now follows DEBUG, still overridable, so one plain-HTTP request no longer puts the session on the wire. --- .../applications/globals/authentication.py | 36 +++++++++++++++++++ FusionIIIT/applications/globals/iam_bridge.py | 19 +++++++--- 2 files changed, 51 insertions(+), 4 deletions(-) create mode 100644 FusionIIIT/applications/globals/authentication.py diff --git a/FusionIIIT/applications/globals/authentication.py b/FusionIIIT/applications/globals/authentication.py new file mode 100644 index 000000000..115fac7cf --- /dev/null +++ b/FusionIIIT/applications/globals/authentication.py @@ -0,0 +1,36 @@ +"""API tokens that stop working. + +DRF's own token never expires, so a copy taken once is a credential for good — +and this one is kept in the browser's localStorage, where any script running on +the origin can read it. An age limit turns a permanent compromise into a bounded +one, and re-authenticating is the only thing the holder has to do about it. + +Switch it on with, in the REST_FRAMEWORK setting: + + 'DEFAULT_AUTHENTICATION_CLASSES': ( + 'applications.globals.authentication.ExpiringTokenAuthentication', + ) +""" +import os +from datetime import timedelta + +from django.utils import timezone +from rest_framework.authentication import TokenAuthentication +from rest_framework.exceptions import AuthenticationFailed + +#: Long enough for a working day, short enough that a stolen token dies. +TOKEN_TTL_HOURS = int(os.environ.get("API_TOKEN_TTL_HOURS", "12")) + + +class ExpiringTokenAuthentication(TokenAuthentication): + def authenticate_credentials(self, key): + user, token = super().authenticate_credentials(key) + + if TOKEN_TTL_HOURS and token.created < ( + timezone.now() - timedelta(hours=TOKEN_TTL_HOURS)): + # Deleted, not just refused: the row is what makes it work, and + # leaving it lets a later change quietly bring the token back. + token.delete() + raise AuthenticationFailed("Session expired. Please sign in again.") + + return user, token diff --git a/FusionIIIT/applications/globals/iam_bridge.py b/FusionIIIT/applications/globals/iam_bridge.py index fab98bc32..2ed0e5d98 100644 --- a/FusionIIIT/applications/globals/iam_bridge.py +++ b/FusionIIIT/applications/globals/iam_bridge.py @@ -24,7 +24,18 @@ #: Blank in development; set to .iiitdmj.ac.in in production. COOKIE_DOMAIN = os.environ.get("IAM_AUTH_COOKIE_DOMAIN", "") COOKIE_MAX_AGE = int(os.environ.get("IAM_AUTH_COOKIE_MAX_AGE", str(12 * 60 * 60))) -COOKIE_SECURE = os.environ.get("IAM_AUTH_COOKIE_SECURE", "0") == "1" +def _cookie_secure(): + """Secure unless this is a debug server, and overridable either way. + + Defaulting it off meant one plain-HTTP request put the session on the wire, + and a cookie that still works is a thing nobody notices. + """ + override = os.environ.get("IAM_AUTH_COOKIE_SECURE") + if override is not None: + return override == "1" + from django.conf import settings + + return not getattr(settings, "DEBUG", False) def _url(path): @@ -51,7 +62,7 @@ def attach_session(response, token): if not token: return response kwargs = {"max_age": COOKIE_MAX_AGE, "httponly": True, - "samesite": "Lax", "secure": COOKIE_SECURE, "path": "/"} + "samesite": "Lax", "secure": _cookie_secure(), "path": "/"} if COOKIE_DOMAIN: kwargs["domain"] = COOKIE_DOMAIN response.set_cookie(COOKIE_NAME, token, **kwargs) @@ -82,7 +93,7 @@ def _me(token): def plugged_modules(token): - """(modules by designation, navigation) from the other services, in one call.""" + """(modules, navigation) from the other services, both keyed by designation.""" payload = _me(token) or {} return (dict(payload.get("modules_by_role") or {}), - list(payload.get("navigation") or [])) + dict(payload.get("navigation_by_role") or {})) From 7d9895449be7d6028589f35eeeef97ab55ab9b11 Mon Sep 17 00:00:00 2001 From: Vikrant Kumar Date: Wed, 30 Sep 2026 12:05:06 +0530 Subject: [PATCH 4/4] Send the OTP off the request path, and cover the security rules with tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reply waited on the full SMTP round-trip although the OTP row is already saved before the send and the caller is told the same thing either way — so the wait bought nothing. It now goes to a thread. Tests for the parts that decide access: the acting role, a stand-in against the absent holder, a role no longer held, an upload served as a download, and a token past its life. --- FusionIIIT/applications/globals/tests.py | 136 ++++++++++++++++++++++- 1 file changed, 135 insertions(+), 1 deletion(-) diff --git a/FusionIIIT/applications/globals/tests.py b/FusionIIIT/applications/globals/tests.py index 19dbde223..6d05c60ca 100644 --- a/FusionIIIT/applications/globals/tests.py +++ b/FusionIIIT/applications/globals/tests.py @@ -1,5 +1,23 @@ -from django.test import SimpleTestCase +import shutil +import tempfile +import threading +import time +from pathlib import Path +from unittest.mock import patch +from datetime import timedelta + +from django.contrib.auth.models import User +from django.utils import timezone +from rest_framework.authtoken.models import Token +from rest_framework.exceptions import AuthenticationFailed +from django.test import RequestFactory, SimpleTestCase, TestCase + +from Fusion.urls import serve_media +from applications.globals.authentication import ExpiringTokenAuthentication +from applications.globals.api.views import _deliver_otp_email +from applications.globals.decorators import active_designation, held_designations +from applications.globals.models import Designation, ExtraInfo, HoldsDesignation from applications.globals.programme_scope import ( canonical_programme_name, programme_display_name, @@ -17,3 +35,119 @@ def test_phd_display_name_is_stable_for_legacy_values(self): programme_display_name(value), 'Doctor of Philosophy', ) + + +class OtpDeliveryTests(SimpleTestCase): + """The reply must not wait on SMTP: the OTP row is saved before the send, + and the caller is told the same thing whether or not delivery works.""" + + class _SlowMail: + def __init__(self): + self.sent = threading.Event() + + def send(self): + time.sleep(0.3) + self.sent.set() + + def test_a_slow_mail_host_does_not_hold_the_request(self): + mail = self._SlowMail() + started = time.perf_counter() + worker = threading.Thread(target=_deliver_otp_email, args=(mail, "u1"), + daemon=True) + worker.start() + dispatch = time.perf_counter() - started + + self.assertLess(dispatch, 0.1) + self.assertTrue(mail.sent.wait(timeout=5)) + + def test_a_delivery_failure_is_logged_and_swallowed(self): + class Broken: + def send(self): + raise OSError("smtp unreachable") + + with patch("applications.globals.api.views._security_log") as log: + _deliver_otp_email(Broken(), "u1") + self.assertTrue(log.exception.called) + + +class ActingRoleTests(TestCase): + """Authority follows the role being acted in, not every role ever held.""" + + def setUp(self): + self.user = User.objects.create_user(username='intern') + ExtraInfo.objects.create(id='intern', user=self.user, user_type='student') + for name in ('student', 'acadadmin'): + HoldsDesignation.objects.create( + user=self.user, working=self.user, + designation=Designation.objects.create(name=name)) + + def _act_as(self, role): + self.user.extrainfo.last_selected_role = role + self.user.extrainfo.save(update_fields=['last_selected_role']) + self.user.refresh_from_db() + + def test_an_office_held_but_not_acted_in_grants_nothing(self): + self._act_as('student') + self.assertEqual(active_designation(self.user), 'student') + + def test_the_office_applies_while_it_is_the_acting_role(self): + self._act_as('acadadmin') + self.assertEqual(active_designation(self.user), 'acadadmin') + + def test_never_having_switched_does_not_lock_the_office_out(self): + """The column is null for anyone who has not used the switcher.""" + self.assertEqual(active_designation(self.user), 'acadadmin') + + def test_a_role_that_is_no_longer_held_is_ignored(self): + self._act_as('acadadmin') + HoldsDesignation.objects.filter( + designation__name='acadadmin').delete() + self.assertIsNone(active_designation(self.user)) + + def test_a_stand_in_is_authorised_and_the_absent_holder_is_not(self): + """`working` is the occupant; `user` is the substantive holder.""" + absent = User.objects.create_user(username='absent') + post = Designation.objects.create(name='registrar') + HoldsDesignation.objects.create( + user=absent, working=self.user, designation=post) + self.assertIn('registrar', held_designations(self.user)) + self.assertNotIn('registrar', held_designations(absent)) + + +class MediaServingTests(TestCase): + """An upload shares this origin with the portal, so it must not be a page.""" + + def setUp(self): + self.media = tempfile.mkdtemp() + Path(self.media, 'evil.html').write_text('') + + def tearDown(self): + shutil.rmtree(self.media, ignore_errors=True) + + def test_an_uploaded_page_is_handed_over_as_a_download(self): + request = RequestFactory().get('/media/evil.html') + response = serve_media(request, 'evil.html', document_root=self.media) + + self.assertEqual(response['Content-Disposition'], 'attachment') + self.assertEqual(response['X-Content-Type-Options'], 'nosniff') + self.assertIn('sandbox', response['Content-Security-Policy']) + + +class ExpiringTokenTests(TestCase): + """A token kept in localStorage must not be a credential for ever.""" + + def setUp(self): + self.user = User.objects.create_user(username='tokenholder') + self.token = Token.objects.create(user=self.user) + self.auth = ExpiringTokenAuthentication() + + def test_a_token_within_its_life_still_works(self): + user, _ = self.auth.authenticate_credentials(self.token.key) + self.assertEqual(user, self.user) + + def test_an_old_token_is_refused_and_destroyed(self): + Token.objects.filter(pk=self.token.pk).update( + created=timezone.now() - timedelta(hours=48)) + with self.assertRaises(AuthenticationFailed): + self.auth.authenticate_credentials(self.token.key) + self.assertFalse(Token.objects.filter(pk=self.token.pk).exists())