diff --git a/backend/blueprints/admin_api.py b/backend/blueprints/admin_api.py index 5a58effd..7eb0aaf1 100644 --- a/backend/blueprints/admin_api.py +++ b/backend/blueprints/admin_api.py @@ -277,6 +277,16 @@ def _backend_java_internal_request(): return headers, {'operatorId': operator_id} +def _resolve_current_operator_id(): + """取当前请求上下文里的操作用户 id（Flask 已登录管理员），无上下文/未登录返回 None。""" + if not has_request_context(): + return None + request_row = getattr(g, '_current_user_row', None) + if request_row and request_row.get('id'): + return request_row.get('id') + return session.get('user_id') + + def _proxy_backend_java( method, path, @@ -287,9 +297,27 @@ def _proxy_backend_java( data=None, timeout=10, headers=None, + use_internal_token=True, ): + """调用 backend-java。 + + use_internal_token 默认 True：以可信内部令牌标识"已通过 Flask 侧校验的 + 管理员"（Java 管理接口在 A1 收敛后普遍要求 require* 身份，而浏览器打到 + 15124 的请求不带 Java JWT/aiimage_token Cookie，若不补内部令牌，Flask + 校验通过后仍会被 Java 判"未登录"）。Java 内部兼容通道要求 + X-Internal-Token + operatorId 双条件，因此同时自动补当前操作用户 id； + 调用方显式传的 operatorId/operator_id 优先。需要转发浏览器原始鉴权材料 + （JWT，且调用方自行带身份参数）的场景应显式传 use_internal_token=False。 + """ url = f"{backend_java_base_url}{path}" forwarded_headers = _backend_java_forward_headers() + if use_internal_token: + forwarded_headers['X-Internal-Token'] = _resolve_internal_token() + if params and not any(key in params for key in ('operatorId', 'operator_id')): + operator_id = _resolve_current_operator_id() + if operator_id: + params = dict(params) + params['operatorId'] = operator_id if headers: forwarded_headers.update(headers) try: diff --git a/backend/tests/test_admin_proxy_internal_token.py b/backend/tests/test_admin_proxy_internal_token.py new file mode 100644 index 00000000..4e7b4e38 --- /dev/null +++ b/backend/tests/test_admin_proxy_internal_token.py @@ -0,0 +1,111 @@ +"""回归测试：Flask 代理转发 Java 时默认携带 X-Internal-Token。 + +背景：A1 管理后台收敛后，Java 管理接口普遍要求身份（requireAdminOrInternal）。 +浏览器打到 15124 的请求不带 Java JWT/aiimage_token Cookie，list_shop_manages 等 +裸代理调用若只转发浏览器 Cookie 会被 Java 判「未登录」，生产后台店铺列表报 +「加载失败:未登录」。本文件锁定 _proxy_backend_java 默认补内部令牌的行为。 +""" +import sys +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +from flask import Flask + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from blueprints import admin_api + + +class _FakeSession: + """记录请求的伪 requests.Session，返回 Java ApiResponse 成功体。""" + + def __init__(self): + self.calls = [] + + def request(self, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + response = Mock() + response.status_code = 200 + response.json.return_value = {'success': True, 'data': {'items': []}} + return response + + +class AdminProxyInternalTokenTest(unittest.TestCase): + + def setUp(self): + self.app = Flask(__name__) + self.session = _FakeSession() + self.patches = [ + patch.object(admin_api, '_get_backend_java_session', return_value=self.session), + patch.object(admin_api, '_resolve_internal_token', return_value='tk-internal-test'), + ] + for p in self.patches: + p.start() + self.addCleanup(p.stop) + + def _call(self, **kwargs): + with self.app.test_request_context('/api/admin/shop-manages?page=1&page_size=15'): + return admin_api._proxy_backend_java('GET', '/api/admin/shop-manages', **kwargs) + + def test_default_carries_internal_token(self): + """默认转发必须带 X-Internal-Token（否则 Java requireAdminOrInternal 判未登录）。""" + result, error_response, status = self._call(params={'page': 1, 'pageSize': 15}) + self.assertEqual(status, 200) + self.assertIsNone(error_response) + self.assertTrue(result.get('success')) + method, url, kwargs = self.session.calls[0] + self.assertEqual(method, 'GET') + self.assertEqual(url, f"{admin_api.backend_java_base_url}/api/admin/shop-manages") + self.assertEqual(kwargs['headers'].get('X-Internal-Token'), 'tk-internal-test') + + def test_auto_injects_operator_id_from_session(self): + """默认转发自动补当前操作用户 operatorId（Java 内部通道要求令牌+operatorId 双条件）。""" + with self.app.test_request_context('/api/admin/shop-manages?page=1&page_size=15'): + with patch.object(admin_api, '_resolve_current_operator_id', return_value=7): + result, error_response, status = admin_api._proxy_backend_java( + 'GET', '/api/admin/shop-manages', params={'page': 1, 'pageSize': 15}) + self.assertEqual(status, 200) + self.assertIsNone(error_response) + method, url, kwargs = self.session.calls[0] + self.assertEqual(kwargs['params'].get('operatorId'), 7) + + def test_explicit_operator_id_wins(self): + """调用方显式传的 operatorId 优先于自动注入（如超管 role 参数场景）。""" + with self.app.test_request_context('/api/admin/shop-manages?page=1&page_size=15'): + with patch.object(admin_api, '_resolve_current_operator_id', return_value=7): + result, error_response, status = admin_api._proxy_backend_java( + 'GET', '/api/admin/shop-manages', + params={'page': 1, 'pageSize': 15, 'operatorId': 99, 'superAdmin': 'true'}) + self.assertEqual(status, 200) + method, url, kwargs = self.session.calls[0] + self.assertEqual(kwargs['params'].get('operatorId'), 99) + + def test_no_session_no_operator_id(self): + """无请求上下文（定时任务线程）时不注入 operatorId，由调用方自行决定。""" + result, error_response, status = admin_api._proxy_backend_java( + 'GET', '/api/admin/shop-manages', params={'page': 1}) + self.assertEqual(status, 200) + method, url, kwargs = self.session.calls[0] + self.assertNotIn('operatorId', kwargs.get('params') or {}) + + def test_opt_out_omits_internal_token(self): + """显式 use_internal_token=False 时不带内部令牌（供转发浏览器原始 JWT 的场景）。""" + result, error_response, status = self._call( + params={'page': 1}, use_internal_token=False) + self.assertEqual(status, 200) + self.assertIsNone(error_response) + method, url, kwargs = self.session.calls[0] + self.assertNotIn('X-Internal-Token', kwargs.get('headers') or {}) + self.assertNotIn('operatorId', kwargs.get('params') or {}) + + def test_explicit_headers_override_merge(self): + """显式传 headers 时与内部令牌合并，不互相覆盖。""" + result, error_response, status = self._call(headers={'X-Forwarded-For': '1.2.3.4'}) + self.assertEqual(status, 200) + method, url, kwargs = self.session.calls[0] + self.assertEqual(kwargs['headers'].get('X-Internal-Token'), 'tk-internal-test') + self.assertEqual(kwargs['headers'].get('X-Forwarded-For'), '1.2.3.4') + + +if __name__ == '__main__': + unittest.main()