become_user API endpoint hardening

arxiv-auth/src/accounts/config.py

@@ -58,6 +58,15 @@
 AUTH_SESSION_COOKIE_DOMAIN = os.environ.get('AUTH_SESSION_COOKIE_DOMAIN', f'.{BASE_SERVER}')
 AUTH_SESSION_COOKIE_SECURE = bool(int(os.environ.get('AUTH_SESSION_COOKIE_SECURE', '1')))
 MASQUERADE_COOKIE_NAME = 'MASQUERADE'
+BECOME_USER_CSRF_COOKIE_NAME = os.environ.get(
+    'BECOME_USER_CSRF_COOKIE_NAME',
+    'ARXIV_BECOME_USER_CSRF'
+)
+"""Cookie name for the CSRF token required by `/become_user`."""
+BECOME_USER_FRESH_AUTH_SECONDS = int(
+    os.environ.get('BECOME_USER_FRESH_AUTH_SECONDS', '600')
+)
+"""Max allowed seconds since auth session start for `/become_user`."""
 
 
 #################### Classic Auth ####################

arxiv-auth/src/accounts/routes/ui.py

@@ -5,9 +5,11 @@
 from functools import wraps
 from pytz import timezone, UTC
 import logging
+import secrets
+import hmac
 
 from flask import Blueprint, render_template, request, \
-    make_response, redirect, current_app, Response
+    make_response, redirect, current_app, Response, jsonify
 
 from arxiv import status
 
@@ -33,6 +35,71 @@
 blueprint = Blueprint('ui', __name__, url_prefix='')
 
 
+def _set_become_user_csrf_cookie(response: Response, token: str) -> None:
+    cookie_name = current_app.config['BECOME_USER_CSRF_COOKIE_NAME']
+    params = {
+        'httponly': False,
+        'domain': current_app.config['AUTH_SESSION_COOKIE_DOMAIN'],
+    }
+    if current_app.config['AUTH_SESSION_COOKIE_SECURE']:
+        params.update({'secure': True, 'samesite': 'lax'})
+    response.set_cookie(cookie_name, token, max_age=3600, **params)
+
+
+def _issue_become_user_csrf_token(response: Response) -> str:
+    token = secrets.token_urlsafe(32)
+    _set_become_user_csrf_cookie(response, token)
+    return token
+
+
+def _fresh_auth_required() -> bool:
+    """Set BECOME_USER_FRESH_AUTH_SECONDS to 0 to disable fresh-auth checks."""
+    return bool(current_app.config.get('BECOME_USER_FRESH_AUTH_SECONDS', 600) > 0)
+
+
+def _is_fresh_auth(session_start_iso: str) -> bool:
+    if not _fresh_auth_required():
+        return True
+    if not session_start_iso:
+        return False
+    try:
+        start = datetime.fromisoformat(session_start_iso)
+    except ValueError:
+        return False
+    if start.tzinfo is None:
+        start = start.replace(tzinfo=UTC)
+    max_age = int(current_app.config['BECOME_USER_FRESH_AUTH_SECONDS'])
+    return datetime.now(tz=UTC) - start <= timedelta(seconds=max_age)
+
+
+def _reauth_redirect() -> Response:
+    referrer = request.referrer or ''
+    next_page = good_next_page(referrer)
+    login_url = f"/login?next_page={next_page}"
+    return make_response(redirect(login_url, code=status.HTTP_303_SEE_OTHER))
+
+
+def _valid_become_user_csrf() -> bool:
+    cookie_name = current_app.config['BECOME_USER_CSRF_COOKIE_NAME']
+    cookie_token = request.cookies.get(cookie_name, '')
+    request_token = request.form.get('csrf_token', '') \
+        or request.headers.get('X-CSRF-Token', '')
+    return bool(
+        cookie_token and request_token
+        and hmac.compare_digest(cookie_token, request_token)
+    )
+
+
+def unset_become_user_csrf_cookie(response: Response) -> None:
+    cookie_name = current_app.config['BECOME_USER_CSRF_COOKIE_NAME']
+    domain = current_app.config['AUTH_SESSION_COOKIE_DOMAIN']
+    # Clear with no domain, with leading-dot domain, and without leading dot
+    # to ensure the cookie is removed regardless of how the browser stored it.
+    response.set_cookie(key=cookie_name, value='', max_age=0, httponly=False)
+    response.set_cookie(key=cookie_name, value='', max_age=0, httponly=False, domain=domain)
+    response.set_cookie(key=cookie_name, value='', max_age=0, httponly=False,
+                        domain=domain.lstrip('.'))
+
 
 def anonymous_only(func: Callable) -> Callable:
     """Redirect logged-in users to their profile."""
