1266 lines
42 KiB
Python
1266 lines
42 KiB
Python
"""Validated, atomic JWeBot configuration profiles."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import tarfile
|
||
import tempfile
|
||
import time
|
||
from dataclasses import dataclass
|
||
from pathlib import Path, PurePosixPath
|
||
from typing import Any, Mapping
|
||
|
||
try:
|
||
import tomllib
|
||
except ModuleNotFoundError: # pragma: no cover - Python 3.9/3.10
|
||
import tomli as tomllib
|
||
|
||
|
||
CONFIG_FORMAT_VERSION = 1
|
||
PROFILE_ID = re.compile(r"^[a-z][a-z0-9-]{0,47}$")
|
||
ENVIRONMENT_KEY = re.compile(r"^(?:JWEBOT|WECHAT)_[A-Z0-9_]+$")
|
||
SAFE_ENVIRONMENT_VALUE = re.compile(r"^[A-Za-z0-9_./:@%+,-]*$")
|
||
MAX_CONFIG_FILE_BYTES = 1_000_000
|
||
MAX_CONTEXT_PEOPLE = 500
|
||
MAX_CONTEXT_VALUES = 64
|
||
|
||
SENSITIVE_ENVIRONMENT_KEYS = frozenset(
|
||
{
|
||
"JWEBOT_EDGE_DEEPSEEK_API_KEY",
|
||
"JWEBOT_EDGE_STEAM_API_KEY",
|
||
"JWEBOT_HEALTH_NOTIFICATION_TOKEN",
|
||
}
|
||
)
|
||
|
||
REQUIRED_PROFILE_FILES = (
|
||
"manifest.toml",
|
||
"components/bot.toml",
|
||
"components/providers.toml",
|
||
"components/runtime.toml",
|
||
"components/debug.toml",
|
||
"components/notification.toml",
|
||
"components/deployment.toml",
|
||
"context/people.json",
|
||
"secrets/runtime.env",
|
||
"overrides/runtime.env",
|
||
"overrides/root.env",
|
||
)
|
||
|
||
DEFAULT_PROFILE_FILES: Mapping[str, str] = {
|
||
"manifest.toml": """format_version = 1
|
||
id = "default"
|
||
description = "Local JWeBot configuration"
|
||
""",
|
||
"components/bot.toml": """[bot]
|
||
mention = "JWeBot"
|
||
identity_note = "You are JWeBot, a concise assistant for one configured group."
|
||
cloud_agent = "wechat-responder"
|
||
memory_remote_path = "/home/node/.openclaw/workspace/MEMORY.md"
|
||
""",
|
||
"components/providers.toml": """[deepseek]
|
||
base_url = "https://api.deepseek.com"
|
||
|
||
[cloud]
|
||
host = ""
|
||
|
||
[steam]
|
||
owner_id64 = ""
|
||
proxy_url = "http://127.0.0.1:7890"
|
||
""",
|
||
"components/runtime.toml": """[features]
|
||
edge_enabled = true
|
||
local_extensions_enabled = true
|
||
telemetry_enabled = true
|
||
ocr_prewarm = true
|
||
delivery_enabled = true
|
||
|
||
[extensions]
|
||
mode = "in-process"
|
||
transport = "unix"
|
||
publisher = ""
|
||
host_timeout_seconds = 8
|
||
fallback_enabled = true
|
||
|
||
[gui]
|
||
sender_timeout_seconds = 55
|
||
""",
|
||
"components/debug.toml": """[debug]
|
||
enabled = false
|
||
capture_payloads = false
|
||
incident_retention_days = 30
|
||
session_retention_days = 14
|
||
""",
|
||
"components/notification.toml": """[notification]
|
||
enabled = false
|
||
command = []
|
||
timeout_seconds = 30
|
||
""",
|
||
"components/deployment.toml": """[deployment]
|
||
runtime_user = "jwebot"
|
||
runtime_home = "/home/jwebot"
|
||
runtime_python = "/opt/jwebot/venv/bin/python"
|
||
backup_root = "/var/backups/jwebot"
|
||
""",
|
||
"context/people.json": """{
|
||
"people": [],
|
||
"clock_scan_targets": []
|
||
}
|
||
""",
|
||
"secrets/runtime.env": """# Secrets only. Keep this file mode 0600.
|
||
JWEBOT_EDGE_DEEPSEEK_API_KEY=
|
||
JWEBOT_EDGE_STEAM_API_KEY=
|
||
""",
|
||
"overrides/runtime.env": """# Optional non-secret runtime overrides.
|
||
""",
|
||
"overrides/root.env": """# Optional root-only deployment overrides.
|
||
""",
|
||
}
|
||
|
||
|
||
class ConfigError(ValueError):
|
||
"""Configuration input is invalid or cannot be activated safely."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ConfigLayout:
|
||
config_root: Path
|
||
state_root: Path
|
||
|
||
@classmethod
|
||
def from_environment(
|
||
cls,
|
||
*,
|
||
home: Path | None = None,
|
||
config_root: Path | None = None,
|
||
state_root: Path | None = None,
|
||
) -> "ConfigLayout":
|
||
base = (home or Path.home()).expanduser()
|
||
configured = config_root or Path(
|
||
os.environ.get(
|
||
"JWEBOT_CONFIG_ROOT",
|
||
str(base / ".config" / "jwebot"),
|
||
)
|
||
)
|
||
state = state_root or Path(
|
||
os.environ.get(
|
||
"JWEBOT_CONFIG_STATE_ROOT",
|
||
str(base / ".local" / "state" / "jwebot"),
|
||
)
|
||
)
|
||
return cls(configured.expanduser(), state.expanduser())
|
||
|
||
@property
|
||
def profiles_root(self) -> Path:
|
||
return self.config_root / "profiles"
|
||
|
||
@property
|
||
def current_link(self) -> Path:
|
||
return self.config_root / "current"
|
||
|
||
def profile(self, profile_id: str) -> Path:
|
||
validate_profile_id(profile_id)
|
||
return self.profiles_root / profile_id
|
||
|
||
def profile_state(self, profile_id: str) -> Path:
|
||
validate_profile_id(profile_id)
|
||
return self.state_root / "profiles" / profile_id
|
||
|
||
def active_profile(self) -> Path | None:
|
||
if not self.current_link.is_symlink():
|
||
return None
|
||
target = self.current_link.resolve(strict=False)
|
||
try:
|
||
target.relative_to(self.profiles_root.resolve(strict=False))
|
||
except ValueError as error:
|
||
raise ConfigError("active profile points outside the profile root") from error
|
||
return target if target.is_dir() else None
|
||
|
||
|
||
def ensure_private_directory(path: Path) -> None:
|
||
path.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||
os.chmod(path, 0o700)
|
||
|
||
|
||
def validate_profile_id(value: object) -> str:
|
||
profile_id = str(value or "")
|
||
if not PROFILE_ID.fullmatch(profile_id):
|
||
raise ConfigError("profile id must match [a-z][a-z0-9-]{0,47}")
|
||
return profile_id
|
||
|
||
|
||
def read_text(path: Path) -> str:
|
||
try:
|
||
size = path.stat().st_size
|
||
except OSError as error:
|
||
raise ConfigError(f"missing configuration file: {path.name}") from error
|
||
if size > MAX_CONFIG_FILE_BYTES:
|
||
raise ConfigError(f"configuration file is too large: {path.name}")
|
||
try:
|
||
return path.read_text(encoding="utf-8")
|
||
except (OSError, UnicodeError) as error:
|
||
raise ConfigError(f"cannot read configuration file: {path.name}") from error
|
||
|
||
|
||
def load_toml(path: Path) -> dict[str, Any]:
|
||
try:
|
||
value = tomllib.loads(read_text(path))
|
||
except tomllib.TOMLDecodeError as error:
|
||
raise ConfigError(f"invalid TOML: {path.name}") from error
|
||
if not isinstance(value, dict):
|
||
raise ConfigError(f"TOML root must be a table: {path.name}")
|
||
return value
|
||
|
||
|
||
def load_json(path: Path) -> dict[str, Any]:
|
||
try:
|
||
value = json.loads(read_text(path))
|
||
except json.JSONDecodeError as error:
|
||
raise ConfigError(f"invalid JSON: {path.name}") from error
|
||
if not isinstance(value, dict):
|
||
raise ConfigError(f"JSON root must be an object: {path.name}")
|
||
return value
|
||
|
||
|
||
def table(value: Mapping[str, Any], name: str) -> dict[str, Any]:
|
||
result = value.get(name, {})
|
||
if not isinstance(result, dict):
|
||
raise ConfigError(f"{name} must be a table")
|
||
return result
|
||
|
||
|
||
def require_string(
|
||
value: Mapping[str, Any],
|
||
key: str,
|
||
*,
|
||
limit: int,
|
||
allow_empty: bool = True,
|
||
) -> str:
|
||
result = value.get(key, "")
|
||
if not isinstance(result, str):
|
||
raise ConfigError(f"{key} must be a string")
|
||
result = result.strip()
|
||
if not allow_empty and not result:
|
||
raise ConfigError(f"{key} must not be empty")
|
||
if len(result) > limit:
|
||
raise ConfigError(f"{key} exceeds {limit} characters")
|
||
if "\n" in result or "\r" in result or "\x00" in result:
|
||
raise ConfigError(f"{key} contains an invalid control character")
|
||
return result
|
||
|
||
|
||
def require_bool(value: Mapping[str, Any], key: str, default: bool) -> bool:
|
||
result = value.get(key, default)
|
||
if not isinstance(result, bool):
|
||
raise ConfigError(f"{key} must be a boolean")
|
||
return result
|
||
|
||
|
||
def require_int(
|
||
value: Mapping[str, Any],
|
||
key: str,
|
||
default: int,
|
||
*,
|
||
minimum: int,
|
||
maximum: int,
|
||
) -> int:
|
||
result = value.get(key, default)
|
||
if not isinstance(result, int) or isinstance(result, bool):
|
||
raise ConfigError(f"{key} must be an integer")
|
||
if not minimum <= result <= maximum:
|
||
raise ConfigError(f"{key} must be between {minimum} and {maximum}")
|
||
return result
|
||
|
||
|
||
def parse_environment(path: Path) -> dict[str, str]:
|
||
values: dict[str, str] = {}
|
||
for line_number, raw in enumerate(read_text(path).splitlines(), start=1):
|
||
line = raw.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
key, separator, value = line.partition("=")
|
||
key = key.strip()
|
||
if not separator or not ENVIRONMENT_KEY.fullmatch(key):
|
||
raise ConfigError(
|
||
f"invalid environment entry in {path.name}:{line_number}"
|
||
)
|
||
if key in values:
|
||
raise ConfigError(f"duplicate environment key: {key}")
|
||
if "\x00" in value or "\n" in value or "\r" in value:
|
||
raise ConfigError(f"invalid environment value: {key}")
|
||
values[key] = value.strip()
|
||
return values
|
||
|
||
|
||
def validate_private_context(value: Mapping[str, Any]) -> None:
|
||
people = value.get("people", [])
|
||
targets = value.get("clock_scan_targets", [])
|
||
if not isinstance(people, list) or len(people) > MAX_CONTEXT_PEOPLE:
|
||
raise ConfigError("people must be a bounded array")
|
||
if not isinstance(targets, list) or len(targets) > MAX_CONTEXT_PEOPLE:
|
||
raise ConfigError("clock_scan_targets must be a bounded array")
|
||
steam_ids: set[str] = set()
|
||
for person in people:
|
||
if not isinstance(person, dict):
|
||
raise ConfigError("each people entry must be an object")
|
||
require_string(person, "canonical_name", limit=120, allow_empty=False)
|
||
require_string(person, "identity_fact", limit=240)
|
||
require_string(person, "display_id", limit=120)
|
||
steam_id64 = require_string(person, "steam_id64", limit=17)
|
||
if steam_id64 and not re.fullmatch(r"\d{17}", steam_id64):
|
||
raise ConfigError("people steam_id64 must be a 17-digit SteamID")
|
||
if steam_id64 in steam_ids:
|
||
raise ConfigError("people steam_id64 values must be unique")
|
||
if steam_id64:
|
||
steam_ids.add(steam_id64)
|
||
for key in ("aliases", "prefixes", "steam_aliases"):
|
||
entries = person.get(key, [])
|
||
if not isinstance(entries, list) or len(entries) > MAX_CONTEXT_VALUES:
|
||
raise ConfigError(f"{key} must be a bounded array")
|
||
if any(
|
||
not isinstance(entry, str)
|
||
or not entry.strip()
|
||
or len(entry.strip()) > 120
|
||
for entry in entries
|
||
):
|
||
raise ConfigError(f"{key} contains an invalid value")
|
||
for target in targets:
|
||
if not isinstance(target, dict):
|
||
raise ConfigError("each clock target must be an object")
|
||
require_string(target, "canonical_name", limit=120, allow_empty=False)
|
||
require_string(target, "display_id", limit=120, allow_empty=False)
|
||
|
||
|
||
def validate_profile(path: Path) -> str:
|
||
root = path.expanduser().resolve()
|
||
if not root.is_dir():
|
||
raise ConfigError("profile directory is missing")
|
||
for child in root.rglob("*"):
|
||
if child.is_symlink():
|
||
raise ConfigError("profile files must not be symlinks")
|
||
for relative in REQUIRED_PROFILE_FILES:
|
||
candidate = root / relative
|
||
if not candidate.is_file():
|
||
raise ConfigError(f"profile is missing {relative}")
|
||
|
||
manifest = load_toml(root / "manifest.toml")
|
||
if manifest.get("format_version") != CONFIG_FORMAT_VERSION:
|
||
raise ConfigError("unsupported configuration format version")
|
||
profile_id = validate_profile_id(manifest.get("id"))
|
||
require_string(manifest, "description", limit=240)
|
||
|
||
bot = table(load_toml(root / "components" / "bot.toml"), "bot")
|
||
require_string(bot, "mention", limit=40, allow_empty=False)
|
||
require_string(bot, "identity_note", limit=500)
|
||
require_string(bot, "cloud_agent", limit=80, allow_empty=False)
|
||
require_string(bot, "memory_remote_path", limit=240, allow_empty=False)
|
||
|
||
providers = load_toml(root / "components" / "providers.toml")
|
||
deepseek = table(providers, "deepseek")
|
||
cloud = table(providers, "cloud")
|
||
steam = table(providers, "steam")
|
||
require_string(deepseek, "base_url", limit=240, allow_empty=False)
|
||
require_string(cloud, "host", limit=240)
|
||
owner_id = require_string(steam, "owner_id64", limit=17)
|
||
if owner_id and not re.fullmatch(r"\d{17}", owner_id):
|
||
raise ConfigError("steam.owner_id64 must be a 17-digit SteamID")
|
||
proxy = require_string(steam, "proxy_url", limit=120, allow_empty=False)
|
||
if not re.fullmatch(
|
||
r"http://(?:127\.0\.0\.1|localhost):(?:[1-9]\d{0,4})",
|
||
proxy,
|
||
):
|
||
raise ConfigError("steam.proxy_url must use a loopback HTTP proxy")
|
||
|
||
runtime = load_toml(root / "components" / "runtime.toml")
|
||
features = table(runtime, "features")
|
||
extensions = table(runtime, "extensions")
|
||
gui = table(runtime, "gui")
|
||
for key in (
|
||
"edge_enabled",
|
||
"local_extensions_enabled",
|
||
"telemetry_enabled",
|
||
"ocr_prewarm",
|
||
"delivery_enabled",
|
||
):
|
||
require_bool(features, key, True)
|
||
mode = require_string(extensions, "mode", limit=24, allow_empty=False)
|
||
if mode not in {"in-process", "prefer-host", "host-only"}:
|
||
raise ConfigError("extensions.mode is invalid")
|
||
transport = require_string(
|
||
extensions,
|
||
"transport",
|
||
limit=16,
|
||
allow_empty=False,
|
||
)
|
||
if transport not in {"unix", "process"}:
|
||
raise ConfigError("extensions.transport is invalid")
|
||
require_string(extensions, "publisher", limit=120)
|
||
require_int(
|
||
extensions,
|
||
"host_timeout_seconds",
|
||
8,
|
||
minimum=1,
|
||
maximum=30,
|
||
)
|
||
require_bool(extensions, "fallback_enabled", True)
|
||
require_int(
|
||
gui,
|
||
"sender_timeout_seconds",
|
||
55,
|
||
minimum=1,
|
||
maximum=300,
|
||
)
|
||
|
||
debug = table(load_toml(root / "components" / "debug.toml"), "debug")
|
||
require_bool(debug, "enabled", False)
|
||
require_bool(debug, "capture_payloads", False)
|
||
require_int(
|
||
debug,
|
||
"incident_retention_days",
|
||
30,
|
||
minimum=1,
|
||
maximum=3650,
|
||
)
|
||
require_int(
|
||
debug,
|
||
"session_retention_days",
|
||
14,
|
||
minimum=1,
|
||
maximum=3650,
|
||
)
|
||
|
||
notification = table(
|
||
load_toml(root / "components" / "notification.toml"),
|
||
"notification",
|
||
)
|
||
notification_enabled = require_bool(notification, "enabled", False)
|
||
command = notification.get("command", [])
|
||
if (
|
||
not isinstance(command, list)
|
||
or len(command) > 32
|
||
or any(
|
||
not isinstance(argument, str)
|
||
or not argument
|
||
or len(argument) > 500
|
||
or "\x00" in argument
|
||
for argument in command
|
||
)
|
||
):
|
||
raise ConfigError("notification.command must be a bounded string array")
|
||
if notification_enabled and (
|
||
not command or not str(command[0]).startswith("/")
|
||
):
|
||
raise ConfigError("enabled notification.command must start with an absolute path")
|
||
require_int(
|
||
notification,
|
||
"timeout_seconds",
|
||
30,
|
||
minimum=1,
|
||
maximum=120,
|
||
)
|
||
|
||
deployment = table(
|
||
load_toml(root / "components" / "deployment.toml"),
|
||
"deployment",
|
||
)
|
||
require_string(deployment, "runtime_user", limit=64, allow_empty=False)
|
||
for key in ("runtime_home", "runtime_python", "backup_root"):
|
||
value = require_string(deployment, key, limit=240, allow_empty=False)
|
||
if not value.startswith("/"):
|
||
raise ConfigError(f"deployment.{key} must be an absolute path")
|
||
|
||
validate_private_context(load_json(root / "context" / "people.json"))
|
||
secrets = parse_environment(root / "secrets" / "runtime.env")
|
||
if any(key not in SENSITIVE_ENVIRONMENT_KEYS for key in secrets):
|
||
raise ConfigError("secrets/runtime.env contains a non-secret key")
|
||
runtime_overrides = parse_environment(root / "overrides" / "runtime.env")
|
||
if any(key in SENSITIVE_ENVIRONMENT_KEYS for key in runtime_overrides):
|
||
raise ConfigError("secret keys belong in secrets/runtime.env")
|
||
parse_environment(root / "overrides" / "root.env")
|
||
return profile_id
|
||
|
||
|
||
def bool_value(value: bool) -> str:
|
||
return "1" if value else "0"
|
||
|
||
|
||
def encode_environment(values: Mapping[str, object]) -> str:
|
||
lines = []
|
||
for key, raw in values.items():
|
||
value = str(raw)
|
||
if not ENVIRONMENT_KEY.fullmatch(key):
|
||
raise ConfigError(f"invalid generated environment key: {key}")
|
||
if not SAFE_ENVIRONMENT_VALUE.fullmatch(value):
|
||
raise ConfigError(f"generated value requires unsupported quoting: {key}")
|
||
lines.append(f"{key}={value}")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def compile_profile(path: Path, layout: ConfigLayout) -> Path:
|
||
profile_id = validate_profile(path)
|
||
root = path.expanduser().resolve()
|
||
ensure_private_directory(layout.config_root)
|
||
ensure_private_directory(layout.profiles_root)
|
||
try:
|
||
root.relative_to(layout.profiles_root.resolve())
|
||
except ValueError:
|
||
pass
|
||
else:
|
||
os.chmod(root, 0o700)
|
||
bot = table(load_toml(root / "components" / "bot.toml"), "bot")
|
||
providers = load_toml(root / "components" / "providers.toml")
|
||
deepseek = table(providers, "deepseek")
|
||
cloud = table(providers, "cloud")
|
||
steam = table(providers, "steam")
|
||
runtime = load_toml(root / "components" / "runtime.toml")
|
||
features = table(runtime, "features")
|
||
extensions = table(runtime, "extensions")
|
||
gui = table(runtime, "gui")
|
||
debug = table(load_toml(root / "components" / "debug.toml"), "debug")
|
||
notification = table(
|
||
load_toml(root / "components" / "notification.toml"),
|
||
"notification",
|
||
)
|
||
deployment = table(
|
||
load_toml(root / "components" / "deployment.toml"),
|
||
"deployment",
|
||
)
|
||
private_context = load_json(root / "context" / "people.json")
|
||
private_context["bot"] = {
|
||
"mention": require_string(bot, "mention", limit=40, allow_empty=False),
|
||
"identity_note": require_string(bot, "identity_note", limit=500),
|
||
}
|
||
|
||
profile_state = layout.profile_state(profile_id)
|
||
runtime_values: dict[str, object] = {
|
||
"JWEBOT_CONFIG_PROFILE": profile_id,
|
||
"JWEBOT_BOT_MENTION": require_string(
|
||
bot,
|
||
"mention",
|
||
limit=40,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_EDGE_ENABLED": bool_value(
|
||
require_bool(features, "edge_enabled", True)
|
||
),
|
||
"JWEBOT_LOCAL_EXTENSIONS_ENABLED": bool_value(
|
||
require_bool(features, "local_extensions_enabled", True)
|
||
),
|
||
"JWEBOT_TELEMETRY_ENABLED": bool_value(
|
||
require_bool(features, "telemetry_enabled", True)
|
||
),
|
||
"WECHAT_OCR_PREWARM": bool_value(
|
||
require_bool(features, "ocr_prewarm", True)
|
||
),
|
||
"JWEBOT_EDGE_DELIVERY_ENABLED": bool_value(
|
||
require_bool(features, "delivery_enabled", True)
|
||
),
|
||
"JWEBOT_EDGE_DEEPSEEK_BASE_URL": require_string(
|
||
deepseek,
|
||
"base_url",
|
||
limit=240,
|
||
allow_empty=False,
|
||
),
|
||
"WECHAT_HEALTH_CLOUD_HOST": require_string(cloud, "host", limit=240),
|
||
"JWEBOT_EDGE_CLOUD_AGENT": require_string(
|
||
bot,
|
||
"cloud_agent",
|
||
limit=80,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_EDGE_MEMORY_REMOTE_PATH": require_string(
|
||
bot,
|
||
"memory_remote_path",
|
||
limit=240,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_EDGE_STEAM_ID64": require_string(
|
||
steam,
|
||
"owner_id64",
|
||
limit=17,
|
||
),
|
||
"JWEBOT_EDGE_STEAM_PROXY_URL": require_string(
|
||
steam,
|
||
"proxy_url",
|
||
limit=120,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_EXTENSION_MODE": require_string(
|
||
extensions,
|
||
"mode",
|
||
limit=24,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_EXTENSION_TRANSPORT": require_string(
|
||
extensions,
|
||
"transport",
|
||
limit=16,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_EXTENSION_PUBLISHER": require_string(
|
||
extensions,
|
||
"publisher",
|
||
limit=120,
|
||
),
|
||
"JWEBOT_EXTENSION_HOST_TIMEOUT_SECONDS": require_int(
|
||
extensions,
|
||
"host_timeout_seconds",
|
||
8,
|
||
minimum=1,
|
||
maximum=30,
|
||
),
|
||
"JWEBOT_EXTENSION_FALLBACK_ENABLED": bool_value(
|
||
require_bool(extensions, "fallback_enabled", True)
|
||
),
|
||
"JWEBOT_GUI_SENDER_TIMEOUT_SECONDS": require_int(
|
||
gui,
|
||
"sender_timeout_seconds",
|
||
55,
|
||
minimum=1,
|
||
maximum=300,
|
||
),
|
||
"JWEBOT_DEBUG_ENABLED": bool_value(
|
||
require_bool(debug, "enabled", False)
|
||
),
|
||
"JWEBOT_DEBUG_CAPTURE_PAYLOADS": bool_value(
|
||
require_bool(debug, "capture_payloads", False)
|
||
),
|
||
"JWEBOT_DEBUG_INCIDENT_RETENTION_DAYS": require_int(
|
||
debug,
|
||
"incident_retention_days",
|
||
30,
|
||
minimum=1,
|
||
maximum=3650,
|
||
),
|
||
"JWEBOT_DEBUG_SESSION_RETENTION_DAYS": require_int(
|
||
debug,
|
||
"session_retention_days",
|
||
14,
|
||
minimum=1,
|
||
maximum=3650,
|
||
),
|
||
"JWEBOT_DEBUG_INCIDENT_ROOT": str(profile_state / "incidents"),
|
||
"JWEBOT_DEBUG_SESSION_ROOT": str(profile_state / "debug" / "sessions"),
|
||
"JWEBOT_HEALTH_NOTIFICATION_CONFIG": str(
|
||
root / "generated" / "notification.json"
|
||
),
|
||
"JWEBOT_RUNTIME_USER": require_string(
|
||
deployment,
|
||
"runtime_user",
|
||
limit=64,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_RUNTIME_HOME": require_string(
|
||
deployment,
|
||
"runtime_home",
|
||
limit=240,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_RUNTIME_PYTHON": require_string(
|
||
deployment,
|
||
"runtime_python",
|
||
limit=240,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_PRIVATE_CONTEXT_FILE": str(
|
||
root / "generated" / "private-context.json"
|
||
),
|
||
}
|
||
runtime_values.update(
|
||
parse_environment(root / "overrides" / "runtime.env")
|
||
)
|
||
runtime_values.update(parse_environment(root / "secrets" / "runtime.env"))
|
||
|
||
root_values: dict[str, object] = {
|
||
"JWEBOT_RUNTIME_USER": require_string(
|
||
deployment,
|
||
"runtime_user",
|
||
limit=64,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_RUNTIME_HOME": require_string(
|
||
deployment,
|
||
"runtime_home",
|
||
limit=240,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_RUNTIME_PYTHON": require_string(
|
||
deployment,
|
||
"runtime_python",
|
||
limit=240,
|
||
allow_empty=False,
|
||
),
|
||
"JWEBOT_PRODUCTION_BACKUP_ROOT": require_string(
|
||
deployment,
|
||
"backup_root",
|
||
limit=240,
|
||
allow_empty=False,
|
||
),
|
||
}
|
||
root_values.update(parse_environment(root / "overrides" / "root.env"))
|
||
|
||
generated = root / "generated"
|
||
temporary = root / f".generated-{os.getpid()}-{time.time_ns()}"
|
||
temporary.mkdir(mode=0o700)
|
||
try:
|
||
runtime_path = temporary / "runtime.env"
|
||
root_path = temporary / "root.env"
|
||
context_path = temporary / "private-context.json"
|
||
notification_path = temporary / "notification.json"
|
||
runtime_path.write_text(
|
||
encode_environment(runtime_values),
|
||
encoding="utf-8",
|
||
)
|
||
root_path.write_text(encode_environment(root_values), encoding="utf-8")
|
||
context_path.write_text(
|
||
json.dumps(
|
||
private_context,
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
sort_keys=True,
|
||
)
|
||
+ "\n",
|
||
encoding="utf-8",
|
||
)
|
||
notification_path.write_text(
|
||
json.dumps(
|
||
{
|
||
"enabled": require_bool(notification, "enabled", False),
|
||
"command": notification.get("command", []),
|
||
"timeout_seconds": require_int(
|
||
notification,
|
||
"timeout_seconds",
|
||
30,
|
||
minimum=1,
|
||
maximum=120,
|
||
),
|
||
},
|
||
indent=2,
|
||
sort_keys=True,
|
||
)
|
||
+ "\n",
|
||
encoding="utf-8",
|
||
)
|
||
generated_files = (
|
||
runtime_path,
|
||
root_path,
|
||
context_path,
|
||
notification_path,
|
||
)
|
||
for candidate in generated_files:
|
||
os.chmod(candidate, 0o600)
|
||
digest_manifest = {
|
||
"format_version": CONFIG_FORMAT_VERSION,
|
||
"profile": profile_id,
|
||
"generated_at": int(time.time()),
|
||
"files": {
|
||
candidate.name: hashlib.sha256(candidate.read_bytes()).hexdigest()
|
||
for candidate in generated_files
|
||
},
|
||
}
|
||
(temporary / "manifest.json").write_text(
|
||
json.dumps(digest_manifest, indent=2, sort_keys=True) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
os.chmod(temporary / "manifest.json", 0o600)
|
||
previous = root / ".generated-previous"
|
||
if previous.exists():
|
||
shutil.rmtree(previous)
|
||
if generated.exists():
|
||
generated.rename(previous)
|
||
temporary.rename(generated)
|
||
if previous.exists():
|
||
shutil.rmtree(previous)
|
||
except Exception:
|
||
shutil.rmtree(temporary, ignore_errors=True)
|
||
raise
|
||
|
||
for directory in (
|
||
layout.state_root,
|
||
layout.state_root / "profiles",
|
||
profile_state,
|
||
profile_state / "debug" / "sessions",
|
||
profile_state / "incidents",
|
||
profile_state / "telemetry",
|
||
profile_state / "archives",
|
||
profile_state / "root",
|
||
):
|
||
ensure_private_directory(directory)
|
||
return generated
|
||
|
||
|
||
def activate_profile(profile_id: str, layout: ConfigLayout) -> Path:
|
||
profile = layout.profile(profile_id)
|
||
generated = compile_profile(profile, layout)
|
||
ensure_private_directory(layout.config_root)
|
||
temporary_link = layout.config_root / f".current-{os.getpid()}"
|
||
temporary_link.unlink(missing_ok=True)
|
||
temporary_link.symlink_to(Path("profiles") / profile_id)
|
||
temporary_link.replace(layout.current_link)
|
||
return generated
|
||
|
||
|
||
def write_default_profile(
|
||
profile_id: str,
|
||
layout: ConfigLayout,
|
||
*,
|
||
force: bool = False,
|
||
) -> Path:
|
||
profile_id = validate_profile_id(profile_id)
|
||
destination = layout.profile(profile_id)
|
||
if destination.exists():
|
||
if not force:
|
||
raise ConfigError("profile already exists")
|
||
shutil.rmtree(destination)
|
||
temporary = layout.profiles_root / f".{profile_id}-{os.getpid()}"
|
||
shutil.rmtree(temporary, ignore_errors=True)
|
||
temporary.mkdir(parents=True, mode=0o700)
|
||
try:
|
||
for relative, content in DEFAULT_PROFILE_FILES.items():
|
||
candidate = temporary / relative
|
||
candidate.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||
if relative == "manifest.toml":
|
||
content = content.replace('id = "default"', f'id = "{profile_id}"')
|
||
candidate.write_text(content, encoding="utf-8")
|
||
os.chmod(candidate, 0o600)
|
||
validate_profile(temporary)
|
||
ensure_private_directory(layout.config_root)
|
||
ensure_private_directory(destination.parent)
|
||
temporary.rename(destination)
|
||
except Exception:
|
||
shutil.rmtree(temporary, ignore_errors=True)
|
||
raise
|
||
return destination
|
||
|
||
|
||
def copy_profile_tree(source: Path, destination: Path) -> None:
|
||
for candidate in source.rglob("*"):
|
||
relative = candidate.relative_to(source)
|
||
target = destination / relative
|
||
if candidate.is_symlink():
|
||
raise ConfigError("imported profiles must not contain symlinks")
|
||
if candidate.is_dir():
|
||
target.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||
elif candidate.is_file():
|
||
target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||
shutil.copyfile(candidate, target)
|
||
os.chmod(target, 0o600)
|
||
|
||
|
||
def safe_extract_profile(archive: Path, destination: Path) -> Path:
|
||
with tarfile.open(archive, "r:*") as bundle:
|
||
members = bundle.getmembers()
|
||
for member in members:
|
||
relative = PurePosixPath(member.name)
|
||
if (
|
||
relative.is_absolute()
|
||
or ".." in relative.parts
|
||
or member.issym()
|
||
or member.islnk()
|
||
or member.isdev()
|
||
):
|
||
raise ConfigError("configuration archive contains an unsafe path")
|
||
bundle.extractall(destination, members=members)
|
||
manifests = list(destination.rglob("manifest.toml"))
|
||
if len(manifests) != 1:
|
||
raise ConfigError("configuration archive must contain one profile")
|
||
return manifests[0].parent
|
||
|
||
|
||
def import_profile(
|
||
source: Path,
|
||
layout: ConfigLayout,
|
||
*,
|
||
replace: bool = False,
|
||
) -> str:
|
||
source = source.expanduser().resolve()
|
||
with tempfile.TemporaryDirectory(prefix="jwebot-config-import-") as directory:
|
||
staging = Path(directory)
|
||
profile_source = (
|
||
source
|
||
if source.is_dir()
|
||
else safe_extract_profile(source, staging)
|
||
)
|
||
profile_id = validate_profile(profile_source)
|
||
destination = layout.profile(profile_id)
|
||
if destination.exists() and not replace:
|
||
raise ConfigError("profile already exists")
|
||
temporary = layout.profiles_root / f".{profile_id}-import-{os.getpid()}"
|
||
shutil.rmtree(temporary, ignore_errors=True)
|
||
temporary.mkdir(parents=True, mode=0o700)
|
||
try:
|
||
copy_profile_tree(profile_source, temporary)
|
||
validate_profile(temporary)
|
||
if destination.exists():
|
||
shutil.rmtree(destination)
|
||
ensure_private_directory(layout.config_root)
|
||
ensure_private_directory(destination.parent)
|
||
temporary.rename(destination)
|
||
except Exception:
|
||
shutil.rmtree(temporary, ignore_errors=True)
|
||
raise
|
||
return profile_id
|
||
|
||
|
||
def redact_environment_file(path: Path) -> None:
|
||
values = parse_environment(path)
|
||
safe = {
|
||
key: "" if is_sensitive_key(key) else value
|
||
for key, value in values.items()
|
||
}
|
||
path.write_text(encode_environment(safe), encoding="utf-8")
|
||
os.chmod(path, 0o600)
|
||
|
||
|
||
def is_sensitive_key(key: str) -> bool:
|
||
return (
|
||
key in SENSITIVE_ENVIRONMENT_KEYS
|
||
or key.endswith(("_KEY", "_TOKEN", "_PASSWORD", "_SECRET"))
|
||
)
|
||
|
||
|
||
def export_profile(
|
||
profile_id: str,
|
||
output: Path,
|
||
layout: ConfigLayout,
|
||
*,
|
||
include_private: bool = False,
|
||
) -> Path:
|
||
source = layout.profile(profile_id)
|
||
validate_profile(source)
|
||
output = output.expanduser().resolve()
|
||
output.parent.mkdir(parents=True, exist_ok=True)
|
||
with tempfile.TemporaryDirectory(prefix="jwebot-config-export-") as directory:
|
||
root = Path(directory) / profile_id
|
||
copy_profile_tree(source, root)
|
||
shutil.rmtree(root / "generated", ignore_errors=True)
|
||
if not include_private:
|
||
redact_environment_file(root / "secrets" / "runtime.env")
|
||
redact_environment_file(root / "overrides" / "runtime.env")
|
||
redact_environment_file(root / "overrides" / "root.env")
|
||
context = load_json(root / "context" / "people.json")
|
||
context["people"] = []
|
||
context["clock_scan_targets"] = []
|
||
(root / "context" / "people.json").write_text(
|
||
json.dumps(context, ensure_ascii=False, indent=2, sort_keys=True)
|
||
+ "\n",
|
||
encoding="utf-8",
|
||
)
|
||
with tarfile.open(output, "w:gz") as bundle:
|
||
bundle.add(root, arcname=profile_id, recursive=True)
|
||
os.chmod(output, 0o600)
|
||
return output
|
||
|
||
|
||
def archive_legacy_material(
|
||
source_root: Path,
|
||
profile_id: str,
|
||
layout: ConfigLayout,
|
||
) -> Path:
|
||
source_root = source_root.expanduser().resolve()
|
||
destination = (
|
||
layout.profile_state(profile_id)
|
||
/ "archives"
|
||
/ time.strftime("legacy-%Y%m%d-%H%M%S")
|
||
)
|
||
ensure_private_directory(layout.state_root)
|
||
ensure_private_directory(layout.state_root / "profiles")
|
||
ensure_private_directory(layout.profile_state(profile_id))
|
||
ensure_private_directory(layout.profile_state(profile_id) / "archives")
|
||
ensure_private_directory(destination)
|
||
candidates = [
|
||
source_root / ".dbg",
|
||
source_root / "dbg",
|
||
source_root / "docs" / "debug",
|
||
source_root / "docs" / "reports",
|
||
source_root / "ops",
|
||
]
|
||
candidates.extend(source_root.glob("debug-*.md"))
|
||
candidates.extend(source_root.glob("homeserver-apikey-*.md"))
|
||
for candidate in candidates:
|
||
if not candidate.exists() or candidate.is_symlink():
|
||
continue
|
||
relative = candidate.relative_to(source_root)
|
||
target = destination / relative
|
||
if candidate.is_dir():
|
||
shutil.copytree(candidate, target)
|
||
else:
|
||
target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||
shutil.copyfile(candidate, target)
|
||
for copied in [target, *target.rglob("*")] if target.is_dir() else [target]:
|
||
if copied.is_dir():
|
||
os.chmod(copied, 0o700)
|
||
elif copied.is_file():
|
||
os.chmod(copied, 0o600)
|
||
return destination
|
||
|
||
|
||
def migrate_legacy_environment(
|
||
source: Path,
|
||
profile_id: str,
|
||
layout: ConfigLayout,
|
||
) -> bool:
|
||
source = source.expanduser().resolve()
|
||
if not source.is_file():
|
||
return False
|
||
profile = layout.profile(profile_id)
|
||
validate_profile(profile)
|
||
values = parse_environment(source)
|
||
secrets = parse_environment(profile / "secrets" / "runtime.env")
|
||
overrides = parse_environment(profile / "overrides" / "runtime.env")
|
||
for key, value in values.items():
|
||
if is_sensitive_key(key):
|
||
secrets[key] = value
|
||
else:
|
||
overrides[key] = value
|
||
(profile / "secrets" / "runtime.env").write_text(
|
||
encode_environment(secrets),
|
||
encoding="utf-8",
|
||
)
|
||
(profile / "overrides" / "runtime.env").write_text(
|
||
encode_environment(overrides),
|
||
encoding="utf-8",
|
||
)
|
||
os.chmod(profile / "secrets" / "runtime.env", 0o600)
|
||
os.chmod(profile / "overrides" / "runtime.env", 0o600)
|
||
archive = layout.profile_state(profile_id) / "archives" / "legacy-environment"
|
||
ensure_private_directory(archive)
|
||
destination = archive / "environment"
|
||
shutil.copyfile(source, destination)
|
||
os.chmod(destination, 0o600)
|
||
return True
|
||
|
||
|
||
def migrate_legacy_context(
|
||
source_root: Path,
|
||
profile_id: str,
|
||
layout: ConfigLayout,
|
||
*,
|
||
environment: Path | None = None,
|
||
) -> bool:
|
||
source = (
|
||
source_root.expanduser().resolve()
|
||
/ "packages"
|
||
/ "jwebot-kernel"
|
||
/ "src"
|
||
/ "jwebot_kernel"
|
||
/ "edge_runtime.py"
|
||
)
|
||
if not source.is_file():
|
||
return False
|
||
try:
|
||
syntax = ast.parse(read_text(source))
|
||
except SyntaxError as error:
|
||
raise ConfigError("legacy context source is not valid Python") from error
|
||
assignments: dict[str, object] = {}
|
||
for node in syntax.body:
|
||
if not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||
continue
|
||
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
||
value_node = node.value
|
||
for target in targets:
|
||
if not isinstance(target, ast.Name):
|
||
continue
|
||
name = target.id
|
||
if (
|
||
name in {
|
||
"PERSON_ALIAS_TEMPLATES",
|
||
"PERSON_DISPLAY_IDS",
|
||
"CLOCK_SCAN_TARGETS",
|
||
}
|
||
or name.endswith("_IDENTITY_NOTE")
|
||
):
|
||
try:
|
||
assignments[name] = ast.literal_eval(value_node)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
|
||
templates = assignments.get("PERSON_ALIAS_TEMPLATES", ())
|
||
display_ids = assignments.get("PERSON_DISPLAY_IDS", {})
|
||
clock_targets = assignments.get("CLOCK_SCAN_TARGETS", ())
|
||
if not isinstance(templates, (list, tuple)) or not isinstance(display_ids, dict):
|
||
return False
|
||
steam_assignments: dict[str, object] = {}
|
||
steam_source = (
|
||
source_root.expanduser().resolve()
|
||
/ "packages"
|
||
/ "jwebot-official-extensions"
|
||
/ "src"
|
||
/ "jwebot_official_extensions"
|
||
/ "steam.py"
|
||
)
|
||
if steam_source.is_file():
|
||
try:
|
||
steam_syntax = ast.parse(read_text(steam_source))
|
||
except SyntaxError as error:
|
||
raise ConfigError("legacy Steam source is not valid Python") from error
|
||
for node in steam_syntax.body:
|
||
if not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||
continue
|
||
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
||
for target in targets:
|
||
if (
|
||
not isinstance(target, ast.Name)
|
||
or target.id
|
||
not in {"STEAM_GROUP_ROSTER", "STEAM_ALIAS_LABELS"}
|
||
):
|
||
continue
|
||
try:
|
||
steam_assignments[target.id] = ast.literal_eval(node.value)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
|
||
roster_by_label: dict[str, str] = {}
|
||
legacy_roster = steam_assignments.get("STEAM_GROUP_ROSTER", ())
|
||
if isinstance(legacy_roster, (list, tuple)):
|
||
for raw in legacy_roster:
|
||
if not isinstance(raw, (list, tuple)) or len(raw) != 2:
|
||
continue
|
||
label = str(raw[0]).strip()
|
||
steam_id64 = str(raw[1]).strip()
|
||
if label and re.fullmatch(r"\d{17}", steam_id64):
|
||
roster_by_label[label] = steam_id64
|
||
|
||
owner_id64 = ""
|
||
if environment is not None and environment.expanduser().is_file():
|
||
owner_id64 = parse_environment(
|
||
environment.expanduser().resolve()
|
||
).get("JWEBOT_EDGE_STEAM_ID64", "").strip()
|
||
display_values = {
|
||
str(value).strip()
|
||
for value in display_ids.values()
|
||
if str(value).strip()
|
||
}
|
||
missing_labels = display_values - set(roster_by_label)
|
||
if re.fullmatch(r"\d{17}", owner_id64) and len(missing_labels) == 1:
|
||
roster_by_label[missing_labels.pop()] = owner_id64
|
||
|
||
aliases_by_label: dict[str, list[str]] = {}
|
||
legacy_aliases = steam_assignments.get("STEAM_ALIAS_LABELS", {})
|
||
if isinstance(legacy_aliases, dict):
|
||
for raw_alias, raw_label in legacy_aliases.items():
|
||
alias = str(raw_alias).strip()
|
||
label = str(raw_label).strip()
|
||
if alias and label in roster_by_label:
|
||
aliases_by_label.setdefault(label, []).append(alias)
|
||
|
||
people = []
|
||
for raw in templates:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
canonical_name = str(raw.get("canonical_name", "")).strip()
|
||
if not canonical_name:
|
||
continue
|
||
display_id = str(display_ids.get(canonical_name, "")).strip()
|
||
person = {
|
||
"canonical_name": canonical_name,
|
||
"aliases": [
|
||
str(value)
|
||
for value in raw.get("aliases", ())
|
||
if str(value).strip()
|
||
],
|
||
"prefixes": [
|
||
str(value)
|
||
for value in raw.get("prefixes", ())
|
||
if str(value).strip()
|
||
],
|
||
"identity_fact": str(raw.get("identity_fact", "")).strip(),
|
||
"display_id": display_id,
|
||
}
|
||
steam_id64 = roster_by_label.get(display_id, "")
|
||
if steam_id64:
|
||
person["steam_id64"] = steam_id64
|
||
person["steam_aliases"] = aliases_by_label.get(display_id, [])
|
||
people.append(person)
|
||
targets = []
|
||
if isinstance(clock_targets, (list, tuple)):
|
||
for raw in clock_targets:
|
||
if (
|
||
isinstance(raw, (list, tuple))
|
||
and len(raw) == 2
|
||
and all(str(value).strip() for value in raw)
|
||
):
|
||
targets.append(
|
||
{
|
||
"canonical_name": str(raw[0]).strip(),
|
||
"display_id": str(raw[1]).strip(),
|
||
}
|
||
)
|
||
profile = layout.profile(profile_id)
|
||
validate_profile(profile)
|
||
context = {
|
||
"people": people,
|
||
"clock_scan_targets": targets,
|
||
}
|
||
validate_private_context(context)
|
||
(profile / "context" / "people.json").write_text(
|
||
json.dumps(context, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
os.chmod(profile / "context" / "people.json", 0o600)
|
||
|
||
identity_note = next(
|
||
(
|
||
str(value).strip()
|
||
for key, value in assignments.items()
|
||
if key.endswith("_IDENTITY_NOTE") and str(value).strip()
|
||
),
|
||
"",
|
||
)
|
||
if identity_note:
|
||
mention = identity_note.split(maxsplit=1)[0].strip("：:，,。.")
|
||
bot = table(load_toml(profile / "components" / "bot.toml"), "bot")
|
||
bot["mention"] = mention or str(bot.get("mention", "JWeBot"))
|
||
bot["identity_note"] = identity_note
|
||
lines = ["[bot]"]
|
||
for key in (
|
||
"mention",
|
||
"identity_note",
|
||
"cloud_agent",
|
||
"memory_remote_path",
|
||
):
|
||
lines.append(
|
||
f"{key} = "
|
||
+ json.dumps(str(bot.get(key, "")), ensure_ascii=False)
|
||
)
|
||
(profile / "components" / "bot.toml").write_text(
|
||
"\n".join(lines) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
os.chmod(profile / "components" / "bot.toml", 0o600)
|
||
return bool(people or targets or identity_note)
|
||
|
||
|
||
def load_active_environment(
|
||
layout: ConfigLayout | None = None,
|
||
) -> dict[str, str]:
|
||
active_layout = layout or ConfigLayout.from_environment()
|
||
profile = active_layout.active_profile()
|
||
if profile is None:
|
||
return {}
|
||
generated = profile / "generated" / "runtime.env"
|
||
return parse_environment(generated) if generated.is_file() else {}
|
||
|
||
|
||
def load_private_context(
|
||
layout: ConfigLayout | None = None,
|
||
) -> dict[str, Any]:
|
||
configured = os.environ.get("JWEBOT_PRIVATE_CONTEXT_FILE", "").strip()
|
||
if configured:
|
||
path = Path(configured).expanduser()
|
||
else:
|
||
active_layout = layout or ConfigLayout.from_environment()
|
||
profile = active_layout.active_profile()
|
||
if profile is None:
|
||
return {"bot": {"mention": "JWeBot", "identity_note": ""}, "people": []}
|
||
path = profile / "generated" / "private-context.json"
|
||
if not path.is_file():
|
||
return {"bot": {"mention": "JWeBot", "identity_note": ""}, "people": []}
|
||
value = load_json(path)
|
||
validate_private_context(value)
|
||
return value
|