import os from pathlib import Path from fastapi import Request from fastapi import HTTPException PUBLIC_API_BASE_URL = os.getenv("PUBLIC_API_BASE_URL", "").strip().rstrip("/") VALID_CLIENT_TOKEN = os.getenv("CLIENT_API_TOKEN") TARGET_PLATFORM = os.getenv("TARGET_PLATFORM", "windows").strip().lower() or "windows" TARGET_ARCH = os.getenv("TARGET_ARCH", "x64").strip() or "x64" def platform_executable_path(value: str, fallback_base_name: str) -> str: path = str(value or "").strip().replace(chr(92), "/").strip("/") if not path: path = fallback_base_name parts = path.split("/") leaf = parts[-1] if TARGET_PLATFORM == "windows": if "." not in leaf: leaf += ".exe" elif TARGET_PLATFORM == "linux": if leaf.casefold().endswith(".exe"): leaf = leaf[:-4] parts[-1] = leaf return "/".join(parts) RELEASE_MAIN_EXECUTABLE = platform_executable_path( os.getenv("RELEASE_MAIN_EXECUTABLE", ""), "bin/SimCAE", ) PUBLISH_MAX_REQUEST_BYTES = int(os.getenv("PUBLISH_MAX_REQUEST_MB", "4096")) * 1024 * 1024 PUBLISH_MAX_FILES = int(os.getenv("PUBLISH_MAX_FILES", "20000")) PUBLISH_MAX_FIELDS = int(os.getenv("PUBLISH_MAX_FIELDS", str(PUBLISH_MAX_FILES + 100))) def public_api_base_url(request: Request) -> str: if PUBLIC_API_BASE_URL: return PUBLIC_API_BASE_URL forwarded_proto = request.headers.get("x-forwarded-proto", "").split(",")[0].strip() forwarded_host = request.headers.get("x-forwarded-host", "").split(",")[0].strip() scheme = forwarded_proto or request.url.scheme host = forwarded_host or request.headers.get("host", "") if host: return f"{scheme}://{host}".rstrip("/") return str(request.base_url).rstrip("/") def default_client_main_executable() -> str: path = RELEASE_MAIN_EXECUTABLE.strip().replace(chr(92), "/").strip("/") if path.casefold().startswith("bin/"): path = path[4:] return platform_executable_path(path, "SimCAE") def default_client_config_values(request: Request) -> dict: return { "api_base_url": public_api_base_url(request), "client_token": VALID_CLIENT_TOKEN or "", "launch_token": os.getenv("CLIENT_LAUNCH_TOKEN", "SimCAE_Launch_Token_2026_ChangeMe_32Bytes"), "request_timeout_ms": os.getenv("CLIENT_REQUEST_TIMEOUT_MS", "5000"), "temp_folder": os.getenv("CLIENT_TEMP_FOLDER", "update_temp"), "install_root": os.getenv("CLIENT_INSTALL_ROOT", ".."), "main_executable": platform_executable_path( os.getenv("CLIENT_MAIN_EXECUTABLE", ""), default_client_main_executable(), ), "launcher_executable": platform_executable_path(os.getenv("CLIENT_LAUNCHER_EXECUTABLE", ""), "Launcher"), "updater_executable": platform_executable_path(os.getenv("CLIENT_UPDATER_EXECUTABLE", ""), "Updater"), "bootstrap_executable": platform_executable_path(os.getenv("CLIENT_BOOTSTRAP_EXECUTABLE", ""), "Bootstrap"), "health_check_timeout_ms": os.getenv("CLIENT_HEALTH_CHECK_TIMEOUT_MS", "15000"), "platform": TARGET_PLATFORM, "arch": TARGET_ARCH, } def crash_report_test_values(api_base_url: str) -> dict: base_url = api_base_url.rstrip("/") crash_admin_token = os.getenv("CRASH_ADMIN_TOKEN", "").strip() admin_token = os.getenv("ADMIN_TOKEN", "").strip() crash_report_token = ( os.getenv("CRASH_REPORT_TOKEN") or os.getenv("SIMCAE_CRASH_CLIENT_TOKEN") or os.getenv("CLIENT_API_TOKEN") or "" ) crash_symbol_token = os.getenv("CRASH_SYMBOL_TOKEN") or os.getenv("SIMCAE_SYMBOL_TOKEN") or "" effective_admin_token = crash_admin_token or admin_token effective_admin_token_source = "CRASH_ADMIN_TOKEN" if crash_admin_token else "ADMIN_TOKEN" return { "说明": "仅用于后台管理员/测试人员联调崩溃报告接口，不要复制进客户端 app_config.json。", "server_base_url": base_url, "tokens": { "CRASH_REPORT_TOKEN": { "用途": "上传崩溃报告 /api/v1/crash-reports", "token": crash_report_token, }, "CRASH_SYMBOL_TOKEN": { "用途": "上传符号包 /api/v1/symbols", "token": crash_symbol_token, }, "CRASH_ADMIN_TOKEN": { "用途": "查询和下载崩溃报告", "token": crash_admin_token, "说明": "如果为空，查询/下载接口会使用 ADMIN_TOKEN。", }, "ADMIN_TOKEN": { "用途": "管理后台令牌；当 CRASH_ADMIN_TOKEN 为空时也用于查询/下载崩溃报告", "token": admin_token, }, }, "upload_crash_report": { "url": f"{base_url}/api/v1/crash-reports", "authorization_header": "Authorization: Bearer ", "token_env": "CRASH_REPORT_TOKEN", "token": crash_report_token, "idempotency_key_rule": "Idempotency-Key 必须等于 metadata.clientReportId", }, "upload_symbols": { "url": f"{base_url}/api/v1/symbols", "authorization_header": "Authorization: Bearer ", "token_env": "CRASH_SYMBOL_TOKEN", "token": crash_symbol_token, }, "admin_query_and_download": { "detail_url_template": f"{base_url}/api/v1/crash-reports/{{report_id}}", "file_url_template": f"{base_url}/api/v1/crash-reports/{{report_id}}/files/{{file_name}}", "authorization_header": f"Authorization: Bearer <{effective_admin_token_source}>", "token_env": effective_admin_token_source, "token": effective_admin_token, "file_name_examples": ["metadata.json", "crash.dmp", "attachments.zip", "server.json"], }, "limits": { "CRASH_SERVICE_VERSION": os.getenv("CRASH_SERVICE_VERSION", "1.0.0"), "CRASH_METADATA_MAX_KB": os.getenv("CRASH_METADATA_MAX_KB", "256"), "CRASH_MINIDUMP_MAX_MB": os.getenv("CRASH_MINIDUMP_MAX_MB", "128"), "CRASH_ATTACHMENTS_MAX_MB": os.getenv("CRASH_ATTACHMENTS_MAX_MB", "64"), "CRASH_REQUEST_MAX_MB": os.getenv("CRASH_REQUEST_MAX_MB", "200"), "CRASH_SYMBOLS_MAX_MB": os.getenv("CRASH_SYMBOLS_MAX_MB", "512"), }, } def crash_report_test_text(api_base_url: str) -> str: values = crash_report_test_values(api_base_url) tokens = values["tokens"] lines = [ "# 说明：下面内容仅用于后台管理员/测试人员联调崩溃报告接口。", "# 注意：不要复制进客户端 app_config.json，也不要放进客户端配置文件。", "", "# 服务端基础地址。curl 示例里的 SERVER 就填这个值。", f"SERVER_BASE_URL={values['server_base_url']}", "", "# 上传崩溃报告使用的 Token。", "# 用途：POST /api/v1/crash-reports", "# 请求头写法：Authorization: Bearer ", f"CRASH_REPORT_TOKEN={tokens['CRASH_REPORT_TOKEN']['token']}", f"CRASH_REPORT_URL={values['upload_crash_report']['url']}", "", "# 上传符号包使用的 Token。", "# 用途：POST /api/v1/symbols", "# 请求头写法：Authorization: Bearer ", f"CRASH_SYMBOL_TOKEN={tokens['CRASH_SYMBOL_TOKEN']['token']}", f"CRASH_SYMBOL_URL={values['upload_symbols']['url']}", "", "# 查询和下载崩溃报告使用的 Token。", "# 如果 CRASH_ADMIN_TOKEN 为空，就使用 ADMIN_TOKEN。", "# 请求头写法：Authorization: Bearer ", f"CRASH_ADMIN_TOKEN={tokens['CRASH_ADMIN_TOKEN']['token']}", f"ADMIN_TOKEN={tokens['ADMIN_TOKEN']['token']}", f"CRASH_REPORT_DETAIL_URL_TEMPLATE={values['admin_query_and_download']['detail_url_template']}", f"CRASH_REPORT_FILE_URL_TEMPLATE={values['admin_query_and_download']['file_url_template']}", "", "# 可下载文件名。", "CRASH_REPORT_FILE_NAMES=metadata.json,crash.dmp,attachments.zip,server.json", "", "# 幂等规则：Idempotency-Key 必须等于 metadata.json 里的 clientReportId。", "CRASH_IDEMPOTENCY_KEY_RULE=Idempotency-Key must equal metadata.clientReportId", "", "# 崩溃报告服务版本和大小限制。", ] for key, value in values["limits"].items(): lines.append(f"{key}={value}") return "\n".join(lines) def normalize_executable_name(value: str) -> str: normalized = str(value or "").strip().replace(chr(92), "/").strip("/") if not normalized: return "" if normalized.startswith("/") or ":" in normalized or any(part in ("", ".", "..") for part in Path(normalized).parts): raise HTTPException(status_code=400, detail="主程序文件名必须是安全的相对路径") if not normalized.casefold().endswith(".exe"): normalized += ".exe" return normalized