diff --git a/astrbot/core/config/astrbot_config.py b/astrbot/core/config/astrbot_config.py index 44030672b7..45b4a735ec 100644 --- a/astrbot/core/config/astrbot_config.py +++ b/astrbot/core/config/astrbot_config.py @@ -86,6 +86,14 @@ def __init__( ) # 检查配置完整性，并插入 has_new = self.check_config_integrity(default_config, conf) + + dashboard_conf = conf.get("dashboard") + if isinstance(dashboard_conf, dict): + host_val = dashboard_conf.get("host") + if isinstance(host_val, str) and host_val: + dashboard_conf["host"] = [host_val] + has_new = True + reset_dashboard_password = self._consume_reset_dashboard_password_flag() if reset_dashboard_password and "dashboard" in conf: self._reset_generated_dashboard_password(conf) diff --git a/astrbot/core/config/default.py b/astrbot/core/config/default.py index 942bcda65f..069ef48a41 100644 --- a/astrbot/core/config/default.py +++ b/astrbot/core/config/default.py @@ -257,7 +257,7 @@ "password_storage_upgraded": False, "password_change_required": False, "jwt_secret": "", - "host": "0.0.0.0", + "host": ["0.0.0.0", "::"], "port": 6185, "disable_access_log": True, "trust_proxy_headers": False, diff --git a/astrbot/core/utils/io.py b/astrbot/core/utils/io.py index a36bda5e4d..560116bd06 100644 --- a/astrbot/core/utils/io.py +++ b/astrbot/core/utils/io.py @@ -361,5 +361,290 @@ def get_local_ip_addresses(): for addr in addrs: if addr.family == socket.AF_INET: # 使用 socket.AF_INET 代替 psutil.AF_INET network_ips.append(addr.address) + elif addr.family == socket.AF_INET6: + address = addr.address + scope_idx = address.find("%") + if scope_idx != -1: + address = address[:scope_idx] + network_ips.append(address) return network_ips + + +def normalize_host_list(host_raw: str | list[str] | None) -> list[str]: + """Normalize a host config value into a list of host strings. + + Args: + host_raw: A string, list of strings, or None. + + Returns: + A list of non-empty host strings. + """ + if host_raw is None: + return [] + if isinstance(host_raw, list): + return [h for h in host_raw if h] + return [h.strip() for h in str(host_raw).split(",") if h.strip()] + + +def get_dashboard_dist_version(dist_dir: str | Path) -> str | None: + """Read the WebUI version from a dashboard dist directory. + + Args: + dist_dir: Dashboard dist directory path. + + Returns: + The version string from assets/version, or None when unavailable. + """ + + version_file = Path(dist_dir) / "assets" / "version" + try: + if version_file.exists(): + return version_file.read_text(encoding="utf-8").strip() + except (OSError, UnicodeDecodeError) as exc: + logger.warning("Failed to read WebUI version from %s: %s", version_file, exc) + return None + + +def get_bundled_dashboard_dist_path() -> Path: + return Path(get_astrbot_path()) / "astrbot" / "dashboard" / "dist" + + +def _normalize_dashboard_version(version: str) -> str: + version = version.strip() + if version[:1].lower() == "v": + version = version[1:] + if not re.match( + r"^[0-9]+(?:\.[0-9]+)*" + r"(?:-[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?" + r"(?:\+.+)?$", + version, + ): + raise ValueError(f"invalid dashboard version: {version!r}") + return version + + +def is_dashboard_version_compatible( + dashboard_version: str | None, current_version: str +) -> bool: + """Check whether a WebUI version matches the current core version. + + Args: + dashboard_version: Version read from the WebUI assets/version file. + current_version: Current AstrBot core version. + + Returns: + True when both versions are valid SemVer values and compare equal. + """ + + if dashboard_version is None: + return False + try: + return ( + VersionComparator.compare_version( + _normalize_dashboard_version(dashboard_version), + _normalize_dashboard_version(current_version), + ) + == 0 + ) + except (TypeError, ValueError): + return False + + +def is_dashboard_dist_compatible(dist_dir: str | Path, current_version: str) -> bool: + """Check whether a WebUI dist is complete and matches the core version. + + Args: + dist_dir: Dashboard dist directory path. + current_version: Current AstrBot core version. + + Returns: + True when the dist has an index file and a compatible assets/version. + """ + + dist_path = Path(dist_dir) + return (dist_path / "index.html").is_file() and is_dashboard_version_compatible( + get_dashboard_dist_version(dist_path), + current_version, + ) + + +def should_use_bundled_dashboard_dist( + user_dist: str | Path, current_version: str +) -> bool: + """Decide whether bundled WebUI should replace a user data dist. + + Args: + user_dist: Runtime dashboard dist directory under data/. + current_version: Current AstrBot core version. + + Returns: + True when user_dist exists but is missing or mismatched against the + current core version, and bundled WebUI matches the current core version. + """ + + user_dist = Path(user_dist) + user_version = get_dashboard_dist_version(user_dist) + bundled_dist = get_bundled_dashboard_dist_path() + if not user_dist.exists() or not is_dashboard_dist_compatible( + bundled_dist, + current_version, + ): + return False + if user_version is None or not (user_dist / "index.html").is_file(): + return True + try: + return not is_dashboard_version_compatible(user_version, current_version) + except (TypeError, ValueError): + return False + + +async def get_dashboard_version(): + """Return the effective WebUI version for the current runtime. + + Returns: + The matching data/dist version, matching bundled version, or the raw + data/dist version when no compatible bundled WebUI is available. + """ + + from astrbot.core.config.default import VERSION + + # First check user data directory (manually updated / downloaded dashboard). + dist_dir = os.path.join(get_astrbot_data_path(), "dist") + if os.path.exists(dist_dir): + user_version = get_dashboard_dist_version(dist_dir) + if is_dashboard_dist_compatible(dist_dir, VERSION): + return user_version + + bundled = get_bundled_dashboard_dist_path() + if is_dashboard_dist_compatible(bundled, VERSION): + return get_dashboard_dist_version(bundled) + return user_version + + bundled = get_bundled_dashboard_dist_path() + if is_dashboard_dist_compatible(bundled, VERSION): + return get_dashboard_dist_version(bundled) + return None + + +async def download_dashboard( + path: str | None = None, + extract_path: str = "data", + latest: bool = True, + version: str | None = None, + proxy: str | None = None, + progress_callback=None, + extract: bool = True, + allow_insecure_ssl_fallback: bool = True, +) -> None: + """Download dashboard assets and optionally extract them. + + Args: + path: Destination zip path. Defaults to the AstrBot data directory. + extract_path: Directory where assets should be extracted. + latest: Whether to download the latest dashboard build. + version: Specific release tag or commit hash to download. + proxy: Optional download proxy prefix. + progress_callback: Optional callback for download progress payloads. + extract: Whether to extract the archive after download. + allow_insecure_ssl_fallback: Whether certificate failures may retry with + TLS certificate verification disabled. + + Returns: + None. + """ + if path is None: + zip_path = Path(get_astrbot_data_path()).absolute() / "dashboard.zip" + else: + zip_path = Path(path).absolute() + ensure_dir(zip_path.parent) + + if latest or len(str(version)) != 40: + ver_name = "latest" if latest else version + dashboard_release_url = f"https://astrbot-registry.soulter.top/download/astrbot-dashboard/{ver_name}/dist.zip" + logger.info( + f"Downloading AstrBot WebUI from {dashboard_release_url}", + ) + try: + await download_file( + dashboard_release_url, + str(zip_path), + show_progress=True, + progress_callback=progress_callback, + allow_insecure_ssl_fallback=allow_insecure_ssl_fallback, + ) + if not zipfile.is_zipfile(zip_path): + raise RuntimeError( + "Downloaded dashboard package is not a valid ZIP file" + ) + except BaseException as _: + if latest: + # Resolve latest release tag from GitHub API to construct correct asset URL + ssl_context = ssl.create_default_context(cafile=certifi.where()) + async with aiohttp.ClientSession( + connector=aiohttp.TCPConnector(ssl=ssl_context), + trust_env=True, + ) as session: + async with session.get( + "https://api.github.com/repos/AstrBotDevs/AstrBot/releases/latest", + timeout=30, + headers={"Accept": "application/vnd.github+json"}, + ) as api_resp: + api_resp.raise_for_status() + release_data = await api_resp.json() + tag = release_data["tag_name"] + else: + tag = version + dashboard_release_url = f"https://github.com/AstrBotDevs/AstrBot/releases/download/{tag}/AstrBot-{tag}-dashboard.zip" + if proxy: + dashboard_release_url = f"{proxy}/{dashboard_release_url}" + await download_file( + dashboard_release_url, + str(zip_path), + show_progress=True, + progress_callback=progress_callback, + allow_insecure_ssl_fallback=allow_insecure_ssl_fallback, + ) + if not zipfile.is_zipfile(zip_path): + raise RuntimeError( + "Downloaded dashboard package is not a valid ZIP file" + ) + else: + url = f"https://github.com/AstrBotDevs/astrbot-release-harbour/releases/download/release-{version}/dist.zip" + logger.info(f"Downloading AstrBot WebUI from {url}") + if proxy: + url = f"{proxy}/{url}" + await download_file( + url, + str(zip_path), + show_progress=True, + progress_callback=progress_callback, + allow_insecure_ssl_fallback=allow_insecure_ssl_fallback, + ) + if not zipfile.is_zipfile(zip_path): + raise RuntimeError("Downloaded dashboard package is not a valid ZIP file") + if extract: + extract_dashboard(zip_path, extract_path) + + +def extract_dashboard(zip_path: str | Path, extract_path: str | Path = "data") -> None: + """Extract a downloaded dashboard archive. + + Args: + zip_path: Dashboard zip archive path. + extract_path: Directory where the archive contents should be extracted. + + Returns: + None. + """ + + extract_root = Path(extract_path).resolve() + ensure_dir(extract_root) + with zipfile.ZipFile(zip_path, "r") as z: + for member in z.infolist(): + target_path = (extract_root / member.filename).resolve() + if not target_path.is_relative_to(extract_root): + raise ValueError( + f"Unsafe dashboard archive path: {member.filename}", + ) + z.extract(member, extract_root) diff --git a/astrbot/dashboard/server.py b/astrbot/dashboard/server.py index 01388ddd34..99154926f0 100644 --- a/astrbot/dashboard/server.py +++ b/astrbot/dashboard/server.py @@ -21,7 +21,15 @@ from astrbot.core.core_lifecycle import AstrBotCoreLifecycle from astrbot.core.dashboard_assets import resolve_dashboard_dist from astrbot.core.db import BaseDatabase -from astrbot.core.utils.io import get_local_ip_addresses +from astrbot.core.utils.astrbot_path import get_astrbot_data_path +from astrbot.core.utils.io import ( + get_bundled_dashboard_dist_path, + get_dashboard_dist_version, + get_local_ip_addresses, + is_dashboard_dist_compatible, + normalize_host_list, + should_use_bundled_dashboard_dist, +) from astrbot.dashboard.asgi_runtime import ( DashboardRequestState, FastAPIAppAdapter, @@ -515,11 +523,12 @@ def run(self): or os.environ.get("ASTRBOT_DASHBOARD_PORT") or dashboard_config.get("port", 6185) ) - host = ( + host_raw = ( os.environ.get("DASHBOARD_HOST") or os.environ.get("ASTRBOT_DASHBOARD_HOST") - or dashboard_config.get("host", "0.0.0.0") + or dashboard_config.get("host", ["0.0.0.0"]) ) + hosts = normalize_host_list(host_raw) enable = dashboard_config.get("enable", True) ssl_config = dashboard_config.get("ssl", {}) if not isinstance(ssl_config, dict): @@ -540,17 +549,36 @@ def run(self): logger.info("WebUI disabled.") return None - logger.info("Starting WebUI at %s://%s:%s", scheme, host, port) - if host == "0.0.0.0": + bound_urls = [ + f"{scheme}://[{h}]:{port}" if ":" in h else f"{scheme}://{h}:{port}" + for h in hosts + ] + logger.info("Starting WebUI at %s", ", ".join(bound_urls)) + all_interfaces = {"0.0.0.0", "::"} + local_hosts = {"localhost", "127.0.0.1", "::1"} + if all_interfaces & set(hosts): logger.info( "WebUI listens on all interfaces. Check security. Set dashboard.host in data/cmd_config.json to change it.", ) - if host not in ["localhost", "127.0.0.1"]: - try: - ip_addr = get_local_ip_addresses() - except Exception as _: - pass + if not set(hosts).issubset(local_hosts): + if all_interfaces & set(hosts): + try: + ip_addr = get_local_ip_addresses() + except Exception as _: + ip_addr = [] + has_v4_wildcard = "0.0.0.0" in hosts + has_v6_wildcard = "::" in hosts + if has_v4_wildcard and not has_v6_wildcard: + specific_v6 = [h for h in hosts if ":" in h and h != "::"] + ip_addr = [ip for ip in ip_addr if ":" not in ip] + specific_v6 + elif has_v6_wildcard and not has_v4_wildcard: + specific_v4 = [h for h in hosts if ":" not in h and h != "0.0.0.0"] + ip_addr = [ip for ip in ip_addr if ":" in ip] + specific_v4 + else: + ip_addr = [h for h in hosts if h not in local_hosts] + else: + ip_addr = [] if isinstance(port, str): port = int(port) @@ -576,7 +604,10 @@ def run(self): parts = [f"\n ✨✨✨\n AstrBot v{VERSION} {webui_status}\n\n"] parts.append(f" ➜ Local: {scheme}://localhost:{port}\n") for ip in ip_addr: - parts.append(f" ➜ Network: {scheme}://{ip}:{port}\n") + if ":" in ip: + parts.append(f" ➜ Network: {scheme}://[{ip}]:{port}\n") + else: + parts.append(f" ➜ Network: {scheme}://{ip}:{port}\n") parts.append(self._build_dashboard_credentials_display()) display = "".join(parts) @@ -589,7 +620,9 @@ def run(self): # 配置 Hypercorn config = HyperConfig() - config.bind = [f"{host}:{port}"] + config.bind = [ + f"[{h}]:{port}" if ":" in h else f"{h}:{port}" for h in hosts + ] if bool(self.config.get("dashboard", {}).get("trust_proxy_headers", False)): config.logger_class = _ProxyAwareHypercornLogger if ssl_enable: diff --git a/astrbot/dashboard/services/auth_service.py b/astrbot/dashboard/services/auth_service.py index 4ff021f06e..3cf08da458 100644 --- a/astrbot/dashboard/services/auth_service.py +++ b/astrbot/dashboard/services/auth_service.py @@ -12,6 +12,7 @@ from astrbot.core import DEMO_MODE from astrbot.core.config.astrbot_config import AstrBotConfig from astrbot.core.db import BaseDatabase +from astrbot.core.utils.io import normalize_host_list from astrbot.core.desktop_runtime import ( is_desktop_session_auth_enabled, is_loopback_client_host, @@ -482,12 +483,17 @@ async def is_setup_required(self) -> bool: def can_skip_default_password_auth(self) -> bool: if not self.env_flag_enabled(SKIP_DEFAULT_PASSWORD_AUTH_ENV): return False - host = ( + host_raw = ( os.environ.get("DASHBOARD_HOST") or os.environ.get("ASTRBOT_DASHBOARD_HOST") or self.config["dashboard"].get("host", "") ) - return str(host).strip().lower() in LOCAL_DASHBOARD_HOSTS + hosts = normalize_host_list(host_raw) + if not hosts: + return False + return all( + str(h).strip().lower() in LOCAL_DASHBOARD_HOSTS for h in hosts + ) @staticmethod def env_flag_enabled(name: str) -> bool: