[EDIT] FILE: jail_config.py
# -*- coding: utf-8 -*- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import typing from dataclasses import dataclass, field if typing.TYPE_CHECKING: from .mount_config import IsolatedRootConfig from .mount_types import MountEntry, MountType # F-13 (CLOS-5951) DiD: characters that would break out of a jail.c # config section header or a mount-entry line. `[` / `]` delimit # section names; `#` starts a comment; CR / LF end a config line; NUL # is rejected on general principle. Reject before rendering so a # tenant-controlled docroot cannot inject extra mount directives into # the root-consumed configuration. _INVALID_JAIL_CONFIG_CHARS = frozenset("\r\n\0[]#") def _reject_jail_config_metachars(value: str) -> None: if not isinstance(value, str) or any(c in _INVALID_JAIL_CONFIG_CHARS for c in value): raise ValueError("Invalid path in jail configuration") @dataclass class MountConfig: """ Builds mount configuration for jail.c implementation. Accumulates MountEntry objects, then renders everything at the end. """ uid: int gid: int _mounts: list[MountEntry] = field(default_factory=list, repr=False) @property def _default_opts(self) -> tuple[str, ...]: return f"uid={self.uid}", f"gid={self.gid}", "mode=0750" @property def mounts(self) -> tuple[MountEntry, ...]: """Read-only access to accumulated mount entries.""" return tuple(self._mounts) def add(self, type_: MountType, source: str, target: str = "", options: tuple[str] = tuple()): """Add a single mount entry.""" self._mounts.append( MountEntry( type=type_, source=source, target=target, options=options, ) ) def comment(self, text: str): """Add a comment line for organization.""" self._mounts.append(MountEntry(MountType.COMMENT, text)) def add_overlay(self, overlay: "IsolatedRootConfig"): """ Add mount operations for a directory overlay. Args: overlay: The overlay configuration """ if not overlay.persistent: # Create temporary in-memory storage self._mounts.append( MountEntry( MountType.BIND, "tmpfs", overlay.root_path, ("mkdir",) + self._default_opts ) ) else: # Persistent real directory storage (bind to self with mkdir) self._mounts.append( MountEntry( MountType.BIND, overlay.root_path, overlay.root_path, ("mkdir",) + self._default_opts, ) ) def close_overlay(self, overlay: "IsolatedRootConfig"): """Add the final mount that closes an overlay.""" self._mounts.extend(overlay.mounts) self._mounts.append( MountEntry(MountType.BIND, overlay.root_path, overlay.target, ("recursive",)) ) def render(self, docroot: str) -> str: """Render complete configuration to jail.c syntax.""" # F-13 (CLOS-5951) DiD: refuse to render if the docroot or any # mount-entry path carries characters that would let a caller # break out of the section header / entry line and inject # additional root-consumed mount directives. _reject_jail_config_metachars(docroot) for m in self._mounts: _reject_jail_config_metachars(m.source) if m.target: _reject_jail_config_metachars(m.target) lines = [f"[{docroot}]"] lines.extend(m.render() for m in self._mounts) return "\n".join(lines)
SAVE
CANCEL