diff --git a/changelog.d/add-from-credentials-redacted.md b/changelog.d/add-from-credentials-redacted.md new file mode 100644 index 00000000..8a4bf807 --- /dev/null +++ b/changelog.d/add-from-credentials-redacted.md @@ -0,0 +1,2 @@ +- **Hardening: `chock add --from` never stores or prints a URL's credentials.** The `user:token@` part of a + catalog URL is dropped before the source is written to `chock.lock`, printed, or put in an error message. diff --git a/changelog.d/add-locate-contained.md b/changelog.d/add-locate-contained.md new file mode 100644 index 00000000..a4e5bc25 --- /dev/null +++ b/changelog.d/add-locate-contained.md @@ -0,0 +1,3 @@ +- **Hardening: `chock add` finds a pack only inside the catalog.** The fallback lookup (`/`) now + applies the same containment check as a `registry.yaml` entry, a pack directory that is itself a symlink + is refused, and the copy never follows a link. diff --git a/changelog.d/catalog-url-never-read-as-path.md b/changelog.d/catalog-url-never-read-as-path.md new file mode 100644 index 00000000..bfd1741c --- /dev/null +++ b/changelog.d/catalog-url-never-read-as-path.md @@ -0,0 +1,3 @@ +- **Hardening: a catalog URL is never read as a path in the working directory.** A source spelled as a URL + or `user@host:path` is always fetched from the remote, and its published-branch checks always run, even + when a directory of the same shape exists under the current directory. A local catalog is a path. diff --git a/changelog.d/eval-export-validates-ids.md b/changelog.d/eval-export-validates-ids.md new file mode 100644 index 00000000..1afa226f --- /dev/null +++ b/changelog.d/eval-export-validates-ids.md @@ -0,0 +1,2 @@ +- **Hardening: `chock eval export` writes only under `--out`.** A policy whose manifest `id` is not a valid + policy id (`^[a-z][a-z0-9-]{2,63}$`) is reported as an error and exports nothing. diff --git a/changelog.d/plugin-build-refuses-links.md b/changelog.d/plugin-build-refuses-links.md new file mode 100644 index 00000000..fef26eb6 --- /dev/null +++ b/changelog.d/plugin-build-refuses-links.md @@ -0,0 +1,2 @@ +- **Hardening: `chock plugin build` in place refuses a symlink under the policy tree.** The build no longer + writes through a link at any path it generates; it stops with an error naming the link and writes nothing. diff --git a/changelog.d/review-checks-from-base.md b/changelog.d/review-checks-from-base.md new file mode 100644 index 00000000..0afcf5c7 --- /dev/null +++ b/changelog.d/review-checks-from-base.md @@ -0,0 +1,5 @@ +- **Hardening: `chock review` defines checks from the base only.** With a base ref, the built-in checks and + the checks the base's `.chock/config.yaml` declares are the whole registry. A check a pull request adds + to its own config, or a built-in it redefines there, is neither run nor part of `command_set_hash`. + A check a pull request both adds and requires is not required of that pull request: `chock review require` + prints that it is required from the merge on. Without a base ref the working tree still defines. diff --git a/changelog.d/sync-refuses-linked-agent-config.md b/changelog.d/sync-refuses-linked-agent-config.md new file mode 100644 index 00000000..3f719add --- /dev/null +++ b/changelog.d/sync-refuses-linked-agent-config.md @@ -0,0 +1,4 @@ +- **Hardening: `chock sync` refuses an agent config that is a symlink.** Writing hooks into + `.claude/settings.json` and the other merged vendor configs now goes through the same no-follow writer as + the MCP registration: a link at the file or any directory above it stops the sync, and nothing is written. + The merged config file is now written with owner-only permissions. diff --git a/changelog.d/yaml-bombs-bounded.md b/changelog.d/yaml-bombs-bounded.md new file mode 100644 index 00000000..16a5c763 --- /dev/null +++ b/changelog.d/yaml-bombs-bounded.md @@ -0,0 +1,4 @@ +- **Hardening: repo YAML cannot hang or crash the guidance server or `chock check`.** Only string entries + of `policies.disabled` are read (an alias-built nested value is no longer stringified), and + the guidance config's nesting limit is measured on parser events, so compact block nesting is refused + like bracket and indent nesting. diff --git a/docs/reviewer-evidence.md b/docs/reviewer-evidence.md index 04afb978..cc17523f 100644 --- a/docs/reviewer-evidence.md +++ b/docs/reviewer-evidence.md @@ -90,8 +90,10 @@ chock: unattestable_paths: [tools/, .github/workflows/] ``` -Config is safe as a source here for the reason evidence is not: it is committed, and CODEOWNERS -routes it for review. An unknown `check` fails verification rather than being skipped — a claim +Only the base branch's config defines a check: a check a pull request adds to its own config is not +run, and a built-in it redefines there keeps the built-in's command. A new check takes effect once it +is merged. Config is safe as a source here for the reason evidence is not: it is committed, and +CODEOWNERS routes it for review. An unknown `check` fails verification rather than being skipped — a claim nobody can re-derive must not pass quietly. ## `chock review require` -- is this PR merge-ready? @@ -144,9 +146,12 @@ judgement that is silent when nothing is declared stays silent, exactly like `un A pull request edits `.chock/config.yaml` like any other file, so `emit`, `verify` and `require` judge by the **stricter of the base branch's policy and the head's**, per setting: the required -set and `unattestable_paths` are the union, `attestation_floor` the higher, and a check both -define runs as the base defines it. Tightening in a PR applies at once; loosening applies after -it merges, in a PR a human approves for that reason (`review/policy.py`). +set and `unattestable_paths` are the union, and `attestation_floor` the higher. Checks are the +exception: only the base defines them, and nothing the head declares ever runs. A check the base +defines and the head requires applies at once. A check the head both adds and requires is not +required of that pull request; `require` says so (` is added by this change and is required from +the merge on`) and it is required from the merge on. Loosening applies after it merges, in a PR a +human approves for that reason (`review/policy.py`). `applies_to` is not enforced by `require` itself; it is there for your own workflow's `if:` condition (§docs/adopting.md), because requiring evidence from everyone is friction that lands on volunteers. diff --git a/src/chock/config.py b/src/chock/config.py index 43425870..86e47cc3 100644 --- a/src/chock/config.py +++ b/src/chock/config.py @@ -96,7 +96,7 @@ def _disabled_list(config: dict[str, Any]) -> set[str]: disabled = _policies(config).get("disabled") if isinstance(disabled, str): return {disabled} - return {str(p) for p in disabled} if isinstance(disabled, list) else set() + return {p for p in disabled if isinstance(p, str)} if isinstance(disabled, list) else set() def policy_status( diff --git a/src/chock/eval/context_report.py b/src/chock/eval/context_report.py index 3d38cb8c..dea69d12 100644 --- a/src/chock/eval/context_report.py +++ b/src/chock/eval/context_report.py @@ -138,6 +138,9 @@ def _frontmatter(data: dict[str, Any], body: str) -> str: return "---\n" + yaml.safe_dump(data, sort_keys=False) + "---\n" + body.rstrip() + "\n" +POLICY_ID = re.compile(r"[a-z][a-z0-9-]{2,63}") + + def _write(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) write_generated(path, text) @@ -210,6 +213,9 @@ def export(repo_root: Path, out_dir: Path, policy_ids: list[str] | None = None) result = ExportResult() policies = _resolve(repo_root, policy_ids, result) for policy in policies: + if not POLICY_ID.fullmatch(policy.id): + result.errors.append(f"{policy.id!r}: not a valid policy id ({POLICY_ID.pattern}); nothing exported") + continue try: notice = _export_one(policy, repo_root, out_dir) except SuiteShapeError as exc: diff --git a/src/chock/guidance/advise.py b/src/chock/guidance/advise.py index 1ac15a98..4bbe34c3 100644 --- a/src/chock/guidance/advise.py +++ b/src/chock/guidance/advise.py @@ -61,7 +61,9 @@ def _too_deep(text: str) -> bool: return True elif char in "]}": depth = max(depth - 1, 0) - return any(len(line) - len(line.lstrip(" ")) > 2 * _MAX_DEPTH for line in text.splitlines()) + if any(len(line) - len(line.lstrip(" ")) > 2 * _MAX_DEPTH for line in text.splitlines()): + return True + return yamlio.nests_deeper_than(text, _MAX_DEPTH) def _parsed(text: str | None, cap: int) -> dict: diff --git a/src/chock/hooks/in_agent_merged.py b/src/chock/hooks/in_agent_merged.py index e6c1d10a..dfba57fc 100644 --- a/src/chock/hooks/in_agent_merged.py +++ b/src/chock/hooks/in_agent_merged.py @@ -15,10 +15,12 @@ from typing import NamedTuple from chock import vendors -from chock.emit import write_generated_json from chock.hooks import heartbeat, stop_check from chock.hooks.in_agent_generic import load_config as _load_config from chock.hooks.runtime_vendor import owned_markers, runtime_rel, vendor_runtime +from chock.scaffold.mcp_write import _lstat_checked, write_file + +_CONFIG_MODE = 0o600 class Wiring(NamedTuple): @@ -186,7 +188,9 @@ def install_merged(repo_root: Path, vendor: str, *, uninstall: bool = False) -> repo_root = Path(repo_root) markers = owned_markers(vendor) - config_path = repo_root / vendors.config_path(vendor) + config_rel = Path(vendors.config_path(vendor)) + config_path = repo_root / config_rel + _lstat_checked(repo_root, config_rel.as_posix()) settings = _load_config(config_path) hooks = settings.setdefault("hooks", {}) if isinstance(settings.get("hooks", {}), dict) else {} settings["hooks"] = hooks @@ -214,8 +218,7 @@ def install_merged(repo_root: Path, vendor: str, *, uninstall: bool = False) -> if not vendor_wiring.write_when_absent and not hooks and not config_path.exists(): return [] - config_path.parent.mkdir(parents=True, exist_ok=True) - write_generated_json(config_path, settings) + write_file(repo_root, config_rel.as_posix(), json.dumps(settings, indent=2).encode("utf-8"), _CONFIG_MODE) return reported diff --git a/src/chock/install/trust.py b/src/chock/install/trust.py index 9c70637b..8384d076 100644 --- a/src/chock/install/trust.py +++ b/src/chock/install/trust.py @@ -8,7 +8,7 @@ from chock.install import package from chock.install.resolve import REFUSED from chock.scaffold.add import IntegrityError -from chock.scaffold.pin import INERT, REMOTE_TIMEOUT, _run +from chock.scaffold.pin import INERT, REMOTE_TIMEOUT, _run, local_source def check_source(source: str, allowed: list[str]) -> None: @@ -25,8 +25,8 @@ def check_source(source: str, allowed: list[str]) -> None: def require_on_branch(source: str, sha: str, into: Path) -> None: """Refuse `sha` unless it is reachable from the source's published branch (a fork's commit is not).""" branch = package.settings()["published_branch"] - local = Path(source).expanduser() - remote = str(local.resolve()) if local.exists() else source + local = local_source(source) + remote = str(local.resolve()) if local is not None else source clone = ["git", *INERT, "clone", "--quiet", "--bare", "--filter=blob:none", "--single-branch", "--branch"] result = _run([*clone, branch, "--", remote, str(into)], REMOTE_TIMEOUT) if result.returncode != 0: diff --git a/src/chock/lock.py b/src/chock/lock.py index a9ec1f8b..5faf5482 100644 --- a/src/chock/lock.py +++ b/src/chock/lock.py @@ -10,6 +10,7 @@ from typing import Any from chock.policies import discover_policy_dirs +from chock.scaffold.pin import is_remote_source from chock.vendored import vendored_differences LOCKFILE_NAME = "chock.lock" @@ -114,6 +115,8 @@ def is_local_source(source: object, repo_root: Path) -> bool: return False if source in {LOCAL_SOURCE, "."}: return True + if is_remote_source(source): + return False try: return Path(source).expanduser().resolve() == repo_root.resolve() except (OSError, RuntimeError): diff --git a/src/chock/plugin/build.py b/src/chock/plugin/build.py index d6ffe58c..b832e54c 100644 --- a/src/chock/plugin/build.py +++ b/src/chock/plugin/build.py @@ -38,7 +38,7 @@ class PluginNameError(ValueError): - """A policy id that cannot be a conformant plugin name.""" + """A policy id that is not a conformant plugin name, or a plugin path that is a link.""" def plugin_name(policy_id: str) -> str: @@ -229,13 +229,26 @@ def plugin_files( return files +def _reject_links(target: Path, rel: Path) -> None: + """Refuse a destination that is, or sits under, a symlink inside the target.""" + here = Path(target) + for part in rel.parts: + here = here / part + if here.is_symlink(): + msg = f"{here} is a symlink; refusing to write through it. Remove the link and build again." + raise PluginNameError(msg) + + def build_plugin( policy_dir: Path, manifest: dict[str, Any], repo_root: Path, out_dir: Path | None = None ) -> list[Path]: """Write the Agent Plugins package for one policy. Defaults to in place.""" target = Path(out_dir) if out_dir else Path(policy_dir) written: list[Path] = [] - for rel, content in plugin_files(Path(policy_dir), manifest, Path(repo_root), packaged=out_dir is not None).items(): + files = plugin_files(Path(policy_dir), manifest, Path(repo_root), packaged=out_dir is not None) + for rel in files: + _reject_links(target, rel) + for rel, content in files.items(): dest = target / rel dest.parent.mkdir(parents=True, exist_ok=True) write_generated(dest, content) diff --git a/src/chock/review/cli.py b/src/chock/review/cli.py index b6f2c901..414a46c7 100644 --- a/src/chock/review/cli.py +++ b/src/chock/review/cli.py @@ -19,6 +19,7 @@ required_checks, verify, ) +from chock.review.policy import deferred_checks DEFAULT_BASE = "origin/main" @@ -93,6 +94,8 @@ def _verify(args: argparse.Namespace) -> int: def _require(args: argparse.Namespace) -> int: root = Path(args.repo).resolve() failures = require(root, args.base) + for name in deferred_checks(root, args.base): + print(f"{name} is added by this change and is required from the merge on.", file=sys.stderr) if failures: print(f"PR is not merge-ready ({len(failures)} problem(s)):", file=sys.stderr) for failure in failures: diff --git a/src/chock/review/policy.py b/src/chock/review/policy.py index b8a3df97..4e598a1c 100644 --- a/src/chock/review/policy.py +++ b/src/chock/review/policy.py @@ -74,14 +74,35 @@ def unattestable_paths(root: Path, base_ref: str | None = None) -> list[str]: return sorted(paths) +def _required_names(review: dict[str, Any] | None) -> set[str]: + configured = (review or {}).get("required_checks") + return {str(n) for n in configured} if isinstance(configured, list) else set() + + +def _base_and_head_required(root: Path, base_ref: str | None) -> tuple[set[str], set[str]]: + """(names the base requires, names the head requires); without a base ref the working tree is the head.""" + head = _required_names(_review_section(load_config(root))) + return (_required_names(_review_at(root, base_ref)) if base_ref else set()), head + + def required_checks(root: Path, base_ref: str | None = None) -> list[str]: - """The repository's own required set, unioned across base and head. Never from evidence.""" - names: set[str] = set() - for review in review_policies(root, base_ref): - configured = review.get("required_checks") - if isinstance(configured, list): - names |= {str(n) for n in configured} - return sorted(names) + """The repository's own required set, unioned across base and head. Never from evidence. + + With a base ref, a name only the head requires and the base does not define is not required yet: see + `deferred_checks`. A name the base requires stays required, defined or not (an undefined one fails as unknown). + """ + base, head = _base_and_head_required(root, base_ref) + if base_ref: + head &= set(check_registry(root, base_ref)) | base + return sorted(base | head) + + +def deferred_checks(root: Path, base_ref: str | None = None) -> list[str]: + """Checks only the head requires and the base does not define: a change that adds one is bound from the merge on.""" + if not base_ref: + return [] + base, head = _base_and_head_required(root, base_ref) + return sorted((head - base) - set(check_registry(root, base_ref))) def attestation_floor(root: Path, base_ref: str | None = None) -> int: @@ -105,9 +126,19 @@ def command_set_hash(root: Path, base_ref: str | None = None) -> str: def check_registry(root: Path, base_ref: str | None = None) -> dict[str, list[str]]: - """Built-in checks plus any the repository declares; a check the base defines runs as the base defines it.""" + """The built-in checks plus the ones the base declares; with a base, nothing the head declares is ever defined. + + A pull request edits `.chock/config.yaml`, so a built-in redefined there, or a check added there, would + run the author's command inside the gate that is judging the author. Without a base ref there is no + other side, and the working tree's own config defines. + """ registry = dict(BUILTIN_CHECKS) - for review in reversed(review_policies(root, base_ref)): + if not base_ref: + defining = [_review_section(load_config(root))] + else: + base = _review_at(root, base_ref) + defining = [] if base is None else [base] + for review in defining: checks = review.get("checks") for name, argv in checks.items() if isinstance(checks, dict) else (): if isinstance(argv, list) and all(isinstance(a, str) for a in argv): diff --git a/src/chock/scaffold/add.py b/src/chock/scaffold/add.py index 53b41203..da36dc3e 100644 --- a/src/chock/scaffold/add.py +++ b/src/chock/scaffold/add.py @@ -16,7 +16,7 @@ from chock.config import agents_from_config as _agents_from_config from chock.lock import LOCAL_SOURCE, compute_pack_hash, is_local_source, pack_bytecode, read_lock, write_lock from chock.output import error -from chock.scaffold.pin import PinError, fetch_catalog +from chock.scaffold.pin import PinError, fetch_catalog, redact_source from chock.scaffold.recompile import BookkeepingError, recompile @@ -42,7 +42,7 @@ def _reject_unsafe_id(artifact_id: str) -> None: def _reject_symlinks(pack_dir: Path, artifact_id: str) -> None: """Refuse a pack carrying a symlink: copying it would read a file the catalog never held.""" - links = [] + links = [artifact_id] if pack_dir.is_symlink() else [] for parent, dirs, files in os.walk(pack_dir): links += [ Path(parent, n).relative_to(pack_dir).as_posix() for n in dirs + files if Path(parent, n).is_symlink() @@ -94,6 +94,8 @@ def locate(catalog_root: Path, artifact_id: str) -> tuple[Path, Path]: for area, dest in _AREAS.items(): candidate = catalog_root / area / artifact_id + if not candidate.resolve().is_relative_to(catalog_resolved): + continue if (candidate / "manifest.yaml").exists() or (candidate / "SKILL.md").exists(): return candidate, dest @@ -145,7 +147,7 @@ def add( if dest.exists(): shutil.rmtree(dest) dest.parent.mkdir(parents=True, exist_ok=True) - shutil.copytree(src, dest) + shutil.copytree(src, dest, symlinks=True) return Added(path=dest, sha256=digest, commit=commit) @@ -160,7 +162,7 @@ def record_provenance(repo_root: Path, artifact_id: str, source: str, ref: str | entry.pop("source_commit", None) entry.pop("source_ref", None) else: - entry["source"] = source + entry["source"] = redact_source(source) entry["source_commit"] = added.commit if ref: entry["source_ref"] = ref @@ -200,7 +202,7 @@ def main(argv: list[str] | None = None) -> int: print(f"Added {args.artifact_id} to {rel}") if added.commit: - print(f" from {args.source} at {added.commit}") + print(f" from {redact_source(args.source)} at {added.commit}") print(f" sha256 {added.sha256}") if not args.ref and added.commit: pin = f"Pin with --ref {added.commit}, or --verify-sha {added.sha256}" diff --git a/src/chock/scaffold/pin.py b/src/chock/scaffold/pin.py index f3a7d9f7..ab3cb775 100644 --- a/src/chock/scaffold/pin.py +++ b/src/chock/scaffold/pin.py @@ -10,6 +10,7 @@ import sys from importlib import metadata from pathlib import Path +from urllib.parse import urlsplit, urlunsplit import chock @@ -93,8 +94,22 @@ def _in(into: Path, *args: str) -> list[str]: return ["git", f"--git-dir={into / '.git'}", f"--work-tree={into}", *INERT, *args] +def redact_source(source: str) -> str: + """The source without the userinfo of a URL: a token typed into `--from` is never stored or printed.""" + if not is_remote_source(source): + return source + parts = urlsplit(source) + if "@" not in parts.netloc or not parts.scheme: + return source + return urlunsplit(parts._replace(netloc=parts.netloc.rpartition("@")[2])) + + +def _scrub(text: str, source: str) -> str: + return text.replace(source, redact_source(source)) + + def _pin_error(source: str, ref: str, why: str) -> PinError: - return PinError(f"chock add: refusing catalog {source} at {ref!r}: {why} (nothing installed)") + return PinError(f"chock add: refusing catalog {redact_source(source)} at {ref!r}: {why} (nothing installed)") def _check_ref_syntax(source: str, ref: str) -> None: @@ -120,7 +135,7 @@ def _fetch_commit(source: str, remote: str, sha: str, into: Path) -> str: except RuntimeError as exc: raise _pin_error(source, sha, str(exc)) from exc if result.returncode != 0: - detail = result.stderr.strip() or "git step failed" + detail = _scrub(result.stderr.strip(), source) or "git step failed" raise _pin_error(source, sha, f"commit not available from the remote by id ({detail})") head = _run(_in(into, "rev-parse", "HEAD")) if head.returncode != 0 or head.stdout.strip().lower() != sha: @@ -141,7 +156,7 @@ def _require_published(source: str, remote: str, sha: str, into: Path) -> None: except RuntimeError as exc: raise _pin_error(source, sha, str(exc)) from exc if found.returncode != 0 or not found.stdout.strip(): - why = f"commit {sha} is not on any branch or tag of {source}; a fork's or a pull request's commit is refused" + why = f"commit {sha} is not on any branch or tag of {redact_source(source)}; a fork's or a pull request's commit is refused" raise _pin_error(source, sha, why) @@ -151,7 +166,11 @@ def _clone_ref(source: str, remote: str, ref: str | None, into: Path) -> str | N args += ["--branch", ref] result = _run([*args, "--", remote, str(into)], REMOTE_TIMEOUT) if result.returncode != 0: - msg = f"could not fetch catalog {source}" + (f" at {ref}" if ref else "") + f":\n{result.stderr.strip()}" + msg = ( + f"could not fetch catalog {redact_source(source)}" + + (f" at {ref}" if ref else "") + + f":\n{_scrub(result.stderr.strip(), source)}" + ) raise RuntimeError(msg) resolved = _run(_in(into, "rev-parse", "HEAD")) return (resolved.stdout.strip() or None) if resolved.returncode == 0 else None @@ -204,20 +223,36 @@ def _snapshot_local(source: str, local: Path, into: Path) -> tuple[Path, str | N return into / prefix, head +_REMOTE_FORM = re.compile(r"^(?:[A-Za-z][A-Za-z0-9+.-]*://|[^/\\:@\s]+@[^/\\:\s]+:)") + + +def is_remote_source(source: str) -> bool: + """True when the text is spelled as a URL or scp-style address: never read from disk, whatever exists there.""" + return bool(_REMOTE_FORM.match(source)) + + +def local_source(source: str) -> Path | None: + """The existing local path a source names, or None for a remote spelling or a path that is not there.""" + if is_remote_source(source): + return None + path = Path(source).expanduser() + return path if path.exists() else None + + def fetch_catalog(source: str, ref: str | None, into: Path) -> tuple[Path, str | None]: """Make the catalog available locally. Returns (root, resolved commit or None). The returned tree holds the bytes a clone of the catalog holds: a local checkout is cloned at its HEAD (uncommitted edits are left out), a plain directory is copied with its text files normalised to LF. """ - local = Path(source).expanduser() - if local.is_dir() and not ref: + local = local_source(source) + if local is not None and local.is_dir() and not ref: return _snapshot_local(source, local.resolve(), into) - remote = str(local.resolve()) if local.exists() else source + remote = str(local.resolve()) if local is not None else source if ref and _FULL_SHA.fullmatch(ref): sha = _fetch_commit(source, remote, ref.lower(), into) - if not local.exists(): + if local is None: _require_published(source, remote, sha, into) return into, sha if ref and _SHORT_SHA.fullmatch(ref): diff --git a/src/chock/yamlio.py b/src/chock/yamlio.py index d017fff0..b02cfc3d 100644 --- a/src/chock/yamlio.py +++ b/src/chock/yamlio.py @@ -24,3 +24,24 @@ def safe_load(source: str | IO[str]) -> Any: """`yaml.safe_load`, with a private copy of the result so no caller sees another's edits.""" text = source if isinstance(source, str) else source.read() return copy.deepcopy(_parsed(text)) + + +def nests_deeper_than(text: str, limit: int) -> bool: + """Whether any node of `text` nests deeper than `limit`, counted from parser events before anything is composed. + + Libyaml's composer can crash the process on deep nesting of any spelling (flow, indented or compact + block), so the depth is measured on the event stream, which it keeps on the heap. Text the parser + rejects is reported as not too deep: the loader then raises its own error for it. + """ + depth = 0 + try: + for event in yaml.parse(text, Loader=SAFE_LOADER): + if isinstance(event, (yaml.SequenceStartEvent, yaml.MappingStartEvent)): + depth += 1 + if depth > limit: + return True + elif isinstance(event, (yaml.SequenceEndEvent, yaml.MappingEndEvent)): + depth -= 1 + except yaml.YAMLError: + return False + return False diff --git a/tests/test_add_locate_containment.py b/tests/test_add_locate_containment.py new file mode 100644 index 00000000..f4186420 --- /dev/null +++ b/tests/test_add_locate_containment.py @@ -0,0 +1,53 @@ +"""A catalog's linked area or pack directory never lets `chock add` copy files from outside the catalog.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from chock.scaffold.add import IntegrityError, locate, verified_pack + + +def _outside(tmp_path: Path) -> Path: + elsewhere = tmp_path / "elsewhere" + (elsewhere / "evil-skill").mkdir(parents=True) + (elsewhere / "evil-skill" / "SKILL.md").write_text("---\nname: evil-skill\n---\n", encoding="utf-8") + return elsewhere + + +def test_a_linked_area_directory_is_not_followed(tmp_path: Path) -> None: + catalog = tmp_path / "catalog" + catalog.mkdir() + (catalog / "skills").symlink_to(_outside(tmp_path), target_is_directory=True) + + with pytest.raises(FileNotFoundError): + locate(catalog, "evil-skill") + + +def test_a_linked_pack_directory_is_not_followed(tmp_path: Path) -> None: + catalog = tmp_path / "catalog" + (catalog / "skills").mkdir(parents=True) + (catalog / "skills" / "evil-skill").symlink_to(_outside(tmp_path) / "evil-skill", target_is_directory=True) + + with pytest.raises(FileNotFoundError): + locate(catalog, "evil-skill") + + +def test_verified_pack_refuses_a_linked_pack_directory(tmp_path: Path) -> None: + catalog = tmp_path / "catalog" + (catalog / "skills").mkdir(parents=True) + (catalog / "skills" / "evil-skill").symlink_to(_outside(tmp_path) / "evil-skill", target_is_directory=True) + + with pytest.raises((FileNotFoundError, IntegrityError)): + verified_pack(catalog, "evil-skill", None) + + +def test_an_ordinary_pack_is_still_found(tmp_path: Path) -> None: + catalog = tmp_path / "catalog" + (catalog / "skills" / "good-skill").mkdir(parents=True) + (catalog / "skills" / "good-skill" / "SKILL.md").write_text("---\nname: good-skill\n---\n", encoding="utf-8") + + src, _ = locate(catalog, "good-skill") + + assert src == catalog / "skills" / "good-skill" diff --git a/tests/test_add_source_credentials.py b/tests/test_add_source_credentials.py new file mode 100644 index 00000000..0fa1fbbd --- /dev/null +++ b/tests/test_add_source_credentials.py @@ -0,0 +1,44 @@ +"""A catalog URL's credentials never reach chock.lock, the console, or an error message.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from chock.lock import write_lock +from chock.scaffold import pin +from chock.scaffold.add import Added, record_provenance + +TOKEN = "s3cr3t-token" +WITH_CREDENTIALS = f"https://x-access-token:{TOKEN}@github.com/org/catalog" +CLEAN = "https://github.com/org/catalog" + + +def test_the_lock_records_the_url_without_userinfo(tmp_path: Path) -> None: + write_lock({"lockfile_version": 1, "engine": "x", "packs": [{"id": "p", "source": "local"}]}, tmp_path) + + record_provenance(tmp_path, "p", WITH_CREDENTIALS, None, Added(path=tmp_path, sha256="0" * 64, commit="a" * 40)) + + text = (tmp_path / "chock.lock").read_text(encoding="utf-8") + assert TOKEN not in text + assert json.loads(text)["packs"][0]["source"] == CLEAN + + +@pytest.mark.parametrize( + ("given", "shown"), + [ + (WITH_CREDENTIALS, CLEAN), + (f"https://{TOKEN}@github.com:8443/org/catalog", "https://github.com:8443/org/catalog"), + (CLEAN, CLEAN), + ("git@github.com:org/catalog.git", "git@github.com:org/catalog.git"), + ("./local/catalog", "./local/catalog"), + ], +) +def test_redact_source_drops_only_userinfo(given: str, shown: str) -> None: + assert pin.redact_source(given) == shown + + +def test_a_pin_error_does_not_echo_credentials() -> None: + assert TOKEN not in str(pin._pin_error(WITH_CREDENTIALS, "a" * 40, "not available")) diff --git a/tests/test_catalog_source_is_a_url.py b/tests/test_catalog_source_is_a_url.py new file mode 100644 index 00000000..41d6c77b --- /dev/null +++ b/tests/test_catalog_source_is_a_url.py @@ -0,0 +1,81 @@ +"""A catalog URL is a URL: a directory of the same shape under the working directory never replaces it.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from chock.install import trust +from chock.scaffold import pin + +URL = "https://github.com/open-coder-ai/chock-catalog" +SCP = "git@github.com:open-coder-ai/chock-catalog.git" + + +def _plant(root: Path, source: str) -> None: + planted = root / Path(source) + planted.mkdir(parents=True) + (planted / "registry.yaml").write_text("planted: true\n", encoding="utf-8") + + +@pytest.mark.parametrize("source", [URL, SCP]) +def test_fetch_catalog_clones_the_remote_not_a_lookalike_directory( + source: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _plant(tmp_path, source) + monkeypatch.chdir(tmp_path) + seen: list[str] = [] + monkeypatch.setattr(pin, "_clone_ref", lambda _s, remote, _r, _i: seen.append(remote)) + monkeypatch.setattr(pin, "_snapshot_local", lambda *_a: pytest.fail("a lookalike directory was snapshotted")) + + pin.fetch_catalog(source, None, tmp_path / "into") + + assert seen == [source] + + +@pytest.mark.parametrize("source", [URL, SCP]) +def test_a_pinned_remote_is_still_checked_as_published( + source: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _plant(tmp_path, source) + monkeypatch.chdir(tmp_path) + sha = "a" * 40 + checked: list[str] = [] + monkeypatch.setattr(pin, "_fetch_commit", lambda _s, _r, s, _i: s) + monkeypatch.setattr(pin, "_require_published", lambda _s, remote, *_a: checked.append(remote)) + + pin.fetch_catalog(source, sha, tmp_path / "into") + + assert checked == [source] + + +@pytest.mark.parametrize("source", [URL, SCP]) +def test_require_on_branch_reads_the_remote_not_a_lookalike_directory( + source: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _plant(tmp_path, source) + monkeypatch.chdir(tmp_path) + cloned: list[list[str]] = [] + + class Done: + returncode = 1 + stderr = "unreachable" + + monkeypatch.setattr(trust, "_run", lambda cmd, *_a: cloned.append(cmd) or Done()) + + with pytest.raises(trust.IntegrityError): + trust.require_on_branch(source, "b" * 40, tmp_path / "bare") + + assert cloned[0][-2] == source + + +def test_an_explicit_local_path_is_still_local(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + (tmp_path / "cat").mkdir() + (tmp_path / "cat" / "registry.yaml").write_text("x: 1\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(pin, "_snapshot_local", lambda source, _l, _i: (Path(source), None)) + + root, _ = pin.fetch_catalog("./cat", None, tmp_path / "into") + + assert root == Path("./cat") diff --git a/tests/test_context_report_export_ids.py b/tests/test_context_report_export_ids.py new file mode 100644 index 00000000..5211a4e0 --- /dev/null +++ b/tests/test_context_report_export_ids.py @@ -0,0 +1,45 @@ +"""`chock eval export` writes under --out only: a manifest id that is not a policy id is an error, not a path.""" + +from __future__ import annotations + +import re +import shutil +from pathlib import Path + +import pytest +from conftest import baseline_policy + +from chock.eval import context_report as cr + + +def _repo_with_id(tmp_path: Path, manifest_id: str) -> Path: + dest = tmp_path / "repo" / ".agents" / "policies" / "git-safety" + dest.parent.mkdir(parents=True) + shutil.copytree(baseline_policy("git-safety"), dest) + manifest = dest / "manifest.yaml" + text = manifest.read_text(encoding="utf-8") + manifest.write_text(re.sub(r"(?m)^id:.*$", f"id: '{manifest_id}'", text, count=1), encoding="utf-8") + return tmp_path / "repo" + + +@pytest.mark.parametrize("bad", ["../../escaped", "ABSOLUTE", "a/b", "UPPER", "x"]) +def test_a_manifest_id_that_is_not_a_policy_id_exports_nothing(tmp_path: Path, bad: str) -> None: + bad = str(tmp_path / "work" / "escaped") if bad == "ABSOLUTE" else bad + repo = _repo_with_id(tmp_path, bad) + out = tmp_path / "work" / "out" + + result = cr.export(repo, out) + + assert result.exported == [] + assert any("not a valid policy id" in e for e in result.errors) + assert not (tmp_path / "escaped.md").exists() + assert not (tmp_path / "work" / "escaped.md").exists() + assert not out.exists() or not any(out.rglob("*.md")) + + +def test_a_valid_id_still_exports(tmp_path: Path) -> None: + repo = _repo_with_id(tmp_path, "git-safety") + + result = cr.export(repo, tmp_path / "out") + + assert result.exported == ["git-safety"] diff --git a/tests/test_install_merged_links.py b/tests/test_install_merged_links.py new file mode 100644 index 00000000..5b90fdb0 --- /dev/null +++ b/tests/test_install_merged_links.py @@ -0,0 +1,52 @@ +"""`sync` never writes chock's hook entries through a linked agent config.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from chock import vendors +from chock.hooks.in_agent_merged import install_merged +from chock.scaffold.mcp_write import McpConfigError + +VENDOR = "claude_code" + + +def test_a_linked_config_file_is_refused_and_left_alone(tmp_path: Path) -> None: + outside = tmp_path / "global-settings.json" + outside.write_text('{"keep": true}', encoding="utf-8") + repo = tmp_path / "repo" + config = repo / vendors.config_path(VENDOR) + config.parent.mkdir(parents=True) + config.symlink_to(outside) + + with pytest.raises(McpConfigError, match="symlink"): + install_merged(repo, VENDOR) + + assert outside.read_text(encoding="utf-8") == '{"keep": true}' + + +def test_a_linked_config_directory_is_refused(tmp_path: Path) -> None: + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + repo = tmp_path / "repo" + config = repo / vendors.config_path(VENDOR) + config.parent.parent.mkdir(parents=True, exist_ok=True) + config.parent.symlink_to(elsewhere, target_is_directory=True) + + with pytest.raises(McpConfigError, match="symlink"): + install_merged(repo, VENDOR) + + assert list(elsewhere.iterdir()) == [] + + +def test_an_ordinary_config_is_still_merged(tmp_path: Path) -> None: + config = tmp_path / vendors.config_path(VENDOR) + config.parent.mkdir(parents=True) + config.write_text(json.dumps({"keep": True}), encoding="utf-8") + + install_merged(tmp_path, VENDOR) + + assert json.loads(config.read_text(encoding="utf-8"))["keep"] is True diff --git a/tests/test_plugin_build_symlink.py b/tests/test_plugin_build_symlink.py new file mode 100644 index 00000000..0317b0c0 --- /dev/null +++ b/tests/test_plugin_build_symlink.py @@ -0,0 +1,53 @@ +"""`chock plugin build` in place never writes through a link the policy tree carries.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from chock.plugin.build import PluginNameError, build_plugin + +MANIFEST = {"id": "evil-policy", "name": "Evil", "version": "0.0.1", "description": "d", "artifact": "rule"} + + +def _policy(repo: Path) -> Path: + pack = repo / ".agents" / "policies" / "evil-policy" + (pack / "skill").mkdir(parents=True) + (pack / "manifest.yaml").write_text(yaml.safe_dump(MANIFEST), encoding="utf-8") + (pack / "skill" / "config").write_text("planted\n", encoding="utf-8") + return pack + + +def test_a_linked_skill_directory_is_refused(tmp_path: Path) -> None: + pack = _policy(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + (pack / "skills").mkdir() + (pack / "skills" / "evil-policy").symlink_to(outside, target_is_directory=True) + + with pytest.raises(PluginNameError, match="symlink"): + build_plugin(pack, MANIFEST, tmp_path) + + assert list(outside.iterdir()) == [] + + +def test_a_linked_destination_file_is_refused(tmp_path: Path) -> None: + pack = _policy(tmp_path) + victim = tmp_path / "victim" + victim.write_text("keep\n", encoding="utf-8") + (pack / "plugin.json").symlink_to(victim) + + with pytest.raises(PluginNameError, match="symlink"): + build_plugin(pack, MANIFEST, tmp_path) + + assert victim.read_text(encoding="utf-8") == "keep\n" + + +def test_a_clean_tree_still_builds(tmp_path: Path) -> None: + pack = _policy(tmp_path) + + build_plugin(pack, MANIFEST, tmp_path) + + assert (pack / "skills" / "evil-policy" / "config").read_text(encoding="utf-8") == "planted\n" diff --git a/tests/test_review_checks_from_base.py b/tests/test_review_checks_from_base.py new file mode 100644 index 00000000..e56277d0 --- /dev/null +++ b/tests/test_review_checks_from_base.py @@ -0,0 +1,114 @@ +"""Only the base defines the checks `chock review` runs; a PR that adds and requires one is bound from the merge on.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from test_review_require import _commit_base_config, _git, _write_config, _write_evidence + +from chock.review.evidence import EvidenceError, build, command_set_hash, require, required_checks +from chock.review.policy import BUILTIN_CHECKS, check_registry, deferred_checks + + +@pytest.fixture +def repo(tmp_path: Path) -> Path: + """A base commit on `main` and one change on a branch.""" + root = tmp_path / "repo" + root.mkdir() + _git(root, "init", "--quiet", ".") + _git(root, "config", "user.email", "a@b.c") + _git(root, "config", "user.name", "t") + _git(root, "symbolic-ref", "HEAD", "refs/heads/main") + (root / "README.md").write_text("base\n", encoding="utf-8") + _git(root, "add", "-A") + _git(root, "-c", "core.hooksPath=/dev/null", "commit", "-q", "-m", "base") + _git(root, "checkout", "-q", "-b", "feat/x") + (root / "feature.txt").write_text("change\n", encoding="utf-8") + _git(root, "add", "-A") + _git(root, "-c", "core.hooksPath=/dev/null", "commit", "-q", "-m", "change") + return root + + +def test_a_branch_cannot_add_a_check_the_base_does_not_define(repo: Path) -> None: + """A check is code the gate runs: only the base defines one; a head-added check is not run, or even known.""" + marker = repo / "ran.txt" + script = ["python", "-c", f"open({str(marker)!r}, 'w').write('x')"] + _commit_base_config(repo, {"required_checks": ["validate"]}) + _write_config(repo, {"checks": {"extra": script}, "required_checks": ["validate", "extra"]}) + + assert "extra" not in check_registry(repo, "main") + assert required_checks(repo, "main") == ["validate"] + with pytest.raises(EvidenceError, match="unknown check"): + build(repo, "main", {"kind": "agent", "id": "t"}, ["validate", "extra"]) + forged = dict(build(repo, "main", {"kind": "agent", "id": "t"}, ["validate"])) + forged["verified"] = [*forged["verified"], {"check": "extra", "command": "chock x", "result": "pass"}] + _write_evidence(repo, forged) + assert require(repo, "main") != [] + assert not marker.exists() + + +def test_a_branch_may_add_and_require_a_check_it_is_required_from_the_merge_on(repo: Path) -> None: + """The PR that introduces a check is not blocked by it, and the check never runs from the head's definition.""" + marker = repo / "ran.txt" + script = ["python", "-c", f"open({str(marker)!r}, 'w').write('x')"] + _commit_base_config(repo, {"required_checks": ["validate"]}) + _write_config(repo, {"checks": {"extra": script}, "required_checks": ["validate", "extra"]}) + + assert required_checks(repo, "main") == ["validate"] + assert deferred_checks(repo, "main") == ["extra"] + _write_evidence(repo, build(repo, "main", {"kind": "agent", "id": "t"}, ["validate"])) + assert require(repo, "main") == [] + assert not marker.exists() + + +def test_a_check_the_base_requires_but_does_not_define_still_fails_as_unknown(repo: Path) -> None: + _commit_base_config(repo, {"required_checks": ["validate", "ghost"]}) + _write_config(repo, {"required_checks": ["validate", "ghost"]}) + + assert required_checks(repo, "main") == ["ghost", "validate"] + assert deferred_checks(repo, "main") == [] + _write_evidence(repo, build(repo, "main", {"kind": "agent", "id": "t"}, ["validate"])) + assert any("ghost" in f for f in require(repo, "main")) + + +def test_a_check_the_base_defines_and_the_head_requires_applies_at_once(repo: Path) -> None: + _commit_base_config(repo, {"checks": {"mine": ["python", "-c", "pass"]}, "required_checks": ["validate"]}) + _write_config(repo, {"required_checks": ["validate", "mine"]}) + + assert required_checks(repo, "main") == ["mine", "validate"] + assert deferred_checks(repo, "main") == [] + _write_evidence(repo, build(repo, "main", {"kind": "agent", "id": "t"}, ["validate"])) + assert any("missing: mine" in f for f in require(repo, "main")) + + +def test_a_branch_cannot_redefine_a_builtin_the_base_requires(repo: Path) -> None: + """The documented setup lists built-ins in `required_checks` without redefining them; the head cannot.""" + marker = repo / "ran.txt" + script = ["python", "-c", f"open({str(marker)!r}, 'w').write('x')"] + _commit_base_config(repo, {"required_checks": ["validate", "eval"]}) + before = command_set_hash(repo, "main") + _write_config(repo, {"checks": {"validate": script, "eval": script}, "required_checks": ["validate", "eval"]}) + + assert check_registry(repo, "main")["validate"] == BUILTIN_CHECKS["validate"] + assert check_registry(repo, "main")["eval"] == BUILTIN_CHECKS["eval"] + assert command_set_hash(repo, "main") == before + assert not marker.exists() + + +def test_a_head_check_is_not_defined_when_the_base_has_no_config(repo: Path) -> None: + _write_config(repo, {"checks": {"validate": ["python", "-c", "pass"], "extra": ["python", "-c", "pass"]}}) + + registry = check_registry(repo, "main") + + assert registry["validate"] == BUILTIN_CHECKS["validate"] + assert "extra" not in registry + + +def test_the_base_may_define_and_redefine_checks(repo: Path) -> None: + _commit_base_config(repo, {"checks": {"validate": ["validate", "{root}", "--strict"], "mine": ["check"]}}) + + registry = check_registry(repo, "main") + + assert registry["validate"] == ["validate", "{root}", "--strict"] + assert registry["mine"] == ["check"] diff --git a/tests/test_review_exit_codes.py b/tests/test_review_exit_codes.py index 50140c31..b218456e 100644 --- a/tests/test_review_exit_codes.py +++ b/tests/test_review_exit_codes.py @@ -11,11 +11,16 @@ def _git(cwd: Path, *args: str) -> None: subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True) -def _repo_with_commit(tmp_path: Path) -> Path: +def _repo_with_commit(tmp_path: Path, config: str | None = None) -> Path: + """A base commit (carrying the review config, when given: the base defines checks) and one change.""" _git(tmp_path, "init", "-q") _git(tmp_path, "config", "user.email", "t@t") _git(tmp_path, "config", "user.name", "t") (tmp_path / "f.txt").write_text("base\n", encoding="utf-8") + if config is not None: + (tmp_path / ".chock").mkdir() + (tmp_path / ".chock" / "config.yaml").write_text(config, encoding="utf-8") + _git(tmp_path, "add", ".chock/config.yaml") _git(tmp_path, "add", "f.txt") _git(tmp_path, "commit", "-q", "-m", "base") _git(tmp_path, "branch", "basepoint") @@ -27,11 +32,9 @@ def _repo_with_commit(tmp_path: Path) -> Path: def test_emit_exits_nonzero_when_a_check_fails(tmp_path): """H2: a failing check must fail the emit exit code, not just be printed.""" - repo = _repo_with_commit(tmp_path) - (repo / ".chock").mkdir() config = "chock:" + chr(10) + " review:" + chr(10) + " checks:" + chr(10) config += ' always-fails: ["python", "-c", "import sys; sys.exit(1)"]' + chr(10) - (repo / ".chock" / "config.yaml").write_text(config, encoding="utf-8") + repo = _repo_with_commit(tmp_path, config) proc = subprocess.run( [ sys.executable, @@ -56,11 +59,9 @@ def test_emit_exits_nonzero_when_a_check_fails(tmp_path): def test_emit_exits_zero_when_checks_pass(tmp_path): - repo = _repo_with_commit(tmp_path) - (repo / ".chock").mkdir() config = "chock:" + chr(10) + " review:" + chr(10) + " checks:" + chr(10) config += ' always-ok: ["python", "-c", "print(1)"]' + chr(10) - (repo / ".chock" / "config.yaml").write_text(config, encoding="utf-8") + repo = _repo_with_commit(tmp_path, config) proc = subprocess.run( [ sys.executable, diff --git a/tests/test_review_require.py b/tests/test_review_require.py index 7b395abc..f82cdda9 100644 --- a/tests/test_review_require.py +++ b/tests/test_review_require.py @@ -118,7 +118,7 @@ def test_require_rejects_a_stale_command_set_hash_even_when_every_named_check_pa def test_require_rejects_a_failing_required_check_while_verify_still_exits_clean(repo: Path) -> None: """Acceptance #3 (H2): validity and outcome are judged separately.""" - _write_config(repo, {"checks": {"always-fails": SCRIPT_ALWAYS_FAILS}, "required_checks": ["always-fails"]}) + _commit_base_config(repo, {"checks": {"always-fails": SCRIPT_ALWAYS_FAILS}, "required_checks": ["always-fails"]}) failing = build(repo, "main", {"kind": "agent", "id": "test"}, ["always-fails"]) _write_evidence(repo, failing) @@ -196,19 +196,8 @@ def test_a_branch_cannot_lower_the_attestation_floor(repo: Path) -> None: assert unattestable_paths(repo, "main") == ["docs/", "infra/"] -def test_a_branch_may_add_a_check_the_base_does_not_know(repo: Path) -> None: - """Tightening applies at once; only loosening waits for the merge.""" - _commit_base_config(repo, {"required_checks": ["validate"]}) - _write_config(repo, {"checks": {"extra": ["python", "-c", "pass"]}, "required_checks": ["validate", "extra"]}) - - assert required_checks(repo, "main") == ["extra", "validate"] - evidence = build(repo, "main", {"kind": "agent", "id": "t"}, ["validate", "extra"]) - _write_evidence(repo, evidence) - assert require(repo, "main") == [] - - def test_a_recorded_failure_is_never_merge_ready_even_without_a_required_set(repo: Path) -> None: - _write_config(repo, {"checks": {"always-fails": SCRIPT_ALWAYS_FAILS}}) + _commit_base_config(repo, {"checks": {"always-fails": SCRIPT_ALWAYS_FAILS}}) _write_evidence(repo, build(repo, "main", {"kind": "agent", "id": "t"}, ["always-fails"])) (failure,) = require(repo, "main") assert failure.startswith("check(s) recorded failing: always-fails") diff --git a/tests/test_yaml_bombs.py b/tests/test_yaml_bombs.py new file mode 100644 index 00000000..6b7b809f --- /dev/null +++ b/tests/test_yaml_bombs.py @@ -0,0 +1,75 @@ +"""A repo's YAML cannot hang or crash chock: alias expansion and compact nesting are bounded.""" + +from __future__ import annotations + +import subprocess +import sys +import textwrap +import time + +import pytest + +from chock.config import policy_status + +CAP = 512 * 1024 + + +def _bomb(levels: int = 40) -> str: + lines = ["policies:", " disabled:", " - &a0 [x, x]"] + lines += [f" - &a{i} [*a{i - 1}, *a{i - 1}]" for i in range(1, levels)] + return "\n".join(lines) + "\n" + + +def test_an_alias_bomb_in_disabled_does_not_expand() -> None: + from chock import yamlio + + config = yamlio.safe_load(_bomb()) + started = time.monotonic() + + status = policy_status(config, "some-policy") + + assert status["state"] != "disabled" + assert time.monotonic() - started < 2 + + +def test_only_string_ids_count_as_disabled() -> None: + config = {"policies": {"disabled": ["a-policy", 7, ["nested"], {"k": "v"}, None]}} + + assert policy_status(config, "a-policy")["state"] == "disabled" + assert policy_status(config, "7")["state"] != "disabled" + + +_CHILD = textwrap.dedent( + """ + import sys + from chock.guidance import advise + text = sys.stdin.read() + try: + advise._parsed(text, 512 * 1024) + except advise.GuidanceError as exc: + print("refused:", exc) + """ +) + + +@pytest.mark.parametrize( + "text", + [ + "- " * 100_000 + "x\n", + "? " * 100_000 + "x\n", + ], + ids=["compact-sequence", "compact-key"], +) +def test_compact_nesting_is_refused_without_crashing_the_process(text: str) -> None: + done = subprocess.run( + [sys.executable, "-I", "-c", _CHILD], input=text, capture_output=True, text=True, timeout=60, check=False + ) + + assert done.returncode == 0, done.stderr[-300:] + assert "refused:" in done.stdout + + +def test_ordinary_nesting_still_parses() -> None: + from chock.guidance import advise + + assert advise._parsed("policies:\n disabled:\n - a\n", CAP) == {"policies": {"disabled": ["a"]}}