@@ -136,6 +203,7 @@ def login() -> Response:
     if code is status.HTTP_303_SEE_OTHER:
         response = make_response(redirect(safe_page, code=code))
         set_cookies(response, data)
+        _issue_become_user_csrf_token(response)
         unset_submission_cookie(response)    # Fix for ARXIVNG-1149
         return response
 
@@ -162,6 +230,7 @@ def logout() -> Response:
         unset_submission_cookie(response)  # Fix for ARXIVNG-1149.
         unset_permanent_cookie(response)  # Partial fix for ARXIVNG-1653, ARXIVNG-1644
         unset_masquerade_cookie(response)
+        unset_become_user_csrf_cookie(response)
         return response
     return redirect(safe_page, code=status.HTTP_302_FOUND)
 
@@ -172,11 +241,37 @@ def auth_status() -> Response:
     return make_response("OK")
 
 
+@blueprint.route('/become_user/csrf', methods=['GET'])
+def become_user_csrf() -> Response:
+    """Issue a CSRF token for `/become_user`. Requires admin (flag_edit_users)."""
+    if not request.auth:
+        return Response("Unauthorized", status=status.HTTP_401_UNAUTHORIZED)
+    user_id = request.auth.user.user_id if request.auth.user else None
+    if not user_id:
+        return Response("Unauthorized", status=status.HTTP_401_UNAUTHORIZED)
+    admin_user = db.session.query(DBUser) \
+        .filter(DBUser.user_id == int(user_id)) \
+        .filter(DBUser.flag_edit_users == 1) \
+        .filter(DBUser.flag_deleted == 0) \
+        .filter(DBUser.flag_banned == 0) \
+        .filter(DBUser.flag_approved == 1) \
+        .first()
+    if not admin_user:
+        return Response("Forbidden", status=status.HTTP_403_FORBIDDEN)
+    token = secrets.token_urlsafe(32)
+    response = make_response(jsonify({'csrf_token': token}), status.HTTP_200_OK)
+    _set_become_user_csrf_cookie(response, token)
+    return response
+
+
 # Only use post in production to avoid caching issues in fastly,
 #   but can include GET in dev for testing.
 @blueprint.route('/become_user', methods=['POST'])
 def become_user_become_user_id() -> Response:
 
+    if not _valid_become_user_csrf():
+        return Response("Forbidden", status=status.HTTP_403_FORBIDDEN)
+
     become_user_id = int(request.args.get('become_user_id'))
 
     classic_cookie_name = current_app.config['CLASSIC_COOKIE_NAME']
@@ -225,9 +320,14 @@ def become_user_become_user_id() -> Response:
     valid_user = False
     if session_cookie:
 
-        data = jwt.decode(session_cookie, secret, algorithms=["HS256"])
+        try:
+            data = jwt.decode(session_cookie, secret, algorithms=["HS256"])
+        except Exception:
+            return _reauth_redirect()
         if DEBUG:
             print("BU-DEBUG: jwt decode session_cookie:", data)
+        if not _is_fresh_auth(data.get('start_time')):
+            return _reauth_redirect()
 
         user_id = f"{ data.get('user_id') }"
         if user_id:
@@ -370,6 +470,7 @@ def become_user_become_user_id() -> Response:
         set_cookies(response, data)
         unset_submission_cookie(response)
         unset_permanent_cookie(response)
