Failed to install: EPERM: operation not permitted, rename
'C:\Users\<user>\.claude\plugins\cache\<plugin>' ->
'C:\Users\<user>\.claude\plugins\cache\<marketplace>\<plugin>\<version>'

---

Add-MpPreference -ExclusionPath "$env:USERPROFILE\.claude\plugins"

---

#!/usr/bin/env python3
"""
Rescue a plugin install that EPERM-failed during the rename to its versioned
slot. Re-fetches the plugin source directly from the marketplace catalog,
places it in the correct cache slot using copy (not rename), and registers
the install in ~/.claude/plugins/installed_plugins.json.

SECURITY MODEL
--------------
Every destructive operation (rmtree, copytree-from-untrusted) is gated by
two independent checks:

  (1) Input validation rejects path-unsafe segment values from CLI args
      and untrusted JSON content (plugin/marketplace/version names).
  (2) safe_rmtree / containment guards refuse to operate on any resolved
      path that does not live strictly under the expected root directory.

Both must pass for the operation to proceed. A bug in one is mitigated by
the other.

Usage:
  python rescue_install.py <plugin>@<marketplace>             # dry-run
  python rescue_install.py <plugin>@<marketplace> --apply     # execute
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime, timezone

home = os.path.expanduser('~')
plugins_root = os.path.join(home, '.claude', 'plugins')
marketplaces_root = os.path.join(plugins_root, 'marketplaces')
cache_root = os.path.join(plugins_root, 'cache')
installed_file = os.path.join(plugins_root, 'installed_plugins.json')


# HARDENING (1): segment validation.
def _validate_segment(value, kind):
    if not isinstance(value, str) or not value:
        raise ValueError(f'{kind} must be a non-empty string')
    if value in ('.', '..'):
        raise ValueError(f'{kind} {value!r} is a parent-directory reference')
    if '/' in value or '\\' in value:
        raise ValueError(f'{kind} {value!r} contains a path separator')
    if os.path.isabs(value):
        raise ValueError(f'{kind} {value!r} is an absolute path')
    if value.startswith('.'):
        raise ValueError(f'{kind} {value!r} starts with a dot')
    return value


# HARDENING (2): containment check.
def _safe_under(path, root):
    real_path = os.path.realpath(path)
    real_root = os.path.realpath(root)
    try:
        common = os.path.commonpath([real_path, real_root])
    except ValueError:
        raise RuntimeError(f'path {path!r} not under {root!r} (different roots)')
    if common != real_root:
        raise RuntimeError(f'refusing op on {real_path!r}; not under {real_root!r}')
    return real_path


# HARDENING (3): rmtree wrapper.
def safe_rmtree(path, root):
    if not os.path.isdir(path):
        return
    _safe_under(path, root)
    shutil.rmtree(path)


def parse_args(argv):
    apply_changes = '--apply' in argv
    positional = [a for a in argv[1:] if not a.startswith('--')]
    if len(positional) != 1 or '@' not in positional[0]:
        print('usage: rescue_install.py <plugin>@<marketplace> [--apply]')
        sys.exit(2)
    plugin_name, marketplace = positional[0].split('@', 1)
    _validate_segment(plugin_name, 'plugin name')
    _validate_segment(marketplace, 'marketplace name')
    return plugin_name, marketplace, apply_changes


def read_marketplace_catalog(marketplace):
    path = os.path.join(marketplaces_root, marketplace, '.claude-plugin', 'marketplace.json')
    if not os.path.isfile(path):
        print(f'marketplace not registered locally: {path}')
        sys.exit(1)
    with open(path) as f:
        return json.load(f), path


def find_plugin_entry(catalog, plugin_name):
    for entry in catalog.get('plugins', []):
        if entry.get('name') == plugin_name:
            return entry
    return None


def run(cmd, cwd=None):
    proc = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
    if proc.returncode != 0:
        raise RuntimeError(f'{" ".join(cmd)} failed:\n{proc.stderr}')
    return proc.stdout


def fetch_source_to(entry, marketplace, dest_dir):
    source = entry.get('source')

    if isinstance(source, str):
        marketplace_root = os.path.join(marketplaces_root, marketplace)
        src = os.path.normpath(os.path.join(marketplace_root, source))
        # HARDENING (4): source-path containment.
        _safe_under(src, marketplace_root)
        if not os.path.isdir(src):
            raise RuntimeError(f'relative source not found: {src}')
        shutil.copytree(src, dest_dir, ignore=shutil.ignore_patterns('.git'))
        return _version_from(entry, dest_dir)

    stype = source.get('source')
    if stype == 'github':
        url = f'https://github.com/{source["repo"]}.git'
        ref = source.get('ref')
        with tempfile.TemporaryDirectory(prefix='rescue_') as td:
            clone_to = os.path.join(td, 'clone')
            args = ['git', 'clone', '--depth', '1']
            if ref:
                args += ['--branch', ref]
            args += [url, clone_to]
            run(args)
            shutil.copytree(clone_to, dest_dir, ignore=shutil.ignore_patterns('.git'))
        return _version_from(entry, dest_dir)

    if stype == 'git-subdir':
        url = source['url']
        path_in_repo = source['path']
        ref = source.get('ref')
        with tempfile.TemporaryDirectory(prefix='rescue_') as td:
            clone_to = os.path.join(td, 'clone')
            args = ['git', 'clone', '--depth', '1', '--filter=blob:none', '--sparse']
            if ref:
                args += ['--branch', ref]
            args += [url, clone_to]
            run(args)
            run(['git', 'sparse-checkout', 'set', path_in_repo], cwd=clone_to)
            sub = os.path.join(clone_to, path_in_repo)
            if not os.path.isdir(sub):
                raise RuntimeError(f'subdir {path_in_repo} not present after sparse checkout')
            shutil.copytree(sub, dest_dir, ignore=shutil.ignore_patterns('.git'))
        return _version_from(entry, dest_dir)

    raise RuntimeError(f'unsupported source type: {stype!r}')


def _version_from(entry, plugin_dir):
    pj = os.path.join(plugin_dir, '.claude-plugin', 'plugin.json')
    if os.path.isfile(pj):
        try:
            with open(pj) as f:
                v = json.load(f).get('version')
                if v:
                    _validate_segment(v, 'version (from plugin.json)')
                    return v
        except ValueError:
            raise
        except Exception:
            pass
    fallback = entry.get('version') or 'unknown'
    _validate_segment(fallback, 'version (from marketplace.json or fallback)')
    return fallback


def register_install(plugin_name, marketplace, install_path, version):
    try:
        with open(installed_file) as f:
            data = json.load(f)
    except FileNotFoundError:
        data = {'version': 2, 'plugins': {}}
    if 'plugins' not in data:
        data['plugins'] = {}

    now = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%fZ')[:-4] + 'Z'
    key = f'{plugin_name}@{marketplace}'
    record = {
        'scope': 'user',
        'installPath': install_path,
        'version': version,
        'installedAt': now,
        'lastUpdated': now,
    }
    existing = data['plugins'].get(key, [])
    existing = [r for r in existing if r.get('scope') != 'user']
    existing.append(record)
    data['plugins'][key] = existing

    with open(installed_file, 'w') as f:
        json.dump(data, f, indent=2)


def main(argv):
    plugin_name, marketplace, apply_changes = parse_args(argv)

    catalog, catalog_path = read_marketplace_catalog(marketplace)
    entry = find_plugin_entry(catalog, plugin_name)
    if entry is None:
        print(f'plugin {plugin_name!r} not found in {catalog_path}')
        return 1

    declared_version = entry.get('version')
    if declared_version is not None:
        _validate_segment(declared_version, 'version (from marketplace entry)')

    version = declared_version or '<from plugin.json after fetch>'
    target = os.path.join(cache_root, marketplace, plugin_name, version)
    print(f'plan: rescue {plugin_name}@{marketplace}')
    print(f'  source: {entry.get("source")}')
    print(f'  target: {target}')

    if not apply_changes:
        print('\nRe-run with --apply to fetch and install.')
        return 0

    orphan = os.path.join(cache_root, plugin_name)
    if os.path.isdir(orphan):
        contents = os.listdir(orphan)
        if any(os.path.isfile(os.path.join(orphan, c)) for c in contents):
            print(f'  removing orphan extract: {orphan}')
            safe_rmtree(orphan, cache_root)

    if os.path.isdir(target):
        print(f'  removing existing target: {target}')
        safe_rmtree(target, cache_root)

    os.makedirs(os.path.dirname(target), exist_ok=True)

    fetch_temp = target + '.fetching'
    if os.path.exists(fetch_temp):
        safe_rmtree(fetch_temp, cache_root)
    try:
        version = fetch_source_to(entry, marketplace, fetch_temp)
        retargeted = os.path.join(cache_root, marketplace, plugin_name, version)
        if retargeted != target:
            os.makedirs(os.path.dirname(retargeted), exist_ok=True)
            target = retargeted
        if os.path.isdir(target):
            safe_rmtree(target, cache_root)
        shutil.copytree(fetch_temp, target)
    finally:
        if os.path.isdir(fetch_temp):
            try:
                safe_rmtree(fetch_temp, cache_root)
            except Exception:
                pass

    register_install(plugin_name, marketplace, target, version)
    print(f'\ninstalled: {plugin_name}@{marketplace} v{version}')
    print(f'  at: {target}')
    print('\nRestart Claude Code so installed_plugins.json is reloaded.')
    return 0


if __name__ == '__main__':
    sys.exit(main(sys.argv))