diff --git a/docs/cli/add.md b/docs/cli/add.md index 0c7777457..ddd775c4b 100644 --- a/docs/cli/add.md +++ b/docs/cli/add.md @@ -92,11 +92,12 @@ change. ## Choosing configuration files -vcspull searches for configuration files in this order: +vcspull resolves configuration from several scopes at once — system, user, +and the `.vcspull.*` files above your working directory. See +{ref}`config-scopes` for the full order. -1. `./.vcspull.yaml` -2. `~/.vcspull.yaml` -3. `~/.config/vcspull/*.yaml` +`add` writes to your user configuration by default, and falls back to +creating `./.vcspull.yaml` when you have none. Specify a file explicitly with `-f/--file`: diff --git a/docs/cli/config.md b/docs/cli/config.md new file mode 100644 index 000000000..5b6e1e238 --- /dev/null +++ b/docs/cli/config.md @@ -0,0 +1,58 @@ +(cli-config)= + +# vcspull config + +vcspull rarely reads just one file. `vcspull config ls` answers the question +that follows from that: standing here, which files are in effect, and which +one wins? Reach for it when `vcspull sync` picks up a repository you did not +expect, or refuses one you did. + +## Command + +```{eval-rst} +.. argparse:: + :module: vcspull.cli + :func: create_parser + :prog: vcspull + :path: config +``` + +## Reading the report + +```vcspull-console +$ vcspull config ls +user ~/.vcspull.yaml 2 repos (1 overridden) +project ~/work/proj/.vcspull.yaml 1 repo + +2 repositories in effect. +``` + +Rows run weakest to strongest, the same order +{ref}`the scopes resolve in `. The count is what each file +declares; "overridden" is how many of those a nearer file replaced. The final +line is the total you will actually sync. + +## Diagnosing a refusal + +A project config that would check a repository out beyond its own directory +needs {ref}`trust ` before it loads. `config ls` shows it rather +than dropping it, and never prompts — this is the command you run *because* +something else refused: + +```vcspull-output +user ~/.vcspull.yaml 2 repos +project ~/work/proj/.vcspull.yaml 1 repo +project ~/work/proj/escaping/.vcspull.yaml 1 repo (untrusted) + +3 repositories in effect. +Untrusted configs are not loaded. Allow one with 'vcspull trust DIR'. +``` + +## Ignoring the project scope + +```console +$ vcspull config ls --no-project +``` + +Both `vcspull --no-project config ls` and `vcspull config ls --no-project` +work. diff --git a/docs/cli/index.md b/docs/cli/index.md index 593c46182..96c147344 100644 --- a/docs/cli/index.md +++ b/docs/cli/index.md @@ -67,6 +67,18 @@ {ref}`vcspull worktree ` manages git worktrees declaratively. ::: +:::{grid-item-card} Config +:link: config +:link-type: doc +{ref}`vcspull config ` shows which config files are in effect. +::: + +:::{grid-item-card} Trust +:link: trust +:link-type: doc +{ref}`vcspull trust ` allows a project config to act outside its directory. +::: + :::{grid-item-card} Completion :link: completion :link-type: doc @@ -89,6 +101,8 @@ status worktree/index fmt migrate +config +trust ``` ```{toctree} @@ -113,5 +127,5 @@ completion :no-description: subparser_name : @replace - See :ref:`cli-sync`, :ref:`cli-add`, :ref:`cli-import`, :ref:`cli-discover`, :ref:`cli-list`, :ref:`cli-search`, :ref:`cli-status`, :ref:`cli-worktree`, :ref:`cli-fmt`, :ref:`cli-migrate` + See :ref:`cli-sync`, :ref:`cli-add`, :ref:`cli-import`, :ref:`cli-discover`, :ref:`cli-list`, :ref:`cli-search`, :ref:`cli-status`, :ref:`cli-worktree`, :ref:`cli-fmt`, :ref:`cli-migrate`, :ref:`cli-config`, :ref:`cli-trust` ``` diff --git a/docs/cli/list.md b/docs/cli/list.md index 74c8ebbbe..e07913e35 100644 --- a/docs/cli/list.md +++ b/docs/cli/list.md @@ -122,10 +122,11 @@ $ vcspull list --ndjson | grep 'study' | jq -r '.name' ## Choosing configuration files -By default, vcspull searches for config files in standard locations -(`~/.vcspull.yaml`, `./.vcspull.yaml`, and +By default, `list` unions every configuration scope in effect: system, your [XDG](https://standards.freedesktop.org/basedir-spec/basedir-spec-latest.html) -config directories). +config directory, `~/.vcspull.yaml`, and any `.vcspull.*` file above your +working directory. See {ref}`config-scopes` for the resolution order, or run +`vcspull config ls` to see what applies where you are standing. Specify a custom config file with `-f/--file`: diff --git a/docs/cli/trust.md b/docs/cli/trust.md new file mode 100644 index 000000000..6062002a5 --- /dev/null +++ b/docs/cli/trust.md @@ -0,0 +1,54 @@ +(cli-trust)= + +# vcspull trust + +A `.vcspull.yaml` inside a repository you cloned is code somebody else wrote, +and it names paths on your disk. vcspull loads such a file silently as long as +every repository it declares lands inside that file's own directory. When one +does not, vcspull asks — and `vcspull trust` is how you answer ahead of time, +or take the answer back. + +Most people never run this command. See {ref}`config-scopes` for when the +question comes up at all. + +## Command + +```{eval-rst} +.. argparse:: + :module: vcspull.cli + :func: create_parser + :prog: vcspull + :path: trust +``` + +## Allowing a project + +```vcspull-console +$ vcspull trust ~/work/api +trusted ~/work/api +``` + +Trust is recorded per *directory*, not per file, so a second config in the +same project does not ask again. The trade-off is real: a configuration +committed to that directory can now direct your checkouts anywhere on disk, +including after somebody else edits it. + +## Reviewing and revoking + +```vcspull-console +$ vcspull trust --show +~/work/api +``` + +```vcspull-console +$ vcspull trust --untrust ~/work/api +untrusted ~/work/api +``` + +The record lives in `$XDG_STATE_HOME/vcspull/trusted`, one directory per line. + +## In automation + +There is no prompt without a terminal — a sync in CI fails loudly rather than +blocking on a question nobody can see. To accept without asking, pass +`--trust-project` or export `VCSPULL_YES=1`. diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 8bb0f72f7..c38bf621c 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -17,6 +17,12 @@ Most users can start with one `~/.vcspull.yaml` file: ::::{grid} 1 1 2 2 :gutter: 2 2 3 3 +:::{grid-item-card} Configuration scopes +:link: scopes +:link-type: doc +Where vcspull looks, in what order, and how project configs are trusted. +::: + :::{grid-item-card} Config Generation :link: generation :link-type: doc @@ -51,6 +57,10 @@ You can place the file in one of three places: 3. Anywhere (and trigger via {ref}`vcspull sync ` with `--file ./path/to/file.yaml [repo_name]`) +vcspull reads all of these, plus a `.vcspull.yaml` in your project, and +resolves them in a fixed order. See {ref}`config-scopes` for that order, how +overlapping entries merge, and when a project config asks before it acts. + [xdg]: https://standards.freedesktop.org/basedir-spec/basedir-spec-latest.html ## Schema @@ -369,5 +379,6 @@ git+ssh://git@github.com/tony/kaptan.git ```{toctree} :hidden: +scopes generation ``` diff --git a/docs/configuration/scopes.md b/docs/configuration/scopes.md new file mode 100644 index 000000000..5901ecd6f --- /dev/null +++ b/docs/configuration/scopes.md @@ -0,0 +1,178 @@ +(config-scopes)= + +# Configuration scopes + +vcspull reads more than one configuration file. Where each file sits decides +how much it can say: a file in `/etc` speaks for the machine, a file in your +home directory speaks for you, and a `.vcspull.yaml` inside a project speaks +for that project. vcspull calls these *scopes*, resolves them in a fixed +order, and unions them into one set of repositories. + +If you keep a single `~/.vcspull.yaml`, that is the whole story and you can +stop reading. Nothing below changes what one file does. + +## The search order + +Weakest to strongest. Every scope that exists contributes, and a repository +named by more than one scope comes from the strongest. + +| Scope | Where | Trust | +| --- | --- | --- | +| system | `/etc/vcspull/*.{yaml,yml,json}` | automatic | +| user directory | `$VCSPULL_CONFIGDIR`, else `$XDG_CONFIG_HOME/vcspull/`, else `~/.config/vcspull/`, else `~/.vcspull/` — the first that exists | automatic | +| user dotfile | `~/.vcspull.yaml` or `~/.vcspull.json` | automatic | +| project | `.vcspull.{yaml,yml,json}` in each ancestor of your working directory, outermost first | gated | + +Passing `-f/--file` replaces that entire stack with the one file you name, so +existing scripts and CI jobs behave exactly as they always have. + +To see what applies where you are standing: + +```vcspull-console +$ vcspull config ls +user ~/.vcspull.yaml 2 repos (1 overridden) +project ~/work/proj/.vcspull.yaml 1 repo + +2 repositories in effect. +``` + +Each row is a file in the stack, weakest first, with the count it declares. +"overridden" is how many of those a nearer file replaced, so the last line is +what you will actually sync. + +## The upward walk + +The project scope walks up from your working directory, so +`~/work/.vcspull.yaml` covers everything under `~/work` — this is what makes +a workspace-of-workspaces manifest usable, and why `vcspull sync` no longer +does nothing from a subdirectory. + +The walk stops at `$HOME` or the filesystem root, whichever comes first. A +`.vcspull.yaml` sitting in `/home` or `/` is far likelier to be an accident +than an intent, and the system scope already covers the machine-wide case. +Override the stop set with `VCSPULL_CEILING_PATHS`, a `:`-separated list of +directories. + +To ignore the project scope entirely for one command: + +```console +$ vcspull --no-project sync --all +``` + +Export `VCSPULL_NO_PROJECT=1` to make that the default for a shell. + +## How the scopes merge + +The identity of a repository is its destination on disk — the workspace root +plus the repository name. Two files that name different destinations union; +a project config adds repositories to your user set. + +When two files name the same destination, the stronger scope replaces the +entry *whole*. The URL, the VCS, and everything under `options:` come from +the winner. That is deliberate rather than convenient: a project config that +repoints `flask` at a fork must not silently inherit the `rev:` pin from your +user config and check out a revision that does not exist in the fork. + +vcspull reports the override only when the two entries actually disagree on +URL or VCS: + +```vcspull-output +~/work/proj/.vcspull.yaml overrides ~/.vcspull.yaml for '~/work/proj/vendor/flask' (url or vcs differs) +``` + +Identical entries across a user and a project file are the common, harmless +case, and stay silent. If every sync warned about them, you would learn to +ignore the channel. + +Replacing the whole entry costs you granularity: a project file cannot tweak +one field of an inherited entry. To change anything about a repository your +user config already names, the project file restates that entry in full. + +## Trusting a project config + +A configuration is a set of destination paths, so a `.vcspull.yaml` inside a +repository you just cloned could name `~/.ssh/` as a workspace root and have +`vcspull sync` clone into it. The project scope is the only scope that can +arrive with untrusted code, so it is the only one vcspull gates. + +The gate is containment: a project config is *contained* when every +repository it declares would be checked out at or beneath the directory +holding that config. The check runs over destinations, not over the workspace +roots you wrote — a repository entry can override its destination with +`path:` and never consult its workspace root at all — and it expands `~`, +environment variables, and symlinks on both sides first. A config targeting +`./vendor/` or a sibling checkout is contained, and loads silently. This is +the overwhelming majority of them, which is what keeps the gate from becoming +a nag. + +A config that escapes its own directory asks once: + +```vcspull-output +! ~/work/proj/escaping/.vcspull.yaml would check repositories out outside its directory: + ~/.ssh/evil + Trust this config? [y/N/always] +``` + +Answering `always` records the project *directory*, not the file, so adding a +second config to a project you already trust does not ask again. The record +lives in `$XDG_STATE_HOME/vcspull/trusted`. + +Without a terminal there is no prompt — vcspull fails with the whole remedy in +one line, because a sync in CI must never block on a question nobody can see: + +```vcspull-output +vcspull: ~/work/proj/escaping/.vcspull.yaml would check repositories out outside its directory (~/.ssh/evil) and there is no terminal to confirm on. Run 'vcspull trust ~/work/proj/escaping' to allow it, pass --trust-project, or use --no-project to skip project configs. +``` + +When you do want to accept non-interactively, pass `--trust-project` or export +`VCSPULL_YES=1`. That switch is deliberately its own word rather than a +general `--yes`: an unrelated confirmation flag must never become a standing +grant to write outside a directory. + +Trusting a directory is not free. It means a configuration committed to that +directory can direct your checkouts anywhere on disk, including next time, +when somebody else has edited it. Trust the project, not just the file you +read. + +The same check guards the commands that write: {ref}`add `, +{ref}`discover `, {ref}`fmt `, and +{ref}`migrate `. A repository that ships a config redirecting +your writes deserves the same question as one redirecting your clones. Naming +a file with `-f/--file` is consent and skips the gate, the same way `--file` +skips the scope stack on the read side. + +Manage the record directly with {ref}`vcspull trust `: + +```console +$ vcspull trust ~/work/api +``` + +```console +$ vcspull trust --untrust ~/work/api +``` + +```console +$ vcspull trust --show +``` + +## For the Python reader + +Everything above is one function. {func}`vcspull.config.load_scoped_configs` +resolves the stack, merges it, and runs the trust gate, returning the same +list of entries the CLI syncs: + +```python +from vcspull.config import load_scoped_configs + +repos = load_scoped_configs() +``` + +Pass `config_path` to replace the stack the way `--file` does, +`include_project=False` for `--no-project`, or `cwd` to resolve as if you were +standing somewhere else. + +Two lower-level pieces are worth knowing if you are building on this. +{func}`vcspull.config.repo_destinations` answers "where would this check +out?" for a list of entries, honouring `path:` overrides — it is what the +containment check reads. {func}`vcspull.config.ensure_config_trusted` is the +gate the write commands call, and returns `False` when the user declines. diff --git a/src/vcspull/_internal/scopes.py b/src/vcspull/_internal/scopes.py new file mode 100644 index 000000000..e53c328d8 --- /dev/null +++ b/src/vcspull/_internal/scopes.py @@ -0,0 +1,642 @@ +"""Configuration scope resolution and project trust. + +vcspull reads configuration from four scopes. Weakest first, every scope is +unioned into one stack: + +============ ========================================================= +Scope Location +============ ========================================================= +``system`` ``/etc/vcspull/*.{yaml,yml,json}`` +``user`` the user config directory, then ``~/.vcspull.{yaml,json}`` +``project`` ``.vcspull.{yaml,yml,json}`` in each ancestor of the + working directory, outermost first +============ ========================================================= + +Everything here is a resolver: it decides *which* files participate and +*whether* a project file may act, never what a file means. Nothing in this +module parses repository entries, and nothing prompts — a caller that needs +consent gets the list of offending destinations from :func:`requires_trust` +and decides how to ask. +""" + +from __future__ import annotations + +import os +import pathlib +import typing as t + +from ..util import get_config_dir + +if t.TYPE_CHECKING: + from collections.abc import Iterable, Mapping + +ConfigScope = t.Literal["system", "user", "project", "external"] +"""Where a configuration file lives, and therefore how much it is trusted.""" + +CONFIG_SUFFIXES: t.Final = (".yaml", ".yml", ".json") +"""Recognized config suffixes, weakest first within a single directory.""" + +SYSTEM_CONFIG_DIR: t.Final = pathlib.Path("/etc/vcspull") +"""Machine-wide configuration directory.""" + +PROJECT_CONFIG_STEM: t.Final = ".vcspull" +"""Basename of a project configuration, before its suffix.""" + +_TRUTHY: t.Final = frozenset({"1", "true", "yes", "on"}) + + +def _real(path: pathlib.Path) -> pathlib.Path: + """Return *path* with ``~`` expanded and every symlink followed. + + Both sides of a containment or ceiling comparison go through this, so a + symlinked ``$HOME`` stops the project walk and a symlinked workspace root + cannot smuggle a destination back out of its directory. + + Examples + -------- + >>> link = tmp_path / "link" + >>> link.symlink_to(tmp_path, target_is_directory=True) + >>> _real(link / "sub") == _real(tmp_path / "sub") + True + """ + return pathlib.Path(os.path.realpath(path.expanduser())) + + +class ConfigSource(t.NamedTuple): + """One configuration file in the resolution stack. + + Sources are ordered weakest to strongest, so a later source overrides an + earlier one for any repository they both name. + """ + + scope: ConfigScope + """Scope this file was resolved from.""" + + path: pathlib.Path + """Absolute path to the configuration file.""" + + trust_root: pathlib.Path + """Directory whose trust governs this file. Its own directory, for a + project config; the containing scope directory otherwise.""" + + @property + def gated(self) -> bool: + """Return ``True`` when this source needs a containment check. + + Examples + -------- + >>> project = ConfigSource( + ... "project", pathlib.Path("/w/.vcspull.yaml"), pathlib.Path("/w") + ... ) + >>> project.gated + True + >>> ConfigSource( + ... "user", pathlib.Path("/h/.vcspull.yaml"), pathlib.Path("/h") + ... ).gated + False + """ + return self.scope == "project" + + +def env_flag(name: str, environ: Mapping[str, str] = os.environ) -> bool: + """Return whether an environment variable is set to a truthy word. + + Parameters + ---------- + name : str + Environment variable to read. + environ : Mapping[str, str] + Environment to read from. Defaults to :data:`os.environ`. + + Examples + -------- + >>> env_flag("VCSPULL_YES", {"VCSPULL_YES": "1"}) + True + >>> env_flag("VCSPULL_YES", {"VCSPULL_YES": "0"}) + False + >>> env_flag("VCSPULL_YES", {}) + False + """ + return environ.get(name, "").strip().lower() in _TRUTHY + + +def ceiling_paths( + environ: Mapping[str, str] = os.environ, + *, + home: pathlib.Path, +) -> frozenset[pathlib.Path]: + """Return the directories the upward walk stops at. + + ``VCSPULL_CEILING_PATHS`` replaces the default stop set. Setting it to an + empty value lets the walk run to the filesystem root. Every entry is + resolved, so a ceiling reached through a symlink still stops the walk. + + Parameters + ---------- + environ : Mapping[str, str] + Environment to read from. Defaults to :data:`os.environ`. + home : pathlib.Path + User home directory, the default ceiling. + + Examples + -------- + >>> ceiling_paths({}, home=tmp_path) == frozenset({_real(tmp_path)}) + True + >>> sorted( + ... ceiling_paths( + ... {"VCSPULL_CEILING_PATHS": f"/srv{os.pathsep}/opt"}, + ... home=tmp_path, + ... ) + ... ) + [PosixPath('/opt'), PosixPath('/srv')] + >>> ceiling_paths({"VCSPULL_CEILING_PATHS": ""}, home=tmp_path) + frozenset() + """ + raw = environ.get("VCSPULL_CEILING_PATHS") + if raw is None: + return frozenset({_real(home)}) + return frozenset( + _real(pathlib.Path(entry)) for entry in raw.split(os.pathsep) if entry + ) + + +def project_dirs( + cwd: pathlib.Path, + *, + ceilings: frozenset[pathlib.Path], +) -> tuple[pathlib.Path, ...]: + """Return the ancestors of *cwd* that may hold a project config. + + Outermost first, so the returned order is weakest to strongest. A ceiling + directory stops the walk and is itself excluded; the filesystem root ends + it either way. + + Parameters + ---------- + cwd : pathlib.Path + Directory to walk up from. + ceilings : frozenset of pathlib.Path + Resolved directories to stop at, from :func:`ceiling_paths`. + + Examples + -------- + >>> home = pathlib.Path("/home/u") + >>> project_dirs(pathlib.Path("/home/u/work/api"), ceilings=frozenset({home})) + (PosixPath('/home/u/work'), PosixPath('/home/u/work/api')) + + Standing on the ceiling yields nothing: + + >>> project_dirs(home, ceilings=frozenset({home})) + () + + Outside the ceiling, the walk runs to the root: + + >>> project_dirs(pathlib.Path("/srv/app"), ceilings=frozenset({home})) + (PosixPath('/'), PosixPath('/srv'), PosixPath('/srv/app')) + """ + walked: list[pathlib.Path] = [] + for directory in (cwd, *cwd.parents): + if _real(directory) in ceilings: + break + walked.append(directory) + walked.reverse() + return tuple(walked) + + +def _config_files_in(directory: pathlib.Path) -> tuple[pathlib.Path, ...]: + """Return sorted config files directly inside *directory*. + + A directory that cannot be listed, or one holding a *directory* named + ``foo.yaml``, contributes nothing rather than raising. + + Examples + -------- + >>> scope_dir = tmp_path / "scope" + >>> (scope_dir / "not-a-file.yaml").mkdir(parents=True) + >>> _ = (scope_dir / "b.json").write_text("{}", encoding="utf-8") + >>> _ = (scope_dir / "a.yaml").write_text("", encoding="utf-8") + >>> [path.name for path in _config_files_in(scope_dir)] + ['a.yaml', 'b.json'] + >>> _config_files_in(tmp_path / "missing") + () + """ + try: + entries = sorted(directory.iterdir()) + except OSError: + return () + return tuple( + entry + for entry in entries + if entry.suffix in CONFIG_SUFFIXES and entry.is_file() + ) + + +def _project_files_in(directory: pathlib.Path) -> tuple[pathlib.Path, ...]: + """Return the project configs directly inside *directory*. + + Examples + -------- + >>> _ = (tmp_path / ".vcspull.yaml").write_text("", encoding="utf-8") + >>> _ = (tmp_path / ".vcspull.json").write_text("{}", encoding="utf-8") + >>> [path.name for path in _project_files_in(tmp_path)] + ['.vcspull.yaml', '.vcspull.json'] + """ + found: list[pathlib.Path] = [] + for suffix in CONFIG_SUFFIXES: + candidate = directory / f"{PROJECT_CONFIG_STEM}{suffix}" + if candidate.is_file(): + found.append(candidate) + return tuple(found) + + +def resolve_sources( + *, + cwd: pathlib.Path, + home: pathlib.Path | None = None, + environ: Mapping[str, str] = os.environ, + include_project: bool = True, +) -> tuple[ConfigSource, ...]: + """Return the configuration stack in effect, weakest source first. + + Parameters + ---------- + cwd : pathlib.Path + Working directory the project walk starts from. + home : pathlib.Path, optional + User home directory, the default project-walk ceiling. Defaults to + :meth:`pathlib.Path.home`. + environ : Mapping[str, str] + Environment to read from. Defaults to :data:`os.environ`. + include_project : bool + Set ``False`` (or export ``VCSPULL_NO_PROJECT=1``) to drop the project + scope and resolve only system and user configuration. + + Returns + ------- + tuple of ConfigSource + Existing files only, ordered weakest to strongest. + + Raises + ------ + vcspull.exc.MultipleConfigWarning + When both ``~/.vcspull.yaml`` and ``~/.vcspull.json`` exist. + + Examples + -------- + A project config in the working directory joins the stack: + + >>> _ = (tmp_path / ".vcspull.yaml").write_text("", encoding="utf-8") + >>> [source.scope for source in resolve_sources(cwd=tmp_path, home=tmp_path.parent)] + ['project'] + + ``VCSPULL_NO_PROJECT`` drops it: + + >>> resolve_sources( + ... cwd=tmp_path, + ... home=tmp_path.parent, + ... environ={"VCSPULL_NO_PROJECT": "1"}, + ... ) + () + """ + from ..config import find_home_config_files + + home = (home or pathlib.Path.home()).expanduser() + sources: list[ConfigSource] = [] + + sources.extend( + ConfigSource("system", path, SYSTEM_CONFIG_DIR) + for path in _config_files_in(SYSTEM_CONFIG_DIR) + ) + + user_dir = get_config_dir() + sources.extend( + ConfigSource("user", path, user_dir) for path in _config_files_in(user_dir) + ) + + # Delegated rather than reimplemented so that owning two home dotfiles + # still raises MultipleConfigWarning, as it did before scopes existed. + sources.extend( + ConfigSource("user", path, path.parent) for path in find_home_config_files() + ) + + if include_project and not env_flag("VCSPULL_NO_PROJECT", environ): + ceilings = ceiling_paths(environ, home=home) + for directory in project_dirs(cwd, ceilings=ceilings): + sources.extend( + ConfigSource("project", path, directory) + for path in _project_files_in(directory) + ) + + return tuple(sources) + + +def classify_scope( + config_path: pathlib.Path, + *, + cwd: pathlib.Path, + home: pathlib.Path, + environ: Mapping[str, str] = os.environ, +) -> ConfigScope: + """Return the scope a single configuration file belongs to. + + Unlike :func:`resolve_sources`, this answers the question for a path the + caller already holds — a ``--file`` argument, or the file a write command + is about to touch. A path that is neither user, system, nor inside the + working directory is ``external``. + + Parameters + ---------- + config_path : pathlib.Path + Configuration file to classify. + cwd : pathlib.Path + Working directory. + home : pathlib.Path + User home directory. + environ : Mapping[str, str] + Environment to read from. Defaults to :data:`os.environ`. + + Examples + -------- + >>> project = tmp_path / "proj" + >>> project.mkdir() + >>> classify_scope( + ... project / ".vcspull.yaml", cwd=project, home=tmp_path / "home" + ... ) + 'project' + >>> classify_scope( + ... tmp_path / "home" / ".vcspull.yaml", + ... cwd=project, + ... home=tmp_path / "home", + ... ) + 'user' + >>> classify_scope( + ... tmp_path / "elsewhere.yaml", cwd=project, home=tmp_path / "home" + ... ) + 'external' + """ + resolved = _real(config_path) + home = _real(home) + cwd = _real(cwd) + + if resolved.parent == home and resolved.name in { + f"{PROJECT_CONFIG_STEM}{suffix}" for suffix in CONFIG_SUFFIXES + }: + return "user" + + xdg_config_home = _real( + pathlib.Path(environ.get("XDG_CONFIG_HOME", home / ".config")), + ) + if _is_within(resolved, xdg_config_home / "vcspull"): + return "user" + + xdg_config_dirs = environ.get("XDG_CONFIG_DIRS") + system_bases = [SYSTEM_CONFIG_DIR] + if xdg_config_dirs: + system_bases.extend( + pathlib.Path(entry) / "vcspull" + for entry in xdg_config_dirs.split(os.pathsep) + if entry + ) + else: + system_bases.append(pathlib.Path("/etc/xdg/vcspull")) + for base in system_bases: + if _is_within(resolved, _real(base)): + return "system" + + return "project" if _is_within(resolved, cwd) else "external" + + +def _is_within(path: pathlib.Path, directory: pathlib.Path) -> bool: + """Return whether *path* is *directory* or lives beneath it. + + Examples + -------- + >>> _is_within(pathlib.Path("/w/vendor/x"), pathlib.Path("/w")) + True + >>> _is_within(pathlib.Path("/w"), pathlib.Path("/w")) + True + >>> _is_within(pathlib.Path("/workspace"), pathlib.Path("/w")) + False + """ + return path == directory or directory in path.parents + + +def escaping_destinations( + destinations: Iterable[pathlib.Path], + *, + config_dir: pathlib.Path, + cwd: pathlib.Path, +) -> tuple[pathlib.Path, ...]: + """Return the repository destinations that resolve outside *config_dir*. + + A configuration is *contained* when every repository it declares would be + checked out at or beneath the directory holding it. Containment is what + keeps the trust gate quiet for the overwhelming majority of project + configs, which target ``./vendor`` or a sibling checkout. + + The check is over destinations rather than the workspace-root keys, because + a repository entry may override its destination with ``path:`` and never + consult its workspace root at all. Symlinks are followed on both sides, so + a destination that points out of the tree through a link is still reported. + + Parameters + ---------- + destinations : Iterable[pathlib.Path] + Checkout destinations, as :func:`vcspull.config.extract_repos` computes + them. + config_dir : pathlib.Path + Directory holding the configuration file. + cwd : pathlib.Path + Working directory, used to resolve relative destinations. + + Returns + ------- + tuple of pathlib.Path + Offending destinations, resolved, in order and without duplicates. + Empty when the configuration is contained. + + Examples + -------- + >>> project = tmp_path / "project" + >>> project.mkdir() + >>> escaping_destinations( + ... [project / "vendor" / "flask"], config_dir=project, cwd=project + ... ) + () + >>> escaping_destinations( + ... [pathlib.Path("~/.ssh/evil")], config_dir=project, cwd=project + ... ) == (_real(pathlib.Path("~/.ssh/evil")),) + True + """ + root = _real(config_dir) + escaping: list[pathlib.Path] = [] + + for destination in destinations: + target = pathlib.Path(os.path.expandvars(str(destination))) + if not target.expanduser().is_absolute(): + target = cwd / target + resolved = _real(target) + if not _is_within(resolved, root): + escaping.append(resolved) + + return tuple(dict.fromkeys(escaping)) + + +def requires_trust( + source: ConfigSource, + destinations: Iterable[pathlib.Path], + *, + cwd: pathlib.Path, + trusted: frozenset[pathlib.Path], +) -> tuple[pathlib.Path, ...]: + """Return the destinations that need explicit consent before loading. + + Empty means the source may load silently: system, user, and explicit + configurations are automatic, an already-trusted project directory is + remembered, and a contained project config never asks. + + Parameters + ---------- + source : ConfigSource + Source under consideration. + destinations : Iterable[pathlib.Path] + Checkout destinations that file declares. + cwd : pathlib.Path + Working directory, used to resolve relative destinations. + trusted : frozenset of pathlib.Path + Resolved directories from :func:`read_trusted`. + + Examples + -------- + >>> project = tmp_path / "project" + >>> project.mkdir() + >>> source = ConfigSource("project", project / ".vcspull.yaml", project) + >>> escaping = [pathlib.Path("~/.ssh/evil")] + >>> requires_trust( + ... source, escaping, cwd=project, trusted=frozenset() + ... ) == (_real(pathlib.Path("~/.ssh/evil")),) + True + + A user-scope source is never gated, and neither is a trusted directory: + + >>> user = ConfigSource("user", tmp_path / "u.yaml", tmp_path) + >>> requires_trust(user, escaping, cwd=project, trusted=frozenset()) + () + >>> requires_trust( + ... source, escaping, cwd=project, trusted=frozenset({_real(project)}) + ... ) + () + """ + if not source.gated: + return () + if _real(source.trust_root) in trusted: + return () + return escaping_destinations( + destinations, + config_dir=source.trust_root, + cwd=cwd, + ) + + +def trust_state_file( + *, + home: pathlib.Path | None = None, + environ: Mapping[str, str] = os.environ, +) -> pathlib.Path: + """Return the file recording trusted project directories. + + Parameters + ---------- + home : pathlib.Path, optional + User home directory. Defaults to :meth:`pathlib.Path.home`. + environ : Mapping[str, str] + Environment to read from. Defaults to :data:`os.environ`. + + Examples + -------- + >>> trust_state_file( + ... home=pathlib.Path("/home/u"), environ={"XDG_STATE_HOME": "/state"} + ... ) + PosixPath('/state/vcspull/trusted') + >>> trust_state_file(home=pathlib.Path("/home/u"), environ={}) + PosixPath('/home/u/.local/state/vcspull/trusted') + """ + home = (home or pathlib.Path.home()).expanduser() + state_home = environ.get("XDG_STATE_HOME") + base = ( + pathlib.Path(state_home).expanduser() + if state_home + else home / ".local" / "state" + ) + return base / "vcspull" / "trusted" + + +def read_trusted(state_file: pathlib.Path) -> frozenset[pathlib.Path]: + """Return the trusted project directories recorded in *state_file*. + + A missing or unreadable state file means nothing is trusted yet. + + Examples + -------- + >>> read_trusted(tmp_path / "never-written") + frozenset() + >>> state = tmp_path / "trusted" + >>> write_trusted(state, [pathlib.Path("/w/api")]) + >>> read_trusted(state) + frozenset({PosixPath('/w/api')}) + """ + try: + content = state_file.read_text(encoding="utf-8") + except OSError: + return frozenset() + return frozenset( + pathlib.Path(line.strip()) for line in content.splitlines() if line.strip() + ) + + +def write_trusted( + state_file: pathlib.Path, + directories: Iterable[pathlib.Path], +) -> None: + """Record *directories* as the complete set of trusted project directories. + + Examples + -------- + >>> state = tmp_path / "state" / "trusted" + >>> write_trusted(state, [pathlib.Path("/w/b"), pathlib.Path("/w/a")]) + >>> state.read_text(encoding="utf-8").splitlines() + ['/w/a', '/w/b'] + """ + state_file.parent.mkdir(parents=True, exist_ok=True) + lines = sorted(str(directory) for directory in directories) + state_file.write_text("".join(f"{line}\n" for line in lines), encoding="utf-8") + + +def trust_directory(state_file: pathlib.Path, directory: pathlib.Path) -> None: + """Add *directory* to the trusted set. + + The project directory is recorded rather than the file, so a second config + in an already-trusted project does not ask again. + + Examples + -------- + >>> state = tmp_path / "trusted" + >>> trust_directory(state, tmp_path / "api") + >>> _real(tmp_path / "api") in read_trusted(state) + True + """ + write_trusted(state_file, read_trusted(state_file) | {_real(directory)}) + + +def untrust_directory(state_file: pathlib.Path, directory: pathlib.Path) -> None: + """Remove *directory* from the trusted set. + + Examples + -------- + >>> state = tmp_path / "trusted" + >>> trust_directory(state, tmp_path / "api") + >>> untrust_directory(state, tmp_path / "api") + >>> read_trusted(state) + frozenset() + """ + write_trusted(state_file, read_trusted(state_file) - {_real(directory)}) diff --git a/src/vcspull/cli/__init__.py b/src/vcspull/cli/__init__.py index b595a7e47..6b1eb8ecb 100644 --- a/src/vcspull/cli/__init__.py +++ b/src/vcspull/cli/__init__.py @@ -5,16 +5,24 @@ import argparse import logging import pathlib +import sys import textwrap import typing as t from libvcs.__about__ import __version__ as libvcs_version +from vcspull import exc from vcspull.__about__ import __version__ from vcspull.log import setup_logger from ._formatter import VcspullHelpFormatter from .add import add_repo, create_add_subparser, handle_add_command +from .config import ( + config_ls, + create_config_subparser, + create_trust_subparser, + trust_command, +) from .discover import create_discover_subparser, discover_repos from .fmt import create_fmt_subparser, format_config_file from .import_cmd import create_import_subparser @@ -299,6 +307,47 @@ def build_description( ), ) +CONFIG_DESCRIPTION = build_description( + """ + Inspect the configuration scopes in effect. + + 'vcspull config ls' prints every configuration file vcspull would load + from the current directory, weakest first, with its scope, its repository + count, and whether a nearer file overrode any of its entries. + """, + ( + ( + None, + [ + "vcspull config ls", + "vcspull --no-project config ls", + ], + ), + ), +) + +TRUST_DESCRIPTION = build_description( + """ + Trust a project directory's configuration. + + A project config that would check a repository out beyond its own + directory needs your consent before vcspull will act on it. Trust is + recorded per directory, so a second config in the same project does not + re-ask. + """, + ( + ( + None, + [ + "vcspull trust", + "vcspull trust ~/work/api", + "vcspull trust --untrust ~/work/api", + "vcspull trust --show", + ], + ), + ), +) + WORKTREE_DESCRIPTION = build_description( """ Manage git worktrees for repositories. @@ -320,6 +369,40 @@ def build_description( ) +def _trust_parser() -> argparse.ArgumentParser: + """Return the shared parent parser carrying ``--trust-project``. + + Every command that may act on a project ``.vcspull.*`` accepts the trust + bypass, and the root parser inherits it too, so both + ``vcspull --trust-project sync`` and ``vcspull sync --trust-project`` + parse. ``SUPPRESS`` keeps the subcommand from clobbering a value the root + already set. + """ + parser = argparse.ArgumentParser(add_help=False) + parser.add_argument( + "--trust-project", + action="store_true", + default=argparse.SUPPRESS, + help=( + "load project configs that check repositories out beyond their " + "own directory (also VCSPULL_YES=1)" + ), + ) + return parser + + +def _scope_parser(trust: argparse.ArgumentParser) -> argparse.ArgumentParser: + """Return the shared parent parser for commands that resolve the stack.""" + parser = argparse.ArgumentParser(add_help=False, parents=[trust]) + parser.add_argument( + "--no-project", + action="store_true", + default=argparse.SUPPRESS, + help="skip .vcspull.* files found above the working directory", + ) + return parser + + @t.overload def create_parser( return_subparsers: t.Literal[True], @@ -334,10 +417,14 @@ def create_parser( return_subparsers: bool = False, ) -> argparse.ArgumentParser | tuple[argparse.ArgumentParser, t.Any]: """Create CLI argument parser for vcspull.""" + trust_parent = _trust_parser() + scope_parent = _scope_parser(trust_parent) + parser = argparse.ArgumentParser( prog="vcspull", formatter_class=VcspullHelpFormatter, description=CLI_DESCRIPTION, + parents=[scope_parent], ) parser.add_argument( "--version", @@ -352,7 +439,6 @@ def create_parser( default="INFO", help="log level (debug, info, warning, error, critical)", ) - subparsers = parser.add_subparsers(dest="subparser_name") # Sync command @@ -361,6 +447,7 @@ def create_parser( help="synchronize repositories", formatter_class=VcspullHelpFormatter, description=SYNC_DESCRIPTION, + parents=[scope_parent], ) create_sync_subparser(sync_parser) @@ -370,6 +457,7 @@ def create_parser( help="list configured repositories", formatter_class=VcspullHelpFormatter, description=LIST_DESCRIPTION, + parents=[scope_parent], ) create_list_subparser(list_parser) @@ -379,6 +467,7 @@ def create_parser( help="check repository status", formatter_class=VcspullHelpFormatter, description=STATUS_DESCRIPTION, + parents=[scope_parent], ) create_status_subparser(status_parser) @@ -388,6 +477,7 @@ def create_parser( help="search configured repositories", formatter_class=VcspullHelpFormatter, description=SEARCH_DESCRIPTION, + parents=[scope_parent], ) create_search_subparser(search_parser) @@ -397,6 +487,7 @@ def create_parser( help="add a single repository", formatter_class=VcspullHelpFormatter, description=ADD_DESCRIPTION, + parents=[trust_parent], ) create_add_subparser(add_parser) @@ -406,6 +497,7 @@ def create_parser( help="discover repositories from filesystem", formatter_class=VcspullHelpFormatter, description=DISCOVER_DESCRIPTION, + parents=[trust_parent], ) create_discover_subparser(discover_parser) @@ -415,6 +507,7 @@ def create_parser( help="format configuration files", formatter_class=VcspullHelpFormatter, description=FMT_DESCRIPTION, + parents=[scope_parent], ) create_fmt_subparser(fmt_parser) @@ -424,6 +517,7 @@ def create_parser( help="migrate configuration files to the options: form", formatter_class=VcspullHelpFormatter, description=MIGRATE_DESCRIPTION, + parents=[scope_parent], ) create_migrate_subparser(migrate_parser) @@ -436,12 +530,32 @@ def create_parser( ) create_import_subparser(import_parser) + # Config command + config_parser = subparsers.add_parser( + "config", + help="inspect the configuration scopes in effect", + formatter_class=VcspullHelpFormatter, + description=CONFIG_DESCRIPTION, + parents=[scope_parent], + ) + create_config_subparser(config_parser, scope_parent) + + # Trust command + trust_parser = subparsers.add_parser( + "trust", + help="trust a project directory's configuration", + formatter_class=VcspullHelpFormatter, + description=TRUST_DESCRIPTION, + ) + create_trust_subparser(trust_parser) + # Worktree command worktree_parser = subparsers.add_parser( "worktree", help="manage git worktrees", formatter_class=VcspullHelpFormatter, description=WORKTREE_DESCRIPTION, + parents=[scope_parent], ) create_worktree_subparser(worktree_parser) @@ -458,12 +572,29 @@ def create_parser( migrate_parser, import_parser, worktree_parser, + config_parser, ) return parser def cli(_args: list[str] | None = None) -> None: - """CLI entry point for vcspull.""" + """CLI entry point for vcspull. + + A :exc:`vcspull.exc.VCSPullException` is the tool telling you something + actionable — an untrusted project config, an unreadable file — so it is + reported as one line rather than a traceback. Run with + ``--log-level debug`` to see the stack. + """ + try: + _run(_args) + except exc.VCSPullException as error: + log.debug("vcspull command failed", exc_info=True) + print(f"vcspull: {error}", file=sys.stderr) + raise SystemExit(1) from error + + +def _run(_args: list[str] | None) -> None: + """Parse arguments and dispatch to the selected subcommand.""" parser, subparsers = create_parser(return_subparsers=True) ( sync_parser, @@ -476,6 +607,7 @@ def cli(_args: list[str] | None = None) -> None: _migrate_parser, _import_parser, _worktree_parser, + config_parser, ) = subparsers args = parser.parse_args(_args) @@ -516,6 +648,8 @@ def cli(_args: list[str] | None = None) -> None: log_file=getattr(args, "log_file", None), no_log_file=getattr(args, "no_log_file", False), panel_lines=getattr(args, "panel_lines", None), + include_project=not getattr(args, "no_project", False), + trust_project=getattr(args, "trust_project", False), ) elif args.subparser_name == "list": list_repos( @@ -527,6 +661,8 @@ def cli(_args: list[str] | None = None) -> None: output_ndjson=args.output_ndjson, color=args.color, include_worktrees=getattr(args, "include_worktrees", False), + include_project=not getattr(args, "no_project", False), + trust_project=getattr(args, "trust_project", False), ) elif args.subparser_name == "status": status_repos( @@ -539,6 +675,8 @@ def cli(_args: list[str] | None = None) -> None: color=args.color, concurrent=not getattr(args, "no_concurrent", False), max_concurrent=getattr(args, "max_concurrent", None), + include_project=not getattr(args, "no_project", False), + trust_project=getattr(args, "trust_project", False), ) elif args.subparser_name == "search": if not args.query_terms: @@ -558,6 +696,8 @@ def cli(_args: list[str] | None = None) -> None: word_regexp=getattr(args, "word_regexp", False), invert_match=getattr(args, "invert_match", False), match_any=getattr(args, "match_any", False), + include_project=not getattr(args, "no_project", False), + trust_project=getattr(args, "trust_project", False), ) elif args.subparser_name == "add": if not args.repo_path: @@ -574,6 +714,7 @@ def cli(_args: list[str] | None = None) -> None: recursive=args.recursive, workspace_root_override=args.workspace_root_path, yes=args.yes, + trust_project=getattr(args, "trust_project", False), dry_run=args.dry_run, merge_duplicates=args.merge_duplicates, include_worktrees=getattr(args, "include_worktrees", False), @@ -587,12 +728,14 @@ def cli(_args: list[str] | None = None) -> None: args.write, args.all, merge_roots=args.merge_roots, + trust_project=getattr(args, "trust_project", False), ) elif args.subparser_name == "migrate": migrate_config_file( args.config, args.write, args.all, + trust_project=getattr(args, "trust_project", False), ) elif args.subparser_name == "import": handler = getattr(args, "import_handler", None) @@ -602,5 +745,19 @@ def cli(_args: list[str] | None = None) -> None: result = handler(args) if result: raise SystemExit(result) + elif args.subparser_name == "config": + if args.config_command != "ls": + config_parser.print_help() + return + config_ls( + include_project=not getattr(args, "no_project", False), + trust_project=getattr(args, "trust_project", False), + ) + elif args.subparser_name == "trust": + trust_command( + args.directory, + untrust=args.untrust, + show=args.show, + ) elif args.subparser_name == "worktree": handle_worktree_command(args) diff --git a/src/vcspull/cli/add.py b/src/vcspull/cli/add.py index 849b1bacd..7cd3a6efa 100644 --- a/src/vcspull/cli/add.py +++ b/src/vcspull/cli/add.py @@ -21,6 +21,7 @@ from vcspull.config import ( build_repo_entry, canonicalize_workspace_path, + ensure_config_trusted, expand_dir, find_home_config_files, get_pin_reason, @@ -144,7 +145,10 @@ def create_add_subparser(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: ~/.vcspull.yaml or ./.vcspull.yaml)", + help=( + "path to config file to write " + "(default: ~/.vcspull.yaml, else ./.vcspull.yaml)" + ), ) parser.add_argument( "-w", @@ -545,6 +549,7 @@ def handle_add_command(args: argparse.Namespace) -> None: workspace_root_path=workspace_root_input, dry_run=args.dry_run, merge_duplicates=args.merge_duplicates, + trust_project=getattr(args, "trust_project", False), rev=getattr(args, "pin", None), shallow=shallow, depth=depth, @@ -560,6 +565,7 @@ def add_repo( dry_run: bool, *, merge_duplicates: bool = True, + trust_project: bool = False, rev: str | None = None, shallow: bool = False, depth: int | None = None, @@ -609,6 +615,9 @@ def add_repo( else: config_file_path = home_configs[0] + if not ensure_config_trusted(config_file_path, trust_project=trust_project): + return + # Load existing config raw_config: dict[str, t.Any] duplicate_root_occurrences: dict[str, list[t.Any]] diff --git a/src/vcspull/cli/config.py b/src/vcspull/cli/config.py new file mode 100644 index 000000000..92242a8dc --- /dev/null +++ b/src/vcspull/cli/config.py @@ -0,0 +1,239 @@ +"""Configuration scope introspection and project trust for vcspull.""" + +from __future__ import annotations + +import logging +import os +import pathlib +import typing as t + +from vcspull._internal import scopes +from vcspull._internal.config_reader import DuplicateAwareConfigReader +from vcspull._internal.private_path import PrivatePath +from vcspull.config import extract_repos, source_escapes + +if t.TYPE_CHECKING: + import argparse + + from vcspull.types import RawConfigDict + +log = logging.getLogger(__name__) + + +def create_config_subparser( + parser: argparse.ArgumentParser, + scope_parent: argparse.ArgumentParser, +) -> None: + """Create ``vcspull config`` argument subparser. + + Parameters + ---------- + parser : argparse.ArgumentParser + The ``config`` command parser. + scope_parent : argparse.ArgumentParser + Shared parent carrying ``--no-project`` and ``--trust-project``, so + they parse after ``ls`` as well as before ``config``. + """ + subparsers = parser.add_subparsers(dest="config_command") + subparsers.add_parser( + "ls", + help="list the configuration scopes in effect", + parents=[scope_parent], + ) + + +def create_trust_subparser(parser: argparse.ArgumentParser) -> None: + """Create ``vcspull trust`` argument subparser.""" + parser.add_argument( + "directory", + metavar="DIR", + nargs="?", + help="project directory to trust (default: current directory)", + ) + group = parser.add_mutually_exclusive_group() + group.add_argument( + "--untrust", + action="store_true", + help="remove the directory from the trusted set", + ) + group.add_argument( + "--show", + action="store_true", + help="print the trusted project directories and exit", + ) + + +def _plural(count: int, singular: str, plural: str | None = None) -> str: + """Return *count* with a correctly inflected noun. + + Examples + -------- + >>> _plural(1, "repo") + '1 repo' + >>> _plural(0, "repo") + '0 repos' + >>> _plural(2, "repository", "repositories") + '2 repositories' + """ + return f"{count} {singular if count == 1 else plural or singular + 's'}" + + +def _note(declared: int, effective: int, *, untrusted: bool) -> str: + """Return the parenthetical explaining why a file's count is not its whole. + + Examples + -------- + >>> _note(2, 2, untrusted=False) + '' + >>> _note(2, 1, untrusted=False) + ' (1 overridden)' + >>> _note(2, 0, untrusted=True) + ' (untrusted)' + """ + if untrusted: + return " (untrusted)" + overridden = declared - effective + return f" ({overridden} overridden)" if overridden else "" + + +def _declared_repos(path: pathlib.Path, cwd: pathlib.Path) -> list[pathlib.Path]: + """Return the destination of every repository a single file declares.""" + try: + content, _duplicates, _items = DuplicateAwareConfigReader.load_with_duplicates( + path, + ) + return [ + pathlib.Path(repo["path"]).parent / str(repo["name"]) + for repo in extract_repos(t.cast("RawConfigDict", content), cwd=cwd) + ] + except Exception: + log.warning("could not read %s", PrivatePath(path)) + return [] + + +class _ScopeRow(t.NamedTuple): + """One line of ``vcspull config ls``.""" + + scope: str + path: str + declared: list[pathlib.Path] + untrusted: bool + + +def config_ls( + *, + include_project: bool = True, + trust_project: bool = False, + cwd: pathlib.Path | None = None, +) -> None: + """Print the configuration scopes in effect, weakest first. + + Each row is a file, its scope, and how many repositories it declares. A + file whose entries are overridden by a nearer one says so, and a project + file awaiting trust is listed as untrusted rather than silently dropped — + this is the command you reach for when ``vcspull list`` refuses to load + something. + + Parameters + ---------- + include_project : bool + Set ``False`` to show the stack as ``--no-project`` resolves it. + trust_project : bool + Treat escaping project configs as trusted, as ``--trust-project`` does. + cwd : pathlib.Path, optional + Working directory. Defaults to :meth:`pathlib.Path.cwd`. + """ + cwd = cwd or pathlib.Path.cwd() + sources = scopes.resolve_sources(cwd=cwd, include_project=include_project) + + if not sources: + print("no configuration files found") + return + + rows: list[_ScopeRow] = [] + winner: dict[pathlib.Path, int] = {} + + for index, source in enumerate(sources): + declared = _declared_repos(source.path, cwd) + untrusted = not trust_project and bool(source_escapes(source, cwd=cwd)) + rows.append( + _ScopeRow(source.scope, str(PrivatePath(source.path)), declared, untrusted), + ) + if untrusted: + continue + for key in declared: + winner[key] = index + + scope_width = max(len(row.scope) for row in rows) + path_width = max(len(row.path) for row in rows) + + for index, row in enumerate(rows): + effective = sum(1 for owner in winner.values() if owner == index) + count = _plural(len(row.declared), "repo") + _note( + len(row.declared), + effective, + untrusted=row.untrusted, + ) + print(f"{row.scope:<{scope_width}} {row.path:<{path_width}} {count}") + + print(f"\n{_plural(len(winner), 'repository', 'repositories')} in effect.") + if any(row.untrusted for row in rows): + print("Untrusted configs are not loaded. Allow one with 'vcspull trust DIR'.") + + +def trust_command( + directory: str | None, + *, + untrust: bool = False, + show: bool = False, + state_file: pathlib.Path | None = None, +) -> None: + """Record, remove, or print trusted project directories. + + Parameters + ---------- + directory : str | None + Directory to trust or untrust. Defaults to the current directory. + untrust : bool + Remove the directory instead of adding it. + show : bool + Print the trusted set and make no change. + state_file : pathlib.Path, optional + File holding the record. Defaults to + :func:`vcspull._internal.scopes.trust_state_file`. + + Examples + -------- + >>> state = tmp_path / "trusted" + >>> trust_command(None, show=True, state_file=state) + no trusted project directories + >>> api = tmp_path / "api" + >>> api.mkdir() + >>> trust_command(str(api), state_file=state) # doctest: +ELLIPSIS + trusted ...api + >>> scopes.read_trusted(state) == frozenset({api}) + True + >>> trust_command(str(api), untrust=True, state_file=state) # doctest: +ELLIPSIS + untrusted ...api + >>> scopes.read_trusted(state) + frozenset() + """ + state_file = state_file or scopes.trust_state_file() + + if show: + trusted = sorted(scopes.read_trusted(state_file)) + if not trusted: + print("no trusted project directories") + for entry in trusted: + print(PrivatePath(entry)) + return + + raw = pathlib.Path(directory).expanduser() if directory else pathlib.Path.cwd() + target = PrivatePath(os.path.realpath(raw)) + + if untrust: + scopes.untrust_directory(state_file, raw) + print(f"untrusted {target}") + else: + scopes.trust_directory(state_file, raw) + print(f"trusted {target}") diff --git a/src/vcspull/cli/discover.py b/src/vcspull/cli/discover.py index 5d91b7a6b..fc42b1361 100644 --- a/src/vcspull/cli/discover.py +++ b/src/vcspull/cli/discover.py @@ -15,9 +15,11 @@ from vcspull._internal.config_reader import DuplicateAwareConfigReader from vcspull._internal.private_path import PrivatePath +from vcspull._internal.scopes import classify_scope from vcspull.config import ( build_repo_entry, canonicalize_workspace_path, + ensure_config_trusted, expand_dir, find_home_config_files, get_pin_reason, @@ -84,66 +86,6 @@ def _classify_discover_action(existing_entry: t.Any) -> DiscoverAction: return DiscoverAction.SKIP_EXISTING -ConfigScope = t.Literal["system", "user", "project", "external"] - - -def _classify_config_scope( - config_path: pathlib.Path, - *, - cwd: pathlib.Path, - home: pathlib.Path, -) -> ConfigScope: - """Determine whether a config lives in user, system, project, or external scope.""" - resolved = config_path.expanduser().resolve() - home = home.expanduser().resolve() - cwd = cwd.expanduser().resolve() - - default_user_configs = { - (home / ".vcspull.yaml").resolve(), - (home / ".vcspull.json").resolve(), - } - if resolved in default_user_configs: - return "user" - - xdg_config_home = ( - pathlib.Path(os.environ.get("XDG_CONFIG_HOME", home / ".config")) - .expanduser() - .resolve() - ) - user_config_root = (xdg_config_home / "vcspull").resolve() - try: - resolved.relative_to(user_config_root) - except ValueError: - pass - else: - return "user" - - xdg_config_dirs_value = os.environ.get("XDG_CONFIG_DIRS") - if xdg_config_dirs_value: - config_dir_bases = [ - pathlib.Path(entry).expanduser().resolve() - for entry in xdg_config_dirs_value.split(os.pathsep) - if entry - ] - else: - config_dir_bases = [pathlib.Path("/etc/xdg").resolve()] - - for base in config_dir_bases: - candidate = (base / "vcspull").resolve() - try: - resolved.relative_to(candidate) - except ValueError: - continue - else: - return "system" - - try: - resolved.relative_to(cwd) - except ValueError: - return "external" - return "project" - - def is_git_worktree(path: pathlib.Path) -> bool: """Check if a directory is a git worktree (not a main repository). @@ -232,7 +174,10 @@ def create_discover_subparser(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: ~/.vcspull.yaml or ./.vcspull.yaml)", + help=( + "path to config file to write " + "(default: ~/.vcspull.yaml, else ./.vcspull.yaml)" + ), ) parser.add_argument( "-w", @@ -345,6 +290,7 @@ def discover_repos( yes: bool, dry_run: bool, *, + trust_project: bool = False, merge_duplicates: bool = True, include_worktrees: bool = False, rev: str | None = None, @@ -413,9 +359,16 @@ def discover_repos( cwd = pathlib.Path.cwd() home = pathlib.Path.home() - config_scope = _classify_config_scope(config_file_path, cwd=cwd, home=home) + config_scope = classify_scope(config_file_path, cwd=cwd, home=home) allow_relative_workspace = config_scope == "project" + if not ensure_config_trusted( + config_file_path, + cwd=cwd, + trust_project=trust_project, + ): + return + raw_config: dict[str, t.Any] duplicate_root_occurrences: dict[str, list[t.Any]] if config_file_path.exists() and config_file_path.is_file(): diff --git a/src/vcspull/cli/fmt.py b/src/vcspull/cli/fmt.py index 858c32a74..2f867db20 100644 --- a/src/vcspull/cli/fmt.py +++ b/src/vcspull/cli/fmt.py @@ -12,10 +12,11 @@ from colorama import Fore, Style +from vcspull._internal import scopes from vcspull._internal.config_reader import DuplicateAwareConfigReader from vcspull._internal.private_path import PrivatePath from vcspull.config import ( - find_config_files, + ensure_config_trusted, find_home_config_files, is_pinned_for_op, merge_duplicate_workspace_roots, @@ -42,7 +43,7 @@ def create_fmt_subparser(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: .vcspull.yaml or ~/.vcspull.yaml)", + help="path to config file (default: ~/.vcspull.yaml, else ./.vcspull.yaml)", ) parser.add_argument( "--write", @@ -232,6 +233,8 @@ def format_single_config( write: bool, *, merge_roots: bool, + trust_project: bool = False, + explicit: bool = False, ) -> bool: """Format a single vcspull configuration file. @@ -243,6 +246,10 @@ def format_single_config( Whether to write changes back to file merge_roots : bool Merge duplicate workspace roots when True (default behavior) + trust_project : bool + Trust an escaping project config without prompting. + explicit : bool + The caller named this file with ``--file``. Returns ------- @@ -264,6 +271,13 @@ def format_single_config( ) return False + if not ensure_config_trusted( + config_file_path, + trust_project=trust_project, + explicit=explicit, + ): + return False + # Load existing config try: raw_config, duplicate_root_occurrences, _top_level_items = ( @@ -484,6 +498,7 @@ def format_config_file( format_all: bool = False, *, merge_roots: bool = True, + trust_project: bool = False, ) -> None: """Format vcspull configuration file(s). @@ -497,20 +512,12 @@ def format_config_file( If True, format all discovered config files merge_roots : bool Merge duplicate workspace roots when True (default) + trust_project : bool + Trust escaping project configs without prompting. """ if format_all: - # Format all discovered config files - config_files = find_config_files(include_home=True) - - # Also check for local .vcspull.yaml - local_yaml = pathlib.Path.cwd() / ".vcspull.yaml" - if local_yaml.exists() and local_yaml not in config_files: - config_files.append(local_yaml) - - # Also check for local .vcspull.json - local_json = pathlib.Path.cwd() / ".vcspull.json" - if local_json.exists() and local_json not in config_files: - config_files.append(local_json) + cwd = pathlib.Path.cwd() + config_files = [source.path for source in scopes.resolve_sources(cwd=cwd)] if not config_files: log.error( @@ -549,6 +556,7 @@ def format_config_file( config_file, write, merge_roots=merge_roots, + trust_project=trust_project, ): success_count += 1 @@ -602,4 +610,6 @@ def format_config_file( config_file_path, write, merge_roots=merge_roots, + trust_project=trust_project, + explicit=bool(config_file_path_str), ) diff --git a/src/vcspull/cli/list.py b/src/vcspull/cli/list.py index dc8c7d57d..a90822d70 100644 --- a/src/vcspull/cli/list.py +++ b/src/vcspull/cli/list.py @@ -7,7 +7,7 @@ import pathlib from vcspull._internal.private_path import PrivatePath -from vcspull.config import filter_repos, find_config_files, load_configs +from vcspull.config import filter_repos, load_scoped_configs from vcspull.types import ConfigDict from ._colors import Colors, get_color_mode @@ -30,7 +30,7 @@ def create_list_subparser(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: ~/.vcspull.yaml or ./.vcspull.yaml)", + help="path to config file (replaces the resolved scope stack)", ) parser.add_argument( "-w", @@ -86,6 +86,8 @@ def list_repos( output_ndjson: bool, color: str, include_worktrees: bool = False, + include_project: bool = True, + trust_project: bool = False, ) -> None: """List configured repositories. @@ -107,12 +109,17 @@ def list_repos( Color mode (auto, always, never) include_worktrees : bool Include configured worktrees in the listing (default: False) + include_project : bool + Include ``.vcspull.*`` files found above the working directory + trust_project : bool + Trust project configs that check repositories out beyond their directory """ # Load configs - if config_path: - configs = load_configs([config_path]) - else: - configs = load_configs(find_config_files(include_home=True)) + configs = load_scoped_configs( + config_path, + include_project=include_project, + trust_project=trust_project, + ) # Filter by patterns if provided if repo_patterns: diff --git a/src/vcspull/cli/migrate.py b/src/vcspull/cli/migrate.py index 73b195288..cfebbe161 100644 --- a/src/vcspull/cli/migrate.py +++ b/src/vcspull/cli/migrate.py @@ -11,11 +11,12 @@ from colorama import Fore, Style +from vcspull._internal import scopes from vcspull._internal.config_reader import DuplicateAwareConfigReader from vcspull._internal.private_path import PrivatePath from vcspull.config import ( LEGACY_REPO_OPTION_KEYS, - find_config_files, + ensure_config_trusted, find_home_config_files, migrate_repo_entry, normalize_config_file_path, @@ -32,7 +33,7 @@ def create_migrate_subparser(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: .vcspull.yaml or ~/.vcspull.yaml)", + help="path to config file (default: ~/.vcspull.yaml, else ./.vcspull.yaml)", ) parser.add_argument( "--write", @@ -89,7 +90,13 @@ def migrate_config(config_data: dict[str, t.Any]) -> tuple[dict[str, t.Any], int return migrated, change_count -def migrate_single_config(config_file_path: pathlib.Path, write: bool) -> bool: +def migrate_single_config( + config_file_path: pathlib.Path, + write: bool, + *, + trust_project: bool = False, + explicit: bool = False, +) -> bool: """Migrate a single vcspull configuration file. Parameters @@ -98,6 +105,10 @@ def migrate_single_config(config_file_path: pathlib.Path, write: bool) -> bool: Path to config file. write : bool Whether to write changes back to file. + trust_project : bool + Trust an escaping project config without prompting. + explicit : bool + The caller named this file with ``--file``. Returns ------- @@ -117,6 +128,13 @@ def migrate_single_config(config_file_path: pathlib.Path, write: bool) -> bool: ) return False + if not ensure_config_trusted( + config_file_path, + trust_project=trust_project, + explicit=explicit, + ): + return False + try: raw_config, _duplicate_root_occurrences, _top_level_items = ( DuplicateAwareConfigReader.load_with_duplicates(config_file_path) @@ -215,6 +233,8 @@ def migrate_config_file( config_file_path_str: str | None, write: bool, migrate_all: bool = False, + *, + trust_project: bool = False, ) -> None: """Migrate vcspull configuration file(s) to the ``options:`` form. @@ -226,17 +246,12 @@ def migrate_config_file( Whether to write changes back to file. migrate_all : bool If True, migrate all discovered config files. + trust_project : bool + Trust escaping project configs without prompting. """ if migrate_all: - config_files = find_config_files(include_home=True) - - local_yaml = pathlib.Path.cwd() / ".vcspull.yaml" - if local_yaml.exists() and local_yaml not in config_files: - config_files.append(local_yaml) - - local_json = pathlib.Path.cwd() / ".vcspull.json" - if local_json.exists() and local_json not in config_files: - config_files.append(local_json) + cwd = pathlib.Path.cwd() + config_files = [source.path for source in scopes.resolve_sources(cwd=cwd)] if not config_files: log.error( @@ -268,7 +283,11 @@ def migrate_config_file( success_count = 0 for config_file in config_files: - if migrate_single_config(config_file, write): + if migrate_single_config( + config_file, + write, + trust_project=trust_project, + ): success_count += 1 if success_count == len(config_files): @@ -313,4 +332,9 @@ def migrate_config_file( else: config_file_path = home_configs[0] - migrate_single_config(config_file_path, write) + migrate_single_config( + config_file_path, + write, + trust_project=trust_project, + explicit=bool(config_file_path_str), + ) diff --git a/src/vcspull/cli/search.py b/src/vcspull/cli/search.py index ad66e1b76..30322ca51 100644 --- a/src/vcspull/cli/search.py +++ b/src/vcspull/cli/search.py @@ -10,7 +10,7 @@ from dataclasses import dataclass from vcspull._internal.private_path import PrivatePath -from vcspull.config import find_config_files, load_configs +from vcspull.config import load_scoped_configs from vcspull.types import ConfigDict from ._colors import Colors, get_color_mode @@ -490,7 +490,7 @@ def create_search_subparser(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: ~/.vcspull.yaml or ./.vcspull.yaml)", + help="path to config file (replaces the resolved scope stack)", ) parser.add_argument( "-w", @@ -581,6 +581,8 @@ def search_repos( invert_match: bool, match_any: bool, emit_output: bool = True, + include_project: bool = True, + trust_project: bool = False, ) -> list[dict[str, t.Any]]: """Search configured repositories. @@ -614,6 +616,10 @@ def search_repos( Match if any term matches emit_output : bool Whether to emit human/JSON output + include_project : bool + Include ``.vcspull.*`` files found above the working directory + trust_project : bool + Trust project configs that check repositories out beyond their directory Returns ------- @@ -647,10 +653,11 @@ def search_repos( >>> [item["name"] for item in results] ['django'] """ - if config_path: - configs = load_configs([config_path]) - else: - configs = load_configs(find_config_files(include_home=True)) + configs = load_scoped_configs( + config_path, + include_project=include_project, + trust_project=trust_project, + ) if workspace_root: configs = filter_by_workspace(configs, workspace_root) diff --git a/src/vcspull/cli/status.py b/src/vcspull/cli/status.py index 55bf8d324..7db5772f3 100644 --- a/src/vcspull/cli/status.py +++ b/src/vcspull/cli/status.py @@ -15,7 +15,7 @@ from time import perf_counter from vcspull._internal.private_path import PrivatePath -from vcspull.config import filter_repos, find_config_files, load_configs +from vcspull.config import filter_repos, load_scoped_configs from vcspull.types import ConfigDict from ._colors import Colors, get_color_mode @@ -111,7 +111,7 @@ def create_status_subparser(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: ~/.vcspull.yaml or ./.vcspull.yaml)", + help="path to config file (replaces the resolved scope stack)", ) parser.add_argument( "-w", @@ -341,6 +341,8 @@ def status_repos( color: str, concurrent: bool = True, max_concurrent: int | None = None, + include_project: bool = True, + trust_project: bool = False, ) -> None: """Check status of configured repositories. @@ -364,12 +366,17 @@ def status_repos( Whether to check repositories concurrently (default: True) max_concurrent : int | None Maximum concurrent status checks (default: based on CPU count) + include_project : bool + Include ``.vcspull.*`` files found above the working directory + trust_project : bool + Trust project configs that check repositories out beyond their directory """ # Load configs - if config_path: - configs = load_configs([config_path]) - else: - configs = load_configs(find_config_files(include_home=True)) + configs = load_scoped_configs( + config_path, + include_project=include_project, + trust_project=trust_project, + ) # Filter by patterns if provided if repo_patterns: diff --git a/src/vcspull/cli/sync.py b/src/vcspull/cli/sync.py index 73ec66ca7..f6724ad43 100644 --- a/src/vcspull/cli/sync.py +++ b/src/vcspull/cli/sync.py @@ -36,7 +36,7 @@ plan_worktree_sync, sync_all_worktrees, ) -from vcspull.config import expand_dir, filter_repos, find_config_files, load_configs +from vcspull.config import expand_dir, filter_repos, load_scoped_configs from vcspull.log import default_debug_log_path, setup_file_logger, teardown_file_logger from vcspull.types import ConfigDict @@ -540,7 +540,7 @@ def create_sync_subparser(parser: argparse.ArgumentParser) -> argparse.ArgumentP "--file", dest="config", metavar="FILE", - help="path to config file (default: ~/.vcspull.yaml or ./.vcspull.yaml)", + help="path to config file (replaces the resolved scope stack)", ) parser.add_argument( "-w", @@ -1339,6 +1339,8 @@ def sync( log_file: str | pathlib.Path | None = None, no_log_file: bool = False, panel_lines: int | None = None, + include_project: bool = True, + trust_project: bool = False, ) -> None: """Entry point for ``vcspull sync``.""" # Prevent git from blocking on credential prompts during batch sync @@ -1392,6 +1394,8 @@ def sync( log_file_path=log_file_path, dry_run=dry_run, panel_lines=resolved_panel_lines, + include_project=include_project, + trust_project=trust_project, ) except KeyboardInterrupt as err: # Catch Ctrl-C from ANY phase of the sync -- the repo loop (where @@ -1446,6 +1450,8 @@ def _sync_impl( repo_timeout: int, log_file_path: pathlib.Path | None, panel_lines: int, + include_project: bool, + trust_project: bool, ) -> None: """Run the core body of :func:`sync`. @@ -1476,13 +1482,12 @@ def _sync_impl( ) plan_config = SyncPlanConfig(fetch=bool(fetch and not offline), offline=offline) - if config: - configs = load_configs([config], warn_legacy_options=True) - else: - configs = load_configs( - find_config_files(include_home=True), - warn_legacy_options=True, - ) + configs = load_scoped_configs( + config, + include_project=include_project, + trust_project=trust_project, + warn_legacy_options=True, + ) found_repos: list[ConfigDict] = [] unmatched_count = 0 diff --git a/src/vcspull/cli/worktree.py b/src/vcspull/cli/worktree.py index e8571011e..abd358b80 100644 --- a/src/vcspull/cli/worktree.py +++ b/src/vcspull/cli/worktree.py @@ -15,7 +15,7 @@ prune_worktrees, sync_all_worktrees, ) -from vcspull.config import expand_dir, filter_repos, find_config_files, load_configs +from vcspull.config import expand_dir, filter_repos, load_scoped_configs from ._colors import Colors, get_color_mode from ._output import OutputFormatter, get_output_mode @@ -91,7 +91,7 @@ def _add_common_args(parser: argparse.ArgumentParser) -> None: "--file", dest="config", metavar="FILE", - help="path to config file (default: ~/.vcspull.yaml or ./.vcspull.yaml)", + help="path to config file (replaces the resolved scope stack)", ) parser.add_argument( "-w", @@ -141,10 +141,11 @@ def handle_worktree_command(args: argparse.Namespace) -> None: # Load configs config_path = pathlib.Path(args.config) if args.config else None - if config_path: - configs = load_configs([config_path]) - else: - configs = load_configs(find_config_files(include_home=True)) + configs = load_scoped_configs( + config_path, + include_project=not getattr(args, "no_project", False), + trust_project=getattr(args, "trust_project", False), + ) # Filter by patterns if args.repo_patterns: diff --git a/src/vcspull/config.py b/src/vcspull/config.py index 6697236df..eb5943d35 100644 --- a/src/vcspull/config.py +++ b/src/vcspull/config.py @@ -10,20 +10,23 @@ import os import pathlib import subprocess +import sys import tempfile import typing as t -from collections.abc import Callable +from collections.abc import Callable, Iterable, Sequence from libvcs.sync.git import GitRemote from vcspull.validator import is_valid_config from . import exc +from ._internal import scopes from ._internal.config_reader import ( ConfigReader, DuplicateAwareConfigReader, config_format_from_path, ) +from ._internal.private_path import PrivatePath from .types import ConfigDict, RawConfigDict, WorktreeConfigDict from .util import get_config_dir, update_dict @@ -459,8 +462,8 @@ def find_home_config_files( def find_config_files( path: list[pathlib.Path] | pathlib.Path | None = None, match: list[str] | str | None = None, - filetype: t.Literal["json", "yaml", "*"] - | list[t.Literal["json", "yaml", "*"]] + filetype: t.Literal["json", "yaml", "yml", "*"] + | list[t.Literal["json", "yaml", "yml", "*"]] | None = None, include_home: bool = False, ) -> list[pathlib.Path]: @@ -488,7 +491,7 @@ def find_config_files( list of absolute paths to config files. """ if filetype is None: - filetype = ["json", "yaml"] + filetype = ["json", "yaml", "yml"] if match is None: match = ["*"] config_files = [] @@ -501,7 +504,6 @@ def find_config_files( if isinstance(path, list): for p in path: config_files.extend(find_config_files(p, match, filetype)) - return config_files else: path = path.expanduser() if isinstance(match, list): @@ -517,36 +519,50 @@ def find_config_files( return config_files +ConfigGate = Callable[[pathlib.Path, Sequence[ConfigDict]], bool] +"""Decides whether a resolved configuration file may contribute. + +Receives the file and the entries it expands to, and returns ``False`` to skip +it. Raising aborts the load. +""" + + def load_configs( files: list[pathlib.Path], cwd: pathlib.Path | Callable[[], pathlib.Path] = pathlib.Path.cwd, *, merge_duplicates: bool = True, warn_legacy_options: bool = False, + gate: ConfigGate | None = None, ) -> list[ConfigDict]: - """Return repos from a list of files. + """Return repos from a list of files, nearest file winning. + + *files* are ordered weakest to strongest. Repositories with distinct + destination paths are unioned; when two files name the same destination the + later one replaces the earlier entry whole, so a project config that + repoints a repository never inherits the weaker file's ``options:``. Parameters ---------- files : list - paths to config file + paths to config file, weakest first cwd : pathlib.Path current path (pass down for :func:`extract_repos` warn_legacy_options : bool If ``True``, log a deprecation warning for entries that still carry top-level ``rev``/``shallow``/``depth`` keys (see :func:`detect_legacy_repo_options`). + gate : ConfigGate, optional + Consulted with each file and the destinations it resolves to before the + file contributes. Returning ``False`` skips the file. Returns ------- list of dict : expanded config dict item - - Todo - ---- - Validate scheme, check for duplicate destinations, VCS urls """ - repos: list[ConfigDict] = [] + merged: dict[pathlib.Path, ConfigDict] = {} + origins: dict[pathlib.Path, pathlib.Path] = {} if callable(cwd): cwd = cwd() @@ -606,20 +622,290 @@ def load_configs( ) assert is_valid_config(config_content) - newrepos = extract_repos(config_content, cwd=cwd) + repos = extract_repos(config_content, cwd=cwd) - if not repos: - repos.extend(newrepos) + # Gated on the expanded entries, not the raw keys: an entry may + # override its destination with ``path:`` and never touch the + # workspace root the key declares. + if gate is not None and not gate(file, repos): continue - dupes = detect_duplicate_repos(repos, newrepos) + for repo in repos: + key = pathlib.Path(repo["path"]).parent / repo["name"] + displaced = merged.get(key) + if displaced is not None and _entries_diverge(displaced, repo): + log.warning( + "%s overrides %s for '%s' (url or vcs differs)", + PrivatePath(file), + PrivatePath(origins[key]), + PrivatePath(key), + ) + merged[key] = repo + origins[key] = file + + return list(merged.values()) - if len(dupes) > 0: - msg = ("repos with same path + different VCS detected!", dupes) - raise exc.VCSPullException(msg) - repos.extend(newrepos) - return repos +def _entries_diverge(displaced: ConfigDict, winner: ConfigDict) -> bool: + """Return whether two entries for one destination really disagree. + + Identical duplicates across a user and a project file are the common, + harmless case, so only a differing URL or VCS is worth a warning. + + Examples + -------- + >>> entry = {"url": "git+https://x", "vcs": "git"} + >>> _entries_diverge(entry, dict(entry)) + False + >>> _entries_diverge(entry, {"url": "git+https://fork", "vcs": "git"}) + True + """ + return (displaced.get("url"), displaced.get("vcs")) != ( + winner.get("url"), + winner.get("vcs"), + ) + + +def load_scoped_configs( + config_path: pathlib.Path | None = None, + *, + cwd: pathlib.Path | None = None, + include_project: bool = True, + trust_project: bool = False, + warn_legacy_options: bool = False, +) -> list[ConfigDict]: + """Return every repository the configuration scopes in effect define. + + An explicit *config_path* replaces the whole stack. Otherwise the system, + user, and project scopes are unioned nearest-wins, and each project config + that would check a repository out beyond its own directory must be trusted + first. + + Parameters + ---------- + config_path : pathlib.Path, optional + Explicit configuration file from ``-f``/``--file``. + cwd : pathlib.Path, optional + Working directory the project walk starts from. + include_project : bool + Set ``False`` to skip the project scope entirely (``--no-project``). + trust_project : bool + Trust escaping project configs without prompting + (``--trust-project``). + warn_legacy_options : bool + Warn about entries still using top-level ``rev``/``shallow``/``depth``. + + Raises + ------ + exc.VCSPullException + When an escaping project config cannot be confirmed because there is no + terminal to ask on. + """ + cwd = cwd or pathlib.Path.cwd() + + if config_path is not None: + return load_configs( + [config_path], + cwd=cwd, + warn_legacy_options=warn_legacy_options, + ) + + sources = scopes.resolve_sources(cwd=cwd, include_project=include_project) + by_path = {source.path: source for source in sources} + state_file = scopes.trust_state_file() + trusted = scopes.read_trusted(state_file) + + def gate(path: pathlib.Path, repos: Sequence[ConfigDict]) -> bool: + source = by_path[path] + escaping = scopes.requires_trust( + source, + repo_destinations(repos), + cwd=cwd, + trusted=trusted, + ) + if not escaping: + return True + return _authorize_config( + source, + escaping, + state_file=state_file, + trust_project=trust_project, + ) + + return load_configs( + [source.path for source in sources], + cwd=cwd, + warn_legacy_options=warn_legacy_options, + gate=gate, + ) + + +def repo_destinations(repos: Iterable[ConfigDict]) -> list[pathlib.Path]: + """Return where each entry would be checked out. + + ``path:`` overrides the workspace root, so this is the only honest answer + to "where does this configuration write?". + + Examples + -------- + >>> repo_destinations( + ... extract_repos( + ... {"./vendor/": {"evil": {"repo": "git+https://e.com/e.git", + ... "path": "~/.ssh/evil"}}}, + ... cwd=pathlib.Path("/w"), + ... ) + ... ) + [PosixPath('~/.ssh/evil')] + """ + return [pathlib.Path(repo["path"]) for repo in repos] + + +def source_escapes( + source: scopes.ConfigSource, + *, + cwd: pathlib.Path, +) -> tuple[pathlib.Path, ...]: + """Return the destinations *source* declares that need consent, if any. + + Never prompts and never raises, so a reporting command can ask the same + question the loader does without changing anything. + + Parameters + ---------- + source : scopes.ConfigSource + Resolved configuration file. + cwd : pathlib.Path + Working directory, used to resolve relative destinations. + """ + try: + content, _duplicates, _items = DuplicateAwareConfigReader.load_with_duplicates( + source.path, + ) + destinations = repo_destinations( + extract_repos(t.cast("RawConfigDict", content), cwd=cwd), + ) + except Exception: + # An unreadable config declares no destinations, so it cannot escape. + # Let the caller's own load report the real parse error. + return () + return scopes.requires_trust( + source, + destinations, + cwd=cwd, + trusted=scopes.read_trusted(scopes.trust_state_file()), + ) + + +def ensure_config_trusted( + config_path: pathlib.Path, + *, + cwd: pathlib.Path | None = None, + trust_project: bool = False, + explicit: bool = False, +) -> bool: + """Return whether a write command may target *config_path*. + + A repository that ships a configuration redirecting your writes deserves + the same gate as one that redirects your clones, so ``add``, ``discover``, + ``fmt``, and ``migrate`` ask this before touching a project file vcspull + found on its own. Files that do not exist yet, and files outside the + project scope, are always allowed. + + Naming a file with ``--file`` is consent, and *explicit* says so. The read + path does not gate ``--file`` either — it replaces the stack rather than + joining the project tier — and gating a reformat of a file you named while + ``vcspull sync --file`` clones from it freely would be incoherent. + + Parameters + ---------- + config_path : pathlib.Path + File the command is about to read or rewrite. + cwd : pathlib.Path, optional + Working directory. Defaults to :meth:`pathlib.Path.cwd`. + trust_project : bool + Trust an escaping config without prompting (``--trust-project``). + explicit : bool + The caller named this file with ``--file``. + + Examples + -------- + A file that does not exist yet is not a project config anybody shipped: + + >>> ensure_config_trusted(tmp_path / "brand-new.yaml", cwd=tmp_path) + True + """ + cwd = cwd or pathlib.Path.cwd() + if explicit or not config_path.is_file(): + return True + + scope = scopes.classify_scope( + config_path, + cwd=cwd, + home=pathlib.Path.home(), + ) + if scope != "project": + return True + + source = scopes.ConfigSource("project", config_path, config_path.parent) + escaping = source_escapes(source, cwd=cwd) + if not escaping: + return True + return _authorize_config( + source, + escaping, + state_file=scopes.trust_state_file(), + trust_project=trust_project, + ) + + +def _authorize_config( + source: scopes.ConfigSource, + escaping: Sequence[pathlib.Path], + *, + state_file: pathlib.Path, + trust_project: bool, +) -> bool: + """Ask whether an escaping project config may act, and remember ``always``. + + Raises + ------ + exc.VCSPullException + When there is no terminal to prompt on. A sync must never block on a + hidden prompt, so this is a hard error rather than a silent skip. + """ + config = PrivatePath(source.path) + directory = PrivatePath(os.path.realpath(source.trust_root)) + + if trust_project or scopes.env_flag("VCSPULL_YES"): + log.debug("trusting %s without prompting", config) + return True + + if not sys.stdin.isatty(): + listed = ", ".join(str(PrivatePath(path)) for path in escaping) + msg = ( + f"{config} would check repositories out outside its directory " + f"({listed}) and there is no terminal to confirm on. " + f"Run 'vcspull trust {directory}' to allow it, pass " + "--trust-project, or use --no-project to skip project configs." + ) + raise exc.VCSPullException(msg) + + print( + f"! {config} would check repositories out outside its directory:", + file=sys.stderr, + ) + for path in escaping: + print(f" {PrivatePath(path)}", file=sys.stderr) + answer = input(" Trust this config? [y/N/always] ").strip().lower() + + if answer == "always": + scopes.trust_directory(state_file, source.trust_root) + return True + if answer in {"y", "yes"}: + return True + + log.warning("skipping untrusted config %s", config) + return False def detect_duplicate_repos( diff --git a/tests/cli/test_config.py b/tests/cli/test_config.py new file mode 100644 index 000000000..f7467afb1 --- /dev/null +++ b/tests/cli/test_config.py @@ -0,0 +1,266 @@ +"""Tests for ``vcspull config ls``, ``vcspull trust``, and the scope flags.""" + +from __future__ import annotations + +import pathlib +import textwrap +import typing as t + +import pytest + +from tests.helpers import write_config +from vcspull._internal import scopes +from vcspull.cli import cli + +if t.TYPE_CHECKING: + from collections.abc import Iterator + + +@pytest.fixture +def scoped_cli( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> Iterator[pathlib.Path]: + """Give one test its own home, trust record, and project tree. + + Yields the project directory, already the working directory. The user + config declares ``flask`` and ``click`` under the project's vendor + directory; the project config repoints ``flask``, so one entry is + overridden and one is not. + """ + home = tmp_path / "home" + project = home / "work" / "proj" + project.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("XDG_CONFIG_HOME", str(home / ".config")) + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "state")) + monkeypatch.delenv("VCSPULL_CEILING_PATHS", raising=False) + monkeypatch.delenv("VCSPULL_NO_PROJECT", raising=False) + monkeypatch.chdir(project) + + write_config( + home / ".vcspull.yaml", + textwrap.dedent( + """\ + ~/work/proj/vendor/: + flask: git+https://github.com/pallets/flask.git + click: git+https://github.com/pallets/click.git + """, + ), + ) + write_config( + project / ".vcspull.yaml", + "./vendor/:\n flask: git+https://github.com/myfork/flask.git\n", + ) + yield project + + +def _escaping(project: pathlib.Path) -> pathlib.Path: + """Write a project config that checks a repository out into ``~/.ssh``.""" + return write_config( + project / ".vcspull.yaml", + textwrap.dedent( + """\ + ./vendor/: + evil: + repo: git+https://example.com/evil.git + path: ~/.ssh/evil + """, + ), + ) + + +def test_config_ls_lists_tiers_weakest_first( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Every file in effect is named, with its scope and repository count.""" + cli(["config", "ls"]) + + lines = capsys.readouterr().out.splitlines() + + assert lines[0].startswith("user") + assert lines[1].startswith("project") + assert lines[-1] == "2 repositories in effect." + + +def test_config_ls_marks_overridden_entries( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """The weaker file says how many of its entries a nearer one displaced.""" + cli(["config", "ls"]) + + out = capsys.readouterr().out + + assert "2 repos (1 overridden)" in out + assert "1 repo\n" in out + + +def test_config_ls_marks_an_untrusted_project_config( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """``config ls`` diagnoses the refusal instead of reproducing it. + + ``vcspull list`` raises on this tree, so the command you reach for to find + out why must not raise, prompt, or quietly omit the file. + """ + _escaping(scoped_cli) + + cli(["config", "ls"]) + + out = capsys.readouterr().out + assert "(untrusted)" in out + assert "vcspull trust" in out + + +def test_config_ls_honours_no_project( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """``--no-project`` after ``ls`` drops the project tier from the report.""" + cli(["config", "ls", "--no-project"]) + + out = capsys.readouterr().out + assert "project" not in out + assert "2 repositories in effect." in out + + +def test_config_ls_without_any_configuration( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A machine with no configuration says so rather than printing a header.""" + home = tmp_path / "empty-home" + (home / "work").mkdir(parents=True) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("XDG_CONFIG_HOME", str(home / ".config")) + monkeypatch.delenv("VCSPULL_CEILING_PATHS", raising=False) + monkeypatch.chdir(home / "work") + + cli(["config", "ls"]) + + assert capsys.readouterr().out == "no configuration files found\n" + + +class ScopeFlagFixture(t.NamedTuple): + """Fixture for a scope flag accepted before and after the subcommand.""" + + test_id: str + argv: list[str] + + +SCOPE_FLAG_FIXTURES: list[ScopeFlagFixture] = [ + ScopeFlagFixture(test_id="root-no-project", argv=["--no-project", "list"]), + ScopeFlagFixture(test_id="subcommand-no-project", argv=["list", "--no-project"]), + ScopeFlagFixture( + test_id="root-trust-project", + argv=["--trust-project", "list"], + ), + ScopeFlagFixture( + test_id="subcommand-trust-project", + argv=["list", "--trust-project"], + ), +] + + +@pytest.mark.parametrize( + list(ScopeFlagFixture._fields), + SCOPE_FLAG_FIXTURES, + ids=[fixture.test_id for fixture in SCOPE_FLAG_FIXTURES], +) +def test_scope_flags_parse_in_either_position( + test_id: str, + argv: list[str], + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Both spellings work: nobody should have to remember which side wins.""" + _escaping(scoped_cli) + + cli(argv) + + assert "flask" in capsys.readouterr().out + + +def test_untrusted_config_reports_one_line_and_exits( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """A refusal is an actionable sentence, not a traceback.""" + _escaping(scoped_cli) + + with pytest.raises(SystemExit) as excinfo: + cli(["list"]) + + captured = capsys.readouterr() + assert excinfo.value.code == 1 + assert len(captured.err.strip().splitlines()) == 1 + assert "vcspull trust" in captured.err + + +def test_trust_records_shows_and_forgets( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """``trust``, ``--show``, and ``--untrust`` round-trip one directory.""" + cli(["trust", str(scoped_cli)]) + assert scopes.read_trusted(scopes.trust_state_file()) == frozenset( + {scoped_cli.resolve()}, + ) + + capsys.readouterr() + cli(["trust", "--show"]) + assert str(scoped_cli.resolve()) in capsys.readouterr().out.replace( + "~", + str(pathlib.Path.home()), + ) + + cli(["trust", "--untrust", str(scoped_cli)]) + assert scopes.read_trusted(scopes.trust_state_file()) == frozenset() + + +def test_trusting_the_project_lets_it_load( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Trust is what turns the refusal into a load.""" + _escaping(scoped_cli) + cli(["trust", str(scoped_cli)]) + capsys.readouterr() + + cli(["list"]) + + assert "evil" in capsys.readouterr().out + + +def test_fmt_write_is_gated_on_a_discovered_project_config( + scoped_cli: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """``fmt --write`` runs through the same gate every other command does.""" + escaping = _escaping(scoped_cli) + before = escaping.read_text(encoding="utf-8") + + with pytest.raises(SystemExit): + cli(["fmt", "--all", "--write"]) + + assert escaping.read_text(encoding="utf-8") == before + + +def test_discover_yes_does_not_grant_trust( + scoped_cli: pathlib.Path, +) -> None: + """``discover --yes`` skips *its own* prompt, never the trust gate. + + The two questions are unrelated, and conflating them would let a routine + import flag authorize a config that writes outside its directory. + """ + (pathlib.Path.home() / ".vcspull.yaml").unlink() + _escaping(scoped_cli) + (scoped_cli / "vendor").mkdir() + + with pytest.raises(SystemExit): + cli(["discover", str(scoped_cli / "vendor"), "--yes"]) diff --git a/tests/cli/test_discover.py b/tests/cli/test_discover.py index 8456be8f4..c3b18d932 100644 --- a/tests/cli/test_discover.py +++ b/tests/cli/test_discover.py @@ -11,7 +11,8 @@ import pytest from vcspull._internal.private_path import PrivatePath -from vcspull.cli.discover import ConfigScope, _classify_config_scope, discover_repos +from vcspull._internal.scopes import ConfigScope, classify_scope +from vcspull.cli.discover import discover_repos if t.TYPE_CHECKING: from _pytest.monkeypatch import MonkeyPatch @@ -667,7 +668,7 @@ def test_discover_detects_numeric_depth( CONFIG_SCOPE_FIXTURES, ids=[fixture.test_id for fixture in CONFIG_SCOPE_FIXTURES], ) -def test_classify_config_scope( +def test_classify_scope( test_id: str, config_template: str, cwd_template: str, @@ -719,7 +720,7 @@ def _expand(template: str) -> pathlib.Path: expanded_value = expanded_value.replace(f"{{{name}}}", str(path)) monkeypatch.setenv(key, expanded_value) - scope = _classify_config_scope(config_path, cwd=cwd_path, home=base_home) + scope = classify_scope(config_path, cwd=cwd_path, home=base_home) assert scope == expected_scope diff --git a/tests/cli/test_fmt.py b/tests/cli/test_fmt.py index 946f5ec25..f6bec953b 100644 --- a/tests/cli/test_fmt.py +++ b/tests/cli/test_fmt.py @@ -400,26 +400,6 @@ def test_format_all_configs( monkeypatch.chdir(local_root) - def fake_find_config_files(include_home: bool = False) -> list[pathlib.Path]: - files: list[pathlib.Path] = [work_config] - if include_home: - files.insert(0, home_config) - return files - - def fake_find_home_config_files( - filetype: list[str] | None = None, - ) -> list[pathlib.Path]: - return [home_config] - - monkeypatch.setattr( - "vcspull.cli.fmt.find_config_files", - fake_find_config_files, - ) - monkeypatch.setattr( - "vcspull.cli.fmt.find_home_config_files", - fake_find_home_config_files, - ) - with caplog.at_level(logging.INFO): format_config_file(None, write=False, format_all=True) diff --git a/tests/cli/test_sync_watchdog.py b/tests/cli/test_sync_watchdog.py index 695cb376e..eb794197d 100644 --- a/tests/cli/test_sync_watchdog.py +++ b/tests/cli/test_sync_watchdog.py @@ -314,7 +314,7 @@ def test_sync_handles_keyboard_interrupt_during_config_load( """Ctrl-C during pre-loop work (e.g. YAML parse) exits cleanly with 130. Regression for the observed traceback where a KeyboardInterrupt raised - inside ``load_configs`` escaped all the way through ``cli.sync`` and + inside ``load_scoped_configs`` escaped all the way through ``cli.sync`` and out to the top-level ``sys.exit`` as an unhandled exception, dumping the entire yaml parser stack to the terminal. The outer ``sync()`` entry point must catch the interrupt, emit a short notice, and exit @@ -328,7 +328,7 @@ def test_sync_handles_keyboard_interrupt_during_config_load( def _raising_load(*_args: t.Any, **_kwargs: t.Any) -> t.Any: raise KeyboardInterrupt - monkeypatch.setattr(sync_module, "load_configs", _raising_load) + monkeypatch.setattr(sync_module, "load_scoped_configs", _raising_load) with pytest.raises(SystemExit) as excinfo: sync_fn( diff --git a/tests/test_cli.py b/tests/test_cli.py index 670f08b1a..a0d97ff24 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2303,14 +2303,9 @@ def test_sync_human_output_redacts_repo_paths( monkeypatch.setattr( sync_module, - "load_configs", + "load_scoped_configs", lambda _paths, **_kwargs: [repo_config], ) - monkeypatch.setattr( - sync_module, - "find_config_files", - lambda include_home=True: [], - ) def _fake_filter_repos( _configs: list[dict[str, t.Any]], diff --git a/tests/test_config_file.py b/tests/test_config_file.py index b51fbc80f..7ba61b824 100644 --- a/tests/test_config_file.py +++ b/tests/test_config_file.py @@ -498,12 +498,15 @@ def test_merge_nested_dict(tmp_path: pathlib.Path, config_path: pathlib.Path) -> ), ) - # Duplicate path + name with different repo URL / remotes raises. - config_files = config.find_config_files( - path=config_path, - match="repoduplicate[1-2]", + # A repeated destination resolves to the nearest file's whole entry, and + # distinct destinations union. + config_files = sorted( + config.find_config_files(path=config_path, match="repoduplicate[1-2]"), ) assert config1 in config_files assert config2 in config_files - with pytest.raises(exc.VCSPullException): - config.load_configs(config_files) + + repos = config.load_configs(config_files) + by_name = {repo["name"]: repo for repo in repos} + assert set(by_name) == {"subRepoDiffVCS", "subRepoSameVCS", "vcsOn1", "vcsOn2"} + assert by_name["subRepoDiffVCS"]["url"] == "git+file:///path/to/diffrepo" diff --git a/tests/test_log.py b/tests/test_log.py index 019e5076b..797556390 100644 --- a/tests/test_log.py +++ b/tests/test_log.py @@ -432,6 +432,7 @@ def test_get_cli_logger_names_includes_base() -> None: "vcspull.cli._progress", "vcspull.cli._workspaces", "vcspull.cli.add", + "vcspull.cli.config", "vcspull.cli.discover", "vcspull.cli.fmt", "vcspull.cli.import_cmd", @@ -500,6 +501,7 @@ def test_setup_logger_leaves_cli_loggers_propagating(caplog: LogCaptureFixture) for logger_name in [ "vcspull.cli.add", + "vcspull.cli.config", "vcspull.cli.sync", ]: logger = logging.getLogger(logger_name) diff --git a/tests/test_scopes.py b/tests/test_scopes.py new file mode 100644 index 000000000..89ff8e6f9 --- /dev/null +++ b/tests/test_scopes.py @@ -0,0 +1,577 @@ +"""Tests for configuration scope resolution, merging, and project trust.""" + +from __future__ import annotations + +import io +import os +import pathlib +import sys +import textwrap +import typing as t + +import pytest + +from vcspull import config, exc +from vcspull._internal import scopes +from vcspull.types import ConfigDict + +from .helpers import write_config + +if t.TYPE_CHECKING: + from collections.abc import Iterator + + +class WalkFixture(t.NamedTuple): + """Fixture for the bounded upward walk.""" + + test_id: str + cwd: str + ceilings: tuple[str, ...] + expected: tuple[str, ...] + + +WALK_FIXTURES: list[WalkFixture] = [ + WalkFixture( + test_id="stops-below-home", + cwd="/home/u/work/api/src", + ceilings=("/home/u",), + expected=("/home/u/work", "/home/u/work/api", "/home/u/work/api/src"), + ), + WalkFixture( + test_id="ceiling-itself-excluded", + cwd="/home/u", + ceilings=("/home/u",), + expected=(), + ), + WalkFixture( + test_id="outside-home-runs-to-root", + cwd="/srv/app", + ceilings=("/home/u",), + expected=("/", "/srv", "/srv/app"), + ), + WalkFixture( + test_id="custom-ceiling-overrides-home", + cwd="/home/u/work/api", + ceilings=("/home/u/work",), + expected=("/home/u/work/api",), + ), + WalkFixture( + test_id="no-ceiling-runs-to-root", + cwd="/home/u/work", + ceilings=(), + expected=("/", "/home", "/home/u", "/home/u/work"), + ), +] + + +@pytest.mark.parametrize( + list(WalkFixture._fields), + WALK_FIXTURES, + ids=[fixture.test_id for fixture in WALK_FIXTURES], +) +def test_project_dirs_walk( + test_id: str, + cwd: str, + ceilings: tuple[str, ...], + expected: tuple[str, ...], +) -> None: + """The walk yields ancestors outermost first and honours its ceiling.""" + result = scopes.project_dirs( + pathlib.Path(cwd), + ceilings=frozenset(pathlib.Path(ceiling) for ceiling in ceilings), + ) + assert result == tuple(pathlib.Path(path) for path in expected) + + +@pytest.fixture(autouse=True) +def trust_state_isolation( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep the trusted-directory record per test. + + ``$HOME`` is session-scoped, so without this the trust state written by one + test would still be there for the next. + """ + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "state")) + + +@pytest.fixture +def scoped_tree( + tmp_path: pathlib.Path, + config_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> Iterator[pathlib.Path]: + """Build a user config plus project configs at two depths, and chdir deep. + + Yields the deep working directory. ``shared`` is declared by both the user + config and the outer project config, which is what makes the override + observable. + """ + # ``$HOME`` is session-scoped, so give this tree its own home; otherwise a + # ``~/.vcspull.yaml`` left by another test joins the user scope. + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + + work = tmp_path / "work" + deep = work / "api" / "src" / "deep" + deep.mkdir(parents=True) + + write_config( + config_path / "main.yaml", + textwrap.dedent( + f"""\ + {work}/vendor/: + shared: + repo: git+https://example.com/upstream.git + options: + rev: v1.0.0 + ~/code/: + flask: git+https://github.com/pallets/flask.git + """, + ), + ) + write_config( + work / ".vcspull.yaml", + textwrap.dedent( + f"""\ + {work}/vendor/: + shared: git+https://example.com/fork.git + outer: git+https://example.com/outer.git + """, + ), + ) + write_config( + work / "api" / ".vcspull.yaml", + textwrap.dedent( + f"""\ + {work}/api/deps/: + inner: git+https://example.com/inner.git + """, + ), + ) + + monkeypatch.setenv("VCSPULL_CEILING_PATHS", str(tmp_path)) + monkeypatch.chdir(deep) + yield deep + + +def _by_name(repos: list[ConfigDict]) -> dict[str, ConfigDict]: + return {str(repo["name"]): repo for repo in repos} + + +def test_scopes_union_with_nearest_tier_winning( + scoped_tree: pathlib.Path, +) -> None: + """From a deep subdirectory every scope contributes, nearest wins outright.""" + repos = _by_name(config.load_scoped_configs(cwd=scoped_tree)) + + assert set(repos) == {"flask", "shared", "outer", "inner"} + assert repos["shared"]["url"] == "git+https://example.com/fork.git" + # Whole-entry replacement: the user config's ``rev`` pin does not survive. + assert "rev" not in repos["shared"] + + +def test_no_project_reproduces_user_only_resolution( + scoped_tree: pathlib.Path, +) -> None: + """``--no-project`` resolves exactly the pre-change user configuration.""" + repos = _by_name( + config.load_scoped_configs(cwd=scoped_tree, include_project=False), + ) + + assert set(repos) == {"flask", "shared"} + assert repos["shared"]["url"] == "git+https://example.com/upstream.git" + assert repos["shared"]["rev"] == "v1.0.0" + + +def test_explicit_file_replaces_the_whole_stack( + scoped_tree: pathlib.Path, + tmp_path: pathlib.Path, +) -> None: + """``--file`` stays exclusive, exactly as it behaves today.""" + explicit = write_config( + tmp_path / "explicit.yaml", + f"{tmp_path}/only/:\n solo: git+https://example.com/solo.git\n", + ) + + repos = config.load_scoped_configs(explicit, cwd=scoped_tree) + + assert [repo["name"] for repo in repos] == ["solo"] + + +def test_escaping_project_config_errors_without_a_tty( + scoped_tree: pathlib.Path, +) -> None: + """No terminal means a hard error naming ``vcspull trust``, never a hang.""" + write_config( + scoped_tree.parent / ".vcspull.yaml", + "~/.ssh/:\n payload: git+https://example.com/payload.git\n", + ) + + with pytest.raises(exc.VCSPullException, match="vcspull trust"): + config.load_scoped_configs(cwd=scoped_tree) + + +def test_escaping_project_config_loads_with_trust_project( + scoped_tree: pathlib.Path, +) -> None: + """``--trust-project`` accepts an escaping config non-interactively.""" + write_config( + scoped_tree.parent / ".vcspull.yaml", + "~/.ssh/:\n payload: git+https://example.com/payload.git\n", + ) + + repos = _by_name(config.load_scoped_configs(cwd=scoped_tree, trust_project=True)) + + assert "payload" in repos + + +class _Tty(io.StringIO): + """Stand-in for an interactive stdin.""" + + def isatty(self) -> bool: + """Report an interactive terminal.""" + return True + + +class TrustAnswerFixture(t.NamedTuple): + """Fixture for the interactive trust prompt.""" + + test_id: str + answer: str + loads: bool + remembers: bool + + +TRUST_ANSWER_FIXTURES: list[TrustAnswerFixture] = [ + TrustAnswerFixture(test_id="declined", answer="n", loads=False, remembers=False), + TrustAnswerFixture( + test_id="empty-declines", answer="", loads=False, remembers=False + ), + TrustAnswerFixture( + test_id="accepted-once", answer="y", loads=True, remembers=False + ), + TrustAnswerFixture(test_id="always", answer="always", loads=True, remembers=True), +] + + +@pytest.mark.parametrize( + list(TrustAnswerFixture._fields), + TRUST_ANSWER_FIXTURES, + ids=[fixture.test_id for fixture in TRUST_ANSWER_FIXTURES], +) +def test_trust_prompt_answers( + test_id: str, + answer: str, + loads: bool, + remembers: bool, + scoped_tree: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The prompt gates the file, and ``always`` remembers the directory.""" + project = scoped_tree.parent + write_config( + project / ".vcspull.yaml", + "~/.ssh/:\n payload: git+https://example.com/payload.git\n", + ) + monkeypatch.setattr(sys, "stdin", _Tty()) + monkeypatch.setattr("builtins.input", lambda _prompt="": answer) + + repos = _by_name(config.load_scoped_configs(cwd=scoped_tree)) + trusted = scopes.read_trusted(scopes.trust_state_file()) + + assert ("payload" in repos) is loads + assert (project.resolve() in trusted) is remembers + + +def test_trusted_directory_skips_the_prompt( + scoped_tree: pathlib.Path, +) -> None: + """A remembered project directory loads without asking again.""" + project = scoped_tree.parent + write_config( + project / ".vcspull.yaml", + "~/.ssh/:\n payload: git+https://example.com/payload.git\n", + ) + scopes.trust_directory(scopes.trust_state_file(), project) + + assert "payload" in _by_name(config.load_scoped_configs(cwd=scoped_tree)) + + +class DivergenceFixture(t.NamedTuple): + """Fixture for the override-conflict warning.""" + + test_id: str + nearest_url: str + warns: bool + + +DIVERGENCE_FIXTURES: list[DivergenceFixture] = [ + DivergenceFixture( + test_id="identical-is-silent", + nearest_url="git+https://example.com/upstream.git", + warns=False, + ), + DivergenceFixture( + test_id="different-url-warns", + nearest_url="git+https://example.com/fork.git", + warns=True, + ), + DivergenceFixture( + test_id="different-vcs-warns", + nearest_url="hg+https://example.com/upstream.git", + warns=True, + ), +] + + +@pytest.mark.parametrize( + list(DivergenceFixture._fields), + DIVERGENCE_FIXTURES, + ids=[fixture.test_id for fixture in DIVERGENCE_FIXTURES], +) +def test_override_warns_only_on_real_divergence( + test_id: str, + nearest_url: str, + warns: bool, + tmp_path: pathlib.Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """An identical duplicate across scopes is the harmless, silent case.""" + weakest = write_config( + tmp_path / "weakest.yaml", + f"{tmp_path}/code/:\n shared: git+https://example.com/upstream.git\n", + ) + nearest = write_config( + tmp_path / "nearest.yaml", + f"{tmp_path}/code/:\n shared: {nearest_url}\n", + ) + + with caplog.at_level("WARNING", logger="vcspull.config"): + repos = config.load_configs([weakest, nearest]) + + assert [repo["name"] for repo in repos] == ["shared"] + assert repos[0]["url"] == nearest_url + assert bool(caplog.records) is warns + + +class ContainmentFixture(t.NamedTuple): + """Fixture for the destination containment check.""" + + test_id: str + destinations: tuple[str, ...] + expected: tuple[str, ...] + + +CONTAINMENT_FIXTURES: list[ContainmentFixture] = [ + ContainmentFixture( + test_id="relative-child-contained", + destinations=("./vendor/flask", "sub/nested/x"), + expected=(), + ), + ContainmentFixture( + test_id="home-rooted-escapes", + destinations=("~/.ssh/evil",), + expected=("~/.ssh/evil",), + ), + ContainmentFixture( + test_id="parent-traversal-escapes", + destinations=("../elsewhere/x",), + expected=("../elsewhere/x",), + ), + ContainmentFixture( + test_id="symlink-target-escapes", + destinations=("./escape-link/x",), + expected=("./escape-link/x",), + ), + ContainmentFixture( + test_id="mixed-reports-only-escaping", + destinations=("./vendor/a", "~/.ssh/b", "./vendor/a"), + expected=("~/.ssh/b",), + ), +] + + +@pytest.mark.parametrize( + list(ContainmentFixture._fields), + CONTAINMENT_FIXTURES, + ids=[fixture.test_id for fixture in CONTAINMENT_FIXTURES], +) +def test_escaping_destinations( + test_id: str, + destinations: tuple[str, ...], + expected: tuple[str, ...], + tmp_path: pathlib.Path, +) -> None: + """Only destinations landing outside the config's own directory report.""" + project = tmp_path / "project" + (project / "sub" / "nested").mkdir(parents=True) + (project / "escape-link").symlink_to(tmp_path / "outside", target_is_directory=True) + + def resolved(entry: str) -> pathlib.Path: + target = pathlib.Path(entry).expanduser() + if not target.is_absolute(): + target = project / target + return pathlib.Path(os.path.realpath(target)) + + assert scopes.escaping_destinations( + [pathlib.Path(entry) for entry in destinations], + config_dir=project, + cwd=project, + ) == tuple(resolved(entry) for entry in expected) + + +def test_path_override_cannot_smuggle_a_destination_out( + scoped_tree: pathlib.Path, +) -> None: + """A contained workspace root does not license an escaping ``path:``. + + ``extract_repos`` honours a per-entry ``path:`` and never consults the + workspace root, so a gate reading only the top-level keys would clone this + into ``~/.ssh`` without asking. + """ + write_config( + scoped_tree.parent / ".vcspull.yaml", + textwrap.dedent( + """\ + ./vendor/: + evil: + repo: git+https://example.com/evil.git + path: ~/.ssh/evil + """, + ), + ) + + with pytest.raises(exc.VCSPullException, match="vcspull trust"): + config.load_scoped_configs(cwd=scoped_tree) + + +def test_symlinked_home_stops_the_walk_and_loads_once( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A ``$HOME`` reached through a symlink is still the ceiling. + + Comparing unresolved paths let the walk run past ``$HOME``, which both + re-read the home dotfile as a project config and exposed everything above + it. + """ + real_home = tmp_path / "real" / "u" + (real_home / "work" / "proj").mkdir(parents=True) + linked_home = tmp_path / "u" + linked_home.symlink_to(real_home, target_is_directory=True) + monkeypatch.setenv("HOME", str(linked_home)) + monkeypatch.delenv("VCSPULL_CEILING_PATHS", raising=False) + + write_config( + real_home / ".vcspull.yaml", + f"{tmp_path}/code/:\n flask: git+https://example.com/flask.git\n", + ) + + for cwd in (real_home / "work" / "proj", linked_home / "work" / "proj"): + sources = scopes.resolve_sources(cwd=cwd) + assert [source.scope for source in sources] == ["user"] + + +def test_both_home_dotfiles_still_raise( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Owning ``~/.vcspull.yaml`` and ``~/.vcspull.json`` is still an error.""" + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + write_config(home / ".vcspull.yaml", "{}\n") + write_config(home / ".vcspull.json", "{}\n") + + with pytest.raises(exc.MultipleConfigWarning): + scopes.resolve_sources(cwd=home / "work") + + +def test_vcspull_no_project_env_drops_the_project_scope( + scoped_tree: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``VCSPULL_NO_PROJECT=1`` resolves the same stack as ``--no-project``.""" + monkeypatch.setenv("VCSPULL_NO_PROJECT", "1") + + repos = _by_name(config.load_scoped_configs(cwd=scoped_tree)) + + assert set(repos) == {"flask", "shared"} + assert repos["shared"]["url"] == "git+https://example.com/upstream.git" + + +class ScopeDirFixture(t.NamedTuple): + """Fixture for scope directories that must not crash the resolver.""" + + test_id: str + build: t.Callable[[pathlib.Path], None] + expected: tuple[str, ...] + + +def _empty_config(directory: pathlib.Path) -> None: + write_config(directory / "a.yaml", "# nothing here\n") + + +def _directory_named_like_a_config(directory: pathlib.Path) -> None: + (directory / "a.yaml").mkdir() + write_config(directory / "b.yaml", "{}\n") + + +def _unreadable(directory: pathlib.Path) -> None: + write_config(directory / "a.yaml", "{}\n") + directory.chmod(0o000) + + +SCOPE_DIR_FIXTURES: list[ScopeDirFixture] = [ + ScopeDirFixture( + test_id="comment-only-file-is-listed", + build=_empty_config, + expected=("a.yaml",), + ), + ScopeDirFixture( + test_id="directory-named-like-a-config-is-skipped", + build=_directory_named_like_a_config, + expected=("b.yaml",), + ), + ScopeDirFixture( + test_id="unreadable-directory-yields-nothing", + build=_unreadable, + expected=(), + ), +] + + +@pytest.mark.skipif(os.getuid() == 0, reason="root ignores directory permissions") +@pytest.mark.parametrize( + list(ScopeDirFixture._fields), + SCOPE_DIR_FIXTURES, + ids=[fixture.test_id for fixture in SCOPE_DIR_FIXTURES], +) +def test_scope_directory_survives_hostile_contents( + test_id: str, + build: t.Callable[[pathlib.Path], None], + expected: tuple[str, ...], + tmp_path: pathlib.Path, +) -> None: + """Listing a scope directory never raises, whatever is in it.""" + scope_dir = tmp_path / "scope" + scope_dir.mkdir() + build(scope_dir) + try: + assert ( + tuple(path.name for path in scopes._config_files_in(scope_dir)) == expected + ) + finally: + scope_dir.chmod(0o755) + + +def test_comment_only_project_config_loads_as_empty( + scoped_tree: pathlib.Path, +) -> None: + """A ``.vcspull.yaml`` with nothing in it contributes nothing.""" + write_config(scoped_tree / ".vcspull.yaml", "# nothing yet\n") + + assert "inner" in _by_name(config.load_scoped_configs(cwd=scoped_tree))