23s2s

✏️ Editing: node_modules_storage.py

# coding:utf-8

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

"""Configuration for the server-wide shared Node modules storage.

The account-readable file contains non-sensitive delivery policy. Account
names excluded by the administrator live in a separate root-only sidecar;
accounts learn only their own current decision from the authenticated daemon.
Unknown JSON keys survive writes so a newer package can add settings without
an older administrator CLI discarding them.
"""

import errno
import fcntl
import json
import math
import os
import pwd
import re
import stat
import tempfile
import time

BUN_BIN = '/opt/cl-bun/bun'
ENGINE_UNAVAILABLE = (
    'Shared node_modules Store requires a usable cl-bun engine. '
    'Ask the administrator to run cl-node-modules-storage enable to restore it, '
    'or cl-node-modules-storage disable to use classic npm.')

# On the CageFS-mounted side on purpose: /etc paths are *copied* into cage
# skeletons and go stale until a refresh, while /opt/cl-bun is bind-mounted
# -- the same filesystem object the daemon's socket already proves is live
# inside every cage. The public switch is therefore current without a refresh;
# account exclusions live in a root-only sidecar read by the daemon.
# /etc/cloudlinux/node-modules-storage.json remains as a symlink for people;
# this constant is the public file.
CONFIG_PATH = '/opt/cl-bun/etc/node-modules-storage.json'
EXCLUSIONS_SUFFIX = '.excluded-users'
DEFAULT_STORAGE_PATH = '/var/lib/cl-node-modules-storage'
MAX_CONFIG_BYTES = 64 * 1024
PROTECTED_HARDLINKS_PATH = '/proc/sys/fs/protected_hardlinks'
CONFIG_MODE = 0o644
EXCLUSIONS_MODE = 0o600
STORAGE_MODE = 0o700
_DANGEROUS_PATHS = frozenset((
    '/', '/boot', '/dev', '/etc', '/home', '/opt', '/proc', '/root', '/run',
    '/sys', os.path.join(os.sep, 'tmp'), '/usr', '/var', '/var/cache', '/var/lib', '/var/log',
    os.path.join(os.sep, 'var', 'tmp')))
LOCK_DIRECTORY = '/run/cloudlinux-node-modules-storage'
LOCK_PATH = os.path.join(LOCK_DIRECTORY, 'storage.lock')
CONFIG_LOCK_PATH = os.path.join(LOCK_DIRECTORY, 'config.lock')
_BUN_TEMPORARY_ENTRY = re.compile(
    r'^\.[0-9a-f]{1,16}-[0-9A-F]+\.[A-Za-z0-9_.-]+$')
_PRUNE_TRASH_ENTRY = re.compile(r'^entry-[0-9a-f]{32}$')
_ASCII_DIGITS = re.compile(r'^[0-9]+$')
PRUNE_TRASH_DIRECTORY = '.prune-trash'
HARDLINK_PROBE_PREFIX = '.cl-node-modules-check-'
PRUNE_BATCH_ENTRIES = 64
PRUNE_LOCK_SLICE_SECONDS = 5
# The daemon retries a busy cache lock with a bounded 25-250 ms backoff.
# Leave a larger scheduling window after each maintenance slice so a waiting
# delivery takes the next turn before prune asks for the lock again.
PRUNE_DELIVERY_YIELD_SECONDS = 0.3

# Administrator-tunable ceilings, all integers. The first four bound delivery
# admission; the queue timeout also bounds direct panel/migration waits for
# the global storage and per-storage cache locks. soft_storage_limit_gb is an
# admission threshold per storage directory (0 = no threshold). A configured
# threshold reads the last complete accounting snapshot under the cache lock;
# usage is measured after delivery for later admission, status and privacy-safe
# summary telemetry.
DEFAULT_LIMITS = {
    'account_deliveries_per_hour': 60,
    'account_deliveries_per_day': 200,
    'concurrent_deliveries': 4,
    'delivery_queue_timeout_seconds': 120,
    'soft_storage_limit_gb': 0,
}
# (min, max) inclusive. Everything is 1-or-more except
# soft_storage_limit_gb, whose 0 is the "disabled" sentinel.
LIMIT_BOUNDS = {
    'account_deliveries_per_hour': (1, 100000),
    'account_deliveries_per_day': (1, 1000000),
    'concurrent_deliveries': (1, 1024),
    'delivery_queue_timeout_seconds': (1, 86400),
    'soft_storage_limit_gb': (0, 1048576),
}
DEFAULT_REGISTRIES = ['https://registry.npmjs.org/']
ROOT_UID = 0
MAX_TREE_WALK_DEPTH = 256
# Removal is one frame per level. The syscalls are linear in the tree but the
# frames are not free -- the delivery-side twin of this walk measured 80 s of
# root CPU at depth 50,000 -- and prune removes while holding both the shared
# population lock and the maintenance lock.
MAX_REMOVAL_DEPTH = 8192
# Accounting walks the whole server-wide cache, not one account's tree: a
# store of ~5000 canonical entries of ~200 files each is ordinary, far past
# anything an account-tree bound would allow. Measurement therefore gets its
# own ceilings, and reaching one is a partial number rather than a refusal --
# see `_measure_tree_usage_fd`.
MEASURE_MAX_ENTRIES = 20000000
MEASURE_MAX_SECONDS = 900

# The size a storage directory is holding, recorded beside it so neither the
# panel process nor `status` has to re-walk the tree on every call.
STORAGE_USAGE_FILE = '.cl-storage-usage.json'
SHARED_NODE_MODULES_MARKER = '.cl-shared-node-modules.json'
MAX_SHARED_NODE_MODULES_MARKER_BYTES = 4096
_SHARED_NODE_MODULES_MARKER_KEYS = frozenset((
    'schema', 'engine', 'delivered_at', 'shared_files', 'private_files',
    'quarantined_files', 'shared_bytes', 'shared_inode'))

class StorageError(Exception):
    """A safe-to-display administrator configuration error."""

class CachePopulationBusy(StorageError):
    """A daemon retry may wait outside its scarce global delivery gate."""

def engine_available(path=BUN_BIN):
    """Check the packaged engine without executing it as root."""
    try:
        info = os.stat(path)
    except OSError:
        return False
    return (stat.S_ISREG(info.st_mode) and info.st_uid == 0
            and not stat.S_IMODE(info.st_mode) & 0o022
            and bool(stat.S_IMODE(info.st_mode) & 0o111))

def _unique_object(pairs):
    """json object_pairs_hook rejecting ambiguous duplicate member names."""
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError('duplicate JSON member')
        result[key] = value
    return result

