# retoor <retoor@molodetz.nl>
from __future__ import annotations
import json
import logging
import shlex
from dataclasses import asdict, dataclass
from pathlib import Path
from devplacepy import config
from devplacepy.database import get_int_setting, get_setting, get_table
from devplacepy.services.containers import store
from devplacepy.services.containers.backend.base import (
WORKSPACE_MOUNT,
WORKSPACE_STATE_MOUNT,
)
from . import quota
logger = logging.getLogger(__name__)
PREFS_TABLE = "workspace_editor_prefs"
APP_NAME = "DevPlace"
WELCOME_TEXT = "Sign in to your DevPlace workspace"
EDITOR_DEFAULT_PORT = 8443
PROFILE_FILE = "devplace-editor.json"
MANAGED_FILE = ".devplace-managed.json"
INHERIT_TEXT = ""
INHERIT_INT = 0
INHERIT_ZOOM = -99
INHERIT_FLAG = -1
THEME_CHOICES = ("devplace-dark", "devplace-light", "system")
LAYOUT_CHOICES = ("standard", "terminal-focus", "zen")
PANEL_CHOICES = ("short", "normal", "tall", "maximized")
WINDOW_CHOICES = ("tab", "window", "fullscreen")
AGENT_CHOICES = ("dpc", "none")
CHOICES = {
"theme": THEME_CHOICES,
"layout": LAYOUT_CHOICES,
"panel_preset": PANEL_CHOICES,
"window_mode": WINDOW_CHOICES,
"boot_agent": AGENT_CHOICES,
}
SENTINELS = {
"zoom_level": INHERIT_ZOOM,
"boot_shell": INHERIT_FLAG,
}
THEME_LABELS = {
"devplace-dark": "DevPlace Dark",
"devplace-light": "DevPlace Light",
}
LAYOUTS = {
"standard": {
"workbench.activityBar.location": "default",
"workbench.sideBar.location": "left",
"workbench.panel.defaultLocation": "bottom",
"editor.minimap.enabled": True,
"breadcrumbs.enabled": True,
},
"terminal-focus": {
"workbench.activityBar.location": "top",
"workbench.sideBar.location": "left",
"workbench.panel.defaultLocation": "bottom",
"editor.minimap.enabled": False,
"breadcrumbs.enabled": True,
},
"zen": {
"workbench.activityBar.location": "hidden",
"workbench.sideBar.location": "left",
"workbench.panel.defaultLocation": "bottom",
"editor.minimap.enabled": False,
"breadcrumbs.enabled": False,
},
}
FOREIGN_AI_SETTINGS = {
"chat.disableAIFeatures": True,
"chat.commandCenter.enabled": False,
"workbench.secondarySideBar.defaultVisibility": "hidden",
}
TRUST_SETTINGS = {
"security.workspace.trust.enabled": False,
"security.workspace.trust.startupPrompt": "never",
"security.workspace.trust.banner": "never",
"security.workspace.trust.emptyWindow": True,
"security.workspace.trust.untrustedFiles": "open",
"task.allowAutomaticTasks": "on",
}
BOUNDS = {
"font_size": (8, 48),
"terminal_font_size": (8, 48),
"zoom_level": (-5, 5),
"window_width": (640, 7680),
"window_height": (480, 4320),
}
SETTING_KEYS = {
"trust_all": "workspace_editor_trust_all",
"theme": "workspace_editor_theme",
"font_size": "workspace_editor_font_size",
"terminal_font_size": "workspace_editor_terminal_font_size",
"zoom_level": "workspace_editor_zoom_level",
"layout": "workspace_editor_layout",
"panel_preset": "workspace_editor_panel_preset",
"boot_agent": "workspace_editor_boot_agent",
"boot_shell": "workspace_editor_boot_shell",
"window_mode": "workspace_editor_window_mode",
"window_width": "workspace_editor_window_width",
"window_height": "workspace_editor_window_height",
}
DEFAULTS = {
"trust_all": True,
"theme": "devplace-dark",
"font_size": 14,
"terminal_font_size": 13,
"zoom_level": 0,
"layout": "standard",
"panel_preset": "tall",
"boot_agent": "dpc",
"boot_shell": True,
"window_mode": "tab",
"window_width": 1600,
"window_height": 1000,
}
PREF_COLUMNS = (
"font_size",
"terminal_font_size",
"zoom_level",
"theme",
"layout",
"panel_preset",
"window_mode",
"window_width",
"window_height",
"boot_agent",
"boot_shell",
)
OPTIONAL_FLAGS = (
"--app-name",
"--welcome-text",
"--disable-getting-started-override",
"--disable-workspace-trust",
)
SOURCE_SITE = "site"
SOURCE_USER = "user"
@dataclass(frozen=True)
class EditorProfile:
trust_all: bool
theme: str
font_size: int
terminal_font_size: int
zoom_level: int
layout: str
panel_preset: str
boot_agent: str
boot_shell: bool
window_mode: str
window_width: int
window_height: int
cpu_millicores: int
memory_mb: int
disk_quota_mb: int
def as_dict(self) -> dict:
return asdict(self)
def cpu_cores(self) -> float:
return round(self.cpu_millicores / 1000, 3)
def cpu_limit(self) -> str:
return quota.format_cpu(self.cpu_millicores)
def mem_limit(self) -> str:
return quota.format_memory(self.memory_mb)
def _clamp(key: str, value: int) -> int:
bounds = BOUNDS.get(key)
if not bounds:
return value
low, high = bounds
return max(low, min(high, value))
def _choice(value: str, choices: tuple[str, ...], fallback: str) -> str:
cleaned = (value or "").strip().lower()
return cleaned if cleaned in choices else fallback
def prefs_for(owner_uid: str) -> dict | None:
if not owner_uid:
return None
return get_table(PREFS_TABLE).find_one(
owner_kind="user", owner_id=owner_uid, deleted_at=None
)
def _inherits(key: str, row: dict | None) -> bool:
if not row:
return True
value = row.get(key)
if value is None:
return True
if key in SENTINELS:
return int(value) == SENTINELS[key]
if isinstance(DEFAULTS[key], str):
return not str(value).strip()
return int(value) == INHERIT_INT
def _site_value(key: str):
default = DEFAULTS[key]
setting = SETTING_KEYS[key]
if isinstance(default, bool):
return get_setting(setting, "1" if default else "0") == "1"
if isinstance(default, int):
return get_int_setting(setting, default)
return get_setting(setting, default)
def _normalize(key: str, value):
default = DEFAULTS[key]
if key in CHOICES:
return _choice(value, CHOICES[key], default)
if isinstance(default, bool):
if isinstance(value, bool):
return value
try:
return bool(int(value))
except (TypeError, ValueError):
return bool(value)
if isinstance(default, int):
try:
return _clamp(key, int(value))
except (TypeError, ValueError):
return default
return str(value)
def source_map(owner_uid: str = "") -> dict[str, str]:
row = prefs_for(owner_uid)
sources = {key: SOURCE_SITE for key in SETTING_KEYS}
for key in PREF_COLUMNS:
if not _inherits(key, row):
sources[key] = SOURCE_USER
return sources
def resolve(owner_uid: str = "", instance: dict | None = None) -> EditorProfile:
row = prefs_for(owner_uid)
values = {}
for key in SETTING_KEYS:
value = _site_value(key)
if key in PREF_COLUMNS and not _inherits(key, row):
value = row.get(key)
values[key] = _normalize(key, value)
limits = quota.resolve(owner_uid, instance)
return EditorProfile(
cpu_millicores=limits.cpu_millicores,
memory_mb=limits.memory_mb,
disk_quota_mb=limits.disk_quota_mb,
**values,
)
def settings_for(profile: EditorProfile) -> dict:
settings = {
"editor.fontSize": profile.font_size,
"terminal.integrated.fontSize": profile.terminal_font_size,
"window.zoomLevel": profile.zoom_level,
"telemetry.telemetryLevel": "off",
"update.mode": "none",
"workbench.tips.enabled": False,
"workbench.startupEditor": "none",
"extensions.autoCheckUpdates": False,
**FOREIGN_AI_SETTINGS,
"terminal.integrated.defaultProfile.linux": "bash",
"terminal.integrated.profiles.linux": {
"bash": {"path": "/bin/bash", "args": ["-l"], "icon": "terminal-bash"},
"DevPlace Code": {"path": "/usr/bin/dpc", "icon": "rocket"},
},
}
settings.update(LAYOUTS[profile.layout])
if profile.theme in THEME_LABELS:
settings["workbench.colorTheme"] = THEME_LABELS[profile.theme]
if profile.trust_all:
settings.update(TRUST_SETTINGS)
return settings
def merge_managed(current: dict, managed: dict, desired: dict) -> tuple[dict, dict]:
merged = dict(current)
for key, value in desired.items():
if key not in merged or merged[key] == managed.get(key):
merged[key] = value
return merged, dict(desired)
def _read_json(path: Path) -> dict:
try:
data = json.loads(path.read_text())
except (OSError, ValueError):
return {}
return data if isinstance(data, dict) else {}
def state_dir(instance: dict) -> Path:
return config.WORKSPACE_STATE_DIR / instance["uid"]
def profile_payload(instance: dict, profile: EditorProfile) -> dict:
return {
"app_name": APP_NAME,
"workspace_uid": instance.get("uid", ""),
"boot_marker": instance.get("boot_marker", ""),
"editor": profile.as_dict(),
}
def seed_state(instance: dict, profile: EditorProfile) -> bool:
root = state_dir(instance)
user_dir = root / "data" / "User"
try:
user_dir.mkdir(parents=True, exist_ok=True)
settings_path = user_dir / "settings.json"
managed_path = user_dir / MANAGED_FILE
settings, managed = merge_managed(
_read_json(settings_path),
_read_json(managed_path),
settings_for(profile),
)
settings_path.write_text(json.dumps(settings, indent=2, sort_keys=True))
managed_path.write_text(json.dumps(managed, indent=2, sort_keys=True))
(root / PROFILE_FILE).write_text(
json.dumps(profile_payload(instance, profile), indent=2, sort_keys=True)
)
except OSError as error:
logger.warning("workspace editor seed failed for %s: %s", instance.get("uid"), error)
return False
return True
def argv(instance: dict, profile: EditorProfile) -> list[str]:
port = int(instance.get("editor_port") or EDITOR_DEFAULT_PORT)
command = [
"code-server",
"--bind-addr",
f"0.0.0.0:{port}",
"--auth",
"password",
"--app-name",
APP_NAME,
"--welcome-text",
WELCOME_TEXT,
"--disable-telemetry",
"--disable-update-check",
"--disable-getting-started-override",
]
if profile.trust_all:
command.append("--disable-workspace-trust")
command += [
"--user-data-dir",
f"{WORKSPACE_STATE_MOUNT}/data",
"--extensions-dir",
f"{WORKSPACE_STATE_MOUNT}/extensions",
WORKSPACE_MOUNT,
]
return command
ENV_EXPORT_FILE = "/etc/profile.d/devplace-env.sh"
_ENV_EXPORT_SCRIPT = (
"import os, pathlib, shlex\n"
f"path = pathlib.Path({ENV_EXPORT_FILE!r})\n"
"lines = ['export ' + k + '=' + shlex.quote(v) for k, v in sorted(os.environ.items()) if k.startswith('DEVPLACE_')]\n"
"path.write_text('\\n'.join(lines) + '\\n' if lines else '')\n"
)
def wrap_with_env_export(command: list[str]) -> list[str]:
export_step = f"umask 022; python3 -c {shlex.quote(_ENV_EXPORT_SCRIPT)} 2>/dev/null || true"
script = f"{export_step}; exec {shlex.join(command)}"
return ["/bin/sh", "-c", script]
def env_for(profile: EditorProfile) -> dict:
return {
"DEVPLACE_EDITOR_APP_NAME": APP_NAME,
"DEVPLACE_EDITOR_PROFILE": f"{WORKSPACE_STATE_MOUNT}/{PROFILE_FILE}",
"DEVPLACE_EDITOR_THEME": profile.theme,
"DEVPLACE_EDITOR_FONT_SIZE": str(profile.font_size),
"DEVPLACE_EDITOR_TERMINAL_FONT_SIZE": str(profile.terminal_font_size),
"DEVPLACE_EDITOR_ZOOM_LEVEL": str(profile.zoom_level),
"DEVPLACE_EDITOR_LAYOUT": profile.layout,
"DEVPLACE_EDITOR_PANEL_PRESET": profile.panel_preset,
"DEVPLACE_EDITOR_BOOT_AGENT": profile.boot_agent,
"DEVPLACE_EDITOR_BOOT_SHELL": "1" if profile.boot_shell else "0",
"DEVPLACE_EDITOR_TRUST_ALL": "1" if profile.trust_all else "0",
}
def booted_profile(instance: dict) -> dict:
return _read_json(state_dir(instance) / PROFILE_FILE).get("editor") or {}
def restart_required(instance: dict, profile: EditorProfile) -> bool:
booted = booted_profile(instance)
if not booted:
return False
return booted != profile.as_dict()
def save_prefs(owner_uid: str, payload: dict) -> dict:
from devplacepy.utils import generate_uid
table = get_table(PREFS_TABLE)
row = prefs_for(owner_uid)
values = {key: payload[key] for key in PREF_COLUMNS if key in payload}
stamp = store.now()
if row:
table.update({"uid": row["uid"], "updated_at": stamp, **values}, ["uid"])
return table.find_one(uid=row["uid"])
uid = generate_uid()
table.insert(
{
"uid": uid,
"owner_kind": "user",
"owner_id": owner_uid,
"created_at": stamp,
"updated_at": stamp,
"deleted_at": None,
"deleted_by": None,
**{key: _blank(key) for key in PREF_COLUMNS},
**values,
}
)
return table.find_one(uid=uid)
def reset_prefs(owner_uid: str, actor_uid: str) -> bool:
row = prefs_for(owner_uid)
if not row:
return False
get_table(PREFS_TABLE).update(
{"uid": row["uid"], "deleted_at": store.now(), "deleted_by": actor_uid}, ["uid"]
)
return True
def _blank(key: str):
if key in SENTINELS:
return SENTINELS[key]
if isinstance(DEFAULTS[key], str):
return INHERIT_TEXT
return INHERIT_INT
def view(owner_uid: str = "", instance: dict | None = None) -> dict:
profile = resolve(owner_uid, instance)
payload = profile.as_dict()
payload["cpu_cores"] = profile.cpu_cores()
payload["sources"] = source_map(owner_uid)
return payload
Reference in New Issue View Git Blame Copy Permalink