309 lines
10 KiB
Python
309 lines
10 KiB
Python
"""MSYNC 服务客户端模块
|
||
|
||
提供 ServiceClient 类，用于微服务间发起认证请求。
|
||
支持自动令牌获取与刷新、OGM 解析、服务发现。
|
||
本模块自包含实现（使用 urllib），不依赖 Django。
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
import uuid
|
||
import logging
|
||
import urllib.request
|
||
import urllib.error
|
||
from urllib.parse import quote
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
MSYNC_AUTH_PATH = '/api/msync/auth/token/'
|
||
MSYNC_AUTH_REFRESH_PATH = '/api/msync/auth/token/refresh/'
|
||
MSYNC_OGM_RESOLVE_PATH = '/api/msync/ogm/{global_id}/'
|
||
MSYNC_REGISTRY_PATH = '/api/msync/registry/{name}/'
|
||
|
||
#: 默认 MasterURL（仅当 settings 与环境变量均未配置时使用）
|
||
DEFAULT_MASTER_URL = 'https://api.gvsds.com'
|
||
|
||
from gvsdsdk.msync.auth import SignPayload, BuildAuthPayload
|
||
|
||
|
||
def _ResolveMasterURL(master_url):
|
||
"""解析 MasterURL：优先参数 → Django settings → 环境变量 → 默认值
|
||
|
||
本模块自包含（不依赖 Django ORM/HTTP），但允许通过 settings 或环境变量覆盖默认地址。
|
||
"""
|
||
if master_url:
|
||
return master_url.rstrip('/')
|
||
# 尝试 Django settings（软依赖，非 Django 环境下跳过）
|
||
try:
|
||
from django.conf import settings as _settings
|
||
url = getattr(_settings, 'GVSDSDK_MSYNC_MASTER_URL', None)
|
||
if url:
|
||
return str(url).rstrip('/')
|
||
except Exception:
|
||
pass
|
||
# 回退到环境变量
|
||
import os as _os
|
||
url = _os.environ.get('GVSDSDK_MSYNC_MASTER_URL')
|
||
if url:
|
||
return url.rstrip('/')
|
||
return DEFAULT_MASTER_URL
|
||
|
||
|
||
def _GenerateChallenge():
|
||
"""生成认证挑战字符串——与 auth.GenerateChallenge 统一格式（时间戳:UUID）"""
|
||
import time as _time
|
||
return f'{_time.time():.6f}:{uuid.uuid4().hex}'
|
||
|
||
|
||
class ServiceClientError(Exception):
|
||
def __init__(self, Message, StatusCode=None, ResponseBody=None):
|
||
super().__init__(Message)
|
||
self.StatusCode = StatusCode
|
||
self.ResponseBody = ResponseBody
|
||
|
||
|
||
class _TokenStore:
|
||
"""内存中的令牌存储"""
|
||
|
||
def __init__(self):
|
||
self._AccessToken = None
|
||
self._RefreshToken = None
|
||
self._ExpiresAt = 0.0
|
||
self._Scope = []
|
||
|
||
@property
|
||
def AccessToken(self):
|
||
return self._AccessToken
|
||
|
||
@property
|
||
def RefreshToken(self):
|
||
return self._RefreshToken
|
||
|
||
@property
|
||
def Scope(self):
|
||
return self._Scope
|
||
|
||
def IsValid(self):
|
||
return self._AccessToken is not None and time.time() < self._ExpiresAt
|
||
|
||
def Store(self, AccessToken, RefreshToken, ExpiresAt, Scope=None):
|
||
self._AccessToken = AccessToken
|
||
self._RefreshToken = RefreshToken
|
||
self._ExpiresAt = ExpiresAt
|
||
self._Scope = Scope or []
|
||
|
||
def Clear(self):
|
||
self._AccessToken = None
|
||
self._RefreshToken = None
|
||
self._ExpiresAt = 0.0
|
||
self._Scope = []
|
||
|
||
|
||
class ServiceClient:
|
||
"""微服务通信客户端——用于向其他服务发起认证请求
|
||
|
||
用法:
|
||
client = ServiceClient(
|
||
ServiceName='gerp',
|
||
ServiceSecret='your-secret',
|
||
)
|
||
client.Authenticate()
|
||
|
||
# 获取其他服务信息
|
||
info = client.DiscoverService('crb')
|
||
|
||
# 解析 OGM
|
||
ogm = client.ResolveOGM('user:abc123')
|
||
|
||
# 向其他服务发请求
|
||
response = client.Request('GET', 'https://crb.gvsds.com/api/wallet', Scope=['ledger:read'])
|
||
"""
|
||
|
||
def __init__(self, ServiceName, ServiceSecret, MasterURL=None):
|
||
self.ServiceName = ServiceName
|
||
self.ServiceSecret = ServiceSecret
|
||
self.MasterURL = _ResolveMasterURL(MasterURL)
|
||
self._tokens = _TokenStore()
|
||
|
||
@property
|
||
def IsAuthenticated(self):
|
||
return self._tokens.IsValid()
|
||
|
||
@property
|
||
def AccessToken(self):
|
||
return self._tokens.AccessToken
|
||
|
||
@property
|
||
def RefreshToken(self):
|
||
return self._tokens.RefreshToken
|
||
|
||
def Authenticate(self):
|
||
if not self._Authenticate():
|
||
raise ServiceClientError('认证失败')
|
||
return self
|
||
|
||
def Refresh(self):
|
||
if not self._RefreshToken():
|
||
raise ServiceClientError('刷新失败')
|
||
return self
|
||
|
||
def _Authenticate(self):
|
||
challenge = _GenerateChallenge()
|
||
payload = BuildAuthPayload(self.ServiceName, challenge)
|
||
signature = SignPayload(payload, self.ServiceSecret)
|
||
|
||
req_body = json.dumps({
|
||
'ServiceName': self.ServiceName,
|
||
'Payload': payload,
|
||
'Signature': signature,
|
||
}).encode('utf-8')
|
||
|
||
req = urllib.request.Request(
|
||
f'{self.MasterURL}{MSYNC_AUTH_PATH}',
|
||
data=req_body,
|
||
headers={'Content-Type': 'application/json'},
|
||
method='POST',
|
||
)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
data = json.loads(resp.read().decode('utf-8'))
|
||
except urllib.error.HTTPError as e:
|
||
body = e.read().decode('utf-8', errors='replace')
|
||
logger.error(f'认证失败 HTTP {e.code}: {body[:200]}') # 限制日志长度，避免泄漏完整 token
|
||
return False
|
||
except (urllib.error.URLError, ValueError, OSError) as e:
|
||
logger.error(f'认证异常: {e}')
|
||
return False
|
||
|
||
if data.get('code') not in (200, 0):
|
||
logger.error(f'认证业务错误 code={data.get("code")} msg={data.get("msg", "")[:200]}') # 仅记录 code 和简短消息
|
||
return False
|
||
|
||
token_data = data.get('data') or {}
|
||
self._tokens.Store(
|
||
AccessToken=token_data.get('AccessToken'),
|
||
RefreshToken=token_data.get('RefreshToken'),
|
||
ExpiresAt=token_data.get('ExpiresAt', 0),
|
||
Scope=token_data.get('Scope', []),
|
||
)
|
||
return self._tokens.AccessToken is not None
|
||
|
||
def _RefreshToken(self):
|
||
if not self._tokens.RefreshToken:
|
||
return False
|
||
|
||
req_body = json.dumps({
|
||
'ServiceName': self.ServiceName,
|
||
'RefreshToken': self._tokens.RefreshToken,
|
||
}).encode('utf-8')
|
||
|
||
req = urllib.request.Request(
|
||
f'{self.MasterURL}{MSYNC_AUTH_REFRESH_PATH}',
|
||
data=req_body,
|
||
headers={'Content-Type': 'application/json'},
|
||
method='POST',
|
||
)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
data = json.loads(resp.read().decode('utf-8'))
|
||
except (urllib.error.URLError, ValueError, OSError) as e:
|
||
logger.error(f'刷新令牌异常: {e}')
|
||
return False
|
||
|
||
if data.get('code') not in (200, 0):
|
||
return False
|
||
|
||
token_data = data.get('data') or {}
|
||
self._tokens.Store(
|
||
AccessToken=token_data.get('AccessToken'),
|
||
RefreshToken=token_data.get('RefreshToken'),
|
||
ExpiresAt=token_data.get('ExpiresAt', 0),
|
||
Scope=token_data.get('Scope', []),
|
||
)
|
||
return self._tokens.AccessToken is not None
|
||
|
||
def _EnsureAuthenticated(self):
|
||
if self.IsAuthenticated:
|
||
return
|
||
if self._tokens.RefreshToken:
|
||
self._RefreshToken()
|
||
return
|
||
self._Authenticate()
|
||
|
||
# 注：本方法约 60 行且含 3 层 try/except 嵌套（M9）。
|
||
# 未拆分原因：401 重试逻辑与首次请求共享 req_headers/body/URL/Method/Timeout 等局部状态，
|
||
# 拆分为独立方法需传递大量参数或改用闭包，收益有限且可能影响现有测试覆盖，故保持原结构。
|
||
def Request(self, Method, URL, Data=None, Headers=None, Scope=None, Timeout=30):
|
||
self._EnsureAuthenticated()
|
||
|
||
req_headers = {
|
||
'Authorization': f'Service {self.AccessToken}',
|
||
'Content-Type': 'application/json',
|
||
'X-Source-Service': self.ServiceName,
|
||
}
|
||
|
||
if Scope:
|
||
req_headers['X-Required-Scope'] = ','.join(Scope)
|
||
|
||
if Headers:
|
||
req_headers.update(Headers)
|
||
|
||
body = None
|
||
if Data is not None:
|
||
if isinstance(Data, dict):
|
||
body = json.dumps(Data).encode('utf-8')
|
||
elif isinstance(Data, bytes):
|
||
body = Data
|
||
elif isinstance(Data, str):
|
||
body = Data.encode('utf-8')
|
||
|
||
req = urllib.request.Request(URL, data=body, headers=req_headers, method=Method)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=Timeout) as resp:
|
||
response_data = resp.read().decode('utf-8')
|
||
content_type = resp.headers.get('Content-Type', '')
|
||
if 'application/json' in content_type:
|
||
return json.loads(response_data)
|
||
return response_data
|
||
except urllib.error.HTTPError as e:
|
||
body_text = e.read().decode('utf-8', errors='replace')
|
||
if e.code == 401:
|
||
logger.info('令牌可能已失效，尝试刷新')
|
||
self._tokens.Clear()
|
||
self._EnsureAuthenticated()
|
||
if not self.AccessToken:
|
||
raise ServiceClientError('重新认证失败，无法重试请求')
|
||
req_headers['Authorization'] = f'Service {self.AccessToken}'
|
||
retry_req = urllib.request.Request(URL, data=body, headers=req_headers, method=Method)
|
||
try:
|
||
with urllib.request.urlopen(retry_req, timeout=Timeout) as resp2:
|
||
response_data = resp2.read().decode('utf-8')
|
||
content_type = resp2.headers.get('Content-Type', '')
|
||
if 'application/json' in content_type:
|
||
return json.loads(response_data)
|
||
return response_data
|
||
except urllib.error.HTTPError as e2:
|
||
body2 = e2.read().decode('utf-8', errors='replace')
|
||
raise ServiceClientError(
|
||
f'重试后仍失败: HTTP {e2.code}',
|
||
StatusCode=e2.code,
|
||
ResponseBody=body2,
|
||
)
|
||
raise ServiceClientError(
|
||
f'请求失败: HTTP {e.code}',
|
||
StatusCode=e.code,
|
||
ResponseBody=body_text,
|
||
)
|
||
|
||
def ResolveOGM(self, GlobalID):
|
||
url = f'{self.MasterURL}{MSYNC_OGM_RESOLVE_PATH.format(global_id=quote(str(GlobalID), safe=""))}'
|
||
return self.Request('GET', url, Scope=['ogm:read'])
|
||
|
||
def DiscoverService(self, ServiceName):
|
||
url = f'{self.MasterURL}{MSYNC_REGISTRY_PATH.format(name=quote(str(ServiceName), safe=""))}'
|
||
return self.Request('GET', url, Scope=['registry:read'])
|