"""Local bootstrap and administrator-only license management.""" import json import secrets import threading from django.http import HttpResponse, JsonResponse, HttpResponseRedirect, FileResponse from django.template import engines from django.views.decorators.csrf import csrf_exempt, ensure_csrf_cookie from django.views.decorators.http import require_http_methods from monitor_runtime.paths import data, resource, DESKTOP from monitor_runtime.licensing import check_license, import_license, request_document from monitor_runtime.bootstrap import has_admin, finish_setup, local_addresses SETUP_TOKEN = secrets.token_urlsafe(32) SETUP_LOCK = threading.Lock() RESTART_REQUESTED = threading.Event() SERVICE_STATE = {'phase': 'starting', 'reason': ''} PAGE = """ Monitor 安装与授权

{% if setup %}首次安装设置{% else %}授权管理{% endif %}

{{ status.reason }}{% if status.license %}
客户：{{ status.license.customer }}
有效期：{{ status.license.expires_at|default:"永久" }}{% endif %}

{% if error %}

{{ error }}

{% endif %}

导出本机申请文件，交给发行方签发，再导入签名授权文件。授权绑定当前电脑。

{% csrf_token %}
{% csrf_token %}
{% if setup %}
{% csrf_token %} 请先导入授权。初始化只允许从安装电脑完成；完成后局域网用户才能登录。
{% else %}

返回登录 · 进入系统

{% endif %}
""" def is_local(request): return request.META.get('REMOTE_ADDR') in ('127.0.0.1', '::1') def admin_allowed(request): user = getattr(request, 'user', None) return bool(user and user.is_authenticated and ( user.is_superuser or user.groups.filter(name='system_admin').exists())) def authorized(request, write=False): if not has_admin(): if not is_local(request): return False return not write or secrets.compare_digest(request.POST.get('setup_token', ''), SETUP_TOKEN) return admin_allowed(request) def page(request, error='', code=200): html = engines['django'].from_string(PAGE).render({ 'setup': not has_admin(), 'status': check_license(), 'error': error, 'addresses': local_addresses() if not has_admin() else [], }, request) response = HttpResponse(html, status=code) response['Cache-Control'] = 'no-store' response['Referrer-Policy'] = 'no-referrer' return response @csrf_exempt @ensure_csrf_cookie @require_http_methods(['GET', 'POST']) def setup(request): if has_admin() or not authorized(request, request.method == 'POST'): return HttpResponse('初始化入口不可用', status=403) if request.method == 'POST': try: with SETUP_LOCK: if request.POST.get('password') != request.POST.get('confirm'): raise ValueError('两次密码输入不一致') finish_setup(request.POST.get('username', ''), request.POST.get('password', ''), request.POST.get('address', '')) RESTART_REQUESTED.set() return HttpResponse('

设置完成，服务正在重启。稍后重新打开桌面快捷方式进入登录页面。

') except Exception as exc: return page(request, str(exc), 400) return page(request) @ensure_csrf_cookie def license_page(request): if not authorized(request): return HttpResponseRedirect('/login') if not getattr(request.user, 'is_authenticated', False) else HttpResponse(status=403) return page(request) @csrf_exempt @require_http_methods(['POST']) def license_request(request): if not authorized(request, True): return HttpResponse(status=403) response = JsonResponse(request_document(), json_dumps_params={'indent': 2}) response['Content-Disposition'] = 'attachment; filename="monitor-license-request.json"' return response @csrf_exempt @require_http_methods(['POST']) def license_import(request): if not authorized(request, True): return HttpResponse(status=403) try: uploaded = request.FILES.get('license') if not uploaded or uploaded.size > 32768: raise ValueError('请选择小于 32KB 的授权文件') import_license(uploaded.read()) if has_admin(): RESTART_REQUESTED.set() return page(request) except Exception as exc: return page(request, str(exc), 400) def license_status(request): if not authorized(request): return HttpResponse(status=403) return JsonResponse({**check_license(), 'service': SERVICE_STATE}) def private_file(request, path, area='storage'): if not request.user.is_authenticated: return HttpResponse(status=403) root = data('static/storage' if area == 'storage' else 'static/upload/audio').resolve() filename = (root / path).resolve() allowed = {'.jpg', '.jpeg', '.png', '.mp4', '.wav', '.mp3', '.ogg'} if not filename.is_relative_to(root) or not filename.is_file() or filename.suffix.lower() not in allowed: return HttpResponse(status=404) response = FileResponse(filename.open('rb')) response['Cache-Control'] = 'private, no-store' response['X-Content-Type-Options'] = 'nosniff' return response class LicenseMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if not DESKTOP: return self.get_response(request) path = request.path_info # WhiteNoise only serves immutable build-time assets, never mutable data. public_asset = path.startswith('/static/') and not path.startswith(('/static/storage/', '/static/upload/')) exceptions = {'/setup', '/license', '/license/status', '/license/request', '/license/import', '/login', '/logout', '/user/openCaptcha'} if not has_admin() and not public_asset and path not in {'/setup', '/license', '/license/request', '/license/import', '/license/status'}: return HttpResponseRedirect('/setup') if is_local(request) else HttpResponse(status=403) if path not in exceptions and not public_asset: status = check_license() if not status['valid']: if '/open' in path or path.startswith('/inner/'): return JsonResponse({'code': 0, 'msg': status['reason'], 'error': 'license_invalid'}, status=403) return HttpResponseRedirect('/license') return self.get_response(request)