Authorise on the acting role, and stop an upload from being a page

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)

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)
 

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

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

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 {}))

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('<script>alert(1)</script>')
+
+    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())