"""Per-stack, file-based compose secrets & configs. Secret/config payloads are stored as files INSIDE the stack's own directory (``/.secrets/`` and ``.configs/``) and referenced from the compose file with a RELATIVE path (``file: ./.secrets/``). Because compose resolves ``file:`` relative to the compose file — which lives in the stack dir (a host bind-mount) — the Docker daemon reads the file correctly with no ``HOST_ROOT_PREFIX`` dependency, exactly as if the user had dropped a secret file next to their ``compose.yaml`` by hand. Content is never returned by the listing API; only metadata (name, size, mtime). """ from __future__ import annotations import os import re from typing import Optional from services import compose_edit_service, compose_service KIND_DIR = {"secret": ".secrets", "config": ".configs"} _NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$") MAX_SECRET_BYTES = 1 * 1024 * 1024 # 1 MiB class SecretError(Exception): """Invalid secret request (bad kind/name, too large).""" def _check_kind(kind: str) -> str: if kind not in KIND_DIR: raise SecretError(f"unknown kind '{kind}'") return kind def _check_name(name: str) -> str: # A single path component, no traversal, no leading dot (keeps it out of the # way of .secrets/.configs themselves and hidden-file surprises). if not name or name.startswith(".") or not _NAME_RE.match(name) or os.path.sep in name: raise SecretError("name must match [A-Za-z0-9._-] and not start with a dot") return name def kind_dir(stack_id: str, kind: str, override: Optional[str] = None) -> str: _check_kind(kind) return os.path.join(compose_service.stack_dir(stack_id, override), KIND_DIR[kind]) def _path(stack_id: str, kind: str, name: str, override: Optional[str] = None) -> str: return os.path.join(kind_dir(stack_id, kind, override), _check_name(name)) def rel_path(kind: str, name: str) -> str: """Path to put in the compose ``file:`` field (relative to the compose file).""" return f"./{KIND_DIR[_check_kind(kind)]}/{_check_name(name)}" def list_secrets(stack_id: str, kind: str, override: Optional[str] = None) -> list[dict]: directory = kind_dir(stack_id, kind, override) if not os.path.isdir(directory): return [] out: list[dict] = [] for name in sorted(os.listdir(directory)): full = os.path.join(directory, name) if os.path.isfile(full): st = os.stat(full) out.append({"name": name, "kind": kind, "size": st.st_size, "modified": st.st_mtime}) return out def list_all(stack_id: str, override: Optional[str] = None) -> list[dict]: items: list[dict] = [] for kind in KIND_DIR: items.extend(list_secrets(stack_id, kind, override)) return items def write_secret(stack_id: str, kind: str, name: str, content: str, override: Optional[str] = None) -> dict: _check_kind(kind) _check_name(name) data = content.encode("utf-8") if len(data) > MAX_SECRET_BYTES: raise SecretError("content exceeds 1 MiB limit") directory = kind_dir(stack_id, kind, override) os.makedirs(directory, exist_ok=True) os.chmod(directory, 0o700) path = _path(stack_id, kind, name, override) with open(path, "wb") as fh: fh.write(data) os.chmod(path, 0o600) return {"name": name, "kind": kind, "size": len(data)} def delete_secret(stack_id: str, kind: str, name: str, override: Optional[str] = None) -> None: path = _path(stack_id, kind, name, override) if os.path.isfile(path): os.remove(path) def exists(stack_id: str, kind: str, name: str, override: Optional[str] = None) -> bool: return os.path.isfile(_path(stack_id, kind, name, override)) def attach(stack_id: str, kind: str, name: str, service: str, target: Optional[str] = None, override: Optional[str] = None) -> str: """Reference a stored secret/config from ``service`` in the stack's compose file (relative ``file:`` path) and persist it. Returns the new YAML.""" _check_kind(kind) _check_name(name) yaml_str = compose_service.read_compose(stack_id, override) file_path = rel_path(kind, name) if kind == "config": new_yaml = compose_edit_service.add_config(yaml_str, service, name, file_path, target or f"/{name}") else: new_yaml = compose_edit_service.add_secret(yaml_str, service, name, file_path) compose_service.write_compose(stack_id, new_yaml, override) return new_yaml def detach(stack_id: str, kind: str, name: str, service: str, override: Optional[str] = None) -> str: _check_kind(kind) yaml_str = compose_service.read_compose(stack_id, override) if kind == "config": new_yaml = compose_edit_service.remove_config(yaml_str, service, name) else: new_yaml = compose_edit_service.remove_secret(yaml_str, service, name) compose_service.write_compose(stack_id, new_yaml, override) return new_yaml