217 lines
7.3 KiB
Python
217 lines
7.3 KiB
Python
import json
|
||
import re
|
||
|
||
from django.http import JsonResponse
|
||
from django.utils import timezone
|
||
from django.views.decorators.csrf import csrf_exempt
|
||
|
||
from . import app_settings
|
||
from .models import ApiToken
|
||
from .versions import is_valid_version
|
||
from .views import generate_custom_client, _get_run_status
|
||
|
||
|
||
# 字段取值约束（与 GenerateForm 保持一致）
|
||
PLATFORM_CHOICES = ['windows', 'windows-x86', 'linux', 'android', 'macos']
|
||
# 版本跟随 rustdesk/rustdesk 官方 tags 动态更新，这里只做格式校验
|
||
DIRECTION_CHOICES = ['incoming', 'outgoing', 'both']
|
||
INSTALLATION_CHOICES = ['installationY', 'installationN']
|
||
SETTINGS_CHOICES = ['settingsY', 'settingsN']
|
||
THEME_CHOICES = ['light', 'dark', 'system']
|
||
THEME_DORO_CHOICES = ['default', 'override']
|
||
PASS_APPROVE_MODE_CHOICES = ['password', 'click', 'password-click']
|
||
PERMISSIONS_DORO_CHOICES = ['default', 'override']
|
||
PERMISSIONS_TYPE_CHOICES = ['custom', 'full', 'view']
|
||
|
||
# 布尔字段
|
||
BOOL_FIELDS = [
|
||
'delayFix', 'xOffline', 'hidecm', 'removeNewVersionNotif',
|
||
'denyLan', 'enableDirectIP', 'autoClose',
|
||
'enableKeyboard', 'enableClipboard', 'enableFileTransfer', 'enableAudio',
|
||
'enableTCP', 'enableRemoteRestart', 'enableRecording', 'enableBlockingInput',
|
||
'enableRemoteModi', 'removeWallpaper', 'enablePrinter', 'enableCamera', 'enableTerminal',
|
||
]
|
||
|
||
# 可选字符串字段
|
||
OPTIONAL_STR_FIELDS = [
|
||
'sh_secret_field', 'serverIP', 'serverPort', 'key', 'apiServer', 'urlLink', 'downloadLink',
|
||
'appname', 'compname', 'androidappid', 'permanentPassword',
|
||
'defaultManual', 'overrideManual',
|
||
'iconbase64', 'logobase64', 'privacybase64',
|
||
]
|
||
|
||
|
||
def _authenticate(request):
|
||
"""通过 Authorization: Token <key> 请求头认证，返回 User 或 None。"""
|
||
auth_header = request.headers.get('Authorization', '')
|
||
if not auth_header.startswith('Token '):
|
||
return None
|
||
key = auth_header[len('Token '):].strip()
|
||
token = (
|
||
ApiToken.objects
|
||
.filter(key=key, is_active=True)
|
||
.select_related('user')
|
||
.first()
|
||
)
|
||
if token and token.user.is_active:
|
||
token.last_used_at = timezone.now()
|
||
token.save(update_fields=['last_used_at'])
|
||
return token.user
|
||
return None
|
||
|
||
|
||
def _unauthorized():
|
||
return JsonResponse(
|
||
{"success": False, "error": "未认证或令牌无效，请在请求头中携带有效的 Token（Authorization: Token <key>）。"},
|
||
status=401,
|
||
headers={'WWW-Authenticate': 'Token'},
|
||
)
|
||
|
||
|
||
def validate_generate_params(data):
|
||
"""
|
||
按与 GenerateForm 相同的约束校验 JSON API 参数。
|
||
|
||
Returns:
|
||
(cleaned_data, errors)
|
||
"""
|
||
errors = {}
|
||
cleaned = {}
|
||
|
||
# 必填字符串字段
|
||
exename = data.get('exename', '')
|
||
if not exename:
|
||
errors['exename'] = '该字段为必填项。'
|
||
else:
|
||
cleaned['exename'] = exename
|
||
|
||
# 枚举字段
|
||
choice_validations = {
|
||
'platform': (PLATFORM_CHOICES, 'windows'),
|
||
'direction': (DIRECTION_CHOICES, 'both'),
|
||
'installation': (INSTALLATION_CHOICES, 'installationY'),
|
||
'settings': (SETTINGS_CHOICES, 'settingsY'),
|
||
'theme': (THEME_CHOICES, 'system'),
|
||
'themeDorO': (THEME_DORO_CHOICES, 'default'),
|
||
'passApproveMode': (PASS_APPROVE_MODE_CHOICES, 'password-click'),
|
||
'permissionsDorO': (PERMISSIONS_DORO_CHOICES, 'default'),
|
||
'permissionsType': (PERMISSIONS_TYPE_CHOICES, 'custom'),
|
||
}
|
||
for field, (choices, default) in choice_validations.items():
|
||
value = data.get(field, default)
|
||
if value not in choices:
|
||
errors[field] = f'取值无效，必须为以下之一：{choices}'
|
||
else:
|
||
cleaned[field] = value
|
||
|
||
# 版本：master 或形如 1.4.9 / 1.5.0-rc.1 的官方 tag（列表动态获取，不做硬枚举）
|
||
version = data.get('version', '1.4.9')
|
||
if not is_valid_version(version):
|
||
errors['version'] = "版本号无效，应为 'master' 或形如 1.4.9 的官方发布 tag。"
|
||
else:
|
||
cleaned['version'] = version
|
||
|
||
# 布尔字段
|
||
for field in BOOL_FIELDS:
|
||
value = data.get(field, False)
|
||
if not isinstance(value, bool):
|
||
errors[field] = '必须为布尔值（true/false）。'
|
||
else:
|
||
cleaned[field] = value
|
||
|
||
# 可选字符串字段
|
||
for field in OPTIONAL_STR_FIELDS:
|
||
cleaned[field] = data.get(field, '')
|
||
|
||
# 自由文本名称会进入 shell sed 脚本（与表单校验规则一致）
|
||
for field in ('appname', 'compname'):
|
||
value = cleaned.get(field, '')
|
||
if isinstance(value, str) and re.search(r'[&\\|\'"$`\r\n]', value):
|
||
errors[field] = '包含构建脚本不支持的字符（& \\ | \' " $ ` 或换行符）。'
|
||
|
||
# API 模式不使用文件上传，改用 base64 字段
|
||
cleaned['iconfile'] = None
|
||
cleaned['logofile'] = None
|
||
cleaned['privacyfile'] = None
|
||
|
||
return cleaned, errors
|
||
|
||
|
||
@csrf_exempt
|
||
def api_generate(request):
|
||
"""
|
||
POST /api/generate
|
||
|
||
接收客户端配置 JSON，触发 GitHub Actions 自定义构建。
|
||
需在请求头携带：Authorization: Token <key>
|
||
"""
|
||
if request.method != 'POST':
|
||
return JsonResponse({"success": False, "error": "仅支持 POST 请求。"}, status=405)
|
||
|
||
user = _authenticate(request)
|
||
if user is None:
|
||
return _unauthorized()
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
except (json.JSONDecodeError, ValueError) as e:
|
||
return JsonResponse({"success": False, "error": f"JSON 格式无效：{str(e)}"}, status=400)
|
||
|
||
cleaned, errors = validate_generate_params(data)
|
||
if errors:
|
||
return JsonResponse({
|
||
"success": False,
|
||
"error": "参数校验未通过",
|
||
"details": errors
|
||
}, status=400)
|
||
|
||
full_url = f"{app_settings.get_value('PROTOCOL')}://{request.get_host()}"
|
||
|
||
result = generate_custom_client(cleaned, full_url, user=user)
|
||
|
||
if result['success']:
|
||
result['status_url'] = f"/api/status?uuid={result['uuid']}&platform={result['platform']}&filename={result['filename']}"
|
||
return JsonResponse(result)
|
||
else:
|
||
return JsonResponse({"success": False, "error": result['error']}, status=result.get('status_code', 500))
|
||
|
||
|
||
@csrf_exempt
|
||
def api_status(request):
|
||
"""
|
||
GET /api/status?uuid=<uuid>
|
||
|
||
查询指定构建任务的状态。需携带 API Token。
|
||
"""
|
||
if request.method != 'GET':
|
||
return JsonResponse({"success": False, "error": "仅支持 GET 请求。"}, status=405)
|
||
|
||
if _authenticate(request) is None:
|
||
return _unauthorized()
|
||
|
||
uuid_val = request.GET.get('uuid')
|
||
if not uuid_val:
|
||
return JsonResponse({"success": False, "error": "缺少必填参数：uuid"}, status=400)
|
||
|
||
filename = request.GET.get('filename', '')
|
||
platform = request.GET.get('platform', '')
|
||
|
||
result = _get_run_status(uuid_val)
|
||
|
||
if not result['found']:
|
||
return JsonResponse({"success": False, "error": "未找到对应的构建任务"}, status=404)
|
||
|
||
response_data = {
|
||
"success": True,
|
||
"status": result['status'],
|
||
"status_label": result['gh_run'].status_label(),
|
||
"uuid": uuid_val,
|
||
"log_url": result['github_log_url'],
|
||
}
|
||
if filename:
|
||
response_data['filename'] = filename
|
||
if platform:
|
||
response_data['platform'] = platform
|
||
|
||
return JsonResponse(response_data)
|