Run terminals and GUI apps as the authenticated user, not root

Atlus runs as root (systemd) but user-facing processes must run under the
authenticated user's identity. Added privilege-dropping via preexec_fn
(os.setgid + os.initgroups + os.setuid) to both terminal PTY spawning
and GUI app launching. System admin operations (services, packages,
network, updates) intentionally remain root.

Autostart apps now support a configurable default_user; without one set,
autostart defers until the first user logs in and runs as that user.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
roberts 2026-03-15 00:35:52 -05:00
parent e44ee2fe64
commit 6a0c8757f8
6 changed files with 95 additions and 15 deletions

View file

@ -64,6 +64,7 @@ _DEFAULT_CONFIG: dict = {
],
"panel_services": [], # systemd unit names to show in panel
"gui_apps": [], # GUI apps: [{id, name, command, icon, args, target_fps}]
"default_user": None, # OS user for autostart apps (None = defer until first login)
"asi_bridge": {
"cifs_share": "//192.168.10.120/share",
"mount_point": "/mnt/asiair",

View file

@ -6,7 +6,6 @@ Each GUI app window is captured independently and streamed as JPEG frames.
"""
import asyncio
import getpass
import logging
import os
import re
@ -17,6 +16,8 @@ import uuid
from dataclasses import dataclass, field
from typing import Optional
from backend.privdrop import get_user_info, make_preexec_fn
log = logging.getLogger("atlus.display")
# ---------------------------------------------------------------------------
@ -477,10 +478,18 @@ class DisplayManager:
display_num = await self.get_or_create_display(username)
# Resolve the user's home directory (not root's)
try:
_uid, _gid, user_home, _shell = get_user_info(username)
except KeyError:
user_home = os.path.expanduser("~")
env = {
**os.environ,
"DISPLAY": f":{display_num}",
"HOME": os.path.expanduser("~"),
"HOME": user_home,
"USER": username,
"LOGNAME": username,
}
# Ensure PATH
path = env.get("PATH", "")
@ -492,11 +501,20 @@ class DisplayManager:
cmd = [cmd_bin] + (args or [])
log.info("Launching GUI app: %s (display :%d, user %s)", " ".join(cmd), display_num, username)
# Drop privileges to the authenticated user
try:
preexec = make_preexec_fn(username)
except KeyError:
preexec = None
log.warning("User %s not found on system — app will run as service user", username)
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=env,
cwd=user_home,
preexec_fn=preexec,
)
app_id = str(uuid.uuid4())[:8]
@ -542,21 +560,29 @@ class DisplayManager:
app.kill()
return [a.to_dict() for a in user_apps.values()]
async def autostart_apps(self, gui_apps: list[dict]):
"""Launch all apps with autostart=True. Called on service startup."""
async def autostart_apps(self, gui_apps: list[dict], default_user: str | None = None):
"""Launch all apps with autostart=True.
If *default_user* is set, apps start immediately as that user.
Otherwise autostart is deferred until the first user logs in.
"""
if not HAS_DISPLAY_DEPS:
log.info("Display deps not available, skipping autostart")
return
username = getpass.getuser()
autostart_list = [a for a in gui_apps if a.get("autostart") and a.get("command")]
if not autostart_list:
return
if not default_user:
log.info("No default_user configured — deferring autostart until first login")
self._pending_autostart = autostart_list
return
username = default_user
started = 0
for app_cfg in gui_apps:
if not app_cfg.get("autostart"):
continue
command = app_cfg.get("command", "")
if not command:
continue
# Skip if already running
for app_cfg in autostart_list:
command = app_cfg["command"]
if self.get_app_by_command(username, command):
log.debug("Autostart skip (already running): %s", command)
continue
@ -569,11 +595,20 @@ class DisplayManager:
target_fps=app_cfg.get("target_fps", 10),
)
started += 1
log.info("Autostarted: %s", command)
log.info("Autostarted: %s (as %s)", command, username)
except Exception as e:
log.warning("Failed to autostart %s: %s", command, e)
if started:
log.info("Autostarted %d application(s)", started)
log.info("Autostarted %d application(s) as %s", started, username)
async def trigger_deferred_autostart(self, username: str):
"""Launch deferred autostart apps on first user login."""
pending = getattr(self, "_pending_autostart", None)
if not pending:
return
self._pending_autostart = None
log.info("Triggering deferred autostart for %s", username)
await self.autostart_apps(pending, default_user=username)
async def shutdown_all(self):
"""Kill all apps and Xvfb displays."""

View file

@ -35,9 +35,10 @@ async def lifespan(app: FastAPI):
# Autostart configured GUI apps (always-on desktop session)
cfg = load_config()
gui_apps = cfg.get("gui_apps", [])
default_user = cfg.get("default_user")
if gui_apps:
try:
await display_manager.autostart_apps(gui_apps)
await display_manager.autostart_apps(gui_apps, default_user=default_user)
except Exception as e:
log.warning("Autostart failed: %s", e)

37
backend/privdrop.py Normal file
View file

@ -0,0 +1,37 @@
"""Privilege-dropping helpers for spawning processes as non-root users.
Atlus runs as root (systemd service) but user-facing processes terminal
PTYs and GUI applications must run as the authenticated user. This
module provides a ``preexec_fn`` factory that drops privileges after
``fork()`` but before ``exec()``.
"""
import os
import pwd
from typing import Callable
def get_user_info(username: str) -> tuple[int, int, str, str]:
"""Return (uid, gid, home_dir, shell) for *username*.
Raises ``KeyError`` if the user does not exist on the system.
"""
pw = pwd.getpwnam(username)
return pw.pw_uid, pw.pw_gid, pw.pw_dir, pw.pw_shell
def make_preexec_fn(username: str) -> Callable[[], None]:
"""Return a closure suitable for ``preexec_fn`` that drops to *username*.
The user lookup (``pwd.getpwnam``) happens eagerly in the **parent**
process so that errors surface immediately. The ``os.set*`` calls
execute in the **child** process after ``fork()``.
"""
uid, gid, home, _shell = get_user_info(username)
def _drop_privileges() -> None:
os.setgid(gid)
os.initgroups(username, gid) # supplementary groups (sudo, video …)
os.setuid(uid) # must be last — can't setgid after this
return _drop_privileges

View file

@ -6,6 +6,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, Query
from pydantic import BaseModel
from backend.auth import get_current_user, decode_token
from backend.display import display_manager
from backend.sessions import manager
router = APIRouter(prefix="/api/session", tags=["session"])
@ -35,6 +36,8 @@ class TerminalCreate(BaseModel):
async def get_session(user: str = Depends(get_current_user)):
"""Get or create the user's desktop session."""
session = manager.get_or_create(user)
# Trigger deferred autostart apps on first login
await display_manager.trigger_deferred_autostart(user)
return session.to_dict()

View file

@ -13,6 +13,8 @@ import signal
import time
import uuid
from collections import deque
from backend.privdrop import make_preexec_fn
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
@ -245,6 +247,7 @@ class SessionManager:
dimensions=(rows, cols),
env=env,
cwd=home,
preexec_fn=make_preexec_fn(username),
)
except Exception:
log.exception("Failed to spawn PTY for %s", username)