b7cf256c44
Helper tests/invariants/forge_credentials.py detects credential carriers via mounts[]/env/secrets_required[]/ssh_hosts[]; today gitea-admin, gitea-ssh, gitea-ssh-accelerators, gitea-ssh-homelab are found and none declare write_targets — the invariant xfails(strict) until the CRS-103 rollout stamps them. Claude-Session: https://claude.ai/code/session_018LRh1S2bHJYhEgrMu3UJx8
127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
"""HC-11 helper: detect harness layers that carry forge (git host) credentials.
|
|
|
|
Spec: /home/paul/dev/claude/projects/agent-runtimes/spec/harness.md HC-11
|
|
(2026-09-13 amendment, bug b1de17b4).
|
|
|
|
A layer "has forge credentials" when ANY of:
|
|
- a ``mounts[]`` entry has one of the keys
|
|
``{"secret", "secret_name", "secret_ref", "vault_path"}`` whose string
|
|
value equals a name in ``FORGE_CREDENTIAL_NAMES`` or ends with
|
|
``"/" + name`` for some such name, OR
|
|
- any ``env`` value contains such a name as a whole token.
|
|
|
|
Comments and Jinja placeholders are never matched (this module only inspects
|
|
parsed YAML string values, never raw file text).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
# Closed set (HC-11) — maintained here per spec.
|
|
FORGE_CREDENTIAL_NAMES = {
|
|
"GITEA_SSH_KEY",
|
|
"GITEA_TOKEN",
|
|
"GITEA_DEPLOY_KEY",
|
|
"AGENT_REPO_DEPLOY_KEY",
|
|
"AGENT_REPO_TOKEN",
|
|
"SSH_KEY_AI_ENABLEMENT",
|
|
"gitea-admin",
|
|
}
|
|
|
|
_MOUNT_KEYS = {"secret", "secret_name", "secret_ref", "vault_path"}
|
|
|
|
# "whole token" — the name bounded by non-alnum/underscore characters (or
|
|
# string edges) so a substring match doesn't false-positive.
|
|
_TOKEN_RE_CACHE: dict[str, re.Pattern[str]] = {}
|
|
|
|
|
|
def _token_pattern(name: str) -> re.Pattern[str]:
|
|
pat = _TOKEN_RE_CACHE.get(name)
|
|
if pat is None:
|
|
pat = re.compile(r"(?<![A-Za-z0-9_])" + re.escape(name) + r"(?![A-Za-z0-9_])")
|
|
_TOKEN_RE_CACHE[name] = pat
|
|
return pat
|
|
|
|
|
|
def _mount_value_matches(value: object) -> bool:
|
|
if not isinstance(value, str):
|
|
return False
|
|
for name in FORGE_CREDENTIAL_NAMES:
|
|
if value == name or value.endswith("/" + name):
|
|
return True
|
|
return False
|
|
|
|
|
|
_JINJA_RE = re.compile(r"\{\{.*?\}\}")
|
|
|
|
|
|
def _env_value_matches(value: object) -> bool:
|
|
if not isinstance(value, str):
|
|
return False
|
|
# Jinja placeholders (e.g. "{{ GITEA_TOKEN }}") are never matched — they
|
|
# name a template variable, not a literal credential reference.
|
|
stripped = _JINJA_RE.sub("", value)
|
|
for name in FORGE_CREDENTIAL_NAMES:
|
|
if _token_pattern(name).search(stripped):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _layer_has_forge_credentials(parsed: dict) -> bool:
|
|
mounts = parsed.get("mounts")
|
|
if isinstance(mounts, list):
|
|
for entry in mounts:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
for key in _MOUNT_KEYS:
|
|
if key in entry and _mount_value_matches(entry[key]):
|
|
return True
|
|
|
|
env = parsed.get("env")
|
|
if isinstance(env, dict):
|
|
for value in env.values():
|
|
if _env_value_matches(value):
|
|
return True
|
|
|
|
secrets_required = parsed.get("secrets_required")
|
|
if isinstance(secrets_required, list):
|
|
for entry in secrets_required:
|
|
if isinstance(entry, dict) and entry.get("name") in FORGE_CREDENTIAL_NAMES:
|
|
return True
|
|
|
|
ssh_hosts = parsed.get("ssh_hosts")
|
|
if isinstance(ssh_hosts, list):
|
|
for entry in ssh_hosts:
|
|
if isinstance(entry, dict) and entry.get("identity_secret") in FORGE_CREDENTIAL_NAMES:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def layers_with_forge_credentials(layers_root: Path) -> list[tuple[str, Path]]:
|
|
"""Walk every ``harnesses/contexts/*/v*/harness.yaml`` under *layers_root*.
|
|
|
|
Returns ``(layer_name, yaml_path)`` for each layer with forge credentials
|
|
per the rules above.
|
|
"""
|
|
out: list[tuple[str, Path]] = []
|
|
contexts_dir = layers_root / "contexts"
|
|
if not contexts_dir.is_dir():
|
|
return out
|
|
|
|
for yaml_path in sorted(contexts_dir.glob("*/v*/harness.yaml")):
|
|
try:
|
|
parsed = yaml.safe_load(yaml_path.read_text(encoding="utf-8"))
|
|
except (OSError, yaml.YAMLError):
|
|
continue
|
|
if not isinstance(parsed, dict):
|
|
continue
|
|
if _layer_has_forge_credentials(parsed):
|
|
name = str(parsed.get("name") or yaml_path.parent.parent.name)
|
|
out.append((name, yaml_path))
|
|
|
|
return out
|