+        unset_become_user_csrf_cookie(response)
         response.set_cookie(key=tracking_cookie_name, value='', max_age=0, httponly=True)
 
         return response

arxiv-auth/src/accounts/tests/test_end_to_end.py

@@ -2,17 +2,19 @@
 
 from flask import request, Blueprint
 from unittest import TestCase
-from datetime import datetime
+from datetime import datetime, timedelta
 from pytz import timezone, UTC
 from dateutil.parser import parse
 import os
 import hashlib
 from base64 import b64encode
 from urllib.parse  import quote_plus
+import jwt
 
 from arxiv import status
 #from accounts.services import legacy, users
 from arxiv_auth.legacy import util, models
+from arxiv_auth.legacy.cookies import pack
 from accounts.factory import create_web_app
 
 
@@ -125,6 +127,46 @@ def setUp(self):
                 session.add(db_nick)
                 session.add(db_demo)
 
+                # Non-admin user for become_user tests
+                db_user2 = models.DBUser(
+                    user_id=2,
+                    first_name='target',
+                    last_name='user',
+                    suffix_name='',
+                    email='target@user.com',
+                    policy_class=2,
+                    flag_edit_users=0,
+                    flag_email_verified=1,
+                    flag_edit_system=0,
+                    flag_approved=1,
+                    flag_deleted=0,
+                    flag_banned=0,
+                    tracking_cookie='targetcookie',
+                )
+                db_nick2 = models.DBUserNickname(
+                    nick_id=2,
+                    nickname='targetuser',
+                    user_id=2,
+                    user_seq=1,
+                    flag_valid=1,
+                    role=0,
+                    policy=0,
+                    flag_primary=1
+                )
+                db_demo2 = models.DBProfile(
+                    user_id=2,
+                    country='US',
+                    affiliation='MIT',
+                    url='http://example.com/target',
+                    rank=2,
+                    original_subject_classes='cs.AI',
+                    archive='cs',
+                    subject_class='AI',
+                )
+                session.add(db_user2)
+                session.add(db_nick2)
+                session.add(db_demo2)
+
     def tearDown(self):
         with self.app.app_context():
             util.drop_all()
@@ -474,3 +516,124 @@ def test_post_login_with_next_page_implicit_protocol(self):
         self.assertEqual(response.status_code, status.HTTP_303_SEE_OTHER)
         assert bad_next_page not in response.headers['Location'] #  redirect should NOT point at value of `bad_next_page` param
         assert "bbc" not in response.headers['Location']
