diff --git a/.apm/architecture/owners/contracts-tooling.json b/.apm/architecture/owners/contracts-tooling.json index cb21eb615..59f77abe7 100644 --- a/.apm/architecture/owners/contracts-tooling.json +++ b/.apm/architecture/owners/contracts-tooling.json @@ -31,8 +31,8 @@ }, { "id": "frontmatter-bom-bounded-yaml", - "decision": "Frontmatter BOM decoding and bounded YAML parsing", - "owner": "utils/yaml_io.py (load_frontmatter, _BoundedYAMLHandler)", + "decision": "Frontmatter delimiter detection, BOM decoding, and bounded YAML parsing", + "owner": "utils/yaml_io.py (load_frontmatter, loads_frontmatter, _BoundedYAMLHandler)", "selectors": ["src/apm_cli/utils/yaml_io.py"], "guards": ["contracts-tooling-frontmatter-yaml"] }, diff --git a/CHANGELOG.md b/CHANGELOG.md index 3009383d1..289a71a88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- `apm install` no longer silently drops instruction Markdown whose + unfenced bodies contain `---` horizontal rules. It now stops the whole package + before deploying any primitive when instruction frontmatter is invalid YAML or + decodes critical hidden characters. `--force` overrides only the critical + character finding, never malformed YAML; warning-level findings do not block. + (by @manideep-malyala, #2666) - `apm pack` now reports unavailable remote package metadata, exposes certifiability in JSON, prevents `--check-clean` from certifying degraded regeneration, and lets `--strict-metadata` fail before writes. (closes #2524) diff --git a/docs/src/content/docs/producer/author-primitives/index.md b/docs/src/content/docs/producer/author-primitives/index.md index fcda795b1..9d255252c 100644 --- a/docs/src/content/docs/producer/author-primitives/index.md +++ b/docs/src/content/docs/producer/author-primitives/index.md @@ -43,10 +43,11 @@ See the [targets matrix](../../reference/targets-matrix/) for the full map. Commands ship as prompts (`.apm/prompts/*.prompt.md`); there is no separate `.apm/commands/` directory. See [Hooks and commands](./hooks-and-commands/). -Every primitive type follows the same pattern: a markdown file (or directory containing a primary markdown file) with frontmatter declaring its name and its trigger conditions. `apm compile` reads `.apm/`, applies any policy, and writes per-target output to the right directories on the target's filesystem. +Markdown-based primitive types use a markdown file (or directory containing a primary markdown file). Frontmatter requirements vary by primitive type and declare metadata such as the primitive's name and trigger conditions. `apm compile` reads `.apm/`, applies any policy, and writes per-target output to the right directories on the target's filesystem. -Encode primitive Markdown as UTF-8. APM accepts files with or without a -leading UTF-8 BOM and strips the BOM before parsing frontmatter. +:::note[Frontmatter fence] +Encode primitive Markdown as UTF-8. When a Markdown primitive has frontmatter, its opening fence of at least three hyphens (for example, `---`) must be the first content on line 1; an optional UTF-8 BOM may precede it and is stripped before parsing. Without that opening fence, APM treats the document as body content and later `---` lines remain Markdown horizontal rules. Malformed instruction frontmatter stops the package before deployment. Critical hidden characters decoded from metadata also prevent installation by default; `--force` overrides only that critical finding, while warning-level findings do not prevent installation. See [`apm install`](../../reference/cli/install/#behavior). +::: ## Recommended reading order diff --git a/docs/src/content/docs/reference/cli/install.md b/docs/src/content/docs/reference/cli/install.md index 01ef88783..3184e35a4 100644 --- a/docs/src/content/docs/reference/cli/install.md +++ b/docs/src/content/docs/reference/cli/install.md @@ -125,6 +125,12 @@ in `apm.yml`, then run `apm install` again. replacements to isolated staging paths and validate them before publication. If download, validation, or activation fails, APM keeps the previous package and lockfile active and exits non-zero with retry guidance. +- **Instruction frontmatter preflight.** Malformed YAML always rejects the + package before any of its primitives are deployed. Critical hidden characters + decoded from metadata also prevent installation by default; `--force` + overrides only that critical finding. Warning-level findings do not prevent + installation. See [Author primitives](../../../producer/author-primitives/) + for fence and UTF-8 BOM syntax. - **MCP-only lock state.** A normal project install creates or updates `apm.lock.yaml` when `apm.yml` declares only MCP dependencies, records the resolved MCP configs and targets, and migrates a legacy `apm.lock` first. Repeating the same install leaves the lockfile and target configs byte-identical. If initial lock creation fails, install exits nonzero and warns with writable-directory and rerun guidance. - **Lockfile replay and Git ref freshness.** Plain and `--frozen` installs may trust `apm.lock.yaml` and the local Git cache, reusing the locked commit for unchanged Git dependencies across the full resolved graph. In contrast, `apm install --update`, `apm install --refresh`, [`apm update`](../update/) with or without `--force`, [`apm lock --update`](../lock/), and [`apm outdated`](../outdated/) establish mutable Git refs from upstream instead of accepting stale refs from a local bare Git cache. APM picks up upstream changes to a transitive package's `apm.yml` only when you regenerate the graph -- run `apm update` or `apm lock --update`. See the [lockfile specification](../../lockfile-spec/) for the replay contract. - **Semver ranges on git deps.** `ref:` accepts semver ranges (`^1.2.0`, `~1.4`, `>=2.0 <3`, `1.5.x`) for git-source deps, including positional virtual-subdirectory references. APM runs `git ls-remote` against the dep, picks the highest tag matching the range, and pins the resolved tag plus commit SHA, version, and original constraint in `apm.lock.yaml`. Subsequent installs replay the lockfile without network; use `--update` (or change the manifest constraint) to re-resolve. See [manage dependencies](../../../consumer/manage-dependencies/#pin-a-semver-range) for the supported syntax. diff --git a/packages/apm-guide/.apm/skills/apm-usage/package-authoring.md b/packages/apm-guide/.apm/skills/apm-usage/package-authoring.md index 6aba33e1c..75289efe8 100644 --- a/packages/apm-guide/.apm/skills/apm-usage/package-authoring.md +++ b/packages/apm-guide/.apm/skills/apm-usage/package-authoring.md @@ -299,8 +299,16 @@ correctly -- the component just records NOASSERTION (genuinely unknown). This warning fires only on the **authoring** path (your own `apm.yml`); installing or exporting other people's dependencies is silent. -Encode primitive Markdown as UTF-8. APM accepts files with or without a -leading UTF-8 BOM and strips the BOM before parsing frontmatter. +Encode primitive Markdown as UTF-8. Frontmatter requirements vary by Markdown +primitive type. When frontmatter is present, its opening fence of at least +three hyphens (for example, `---`) must be the first content on line 1; an +optional UTF-8 BOM may precede it and is stripped before parsing. Without that +opening fence, APM treats the whole document as body content, and later `---` +lines stay Markdown horizontal rules. Malformed instruction frontmatter stops +the package before any primitive is deployed. Critical hidden characters +decoded from metadata also prevent installation by default; `--force` +overrides only that critical finding, while warning-level findings do not +prevent installation. ## The 7 primitive types diff --git a/scripts/architecture_linter/checks/contracts_test_taxonomy.py b/scripts/architecture_linter/checks/contracts_test_taxonomy.py index b62e19ad4..f13bec7e5 100644 --- a/scripts/architecture_linter/checks/contracts_test_taxonomy.py +++ b/scripts/architecture_linter/checks/contracts_test_taxonomy.py @@ -591,10 +591,142 @@ def check_apply_to_placement(provider: FactsProvider) -> tuple[Violation, ...]: _FRONTMATTER_OWNER = "src/apm_cli/utils/yaml_io.py" +_INSTRUCTION_INTEGRATOR = "src/apm_cli/integration/instruction_integrator.py" +_CONTENT_SCANNER = "src/apm_cli/security/content_scanner.py" +_INSTALL_SERVICES = "src/apm_cli/install/services.py" +_FRONTMATTER_METHODS = frozenset({"load", "loads", "parse"}) + + +def _frontmatter_aliases(nodes: Sequence[ast.AST]) -> tuple[set[str], dict[str, str]]: + """Return module aliases and imported parser-function aliases.""" + modules: set[str] = set() + functions: dict[str, str] = {} + for node in nodes: + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "frontmatter": + modules.add(alias.asname or alias.name) + elif isinstance(node, ast.ImportFrom) and node.module == "frontmatter": + for alias in node.names: + if alias.name in _FRONTMATTER_METHODS: + functions[alias.asname or alias.name] = alias.name + return modules, functions + + +def _frontmatter_call_name( + node: ast.Call, + modules: set[str], + functions: dict[str, str], +) -> str | None: + """Return the frontmatter parser entry point called by *node*.""" + if ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id in modules + and node.func.attr in _FRONTMATTER_METHODS + ): + return node.func.attr + if isinstance(node.func, ast.Name): + return functions.get(node.func.id) + return None + + +def _is_bounded_detect(node: ast.Call) -> bool: + return ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "_BOUNDED_FRONTMATTER_HANDLER" + and node.func.attr == "detect" + and len(node.args) == 1 + and isinstance(node.args[0], ast.Name) + and node.args[0].id == "text" + ) + + +def _is_bounded_loads(node: ast.Call, modules: set[str], functions: dict[str, str]) -> bool: + if _frontmatter_call_name(node, modules, functions) != "loads": + return False + handler = next((item.value for item in node.keywords if item.arg == "handler"), None) + return ( + len(node.args) == 1 + and isinstance(node.args[0], ast.Name) + and node.args[0].id == "text" + and isinstance(handler, ast.Name) + and handler.id == "_BOUNDED_FRONTMATTER_HANDLER" + ) + + +def _manual_frontmatter_detector(node: ast.Call) -> bool: + if ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "re" + and node.func.attr in {"compile", "match"} + and node.args + and isinstance(node.args[0], ast.Constant) + and isinstance(node.args[0].value, str) + and node.args[0].value.startswith("^---") + ): + return True + return ( + isinstance(node.func, ast.Attribute) + and node.func.attr in {"startswith", "split"} + and node.args + and isinstance(node.args[0], ast.Constant) + and isinstance(node.args[0].value, str) + and node.args[0].value.startswith("---") + ) + + +def _self_method_call(node: ast.Call, method: str) -> bool: + return ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "self" + and node.func.attr == method + ) + + +def _inside_matching_if(tree, node: ast.AST, predicate) -> bool: + """Return whether *node* is nested in an if whose test matches.""" + parent = tree.parent(node) + while parent is not None: + if isinstance(parent, ast.If) and predicate(parent.test): + return True + parent = tree.parent(parent) + return False + + +def _is_no_targets_test(node: ast.AST) -> bool: + return ( + isinstance(node, ast.UnaryOp) + and isinstance(node.op, ast.Not) + and isinstance(node.operand, ast.Name) + and node.operand.id == "targets" + ) + + +def _is_native_plugin_test(node: ast.AST) -> bool: + return ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "admits_native_plugin" + ) + + +def _prepared_identity_content(node: ast.AST) -> bool: + candidate = node.value if isinstance(node, ast.Attribute) and node.attr == "content" else node + return ( + isinstance(candidate, ast.Subscript) + and isinstance(candidate.value, ast.Name) + and candidate.value.id == "prepared_instructions" + and isinstance(candidate.slice, ast.Name) + and candidate.slice.id == "source_file" + ) def check_frontmatter_yaml(provider: FactsProvider) -> tuple[Violation, ...]: - """Frontmatter BOM decoding must route through utils/yaml_io.py.""" + """Frontmatter detection, BOM decoding, and parsing must use yaml_io.py.""" rule_id = _GUARD_FRONTMATTER owner, owner_fail = _facts_for(provider, _FRONTMATTER_OWNER, rule_id) if owner_fail: @@ -620,6 +752,273 @@ def check_frontmatter_yaml(provider: FactsProvider) -> tuple[Violation, ...]: "Frontmatter BOM decoding must route through utils/yaml_io.py", ) ) + scanner, scanner_fail = _facts_for(provider, _CONTENT_SCANNER, rule_id) + findings.extend(scanner_fail) + if not scanner_fail and ( + not _present(scanner, "content = _combine_surrogate_pairs(content)") + or not _present(scanner, "0xD800,") + or not _present(scanner, "0xDFFF,") + ): + findings.append( + _summary( + rule_id, + _CONTENT_SCANNER, + "decoded frontmatter scanning must normalize and reject UTF-16 surrogates", + ) + ) + services, services_fail = _facts_for(provider, _INSTALL_SERVICES, rule_id) + findings.extend(services_fail) + if not services_fail and services.tree_index is not None: + service_tree = services.tree_index + integration = service_tree.function("integrate_package_primitives") + service_scope = service_tree.own_scope(integration) if integration is not None else () + preflight_calls = [ + node + for node in service_scope + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "preflight_instructions_for_targets" + ] + reconcile_calls = [ + node + for node in service_scope + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "_reconcile_excluded_targets" + ] + no_target_calls = [ + node + for node in reconcile_calls + if _inside_matching_if(service_tree, node, _is_no_targets_test) + ] + native_calls = [ + node + for node in reconcile_calls + if _inside_matching_if(service_tree, node, _is_native_plugin_test) + ] + post_preflight_calls = [ + node + for node in reconcile_calls + if node not in no_target_calls and node not in native_calls + ] + if ( + len(preflight_calls) != 1 + or len(no_target_calls) != 1 + or len(native_calls) != 1 + or len(post_preflight_calls) != 1 + or post_preflight_calls[0].lineno <= preflight_calls[0].lineno + ): + findings.append( + _summary( + rule_id, + _INSTALL_SERVICES, + "instruction preflight must precede non-empty target reconciliation", + ) + ) + + tree = owner.tree_index + loads_function = tree.function("loads_frontmatter") if tree is not None else None + load_function = tree.function("load_frontmatter") if tree is not None else None + if tree is None or loads_function is None or load_function is None: + findings.append( + _summary( + rule_id, + _FRONTMATTER_OWNER, + "Frontmatter parsing must expose load_frontmatter and loads_frontmatter", + ) + ) + else: + modules, functions = _frontmatter_aliases(tree.nodes) + loads_scope = tree.own_scope(loads_function) + parser_calls = [ + node + for node in loads_scope + if isinstance(node, ast.Call) + and _frontmatter_call_name(node, modules, functions) is not None + ] + detect_calls = [ + node for node in loads_scope if isinstance(node, ast.Call) and _is_bounded_detect(node) + ] + bounded_calls = [ + node for node in parser_calls if _is_bounded_loads(node, modules, functions) + ] + load_delegates = [ + node + for node in tree.own_scope(load_function) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "loads_frontmatter" + ] + if len(detect_calls) != 1 or len(parser_calls) != 1 or len(bounded_calls) != 1: + findings.append( + _summary( + rule_id, + _FRONTMATTER_OWNER, + "loads_frontmatter must gate exactly one bounded frontmatter.loads call", + ) + ) + if len(load_delegates) != 1: + findings.append( + _summary( + rule_id, + _FRONTMATTER_OWNER, + "load_frontmatter must delegate parsed text to loads_frontmatter", + ) + ) + + for path in _python_paths(provider, _SRC_PREFIX): + facts, facts_fail = _facts_for(provider, path, rule_id) + findings.extend(facts_fail) + if facts_fail or facts.tree_index is None: + continue + modules, functions = _frontmatter_aliases(facts.tree_index.nodes) + for node in facts.tree_index.nodes: + if not isinstance(node, ast.Call): + continue + parser_name = _frontmatter_call_name(node, modules, functions) + if path != _FRONTMATTER_OWNER and parser_name in _FRONTMATTER_METHODS: + findings.append( + violation( + rule_id, + path, + "direct frontmatter parsing must route through utils/yaml_io.py", + line=node.lineno, + column=node.col_offset + 1, + ) + ) + if path == _INSTRUCTION_INTEGRATOR and _manual_frontmatter_detector(node): + findings.append( + violation( + rule_id, + path, + "instruction frontmatter detection must route through loads_frontmatter", + line=node.lineno, + column=node.col_offset + 1, + ) + ) + if path == _INSTRUCTION_INTEGRATOR: + integrate_function = facts.tree_index.function( + "InstructionIntegrator.integrate_instructions_for_target" + ) + integrate_scope = ( + facts.tree_index.own_scope(integrate_function) + if integrate_function is not None + else () + ) + prepare_calls = [ + node + for node in integrate_scope + if isinstance(node, ast.Call) and _self_method_call(node, "_prepare_instruction") + ] + identity_renders = [ + node + for node in integrate_scope + if isinstance(node, ast.Call) and _self_method_call(node, "_render_instruction") + ] + adoption_calls = [ + node + for node in integrate_scope + if isinstance(node, ast.Call) and _self_method_call(node, "_check_adopt_or_skip") + ] + expected_content = ( + next( + ( + item.value + for item in adoption_calls[0].keywords + if item.arg == "expected_content" + ), + None, + ) + if len(adoption_calls) == 1 + else None + ) + prepared_value = ( + next( + (item.value for item in identity_renders[0].keywords if item.arg == "prepared"), + None, + ) + if len(identity_renders) == 1 + else None + ) + if ( + len(prepare_calls) != 1 + or len(identity_renders) != 1 + or not _prepared_identity_content(prepared_value) + or not isinstance(expected_content, ast.Name) + or expected_content.id != "new_content" + ): + findings.append( + _summary( + rule_id, + path, + "identity instructions must materialize the prepared canonical parse", + ) + ) + prepare_function = facts.tree_index.function( + "InstructionIntegrator._prepare_instruction" + ) + prepare_scope = ( + facts.tree_index.own_scope(prepare_function) if prepare_function is not None else () + ) + security_calls = [ + node + for node in prepare_scope + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "SecurityGate" + and node.func.attr == "scan_text" + ] + json_calls = [ + node + for node in prepare_scope + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "json" + and node.func.attr == "dumps" + ] + force_value = ( + next( + (item.value for item in security_calls[0].keywords if item.arg == "force"), + None, + ) + if len(security_calls) == 1 + else None + ) + ensure_ascii = ( + next( + (item.value for item in json_calls[0].keywords if item.arg == "ensure_ascii"), + None, + ) + if len(json_calls) == 1 + else None + ) + block_checks = [ + node + for node in prepare_scope + if isinstance(node, ast.If) + and isinstance(node.test, ast.Attribute) + and isinstance(node.test.value, ast.Name) + and node.test.value.id == "verdict" + and node.test.attr == "should_block" + ] + if ( + len(security_calls) != 1 + or not isinstance(force_value, ast.Name) + or force_value.id != "force" + or len(json_calls) != 1 + or not isinstance(ensure_ascii, ast.Constant) + or ensure_ascii.value is not False + or len(block_checks) != 1 + ): + findings.append( + _summary( + rule_id, + path, + "decoded frontmatter metadata must cross SecurityGate with force policy", + ) + ) return tuple(findings) @@ -720,7 +1119,7 @@ def _structural_rule(rule_id: str, description: str, check) -> Rule: ), _owner_rule( _GUARD_FRONTMATTER, - "Frontmatter BOM decoding and bounded YAML parsing stay owned by utils/yaml_io.py.", + "Frontmatter delimiter detection, BOM decoding, and bounded YAML parsing stay owned by utils/yaml_io.py.", check_frontmatter_yaml, ), _owner_rule( diff --git a/src/apm_cli/install/deployed_paths.py b/src/apm_cli/install/deployed_paths.py index edb1e2b5e..d5372b105 100644 --- a/src/apm_cli/install/deployed_paths.py +++ b/src/apm_cli/install/deployed_paths.py @@ -9,6 +9,20 @@ from apm_cli.utils.paths import portable_relpath +def format_target_collapse(paths: list[str], verbose: bool) -> tuple[str, list[str]]: + """Format one target path, two paths, or a collapsed multi-target count.""" + deduped = list(dict.fromkeys(paths)) + if verbose and len(deduped) >= 2: + return "", [f" | -> {path}" for path in deduped] + if not deduped: + return "", [] + if len(deduped) == 1: + return deduped[0], [] + if len(deduped) == 2: + return f"{deduped[0]}, {deduped[1]}", [] + return f"{len(deduped)} targets", [] + + def deployed_path_entry( target_path: Path, project_root: Path, diff --git a/src/apm_cli/install/services.py b/src/apm_cli/install/services.py index 168c6d0dd..84114d6d6 100644 --- a/src/apm_cli/install/services.py +++ b/src/apm_cli/install/services.py @@ -26,6 +26,7 @@ from apm_cli.agent_plugins.errors import enforce_agent_plugin_deployment_boundary from .deployed_paths import deployed_path_entry as _deployed_path_entry +from .deployed_paths import format_target_collapse as _format_target_collapse from .deployed_paths import skill_bundle_file_entries as _skill_bundle_file_entries from .exec_gate import check_executable_approval from .exec_gate import plugin_bin_deployable as _plugin_bin_deployable @@ -35,6 +36,9 @@ from .local_bundle_paths import bundle_slug_validation_error as _bundle_slug_error from .local_bundle_paths import known_bundle_deploy_prefixes as _known_bundle_prefixes from .local_bundle_paths import target_bundle_deploy_prefixes as _target_bundle_prefixes +from .target_filter import ( + log_package_target_restriction as _log_package_target_restriction, +) from .target_filter import resolve_effective_package_targets if TYPE_CHECKING: @@ -199,21 +203,6 @@ def _warn_target_reconcile_failure( ) -def _log_package_target_restriction(logger: InstallLogger | None, target_selection: Any) -> None: - """Name the declared and effective target sets when a package narrows them.""" - if logger is None or not target_selection.package_restriction_active: - return - declared = ( - ", ".join(target_selection.package_declared_targets) - if target_selection.package_declared_targets - else "unrestricted" - ) - effective = ", ".join(target.name for target in target_selection.targets) or "none" - logger.verbose_detail( - f"Package target restriction: [{declared}]; effective targets: [{effective}]" - ) - - def integrate_package_primitives( # noqa: PLR0913 package_info: Any, project_root: Path, @@ -318,14 +307,18 @@ def integrate_package_primitives( # noqa: PLR0913 "reconcile_package_target_restriction", None, ) - if target_selection.excluded_targets and callable(reconcile_package_targets): - reconcile_stats = reconcile_package_targets( - package_info, - project_root, - target_selection.excluded_targets, - ) - _warn_target_reconcile_failure(diagnostics, package_name, reconcile_stats) + + def _reconcile_excluded_targets() -> None: + if target_selection.excluded_targets and callable(reconcile_package_targets): + reconcile_stats = reconcile_package_targets( + package_info, + project_root, + target_selection.excluded_targets, + ) + _warn_target_reconcile_failure(diagnostics, package_name, reconcile_stats) + if not targets: + _reconcile_excluded_targets() return result # Executable approval gate (npm v12-style default-deny); all five verdicts feed the gates. @@ -384,6 +377,7 @@ def integrate_package_primitives( # noqa: PLR0913 from apm_cli.install.native_plugin_admission import finalize_native_plugin if admits_native_plugin(package_info): + _reconcile_excluded_targets() return finalize_native_plugin( result, package_info, @@ -414,37 +408,6 @@ def _log_integration(msg): if logger: logger.tree_item(msg) - def _format_target_collapse(paths: list[str], verbose: bool) -> tuple[str, list[str]]: - """Apply the 1/2/3+ multi-target collapse rule. - - Returns a tuple ``(suffix, expansion_lines)``: - - * ``suffix`` -- the text appended after ``-> `` on the aggregate line. - * ``expansion_lines`` -- extra `` | -> `` lines emitted - AFTER the aggregate line when ``verbose`` is True. Empty list when - collapsed. - - The rule: - 1 target -> ```` - 2 targets -> ``, `` - 3+ -> ``N targets`` (verbose forces full enumeration) - """ - deduped: list[str] = [] - seen: set = builtins.set() - for p in paths: - if p not in seen: - seen.add(p) - deduped.append(p) - if verbose and len(deduped) >= 2: - return "", [f" | -> {p}" for p in deduped] - if len(deduped) == 0: - return "", [] - if len(deduped) == 1: - return deduped[0], [] - if len(deduped) == 2: - return f"{deduped[0]}, {deduped[1]}", [] - return f"{len(deduped)} targets", [] - _verbose = bool(getattr(ctx, "verbose", False)) if ctx is not None else False _INTEGRATOR_KWARGS = { @@ -457,6 +420,21 @@ def _format_target_collapse(paths: list[str], verbose: bool) -> tuple[str, list[ "skills": integrators.skill, } + # Validate every converted instruction target before any primitive kind can + # write. A rejected instruction must not leave prompts, agents, commands, + # or identity-target instructions from the same package active. + if integrators.instruction is not None: + integrators.instruction.preflight_instructions_for_targets( + targets, + package_info, + project_root, + source_plan, + force=force, + diagnostics=diagnostics, + ) + + _reconcile_excluded_targets() + # Aggregate per-primitive across targets so we emit ONE line per kind # (per the 1/2/3+ collapse rule), not one per target. # Structure: { prim_name: {"files": int, "adopted": int, "label": str, "paths": [str]} } diff --git a/src/apm_cli/install/target_filter.py b/src/apm_cli/install/target_filter.py index 5a8a7fe18..14b196d51 100644 --- a/src/apm_cli/install/target_filter.py +++ b/src/apm_cli/install/target_filter.py @@ -9,6 +9,7 @@ from apm_cli.models.apm_package import canonical_package_targets if TYPE_CHECKING: + from apm_cli.core.command_logger import InstallLogger from apm_cli.integration.targets import TargetProfile from ..utils.diagnostics import DiagnosticCollector @@ -27,6 +28,24 @@ class EffectivePackageTargets: package_restriction_active: bool +def log_package_target_restriction( + logger: InstallLogger | None, + target_selection: EffectivePackageTargets, +) -> None: + """Name declared and effective targets when a package narrows them.""" + if logger is None or not target_selection.package_restriction_active: + return + declared = ( + ", ".join(target_selection.package_declared_targets) + if target_selection.package_declared_targets + else "unrestricted" + ) + effective = ", ".join(target.name for target in target_selection.targets) or "none" + logger.verbose_detail( + f"Package target restriction: [{declared}]; effective targets: [{effective}]" + ) + + def filter_targets_for_dependency( targets: list[TargetProfile], dep_target_subset: list[str] | None, diff --git a/src/apm_cli/integration/base_integrator.py b/src/apm_cli/integration/base_integrator.py index fc6ad18e2..52396f289 100644 --- a/src/apm_cli/integration/base_integrator.py +++ b/src/apm_cli/integration/base_integrator.py @@ -387,6 +387,28 @@ def try_adopt_identical( return True return False + @staticmethod + def is_content_identical_to_text( + target_path: Path, + expected_content: str, + *, + lf_normalized_deploy: bool = False, + ) -> bool: + """Return whether a target matches already-prepared deployment text.""" + try: + if not target_path.exists() or target_path.is_symlink(): + return False + try: + target_bytes = _read_bytes_no_follow(target_path) + except _SymlinkRaceError: + return False + expected = ( + normalize_crlf_to_lf(expected_content) if lf_normalized_deploy else expected_content + ).encode("utf-8") + return target_bytes == expected + except OSError: + return False + def _check_adopt_or_skip( self, target_path: Path, @@ -396,6 +418,8 @@ def _check_adopt_or_skip( force: bool, diagnostics, target_paths: list, + *, + expected_content: str | None = None, ) -> tuple[bool, bool]: """Check whether *target_path* should be adopted or skipped. @@ -431,9 +455,20 @@ def _check_adopt_or_skip( is ``True`` only when the existing file already matched the deployed content and has been silently adopted. """ - if self.is_content_identical_to_source( - target_path, source_file, lf_normalized_deploy=self._LF_NORMALIZED_DEPLOY - ): + identical = ( + self.is_content_identical_to_text( + target_path, + expected_content, + lf_normalized_deploy=self._LF_NORMALIZED_DEPLOY, + ) + if expected_content is not None + else self.is_content_identical_to_source( + target_path, + source_file, + lf_normalized_deploy=self._LF_NORMALIZED_DEPLOY, + ) + ) + if identical: target_paths.append(target_path) return True, True if self.check_collision( diff --git a/src/apm_cli/integration/instruction_integrator.py b/src/apm_cli/integration/instruction_integrator.py index 0d88bd8ce..497f7ad38 100644 --- a/src/apm_cli/integration/instruction_integrator.py +++ b/src/apm_cli/integration/instruction_integrator.py @@ -9,10 +9,14 @@ from __future__ import annotations +import json import re +from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, ClassVar +import yaml + from apm_cli.integration.base_integrator import BaseIntegrator, IntegrationResult from apm_cli.integration.targets import RULE_FORMATS from apm_cli.utils.atomic_io import normalize_crlf_to_lf, write_text_lf @@ -20,11 +24,21 @@ from apm_cli.utils.path_security import ensure_path_within from apm_cli.utils.paths import portable_relpath from apm_cli.utils.patterns import normalize_apply_to, parse_apply_to, yaml_double_quote +from apm_cli.utils.yaml_io import loads_frontmatter if TYPE_CHECKING: + from apm_cli.install.deployable_source_plan import DeployableSourcePlan from apm_cli.integration.targets import TargetProfile +@dataclass(frozen=True) +class _PreparedInstruction: + """Validated instruction source reused across target plans.""" + + content: str + body: str + + class InstructionIntegrator(BaseIntegrator): """Handles integration of APM package instructions. @@ -39,6 +53,14 @@ class InstructionIntegrator(BaseIntegrator): # Deploys via write_text_lf -> compare adopt candidates in LF mode. _LF_NORMALIZED_DEPLOY = True + def __init__(self): + super().__init__() + self._prepared_rule_plans: dict[ + tuple[str, str, str, str, str], + dict[Path, tuple[Path, str, int]], + ] = {} + self._prepared_instructions: dict[str, dict[Path, _PreparedInstruction]] = {} + # Map format_id -> converter method. Built once at class load time; # avoids rebuilding the dict on every ``_render_instruction`` call. _FORMAT_CONVERTERS: ClassVar[dict[str, str]] = { @@ -50,15 +72,11 @@ class InstructionIntegrator(BaseIntegrator): } @staticmethod - def _normalize_frontmatter_apply_to(frontmatter: str) -> str: - """Return canonical applyTo text from a bounded YAML frontmatter block.""" - from apm_cli.utils.yaml_io import load_yaml_str - - try: - metadata = load_yaml_str(frontmatter) or {} - except Exception: - return "" - return normalize_apply_to(metadata.get("applyTo"), default="") + def _parse_frontmatter(content: str) -> tuple[dict, str]: + """Return bounded frontmatter metadata and body for instruction text.""" + post = loads_frontmatter(content, preserve_body=True) + metadata = post.metadata if isinstance(post.metadata, dict) else {} + return metadata, post.content def find_instruction_files(self, package_path: Path, source_plan=None) -> list[Path]: """Find all .instructions.md files in a package. @@ -74,17 +92,78 @@ def find_instruction_files(self, package_path: Path, source_plan=None) -> list[P source_plan, ) - def copy_instruction(self, source: Path, target: Path) -> int: + def copy_instruction(self, source: Path, target: Path, content: str | None = None) -> int: """Copy instruction file with link resolution. Preserves applyTo: frontmatter and all content as-is. """ - content = source.read_text(encoding="utf-8") + content = content if content is not None else source.read_text(encoding="utf-8") content, links_resolved = self.resolve_links(content, source, target) write_text_lf(target, content) return links_resolved - def _render_instruction(self, source: Path, target: Path, fmt: str) -> tuple[str, int]: + @staticmethod + def _package_key(package_path: Path) -> str: + """Return the stable cache key for one package source tree.""" + return str(package_path.resolve()) + + @staticmethod + def _raise_frontmatter_error(source: Path, exc: yaml.YAMLError) -> None: + """Raise the package-facing malformed-frontmatter error.""" + raise yaml.YAMLError( + f"Rejected frontmatter in {source.name}: {exc}\n" + "Fix or remove the invalid frontmatter, then rerun apm install." + ) from exc + + def _prepare_instruction( + self, + source: Path, + *, + force: bool = False, + diagnostics=None, + package_name: str = "", + ) -> _PreparedInstruction: + """Read and validate one instruction without writing target files.""" + from apm_cli.security.gate import BLOCK_POLICY, SecurityGate + + content = source.read_text(encoding="utf-8") + try: + metadata, body = self._parse_frontmatter(content) + except yaml.YAMLError as exc: + self._raise_frontmatter_error(source, exc) + decoded_metadata = json.dumps(metadata, ensure_ascii=False, sort_keys=True, default=str) + verdict = SecurityGate.scan_text( + decoded_metadata, + f"{source}#decoded-frontmatter", + policy=BLOCK_POLICY, + force=force, + ) + if verdict.has_findings and diagnostics is not None: + SecurityGate.report( + verdict, + diagnostics, + package=package_name, + force=force, + force_action="Allowed by preflight", + force_detail=( + "Decoded frontmatter contains critical hidden characters; " + f"edit the escaped value in {source.name}" + ), + ) + if verdict.should_block: + raise yaml.YAMLError( + f"Rejected decoded frontmatter in {source.name}: critical hidden Unicode characters" + ) + return _PreparedInstruction(content=content, body=body) + + def _render_instruction( + self, + source: Path, + target: Path, + fmt: str, + *, + prepared: _PreparedInstruction | None = None, + ) -> tuple[str, int]: """Render *source* to the content it would deploy for *fmt*, WITHOUT writing. Returns ``(content, links_resolved)``. @@ -96,13 +175,119 @@ def _render_instruction(self, source: Path, target: Path, fmt: str) -> tuple[str the single home for "which formats transform". Any ``fmt`` outside it is copied verbatim (identity transform). """ - content = source.read_text(encoding="utf-8") + content = prepared.content if prepared is not None else source.read_text(encoding="utf-8") if fmt in RULE_FORMATS: converter = getattr(self, self._FORMAT_CONVERTERS[fmt]) - content = converter(content) + try: + content = converter(content) + except yaml.YAMLError as exc: + self._raise_frontmatter_error(source, exc) content, links_resolved = self.resolve_links(content, source, target) return content, links_resolved + @staticmethod + def _plan_key( + package_path: Path, + project_root: Path, + deploy_dir: Path, + extension: str, + fmt: str, + ) -> tuple[str, str, str, str, str]: + """Return the per-package, per-target identity for a prepared rule plan.""" + return ( + str(package_path.resolve()), + str(project_root.resolve()), + str(deploy_dir.resolve()), + extension, + fmt, + ) + + def _prepare_rule_plan( + self, + instruction_files: list[Path], + deploy_dir: Path, + extension: str, + fmt: str, + prepared_instructions: dict[Path, _PreparedInstruction] | None = None, + ) -> dict[Path, tuple[Path, str, int]]: + """Render every converted rule without writing target files.""" + plan: dict[Path, tuple[Path, str, int]] = {} + for source_file in instruction_files: + stem = source_file.name + if stem.endswith(".instructions.md"): + stem = stem[: -len(".instructions.md")] + target_path = deploy_dir / f"{stem}{extension}" + ensure_path_within(target_path, deploy_dir) + content, links_resolved = self._render_instruction( + source_file, + target_path, + fmt, + prepared=(prepared_instructions or {}).get(source_file), + ) + plan[source_file] = (target_path, content, links_resolved) + return plan + + def preflight_instructions_for_targets( + self, + targets: list[TargetProfile], + package_info, + project_root: Path, + source_plan: DeployableSourcePlan, + *, + force: bool = False, + diagnostics=None, + ) -> None: + """Validate authorized instructions and prepare conversions before writes.""" + self._prepared_rule_plans.clear() + self._prepared_instructions.clear() + eligible_targets = [] + for target in targets: + mapping = target.primitives.get("instructions") + if not mapping: + continue + if not target.auto_create and not (project_root / target.root_dir).is_dir(): + continue + eligible_targets.append((target, mapping)) + if not eligible_targets: + return + + package_path = Path(package_info.install_path) + instruction_files = self.find_instruction_files(package_path, source_plan) + if not instruction_files: + return + + self.init_link_resolver(package_info, project_root) + package_name = getattr(getattr(package_info, "package", None), "name", "") + prepared_instructions = { + source_file: self._prepare_instruction( + source_file, + force=force, + diagnostics=diagnostics, + package_name=package_name, + ) + for source_file in instruction_files + } + self._prepared_instructions[self._package_key(package_path)] = prepared_instructions + for target, mapping in eligible_targets: + if not mapping.output_compare: + continue + effective_root = mapping.deploy_root or target.root_dir + deploy_dir = project_root / effective_root / mapping.subdir + key = self._plan_key( + package_path, + project_root, + deploy_dir, + mapping.extension, + mapping.format_id, + ) + self._prepared_rule_plans[key] = self._prepare_rule_plan( + instruction_files, + deploy_dir, + mapping.extension, + mapping.format_id, + prepared_instructions, + ) + # ------------------------------------------------------------------ # Target-driven API (data-driven dispatch) # ------------------------------------------------------------------ @@ -145,9 +330,26 @@ def integrate_instructions_for_target( return IntegrationResult(0, 0, 0, []) self.init_link_resolver(package_info, project_root) - instruction_files = self.find_instruction_files(package_info.install_path, source_plan) + package_path = Path(package_info.install_path) + prepared_instructions = self._prepared_instructions.get(self._package_key(package_path)) + instruction_files = ( + list(prepared_instructions) + if prepared_instructions is not None + else self.find_instruction_files(package_path, source_plan) + ) if not instruction_files: return IntegrationResult(0, 0, 0, []) + if prepared_instructions is None: + package_name = getattr(getattr(package_info, "package", None), "name", "") + prepared_instructions = { + source_file: self._prepare_instruction( + source_file, + force=force, + diagnostics=diagnostics, + package_name=package_name, + ) + for source_file in instruction_files + } deploy_dir = target_root / mapping.subdir deploy_dir.mkdir(parents=True, exist_ok=True) @@ -163,6 +365,7 @@ def integrate_instructions_for_target( managed_files=managed_files, diagnostics=diagnostics, pkg_source=getattr(getattr(package_info, "package", None), "source", None), + prepared_instructions=prepared_instructions, ) # APM-owned rule dirs (.claude/rules, .cursor/rules, .windsurf/rules): @@ -177,22 +380,42 @@ def integrate_instructions_for_target( files_adopted = 0 target_paths: list[Path] = [] total_links_resolved = 0 + plan_key = self._plan_key( + package_path, + project_root, + deploy_dir, + mapping.extension, + fmt, + ) + rendered_rules = self._prepared_rule_plans.pop(plan_key, None) + if apm_owned_rule_dir: + # Direct integrator callers may bypass install-level preflight. + if rendered_rules is None: + rendered_rules = self._prepare_rule_plan( + instruction_files, + deploy_dir, + mapping.extension, + fmt, + prepared_instructions, + ) + else: + rendered_rules = {} for source_file in instruction_files: if apm_owned_rule_dir: - stem = source_file.name - if stem.endswith(".instructions.md"): - stem = stem[: -len(".instructions.md")] - target_name = f"{stem}{mapping.extension}" + target_path, new_content, links_resolved = rendered_rules[source_file] else: - target_name = source_file.name - - target_path = deploy_dir / target_name - # target_name is Path.name (no separators), so traversal via - # deploy_dir is impossible. Validated against deploy_dir (not - # project_root) so user-scope targets whose root resolves - # outside the workspace still work correctly. - ensure_path_within(target_path, deploy_dir) + target_path = deploy_dir / source_file.name + # source_file.name has no separators, so traversal via + # deploy_dir is impossible. Validate against deploy_dir (not + # project_root) so user-scope targets outside the workspace work. + ensure_path_within(target_path, deploy_dir) + new_content, links_resolved = self._render_instruction( + source_file, + target_path, + fmt, + prepared=prepared_instructions[source_file], + ) rel_path = portable_relpath(target_path, project_root) @@ -210,9 +433,6 @@ def integrate_instructions_for_target( # transformed *output*: adopt when up-to-date (no churn), else # (re)write. Always record the path so it stays managed on the # next run. - new_content, links_resolved = self._render_instruction( - source_file, target_path, fmt - ) # Compare the on-disk bytes against the exact bytes # write_text_lf would emit (LF-normalized). A text-mode # read_text() comparison would collapse CRLF->LF and wrongly @@ -238,7 +458,14 @@ def integrate_instructions_for_target( continue skip, adopted = self._check_adopt_or_skip( - target_path, source_file, rel_path, managed_files, force, diagnostics, target_paths + target_path, + source_file, + rel_path, + managed_files, + force, + diagnostics, + target_paths, + expected_content=new_content, ) if skip: if adopted: @@ -247,7 +474,7 @@ def integrate_instructions_for_target( files_skipped += 1 continue - links_resolved = self.copy_instruction(source_file, target_path) + write_text_lf(target_path, new_content) total_links_resolved += links_resolved files_integrated += 1 target_paths.append(target_path) @@ -328,10 +555,8 @@ def _strip_frontmatter(content: str) -> str: If no frontmatter is present, returns the content unchanged. Handles both LF and CRLF line endings. """ - fm_match = re.match(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", content, re.DOTALL) - if fm_match: - return content[fm_match.end() :] - return content + _, body = InstructionIntegrator._parse_frontmatter(content) + return body @classmethod def _is_apm_managed_copilot(cls, content: str) -> bool: @@ -369,6 +594,7 @@ def _integrate_copilot_user_instructions( managed_files: set[str] | None = None, diagnostics=None, pkg_source: str | None = None, + prepared_instructions: dict[Path, _PreparedInstruction] | None = None, ) -> IntegrationResult: """Concatenate all instruction files into ~/.copilot/copilot-instructions.md. @@ -396,8 +622,12 @@ def _integrate_copilot_user_instructions( bodies: list[str] = [] for source_file in instruction_files: - raw = source_file.read_text(encoding="utf-8") - body = self._strip_frontmatter(raw).strip() + prepared = (prepared_instructions or {}).get(source_file) + body = ( + prepared.body + if prepared is not None + else self._strip_frontmatter(source_file.read_text(encoding="utf-8")) + ).strip() if body: bodies.append(body) @@ -494,21 +724,9 @@ def _convert_to_cursor_rules(content: str) -> str: extracts or generates a ``description``, and rewrites the frontmatter in Cursor's expected format. """ - body = content - apply_to = "" - description = "" - - # Parse existing frontmatter - fm_match = re.match(r"^---\s*\n(.*?)\n---\s*\n?", content, re.DOTALL) - if fm_match: - fm_block = fm_match.group(1) - body = content[fm_match.end() :] - apply_to = InstructionIntegrator._normalize_frontmatter_apply_to(fm_block) - - for line in fm_block.splitlines(): - line_stripped = line.strip() - if line_stripped.startswith("description:"): - description = line_stripped[len("description:") :].strip().strip("'\"") + metadata, body = InstructionIntegrator._parse_frontmatter(content) + apply_to = normalize_apply_to(metadata.get("applyTo"), default="") + description = str(metadata.get("description", "")).strip() # Generate description from first content sentence if missing if not description: @@ -521,7 +739,7 @@ def _convert_to_cursor_rules(content: str) -> str: # Build Cursor Rules frontmatter parts = ["---"] if description: - parts.append(f"description: {description}") + parts.append(f"description: {yaml_double_quote(description)}") globs = parse_apply_to(apply_to) if len(globs) == 1: parts.append(f"globs: {yaml_double_quote(globs[0])}") @@ -588,21 +806,14 @@ def sync_integration_cursor( # pylint: disable=duplicate-code # deprecated shi def _convert_to_windsurf_rules(content: str) -> str: """Convert APM instruction content to Windsurf rules ``.md`` format. - Parses existing YAML frontmatter via ``yaml.safe_load``, maps + Parses existing YAML frontmatter through the canonical bounded loader, maps ``applyTo`` to Windsurf's ``trigger: glob`` + ``globs`` frontmatter. Instructions without ``applyTo`` become ``trigger: always_on`` rules. Ref: https://docs.windsurf.com/windsurf/cascade/memories """ - body = content - apply_to = "" - - # Parse existing frontmatter with the bounded loader so a hostile - # frontmatter block in an untrusted package cannot hang the parser. - fm_match = re.match(r"^---\s*\n(.*?)\n---\s*\n?", content, re.DOTALL) - if fm_match: - body = content[fm_match.end() :] - apply_to = InstructionIntegrator._normalize_frontmatter_apply_to(fm_match.group(1)) + metadata, body = InstructionIntegrator._parse_frontmatter(content) + apply_to = normalize_apply_to(metadata.get("applyTo"), default="") # Build Windsurf rules frontmatter parts = ["---"] @@ -646,28 +857,10 @@ def _convert_to_kiro_steering(content: str) -> str: path-scoped guidance. APM's ``applyTo`` frontmatter is the source of truth for that scoping. """ - from ..utils.yaml_io import load_yaml_str - - body = content - globs = [] - - fm_match = re.match(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", content, re.DOTALL) - if fm_match: - body = content[fm_match.end() :] - try: - fm = load_yaml_str(fm_match.group(1)) or {} - except Exception: - fm = {} - raw_apply_to = fm.get("applyTo", "") - if isinstance(raw_apply_to, list): - globs = [ - s - for item in raw_apply_to - if (s := str(item).replace("\n", " ").replace("\r", " ").strip()) - ] - else: - safe_apply_to = str(raw_apply_to).replace("\n", " ").replace("\r", " ").strip() - globs = parse_apply_to(safe_apply_to) + metadata, body = InstructionIntegrator._parse_frontmatter(content) + apply_to = normalize_apply_to(metadata.get("applyTo"), default="") + safe_apply_to = apply_to.replace("\n", " ").replace("\r", " ").strip() + globs = parse_apply_to(safe_apply_to) parts = ["---"] if globs: @@ -698,15 +891,8 @@ def _convert_to_claude_rules(content: str) -> str: Ref: https://code.claude.com/docs/en/memory#organize-rules-with-claude%2Frules%2F """ - body = content - apply_to = "" - - # Parse existing frontmatter - fm_match = re.match(r"^---\s*\n(.*?)\n---\s*\n?", content, re.DOTALL) - if fm_match: - fm_block = fm_match.group(1) - body = content[fm_match.end() :] - apply_to = InstructionIntegrator._normalize_frontmatter_apply_to(fm_block) + metadata, body = InstructionIntegrator._parse_frontmatter(content) + apply_to = normalize_apply_to(metadata.get("applyTo"), default="") # Build Claude rules frontmatter (only when path-scoped) globs = parse_apply_to(apply_to) @@ -726,34 +912,10 @@ def _convert_to_antigravity_rules(content: str) -> str: Parses existing YAML frontmatter, maps ``applyTo`` to Antigravity's ``trigger: glob`` + ``globs`` frontmatter. """ - from ..utils.yaml_io import load_yaml_str - - body = content - globs = [] - - # Parse existing frontmatter - fm_match = re.match(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", content, re.DOTALL) - if fm_match: - body = content[fm_match.end() :] - try: - fm = load_yaml_str(fm_match.group(1)) or {} - except Exception as e: - import logging - - logging.getLogger(__name__).warning( - "Failed to parse instruction frontmatter YAML: %s", e - ) - fm = {} - raw_apply_to = fm.get("applyTo", "") - if isinstance(raw_apply_to, list): - globs = [ - s - for item in raw_apply_to - if (s := str(item).replace("\n", " ").replace("\r", " ").strip()) - ] - else: - safe_apply_to = str(raw_apply_to).replace("\n", " ").replace("\r", " ").strip() - globs = parse_apply_to(safe_apply_to) + metadata, body = InstructionIntegrator._parse_frontmatter(content) + apply_to = normalize_apply_to(metadata.get("applyTo"), default="") + safe_apply_to = apply_to.replace("\n", " ").replace("\r", " ").strip() + globs = parse_apply_to(safe_apply_to) # Build Antigravity rules frontmatter parts = ["---"] diff --git a/src/apm_cli/security/content_scanner.py b/src/apm_cli/security/content_scanner.py index a3ebcd82d..6e55fc17b 100644 --- a/src/apm_cli/security/content_scanner.py +++ b/src/apm_cli/security/content_scanner.py @@ -9,6 +9,7 @@ be tested and used independently. """ +import re import unicodedata from dataclasses import dataclass from pathlib import Path @@ -32,6 +33,13 @@ class ScanFinding: # range_end is inclusive. _SUSPICIOUS_RANGES: list[tuple[int, int, str, str, str]] = [ # ── Critical: no legitimate use in prompt/instruction files ── + ( + 0xD800, + 0xDFFF, + "critical", + "invalid-surrogate", + "Unpaired UTF-16 surrogate code unit", + ), # Unicode tag characters — invisible ASCII mapping ( 0xE0001, @@ -145,6 +153,28 @@ def _zwj_in_emoji_context(text: str, idx: int) -> bool: return prev_ok and next_ok +_SURROGATE_RE = re.compile(r"[\ud800-\udfff]") + + +def _combine_surrogate_pairs(text: str) -> str: + """Combine valid UTF-16 surrogate pairs while preserving unpaired units.""" + if _SURROGATE_RE.search(text) is None: + return text + combined: list[str] = [] + index = 0 + while index < len(text): + high = ord(text[index]) + if 0xD800 <= high <= 0xDBFF and index + 1 < len(text): + low = ord(text[index + 1]) + if 0xDC00 <= low <= 0xDFFF: + combined.append(chr(0x10000 + ((high - 0xD800) << 10) + low - 0xDC00)) + index += 2 + continue + combined.append(text[index]) + index += 1 + return "".join(combined) + + class ContentScanner: """Scans text content for hidden or suspicious Unicode characters.""" @@ -165,6 +195,7 @@ def scan_text(content: str, filename: str = "") -> list[ScanFinding]: if content.isascii(): return [] + content = _combine_surrogate_pairs(content) findings: list[ScanFinding] = [] lines = content.split("\n") diff --git a/src/apm_cli/security/gate.py b/src/apm_cli/security/gate.py index 1c5034a41..692fcd8e4 100644 --- a/src/apm_cli/security/gate.py +++ b/src/apm_cli/security/gate.py @@ -172,6 +172,7 @@ def scan_text( filename: str, *, policy: ScanPolicy = BLOCK_POLICY, + force: bool = False, ) -> ScanVerdict: """Scan in-memory text (compiled output, generated files).""" file_findings = ContentScanner.scan_text(content, filename=filename) @@ -182,7 +183,7 @@ def scan_text( findings_by_file, 1, policy, - force=False, + force=force, scanned_files=frozenset({filename}), ) @@ -191,6 +192,7 @@ def scan_texts( contents: dict[str, str], *, policy: ScanPolicy = BLOCK_POLICY, + force: bool = False, ) -> ScanVerdict: """Scan a complete in-memory output batch with one policy decision.""" findings_by_file: dict[str, list[ScanFinding]] = {} @@ -202,7 +204,7 @@ def scan_texts( findings_by_file, len(contents), policy, - force=False, + force=force, scanned_files=frozenset(contents), ) @@ -213,6 +215,8 @@ def report( *, package: str = "", force: bool = False, + force_action: str = "Deployed", + force_detail: str | None = None, ) -> None: """Record findings into a DiagnosticCollector with consistent messaging.""" if not verdict.has_findings: @@ -221,9 +225,10 @@ def report( if verdict.has_critical and not verdict.should_block and force: # --force: deployed despite critical diagnostics.security( - message=("Deployed with --force despite critical hidden characters"), + message=(f"{force_action} with --force despite critical hidden characters"), package=package, - detail=( + detail=force_detail + or ( f"{verdict.critical_count} critical finding(s) — " "run 'apm audit --strip' to clean up" ), diff --git a/src/apm_cli/utils/diagnostics.py b/src/apm_cli/utils/diagnostics.py index cd5de766b..6ede707b2 100644 --- a/src/apm_cli/utils/diagnostics.py +++ b/src/apm_cli/utils/diagnostics.py @@ -356,6 +356,8 @@ def _render_security_group(self, items: list[Diagnostic]) -> None: bold=True, ) _rich_info(" Run 'apm audit' for full details") + for detail in dict.fromkeys(d.detail for d in critical if d.detail): + _rich_info(f" {detail}") if self.verbose: by_pkg = _group_by_package(critical) for pkg, diags in by_pkg.items(): diff --git a/src/apm_cli/utils/patterns.py b/src/apm_cli/utils/patterns.py index c557de5b3..a9f4d35cd 100644 --- a/src/apm_cli/utils/patterns.py +++ b/src/apm_cli/utils/patterns.py @@ -7,6 +7,7 @@ from __future__ import annotations +import json from collections.abc import Iterable _APPLY_TO_ESCAPE = "\\" @@ -76,9 +77,9 @@ def yaml_double_quote(value: str) -> str: Defence-in-depth for the instruction integrators that emit YAML frontmatter via f-strings (``f' - "{g}"'``). A glob containing a literal backslash, double-quote, or control character would break - the surrounding YAML if inlined verbatim; this helper escapes the - minimal set needed for the YAML 1.2 double-quoted form. Returns the - value already wrapped in the surrounding double quotes. + the surrounding YAML if inlined verbatim. JSON string serialization + is a strict subset of YAML double-quoted scalar syntax and covers every + control character. Returns the value wrapped in double quotes. Note: ``parse_apply_to`` already strips leading/trailing whitespace per segment, and the Windsurf integrator strips newlines from the @@ -86,14 +87,7 @@ def yaml_double_quote(value: str) -> str: is near-zero -- this exists so emitted YAML stays well-formed even on adversarial or copy-paste-mangled inputs. """ - escaped = ( - value.replace("\\", "\\\\") - .replace('"', '\\"') - .replace("\n", "\\n") - .replace("\r", "\\r") - .replace("\t", "\\t") - ) - return f'"{escaped}"' + return json.dumps(value, ensure_ascii=True) def normalize_apply_to(value: object, default: str = "") -> str: diff --git a/src/apm_cli/utils/yaml_io.py b/src/apm_cli/utils/yaml_io.py index 281a3ecb0..43818e254 100644 --- a/src/apm_cli/utils/yaml_io.py +++ b/src/apm_cli/utils/yaml_io.py @@ -437,6 +437,10 @@ class _BoundedYAMLHandler(_FrontmatterYAMLHandler): ``apm install`` / ``apm audit``. """ + def detect(self, text: str) -> bool: + """Strip one leading UTF-8 BOM before detecting front matter.""" + return super().detect(text.removeprefix("\ufeff")) + def split(self, text: str) -> tuple[str, str]: """Strip one leading UTF-8 BOM before locating the front matter.""" return super().split(text.removeprefix("\ufeff")) @@ -456,6 +460,35 @@ def load(self, fm: str, **kwargs: Any) -> Any: _BOUNDED_FRONTMATTER_HANDLER = _BoundedYAMLHandler() +def loads_frontmatter(text: str, *, preserve_body: bool = False) -> Any: + """Parse Markdown text through the bounded handler. + + ``preserve_body`` retains body whitespace after consuming the delimiter's + first line break. Integrators use it when target conversion must not alter + the authored body. + """ + import frontmatter + + if not _BOUNDED_FRONTMATTER_HANDLER.detect(text): + return frontmatter.Post(text) + + try: + split = _BOUNDED_FRONTMATTER_HANDLER.split(text) + post = frontmatter.loads(text, handler=_BOUNDED_FRONTMATTER_HANDLER) + except yaml.YAMLError: + raise + except (IndexError, ValueError) as exc: + raise yaml.YAMLError(f"malformed frontmatter delimiters: {exc}") from exc + if preserve_body: + _, body = split + if body.startswith("\r\n"): + body = body[2:] + elif body.startswith("\n"): + body = body[1:] + post.content = body + return post + + def load_frontmatter(fd: Any, encoding: str = "utf-8-sig") -> Any: """Parse Markdown front matter with the bounded YAML loader. @@ -473,9 +506,17 @@ def load_frontmatter(fd: Any, encoding: str = "utf-8-sig") -> Any: (written by PowerShell's ``Out-File``, ``>``, or Notepad) therefore cannot hide the ``---`` fence and silently drop its ``applyTo`` scope (apm#2683). """ - import frontmatter - - return frontmatter.load(fd, encoding=encoding, handler=_BOUNDED_FRONTMATTER_HANDLER) + text = "" + if isinstance(fd, (str, Path)): + text = Path(fd).read_text(encoding=encoding) + elif hasattr(fd, "read"): + text = fd.read() + if isinstance(text, bytes): + text = text.decode(encoding) + else: + text = str(fd) + + return loads_frontmatter(text) def dump_yaml( diff --git a/tests/integration/test_architecture_frontmatter_bom.py b/tests/integration/test_architecture_frontmatter_bom.py index 9b2bdea6b..ce0c82702 100644 --- a/tests/integration/test_architecture_frontmatter_bom.py +++ b/tests/integration/test_architecture_frontmatter_bom.py @@ -1,4 +1,4 @@ -"""Architecture guards for canonical frontmatter BOM decoding.""" +"""Architecture guards for canonical frontmatter detection and BOM decoding.""" from __future__ import annotations @@ -36,11 +36,14 @@ def test_frontmatter_bom_decoding_has_single_owner() -> None: owner for owner in registry.owners if owner.id == "frontmatter-bom-bounded-yaml" ) rule = _RULES_BY_ID["contracts-tooling-frontmatter-yaml"] + report = run_selected_rules(root, ("contracts-tooling-frontmatter-yaml",)) + assert report.violations == () + assert report.failures == () assert 'def load_frontmatter(fd: Any, encoding: str = "utf-8-sig")' in owner assert 'text.removeprefix("\\ufeff")' in owner assert ( - "Frontmatter BOM decoding and bounded YAML parsing stay owned by utils/yaml_io.py" + "Frontmatter delimiter detection, BOM decoding, and bounded YAML parsing stay owned by utils/yaml_io.py" in rule.description ) assert registry_owner.selectors == ("src/apm_cli/utils/yaml_io.py",) @@ -85,3 +88,149 @@ def test_frontmatter_bom_guard_rejects_caller_owned_encoding(tmp_path: Path) -> assert report.exit_code != 0 assert _violated(report, "contracts-tooling-frontmatter-yaml") + + +@pytest.mark.parametrize( + "mutation", + [ + "local-detector", + "aliased-parser-bypass", + "identity-reread", + "identity-adoption-reread", + "decoded-security-bypass", + "decoded-force-bypass", + "surrogate-normalization-bypass", + "reconcile-before-preflight", + "native-reconcile-missing", + "no-target-reconcile-escapes", + ], +) +def test_frontmatter_authority_guard_rejects_split_owners( + tmp_path: Path, + mutation: str, +) -> None: + """The registered guard rejects local delimiter grammar and parser bypasses.""" + root = Path(__file__).parents[2] + sandbox = tmp_path / "repo" + shutil.copytree( + root, + sandbox, + ignore=shutil.ignore_patterns( + ".git", + ".venv", + ".pytest_cache", + "__pycache__", + "build", + "dist", + "node_modules", + ), + ) + if mutation == "local-detector": + owner = sandbox / "src/apm_cli/utils/yaml_io.py" + owner.write_text( + owner.read_text(encoding="utf-8").replace( + "_BOUNDED_FRONTMATTER_HANDLER.detect(text)", + 'text.startswith("---")', + 1, + ), + encoding="utf-8", + ) + elif mutation == "aliased-parser-bypass": + bypass = sandbox / "src/apm_cli/frontmatter_bypass.py" + bypass.write_text( + "from frontmatter import loads as parse\n\n" + "def read(text: str):\n" + " return parse(text)\n", + encoding="utf-8", + ) + elif mutation == "identity-reread": + integrator = sandbox / "src/apm_cli/integration/instruction_integrator.py" + integrator.write_text( + integrator.read_text(encoding="utf-8").replace( + " prepared=prepared_instructions[source_file],\n", + "", + 1, + ), + encoding="utf-8", + ) + elif mutation == "identity-adoption-reread": + integrator = sandbox / "src/apm_cli/integration/instruction_integrator.py" + integrator.write_text( + integrator.read_text(encoding="utf-8").replace( + " expected_content=new_content,\n", + "", + 1, + ), + encoding="utf-8", + ) + elif mutation == "decoded-security-bypass": + integrator = sandbox / "src/apm_cli/integration/instruction_integrator.py" + integrator.write_text( + integrator.read_text(encoding="utf-8").replace( + " if verdict.should_block:\n", + " if False:\n", + 1, + ), + encoding="utf-8", + ) + elif mutation == "decoded-force-bypass": + integrator = sandbox / "src/apm_cli/integration/instruction_integrator.py" + integrator.write_text( + integrator.read_text(encoding="utf-8").replace( + " force=force,\n", + " force=False,\n", + 1, + ), + encoding="utf-8", + ) + elif mutation == "surrogate-normalization-bypass": + scanner = sandbox / "src/apm_cli/security/content_scanner.py" + scanner.write_text( + scanner.read_text(encoding="utf-8").replace( + " content = _combine_surrogate_pairs(content)\n", + "", + 1, + ), + encoding="utf-8", + ) + elif mutation == "reconcile-before-preflight": + services = sandbox / "src/apm_cli/install/services.py" + source = services.read_text(encoding="utf-8") + post_call = "\n _reconcile_excluded_targets()\n\n # Aggregate per-primitive" + assert post_call in source + services.write_text( + source.replace(post_call, "\n\n # Aggregate per-primitive", 1).replace( + " if integrators.instruction is not None:\n", + " _reconcile_excluded_targets()\n if integrators.instruction is not None:\n", + 1, + ), + encoding="utf-8", + ) + elif mutation == "native-reconcile-missing": + services = sandbox / "src/apm_cli/install/services.py" + services.write_text( + services.read_text(encoding="utf-8").replace( + " if admits_native_plugin(package_info):\n" + " _reconcile_excluded_targets()\n", + " if admits_native_plugin(package_info):\n", + 1, + ), + encoding="utf-8", + ) + else: + services = sandbox / "src/apm_cli/install/services.py" + services.write_text( + services.read_text(encoding="utf-8").replace( + " if not targets:\n" + " _reconcile_excluded_targets()\n" + " return result\n", + " _reconcile_excluded_targets()\n if not targets:\n return result\n", + 1, + ), + encoding="utf-8", + ) + + report = run_selected_rules(sandbox, ("contracts-tooling-frontmatter-yaml",)) + + assert report.exit_code != 0 + assert _violated(report, "contracts-tooling-frontmatter-yaml") diff --git a/tests/integration/test_coverage_gaps_phase4w2.py b/tests/integration/test_coverage_gaps_phase4w2.py index a0c27d287..a54b7fb46 100644 --- a/tests/integration/test_coverage_gaps_phase4w2.py +++ b/tests/integration/test_coverage_gaps_phase4w2.py @@ -364,7 +364,7 @@ def test_scan_picks_up_mcp_servers(self, tmp_path: Path) -> None: ), ), patch("builtins.open", MagicMock()), - patch("frontmatter.load", return_value=mock_metadata), + patch("apm_cli.deps.aggregator.load_frontmatter", return_value=mock_metadata), ): result = scan_workflows_for_dependencies() assert "server-a" in result @@ -385,7 +385,7 @@ def test_scan_skips_non_list_mcp(self, tmp_path: Path) -> None: ), ), patch("builtins.open", MagicMock()), - patch("frontmatter.load", return_value=mock_metadata), + patch("apm_cli.deps.aggregator.load_frontmatter", return_value=mock_metadata), ): result = scan_workflows_for_dependencies() assert result == set() @@ -424,7 +424,7 @@ def _glob(pattern: str, recursive: bool) -> list[str]: with ( patch("glob.glob", side_effect=_glob), patch("builtins.open", MagicMock()), - patch("frontmatter.load", return_value=mock_metadata), + patch("apm_cli.deps.aggregator.load_frontmatter", return_value=mock_metadata), ): result = scan_workflows_for_dependencies() # Even though glob returned same file twice, should only count once diff --git a/tests/integration/test_local_install.py b/tests/integration/test_local_install.py index 36a28ab11..8ed1e7c4a 100644 --- a/tests/integration/test_local_install.py +++ b/tests/integration/test_local_install.py @@ -4,7 +4,7 @@ These tests create real file structures and invoke CLI commands via subprocess. """ -import os # noqa: F401 +import os import subprocess import sys # noqa: F401 import tempfile # noqa: F401 @@ -12,6 +12,9 @@ import pytest import yaml +from apm_cli.primitives.parser import parse_primitive_file +from apm_cli.utils.yaml_io import loads_frontmatter + pytestmark = pytest.mark.requires_apm_binary # --------------------------------------------------------------------------- @@ -165,6 +168,31 @@ def test_install_local_package_absolute_path(self, temp_workspace, apm_binary_pa def test_install_local_deploys_instructions(self, temp_workspace, apm_binary_path): """Verify that instructions from a local package are deployed to .github/instructions/.""" consumer = temp_workspace / "consumer" + source = ( + temp_workspace + / "packages" + / "local-skills" + / ".apm" + / "instructions" + / "test-skill.instructions.md" + ) + unfenced_markdown = """# Test Skill + +This instruction has no frontmatter. + +--- + +## Examples + +```python +print("ordinary Markdown") +``` + +--- + +## Next steps +""" + source.write_text(unfenced_markdown, encoding="utf-8") result = subprocess.run( [apm_binary_path, "install", "../packages/local-skills"], cwd=consumer, @@ -180,6 +208,364 @@ def test_install_local_deploys_instructions(self, temp_workspace, apm_binary_pat assert deployed.exists(), ( f"Instructions not deployed. Files in .github/: {all_files}\nstdout: {result.stdout}" ) + assert deployed.read_text(encoding="utf-8") == unfenced_markdown + assert parse_primitive_file(deployed).content == unfenced_markdown + + @pytest.mark.parametrize( + ("target", "deployed_path", "scope_marker"), + [ + ("claude", ".claude/rules/test-skill.md", ' - "src/**"'), + ("cursor", ".cursor/rules/test-skill.mdc", 'globs: "src/**"'), + ("windsurf", ".windsurf/rules/test-skill.md", 'globs: "src/**"'), + ("kiro", ".kiro/steering/test-skill.md", 'fileMatchPattern: "src/**"'), + ("antigravity", ".agents/rules/test-skill.md", 'globs: "src/**"'), + ], + ) + def test_install_four_hyphen_frontmatter_preserves_scope_across_targets( + self, + temp_workspace, + apm_binary_path, + target, + deployed_path, + scope_marker, + ): + """Converted targets consume the canonical three-or-more-hyphen grammar.""" + consumer = temp_workspace / "consumer" + source = ( + temp_workspace + / "packages" + / "local-skills" + / ".apm" + / "instructions" + / "test-skill.instructions.md" + ) + source.write_text( + "----\napplyTo: src/**\n----\n# Scoped rule\n", + encoding="utf-8", + ) + + result = subprocess.run( + [ + apm_binary_path, + "install", + "../packages/local-skills", + "--target", + target, + ], + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode == 0, result.stdout + result.stderr + deployed = consumer / deployed_path + assert deployed.exists() + rendered = deployed.read_text(encoding="utf-8") + assert scope_marker in rendered + assert "# Scoped rule" in rendered + assert "----" not in rendered + + def test_install_multi_target_consumes_all_preflight_plans( + self, + temp_workspace, + apm_binary_path, + ): + """One preflight prepares every selected converted target.""" + consumer = temp_workspace / "consumer" + source = ( + temp_workspace + / "packages" + / "local-skills" + / ".apm" + / "instructions" + / "test-skill.instructions.md" + ) + source.write_text( + "----\napplyTo: src/**\n----\n# Scoped rule\n", + encoding="utf-8", + ) + + result = subprocess.run( + [ + apm_binary_path, + "install", + "../packages/local-skills", + "--target", + "claude,cursor,windsurf,kiro,antigravity", + ], + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode == 0, result.stdout + result.stderr + expected_paths = [ + ".claude/rules/test-skill.md", + ".cursor/rules/test-skill.mdc", + ".windsurf/rules/test-skill.md", + ".kiro/steering/test-skill.md", + ".agents/rules/test-skill.md", + ] + for path in expected_paths: + rendered = (consumer / path).read_text(encoding="utf-8") + assert "# Scoped rule" in rendered + + def test_install_ignores_unplanned_root_instruction_during_preflight( + self, + temp_workspace, + apm_binary_path, + ): + """Only instructions authorized by the deployment plan affect preflight.""" + consumer = temp_workspace / "consumer" + package = temp_workspace / "packages" / "local-skills" + (package / "ignored.instructions.md").write_text( + "---\napplyTo: [\n---\n# Excluded rule\n", + encoding="utf-8", + ) + + result = subprocess.run( + [ + apm_binary_path, + "install", + "../packages/local-skills", + "--target", + "cursor", + ], + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode == 0, result.stdout + result.stderr + assert (consumer / ".cursor/rules/test-skill.mdc").exists() + assert not (consumer / ".cursor/rules/ignored.mdc").exists() + + def test_install_copilot_rejects_frontmatter_before_project_writes( + self, + temp_workspace, + apm_binary_path, + ): + """Copilot identity deployment validates instructions before prompts write.""" + consumer = temp_workspace / "consumer" + package = temp_workspace / "packages" / "local-skills" + source = package / ".apm/instructions/test-skill.instructions.md" + source.write_text("---\napplyTo: [\n---\n# Invalid rule\n", encoding="utf-8") + prompt = package / ".apm/prompts/good.prompt.md" + prompt.parent.mkdir(parents=True, exist_ok=True) + prompt.write_text("# Good prompt\n", encoding="utf-8") + + result = subprocess.run( + [ + apm_binary_path, + "install", + "../packages/local-skills", + "--target", + "copilot", + ], + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode != 0, result.stdout + result.stderr + assert not (consumer / ".github/prompts/good.prompt.md").exists() + assert not (consumer / ".github/instructions/test-skill.instructions.md").exists() + + def test_install_copilot_user_rejects_frontmatter_before_global_writes( + self, + temp_workspace, + apm_binary_path, + ): + """Copilot user deployment validates instructions before prompts write.""" + consumer = temp_workspace / "consumer" + package = temp_workspace / "packages" / "local-skills" + source = package / ".apm/instructions/test-skill.instructions.md" + source.write_text("---\napplyTo: [\n---\n# Invalid rule\n", encoding="utf-8") + prompt = package / ".apm/prompts/good.prompt.md" + prompt.parent.mkdir(parents=True, exist_ok=True) + prompt.write_text("# Good prompt\n", encoding="utf-8") + fake_home = temp_workspace / "home" + fake_home.mkdir() + env = os.environ.copy() + env["HOME"] = str(fake_home) + + result = subprocess.run( + [ + apm_binary_path, + "install", + str(package.resolve()), + "--target", + "copilot", + "--global", + ], + cwd=consumer, + env=env, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode != 0, result.stdout + result.stderr + assert not (fake_home / ".copilot/prompts/good.prompt.md").exists() + assert not (fake_home / ".copilot/copilot-instructions.md").exists() + + @pytest.mark.parametrize( + "escaped_hidden_unicode", + [r"\u202e", r"\uDB40\uDC01"], + ids=["bidi-override", "surrogate-pair-tag"], + ) + def test_install_rejects_yaml_escaped_hidden_unicode_before_writes( + self, + temp_workspace, + apm_binary_path, + escaped_hidden_unicode, + ): + """Decoded frontmatter metadata crosses the same security gate as source.""" + consumer = temp_workspace / "consumer" + package = temp_workspace / "packages" / "local-skills" + source = package / ".apm/instructions/test-skill.instructions.md" + source.write_text( + f'---\napplyTo: src/**\ndescription: "{escaped_hidden_unicode}hidden"\n---\n# Rule\n', + encoding="utf-8", + ) + prompt = package / ".apm/prompts/good.prompt.md" + prompt.parent.mkdir(parents=True, exist_ok=True) + prompt.write_text("# Good prompt\n", encoding="utf-8") + command = [ + apm_binary_path, + "install", + "../packages/local-skills", + "--target", + "cursor", + ] + + rejected = subprocess.run( + command, + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert rejected.returncode != 0, rejected.stdout + rejected.stderr + assert not (consumer / ".github/prompts/good.prompt.md").exists() + assert not (consumer / ".cursor/rules/test-skill.mdc").exists() + + forced = subprocess.run( + [*command, "--force"], + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert forced.returncode == 0, forced.stdout + forced.stderr + output = " ".join((forced.stdout + forced.stderr).split()) + assert "hidden characters detected" in output + assert "edit the escaped value in test-skill.instructions.md" in output + assert "Deployed with --force" not in output + assert (consumer / ".cursor/rules/test-skill.mdc").exists() + + def test_install_rejects_bounded_frontmatter_bomb( + self, + temp_workspace, + apm_binary_path, + ): + """Rejected YAML must not be copied into an agent-readable target.""" + consumer = temp_workspace / "consumer" + source = ( + temp_workspace + / "packages" + / "local-skills" + / ".apm" + / "instructions" + / "test-skill.instructions.md" + ) + good_source = source.with_name("a-good.instructions.md") + good_source.write_text("# Good rule\n", encoding="utf-8") + prompt_source = source.parents[1] / "prompts" / "good.prompt.md" + prompt_source.parent.mkdir(parents=True, exist_ok=True) + prompt_source.write_text("# Good prompt\n", encoding="utf-8") + lines = ["a0: &a0 {k: v}"] + previous = "a0" + for index in range(1, 40): + current = f"a{index}" + lines.append(f"{current}: &{current}") + lines.append(f" <<: [*{previous}, *{previous}]") + previous = current + source.write_text( + "---\n" + "\n".join(lines) + "\n---\n# Hostile rule\n", + encoding="utf-8", + ) + + result = subprocess.run( + [ + apm_binary_path, + "install", + "../packages/local-skills", + "--target", + "copilot,cursor", + ], + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode != 0, result.stdout + result.stderr + output = result.stdout + result.stderr + assert not (consumer / ".github/prompts/good.prompt.md").exists() + assert not (consumer / ".github/instructions/a-good.instructions.md").exists() + assert not (consumer / ".github/instructions/test-skill.instructions.md").exists() + assert not (consumer / ".cursor/rules/a-good.mdc").exists() + assert not (consumer / ".cursor/rules/test-skill.mdc").exists() + assert "Fix or remove the invalid frontmatter, then rerun apm install." in output + + def test_install_cursor_quotes_multiline_control_characters( + self, + temp_workspace, + apm_binary_path, + ): + """Cursor output remains valid YAML for decoded control characters.""" + consumer = temp_workspace / "consumer" + source = ( + temp_workspace + / "packages" + / "local-skills" + / ".apm" + / "instructions" + / "test-skill.instructions.md" + ) + source.write_text( + '---\napplyTo: src/**\ndescription: "safe\\n---\\n\\0suffix"\n---\n# Scoped rule\n', + encoding="utf-8", + ) + + result = subprocess.run( + [ + apm_binary_path, + "install", + "../packages/local-skills", + "--target", + "cursor", + ], + cwd=consumer, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode == 0, result.stdout + result.stderr + deployed = consumer / ".cursor/rules/test-skill.mdc" + rendered = deployed.read_text(encoding="utf-8") + assert "\0" not in rendered + post = loads_frontmatter(rendered) + assert post.metadata["description"] == "safe\n---\n\0suffix" + assert post.metadata["globs"] == "src/**" def test_install_local_package_no_manifest_fails(self, temp_workspace, apm_binary_path): """Installing a path with no apm.yml or SKILL.md should fail gracefully.""" diff --git a/tests/integration/test_package_target_hook_routing_e2e.py b/tests/integration/test_package_target_hook_routing_e2e.py index 11fbde2a0..586a41739 100644 --- a/tests/integration/test_package_target_hook_routing_e2e.py +++ b/tests/integration/test_package_target_hook_routing_e2e.py @@ -134,12 +134,23 @@ def _publish( def _restrict_published_package_to_claude( scenario: _Scenario, published: _PublishedPackage, + *, + malformed_instruction: bool = False, ) -> _PublishedPackage: """Advance a universal package manifest to a Claude-only release.""" manifest_path = published.repository.worktree / "apm.yml" manifest = load_yaml(manifest_path) manifest["targets"] = ["claude"] dump_yaml(manifest, manifest_path) + if malformed_instruction: + instruction = ( + published.repository.worktree / ".apm" / "instructions" / "broken.instructions.md" + ) + instruction.parent.mkdir(parents=True, exist_ok=True) + instruction.write_text( + "---\napplyTo: [\n---\n# Invalid instruction\n", + encoding="utf-8", + ) commit = scenario.repositories.commit( published.repository, message=f"restrict {published.name} to claude", @@ -455,3 +466,43 @@ def test_package_target_transition_repairs_cursor_and_uninstall_preserves_user_h environment=claude_only.environment, scenario_id="target-transition-uninstalled-audit", ) + + +def test_failed_restricted_update_preserves_existing_hook_state( + tmp_path: Path, + apm_binary_path: Path, +) -> None: + """Malformed instruction preflight precedes excluded-target hook cleanup.""" + scenario = _new_scenario(tmp_path / "failed-target-transition", apm_binary_path) + universal = _publish(scenario, "failed-transition-hooks", targets=()) + consumer = _consumer(scenario, "failed-transition-consumer", universal.dependency) + _run_success( + scenario, + consumer, + _INSTALL_ARGS, + environment=universal.environment, + scenario_id="failed-transition-universal-install", + ) + claude_only = _restrict_published_package_to_claude( + scenario, + universal, + malformed_instruction=True, + ) + scenario.consumers.replace_apm_dependencies(consumer, (claude_only.dependency,)) + before = _snapshot(consumer) + assert set(_cursor_commands(before)) == set(_EVENT_COMMANDS.values()) + assert before.file(".cursor/apm-hooks.json").kind == "file" + + result = scenario.runner.run( + (*_INSTALL_ARGS, "--update"), + scenario_id="failed-transition-restrict-update", + cwd=consumer.root, + env=claude_only.environment, + ) + + assert result.returncode != 0, ( + f"command={result.command!r}\nstdout={result.stdout!r}\nstderr={result.stderr!r}" + ) + output = " ".join((result.stdout + result.stderr).split()) + assert "Rejected frontmatter in broken.instructions.md" in output + assert _snapshot(consumer) == before diff --git a/tests/integration/test_validation_rules.py b/tests/integration/test_validation_rules.py index 937c93ae9..23edd7109 100644 --- a/tests/integration/test_validation_rules.py +++ b/tests/integration/test_validation_rules.py @@ -372,7 +372,10 @@ def test_validate_claude_skill_uses_directory_name_when_name_missing( def test_validate_claude_skill_surfaces_exception(self, tmp_path: Path) -> None: skill_md = _write_skill_md(tmp_path) - with patch("frontmatter.load", side_effect=ValueError("broken frontmatter")): + with patch( + "apm_cli.utils.yaml_io.load_frontmatter", + side_effect=ValueError("broken frontmatter"), + ): result = _validate_claude_skill(tmp_path, skill_md, ValidationResult()) assert result.is_valid is False assert "broken frontmatter" in result.errors[0] @@ -410,7 +413,10 @@ def test_validate_skill_bundle_surfaces_ensure_path_within_error(self, tmp_path: def test_validate_skill_bundle_surfaces_frontmatter_parse_error(self, tmp_path: Path) -> None: _write(tmp_path / "skills" / "alpha" / "SKILL.md", "---\nname: alpha\n") - with patch("frontmatter.load", side_effect=ValueError("bad frontmatter")): + with patch( + "apm_cli.utils.yaml_io.load_frontmatter", + side_effect=ValueError("bad frontmatter"), + ): result = _validate_skill_bundle(tmp_path, ValidationResult()) assert result.is_valid is False assert "failed to parse frontmatter" in result.errors[0] @@ -493,7 +499,10 @@ def test_validate_hybrid_requires_skill_md(self, tmp_path: Path) -> None: def test_validate_hybrid_warns_when_frontmatter_cannot_be_parsed(self, tmp_path: Path) -> None: _write_apm_yml(tmp_path) skill_md = _write_skill_md(tmp_path) - with patch("frontmatter.load", side_effect=ValueError("bad frontmatter")): + with patch( + "apm_cli.utils.yaml_io.load_frontmatter", + side_effect=ValueError("bad frontmatter"), + ): result = _validate_hybrid_package(tmp_path, tmp_path / "apm.yml", ValidationResult()) assert result.package is not None assert any(str(skill_md.name) in warning for warning in result.warnings) diff --git a/tests/unit/integration/test_dep_target_intersection.py b/tests/unit/integration/test_dep_target_intersection.py index b16860a74..d394f3fed 100644 --- a/tests/unit/integration/test_dep_target_intersection.py +++ b/tests/unit/integration/test_dep_target_intersection.py @@ -91,6 +91,11 @@ def reconcile_package_target_restriction( return self.reconcile_result +class _RejectingInstructionIntegrator: + def preflight_instructions_for_targets(self, *_args, **_kwargs) -> None: + raise ValueError("malformed instruction frontmatter") + + def _bundle(hook_integrator) -> IntegratorBundle: return IntegratorBundle( prompt=None, @@ -103,6 +108,61 @@ def _bundle(hook_integrator) -> IntegratorBundle: ) +def test_instruction_preflight_precedes_excluded_hook_reconciliation(tmp_path: Path) -> None: + """A rejected instruction cannot leave excluded-target hook mutations.""" + hook_integrator = _RecordingHookIntegrator() + bundle = _bundle(hook_integrator) + bundle = replace(bundle, instruction=_RejectingInstructionIntegrator()) + package_path = tmp_path / "pkg" + package_path.mkdir() + + with pytest.raises(ValueError, match="malformed instruction frontmatter"): + integrate_package_primitives( + _package_info(package_path, target="copilot"), + tmp_path / "project", + targets=[KNOWN_TARGETS["copilot"], KNOWN_TARGETS["claude"]], + integrators=bundle, + force=False, + managed_files=set(), + diagnostics=DiagnosticCollector(), + package_name="targeted-hooks", + ) + + assert hook_integrator.reconciled_targets == [] + + +def test_native_plugin_reconciles_excluded_targets( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Native plugin admission still removes excluded-target hook ownership.""" + from apm_cli.copilot_plugins import capability + from apm_cli.install import native_plugin_admission + + hook_integrator = _RecordingHookIntegrator() + monkeypatch.setattr(capability, "admits_native_plugin", lambda _package: True) + monkeypatch.setattr( + native_plugin_admission, + "finalize_native_plugin", + lambda result, *_args, **_kwargs: result, + ) + package_path = tmp_path / "pkg" + package_path.mkdir() + + integrate_package_primitives( + _package_info(package_path, target="copilot"), + tmp_path / "project", + targets=[KNOWN_TARGETS["copilot"], KNOWN_TARGETS["claude"]], + integrators=_bundle(hook_integrator), + force=False, + managed_files=set(), + diagnostics=DiagnosticCollector(), + package_name="targeted-hooks", + ) + + assert hook_integrator.reconciled_targets == ["claude"] + + def _run_with_targets( tmp_path: Path, install_target_names: list[str], diff --git a/tests/unit/integration/test_instruction_integrator.py b/tests/unit/integration/test_instruction_integrator.py index c3b517ebb..143a574db 100644 --- a/tests/unit/integration/test_instruction_integrator.py +++ b/tests/unit/integration/test_instruction_integrator.py @@ -7,10 +7,13 @@ from unittest.mock import Mock import pytest +import yaml +from apm_cli.install.deployable_source_plan import DeployableSourcePlan from apm_cli.integration.base_integrator import IntegrationResult from apm_cli.integration.instruction_integrator import InstructionIntegrator from apm_cli.models.apm_package import APMPackage, GitReferenceType, PackageInfo, ResolvedReference +from apm_cli.utils.yaml_io import loads_frontmatter def _make_package_info(package_dir, name="test-pkg"): @@ -125,6 +128,107 @@ def test_copy_instruction_preserves_frontmatter(self): self.integrator.copy_instruction(source, target) assert target.read_text() == content + def test_direct_identity_target_rejects_invalid_frontmatter_before_writes(self): + """Direct target integration keeps the same validation boundary as install.""" + from apm_cli.integration.targets import KNOWN_TARGETS + + package_dir = self.project_root / "package" + instruction_dir = package_dir / ".apm" / "instructions" + instruction_dir.mkdir(parents=True) + (instruction_dir / "broken.instructions.md").write_text( + "---\napplyTo: [\n---\n# Invalid\n", + encoding="utf-8", + ) + + with pytest.raises(yaml.YAMLError, match="Rejected frontmatter"): + self.integrator.integrate_instructions_for_target( + KNOWN_TARGETS["copilot"], + self._make_package_info(package_dir), + self.project_root, + ) + + assert not (self.project_root / ".github/instructions").exists() + + def test_preflight_force_report_does_not_claim_deployment(self): + """Preflight labels a force override without claiming files were written.""" + from unittest.mock import patch + + from apm_cli.security.gate import SecurityGate + + source = self.project_root / "forced.instructions.md" + source.write_text("# Rule\n", encoding="utf-8") + diagnostics = Mock() + verdict = Mock(has_findings=True, should_block=False) + + with ( + patch.object(SecurityGate, "scan_text", return_value=verdict), + patch.object(SecurityGate, "report") as report, + ): + self.integrator._prepare_instruction( + source, + force=True, + diagnostics=diagnostics, + package_name="pkg", + ) + + report.assert_called_once_with( + verdict, + diagnostics, + package="pkg", + force=True, + force_action="Allowed by preflight", + force_detail=( + "Decoded frontmatter contains critical hidden characters; " + "edit the escaped value in forced.instructions.md" + ), + ) + + def test_identity_target_materializes_preflight_validated_content(self): + """Identity deployment writes the bytes validated during preflight.""" + from unittest.mock import patch + + from apm_cli.integration.targets import KNOWN_TARGETS + + package_dir = self.project_root / "package" + instruction_dir = package_dir / ".apm" / "instructions" + instruction_dir.mkdir(parents=True) + source = instruction_dir / "python.instructions.md" + source.write_text("# Validated content\n", encoding="utf-8") + package_info = self._make_package_info(package_dir) + source_plan = DeployableSourcePlan( + package_dir.resolve(), + frozenset({".apm/instructions/python.instructions.md"}), + ) + + self.integrator.preflight_instructions_for_targets( + [KNOWN_TARGETS["copilot"]], + package_info, + self.project_root, + source_plan, + ) + changed = "---\napplyTo: [\n---\n# Changed after preflight\n" + source.write_text(changed, encoding="utf-8") + deployed = self.project_root / ".github/instructions/python.instructions.md" + deployed.parent.mkdir(parents=True) + deployed.write_text("# Resolved link\n", encoding="utf-8") + + def resolve_prepared(content, _source, _target): + assert content == "# Validated content\n" + return "# Resolved link\n", 1 + + with patch.object(self.integrator, "resolve_links", side_effect=resolve_prepared): + result = self.integrator.integrate_instructions_for_target( + KNOWN_TARGETS["copilot"], + package_info, + self.project_root, + source_plan=source_plan, + ) + + assert result.files_integrated == 0 + assert result.files_adopted == 1 + assert result.target_paths == [deployed] + assert deployed.read_text(encoding="utf-8") == "# Resolved link\n" + # ===== Integration ===== def test_integrate_creates_target_directory(self): @@ -522,24 +626,24 @@ def test_maps_apply_to_to_globs(self): def test_preserves_description(self): content = "---\napplyTo: '**/*.ts'\ndescription: TypeScript guidelines\n---\n\n# TS Rules" result = InstructionIntegrator._convert_to_cursor_rules(content) - assert "description: TypeScript guidelines" in result + assert 'description: "TypeScript guidelines"' in result assert 'globs: "**/*.ts"' in result def test_generates_description_from_heading(self): content = "---\napplyTo: '**/*.py'\n---\n\n# Python coding standards\n\nUse type hints." result = InstructionIntegrator._convert_to_cursor_rules(content) - assert "description: Python coding standards" in result + assert 'description: "Python coding standards"' in result def test_generates_description_from_first_sentence(self): content = "---\napplyTo: '**'\n---\n\nAlways use descriptive names. Follow PEP8." result = InstructionIntegrator._convert_to_cursor_rules(content) - assert "description: Always use descriptive names" in result + assert 'description: "Always use descriptive names"' in result def test_no_frontmatter(self): content = "# Simple rules\n\nJust some guidelines." result = InstructionIntegrator._convert_to_cursor_rules(content) assert result.startswith("---\n") - assert "description: Simple rules" in result + assert 'description: "Simple rules"' in result # No globs when no applyTo assert "globs" not in result @@ -553,7 +657,7 @@ def test_empty_apply_to_omits_globs(self): content = "---\ndescription: General rules\n---\n\n# Rules" result = InstructionIntegrator._convert_to_cursor_rules(content) assert "globs" not in result - assert "description: General rules" in result + assert 'description: "General rules"' in result class TestCursorRulesIntegration: @@ -768,7 +872,7 @@ def test_frontmatter_conversion_in_deployed_file(self): deployed = (self.project_root / ".cursor" / "rules" / "ts.mdc").read_text() assert 'globs: "src/**/*.ts"' in deployed - assert "description: TypeScript rules" in deployed + assert 'description: "TypeScript rules"' in deployed assert "applyTo" not in deployed assert "# TypeScript" in deployed assert "Use strict mode." in deployed @@ -1307,7 +1411,41 @@ def test_double_quoted_apply_to(self): class TestApplyToCommaSplitting: - """Verify all three converters split comma-separated applyTo globs.""" + """Verify target converters preserve canonical applyTo semantics.""" + + @pytest.mark.parametrize( + ("converter", "scope_marker"), + [ + ("_convert_to_claude_rules", ' - "src/**"'), + ("_convert_to_cursor_rules", 'globs: "src/**"'), + ("_convert_to_windsurf_rules", 'globs: "src/**"'), + ("_convert_to_kiro_steering", 'fileMatchPattern: "src/**"'), + ("_convert_to_antigravity_rules", 'globs: "src/**"'), + ], + ) + def test_four_hyphen_frontmatter_uses_canonical_parser(self, converter, scope_marker): + content = "----\napplyTo: src/**\n----\n# Scoped rule\n" + + result = getattr(InstructionIntegrator, converter)(content) + + assert scope_marker in result + assert "----" not in result + assert "# Scoped rule" in result + + def test_strip_frontmatter_supports_four_hyphen_fence(self): + content = "----\napplyTo: src/**\n----\n# Scoped rule\n" + + assert InstructionIntegrator._strip_frontmatter(content) == "# Scoped rule\n" + + def test_cursor_quotes_multiline_description(self): + content = '---\napplyTo: src/**\ndescription: "safe\\n---\\n"\n---\n# Scoped rule\n' + + result = InstructionIntegrator._convert_to_cursor_rules(content) + post = loads_frontmatter(result) + + assert post.metadata["globs"] == "src/**" + assert post.metadata["description"] == "safe\n---" + assert post.content == "# Scoped rule" # ---- Claude ---- @@ -1516,15 +1654,10 @@ def test_antigravity_list_valued_apply_to_with_literal_commas(self): assert ' - "src/foo,bar/*.py"' in result assert ' - "tests/**/*.py"' in result - def test_antigravity_malformed_yaml_fallback(self, caplog): - import logging - + def test_antigravity_malformed_yaml_fails_closed(self): content = "---\napplyTo: [\n---\n\n# Body" - with caplog.at_level(logging.WARNING): - result = InstructionIntegrator._convert_to_antigravity_rules(content) - assert "Failed to parse instruction frontmatter YAML" in caplog.text - assert "trigger: glob" not in result - assert "# Body" in result + with pytest.raises(yaml.YAMLError): + InstructionIntegrator._convert_to_antigravity_rules(content) class TestWindsurfRulesIntegration: diff --git a/tests/unit/test_content_scanner.py b/tests/unit/test_content_scanner.py index 99f2dc3f5..7d531b789 100644 --- a/tests/unit/test_content_scanner.py +++ b/tests/unit/test_content_scanner.py @@ -5,7 +5,11 @@ import pytest # noqa: F401 -from apm_cli.security.content_scanner import ContentScanner, ScanFinding +from apm_cli.security.content_scanner import ( + ContentScanner, + ScanFinding, + _combine_surrogate_pairs, +) class TestScanText: @@ -21,6 +25,11 @@ def test_empty_string_returns_empty(self): findings = ContentScanner.scan_text("") assert findings == [] + def test_surrogate_normalizer_returns_common_non_ascii_input_unchanged(self): + content = "caf\u00e9" * 10_000 + + assert _combine_surrogate_pairs(content) is content + def test_whitespace_only_returns_empty(self): findings = ContentScanner.scan_text(" \n\n\t\t\n") assert findings == [] @@ -37,6 +46,21 @@ def test_tag_character_detected_as_critical(self): assert findings[0].codepoint == "U+E0001" assert findings[0].file == "test.md" + def test_utf16_surrogate_pair_is_scanned_as_unicode_scalar(self): + findings = ContentScanner.scan_text("\udb40\udc01", filename="decoded.yml") + + assert len(findings) == 1 + assert findings[0].severity == "critical" + assert findings[0].category == "tag-character" + assert findings[0].codepoint == "U+E0001" + + def test_unpaired_utf16_surrogate_is_critical(self): + findings = ContentScanner.scan_text("\udb40", filename="decoded.yml") + + assert len(findings) == 1 + assert findings[0].severity == "critical" + assert findings[0].category == "invalid-surrogate" + def test_multiple_tag_characters(self): """Full range of tag chars embedded in text.""" # Embed a few tag characters that map to invisible ASCII diff --git a/tests/unit/test_deps.py b/tests/unit/test_deps.py index 148ddef0f..adba72564 100644 --- a/tests/unit/test_deps.py +++ b/tests/unit/test_deps.py @@ -6,7 +6,6 @@ import unittest from unittest.mock import mock_open, patch -import frontmatter # noqa: F401 import yaml from apm_cli.deps.aggregator import ( @@ -25,7 +24,7 @@ class TestDependenciesAggregator(unittest.TestCase): @patch("glob.glob") @patch("builtins.open", new_callable=mock_open) - @patch("frontmatter.load") + @patch("apm_cli.deps.aggregator.load_frontmatter") def test_scan_workflows_for_dependencies(self, mock_frontmatter_load, mock_file, mock_glob): """Test scanning workflows for dependencies.""" # Mock glob to return workflow files diff --git a/tests/unit/test_security_gate.py b/tests/unit/test_security_gate.py index 2a752d961..61c431217 100644 --- a/tests/unit/test_security_gate.py +++ b/tests/unit/test_security_gate.py @@ -199,6 +199,29 @@ def test_force_critical_reports_deployed(self): detail = call_args.kwargs.get("detail", call_args[1].get("detail", "")) assert "apm audit --strip" in detail + def test_force_report_accepts_preflight_action(self): + diag = MagicMock() + verdict = ScanVerdict( + findings_by_file={"decoded.yml": [MagicMock(severity="critical")]}, + has_critical=True, + should_block=False, + critical_count=1, + warning_count=0, + ) + + SecurityGate.report( + verdict, + diag, + package="pkg", + force=True, + force_action="Allowed by preflight", + force_detail="Edit the escaped value in decoded.yml", + ) + + message = diag.security.call_args.kwargs["message"] + assert message == "Allowed by preflight with --force despite critical hidden characters" + assert diag.security.call_args.kwargs["detail"] == "Edit the escaped value in decoded.yml" + def test_warning_only_reports(self): diag = MagicMock() v = ScanVerdict( diff --git a/tests/unit/utils/test_frontmatter_horizontal_rules.py b/tests/unit/utils/test_frontmatter_horizontal_rules.py new file mode 100644 index 000000000..2093537e8 --- /dev/null +++ b/tests/unit/utils/test_frontmatter_horizontal_rules.py @@ -0,0 +1,95 @@ +"""Unit tests for load_frontmatter handling of Markdown horizontal rules.""" + +import pytest +import yaml + +from apm_cli.utils.yaml_io import load_frontmatter + + +def test_load_frontmatter_with_valid_line1_header(tmp_path): + md_content = """--- +name: sample-skill +description: A sample skill +--- +# Main Content + +This is body content. +""" + file_path = tmp_path / "sample.skill.md" + file_path.write_text(md_content, encoding="utf-8") + + post = load_frontmatter(file_path) + assert post.metadata.get("name") == "sample-skill" + assert post.metadata.get("description") == "A sample skill" + assert "# Main Content" in post.content + + +def test_load_frontmatter_with_middle_horizontal_rules(tmp_path): + md_content = """# Dataverse Guide + +Overview text... + +--- + +## 1. Code Examples + +```python +def foo(): + pass +``` + +--- + +## 2. Next Steps +""" + file_path = tmp_path / "guide.instructions.md" + file_path.write_text(md_content, encoding="utf-8") + + # Before fix, middle horizontal rules caused ScannerError + post = load_frontmatter(file_path) + assert post.metadata == {} + assert "# Dataverse Guide" in post.content + assert "def foo():" in post.content + assert "## 2. Next Steps" in post.content + + +def test_load_frontmatter_with_supported_four_hyphen_fence(tmp_path): + md_content = """---- +name: sample-skill +---- +# Main Content +""" + file_path = tmp_path / "sample.skill.md" + file_path.write_text(md_content, encoding="utf-8") + + post = load_frontmatter(file_path) + assert post.metadata == {"name": "sample-skill"} + assert "# Main Content" in post.content + + +def test_load_frontmatter_with_indented_horizontal_rule_on_line1(tmp_path): + md_content = """ --- +# Guide + +--- + +interval: daily + +--- + +Body content. +""" + file_path = tmp_path / "guide.instructions.md" + file_path.write_text(md_content, encoding="utf-8") + + post = load_frontmatter(file_path) + assert post.metadata == {} + assert post.content == md_content + + +def test_load_frontmatter_with_unterminated_fence_raises_yaml_error(tmp_path): + file_path = tmp_path / "broken.instructions.md" + file_path.write_text("---\napplyTo: src/**\n# Missing closing fence\n", encoding="utf-8") + + with pytest.raises(yaml.YAMLError, match="malformed frontmatter delimiters"): + load_frontmatter(file_path) diff --git a/tests/unit/utils/test_patterns.py b/tests/unit/utils/test_patterns.py index b6e74a27c..02d12b42c 100644 --- a/tests/unit/utils/test_patterns.py +++ b/tests/unit/utils/test_patterns.py @@ -173,6 +173,10 @@ def test_escapes_tab(self): def test_yaml_safe_load_roundtrip(self): import yaml - for value in ['a"b', "a\\b", "a\nb", "**/src/**", "**/*.{css,scss}"]: + values = ['a"b', "a\\b", "a\nb", "**/src/**", "**/*.{css,scss}"] + values.extend(chr(codepoint) for codepoint in range(32)) + values.append(chr(127)) + for value in values: yaml_doc = f"k: {yaml_double_quote(value)}\n" + assert yaml_doc.isascii() assert yaml.safe_load(yaml_doc) == {"k": value}