diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index bb3490ca..0b81379d 100644 --- a/services/hermes/agent-configmap.yaml +++ b/services/hermes/agent-configmap.yaml @@ -330,6 +330,9 @@ data: is `/opt/coordinator/jenkins_build_evidence.py JOB [--branch BRANCH] [--commit SHA] --wait`; it distinguishes a genuinely terminal build from nested Jenkins execution metadata and returns bounded JSON/log evidence. + With a branch, the helper also tries the conventional `JOB-branches` + multibranch name. For JSON Gitea mutations, prefer repeatable shell-safe + `--field KEY=VALUE` arguments over hand-quoted JSON. After merging, observe the default-branch build to a terminal result and report exact commit, build, and test evidence. diff --git a/services/hermes/scripts/gitea_api.py b/services/hermes/scripts/gitea_api.py index a2188ba6..5544c398 100644 --- a/services/hermes/scripts/gitea_api.py +++ b/services/hermes/scripts/gitea_api.py @@ -78,15 +78,40 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: data_group.add_argument( "--data-file", type=Path, help="path to a JSON request body" ) + data_group.add_argument( + "--field", + action="append", + metavar="KEY=VALUE", + help=( + "repeatable JSON field; bare text remains a string while true, false, " + "null, numbers, objects, and arrays are decoded as JSON" + ), + ) return parser.parse_args(argv) +def parse_fields(values: list[str]) -> dict[str, object]: + """Build a JSON object from shell-safe, repeatable key/value arguments.""" + result: dict[str, object] = {} + for value in values: + key, separator, raw = value.partition("=") + if not separator or not key or any(char.isspace() for char in key): + raise ValueError("each --field must be KEY=VALUE with a non-space key") + try: + result[key] = json.loads(raw) + except json.JSONDecodeError: + result[key] = raw + return result + + def load_data(args: argparse.Namespace) -> object | None: """Decode the optional JSON body without involving a shell expansion.""" if args.data_json is not None: return json.loads(args.data_json) if args.data_file is not None: return json.loads(args.data_file.read_text(encoding="utf-8")) + if args.field is not None: + return parse_fields(args.field) return None diff --git a/services/hermes/scripts/jenkins_build_evidence.py b/services/hermes/scripts/jenkins_build_evidence.py index 8a7e621d..19fbf590 100644 --- a/services/hermes/scripts/jenkins_build_evidence.py +++ b/services/hermes/scripts/jenkins_build_evidence.py @@ -16,6 +16,8 @@ from dataclasses import asdict, dataclass JENKINS_HOME = "/var/jenkins_home/jobs" SAFE_JOB = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") SAFE_BRANCH = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$") +HIDDEN_ANSI = re.compile(r"\x1b\[8m.*?\x1b\[0m", re.DOTALL) +ANSI_ESCAPE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") @dataclass(frozen=True) @@ -143,7 +145,12 @@ def parse_build_xml( ) -def find_build( +def sanitize_log(value: str) -> str: + """Remove Jenkins hidden hyperlinks and terminal control sequences.""" + return ANSI_ESCAPE.sub("", HIDDEN_ANSI.sub("", value)) + + +def _find_build_for_job( job: str, branch: str | None, commit: str | None, @@ -168,11 +175,34 @@ def find_build( continue if commit and not (evidence.revision or "").startswith(commit): continue - log_tail = _kubectl("tail", "-n", str(log_lines), f"{build_path}/log") + log_tail = sanitize_log( + _kubectl("tail", "-n", str(log_lines), f"{build_path}/log") + ) return BuildEvidence(**{**asdict(evidence), "log_tail": log_tail}) return None +def find_build( + job: str, + branch: str | None, + commit: str | None, + log_lines: int, +) -> BuildEvidence | None: + """Find a build, accepting the common omitted ``-branches`` suffix.""" + candidates = [job] + if branch is not None and not job.endswith("-branches"): + candidates.append(f"{job}-branches") + last_error: Exception | None = None + for candidate in candidates: + try: + return _find_build_for_job(candidate, branch, commit, log_lines) + except (OSError, ValueError, subprocess.SubprocessError) as exc: + last_error = exc + if last_error is not None: + raise last_error + return None + + def parse_args(argv: list[str] | None = None) -> argparse.Namespace: """Parse the bounded Jenkins evidence request.""" parser = argparse.ArgumentParser(description=__doc__) diff --git a/testing/tests/test_hermes_runtime_access.py b/testing/tests/test_hermes_runtime_access.py index d7be46ff..7e455eb6 100644 --- a/testing/tests/test_hermes_runtime_access.py +++ b/testing/tests/test_hermes_runtime_access.py @@ -61,6 +61,27 @@ def test_gitea_api_rejects_foreign_or_non_api_targets(path: str): gitea_api.api_url("https://scm.bstein.dev", path) +def test_gitea_api_parses_shell_safe_fields_as_json_values(): + gitea_api = _load("gitea_api") + + assert gitea_api.parse_fields( + ["Do=manually-merged", "MergeCommitID=abc123", "enabled=true", "count=2"] + ) == { + "Do": "manually-merged", + "MergeCommitID": "abc123", + "enabled": True, + "count": 2, + } + + +@pytest.mark.parametrize("value", ["missing-separator", "=missing-key", "bad key=x"]) +def test_gitea_api_rejects_invalid_shell_safe_fields(value: str): + gitea_api = _load("gitea_api") + + with pytest.raises(ValueError): + gitea_api.parse_fields([value]) + + def test_jenkins_evidence_ignores_nested_execution_result(): evidence_reader = _load("jenkins_build_evidence") evidence = evidence_reader.parse_build_xml( @@ -102,6 +123,33 @@ def test_jenkins_evidence_reads_only_top_level_terminal_result(): assert evidence.duration_ms == 250 +def test_jenkins_evidence_removes_hidden_links_and_ansi_control_sequences(): + evidence_reader = _load("jenkins_build_evidence") + + value = ( + "before\n\x1b[8mha:////private-jenkins-payload\x1b[0m[Pipeline] sh\n" + "\x1b[31m5 passed\x1b[0m\n" + ) + + assert evidence_reader.sanitize_log(value) == "before\n[Pipeline] sh\n5 passed\n" + + +def test_jenkins_evidence_falls_back_to_multibranch_suffix(monkeypatch): + evidence_reader = _load("jenkins_build_evidence") + calls = [] + + def fake_find(job, branch, commit, log_lines): + calls.append((job, branch, commit, log_lines)) + if job == "demo": + raise ValueError("not a multibranch job") + return "evidence" + + monkeypatch.setattr(evidence_reader, "_find_build_for_job", fake_find) + + assert evidence_reader.find_build("demo", "master", "abc", 20) == "evidence" + assert [call[0] for call in calls] == ["demo", "demo-branches"] + + @pytest.mark.parametrize("branch", ["../master", "/master", "feature//unsafe"]) def test_jenkins_evidence_rejects_unsafe_branch_paths(branch: str): evidence_reader = _load("jenkins_build_evidence")