+
+    def test_become_user_requires_csrf(self):
+        """POST /become_user requires a CSRF token."""
+        client = self.app.test_client()
+        response = client.post('/become_user?become_user_id=1')
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_become_user_stale_auth_redirects_to_login(self):
+        """POST /become_user requires a fresh auth session."""
+        client = self.app.test_client()
+        now_utc = datetime.now(tz=UTC)
+        stale_start = (now_utc - timedelta(hours=2)).replace(microsecond=0)
+        auth_cookie_name = self.app.config['AUTH_SESSION_COOKIE_NAME']
+        auth_cookie_value = jwt.encode(
+            {
+                'user_id': 1,
+                'session_id': 'stale-session',
+                'nonce': 'nonce',
+                'start_time': stale_start.isoformat(),
+                'expires': (now_utc + timedelta(hours=1)).isoformat(),
+            },
+            self.secret,
+            algorithm='HS256'
+        )
+        client.set_cookie(key=auth_cookie_name, value=auth_cookie_value, domain='localhost')
+
+        classic_cookie_name = self.app.config['CLASSIC_COOKIE_NAME']
+        classic_cookie_value = pack('123', '1', self.ip_address, now_utc, '1')
+        client.set_cookie(key=classic_cookie_name, value=classic_cookie_value, domain='localhost')
+
+        csrf_cookie_name = self.app.config['BECOME_USER_CSRF_COOKIE_NAME']
+        csrf_token = 'csrf-token'
+        client.set_cookie(key=csrf_cookie_name, value=csrf_token, domain='localhost')
+
+        response = client.post(
+            '/become_user?become_user_id=2',
+            data={'csrf_token': csrf_token},
+            headers={'Referer': 'https://arxiv.org/admin'}
+        )
+        self.assertEqual(response.status_code, status.HTTP_303_SEE_OTHER)
+        self.assertTrue(response.headers['Location'].startswith('/login?next_page='))
+
+    def test_become_user_happy_path(self):
+        """POST /become_user with valid CSRF and fresh auth succeeds."""
+        client = self.app.test_client()
+        client.environ_base = self.environ_base
+        now_utc = datetime.now(tz=UTC)
+        fresh_start = (now_utc - timedelta(seconds=30)).replace(microsecond=0)
+
+        auth_cookie_name = self.app.config['AUTH_SESSION_COOKIE_NAME']
+        auth_cookie_value = jwt.encode(
+            {
+                'user_id': 1,
+                'session_id': 'fresh-session',
+                'nonce': 'nonce',
+                'start_time': fresh_start.isoformat(),
+                'expires': (now_utc + timedelta(hours=1)).isoformat(),
+            },
+            self.secret,
+            algorithm='HS256'
+        )
+        client.set_cookie(key=auth_cookie_name, value=auth_cookie_value, domain='localhost')
+
+        classic_cookie_name = self.app.config['CLASSIC_COOKIE_NAME']
+        classic_cookie_value = pack('123', '1', self.ip_address, now_utc, '1')
+        client.set_cookie(key=classic_cookie_name, value=classic_cookie_value, domain='localhost')
+
+        csrf_cookie_name = self.app.config['BECOME_USER_CSRF_COOKIE_NAME']
+        csrf_token = 'valid-csrf-token'
+        client.set_cookie(key=csrf_cookie_name, value=csrf_token, domain='localhost')
+
+        response = client.post(
+            '/become_user?become_user_id=2',
+            data={'csrf_token': csrf_token},
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        # CSRF cookie should be invalidated after use
+        cookies = _parse_cookies(response.headers.getlist('Set-Cookie'))
+        self.assertIn(csrf_cookie_name, cookies)
+        self.assertEqual(cookies[csrf_cookie_name]['Max-Age'], '0',
+                         'CSRF cookie should be cleared after use')
+
+    def test_become_user_csrf_endpoint_requires_admin(self):
+        """GET /become_user/csrf requires an admin user."""
+        client = self.app.test_client()
+        client.environ_base = self.environ_base
+
+        # Unauthenticated request
+        response = client.get('/become_user/csrf')
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+    def test_become_user_csrf_endpoint_issues_token(self):
+        """GET /become_user/csrf returns a CSRF token for admin users."""
+        client = self.app.test_client()
+        client.environ_base = self.environ_base
+
+        # Log in as admin user (user_id=1 has flag_edit_users=1)
+        form_data = {'username': 'foouser', 'password': 'thepassword'}
+        response = client.post('/login', data=form_data)
+        self.assertEqual(response.status_code, status.HTTP_303_SEE_OTHER)
+
+        # Set auth cookies on client for subsequent requests
+        cookies = _parse_cookies(response.headers.getlist('Set-Cookie'))
+        auth_cookie_name = self.app.config['AUTH_SESSION_COOKIE_NAME']
+        client.set_cookie(key=auth_cookie_name, value=cookies[auth_cookie_name]['value'],
+                          domain='localhost')
+        classic_cookie_name = self.app.config['CLASSIC_COOKIE_NAME']
+        client.set_cookie(key=classic_cookie_name, value=cookies[classic_cookie_name]['value'],
+                          domain='localhost')
+
+        response = client.get('/become_user/csrf')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.get_json()
+        self.assertIn('csrf_token', data)
+        self.assertTrue(len(data['csrf_token']) > 0)
+
+        # Verify CSRF cookie was set
+        csrf_cookies = _parse_cookies(response.headers.getlist('Set-Cookie'))
+        csrf_cookie_name = self.app.config['BECOME_USER_CSRF_COOKIE_NAME']
+        self.assertIn(csrf_cookie_name, csrf_cookies)