def read_shared_node_modules_marker(node_modules):
    """Return a valid delivered-tree marker, or ``None`` for inactive.

The marker is account-readable input when the selector lists applications.
    It is deliberately a tiny, flat, exact schema: malformed data, a symlink,
    or any read failure is an inactive shared-node-modules state rather than an
    application-listing failure.
    """
    root_fd = None
    try:
        if isinstance(node_modules, int):
            root_fd = os.dup(node_modules)
        else:
            root_fd = os.open(node_modules,
                              os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        return read_shared_node_modules_marker_fd(root_fd)
    except (OSError, TypeError):
        return None
    finally:
        if root_fd is not None:
            try:
                os.close(root_fd)
            except OSError:
                pass

def read_shared_node_modules_marker_fd(root_fd):
    """Read a marker below an already-pinned node_modules directory."""
    marker_fd = None
    try:
        marker_fd = os.open(SHARED_NODE_MODULES_MARKER,
                            os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW,
                            dir_fd=root_fd)
        info = os.fstat(marker_fd)
        if (not stat.S_ISREG(info.st_mode)
                or info.st_size > MAX_SHARED_NODE_MODULES_MARKER_BYTES):
            return None
        chunks = []
        remaining = MAX_SHARED_NODE_MODULES_MARKER_BYTES + 1
        while remaining:
            chunk = os.read(marker_fd, min(4096, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        data = b''.join(chunks)
        if len(data) > MAX_SHARED_NODE_MODULES_MARKER_BYTES:
            return None
        value = json.loads(data.decode('utf-8'), object_pairs_hook=_unique_object)
    except (OSError, UnicodeError, ValueError, RecursionError):
        return None
    finally:
        if marker_fd is not None:
            try:
                os.close(marker_fd)
            except OSError:
                pass
    if not isinstance(value, dict) or set(value) != _SHARED_NODE_MODULES_MARKER_KEYS:
        return None
    if type(value.get('schema')) is not int or value['schema'] != 2:
        return None
    if value.get('engine') != 'bun':
        return None
    delivered_at = value.get('delivered_at')
    if type(delivered_at) is int:
        try:
            timestamp_finite = math.isfinite(float(delivered_at))
        except OverflowError:
            timestamp_finite = False
        if delivered_at < 0 or not timestamp_finite:
            return None
    elif (type(delivered_at) is not float
          or not math.isfinite(delivered_at) or delivered_at < 0):
        return None
    for name in ('shared_files', 'private_files', 'quarantined_files',
                 'shared_bytes'):
        if type(value.get(name)) is not int or value[name] < 0:
            return None
    proof = value['shared_inode']
    if value['shared_files'] == 0 or not _shared_inode_matches(root_fd, proof):
        return None
    return value

def _shared_inode_matches(root_fd, proof):
    """Verify one inode that proves the restored tree is still shared."""
    if (not isinstance(proof, dict)
            or set(proof) != {'path', 'device', 'inode'}
            or not isinstance(proof.get('path'), str)
            or type(proof.get('device')) is not int or proof['device'] < 0
            or type(proof.get('inode')) is not int or proof['inode'] < 0):
        return False
    path = proof['path']
    parts = path.split('/')
    if (not path or os.path.isabs(path)
            or any(part in ('', '.', '..') for part in parts)):
        return False
    current = os.dup(root_fd)
    try:
        for part in parts[:-1]:
            child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                            dir_fd=current)
            os.close(current)
            current = child
        info = os.stat(parts[-1], dir_fd=current, follow_symlinks=False)
        return (stat.S_ISREG(info.st_mode)
                and info.st_dev == proof['device']
                and info.st_ino == proof['inode']
                and info.st_nlink > 1
                and is_sound_shared_inode(info))
    except (OSError, ValueError):
        return False
    finally:
        os.close(current)

def nodejs_selector_supported():
    """Return the panel capability without hiding probe failures.

``False`` is reserved for a successful, authoritative "unsupported"
    answer.  Callers must refuse on ``StorageError`` so a transient panel API
    failure can never select classic npm and detach a live shared tree.
    """
    try:
        from clcommon.cpapi import Feature, is_panel_feature_supported
        supported = is_panel_feature_supported(Feature.NODEJS_SELECTOR)
    except Exception as exc:
        raise StorageError('cannot determine Node.js Selector support') from exc
    if type(supported) is not bool:
        raise StorageError('cannot determine Node.js Selector support')
    return supported

def _open_user_directory_beneath(home, uid, components):
    """Pin a user-owned directory below an account home without symlinks."""
    if (not isinstance(home, str) or not os.path.isabs(home)
            or type(uid) is not int or not isinstance(components, (list, tuple))):
        return None
    if any(not isinstance(component, str) or component in ('', '.', '..')
           or os.path.sep in component
           or (os.path.altsep is not None and os.path.altsep in component)
           for component in components):
        return None
    directory_fd = None
    try:
        directory_fd = os.open(home, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        if os.fstat(directory_fd).st_uid != uid:
            return None
        for component in components:
            child_fd = os.open(component,
                               os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                               dir_fd=directory_fd)
            os.close(directory_fd)
            directory_fd = child_fd
            if os.fstat(directory_fd).st_uid != uid:
                return None
        result = directory_fd
        directory_fd = None
        return result
    except OSError:
        return None
    finally:
        if directory_fd is not None:
            os.close(directory_fd)

def user_directory_device_beneath(home, uid, components):
    """Return a safely-pinned user directory's device, or ``None``."""
    directory_fd = _open_user_directory_beneath(home, uid, components)
    if directory_fd is None:
        return None
    try:
        return os.fstat(directory_fd).st_dev
    except OSError:
        return None
    finally:
        os.close(directory_fd)

def read_shared_node_modules_marker_beneath(home, uid, components):
    """Read a marker below user-owned, component-wise no-follow directories."""
    directory_fd = _open_user_directory_beneath(home, uid, components)
    if directory_fd is None:
        return None
    try:
        return read_shared_node_modules_marker_fd(directory_fd)
    finally:
        os.close(directory_fd)

def is_sound_shared_inode(info):
    """Whether a regular file is still what bun made it and may be shared.

The delivery hardlinks cache inodes into every account, so an inode is
    only safe to share while it is root-owned, writable by nobody but root,
    readable by others and carrying no setuid/setgid/sticky bit. An administrator's ``chown -R`` over a home (or
    Plesk's ``repair fs``) retitles the inode behind every account's link,
    and a tarball may ship a world-writable or owner-only file; in both
    cases the delivery quarantines a private copy (B14) and prune reports
    the entry as ``poisoned`` rather than removing or hiding it.
    """
    mode = stat.S_IMODE(info.st_mode)
    # 0o7000 are the setuid/setgid/sticky bits: setuid is a property of the
    # inode, so a root-owned 04755 file would run as root from the account's
    # hardlink as well. A tarball can ship the bit; it never ships shared.
    return info.st_uid == ROOT_UID and not mode & 0o7022 and bool(mode & 0o004)

def _read_bytes(path, required_mode=None):
    try:
        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
    except OSError:
        return None
    try:
        info = os.fstat(fd)
        if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0
                or stat.S_IMODE(info.st_mode) & 0o022
                or (required_mode is not None
                    and stat.S_IMODE(info.st_mode) != required_mode)
                or info.st_size > MAX_CONFIG_BYTES):
            return None
        data = b''
        while len(data) <= MAX_CONFIG_BYTES:
            part = os.read(fd, min(8192, MAX_CONFIG_BYTES + 1 - len(data)))
            if not part:
                return data
            data += part
        return None
    finally:
        os.close(fd)

def protected_hardlinks_enabled(path=PROTECTED_HARDLINKS_PATH):
    """Return true only for the kernel's exact enabled sysctl value.

The kernel state is external input: read it with the same bounded,
    no-follow, root-owned-file contract used for administrator configuration.
    Missing or malformed state fails closed.
    """
    value = _read_bytes(path)
    return value in (b'1', b'1\n')

def _read_raw(path, required_mode=None):
    data = _read_bytes(path, required_mode=required_mode)
    if data is None:
        return None
    try:
        value = json.loads(data.decode('utf-8'))
    except (UnicodeError, ValueError, RecursionError):
        return None
    return value if isinstance(value, dict) else None

def _user_list(raw, key):
    """A list of account names, or nothing. Wrong types are ignored per
    key -- same policy as every other key in this file -- so a hand-edited
    `"excluded_users": "bob"` reads as an empty list, not as a crash and
    not as the letters b, o, b."""
    value = raw.get(key)
    if not isinstance(value, list):
        return []
    return [entry for entry in value if isinstance(entry, str)]

def exclusions_path(path=CONFIG_PATH):
    """Root-only account-exclusion authority beside the public policy."""
    return path + EXCLUSIONS_SUFFIX

def _exclusions(path, public_raw):
    """Return (account names, valid).

A pre-release configuration may still carry the list in its public JSON;
    reconcile migrates it. Once the private sidecar exists, malformed or
    unreadable content is never interpreted as an empty allow-all list.
    """
    private_path = exclusions_path(path)
    if not os.path.lexists(private_path):
        return _user_list(public_raw, 'excluded_users'), True
    private_raw = _read_raw(private_path, required_mode=EXCLUSIONS_MODE)
    if private_raw is None:
        return [], False
    value = private_raw.get('excluded_users')
    if (not isinstance(value, list)
            or any(not isinstance(entry, str) for entry in value)):
        return [], False
    return list(value), True

def _limits(raw):
    """Fill every ceiling from defaults, taking only values that are sane
    integers within bounds. A hand-edited garbage value reads as the
    default for that key alone -- the same per-key tolerance as the lists."""
    limits = dict(DEFAULT_LIMITS)
    given = raw.get('limits')
    if isinstance(given, dict):
        for key, (low, high) in LIMIT_BOUNDS.items():
            value = given.get(key)
            if (isinstance(value, int) and not isinstance(value, bool)
                    and low <= value <= high):
                limits[key] = value
    return limits

def _registries(raw):
    """The allowlist of registries a lockfile may name.

Preserve an exact empty or multi-entry list so status can diagnose a
    hand-edited or legacy configuration and delivery can fail closed. Missing
    and malformed values retain the safe npm-default behaviour.
    """
    given = raw.get('allowed_registries')
    if not isinstance(given, list):
        return list(DEFAULT_REGISTRIES)
    if not given:
        return []
    registries = [entry for entry in given
                  if isinstance(entry, str) and entry.strip()]
    return registries or list(DEFAULT_REGISTRIES)

def _storage_paths(raw):
    """The list of storage directories, newest key first, legacy folded in.

`storage_paths` (a list) is authoritative; a legacy single `storage_path`
    string is read as a one-element list. With neither key present the store
    defaults to DEFAULT_STORAGE_PATH (unchanged behaviour). Only syntactically
    valid, de-duplicated paths survive; a present-but-all-invalid setting
    yields the empty result, which the caller then treats as off -- exactly
    what an invalid single path did before."""
    if 'storage_paths' not in raw and 'storage_path' not in raw:
        return [DEFAULT_STORAGE_PATH]
    candidates = raw.get('storage_paths')
    if not isinstance(candidates, list):
        legacy = raw.get('storage_path')
        candidates = [legacy] if isinstance(legacy, str) else []
    paths = []
    for entry in candidates:
        if (isinstance(entry, str) and _valid_path_syntax(entry)
                and entry not in paths):
            paths.append(entry)
    return paths

def _storage_devices(raw):
    """Root-published device IDs used by account-side eligibility checks.

Storage is deliberately root-private.  The identifiers are not sensitive,
    but they let a selector running as an account determine same-filesystem
    eligibility without opening a protected storage path.  ``None`` means an
    older configuration did not publish the information yet.
    """
    value = raw.get('storage_devices')
    if not isinstance(value, list):
        return None
    devices = []
    for device in value:
        if (isinstance(device, int) and not isinstance(device, bool)
                and device >= 0 and device not in devices):
            devices.append(device)
    return devices

def _effective_config(raw, excluded_users, exclusions_valid):
    """Build the complete caller shape from already selected authorities."""
    paths = _storage_paths(raw) if raw is not None else []
    # Whether the administrator ever set a storage path, as opposed to the
    # displayed default: the CLI uses this to know if `--storage-path X` on a
    # fresh server means "use X" rather than "default plus X".
    configured = bool(raw is not None and paths
                      and ('storage_paths' in raw or 'storage_path' in raw))
    if raw is None or not paths:
        return {'enabled': False,
                'storage_paths': [DEFAULT_STORAGE_PATH],
                'storage_path': DEFAULT_STORAGE_PATH,
                'storage_devices': None,
                'storage_configured': False,
                'excluded_users': excluded_users,
                'exclusions_valid': exclusions_valid,
                'limits': dict(DEFAULT_LIMITS),
                'allowed_registries': list(DEFAULT_REGISTRIES)}
    return {'enabled': raw.get('enabled') is True,
            'storage_paths': paths,
            'storage_path': paths[0],
            'storage_devices': _storage_devices(raw),
            'storage_configured': configured,
            'excluded_users': excluded_users,
            'exclusions_valid': exclusions_valid,
            'limits': _limits(raw),
            'allowed_registries': _registries(raw)}

def read_config(path=CONFIG_PATH):
    """Return root-authoritative policy; invalid input is always off.

Every key is present on every return path so no caller has to guess: a
    partial or malformed public file yields the safe disabled shape with
    defaults. A malformed private exclusion authority refuses every account.
    """
    raw = _read_raw(path)
    excluded_users, exclusions_valid = _exclusions(path, raw or {})
    return _effective_config(raw, excluded_users, exclusions_valid)

def read_public_config(path=CONFIG_PATH):
    """Return only CageFS-readable policy without opening account names.

``exclusions_valid`` is unknown rather than true. Passing this view to
    ``is_user_included`` therefore fails closed; callers that need an account
    decision must ask the authenticated daemon.
    """
    return _effective_config(_read_raw(path), [], None)

def validate_limits(partial):
    """Range-check a partial {key: int} the CLI is about to write. Raises
    StorageError naming the offending key and its bounds."""
    for key, value in partial.items():
        if key not in LIMIT_BOUNDS:
            raise StorageError('unknown limit: %s' % key)
        low, high = LIMIT_BOUNDS[key]
        if isinstance(value, bool) or not isinstance(value, int) or not low <= value <= high:
            raise StorageError('%s must be an integer between %d and %d'
                               % (key, low, high))

def is_user_included(user, config):
    """Per-account carve-out on top of the global `enabled` flag.

The feature is for everyone; `excluded_users` is the handle for the
    account it cannot serve -- a git or tarball dependency, say -- carved
    out without touching the flag for the rest of the server. An excluded
    account takes the ordinary npm path: that is the point of the knob,
    not a failure. There is deliberately no allowlist counterpart: an
    administrator cannot know in advance which accounts are safe, so an
    allowlist rollout degenerates into more work than excluding the few
    that surface (and the config's unknown-key tolerance leaves room to
    add one later without migration, should a real need appear).

Takes the dict `read_config` returned; an absent key reads as an empty
    list, so a partial dict from an older caller means everyone included.
    """
    return (config.get('exclusions_valid', True)
            and user not in (config.get('excluded_users') or []))

def _valid_path_syntax(path):
    return (bool(path) and os.path.isabs(path) and path == os.path.normpath(path)
            and path not in _DANGEROUS_PATHS)

def _reject_symlink_ancestors(path):
    """Reject every lexical symlink component before creating or using it."""
    current = os.sep
    for component in path.split(os.sep)[1:]:
        current = os.path.join(current, component)
        try:
            info = os.lstat(current)
        except OSError:
            return
        if stat.S_ISLNK(info.st_mode):
            raise StorageError('path must not traverse a symlink')

def _require_root_private_dir(fd, label):
    info = os.fstat(fd)
    if (not stat.S_ISDIR(info.st_mode) or info.st_uid != 0
            or stat.S_IMODE(info.st_mode) & 0o022):
        raise StorageError('%s must be a root-owned non-writable directory' % label)

def _open_storage_dir(path, create=False):
    """Open a configured cache directory without ever following a component.

The fd is the authority for all later cache work.  Rechecking names with
    lstat and then reopening them would leave a replacement window for a
    hostile path component.
    """
    if not isinstance(path, str) or not _valid_path_syntax(path):
        raise StorageError('storage path must be an absolute, non-system directory')
    components = path.split(os.sep)[1:]
    created = False
    fd = os.open(os.sep, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        _require_root_private_dir(fd, 'storage path ancestor')
        for index, component in enumerate(components):
            try:
                child = os.open(component,
                                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                dir_fd=fd)
            except OSError as exc:
                if not create or index != len(components) - 1:
                    raise StorageError('storage path is not usable: %s' % exc)
                try:
                    os.mkdir(component, STORAGE_MODE, dir_fd=fd)
                    created = True
                    child = os.open(component,
                                    os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                    dir_fd=fd)
                except OSError as create_exc:
                    raise StorageError('cannot create storage path: %s' % create_exc)
            _require_root_private_dir(child, 'storage path component')
            os.close(fd)
            fd = child
        return fd, created
    except Exception:
        os.close(fd)
        raise

def validate_storage_path(path):
    """Prepare one explicit storage directory and prove local hardlinks work.

Only an absolute, normalized path below an existing non-symlink directory
    is accepted.  The final directory may be created root-private; existing
    storage is required to be a real root-owned mode-0700 directory and is
    never chmodded by this command. A probe proves hardlinks work *inside* it.
    It cannot prove compatibility with every account home, so delivery reports
    a clear cross-device failure when its own staging/cache operation needs it.
    """
    fd, created = _open_storage_dir(path, create=True)
    try:
        try:
            if created:
                os.fchmod(fd, STORAGE_MODE)
            elif stat.S_IMODE(os.fstat(fd).st_mode) != STORAGE_MODE:
                raise StorageError('existing storage path must already have mode 0700')
        except OSError as exc:
            raise StorageError('cannot secure storage path: %s' % exc)
        _check_hardlinks_fd(fd)
    finally:
        os.close(fd)

def storage_cache_is_empty(path):
    """Whether a registry may safely be changed for this storage.

Bun does not include the registry in its cache identity. Keep the check
    fd-relative and conservative: the usage snapshot, an empty delivery
    staging directory, an empty prune quarantine and a leftover hardlink
    probe file are metadata this code wrote itself, never content from the
    former registry; every other entry means the cache is not safe to reuse
    for another registry.
    """
    root_fd, _created = _open_storage_dir(path)
    try:
        try:
            names = os.listdir(root_fd)
        except OSError as exc:
            raise StorageError('cannot inspect storage cache: %s' % exc)
        for name in names:
            if (name == STORAGE_USAGE_FILE
                    or name.startswith(HARDLINK_PROBE_PREFIX)):
                # An interrupted `_check_hardlinks_fd` leaves its own
                # root-owned probe file behind. It holds nothing.
                try:
                    info = os.stat(name, dir_fd=root_fd, follow_symlinks=False)
                except OSError:
                    return False
                if stat.S_ISREG(info.st_mode) and info.st_uid == ROOT_UID:
                    continue
                return False
            if name not in ('.delivery-staging', '.tmp',
                            PRUNE_TRASH_DIRECTORY):
                return False
            try:
                staging_fd = os.open(
                    name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                    dir_fd=root_fd)
            except OSError:
                return False
            try:
                info = os.fstat(staging_fd)
                if name == '.tmp':
                    _require_root_private_dir(staging_fd,
                                              'Bun temporary directory')
                elif (info.st_uid != ROOT_UID
                      or stat.S_IMODE(info.st_mode) != STORAGE_MODE):
                    return False
                if os.listdir(staging_fd):
                    return False
            except (OSError, StorageError):
                return False
            finally:
                os.close(staging_fd)
        return True
    finally:
        os.close(root_fd)

def _check_hardlinks_fd(directory_fd):
    source = target = None
    created = False
    linked = False
    try:
        for _unused in range(16):
            source = HARDLINK_PROBE_PREFIX + os.urandom(12).hex()
            try:
                fd = os.open(source,
                             os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
                             0o600, dir_fd=directory_fd)
                created = True
                break
            except OSError as exc:
                if exc.errno != errno.EEXIST:
                    raise
        else:
            raise StorageError('cannot create unique hardlink probe')
        os.close(fd)
        target = source + '.link'
        os.link(source, target, src_dir_fd=directory_fd, dst_dir_fd=directory_fd,
                follow_symlinks=False)
        linked = True
    except OSError as exc:
        raise StorageError('storage path does not support hardlinks: %s' % exc)
    finally:
        for candidate in (target if linked else None, source if created else None):
            if candidate:
                try:
                    os.unlink(candidate, dir_fd=directory_fd)
                except OSError:
                    pass

def _open_lock_directory(path):
    """Open the root-private parent for a lock without trusting its name."""
    if (not isinstance(path, str) or not os.path.isabs(path)
            or path != os.path.normpath(path)):
        raise StorageError('storage lock path is invalid')
    directory, name = os.path.split(path)
    if not name:
        raise StorageError('storage lock path is invalid')
    fd = os.open(os.sep, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        _require_root_private_dir(fd, 'storage lock ancestor')
        components = directory.split(os.sep)[1:]
        for index, component in enumerate(components):
            try:
                child = os.open(component,
                                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                dir_fd=fd)
            except OSError as exc:
                if index != len(components) - 1 or exc.errno != errno.ENOENT:
                    raise StorageError('storage lock directory is not usable: %s' % exc)
                try:
                    os.mkdir(component, STORAGE_MODE, dir_fd=fd)
                except OSError as mkdir_exc:
                    if mkdir_exc.errno != errno.EEXIST:
                        raise StorageError('cannot create storage lock directory: %s' % mkdir_exc)
                try:
                    child = os.open(component,
                                    os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                    dir_fd=fd)
                except OSError as open_exc:
                    raise StorageError('storage lock directory is not usable: %s' % open_exc)
            try:
                _require_root_private_dir(child, 'storage lock directory')
                if (index == len(components) - 1
                        and stat.S_IMODE(os.fstat(child).st_mode) != STORAGE_MODE):
                    raise StorageError('storage lock directory must have mode 0700')
            except Exception:
                os.close(child)
                raise
            os.close(fd)
            fd = child
        return fd, name
    except Exception:
        os.close(fd)
        raise

class _StorageLock(object):
    def __init__(self, exclusive, path=LOCK_PATH, nonblocking=False,
                 timeout=None):
        self._exclusive = exclusive
        self._path = path
        self._nonblocking = nonblocking
        self._timeout = timeout
        self._fd = None

def __enter__(self):
        directory_fd, name = _open_lock_directory(self._path)
        try:
            try:
                self._fd = os.open(name, os.O_RDWR | os.O_NOFOLLOW,
                                   dir_fd=directory_fd)
                created = False
            except OSError as exc:
                if exc.errno != errno.ENOENT:
                    raise
                try:
                    self._fd = os.open(
                        name, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
                        0o600, dir_fd=directory_fd)
                    created = True
                except OSError as create_exc:
                    if create_exc.errno != errno.EEXIST:
                        raise
                    self._fd = os.open(name, os.O_RDWR | os.O_NOFOLLOW,
                                       dir_fd=directory_fd)
                    created = False
            if created:
                os.fchmod(self._fd, 0o600)
            info = os.fstat(self._fd)
            if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0
                    or stat.S_IMODE(info.st_mode) & 0o077):
                raise StorageError('storage lock must be a root-owned private regular file')
            flags = fcntl.LOCK_EX if self._exclusive else fcntl.LOCK_SH
            deadline = None
            if self._timeout is not None:
                if self._timeout < 0:
                    raise StorageError('storage lock timeout must be non-negative')
                deadline = time.monotonic() + self._timeout
            if self._nonblocking or deadline is not None:
                flags |= fcntl.LOCK_NB
            while True:
                try:
                    fcntl.flock(self._fd, flags)
                    break
                except OSError as exc:
                    if exc.errno not in (
                            errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK):
                        raise
                    if self._nonblocking:
                        raise CachePopulationBusy('storage lock is busy')
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        raise StorageError('storage lock wait timed out')
                    time.sleep(min(0.05, remaining))
        except StorageError:
            if self._fd is not None:
                os.close(self._fd)
                self._fd = None
            raise
        except OSError as exc:
            if self._fd is not None:
                os.close(self._fd)
                self._fd = None
            raise StorageError('cannot open storage lock: %s' % exc)
        finally:
            os.close(directory_fd)
        return self

def __exit__(self, *exc):
        if self._fd is not None:
            os.close(self._fd)
            self._fd = None

def storage_lock(exclusive, path=LOCK_PATH, timeout=None):
    return _StorageLock(exclusive, path, timeout=timeout)

def config_lock(path=CONFIG_LOCK_PATH, timeout=None):
    """Serialize the supported CLI's complete config read-modify-write."""
    return _StorageLock(True, path, timeout=timeout)

def cache_population_lock(path, lock_directory=None, nonblocking=False,
                          timeout=None):
    """Serialize Bun cache writers for one pinned storage directory.

Delivery keeps its existing shared storage lock for the complete
    transaction, while this narrower lock covers only Bun's cache mutation.
    Distinct storage directories therefore remain independent.
    """
    root_fd, _created = _open_storage_dir(path)
    try:
        info = os.fstat(root_fd)
    finally:
        os.close(root_fd)
    if lock_directory is None:
        lock_directory = LOCK_DIRECTORY
    name = 'bun-cache-%x-%x.lock' % (info.st_dev, info.st_ino)
    return _StorageLock(True, os.path.join(lock_directory, name),
                        nonblocking=nonblocking, timeout=timeout)

def _prune_maintenance_lock(path, lock_directory=None, timeout=None):
    """Serialize prune commands without making deliveries wait on this lock."""
    root_fd, _created = _open_storage_dir(path)
    try:
        info = os.fstat(root_fd)
    finally:
        os.close(root_fd)
    if lock_directory is None:
        lock_directory = LOCK_DIRECTORY
    name = 'bun-prune-%x-%x.lock' % (info.st_dev, info.st_ino)
    return _StorageLock(True, os.path.join(lock_directory, name),
                        timeout=timeout)

def _is_canonical_name(name):
    """Measured Bun 1.4 name: package@version@@@cache-generation."""
    marker = name.rfind('@@@')
    if marker <= 0 or not _ASCII_DIGITS.match(name[marker + 3:]):
        return False
    package_version = name[:marker]
    separator = package_version.rfind('@')
    version = package_version[separator + 1:]
    return (separator > 0 and version
            and ord('0') <= ord(version[0]) <= ord('9')
            and '/' not in package_version and not package_version.startswith('.'))

def _canonical_parts(name):
    marker = name.rfind('@@@')
    package_version = name[:marker]
    separator = package_version.rfind('@')
    return package_version[:separator], package_version[separator + 1:] + name[marker:]

def _safe_entry_tree(fd, deadline=None):
    """Return (state, bytes): removable, kept (linked), poisoned, or skipped.

``poisoned`` is an owner/mode anomaly on a directory or regular file --
    the trace a ``chown -R`` leaves -- and is reported separately from the
    other anomalies because it is the one the administrator has to act on.
    """
    total = 0
    seen = set()
    has_regular = False
    stack = [os.dup(fd)]
    try:
        while stack:
            if deadline is not None and time.monotonic() >= deadline:
                return 'skipped', 0
            current = stack.pop()
            try:
                for name in os.listdir(current):
                    if deadline is not None and time.monotonic() >= deadline:
                        return 'skipped', 0
                    try:
                        info = os.stat(name, dir_fd=current, follow_symlinks=False)
                    except OSError:
                        return 'skipped', 0
                    if stat.S_ISDIR(info.st_mode):
                        if info.st_uid != 0 or stat.S_IMODE(info.st_mode) & 0o022:
                            return 'poisoned', 0
                        try:
                            child = os.open(name,
                                            os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                            dir_fd=current)
                        except OSError:
                            return 'skipped', 0
                        stack.append(child)
                    elif stat.S_ISREG(info.st_mode):
                        has_regular = True
                        if not is_sound_shared_inode(info):
                            return 'poisoned', 0
                        if info.st_nlink > 1:
                            return 'kept', 0
                        inode = (info.st_dev, info.st_ino)
                        if inode not in seen:
                            seen.add(inode)
                            total += info.st_size
                    else:
                        return 'skipped', 0
            finally:
                os.close(current)
    except OSError:
        return 'skipped', 0
    finally:
        while stack:
            os.close(stack.pop())
    return ('removable', total) if has_regular else ('skipped', 0)

def _remove_tree(fd, allow_symlinks=False, deadline=None):
    """Remove a previously scanned tree, still refusing all unsafe nodes.

A tree deeper than `MAX_REMOVAL_DEPTH` is refused rather than removed:
    every caller counts that as `skipped`, so the entry stays quarantined for
    the next prune instead of costing this one unbounded time under the locks.
    """
    # Post-order traversal without recursion: cache entries are untrusted
    # filesystem trees and a deeply nested package must not exhaust Python's
    # recursion limit while a root pruning command is running.
    #
    # One single pass, too: re-deriving each directory's path from the scan
    # root cost one `open` per component, twice per directory, which is
    # quadratic in depth while this runs under the delivery population lock.
    # Only the directory being emptied holds a descriptor; a popped frame
    # reaches its parent through `..` and proves it is the inode it
    # descended from.
    def open_dir(dir_fd, name):
        child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                        dir_fd=dir_fd)
        try:
            return child, os.fstat(child)
        except BaseException:
            os.close(child)
            raise

current = os.dup(fd)
    root_info = os.fstat(current)
    frames = [{'name': None, 'names': None, 'index': 0,
               'dev': root_info.st_dev, 'ino': root_info.st_ino}]
    try:
        while frames:
            if deadline is not None and time.monotonic() >= deadline:
                raise StorageError('prune maintenance slice expired')
            frame = frames[-1]
            if frame['names'] is None:
                frame['names'] = os.listdir(current)
            if frame['index'] >= len(frame['names']):
                if frame['name'] is None:
                    frames.pop()
                    continue
                parent, parent_info = open_dir(current, '..')
                above = frames[-2]
                try:
                    if (parent_info.st_dev != above['dev']
                            or parent_info.st_ino != above['ino']):
                        raise StorageError('entry changed while pruning')
                    os.rmdir(frame['name'], dir_fd=parent)
                except BaseException:
                    os.close(parent)
                    raise
                os.close(current)
                current = parent
                frames.pop()
                continue
            name = frame['names'][frame['index']]
            frame['index'] += 1
            info = os.stat(name, dir_fd=current, follow_symlinks=False)
            if stat.S_ISDIR(info.st_mode):
                if len(frames) >= MAX_REMOVAL_DEPTH:
                    raise StorageError('entry is too deep to prune')
                child, child_info = open_dir(current, name)
                os.close(current)
                current = child
                frames.append({'name': name, 'names': None, 'index': 0,
                               'dev': child_info.st_dev,
                               'ino': child_info.st_ino})
            elif stat.S_ISREG(info.st_mode) and info.st_nlink == 1:
                os.unlink(name, dir_fd=current)
            elif allow_symlinks and stat.S_ISLNK(info.st_mode):
                os.unlink(name, dir_fd=current)
            else:
                raise StorageError('entry changed while pruning')
    finally:
        os.close(current)

def _safe_temporary_tree(fd, deadline=None):
    """Classify one observed Bun 1.4 temporary cache entry.

Symlinks are data inside npm tarballs and are safe to unlink without
    following. A regular file with another hardlink means a delivered tree
    still uses this otherwise non-reusable temporary entry, so retain the
    whole directory until a later prune.
    """
    total = 0
    seen = set()
    stack = [os.dup(fd)]
    try:
        while stack:
            if deadline is not None and time.monotonic() >= deadline:
                return 'skipped', 0
            current = stack.pop()
            try:
                for name in os.listdir(current):
                    if deadline is not None and time.monotonic() >= deadline:
                        return 'skipped', 0
                    try:
                        info = os.stat(name, dir_fd=current, follow_symlinks=False)
                    except OSError:
                        return 'skipped', 0
                    if stat.S_ISDIR(info.st_mode):
                        if info.st_uid != ROOT_UID or stat.S_IMODE(info.st_mode) & 0o022:
                            return 'poisoned', 0
                        try:
                            child = os.open(
                                name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                dir_fd=current)
                        except OSError:
                            return 'skipped', 0
                        stack.append(child)
                    elif stat.S_ISREG(info.st_mode):
                        # Bun preserves package payload modes. A 0600 file is
                        # unsuitable for sharing but still safe to unlink once
                        # it is root-owned and has no other hardlink.
                        if info.st_uid != ROOT_UID:
                            return 'poisoned', 0
                        if info.st_nlink > 1:
                            return 'kept', 0
                        inode = (info.st_dev, info.st_ino)
                        if inode not in seen:
                            seen.add(inode)
                            total += info.st_size
                    elif stat.S_ISLNK(info.st_mode):
                        if info.st_uid != ROOT_UID:
                            return 'poisoned', 0
                    else:
                        return 'skipped', 0
            finally:
                os.close(current)
    except OSError:
        return 'skipped', 0
    finally:
        while stack:
            os.close(stack.pop())
    return 'removable', total

def _prune_bun_temporary_entries(root_fd, summary, deadline=None):
    """Remove only inactive Bun 1.4 race/crash residue below ``.tmp``."""
    try:
        temporary_fd = os.open(
            '.tmp', os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
            dir_fd=root_fd)
    except OSError as exc:
        if exc.errno != errno.ENOENT:
            summary['skipped'] += 1
        return
    try:
        try:
            _require_root_private_dir(temporary_fd, 'Bun temporary directory')
        except StorageError:
            summary['poisoned'] += 1
            return
        try:
            names = sorted(os.listdir(temporary_fd))
        except OSError:
            summary['skipped'] += 1
            return
        for name in names:
            if deadline is not None and time.monotonic() >= deadline:
                break
            if not _BUN_TEMPORARY_ENTRY.match(name):
                summary['skipped'] += 1
                continue
            try:
                entry_fd = os.open(
                    name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                    dir_fd=temporary_fd)
            except OSError:
                summary['skipped'] += 1
                continue
            try:
                try:
                    _require_root_private_dir(entry_fd, 'Bun temporary entry')
                except StorageError:
                    summary['poisoned'] += 1
                    continue
                state, reclaimed = _safe_temporary_tree(
                    entry_fd, deadline=deadline)
                if state != 'removable':
                    summary[state] += 1
                    continue
                _remove_tree(entry_fd, allow_symlinks=True,
                             deadline=deadline)
                os.rmdir(name, dir_fd=temporary_fd)
                summary['removed'] += 1
                summary['bytes'] += reclaimed
            except (OSError, StorageError):
                summary['skipped'] += 1
            finally:
                os.close(entry_fd)
    finally:
        os.close(temporary_fd)
    try:
        os.rmdir('.tmp', dir_fd=root_fd)
    except OSError:
        pass

def _remove_lookup(parent_fd, package, version, canonical, expected_target):
    """Drop only Bun's exact lookup symlink; unknown lookup data survives."""
    try:
        lookup_fd = os.open(package,
                            os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                            dir_fd=parent_fd)
    except OSError:
        return
    try:
        _require_root_private_dir(lookup_fd, 'Bun lookup directory')
        if os.readlink(version, dir_fd=lookup_fd) != expected_target:
            return
        os.unlink(version, dir_fd=lookup_fd)
    except OSError:
        return
    finally:
        os.close(lookup_fd)
    try:
        os.rmdir(package, dir_fd=parent_fd)
    except OSError:
        pass

def _open_prune_trash(root_fd, create=False):
    try:
        fd = os.open(PRUNE_TRASH_DIRECTORY,
                     os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                     dir_fd=root_fd)
    except OSError as exc:
        if not create:
            return None
        if exc.errno != errno.ENOENT:
            raise StorageError('cannot open prune quarantine: %s' % exc)
        try:
            os.mkdir(PRUNE_TRASH_DIRECTORY, STORAGE_MODE, dir_fd=root_fd)
            fd = os.open(PRUNE_TRASH_DIRECTORY,
                         os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                         dir_fd=root_fd)
        except OSError as create_exc:
            raise StorageError('cannot create prune quarantine: %s' % create_exc)
    try:
        _require_root_private_dir(fd, 'prune quarantine')
    except Exception:
        os.close(fd)
        raise
    return fd

def _clean_prune_trash(root_fd, summary):
    """Finish deletions whose canonical name was already quarantined."""
    trash_fd = _open_prune_trash(root_fd)
    if trash_fd is None:
        return
    try:
        try:
            names = os.listdir(trash_fd)
        except OSError as exc:
            raise StorageError('cannot list prune quarantine: %s' % exc)
        for name in names:
            if not _PRUNE_TRASH_ENTRY.match(name):
                summary['skipped'] += 1
                continue
            entry_fd = None
            try:
                entry_fd = os.open(name,
                                   os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                   dir_fd=trash_fd)
                _require_root_private_dir(entry_fd, 'quarantined cache entry')
                _remove_tree(entry_fd)
                os.rmdir(name, dir_fd=trash_fd)
            except (OSError, StorageError):
                summary['skipped'] += 1
            finally:
                if entry_fd is not None:
                    os.close(entry_fd)
    finally:
        os.close(trash_fd)
    try:
        os.rmdir(PRUNE_TRASH_DIRECTORY, dir_fd=root_fd)
    except OSError:
        pass

def _prune_canonical_entry(root_fd, candidate, summary, deadline=None):
    """Classify and remove one canonical entry while population is paused."""
    parent_fd, canonical, package, version, expected_target = candidate
    if deadline is not None and time.monotonic() >= deadline:
        summary['skipped'] += 1
        return
    # A lookup hierarchy anomaly is a safe leak: the canonical content
    # remains so we never disturb unexpected Bun data.
    try:
        lookup_fd = os.open(package,
                            os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                            dir_fd=parent_fd)
    except OSError:
        lookup_fd = None
    if lookup_fd is not None:
        try:
            _require_root_private_dir(lookup_fd, 'Bun lookup directory')
        except StorageError:
            summary['skipped'] += 1
            return
        finally:
            os.close(lookup_fd)
    try:
        entry_fd = os.open(canonical,
                           os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                           dir_fd=parent_fd)
    except OSError:
        summary['skipped'] += 1
        return
    try:
        try:
            _require_root_private_dir(entry_fd, 'Bun canonical entry')
        except StorageError:
            summary['poisoned'] += 1
            return
        state, reclaimed = _safe_entry_tree(entry_fd, deadline=deadline)
        if state != 'removable':
            summary[state] += 1
            return
        trash_fd = _open_prune_trash(root_fd, create=True)
        trash_name = 'entry-%s' % os.urandom(16).hex()
        try:
            os.rename(canonical, trash_name,
                      src_dir_fd=parent_fd, dst_dir_fd=trash_fd)
            # Quarantine canonical bytes first. The lookup disappears only
            # after the rename, so interruption cannot publish partial data.
            _remove_lookup(parent_fd, package, version, canonical,
                           expected_target)
            _remove_tree(entry_fd, deadline=deadline)
            os.rmdir(trash_name, dir_fd=trash_fd)
        finally:
            os.close(trash_fd)
        summary['removed'] += 1
        summary['bytes'] += reclaimed
    except (OSError, StorageError):
        summary['skipped'] += 1
    finally:
        os.close(entry_fd)

def prune_storage(path, lock_path=LOCK_PATH, timeout=None):
    """Prune unused, measured Bun 1.4 canonical cache entries only."""
    summary = {'removed': 0, 'bytes': 0, 'kept': 0, 'skipped': 0, 'poisoned': 0}
    if timeout is None:
        timeout = DEFAULT_LIMITS['delivery_queue_timeout_seconds']
    # Keep registry/storage configuration stable with the global shared lock,
    # but serialize destructive cache work only against population of this
    # exact store. Daily maintenance must not stop deliveries using another
    # configured filesystem.
    lock_directory = os.path.dirname(lock_path)
    with storage_lock(False, lock_path, timeout=timeout), \
         _prune_maintenance_lock(path, lock_directory, timeout=timeout):
        root_fd, _created = _open_storage_dir(path)
        scope_fds = []
        candidate_scopes = set()
        try:
            if stat.S_IMODE(os.fstat(root_fd).st_mode) != STORAGE_MODE:
                raise StorageError('storage path must already have mode 0700')
            try:
                top_entries = sorted(os.listdir(root_fd))
            except OSError as exc:
                raise StorageError('cannot scan storage path: %s' % exc)
            _clean_prune_trash(root_fd, summary)
            with cache_population_lock(
                    path, lock_directory, timeout=timeout):
                _prune_bun_temporary_entries(
                    root_fd, summary,
                    deadline=time.monotonic() + PRUNE_LOCK_SLICE_SECONDS)
            # Trash is outside Bun's canonical namespace. Finish any deletion
            # after releasing the population lock so delivery gets priority.
            _clean_prune_trash(root_fd, summary)
            candidates = []
            for name in top_entries:
                try:
                    info = os.stat(name, dir_fd=root_fd, follow_symlinks=False)
                except OSError:
                    continue
                if stat.S_ISDIR(info.st_mode) and _is_canonical_name(name):
                    package, version = _canonical_parts(name)
                    candidates.append((root_fd, name, package, version,
                                       os.path.join(path, name)))
                elif (stat.S_ISDIR(info.st_mode) and name.startswith('@')
                      and '/' not in name and not name.startswith('.')):
                    scope_fd = None
                    try:
                        scope_fd = os.open(name,
                                           os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                           dir_fd=root_fd)
                        _require_root_private_dir(scope_fd, 'Bun scope directory')
                        scoped_entries = sorted(os.listdir(scope_fd))
                    except (OSError, StorageError):
                        if scope_fd is not None:
                            os.close(scope_fd)
                        summary['skipped'] += 1
                        continue
                    scope_fds.append(scope_fd)
                    for child in scoped_entries:
                        try:
                            child_info = os.stat(child, dir_fd=scope_fd,
                                                 follow_symlinks=False)
                        except OSError:
                            continue
                        if stat.S_ISDIR(child_info.st_mode) and _is_canonical_name(child):
                            package, version = _canonical_parts(child)
                            candidates.append((scope_fd, child, package, version,
                                               os.path.join(path, name, child)))
                            candidate_scopes.add(name)
                    # Keep the fd for candidate deletion; closing it would
                    # reintroduce a path-based reopen.
            offset = 0
            while offset < len(candidates):
                batch_end = min(offset + PRUNE_BATCH_ENTRIES, len(candidates))
                with cache_population_lock(
                        path, lock_directory, timeout=timeout):
                    deadline = time.monotonic() + PRUNE_LOCK_SLICE_SECONDS
                    while offset < batch_end:
                        _prune_canonical_entry(
                            root_fd, candidates[offset], summary,
                            deadline=deadline)
                        offset += 1
                        if time.monotonic() >= deadline:
                            break
                # Quarantined content no longer participates in Bun's cache.
                # Its potentially slow deletion must not occupy the lock that
                # a delivery needs to populate this store.
                _clean_prune_trash(root_fd, summary)
                if offset < len(candidates):
                    time.sleep(PRUNE_DELIVERY_YIELD_SECONDS)
            # A scope is a Bun lookup container only when it contained a
            # canonical child we just considered.  rmdir is safe and leaves
            # any non-empty (including unknown) scope untouched.
            with cache_population_lock(
                    path, lock_directory, timeout=timeout):
                for scope in candidate_scopes:
                    try:
                        os.rmdir(scope, dir_fd=root_fd)
                    except OSError:
                        pass
            # Pruning is exactly when the recorded size is most wrong; refresh
            # it after cache mutation. The usage record is telemetry, not an
            # admission transaction; delivery must not wait behind this walk.
            try:
                usage = _measure_tree_usage_fd(root_fd)
                write_usage(path, usage)
            except StorageError:
                # Reclaim already completed safely. Keep that result useful to
                # the administrator and make the stale telemetry explicit.
                summary['warnings'] = [
                    'storage usage snapshot could not be refreshed']
            return summary
        finally:
            for scope_fd in scope_fds:
                os.close(scope_fd)
            os.close(root_fd)

# ------------------------------------------------------------ size accounting

def _measure_tree_usage_fd(root_fd):
    """Unique logical and allocated bytes below a pinned storage root fd.

Hardlinks are the whole point of the store, so a file linked N times must
    count once: dedup by (dev, ino). Odd nodes do not contribute bytes; inside
    a canonical entry they mark that entry poisoned. This is accounting
    telemetry, never a delivery decision.

Which is why its own entry/time bounds do not refuse. Raising on a real
    server-wide cache stopped the snapshot advancing at all: every delivery
    and prune logged the same failure, `status` froze on an old number, and a
    missing record made `over_soft_storage_limit` refuse every delivery with
    a "run prune" that could not help. A bound returns what it counted so
    far, marked `stale` with the bound it hit -- a floor, not a size."""
    used = allocated = shared_inodes = shared_links = saved = 0
    seen = set()
    poisoned_entries = set()
    stale_reason = None
    # Retain one pinned directory descriptor per DFS level. Reopening every
    # descendant from root is quadratic on a hostile deep tree.
    frames = []
    entries = 0
    deadline = time.monotonic() + MEASURE_MAX_SECONDS
    try:
        frames.append({'fd': os.dup(root_fd), 'names': None, 'index': 0,
                       'depth': 0, 'canonical': None})
        while frames:
            if time.monotonic() > deadline:
                stale_reason = 'maximum measurement time exceeded'
                break
            frame = frames[-1]
            if frame['names'] is None:
                try:
                    frame['names'] = os.listdir(frame['fd'])
                except OSError:
                    os.close(frame['fd'])
                    frames.pop()
                    continue
            if frame['index'] >= len(frame['names']):
                os.close(frame['fd'])
                frames.pop()
                continue
            name = frame['names'][frame['index']]
            frame['index'] += 1
            entries += 1
            if entries > MEASURE_MAX_ENTRIES:
                stale_reason = 'maximum measurement entry count exceeded'
                break
            try:
                info = os.stat(name, dir_fd=frame['fd'], follow_symlinks=False)
            except OSError:
                continue
            canonical = frame['canonical']
            if stat.S_ISDIR(info.st_mode):
                depth = frame['depth'] + 1
                if depth > MAX_TREE_WALK_DEPTH:
                    raise StorageError('maximum tree depth exceeded')
                child_canonical = canonical
                if canonical is None and (
                        (frame['depth'] == 0 and _is_canonical_name(name))
                        or (frame['depth'] == 1
                            and frame.get('child_name', '').startswith('@')
                            and _is_canonical_name(name))):
                    child_canonical = (info.st_dev, info.st_ino)
                if (child_canonical is not None
                        and (info.st_uid != ROOT_UID
                             or stat.S_IMODE(info.st_mode) & 0o022)):
                    poisoned_entries.add(child_canonical)
                child_fd = None
                try:
                    child_fd = os.open(
                        name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                        dir_fd=frame['fd'])
                    opened = os.fstat(child_fd)
                    if (not stat.S_ISDIR(opened.st_mode)
                            or opened.st_dev != info.st_dev
                            or opened.st_ino != info.st_ino):
                        os.close(child_fd)
                        child_fd = None
                        continue
                except OSError:
                    if child_fd is not None:
                        os.close(child_fd)
                    continue
                frames.append({'fd': child_fd, 'names': None, 'index': 0,
                               'depth': depth, 'canonical': child_canonical,
                               'child_name': name})
            elif stat.S_ISREG(info.st_mode):
                if canonical is not None and not is_sound_shared_inode(info):
                    poisoned_entries.add(canonical)
                key = (info.st_dev, info.st_ino)
                if key not in seen:
                    seen.add(key)
                    used += info.st_size
                    blocks = info.st_blocks * 512
                    allocated += blocks
                    if info.st_nlink > 1:
                        shared_inodes += 1
                        shared_links += info.st_nlink - 1
                        saved += blocks * (info.st_nlink - 1)
            elif canonical is not None:
                poisoned_entries.add(canonical)
    finally:
        for frame in frames:
            try:
                os.close(frame['fd'])
            except OSError:
                pass
    return {
        'used_bytes': used,
        'allocated_bytes': allocated,
        'shared_inodes': shared_inodes,
        'shared_links': shared_links,
        'estimated_saved_bytes': saved,
        'poisoned_entries': len(poisoned_entries),
        'deduplication_ratio': ((allocated + saved) / float(allocated)
                                if allocated else 1.0),
        'stale': stale_reason is not None,
        'stale_reason': stale_reason,
    }

def _measure_tree_fd(root_fd):
    """Compatibility helper for the existing logical-size API."""
    return _measure_tree_usage_fd(root_fd)['used_bytes']

def measure_storage(path):
    """Walk one storage directory and return its unique byte size."""
    return measure_storage_usage(path)['used_bytes']

def measure_storage_usage(path):
    """Walk one storage directory and return its accounting record."""
    root_fd, _created = _open_storage_dir(path)
    try:
        return _measure_tree_usage_fd(root_fd)
    finally:
        os.close(root_fd)

def has_storage_for_device(storage_paths, device):
    """Whether a configured, presently usable storage shares ``device``.

This is a read-only eligibility check for selector listings; unlike
    ``validate_storage_path`` it never creates or changes a directory.
    """
    for path in storage_paths:
        try:
            fd, _created = _open_storage_dir(path)
        except StorageError:
            continue
        try:
            if os.fstat(fd).st_dev == device:
                return True
        finally:
            os.close(fd)
    return False

def storage_device_ids(storage_paths):
    """Safely obtain device IDs for storage paths validated by root."""
    devices = []
    for path in storage_paths:
        fd, _created = _open_storage_dir(path)
        try:
            device = os.fstat(fd).st_dev
        finally:
            os.close(fd)
        if device not in devices:
            devices.append(device)
    return devices

def reconcile_storage_devices(path=CONFIG_PATH):
    """Publish device IDs and migrate legacy public account exclusions.

The account-side selector cannot inspect a root-private storage directory,
    so a config written before ``storage_devices`` existed must be upgraded by
    root. A legacy ``excluded_users`` list is moved to the root-only sidecar
    even while delivery is disabled. Revalidate every existing store first and
    write only when the full current snapshot differs. A bad store raises
    without altering the config, leaving account-side eligibility safely false.
    """
    raw = _read_raw(path)
    if raw is None:
        return False
    migrate_exclusions = 'excluded_users' in raw
    excluded_users = []
    if migrate_exclusions:
        excluded_users, exclusions_valid = _exclusions(path, raw)
        if not exclusions_valid:
            raise StorageError('account exclusion policy is unreadable')
    has_configured_paths = ('storage_paths' in raw or 'storage_path' in raw)
    reconcile_devices = raw.get('enabled') is True or has_configured_paths
    if not reconcile_devices:
        if not migrate_exclusions:
            return False
        raw.pop('excluded_users', None)
        _write_config_and_exclusions(
            path, raw, excluded_users, write_exclusions=True)
        return True
    paths = _storage_paths(raw)
    if not paths:
        raise StorageError('storage paths are not usable')
    devices = []
    for storage_path in paths:
        fd, _created = _open_storage_dir(storage_path)
        try:
            if stat.S_IMODE(os.fstat(fd).st_mode) != STORAGE_MODE:
                raise StorageError('existing storage path must already have mode 0700')
            _check_hardlinks_fd(fd)
            device = os.fstat(fd).st_dev
        finally:
            os.close(fd)
        if device not in devices:
            devices.append(device)
    devices_changed = _storage_devices(raw) != devices
    if not devices_changed and not migrate_exclusions:
        return False
    if devices_changed:
        raw['storage_devices'] = devices
    raw.pop('excluded_users', None)
    _write_config_and_exclusions(
        path, raw, excluded_users, write_exclusions=migrate_exclusions)
    return True

def _usage_path(path):
    return os.path.join(path, STORAGE_USAGE_FILE)

def read_usage(path):
    """The last recorded size of a storage directory, or None if never
    measured / unreadable. The file is 0644, but it sits inside the 0700
    store, so only root can read it. Non-root status reports unknown rather
    than activating the privileged daemon for server-wide statistics."""
    try:
        fd = os.open(_usage_path(path), os.O_RDONLY | os.O_NOFOLLOW)
    except OSError:
        return None
    try:
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size > 4096:
            return None
        data = os.read(fd, 4096)
    finally:
        os.close(fd)
    try:
        value = json.loads(data.decode('utf-8'))
    except (UnicodeError, ValueError):
        return None
    if (not isinstance(value, dict)
            or type(value.get('used_bytes')) is not int
            or value['used_bytes'] < 0):
        return None
    result = {'used_bytes': value['used_bytes'],
              'measured_at': value.get('measured_at'),
              # A record written before measurement could stop at a bound was
              # always complete; absent means complete, never unknown.
              'stale': value.get('stale') is True,
              'stale_reason': value.get('stale_reason')}
    for name in ('allocated_bytes', 'shared_inodes', 'shared_links',
                 'estimated_saved_bytes', 'deduplication_ratio',
                 'poisoned_entries'):
        result[name] = value.get(name)
    return result

def _active_shared_node_modules_counts():
    """Return aggregate active user/application counts, never identities."""
    try:
        from clselect.clselectnodejs.apps_manager import ApplicationsManager
        info = ApplicationsManager().get_applications_users_info()
        if ('warning' in info or 'available_versions' not in info
                or not isinstance(info['available_versions'], dict)):
            return None, None
        active_users = set()
        active_applications = 0
        for version in info['available_versions'].values():
            if not isinstance(version, dict):
                return None, None
            # An installed interpreter that no account uses carries no
            # `users` key at all (only versions with applications get one),
            # exactly like an explicit `None`: an empty version, never an
            # unknown count for the whole server.
            users = version.get('users')
            if users is None:
                continue
            if not isinstance(users, dict):
                return None, None
            for user, user_info in users.items():
                if not isinstance(user_info, dict):
                    return None, None
                applications = user_info.get('applications')
                if not isinstance(applications, dict):
                    return None, None
                for application in applications.values():
                    if (isinstance(application, dict)
                            and application.get('shared_node_modules', {}).get('active') is True):
                        active_users.add(user)
                        active_applications += 1
        return len(active_users), active_applications
    except Exception:
        return None, None

def _sum_recorded_usage(records, name):
    """Sum one non-negative integer field, or return unknown."""
    values = [record.get(name) if isinstance(record, dict) else None
              for record in records]
    if not values or not all(type(value) is int and value >= 0
                             for value in values):
        return None
    return sum(values)

def _measurement_stale(records):
    """Whether any contributing measurement stopped at one of its own bounds.

A measurement that hit MEASURE_MAX_ENTRIES / MEASURE_MAX_SECONDS records a
    floor, so every aggregate built from it is a floor too. One known floor is
    a more useful answer than "cannot tell" and outranks an unreadable record;
    with no floor but a record missing, completeness is simply unknown.
    """
    if any(isinstance(record, dict) and record.get('stale') is True
           for record in records):
        return True
    if any(not isinstance(record, dict) for record in records):
        return None
    return False

def get_shared_node_modules_summary(path=CONFIG_PATH, active_counts=None):
    """Return privacy-safe aggregates for ``cloudlinux-summary``.

Storage sizes come only from existing accounting snapshots. This function
    deliberately never walks the shared storage tree; delivery and prune own
    those comparatively expensive measurements.
    """
    config = read_config(path)
    if (isinstance(active_counts, tuple) and len(active_counts) == 2
            and all(type(value) is int and value >= 0
                    for value in active_counts)):
        active_users, active_applications = active_counts
    elif active_counts is None:
        active_users, active_applications = \
            _active_shared_node_modules_counts()
    else:
        active_users, active_applications = None, None
    records = [read_usage(storage_path)
               for storage_path in config['storage_paths']]
    stale = _measurement_stale(records)
    # Totals stay: a sum of floors is still a lower bound, and the flag says
    # so. That holds for poisoned entries a truncated walk did reach; only a
    # zero is inconclusive -- "none in the part it got through" is not "none".
    poisoned = _sum_recorded_usage(records, 'poisoned_entries')
    if stale is True and not poisoned:
        poisoned = None
    return {
        'enabled': config['enabled'],
        'applications_amount': active_applications,
        'users_amount': active_users,
        'storage_allocated_bytes': _sum_recorded_usage(
            records, 'allocated_bytes'),
        'estimated_saved_bytes': _sum_recorded_usage(
            records, 'estimated_saved_bytes'),
        'poisoned_entries_amount': poisoned,
        'measurement_stale': stale,
    }

def write_usage(path, usage):
    """Record a storage directory's size beside it, atomically and
    root-owned but world-readable."""
    root_fd, _created = _open_storage_dir(path)
    temporary = '.cl-storage-usage-%s' % os.urandom(8).hex()
    if isinstance(usage, dict):
        record = dict(usage)
    else:
        record = {'used_bytes': int(usage)}
    record['used_bytes'] = int(record['used_bytes'])
    # A measurement that stopped at one of its own bounds records a floor,
    # and says so, so that later readers do not mistake it for a size.
    record['stale'] = bool(record.get('stale'))
    record['stale_reason'] = record.get('stale_reason') or None
    record['measured_at'] = time.time()
    body = (json.dumps(record) + '\n').encode('utf-8')
    fd = None
    created = False
    try:
        fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
                     0o644, dir_fd=root_fd)
        created = True
        os.fchmod(fd, 0o644)
        os.fchown(fd, 0, 0)
        written = 0
        while written < len(body):
            written += os.write(fd, body[written:])
        os.fsync(fd)
        os.close(fd)
        fd = None
        os.replace(temporary, STORAGE_USAGE_FILE,
                   src_dir_fd=root_fd, dst_dir_fd=root_fd)
        created = False
    except OSError as exc:
        raise StorageError('cannot record storage usage: %s' % exc)
    finally:
        if fd is not None:
            os.close(fd)
        if created:
            try:
                os.unlink(temporary, dir_fd=root_fd)
            except OSError:
                pass
        os.close(root_fd)

def refresh_usage(path):
    """Re-measure and record a storage directory's size, every time.

This runs after the account already has its answer. The soft threshold is
    deliberately a post-fact guardrail rather than a synchronous filesystem
    quota: a delivery reads this last complete snapshot and the next deferred
    refresh advances it."""
    usage = measure_storage_usage(path)
    write_usage(path, usage)
    return usage['used_bytes']

def over_soft_storage_limit(path, soft_storage_limit_gb):
    """Whether recorded usage has reached the configured soft threshold.

Reads the recorded size, never walks: deferred refresh can lag admissions
    (the trees that tip it over still land), which is fine for a guardrail
    against a shared filesystem, not a billing quota. A threshold of 0
    disables the check entirely. Missing or malformed
    accounting fails closed. A snapshot that stopped at a measurement bound
    counted part of the store, so its number is a floor rather than a size --
    and a floor is compared before it is doubted. Over the threshold it is
    conclusive: 300 GB counted is 300 GB there, whatever the rest holds.
    Below it, it says nothing about the whole, and an advisory guardrail
    must not turn a partial count into a server-wide refusal. Checking
    staleness first switched the guardrail off permanently for exactly the
    store that outgrew its own accounting, because every later walk stops at
    the same bound. Administrator operations create a fresh snapshot
    before enabling a non-zero threshold, and delivery refreshes it after Bun
    has changed the store. This is intentionally not a hard quota: a delivery
    that crosses the threshold and already-admitted concurrent deliveries may
    exceed it."""
    if not soft_storage_limit_gb:
        return False
    usage = read_usage(path)
    if usage is None:
        raise StorageError(
            'shared storage usage snapshot is unavailable for %s; run '
            'cl-node-modules-storage prune before retrying' % path)
    # A floor at or above the threshold is definitive whether or not the
    # snapshot is complete; below it, a partial count proves nothing about
    # the rest and a complete one is a plain no.
    return usage['used_bytes'] >= soft_storage_limit_gb * (1 << 30)

def _cpusers():
    """The panel's hosting accounts. Imported here rather than at module
    scope because `read_config` runs as the account inside a cage on every
    `npm install`: importing cpapi costs a panel probe (0.17 s measured)
    that has no business happening there."""
    from clcommon.cpapi import cpusers
    return cpusers()

def home_device_coverage(storage_paths):
    """(total_accounts, sorted uncovered account names): hosting accounts
    whose home lives on a filesystem no storage directory sits on. Hardlinks
    do not cross filesystems, so those accounts cannot be delivered to.

The accounts are the panel's, not passwd's. A uid-floor sweep of
    `getpwall` counts every service account the distribution ships (nobody,
    chrony, redis, polkitd...), whose homes sit under /var -- so on any
    server whose homes are a separate filesystem it reported two dozen
    accounts to exclude before the feature could be switched on at all.
    """
    devices = set()
    for path in storage_paths:
        try:
            devices.add(os.stat(path).st_dev)
        except OSError:
            continue
    try:
        accounts = _cpusers()
    except Exception as exc:  # noqa: BLE001 - cpapi raises panel-specific errors
        # Never a silent fallback to a passwd scan: an empty or invented
        # account list reads as "everyone is covered" and would enable the
        # feature on a server nobody could actually check.
        raise StorageError('cannot list hosting accounts: %s' % exc)
    total = 0
    uncovered = []
    for name in accounts:
        try:
            account = pwd.getpwnam(name)
        except KeyError:
            continue
        try:
            home_info = os.lstat(account.pw_dir)
        except OSError:
            continue
        total += 1
        # Delivery pins the account home with O_NOFOLLOW and checks its owner.
        # Admission must apply the same boundary: following a home symlink here
        # would promise delivery for an account that the real operation always
        # refuses. Such an account remains explicitly excludable by the admin.
        if (not stat.S_ISDIR(home_info.st_mode)
                or home_info.st_uid != account.pw_uid):
            uncovered.append(name)
            continue
        device = home_info.st_dev
        if device not in devices:
            uncovered.append(name)
    return total, sorted(uncovered)

def update_config(path=CONFIG_PATH, enabled=None, storage_paths=None,
                  excluded_users=None, limits=None, allowed_registries=None):
    """Atomically update known keys, preserving valid unknown keys.

The existing config is opened with ``O_NOFOLLOW``; a symlink is refused.
    An existing file that does not satisfy the reader contract is preserved
    and refused rather than replaced with defaults. The private exclusion
    authority is read and rewritten only by an explicit exclusion update.
    The replacement is a root-owned, mode-0644 temp file in the same directory,
    fsynced and atomically renamed into place. ``storage_paths`` is always
    written as a list; a legacy single ``storage_path`` key is dropped so the
    two can never disagree.
    """
    if os.path.lexists(path) and os.path.islink(path):
        raise StorageError('configuration path must not be a symlink')
    parent = os.path.dirname(path)
    try:
        _reject_symlink_ancestors(parent)
        os.makedirs(parent, mode=0o755, exist_ok=True)
        parent_info = os.lstat(parent)
    except OSError as exc:
        raise StorageError('configuration directory is not usable: %s' % exc)
    if stat.S_ISLNK(parent_info.st_mode) or not stat.S_ISDIR(parent_info.st_mode):
        raise StorageError('configuration directory must be a real directory')
    raw = _read_raw(path)
    if raw is None:
        if os.path.lexists(path):
            raise StorageError('configuration is unreadable')
        raw = {}
    if storage_paths is not None:
        if not storage_paths:
            raise StorageError('at least one storage path is required')
        for storage_path in storage_paths:
            validate_storage_path(storage_path)
        raw['storage_paths'] = list(storage_paths)
        raw['storage_devices'] = storage_device_ids(storage_paths)
        raw.pop('storage_path', None)
    if enabled is not None:
        raw['enabled'] = bool(enabled)
    target_excluded = []
    write_exclusions = excluded_users is not None
    if write_exclusions:
        _, exclusions_valid = _exclusions(path, raw)
        if not exclusions_valid:
            raise StorageError('account exclusion policy is unreadable')
        target_excluded = sorted(excluded_users)
        raw.pop('excluded_users', None)
    if limits is not None:
        validate_limits(limits)
        merged = _limits(raw)
        merged.update(limits)
        raw['limits'] = merged
    if allowed_registries is not None:
        raw['allowed_registries'] = list(allowed_registries)
    _write_config_and_exclusions(
        path, raw, target_excluded, write_exclusions=write_exclusions)

def _encoded_config(raw):
    encoded = (json.dumps(raw, sort_keys=True, separators=(',', ':'))
               + '\n').encode('utf-8')
    if len(encoded) > MAX_CONFIG_BYTES:
        raise StorageError(
            'configuration exceeds the %d byte limit' % MAX_CONFIG_BYTES)
    return encoded

def _write_config_and_exclusions(path, raw, excluded_users,
                                 write_exclusions=False):
    """Publish a requested private authority, then the public snapshot."""
    public_encoded = _encoded_config(raw)
    if write_exclusions:
        private = {'excluded_users': list(excluded_users)}
        private_encoded = _encoded_config(private)
        _atomic_write(exclusions_path(path), private,
                      mode=EXCLUSIONS_MODE, encoded=private_encoded)
    _atomic_write(path, raw, encoded=public_encoded)

def _atomic_write(path, raw, mode=CONFIG_MODE, encoded=None):
    directory = os.path.dirname(path)
    if encoded is None:
        encoded = _encoded_config(raw)
    fd = None
    temporary = None
    try:
        fd, temporary = tempfile.mkstemp(dir=directory, prefix='.node-modules-storage-', text=False)
        os.fchmod(fd, mode)
        os.fchown(fd, 0, 0)
        written = 0
        while written < len(encoded):
            written += os.write(fd, encoded[written:])
        os.fsync(fd)
        os.close(fd)
        fd = None
        os.replace(temporary, path)
        directory_fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)
        temporary = None
    except OSError as exc:
        raise StorageError('cannot write configuration: %s' % exc)
    finally:
        if fd is not None:
            os.close(fd)
        if temporary is not None:
            try:
                os.unlink(temporary)
            except OSError:
                pass

Sid Gifari From Gifari Industries - BD Cyber Security Team