diff --git a/CHANGELOG.md b/CHANGELOG.md index b2ed0df..b37d853 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,47 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + + +## [1.2.0] - 2026-08-03 + +### Added +- `tirith platform check`: run an organization's policies against a plan, state or arbitrary JSON + document from CI or a laptop. Masks the document locally, packs it with the terraform source into + an archive, uploads it, creates a StackGuardian run, polls it and reports the verdict as JSON + and/or markdown. +- `ExitStatus.ERROR_POLICY_FAILED` (3), so a caller can tell "a policy said no" from "tirith could + not reach the platform". Exit 1 stays reserved for the latter, and applies even without + `--fail-on-error`: a run that produced no verdict must never look like a pass. + +### Changed +- `cli.main(args=...)` is now honoured. It previously called `parse_args()` with no argument, so + the parameter was ignored and the CLI could only ever read `sys.argv`. + +### Notes +- The local evaluation surface is unchanged, including its single-dash long options. Subcommands + are dispatched before the flat parser sees anything, so `--json` output stays byte-identical. +- No new runtime dependencies: the platform integration is stdlib-only. + +## [1.1.0] - 2026-08-01 + +### Added +- `core`: Policy metadata passthrough — `meta.id`, `meta.name`, `meta.description`, + `meta.severity`, `meta.enforcement`, `meta.tags` and `meta.remediation` now reach the result + document when a policy declares them. Keys that are absent are omitted, so the output of a + policy declaring none of them is unchanged. `{{ var.x }}` substitution works in all of them. + +### Fixed +- `core`: Variable substitution no longer mutates the caller's policy dictionary. Evaluating the + same parsed policy more than once (a policy set, or a retry) previously leaked substituted + values from one evaluation into the next. +- `core`: An unsupported `condition.type` now populates `result` instead of returning without it, + which raised `KeyError` in the pretty printer far from the real cause. +- `core`: Provider errors reported without a `ProviderError` severity are now surfaced instead of + being discarded and `None` evaluated against the condition — a typo'd `operation_type` read as + a genuine policy violation. These are treated as malformed provider calls and are deliberately + not subject to `error_tolerance`. + ## [1.0.5] - 2025-11-19 ### Fixed diff --git a/setup.py b/setup.py index 7d07cb9..667e0b5 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ def read(*names, **kwargs): setup( name="py-tirith", - version="1.0.5", + version="1.2.0", license="Apache", description="Tirith simplifies defining Policy as Code.", long_description_content_type="text/markdown", diff --git a/src/tirith/__init__.py b/src/tirith/__init__.py index 151dee5..4c2aac7 100644 --- a/src/tirith/__init__.py +++ b/src/tirith/__init__.py @@ -2,6 +2,6 @@ tirith: Execute policies defined using Tirith (StackGuardian Policy Framework) """ -__version__ = "1.0.5" +__version__ = "1.2.0" __author__ = "StackGuardian" __license__ = "Apache" diff --git a/src/tirith/cli.py b/src/tirith/cli.py index 6642e31..1b314f8 100755 --- a/src/tirith/cli.py +++ b/src/tirith/cli.py @@ -15,7 +15,6 @@ from .core import start_policy_evaluation - logger = logging.getLogger(__name__) @@ -27,6 +26,13 @@ def eprint(*args, **kwargs): print(*args, file=sys.stderr, **kwargs) +# Subcommands are dispatched before the flat parser sees anything. argparse cannot express an +# optional subcommand alongside options like `-policy-path` (a single dash and a long name), and the +# local-evaluation surface is a contract: tests/core/test_output_compatibility.py asserts its --json +# output is byte-identical to a golden file. An explicit pre-dispatch leaves that untouched. +SUBCOMMANDS = {"platform"} + + def main(args=None) -> ExitStatus: """ The main function. @@ -36,6 +42,13 @@ def main(args=None) -> ExitStatus: Return exit status code. """ + argv = list(sys.argv[1:] if args is None else args) + + if argv and argv[0] in SUBCOMMANDS: + from tirith.platform import cli as platform_cli + + return platform_cli.main(argv) + try: class _WidthFormatter(argparse.RawTextHelpFormatter): @@ -45,8 +58,7 @@ def __init__(self, prog="PROG") -> None: parser = argparse.ArgumentParser( description="Tirith (StackGuardian Policy Framework)", formatter_class=_WidthFormatter, - epilog=textwrap.dedent( - """\ + epilog=textwrap.dedent("""\ About Tirith: * Abstract away the implementation complexity of policy engine underneath. @@ -55,8 +67,7 @@ def __init__(self, prog="PROG") -> None: * Provide modularity to enable easy extensibility * Github - https://github.com/StackGuardian/tirith * Docs - https://docs.stackguardian.io/docs/tirith/overview - """ - ), + """), ) parser.add_argument( "-policy-path", @@ -104,9 +115,9 @@ def __init__(self, prog="PROG") -> None: ) parser.add_argument("--version", action="version", version=__version__) - args = parser.parse_args() + args = parser.parse_args(argv) - if len(sys.argv) == 1: + if not argv: parser.print_help() sys.exit(0) diff --git a/src/tirith/core/core.py b/src/tirith/core/core.py index 27c6064..5c49afe 100644 --- a/src/tirith/core/core.py +++ b/src/tirith/core/core.py @@ -12,7 +12,6 @@ from .evaluators import EVALUATORS_DICT from .policy_parameterization import get_policy_with_vars_replaced - logger = logging.getLogger(__name__) @@ -50,6 +49,10 @@ def generate_evaluator_result(evaluator_obj, input_data, provider_module): evaluator_class = EVALUATORS_DICT.get(evaluator_name) if evaluator_class is None: logger.error(f"{evaluator_name} is not a supported evaluator") + # Always populate "result" before returning. Consumers (the pretty printer, the + # workflow-step templates, the platform) index into it unconditionally, and an + # early return without it used to raise KeyError far away from the real cause. + result["result"] = [{"passed": False, "message": f"`{evaluator_name}` is not a supported evaluator"}] return result evaluator_instance = evaluator_class() @@ -66,6 +69,17 @@ def generate_evaluator_result(evaluator_obj, input_data, provider_module): has_valid_evaluation = False for evaluator_input in evaluator_inputs: + # A provider reported an error without attaching a ProviderError severity. That means a + # malformed provider call -- an unsupported operation_type, a missing required argument -- + # not a policy violation. Surface the message and fail hard: error_tolerance exists to + # tolerate missing data, never to mask a broken policy. Without this branch the error text + # is discarded and `None` is evaluated against the condition, so a typo'd operation_type + # reads as a genuine violation. + if evaluator_input.get("err") and not isinstance(evaluator_input["value"], ProviderError): + evaluation_results.append({"passed": False, "message": evaluator_input["err"]}) + has_evaluation_passed = False + continue + if isinstance(evaluator_input["value"], ProviderError) and evaluator_input.get("err", None): severity_value = evaluator_input["value"].severity_value err_result = dict(message=evaluator_input["err"]) @@ -302,8 +316,16 @@ def start_policy_evaluation_from_dict(policy_dict: Dict, input_dict: Dict, var_d eval_results.append(eval_result) final_evaluation_result, errors = final_evaluator(final_evaluation_policy_string, eval_results_obj) + # Pass policy-declared metadata through to the result, but only the keys that are actually + # present. Absent keys are omitted rather than emitted as null, so the output of a policy + # that declares none of them is byte-identical to what it was before this was added. + final_output_meta = {"version": policy_meta.get("version"), "required_provider": provider_module} + for meta_key in ("id", "name", "description", "severity", "enforcement", "tags", "remediation"): + if meta_key in policy_meta: + final_output_meta[meta_key] = policy_meta[meta_key] + final_output = { - "meta": {"version": policy_meta.get("version"), "required_provider": provider_module}, + "meta": final_output_meta, "final_result": final_evaluation_result, "evaluators": eval_results, "errors": errors, diff --git a/src/tirith/core/policy_parameterization.py b/src/tirith/core/policy_parameterization.py index ce81daf..c34092a 100644 --- a/src/tirith/core/policy_parameterization.py +++ b/src/tirith/core/policy_parameterization.py @@ -1,3 +1,4 @@ +import copy import re import pydash @@ -52,11 +53,17 @@ def get_policy_with_vars_replaced(policy_dict: dict, var_dict: dict) -> Tuple[di """ Replace the variables in the policy_dict with the values from the var_dict + The caller's `policy_dict` is never mutated: substitution happens on a deep copy. This + matters when the same parsed policy is evaluated more than once (for example a policy set + run against several inputs, or a retry), where substituted values would otherwise leak + from one evaluation into the next. + :param policy_dict: The policy dictionary :param var_dict: The dictionary containing the variables - :return: The policy dictionary with the variables replaced + :return: A copy of the policy dictionary with the variables replaced and the list of variables that are not found """ + policy_dict = copy.deepcopy(policy_dict) not_found_vars = [] # Replace vars in the meta key _replace_vars_in_dict(policy_dict["meta"], var_dict, not_found_vars) diff --git a/src/tirith/platform/__init__.py b/src/tirith/platform/__init__.py new file mode 100644 index 0000000..ae9467b --- /dev/null +++ b/src/tirith/platform/__init__.py @@ -0,0 +1,6 @@ +""" +StackGuardian platform integration. + +Everything here is stdlib-only on purpose: tirith has three runtime dependencies and none of them +are an HTTP library, so a CI runner needs nothing installed beyond tirith itself. +""" diff --git a/src/tirith/platform/archive.py b/src/tirith/platform/archive.py new file mode 100644 index 0000000..f1e15e2 --- /dev/null +++ b/src/tirith/platform/archive.py @@ -0,0 +1,202 @@ +""" +Build the gzipped tar that carries a run's inputs to StackGuardian. + +The archive is what the run controller unpacks in place of a VCS checkout, so it holds both the +terraform source and the documents to evaluate, at the fixed names the step looks for: + + plan.json terraform plan JSON -- the primary policy input + tfstate.json terraform state JSON + infracost.json cost breakdown + +Two things here are easy to get wrong and expensive to get wrong. + +**The masked documents go in, never the originals.** `pack()` takes already-redacted objects and +serializes them itself; it never copies plan.json off disk. A caller that packed the source +directory *first* and masked afterwards would ship the plaintext file alongside the masked one. The +tests assert on the bytes inside the resulting tarball for this reason -- asserting on the dict +that was passed in would pass while the archive leaked. + +**`.terraform/` must be excluded.** A provider cache is routinely hundreds of megabytes; including +it would make every run upload the AWS provider. `*.tfstate*` is excluded for the same reason as +the first point: an unmasked state file sitting in the working directory would otherwise travel +next to the masked copy. +""" + +import fnmatch +import io +import os +import tarfile + +# Fixed names the policy-only step looks for at the archive root. +PLAN_DOCUMENT = "plan.json" +STATE_DOCUMENT = "tfstate.json" +INFRACOST_DOCUMENT = "infracost.json" + +# These names are ALWAYS written by pack(), never copied from the source tree -- whether or not a +# masked document was supplied for them. A file called tfstate.json in the working directory is raw, +# unmasked state; see the note in pack(). +RESERVED_DOCUMENTS = frozenset((PLAN_DOCUMENT, STATE_DOCUMENT, INFRACOST_DOCUMENT)) + +# Always excluded, regardless of .gitignore. +# +# .terraform/ provider binaries and modules; hundreds of MB, and the runner does its own init +# .git/ full history, so anything ever committed would ship +# *.tfstate* raw state -- unmasked by definition, including .backup files +# .terraform.lock.hcl is deliberately NOT excluded: it pins provider versions and is small. +DEFAULT_EXCLUDES = ( + ".git", + ".terraform", + "*.tfstate", + "*.tfstate.*", + "*.tfstate.backup", + "__pycache__", + "*.pyc", + ".venv", + "node_modules", +) + +# Refuse to build anything larger than this. A runaway archive is nearly always an exclusion that +# did not fire, and failing loudly beats a five-minute upload that times out the run. +MAX_ARCHIVE_BYTES = 100 * 1024 * 1024 + + +class ArchiveError(Exception): + """The archive could not be built.""" + + +def _load_gitignore_patterns(source_dir): + """ + Read .gitignore into fnmatch patterns. + + Deliberately simple: leading `/` and trailing `/` are stripped, negations (`!`) are ignored. + A full gitignore implementation is not worth it here -- DEFAULT_EXCLUDES covers the cases that + actually matter, and .gitignore is a convenience on top. + """ + path = os.path.join(source_dir, ".gitignore") + patterns = [] + try: + with open(path, "r", encoding="utf-8", errors="replace") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#") or line.startswith("!"): + continue + patterns.append(line.strip("/")) + except OSError: + return [] + return patterns + + +def _is_excluded(relative_path, name, patterns): + """Match a path against the exclusion patterns, by both basename and full relative path.""" + for pattern in patterns: + if fnmatch.fnmatch(name, pattern) or fnmatch.fnmatch(relative_path, pattern): + return True + # A directory pattern excludes everything beneath it. + if relative_path.startswith(pattern + os.sep): + return True + return False + + +def pack(source_dir, plan=None, state=None, infracost=None, extra_excludes=(), respect_gitignore=True): + """ + Build the archive in memory and return its bytes. + + `plan`, `state` and `infracost` are already-redacted objects. They are serialized here and + written at the archive root, overriding any same-named file in `source_dir` -- so a stale + plan.json lying around cannot displace the masked one. + + Returns (archive_bytes, manifest) where manifest lists what went in, for logging. + """ + if source_dir and not os.path.isdir(source_dir): + raise ArchiveError(f"Source directory does not exist: {source_dir}") + + patterns = list(DEFAULT_EXCLUDES) + list(extra_excludes) + if respect_gitignore and source_dir: + patterns += _load_gitignore_patterns(source_dir) + + documents = {} + if plan is not None: + documents[PLAN_DOCUMENT] = plan + if state is not None: + documents[STATE_DOCUMENT] = state + if infracost is not None: + documents[INFRACOST_DOCUMENT] = infracost + + buffer = io.BytesIO() + manifest = {"documents": sorted(documents), "files": 0, "skipped": 0} + + with tarfile.open(fileobj=buffer, mode="w:gz") as tar: + if source_dir: + # RESERVED_DOCUMENTS, not just the ones being written. A file named tfstate.json in the + # working directory is unmasked by definition -- `terraform state pull > state.json` is + # the documented way to produce one -- so packing it would ship every attribute in + # plaintext beside the masked copy. If the caller wants it evaluated they pass + # --state-path, which masks it first. + manifest["files"], manifest["skipped"] = _add_tree(tar, source_dir, patterns, RESERVED_DOCUMENTS) + for name, document in documents.items(): + _add_document(tar, name, document) + + archive = buffer.getvalue() + if len(archive) > MAX_ARCHIVE_BYTES: + raise ArchiveError( + f"Archive is {len(archive) // (1024 * 1024)} MB, over the {MAX_ARCHIVE_BYTES // (1024 * 1024)} MB " + "limit. This usually means a large directory was not excluded -- check for provider " + "caches or build output, and pass extra excludes if needed." + ) + + manifest["bytes"] = len(archive) + return archive, manifest + + +def _add_tree(tar, source_dir, patterns, reserved_names): + """Walk `source_dir`, adding everything not excluded. Returns (added, skipped).""" + added = 0 + skipped = 0 + + for root, dirs, files in os.walk(source_dir): + relative_root = os.path.relpath(root, source_dir) + relative_root = "" if relative_root == "." else relative_root + + # Prune in place so os.walk does not descend into excluded directories at all -- the point + # of excluding .terraform is not to read it. + kept_dirs = [] + for d in dirs: + relative = os.path.join(relative_root, d) if relative_root else d + if _is_excluded(relative, d, patterns): + skipped += 1 + else: + kept_dirs.append(d) + dirs[:] = kept_dirs + + for name in files: + relative = os.path.join(relative_root, name) if relative_root else name + if _is_excluded(relative, name, patterns): + skipped += 1 + continue + # The masked documents are written separately and must win. + if relative in reserved_names: + skipped += 1 + continue + full = os.path.join(root, name) + if os.path.islink(full): + # A symlink out of the tree would either break on extraction or smuggle a file in. + skipped += 1 + continue + try: + tar.add(full, arcname=relative) + added += 1 + except OSError: + skipped += 1 + + return added, skipped + + +def _add_document(tar, name, document): + """Serialize one document straight into the tar, never via a file on disk.""" + import json + + payload = document if isinstance(document, bytes) else json.dumps(document).encode("utf-8") + info = tarfile.TarInfo(name=name) + info.size = len(payload) + info.mode = 0o644 + tar.addfile(info, io.BytesIO(payload)) diff --git a/src/tirith/platform/check.py b/src/tirith/platform/check.py new file mode 100644 index 0000000..3ea45dd --- /dev/null +++ b/src/tirith/platform/check.py @@ -0,0 +1,224 @@ +""" +Orchestration for `tirith platform check`. + + read -> mask -> pack -> ensure workflow -> upload archive -> create run -> poll -> fetch -> report + +The masking is the part that matters most and it happens *here*, on the caller's machine, before +anything leaves it. Masking server-side would be theatre: once the bytes arrive the exposure has +already happened. +""" + +import json +import os +import sys + +from . import archive, redact, report +from .client import SGClient, SGError + +DEFAULT_WORKFLOW_GROUP = "default" +DEFAULT_TERRAFORM_VERSION = "1.5.7" + +# What the CLI understands as an input document. `terraform_state` exists as a distinct kind from +# `json` purely so this side knows to mask it -- tirith itself has no state provider, and the step +# routes it to the json provider. +INPUT_KINDS = ("terraform_plan", "terraform_state", "kubernetes", "json") + + +class CheckError(Exception): + """The check could not be completed. Always fails closed.""" + + +def log(message): + """Progress goes to stderr so stdout stays clean for machine-readable output.""" + print(message, file=sys.stderr, flush=True) + + +def read_json(path, label): + if not os.path.exists(path): + raise CheckError(f"{label} not found: {path}") + try: + with open(path, "r") as f: + return json.load(f) + except json.JSONDecodeError as e: + raise CheckError(f"{label} is not valid JSON ({path}): {e}") + except OSError as e: + raise CheckError(f"Could not read {label} ({path}): {e}") + + +def prepare_documents(input_path, input_kind, state_path, infracost_path): + """ + Read and mask everything that will go into the archive. + + Returns (plan, state, infracost, redaction_count). The returned objects are the *masked* ones; + nothing downstream should ever touch the originals again. + """ + plan = None + state = None + redactions = 0 + + if input_path: + document = read_json(input_path, "input document") + if input_kind == "terraform_plan": + plan = redact.redact_plan(document) + redactions += redact.count_redactions(plan) + elif input_kind == "terraform_state": + state = redact.redact_state(document) + redactions += redact.count_redactions(state) + else: + # kubernetes / json: no marker structure to drive masking, so it goes as-is. Warn if it + # looks like state, because that is the mistake that would ship every attribute in + # plaintext. + if isinstance(document, dict) and {"version", "lineage", "resources"} <= set(document): + log( + "WARNING: this document looks like terraform state but --input-kind is " + f"'{input_kind}', so it will NOT be masked. Use --input-kind terraform_state." + ) + plan = document + + if state_path: + state_document = read_json(state_path, "state document") + masked_state = redact.redact_state(state_document) + redactions += redact.count_redactions(masked_state) + if state is None: + state = masked_state + else: + log("Both --input-path and --state-path are state documents; using --input-path") + + infracost = read_json(infracost_path, "cost breakdown") if infracost_path else None + + return plan, state, infracost, redactions + + +def terraform_config(terraform_version, policy_input_kind, step_template_id): + """ + The workflow's stored configuration. + + core synthesises the run's steps from this plus the per-run TerraformAction, so anything the + step needs that does not vary per run belongs here. + """ + config = { + "terraformVersion": terraform_version or DEFAULT_TERRAFORM_VERSION, + "managedTerraformState": False, + "policyInputKind": policy_input_kind, + } + if step_template_id: + config["wfStepTemplateRevisionId"] = step_template_id + return config + + +def write_output_json(path, payload): + if not path: + return + try: + with open(path, "w") as f: + json.dump(payload, f, indent=2) + except OSError as e: + log(f"WARNING: could not write {path}: {e}") + + +def run_check(opts): + """ + Execute the check. Returns the result document. + + Raises CheckError for anything that leaves the verdict unknown -- the caller maps that to a + non-zero exit regardless of --fail-on-error, because a run that produced no verdict must never + look like a pass. + """ + client = SGClient(opts.api_url, opts.org, opts.api_key, timeout=60) + + plan, state, infracost, redactions = prepare_documents( + opts.input_path, opts.input_kind, opts.state_path, opts.infracost_path + ) + if redactions: + log(f"Masked {redactions} sensitive value(s) before upload") + + archive_bytes, manifest = archive.pack( + source_dir=opts.source_dir, + plan=plan, + state=state, + infracost=infracost, + ) + log( + f"Packed {manifest['files']} file(s) and {len(manifest['documents'])} document(s) " + f"into {manifest['bytes'] // 1024} KB" + ) + + try: + client.ensure_workflow_group(opts.workflow_group) + client.ensure_workflow( + opts.workflow_group, + opts.workflow_id, + f"Policy checks for {opts.workflow_id}", + terraform_config(opts.terraform_version, opts.input_kind, opts.step_template_id), + ) + + key = client.upload_archive( + opts.workflow_group, + opts.workflow_id, + f"{opts.artifact_tag}.tar.gz", + opts.sha[:7] if opts.sha else "latest", + archive_bytes, + ) + log(f"Uploaded the project archive: {key}") + + run_id, _data = client.create_run(opts.workflow_group, opts.workflow_id, key, opts.trigger_details) + except SGError as e: + raise CheckError(str(e)) + + run_url = ( + f"{opts.dashboard_url.rstrip('/')}/orchestrator/orgs/{opts.org}" + f"/wfgrps/{opts.workflow_group}/wfs/{opts.workflow_id}/wfruns/{run_id}" + ) + log(f"Run created: {run_url}") + + # Written before polling so a timeout still leaves the run discoverable. + write_output_json(opts.output_json, {"status": "RUNNING", "wfrun_id": run_id, "wfrun_url": run_url}) + + try: + status, _run = client.wait_for_run( + opts.workflow_group, + opts.workflow_id, + run_id, + timeout=opts.timeout, + on_poll=lambda s: log(f"Run status: {s}"), + ) + except SGError as e: + raise CheckError(f"{e} (run: {run_url})") + + policy_results = client.get_results_artifact(opts.workflow_group, opts.workflow_id, f"{run_id}/tirith-results.json") + if policy_results is None: + policy_results = client.get_policy_results(opts.workflow_group, opts.workflow_id, run_id) + + counts, _findings = report.summarize(policy_results) + verdict_value = report.verdict(counts, status) + + result = { + "status": status, + "verdict": verdict_value, + "counts": { + "passed": counts.get(report.PASS, 0), + "failed": counts.get(report.FAIL, 0), + "warned": counts.get(report.WARN, 0), + "approval_required": counts.get(report.APPROVAL_REQUIRED, 0), + "skipped": counts.get("SKIPPED", 0), + }, + "headline": report.headline(counts, verdict_value), + "wfrun_id": run_id, + "wfrun_url": run_url, + "policy_results": policy_results or {}, + } + + write_output_json(opts.output_json, result) + + if opts.output_markdown: + body = report.render_markdown( + policy_results, status, run_url, marker=opts.comment_marker, limit=opts.markdown_limit + ) + try: + with open(opts.output_markdown, "w") as f: + f.write(body) + except OSError as e: + log(f"WARNING: could not write {opts.output_markdown}: {e}") + + log(result["headline"]) + return result diff --git a/src/tirith/platform/cli.py b/src/tirith/platform/cli.py new file mode 100644 index 0000000..bd4e6bd --- /dev/null +++ b/src/tirith/platform/cli.py @@ -0,0 +1,175 @@ +""" +`tirith platform ...` -- run policy checks against a StackGuardian organization. + +Flag and environment names follow sg-cli (SG_API_TOKEN, SG_BASE_URL, SG_ORG, SG_DASHBOARD_URL) so +someone who knows one tool knows the other. +""" + +import argparse +import json +import os +import sys + +from ..status import ExitStatus +from .check import DEFAULT_WORKFLOW_GROUP, INPUT_KINDS, CheckError, log, run_check + +DEFAULT_API_URL = "https://api.app.stackguardian.io/api/v1" +DEFAULT_DASHBOARD_URL = "https://app.stackguardian.io" + + +def _resolve_api_key(value): + """ + Resolve the API key, preferring the environment. + + A key on argv is visible in `ps` for the lifetime of the process, so `-` reads it from stdin + and $SG_API_TOKEN is the documented default. + """ + if value == "-": + return sys.stdin.readline().strip() + return value or os.environ.get("SG_API_TOKEN", "") + + +def _load_trigger_details(opts): + if opts.trigger_details_json: + source, raw = "--trigger-details-json", opts.trigger_details_json + elif opts.trigger_details_file: + source = f"--trigger-details-file {opts.trigger_details_file}" + try: + with open(opts.trigger_details_file) as f: + raw = f.read() + except OSError as e: + raise CheckError(f"Could not read {opts.trigger_details_file}: {e}") + else: + return {"type": "cli"} + + try: + details = json.loads(raw) + except json.JSONDecodeError as e: + raise CheckError(f"{source} is not valid JSON: {e}") + if not isinstance(details, dict): + raise CheckError(f"{source} must be a JSON object") + details.setdefault("type", "cli") + return details + + +def build_parser(): + parser = argparse.ArgumentParser( + prog="tirith platform", + description="Run StackGuardian policy checks from a CI pipeline or a laptop.", + ) + sub = parser.add_subparsers(dest="subcommand") + + check = sub.add_parser( + "check", + help="Evaluate the organization's policies against a document and report the verdict.", + description=( + "Masks the document, packs it with the terraform source into an archive, uploads it, " + "runs the policies on StackGuardian and reports the verdict." + ), + ) + + identity = check.add_argument_group("identity") + identity.add_argument( + "--api-key", default=None, help="API key, or '-' to read it from stdin. Default: $SG_API_TOKEN" + ) + identity.add_argument("--org", default=None, help="Organization name. Default: $SG_ORG") + identity.add_argument("--api-url", default=None, help=f"API base URL. Default: $SG_BASE_URL or {DEFAULT_API_URL}") + identity.add_argument("--dashboard-url", default=None, help="Dashboard base URL, used to build run links.") + + workflow = check.add_argument_group("workflow") + workflow.add_argument("--workflow-id", required=True, help="Slug identifying the workflow. Created if absent.") + workflow.add_argument("--workflow-group", default=DEFAULT_WORKFLOW_GROUP, help="Workflow group. Created if absent.") + workflow.add_argument("--terraform-version", default=None, help="Stored on the workflow at creation.") + workflow.add_argument( + "--step-template-id", + default=None, + help="Override the terraform step template. Omit to use the platform's own default.", + ) + + inputs = check.add_argument_group("inputs") + inputs.add_argument("--input-path", default=None, help="Document to evaluate, e.g. `terraform show -json tfplan`.") + inputs.add_argument("--input-kind", default="terraform_plan", choices=INPUT_KINDS) + inputs.add_argument("--state-path", default=None, help="Optional terraform state, masked before upload.") + inputs.add_argument("--infracost-path", default=None, help="Optional `infracost breakdown --format json`.") + inputs.add_argument("--source-dir", default=".", help="Terraform source to pack alongside the documents.") + inputs.add_argument("--no-source", action="store_true", help="Send only the documents, not the source tree.") + + run = check.add_argument_group("run") + run.add_argument("--sha", default=None, help="Commit SHA, used to namespace the uploaded archive.") + run.add_argument("--artifact-tag", default="default", help="Namespaces the archive within a commit.") + run.add_argument("--trigger-details-json", default=None, help="JSON object describing what triggered this run.") + run.add_argument("--trigger-details-file", default=None, help="File containing that JSON object.") + run.add_argument("--timeout", type=int, default=1800, help="Seconds to wait for the run. Default: 1800") + + output = check.add_argument_group("output") + output.add_argument("--output-json", default=None, help="Write the result document here.") + output.add_argument("--output-markdown", default=None, help="Write a markdown report here.") + output.add_argument("--comment-marker", default=None, help="Opaque first line of the markdown, for stickiness.") + output.add_argument("--markdown-limit", type=int, default=60000, help="Truncate the markdown to this length.") + output.add_argument( + "--fail-on-error", + action="store_true", + help=( + "Exit non-zero when a policy fails. An unreachable platform or a run that produced no " + "verdict always exits non-zero regardless of this flag." + ), + ) + + return parser + + +def main(argv): + parser = build_parser() + opts = parser.parse_args(argv[1:]) + + if opts.subcommand != "check": + parser.print_help() + return ExitStatus.SUCCESS + + opts.api_key = _resolve_api_key(opts.api_key) + opts.org = opts.org or os.environ.get("SG_ORG", "") + opts.api_url = opts.api_url or os.environ.get("SG_BASE_URL") or DEFAULT_API_URL + opts.dashboard_url = opts.dashboard_url or os.environ.get("SG_DASHBOARD_URL") or DEFAULT_DASHBOARD_URL + opts.source_dir = None if opts.no_source else opts.source_dir + + missing = [name for name, value in (("--api-key", opts.api_key), ("--org", opts.org)) if not value] + if missing: + log(f"ERROR: missing required {' and '.join(missing)}") + return ExitStatus.ERROR + + if not opts.input_path and not opts.state_path: + log("ERROR: at least one of --input-path or --state-path is required") + return ExitStatus.ERROR + + if opts.api_key.startswith("sgu_"): + log( + "WARNING: sgu_ tokens are non-functional for SSO-group-only users and inherit only " + "direct permissions for hybrid SSO users. Prefer an organization (sgo_) token." + ) + + try: + opts.trigger_details = _load_trigger_details(opts) + result = run_check(opts) + except CheckError as e: + # Fails closed: a run that produced no verdict must never look like a pass, whatever + # --fail-on-error says. + log(f"ERROR: {e}") + return ExitStatus.ERROR + except KeyboardInterrupt: + log("Interrupted") + return ExitStatus.ERROR_CTRL_C + + verdict = result["verdict"] + if verdict == "errored": + # Fails closed regardless of --fail-on-error: the flag governs policy verdicts, not tool + # health, and a run that produced no verdict must never look like a pass. + log("The run did not produce a verdict") + return ExitStatus.ERROR + if verdict in ("failed", "approval-required") and opts.fail_on_error: + return ExitStatus.ERROR_POLICY_FAILED + if verdict == "failed": + log("Policies failed, but --fail-on-error was not set") + if verdict == "approval-required": + log("The run is waiting for approval; --fail-on-error was not set") + + return ExitStatus.SUCCESS diff --git a/src/tirith/platform/client.py b/src/tirith/platform/client.py new file mode 100644 index 0000000..6996d53 --- /dev/null +++ b/src/tirith/platform/client.py @@ -0,0 +1,319 @@ +""" +StackGuardian API client. + +stdlib only -- urllib rather than requests -- so this adds no dependency to a package that has +three, and a CI runner needs nothing installed beyond tirith itself. + + POST /orgs//wfgrps/ create the workflow group + POST /orgs//wfgrps//wfs/ create the workflow + GET /orgs//wfgrps//wfs//configuration_upload_url/ presigned PUT (5 min) + key + POST /orgs//wfgrps//wfs//wfruns/ create the run + GET /orgs//wfgrps//wfs//wfruns// poll + GET /orgs//wfgrps//wfs//artifacts// fetch the results artifact + GET .../wfruns//wfrunfacts// fallback -> PolicyEvalResults +""" + +import gzip +import json +import time +import urllib.error +import urllib.parse +import urllib.request + +DEFAULT_API_URL = "https://api.app.stackguardian.io/api/v1" + +# Terminal run states. QUEUED/PENDING/RUNNING are transient; a run can sit in QUEUED for a long +# while behind the per-workflow concurrency gate, which is why the caller logs each poll. +# +# APPROVAL_REQUIRED is terminal *for polling purposes*: it is a resting state, reached when a +# policy's onFail is APPROVAL_REQUIRED, and nothing further happens without a human. Treating it as +# transient would spin until the timeout and then report a tool failure for what is actually a +# completed evaluation. sg-cli treats it the same way. +TERMINAL_STATUSES = ("COMPLETED", "ERRORED", "CANCELLED", "APPROVAL_REQUIRED") + +RETRYABLE_STATUS = (408, 429, 500, 502, 503, 504) + + +class SGError(Exception): + """An API call failed in a way the caller cannot recover from.""" + + +def _extract_signed_url(payload): + """ + Pull the presigned URL out of an upload-url response. + + The shape varies by endpoint and deployment: the tfstate/file upload endpoints return the URL + as a bare string in `msg`, while the newer template-artifact endpoints nest it under + `data.signedUrl`. Accept either rather than depending on one. + """ + if not isinstance(payload, dict): + return None + + for container_key in ("data", "msg"): + container = payload.get(container_key) + if isinstance(container, str) and container.startswith("http"): + return container + if isinstance(container, dict): + for url_key in ("signedUrl", "signed_url", "url"): + candidate = container.get(url_key) + if isinstance(candidate, str) and candidate.startswith("http"): + return candidate + return None + + +class SGClient: + def __init__(self, api_url, org, api_key, user_agent="tirith-action", timeout=60): + self.api_url = (api_url or DEFAULT_API_URL).rstrip("/") + self.org = org + self.api_key = api_key + self.user_agent = user_agent + self.timeout = timeout + + # -- plumbing ------------------------------------------------------------------------------ + + def _request(self, method, path, body=None, retries=4): + url = f"{self.api_url}/orgs/{urllib.parse.quote(self.org)}{path}" + data = json.dumps(body).encode() if body is not None else None + + last_error = None + for attempt in range(retries + 1): + request = urllib.request.Request(url, data=data, method=method) + # SG's documented scheme. Must be an sgo_ (org) token: sgu_ tokens are non-functional + # for SSO-group-only users and inherit only direct permissions for hybrid SSO users, + # which surfaces as a confusing 403. + request.add_header("Authorization", f"apikey {self.api_key}") + request.add_header("Content-Type", "application/json") + request.add_header("X-SG-Client", self.user_agent) + + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + raw = response.read() + return response.status, (json.loads(raw) if raw else {}) + except urllib.error.HTTPError as e: + raw = e.read() + try: + payload = json.loads(raw) if raw else {} + except json.JSONDecodeError: + payload = {"msg": raw.decode("utf-8", "replace")[:500]} + + if e.code in RETRYABLE_STATUS and attempt < retries: + last_error = f"HTTP {e.code}: {payload.get('msg', '')}" + time.sleep(min(2**attempt, 8)) + continue + return e.code, payload + except (urllib.error.URLError, TimeoutError) as e: + # Never treat a network failure as a pass -- the caller maps this to a red check. + last_error = str(e) + if attempt < retries: + time.sleep(min(2**attempt, 8)) + continue + raise SGError(f"Could not reach StackGuardian at {self.api_url}: {last_error}") + + raise SGError(f"StackGuardian request failed after {retries + 1} attempts: {last_error}") + + # -- resources ----------------------------------------------------------------------------- + + def ensure_workflow_group(self, name): + """ + Create the workflow group if absent. + + Needed because `createIfNotExists` on run creation auto-creates the *workflow*, not the + group -- core's own error for a missing group reads "Workflow Group does not exist and + cannot be created". A 409 means someone else already made it, which is success here. + """ + status, payload = self._request( + "POST", + "/wfgrps/", + {"ResourceName": name, "Description": "Created by tirith", "Tags": ["sg-created"]}, + ) + if status in (200, 201, 409): + return status + raise SGError(f"Could not create workflow group '{name}' (HTTP {status}): {payload.get('msg')}") + + def ensure_workflow(self, wfgrp, workflow_id, description, terraform_config): + """ + Create the workflow if absent, keyed on `Id`. + + `Id` is the stable slug identity and what goes in the URL; `ResourceName` is a display name + and is not unique. Both are set to the same string so there is one name to reason about. + Note `Id` is a DRF SlugField, so it cannot contain dots. + + The workflow is `TERRAFORM`, not `CUSTOM`. For a terraform workflow core synthesises the + steps from the stored TerraformConfig plus the per-run TerraformAction and *ignores* any + WfStepsConfig in the request -- so the step configuration has to live here, once, rather + than being sent on every run. It also means the run renders as a real terraform run in the + dashboard rather than as opaque custom steps. + """ + status, payload = self._request( + "POST", + f"/wfgrps/{urllib.parse.quote(wfgrp)}/wfs/", + { + "Id": workflow_id, + "ResourceName": workflow_id, + "Description": description, + "Tags": ["sg-created", "tirith"], + "WfType": "TERRAFORM", + "TerraformConfig": terraform_config, + }, + ) + if status in (200, 201, 409): + return status + raise SGError(f"Could not create workflow '{workflow_id}' (HTTP {status}): {payload.get('msg')}") + + def upload_archive(self, wfgrp, workflow_id, filename, folder, archive_bytes): + """ + Upload the project archive via a presigned PUT, returning its storage key. + + The key is what the caller passes back as `terraformProjectZip` when creating the run. It + comes from the response rather than being rebuilt here: the layout is runner-aware (a + private runner's own S3 bucket or Azure container rather than the shared bucket), so a + client-side guess would be wrong for exactly the customers who are hardest to debug. + + `folder` must be a flat token -- the endpoint rejects `/`, `\\` and `..` to prevent path + traversal. + """ + query = urllib.parse.urlencode({"filename": filename, "folder": folder}) + status, payload = self._request( + "GET", f"/wfgrps/{urllib.parse.quote(wfgrp)}/wfs/{workflow_id}/configuration_upload_url/?{query}" + ) + if status != 200: + raise SGError(f"Could not get an upload URL for {filename} (HTTP {status}): {payload.get('msg')}") + + msg = payload.get("msg") + if not isinstance(msg, dict) or not msg.get("key"): + raise SGError( + f"The upload response for {filename} carried no storage key. The platform may " + f"predate the configuration_upload_url endpoint. Response: {payload}" + ) + signed_url = _extract_signed_url({"msg": msg.get("signedUrl")}) + if not signed_url: + raise SGError(f"No signed URL in the upload response for {filename}: {payload}") + + # Must match the content type the URL was signed with, or S3 rejects it as a signature + # mismatch. + put = urllib.request.Request(signed_url, data=archive_bytes, method="PUT") + put.add_header("Content-Type", "application/gzip") + try: + with urllib.request.urlopen(put, timeout=self.timeout) as response: + if response.status not in (200, 204): + raise SGError(f"Upload of {filename} returned HTTP {response.status}") + except urllib.error.HTTPError as e: + # The signed URL is valid for 5 minutes; an expiry shows up here as a 403. + raise SGError(f"Upload of {filename} failed (HTTP {e.code}): {e.read()[:300]!r}") + except (urllib.error.URLError, TimeoutError) as e: + raise SGError(f"Upload of {filename} failed: {e}") + + return msg["key"] + + def create_run(self, wfgrp, workflow_id, project_zip_key, trigger_details, action="policy-only"): + """ + Create one workflow run. Every invocation makes a new run. + + Deliberately carries no WfStepsConfig: core ignores it for TERRAFORM workflows and + synthesises the steps from the workflow's TerraformConfig and this TerraformAction. The + only per-run state is the archive key and where the run came from. + """ + body = { + "TerraformAction": {"action": action}, + "terraformProjectZip": project_zip_key, + "TriggerDetails": trigger_details, + } + status, payload = self._request("POST", f"/wfgrps/{urllib.parse.quote(wfgrp)}/wfs/{workflow_id}/wfruns/", body) + if status not in (200, 201): + raise SGError(f"Could not create the workflow run (HTTP {status}): {payload.get('msg')}") + + data = payload.get("data") or {} + run_name = data.get("ResourceName") + if not run_name: + raise SGError(f"No ResourceName in the run-creation response: {payload}") + return run_name, data + + def get_run(self, wfgrp, workflow_id, run_id): + status, payload = self._request( + "GET", f"/wfgrps/{urllib.parse.quote(wfgrp)}/wfs/{workflow_id}/wfruns/{run_id}/" + ) + if status != 200: + raise SGError(f"Could not read run {run_id} (HTTP {status}): {payload.get('msg')}") + # This endpoint returns the run object under "msg" rather than "data". + return payload.get("msg") or payload.get("data") or {} + + def wait_for_run(self, wfgrp, workflow_id, run_id, timeout=1800, interval=10, on_poll=None): + """ + Poll until the run reaches a terminal state. + + A timeout is a failure, never a pass: the caller maps it to a red check. `on_poll` exists + so the caller can log each status -- a run stuck in QUEUED behind another run on the same + workflow looks identical to a hung run otherwise. + """ + deadline = time.time() + timeout + last_status = None + + while time.time() < deadline: + run = self.get_run(wfgrp, workflow_id, run_id) + status = run.get("LatestStatus") + if status != last_status and on_poll: + on_poll(status) + last_status = status + + if status in TERMINAL_STATUSES: + return status, run + time.sleep(interval) + + raise SGError( + f"Run {run_id} did not finish within {timeout}s (last status: {last_status}). " + f"Runs on one workflow serialize, so it may be queued behind another run." + ) + + def get_results_artifact(self, wfgrp, workflow_id, artifact_path): + """ + Read the results artifact the tirith step publishes next to the inputs. + + This is the primary source. The run controller no longer creates a WorkflowRunFacts + record -- it forwards the facts to the report-aggregator lambda and leaves only a pointer + on the workflow object -- so the wfrunfacts endpoint answers "does not exist" for runs it + did produce results for. The artifact is written by our own step, so it is a contract we + control end to end. + """ + status, payload = self._request( + "GET", + f"/wfgrps/{urllib.parse.quote(wfgrp)}/wfs/{workflow_id}/artifacts/{artifact_path}/", + ) + if status != 200: + return None + + # This endpoint returns the artifact body directly rather than an envelope. + if isinstance(payload, dict) and "PolicyEvalResults" in payload: + return payload.get("PolicyEvalResults") or {} + return None + + def get_policy_results(self, wfgrp, workflow_id, run_id): + """ + Fetch PolicyEvalResults from the run fact. + + Retained as a fallback for deployments where the run controller still writes the record. + The endpoint hands back a presigned GET rather than the payload inline, because the facts + document embeds the whole plan and can be large. + """ + status, payload = self._request( + "GET", + f"/wfgrps/{urllib.parse.quote(wfgrp)}/wfs/{workflow_id}/wfruns/{run_id}/wfrunfacts/default/", + ) + if status != 200: + return {} + + body = payload.get("msg") or payload.get("data") or {} + if isinstance(body, dict) and body.get("PolicyEvalResults"): + return body["PolicyEvalResults"] + + signed_url = body.get("signedUrl") if isinstance(body, dict) else None + if not signed_url: + return {} + + try: + with urllib.request.urlopen(signed_url, timeout=self.timeout) as response: + raw = response.read() + if response.info().get("Content-Encoding") == "gzip" or raw[:2] == b"\x1f\x8b": + raw = gzip.decompress(raw) + return (json.loads(raw) or {}).get("PolicyEvalResults") or {} + except Exception: + return {} diff --git a/src/tirith/platform/redact.py b/src/tirith/platform/redact.py new file mode 100644 index 0000000..53f4f3a --- /dev/null +++ b/src/tirith/platform/redact.py @@ -0,0 +1,383 @@ +""" +Slim and mask terraform documents before they leave the runner. + +This runs client-side on purpose. Once bytes reach StackGuardian the exposure has already +happened, so masking on the server would be theatre. Everything here is a pure function over +parsed JSON so it can be tested exhaustively. + +A caveat worth stating plainly, and repeated in the README: terraform's `*_sensitive` markers are +NOT exhaustive. A value that flows through `locals`, or comes from a provider that did not mark +its schema, arrives marked `false` and will not be masked by marker-driven redaction. Slimming and +the `variables` drop below exist partly to limit that blast radius. +""" + +import copy + +SENTINEL = "__SG_REDACTED__" + +# Top-level plan sections tirith's terraform_plan provider never reads, verified against +# providers/terraform_plan/handler.py: +# +# resource_changes -> attribute / action / count operations +# configuration -> direct_dependencies, direct_references, provider_config (KEPT) +# terraform_version -> terraform_version operation +# +# `planned_values` is the dangerous one. It mirrors every resource's values in a second place and +# carries NO sensitivity markers of its own, so marker-driven redaction of `resource_changes` +# leaves the same secret in plaintext here. Dropping it is lossless for evaluation and closes that +# hole; a real plan leaked a `local_sensitive_file` body through exactly this path. +SLIM_DROP_KEYS = ("prior_state", "planned_values") + +# Provider blocks whose `expressions` can hold hardcoded credentials. `configuration` cannot be +# dropped wholesale -- three tirith operations read it -- so the credential-bearing part is +# scrubbed instead, keeping the two fields provider_config_operator actually consults. +_PROVIDER_CONFIG_KEEP = ("name", "full_name", "version_constraint", "module_address", "alias") + + +def slim_plan(plan): + """ + Drop plan sections that are irrelevant to evaluation. + + Typically removes 60-90% of the bytes. `configuration` is deliberately retained but scrubbed + (see `_scrub_configuration`), because dropping it would silently break the + `direct_dependencies`, `direct_references` and `provider_config` operations -- policies would + stop finding what they are looking for rather than failing loudly. + """ + if not isinstance(plan, dict): + return plan + + slimmed = {k: v for k, v in plan.items() if k not in SLIM_DROP_KEYS} + if isinstance(slimmed.get("configuration"), dict): + slimmed["configuration"] = _scrub_configuration(slimmed["configuration"]) + return slimmed + + +def _scrub_configuration(configuration): + """ + Strip credential-bearing expressions from `configuration` while keeping what tirith reads. + + Two places hold literals, and both have to be scrubbed: + + `provider_config[].expressions` -- `provider_config_operator` reads only `version_constraint` + and `expressions.region.constant_value`, so access keys, tokens and assume-role blocks can go. + + `root_module.resources[].expressions[].constant_value` -- every literal written in the HCL, + including a hardcoded password. This is a third instance of the `planned_values` pattern: a + place values live that carries no sensitivity markers, so marker-driven masking of + `resource_changes` never touches it. Caught in QA -- a `local_sensitive_file` body was masked + in `resource_changes` and sat in plaintext here in the same document. + + Dropping `constant_value` is lossless: `direct_references_operator` reads only `references` + from these expressions, and `direct_dependencies_operator` reads only `depends_on` + (providers/terraform_plan/handler.py:329, :385-388). + """ + scrubbed = dict(configuration) + + provider_config = scrubbed.get("provider_config") + if isinstance(provider_config, dict): + cleaned = {} + for name, block in provider_config.items(): + if not isinstance(block, dict): + cleaned[name] = block + continue + kept = {k: v for k, v in block.items() if k in _PROVIDER_CONFIG_KEEP} + region = (block.get("expressions") or {}).get("region") + if region is not None: + kept["expressions"] = {"region": region} + cleaned[name] = kept + scrubbed["provider_config"] = cleaned + + root_module = scrubbed.get("root_module") + if isinstance(root_module, dict): + scrubbed["root_module"] = _scrub_config_module(root_module) + + return scrubbed + + +def _scrub_config_module(module): + """Recursively drop literal values from a configuration module, keeping the reference graph.""" + scrubbed = dict(module) + + resources = scrubbed.get("resources") + if isinstance(resources, list): + scrubbed["resources"] = [_scrub_config_resource(r) for r in resources] + + # Child modules nest the same shape under module_calls[].module. + module_calls = scrubbed.get("module_calls") + if isinstance(module_calls, dict): + calls = {} + for name, call in module_calls.items(): + if isinstance(call, dict) and isinstance(call.get("module"), dict): + call = {**call, "module": _scrub_config_module(call["module"])} + # A module's own arguments are literals too. + call.pop("expressions", None) + calls[name] = call + scrubbed["module_calls"] = calls + + # Variable defaults and output values are literals with no operation reading them. + for section in ("variables", "outputs"): + if isinstance(scrubbed.get(section), dict): + scrubbed[section] = _scrub_config_section(scrubbed[section]) + + return scrubbed + + +def _scrub_config_resource(resource): + if not isinstance(resource, dict): + return resource + + expressions = resource.get("expressions") + if not isinstance(expressions, dict): + return resource + + return {**resource, "expressions": {k: _keep_references(v) for k, v in expressions.items()}} + + +def _keep_references(expression): + """ + Reduce one expression to just its `references`, dropping every literal. + + Terraform nests expressions arbitrarily: a block argument is a dict of expressions, and a + repeated block is a list of them, so this recurses rather than looking one level deep. + """ + if isinstance(expression, list): + return [_keep_references(item) for item in expression] + if not isinstance(expression, dict): + return expression + if "references" in expression or "constant_value" in expression: + # A leaf: keep only the reference graph. + return {"references": expression["references"]} if "references" in expression else {} + return {k: _keep_references(v) for k, v in expression.items()} + + +def _scrub_config_section(section): + """Drop `default` / `expression` literals from variables and outputs.""" + cleaned = {} + for name, entry in section.items(): + if isinstance(entry, dict): + entry = {k: v for k, v in entry.items() if k not in ("default", "expression", "value")} + cleaned[name] = entry + return cleaned + + +def _mask_by_marker(value, marker): + """ + Walk `value` alongside terraform's parallel sensitivity structure `marker`. + + A marker node of `true` masks the whole subtree beneath it. Dicts and lists are walked in + lockstep; anything else is returned untouched. + """ + if marker is True: + return SENTINEL + + if isinstance(marker, dict) and isinstance(value, dict): + return {k: _mask_by_marker(v, marker.get(k)) for k, v in value.items()} + + if isinstance(marker, list) and isinstance(value, list): + # Terraform emits a marker list positionally aligned with the value list. A shorter + # marker list means the tail is not sensitive. + return [_mask_by_marker(item, marker[i] if i < len(marker) else None) for i, item in enumerate(value)] + + return value + + +def redact_plan(plan): + """ + Slim, then mask every value terraform flagged sensitive, then drop root `variables`. + + `variables` goes wholesale because the plan does not reliably mark which root variables were + declared `sensitive = true` -- so the only safe assumption is that all of them might be. + """ + plan = slim_plan(plan) + if not isinstance(plan, dict): + return plan + + redacted = dict(plan) + redacted.pop("variables", None) + + resource_changes = redacted.get("resource_changes") + if isinstance(resource_changes, list): + masked_changes = [] + for resource_change in resource_changes: + if not isinstance(resource_change, dict): + masked_changes.append(resource_change) + continue + + masked = dict(resource_change) + change = masked.get("change") + if isinstance(change, dict): + masked_change = dict(change) + for value_key, marker_key in (("before", "before_sensitive"), ("after", "after_sensitive")): + if value_key in masked_change: + masked_change[value_key] = _mask_by_marker( + masked_change[value_key], masked_change.get(marker_key) + ) + masked["change"] = masked_change + masked_changes.append(masked) + redacted["resource_changes"] = masked_changes + + output_changes = redacted.get("output_changes") + if isinstance(output_changes, dict): + redacted["output_changes"] = {name: _redact_output_change(change) for name, change in output_changes.items()} + + return redacted + + +def _redact_output_change(change): + """ + Mask a sensitive output's before/after values. + + Terraform spells the marker differently across versions: older plans carry a single + `sensitive`, newer ones carry `before_sensitive` / `after_sensitive` per side. Checking only + `sensitive` silently missed every modern plan, so all three are honoured -- and each side is + masked independently, since an output can become sensitive without having been so before. + + Only keys that are actually present are replaced. Adding an `after` to a create whose value is + still unknown (`after_unknown: true`) would invent data the plan never contained. + """ + if not isinstance(change, dict): + return change + + masked = dict(change) + whole = bool(change.get("sensitive")) + + for side in ("before", "after"): + if side not in masked: + continue + if whole or change.get(f"{side}_sensitive") is True: + masked[side] = SENTINEL + + return masked + + +def redact_state(state): + """ + Mask a terraform state document. + + State is more dangerous than a plan: it holds every resource attribute in plaintext, including + values no plan would surface. Two rules, matching what the platform's terraform step applies: + + - `outputs[k].sensitive` is true -> replace that output's value + - each key named in an instance's `sensitive_attributes` -> replace that attribute + + Expects the raw state shape (top-level `resources` / `outputs`), not `terraform show -json` + output, which nests resources under `values.root_module.resources`. + """ + if not isinstance(state, dict): + return state + + redacted = dict(state) + + outputs = redacted.get("outputs") + if isinstance(outputs, dict): + masked_outputs = {} + for name, output in outputs.items(): + if isinstance(output, dict) and output.get("sensitive"): + masked_outputs[name] = {**output, "value": SENTINEL} + else: + masked_outputs[name] = output + redacted["outputs"] = masked_outputs + + resources = redacted.get("resources") + if isinstance(resources, list): + redacted["resources"] = [_redact_state_resource(r) for r in resources] + + return redacted + + +def _redact_state_resource(resource): + if not isinstance(resource, dict): + return resource + + instances = resource.get("instances") + if not isinstance(instances, list): + return resource + + masked_instances = [] + for instance in instances: + if not isinstance(instance, dict): + masked_instances.append(instance) + continue + + masked = dict(instance) + attributes = masked.get("attributes") + sensitive_attributes = masked.get("sensitive_attributes") or [] + + if isinstance(attributes, dict) and sensitive_attributes: + masked_attributes = copy.deepcopy(attributes) + for sensitive_attribute in sensitive_attributes: + _mask_attribute_path(masked_attributes, _attribute_steps(sensitive_attribute)) + masked["attributes"] = masked_attributes + + masked_instances.append(masked) + + return {**resource, "instances": masked_instances} + + +def _attribute_steps(sensitive_attribute): + """ + Normalise one `sensitive_attributes` entry into a list of path steps. + + Terraform writes each entry as a PATH -- a list of steps -- not a single key: + + [[{"type": "get_attr", "value": "content_base64"}], + [{"type": "get_attr", "value": "content"}]] + + Reading only the flat forms silently masked nothing at all on real state, because a list is + neither a dict nor a string. Verified against `terraform state pull` output for a + `local_sensitive_file`; the earlier unit tests passed only because their fixture invented the + flat shape. + + The two flat forms are still accepted: some providers and older state versions emit them. + """ + if isinstance(sensitive_attribute, list): + entries = sensitive_attribute + else: + entries = [sensitive_attribute] + + steps = [] + for entry in entries: + if isinstance(entry, dict): + steps.append(entry.get("value")) + elif isinstance(entry, (str, int)): + steps.append(entry) + else: + # An unrecognised step means the path cannot be trusted; masking a guessed location + # would be worse than reporting nothing. + return [] + return steps + + +def _mask_attribute_path(container, steps): + """ + Replace the value at `steps` within `container` with the sentinel. + + A path may descend through nested objects and list indices -- `[{"get_attr": "config"}, + {"index": 0}, {"get_attr": "token"}]` -- so this walks rather than assuming one level. + """ + if not steps: + return + + *parents, leaf = steps + node = container + for step in parents: + if isinstance(node, dict) and step in node: + node = node[step] + elif isinstance(node, list) and isinstance(step, int) and 0 <= step < len(node): + node = node[step] + else: + return + + if isinstance(node, dict) and leaf in node: + node[leaf] = SENTINEL + elif isinstance(node, list) and isinstance(leaf, int) and 0 <= leaf < len(node): + node[leaf] = SENTINEL + + +def count_redactions(document): + """Count sentinel occurrences, for the attestation the action sends with the upload.""" + if isinstance(document, dict): + return sum(count_redactions(v) for v in document.values()) + if isinstance(document, list): + return sum(count_redactions(v) for v in document) + return 1 if document == SENTINEL else 0 diff --git a/src/tirith/platform/report.py b/src/tirith/platform/report.py new file mode 100644 index 0000000..6639549 --- /dev/null +++ b/src/tirith/platform/report.py @@ -0,0 +1,240 @@ +""" +Turn PolicyEvalResults into a PR comment body, a check-run summary, and a verdict. + +Pure functions over the results document so the layout and the truncation arithmetic can be tested +without touching a network. +""" + +FAIL = "FAIL" +WARN = "WARN" +PASS = "PASS" +APPROVAL_REQUIRED = "APPROVAL_REQUIRED" + +# GitHub rejects an issue-comment body over 65536 characters and a check-run output.summary over +# 65535. Budget well under both: the count that matters is characters after rendering, and a +# 422 at the end of a run is a bad way to find out. +COMMENT_LIMIT = 60000 + +_ICONS = {FAIL: "❌", WARN: "⚠️", APPROVAL_REQUIRED: "⏳", PASS: "✅"} + + +def summarize(policy_results): + """ + Collapse the results into counts plus a flat finding list. + + A rule marked `skip` carries no verdict, so it is counted separately rather than being + folded into passes -- reporting a skipped control as passing is the kind of quiet + inaccuracy this whole design exists to avoid. + """ + counts = {FAIL: 0, WARN: 0, APPROVAL_REQUIRED: 0, PASS: 0, "SKIPPED": 0} + findings = [] + + for policy_id, rules in sorted((policy_results or {}).items()): + for rule in rules or []: + if rule.get("skip"): + counts["SKIPPED"] += 1 + findings.append( + { + "policy_id": policy_id, + "rule_name": rule.get("rule_name", ""), + "result": "SKIPPED", + "messages": [], + "resources": [], + } + ) + continue + + result = rule.get("result", PASS) + counts[result] = counts.get(result, 0) + 1 + messages, resources = _extract_detail(rule) + findings.append( + { + "policy_id": policy_id, + "rule_name": rule.get("rule_name", ""), + "result": result, + "messages": messages, + "resources": resources, + } + ) + + return counts, findings + + +def _extract_detail(rule): + """Pull human-readable messages and resource addresses out of a rule's evaluations.""" + messages = [] + resources = [] + + for entry in (rule.get("evaluations") or {}).get("fails") or []: + if "exec_err" in entry: + # An engine/config problem rather than a policy violation -- surfaced verbatim so a + # malformed policy is not mistaken for a real finding. + messages.append(f"engine: {entry['exec_err']}") + continue + + for evaluation in entry.get("result") or []: + message = evaluation.get("message") + if message: + messages.append(message) + # Only the terraform_plan provider populates meta; others set it to None. + meta = evaluation.get("meta") or {} + address = meta.get("address") if isinstance(meta, dict) else None + if address and address not in resources: + resources.append(address) + + return messages, resources + + +def verdict(counts, run_status): + """ + Reduce counts and run status to one word. + + failed | warned | passed | no-policies | approval-required | errored + + `errored` covers a run that never produced a verdict -- an ERRORED/CANCELLED run, or results + that came back empty. It is deliberately distinct from `failed` so the caller can tell "a + policy said no" from "we do not know", and never conflate either with a pass. + + `approval-required` is a resting state, not a failure: the evaluation finished and a human now + has to act. Reporting it as `errored` would blame the tool for a working evaluation. + + It is reached two ways, and both matter. The run status is APPROVAL_REQUIRED when the platform + itself gated the run. A *rule* result of APPROVAL_REQUIRED means a policy author wrote + `onFail: APPROVAL_REQUIRED`, which the policy-only step records without pausing the run -- so + the run comes back COMPLETED and only the counts carry the intent. + + Folding that into `warned` was wrong: `warned` maps to a `neutral` check, which SATISFIES a + required status check, so a policy demanding human sign-off silently did not block. Ranking it + above `warned` keeps the author's intent without implementing the approval workflow, which is + out of scope here. + """ + if run_status == "APPROVAL_REQUIRED": + return "approval-required" + if run_status not in ("COMPLETED",): + return "errored" + if counts.get(FAIL): + return "failed" + if counts.get(APPROVAL_REQUIRED): + return "approval-required" + if counts.get(WARN): + return "warned" + if counts.get(PASS) or counts.get("SKIPPED"): + return "passed" + # A COMPLETED run with no policy results at all: nothing was in scope. Report it rather than + # implying a clean bill of health. + return "no-policies" + + +def headline(counts, verdict_value): + if verdict_value == "errored": + return "Tirith could not evaluate policies" + if verdict_value == "no-policies": + return "Tirith — no policies in scope for this workflow" + + parts = [] + for key, label in ((FAIL, "failed"), (APPROVAL_REQUIRED, "need approval"), (WARN, "warned")): + if counts.get(key): + parts.append(f"{counts[key]} {label}") + if counts.get(PASS): + parts.append(f"{counts[PASS]} passed") + if counts.get("SKIPPED"): + parts.append(f"{counts['SKIPPED']} skipped") + return "Tirith — " + (", ".join(parts) if parts else "nothing evaluated") + + +def render_markdown(policy_results, run_status, run_url, marker=None, limit=COMMENT_LIMIT): + """ + Render the results as markdown, truncating detail before the summary table. + + `marker` is an opaque first line the caller can use to find this document again -- GitHub's + sticky-comment marker, for instance. Kept as a parameter rather than built here so this module + stays VCS-agnostic. + """ + counts, findings = summarize(policy_results) + verdict_value = verdict(counts, run_status) + + header = ([marker, ""] if marker else []) + [ + f"## 🛡️ {headline(counts, verdict_value)}", + "", + ] + + if verdict_value == "errored": + header += [ + f"The workflow run finished as `{run_status}` without producing policy results.", + "This is reported as a failure rather than a pass: no verdict is not the same as a clean one.", + "", + ] + + table = _render_table(findings) + footer = _render_footer(counts, run_url) + + detail_sections = [_render_detail(f) for f in findings if f["result"] in (FAIL, APPROVAL_REQUIRED, WARN)] + + body = "\n".join(header + table + detail_sections + footer) + if len(body) <= limit: + return body + + # Drop detail sections from the end until it fits, keeping the summary table intact -- the + # table is the part a reviewer scans first. + kept = list(detail_sections) + while kept and len(body) > limit: + kept.pop() + omitted = len(detail_sections) - len(kept) + note = [f"", f"_… and {omitted} more finding(s). See the full run in StackGuardian._", ""] + body = "\n".join(header + table + kept + note + footer) + + if len(body) > limit: + # Even the table is too large; truncate hard rather than risk a 422. + body = body[: limit - 200] + "\n\n_… truncated. See the full run in StackGuardian._\n" + + return body + + +def _render_table(findings): + if not findings: + return [] + rows = [ + "| | Policy | Rule | Resource |", + "|---|---|---|---|", + ] + for finding in findings: + icon = _ICONS.get(finding["result"], "⚪") + resources = ", ".join(f"`{r}`" for r in finding["resources"][:3]) or "—" + if len(finding["resources"]) > 3: + resources += f" _+{len(finding['resources']) - 3}_" + rows.append(f"| {icon} | `{finding['policy_id']}` | {finding['rule_name']} | {resources} |") + rows.append("") + return rows + + +def _render_detail(finding): + icon = _ICONS.get(finding["result"], "⚪") + lines = [ + "
", + f"{icon} {finding['policy_id']} › {finding['rule_name']}", + "", + ] + for message in finding["messages"][:20]: + lines.append(f"- {message}") + if len(finding["messages"]) > 20: + lines.append(f"- _… and {len(finding['messages']) - 20} more_") + if finding["resources"]: + lines += ["", "Resources:"] + [f"- `{r}`" for r in finding["resources"][:20]] + lines += ["", "
", ""] + return "\n".join(lines) + + +def _render_footer(counts, run_url): + bits = [] + if counts.get(PASS): + bits.append(f"✅ {counts[PASS]} passed") + if counts.get("SKIPPED"): + bits.append(f"⚪ {counts['SKIPPED']} skipped") + if run_url: + bits.append(f'View run in StackGuardian') + return ["", f"{' · '.join(bits)}"] if bits else [] + + +def strip_marker(body): + """Drop the marker line, for a rendering target that has no use for it.""" + return "\n".join(line for line in body.split("\n") if not line.startswith("[//]: <>")) diff --git a/src/tirith/prettyprinter.py b/src/tirith/prettyprinter.py index 4134ba7..599f410 100644 --- a/src/tirith/prettyprinter.py +++ b/src/tirith/prettyprinter.py @@ -97,7 +97,7 @@ def pretty_print_result_dict(final_result_dict: Dict) -> None: print(f" {TermStyle.fail('FAILED')}") num_failed_checks += 1 - for result_num, result_dict in enumerate(check_dict["result"]): + for result_num, result_dict in enumerate(check_dict.get("result", [])): result_message = result_dict["message"] if result_dict["passed"]: print(TermStyle.green(f" {result_num+1}. PASSED: {result_message}")) diff --git a/src/tirith/status.py b/src/tirith/status.py index d7ee321..b690243 100644 --- a/src/tirith/status.py +++ b/src/tirith/status.py @@ -9,6 +9,11 @@ class ExitStatus(IntEnum): ERROR = 1 ERROR_TIMEOUT = 2 + # A policy said no, under `platform check --fail-on-error`. Distinct from ERROR so a caller can + # tell "your infrastructure violates a policy" from "tirith could not reach the platform" -- + # the same distinction --fail-on-error exists to draw, one level up. + ERROR_POLICY_FAILED = 3 + # # 128+2 SIGINT ERROR_CTRL_C = 130 diff --git a/tests/cli/test_dispatch.py b/tests/cli/test_dispatch.py new file mode 100644 index 0000000..8314411 --- /dev/null +++ b/tests/cli/test_dispatch.py @@ -0,0 +1,87 @@ +""" +Tests for subcommand dispatch. + +The local-evaluation surface is a contract: the platform and the workflow-step templates parse its +--json output, and tests/core/test_output_compatibility.py asserts that output byte-for-byte. +Adding `tirith platform` must leave it completely untouched, including its single-dash long +options, which argparse cannot express alongside a subparser. +""" + +import json +import os + +import pytest + +from tirith import cli +from tirith.status import ExitStatus + +FIXTURES = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "providers", "json") +POLICY = os.path.join(FIXTURES, "policy.json") +INPUT = os.path.join(FIXTURES, "input.json") + + +def test_legacy_invocation_still_works(capsys): + """The flat parser must keep working exactly as before, driven through main(args=...).""" + status = cli.main(["-policy-path", POLICY, "-input-path", INPUT, "--json"]) + + assert status == ExitStatus.SUCCESS + document = json.loads(capsys.readouterr().out) + assert "final_result" in document + assert "evaluators" in document + + +def test_main_honours_its_args_parameter(capsys): + """ + It did not before: parse_args() was called with no argument, so main(args=...) was ignored and + the CLI always read sys.argv. That made it untestable and undrivable from another program. + """ + status = cli.main(["-policy-path", POLICY, "-input-path", INPUT, "--json"]) + + assert status == ExitStatus.SUCCESS + assert capsys.readouterr().out.strip().startswith("{") + + +def test_no_arguments_prints_help(capsys): + """ + Pre-existing behaviour, asserted so the dispatcher does not change it: the sys.exit(0) is + caught by main's own SystemExit handler, which returns None for a zero code. __main__ treats + that as success. + """ + status = cli.main([]) + + assert not status + assert "usage" in capsys.readouterr().out.lower() + + +def test_platform_is_dispatched_to_the_subcommand(capsys): + """`platform` with no subcommand prints the platform help, not the local-evaluation help.""" + status = cli.main(["platform"]) + + assert status == ExitStatus.SUCCESS + assert "tirith platform" in capsys.readouterr().out + + +def test_platform_check_requires_credentials(capsys, monkeypatch): + monkeypatch.delenv("SG_API_TOKEN", raising=False) + monkeypatch.delenv("SG_ORG", raising=False) + + status = cli.main(["platform", "check", "--workflow-id", "wf", "--input-path", INPUT]) + + assert status == ExitStatus.ERROR + assert "--api-key" in capsys.readouterr().err + + +def test_platform_check_requires_a_document(capsys, monkeypatch): + monkeypatch.setenv("SG_API_TOKEN", "sgo_x") + monkeypatch.setenv("SG_ORG", "acme") + + status = cli.main(["platform", "check", "--workflow-id", "wf"]) + + assert status == ExitStatus.ERROR + assert "--input-path" in capsys.readouterr().err + + +def test_a_bare_word_is_not_mistaken_for_a_subcommand(capsys): + """Only names in SUBCOMMANDS dispatch; anything else goes to the flat parser.""" + assert "platform" in cli.SUBCOMMANDS + assert "check" not in cli.SUBCOMMANDS diff --git a/tests/core/test_core.py b/tests/core/test_core.py index 3afdc41..ec09ea3 100644 --- a/tests/core/test_core.py +++ b/tests/core/test_core.py @@ -151,3 +151,67 @@ def test_generate_evaluator_result_multiple_resources_one_failing(): assert len(result["result"]) == 2 assert result["result"][0]["passed"] is True assert result["result"][1]["passed"] is False + + +@mark.passing +def test_generate_evaluator_result_unsupported_evaluator_populates_result(): + """ + An unsupported condition.type must still produce a "result" list. Consumers index into + it unconditionally, so an early return without it used to raise KeyError far from the cause. + """ + evaluator_obj = { + "id": "test_evaluator", + "provider_args": {"operation_type": "attribute", "key": "value"}, + "condition": {"type": "NotAnEvaluator", "value": True}, + } + + with patch("tirith.core.core.get_evaluator_inputs_from_provider_inputs", return_value=[{"value": "x"}]): + result = generate_evaluator_result(evaluator_obj, {}, "test_provider") + + assert result["passed"] is False + assert result["result"] == [{"passed": False, "message": "`NotAnEvaluator` is not a supported evaluator"}] + + +@mark.passing +def test_generate_evaluator_result_bare_provider_err_is_surfaced(): + """ + A provider that reports "err" without a ProviderError is a malformed provider call (bad + operation_type, missing arg), not a policy violation. The message must reach the output + instead of being dropped and None evaluated against the condition. + """ + evaluator_obj = { + "id": "test_evaluator", + "provider_args": {"operation_type": "gt_value", "key": "value"}, + "condition": {"type": "Equals", "value": "us-east-1"}, + } + + bare_err = {"value": None, "meta": None, "err": "operation_type: gt_value is not supported"} + + with patch("tirith.core.core.get_evaluator_inputs_from_provider_inputs", return_value=[bare_err]): + with patch("tirith.core.core.EVALUATORS_DICT", {"Equals": MockEvaluator}): + result = generate_evaluator_result(evaluator_obj, {}, "test_provider") + + assert result["passed"] is False + assert len(result["result"]) == 1 + assert result["result"][0]["passed"] is False + assert result["result"][0]["message"] == "operation_type: gt_value is not supported" + + +@mark.passing +def test_generate_evaluator_result_bare_provider_err_ignores_error_tolerance(): + """error_tolerance tolerates missing data; it must never mask a malformed provider call.""" + evaluator_obj = { + "id": "test_evaluator", + "provider_args": {"operation_type": "gt_value", "key": "value"}, + # A tolerance high enough to swallow every documented severity, including 99. + "condition": {"type": "Equals", "value": "us-east-1", "error_tolerance": 100}, + } + + bare_err = {"value": None, "meta": None, "err": "operation_type: gt_value is not supported"} + + with patch("tirith.core.core.get_evaluator_inputs_from_provider_inputs", return_value=[bare_err]): + with patch("tirith.core.core.EVALUATORS_DICT", {"Equals": MockEvaluator}): + result = generate_evaluator_result(evaluator_obj, {}, "test_provider") + + assert result["passed"] is False, "a malformed provider call must not be skipped" + assert result["result"][0]["passed"] is False diff --git a/tests/core/test_output_compatibility.py b/tests/core/test_output_compatibility.py new file mode 100644 index 0000000..4dc6454 --- /dev/null +++ b/tests/core/test_output_compatibility.py @@ -0,0 +1,121 @@ +""" +Guardrails on the shape of the result document. + +The StackGuardian platform and the workflow-step templates parse this output, so its shape is a +contract rather than an implementation detail. `test_legacy_json_output_is_byte_identical` holds +the line: the golden file was captured before the engine changes landed, so any drift in the +single-policy output is a regression until proven otherwise. +""" + +import json +import os + +from pytest import mark + +from tirith.core.core import start_policy_evaluation_from_dict + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +GOLDEN_PATH = os.path.join(REPO_ROOT, "tests", "golden", "json_policy_output.json") + + +@mark.passing +def test_legacy_json_output_is_byte_identical(): + with open(os.path.join(REPO_ROOT, "tests", "providers", "json", "policy.json")) as f: + policy = json.load(f) + with open(os.path.join(REPO_ROOT, "tests", "providers", "json", "input.json")) as f: + input_data = json.load(f) + + result = start_policy_evaluation_from_dict(policy, input_data) + + with open(GOLDEN_PATH) as f: + # The golden file was captured from the CLI, whose print() adds a trailing newline + # that json.dumps does not produce. + expected = f.read().rstrip("\n") + + # indent=3 matches what the CLI emits (cli.py), so the golden file doubles as a + # record of the exact bytes a --json consumer receives. + assert json.dumps(result, indent=3) == expected + + +@mark.passing +def test_meta_passthrough_omits_absent_keys(): + """A policy declaring no optional metadata must produce exactly the two original keys.""" + policy = { + "meta": {"version": "v1", "required_provider": "stackguardian/json"}, + "evaluators": [ + { + "id": "check0", + "provider_args": {"operation_type": "get_value", "key_path": "a"}, + "condition": {"type": "Equals", "value": 1}, + } + ], + "eval_expression": "check0", + } + + result = start_policy_evaluation_from_dict(policy, {"a": 1}) + + assert result["meta"] == {"version": "v1", "required_provider": "stackguardian/json"} + + +@mark.passing +def test_meta_passthrough_carries_declared_keys(): + policy = { + "meta": { + "version": "v1", + "required_provider": "stackguardian/json", + "id": "no-public-ingress", + "name": "No 0.0.0.0/0 ingress", + "description": "Public ingress is not permitted", + "severity": "HIGH", + "enforcement": "hard_mandatory", + "tags": ["cis", "network"], + "remediation": "Restrict the CIDR or use a security-group reference", + }, + "evaluators": [ + { + "id": "check0", + "provider_args": {"operation_type": "get_value", "key_path": "a"}, + "condition": {"type": "Equals", "value": 1}, + } + ], + "eval_expression": "check0", + } + + result = start_policy_evaluation_from_dict(policy, {"a": 1}) + + assert result["meta"]["id"] == "no-public-ingress" + assert result["meta"]["name"] == "No 0.0.0.0/0 ingress" + assert result["meta"]["severity"] == "HIGH" + assert result["meta"]["enforcement"] == "hard_mandatory" + assert result["meta"]["tags"] == ["cis", "network"] + assert result["meta"]["remediation"] == "Restrict the CIDR or use a security-group reference" + # The originals survive alongside the additions. + assert result["meta"]["version"] == "v1" + assert result["meta"]["required_provider"] == "stackguardian/json" + + +@mark.passing +def test_meta_passthrough_supports_variables(): + """ + Variable substitution already covers the whole meta dict, so the new fields get + {{ var.x }} support without any extra plumbing. This pins that behaviour. + """ + policy = { + "meta": { + "version": "v1", + "required_provider": "stackguardian/json", + "severity": "{{ var.sev }}", + }, + "evaluators": [ + { + "id": "check0", + "provider_args": {"operation_type": "get_value", "key_path": "a"}, + "condition": {"type": "Equals", "value": 1}, + } + ], + "eval_expression": "check0", + } + + result = start_policy_evaluation_from_dict(policy, {"a": 1}, {"sev": "CRITICAL"}) + + assert result["meta"]["severity"] == "CRITICAL" diff --git a/tests/core/test_policy_parameterization.py b/tests/core/test_policy_parameterization.py index db9fcc0..08a5568 100644 --- a/tests/core/test_policy_parameterization.py +++ b/tests/core/test_policy_parameterization.py @@ -48,6 +48,56 @@ def test_not_found_variable(processed_policy): assert processed_policy[1] == ["key_path"] +def test_caller_policy_is_not_mutated(): + """Substitution must not write through to the caller's dict.""" + policy = { + "meta": {"version": "", "required_provider": "{{var.provider}}"}, + "evaluators": [ + { + "id": "check0", + "provider_args": {"operation_type": "get_value", "key_path": "a.b"}, + "condition": {"type": "Equals", "value": "{{var.expected}}"}, + } + ], + "eval_expression": "check0", + } + + replaced, not_found = get_policy_with_vars_replaced(policy, {"provider": "stackguardian/json", "expected": "yes"}) + + assert not_found == [] + # The copy carries the substituted values ... + assert replaced["meta"]["required_provider"] == "stackguardian/json" + assert replaced["evaluators"][0]["condition"]["value"] == "yes" + # ... while the original still carries the placeholders. + assert policy["meta"]["required_provider"] == "{{var.provider}}" + assert policy["evaluators"][0]["condition"]["value"] == "{{var.expected}}" + + +def test_same_policy_reused_with_different_vars(): + """ + A policy dict evaluated twice with different vars must not leak values between runs. + This is the multi-policy / retry case: without a deep copy the second call sees the + first call's substitutions already baked in and reports nothing to substitute. + """ + policy = { + "meta": {"version": "", "required_provider": "stackguardian/json"}, + "evaluators": [ + { + "id": "check0", + "provider_args": {"operation_type": "get_value", "key_path": "{{var.path}}"}, + "condition": {"type": "Equals", "value": True}, + } + ], + "eval_expression": "check0", + } + + first, _ = get_policy_with_vars_replaced(policy, {"path": "first.path"}) + second, _ = get_policy_with_vars_replaced(policy, {"path": "second.path"}) + + assert first["evaluators"][0]["provider_args"]["key_path"] == "first.path" + assert second["evaluators"][0]["provider_args"]["key_path"] == "second.path" + + # TODO: Create testcases for: # - test inline vars precendece over var files # - test undefined vars diff --git a/tests/golden/json_policy_output.json b/tests/golden/json_policy_output.json new file mode 100644 index 0000000..d0afad4 --- /dev/null +++ b/tests/golden/json_policy_output.json @@ -0,0 +1,87 @@ +{ + "meta": { + "version": "v1", + "required_provider": "stackguardian/json" + }, + "final_result": true, + "evaluators": [ + { + "id": "check0", + "passed": null, + "result": [ + { + "message": "key_path: `z.b` is not found (severity: 2)", + "passed": null + } + ], + "description": null + }, + { + "id": "check1", + "passed": true, + "result": [ + { + "passed": true, + "message": "`1` is less than equal to `1`", + "meta": null + } + ], + "description": null + }, + { + "id": "check2", + "passed": true, + "result": [ + { + "passed": true, + "message": "Found `\"aa\"` inside `[\"aa\", \"bb\"]`", + "meta": null + } + ], + "description": null + }, + { + "id": "check3", + "passed": true, + "result": [ + { + "passed": true, + "message": "`\"3\"` is equal to `\"3\"`", + "meta": null + } + ], + "description": null + }, + { + "id": "check4", + "passed": true, + "result": [ + { + "passed": true, + "message": "`\"value1\"` is equal to `\"value1\"`", + "meta": null + }, + { + "passed": true, + "message": "`\"value1\"` is equal to `\"value1\"`", + "meta": null + } + ], + "description": null + }, + { + "id": "check5", + "passed": true, + "result": [ + { + "passed": true, + "message": "`{\"e\": {\"f\": \"3\"}}` is equal to `{\"e\": {\"f\": \"3\"}}`", + "meta": null + } + ], + "description": null + } + ], + "errors": [], + "eval_expression": "check1 && check2 && check3 && check4 && check5" +} diff --git a/tests/platform/test_archive.py b/tests/platform/test_archive.py new file mode 100644 index 0000000..326c362 --- /dev/null +++ b/tests/platform/test_archive.py @@ -0,0 +1,248 @@ +""" +Tests for the project archive. + +The assertions that matter read the bytes *inside the built tarball*, not the objects handed to +pack(). That distinction is the whole point: a previous iteration of this code masked a plan +correctly in memory and still shipped the plaintext, because the secret lived in a second place +nobody had looked at. Asserting on the input would have passed. +""" + +import io +import json +import os +import tarfile + +import pytest + +from tirith.platform import archive + +SECRET = "hunter2-this-must-never-leave-the-runner" + + +def members(archive_bytes): + with tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:gz") as tar: + return sorted(tar.getnames()) + + +def read_member(archive_bytes, name): + with tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:gz") as tar: + return tar.extractfile(name).read() + + +def raw_bytes(archive_bytes): + """Everything in the archive, decompressed, as one blob -- for leak assertions.""" + blob = b"" + with tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:gz") as tar: + for member in tar.getmembers(): + blob += member.name.encode() + if member.isfile(): + blob += tar.extractfile(member).read() + return blob + + +# --- documents --------------------------------------------------------------------------------- + + +def test_documents_land_at_the_fixed_names_the_step_looks_for(tmp_path): + body, _manifest = archive.pack(source_dir=None, plan={"a": 1}, state={"b": 2}, infracost={"c": 3}) + + assert members(body) == ["infracost.json", "plan.json", "tfstate.json"] + assert json.loads(read_member(body, "plan.json")) == {"a": 1} + + +def test_absent_documents_are_simply_not_written(): + body, _manifest = archive.pack(source_dir=None, state={"version": 4}) + + assert members(body) == ["tfstate.json"] + + +def test_masked_document_wins_over_a_stale_file_on_disk(tmp_path): + """ + The dangerous ordering: a plan.json left in the working directory from an earlier run would + otherwise be packed *and* the masked one written, shipping both. + """ + (tmp_path / "plan.json").write_text(json.dumps({"leaked": SECRET})) + + body, _manifest = archive.pack(source_dir=str(tmp_path), plan={"masked": "__SG_REDACTED__"}) + + assert json.loads(read_member(body, "plan.json")) == {"masked": "__SG_REDACTED__"} + assert SECRET.encode() not in raw_bytes(body) + + +@pytest.mark.parametrize("name", ["plan.json", "tfstate.json", "infracost.json"]) +def test_reserved_names_on_disk_are_never_packed(tmp_path, name): + """ + The leak this closes: `terraform state pull > state.json` is the documented way to produce a + state file, so one routinely sits in the working directory -- raw and unmasked. Packing the + source tree naively shipped it in full, right next to the masked copy. + + These names are only ever written by pack() from an already-masked object. A caller who wants + the file evaluated passes --state-path / --input-path, which masks it first. + """ + (tmp_path / name).write_text(json.dumps({"outputs": {"db": {"value": SECRET}}})) + (tmp_path / "main.tf").write_text("") + + body, _manifest = archive.pack(source_dir=str(tmp_path), plan={"masked": True}) + + assert SECRET.encode() not in raw_bytes(body) + assert members(body) == ["main.tf", "plan.json"] + + +def test_masked_document_is_what_gets_written(tmp_path): + """The counterpart: a supplied document really does reach the archive.""" + (tmp_path / "tfstate.json").write_text(json.dumps({"secret": SECRET})) + + body, _manifest = archive.pack(source_dir=str(tmp_path), state={"masked": True}) + + assert json.loads(read_member(body, "tfstate.json")) == {"masked": True} + assert SECRET.encode() not in raw_bytes(body) + + +# --- exclusions -------------------------------------------------------------------------------- + + +def test_terraform_provider_cache_is_excluded(tmp_path): + """A provider cache is routinely hundreds of MB; shipping it would make every run unusable.""" + provider = tmp_path / ".terraform" / "providers" / "registry.terraform.io" + provider.mkdir(parents=True) + (provider / "terraform-provider-aws").write_bytes(b"x" * 1024) + (tmp_path / "main.tf").write_text('resource "null_resource" "a" {}') + + body, manifest = archive.pack(source_dir=str(tmp_path)) + + assert members(body) == ["main.tf"] + assert manifest["skipped"] >= 1 + + +def test_git_directory_is_excluded(tmp_path): + """.git carries full history, so anything ever committed would ship.""" + (tmp_path / ".git").mkdir() + (tmp_path / ".git" / "config").write_text(f"token = {SECRET}") + (tmp_path / "main.tf").write_text("") + + body, _manifest = archive.pack(source_dir=str(tmp_path)) + + assert members(body) == ["main.tf"] + assert SECRET.encode() not in raw_bytes(body) + + +@pytest.mark.parametrize("name", ["terraform.tfstate", "terraform.tfstate.backup", "prod.tfstate"]) +def test_raw_state_files_are_excluded(tmp_path, name): + """ + Raw state is unmasked by definition. Left in, it would travel next to the masked copy and + undo the masking entirely. + """ + (tmp_path / name).write_text(json.dumps({"outputs": {"db": {"value": SECRET}}})) + (tmp_path / "main.tf").write_text("") + + body, _manifest = archive.pack(source_dir=str(tmp_path)) + + assert name not in members(body) + assert SECRET.encode() not in raw_bytes(body) + + +def test_gitignore_is_honoured(tmp_path): + (tmp_path / ".gitignore").write_text("secrets.auto.tfvars\nbuild/\n") + (tmp_path / "secrets.auto.tfvars").write_text(f'password = "{SECRET}"') + (tmp_path / "build").mkdir() + (tmp_path / "build" / "out.bin").write_text("junk") + (tmp_path / "main.tf").write_text("") + + body, _manifest = archive.pack(source_dir=str(tmp_path)) + + assert "secrets.auto.tfvars" not in members(body) + assert "build/out.bin" not in members(body) + assert SECRET.encode() not in raw_bytes(body) + + +def test_gitignore_can_be_turned_off(tmp_path): + (tmp_path / ".gitignore").write_text("keep-me.tf\n") + (tmp_path / "keep-me.tf").write_text("") + + body, _manifest = archive.pack(source_dir=str(tmp_path), respect_gitignore=False) + + assert "keep-me.tf" in members(body) + + +def test_extra_excludes_are_applied(tmp_path): + (tmp_path / "big.zip").write_text("junk") + (tmp_path / "main.tf").write_text("") + + body, _manifest = archive.pack(source_dir=str(tmp_path), extra_excludes=("*.zip",)) + + assert members(body) == ["main.tf"] + + +def test_lock_file_is_kept(tmp_path): + """It pins provider versions, is small, and the run controller's init wants it.""" + (tmp_path / ".terraform.lock.hcl").write_text("provider ...") + + body, _manifest = archive.pack(source_dir=str(tmp_path)) + + assert ".terraform.lock.hcl" in members(body) + + +def test_symlinks_are_skipped(tmp_path): + """A symlink out of the tree either breaks on extraction or smuggles a file in.""" + outside = tmp_path.parent / "outside.txt" + outside.write_text(SECRET) + source = tmp_path / "src" + source.mkdir() + (source / "main.tf").write_text("") + os.symlink(str(outside), str(source / "link.txt")) + + body, _manifest = archive.pack(source_dir=str(source)) + + assert members(body) == ["main.tf"] + assert SECRET.encode() not in raw_bytes(body) + + +# --- structure --------------------------------------------------------------------------------- + + +def test_nested_directories_keep_their_relative_paths(tmp_path): + (tmp_path / "modules" / "vpc").mkdir(parents=True) + (tmp_path / "modules" / "vpc" / "main.tf").write_text("") + (tmp_path / "main.tf").write_text("") + + body, _manifest = archive.pack(source_dir=str(tmp_path)) + + assert "modules/vpc/main.tf" in members(body) + + +def test_no_source_dir_is_allowed(): + """--no-source: send only the documents.""" + body, manifest = archive.pack(source_dir=None, plan={"a": 1}) + + assert members(body) == ["plan.json"] + assert manifest["files"] == 0 + + +def test_missing_source_dir_is_an_error(tmp_path): + with pytest.raises(archive.ArchiveError): + archive.pack(source_dir=str(tmp_path / "does-not-exist")) + + +def test_oversized_archive_is_refused(tmp_path, monkeypatch): + """ + Failing loudly beats a five-minute upload that times out the run. A runaway archive is nearly + always an exclusion that did not fire. + """ + monkeypatch.setattr(archive, "MAX_ARCHIVE_BYTES", 512) + (tmp_path / "big.tf").write_text("resource {}\n" * 20000) + + with pytest.raises(archive.ArchiveError, match="limit"): + archive.pack(source_dir=str(tmp_path)) + + +def test_manifest_reports_what_went_in(tmp_path): + (tmp_path / "main.tf").write_text("") + (tmp_path / ".terraform").mkdir() + (tmp_path / ".terraform" / "x").write_text("") + + _body, manifest = archive.pack(source_dir=str(tmp_path), plan={"a": 1}) + + assert manifest["files"] == 1 + assert manifest["documents"] == ["plan.json"] + assert manifest["skipped"] >= 1 + assert manifest["bytes"] > 0 diff --git a/tests/platform/test_client.py b/tests/platform/test_client.py new file mode 100644 index 0000000..ba9b8af --- /dev/null +++ b/tests/platform/test_client.py @@ -0,0 +1,226 @@ +""" +Tests for the StackGuardian client. + +The polling contract is the part worth pinning: a run that rests in a state the poller does not +recognise as terminal spins until the timeout and is then reported as a tool failure -- turning a +completed evaluation into what looks like an outage. +""" + +import json + +import pytest + +from tirith.platform import client +from tirith.platform.client import SGClient, SGError, _extract_signed_url + +# --- terminal statuses ------------------------------------------------------------------------- + + +def test_approval_required_is_terminal(): + """ + A regression test. APPROVAL_REQUIRED is a resting state -- reached when a policy's onFail is + APPROVAL_REQUIRED -- and nothing further happens without a human. Treating it as transient + made the poller spin to its timeout and report a tool failure for a finished evaluation. + """ + assert "APPROVAL_REQUIRED" in client.TERMINAL_STATUSES + + +@pytest.mark.parametrize("status", ["COMPLETED", "ERRORED", "CANCELLED", "APPROVAL_REQUIRED"]) +def test_terminal_statuses_stop_the_poll(status): + assert status in client.TERMINAL_STATUSES + + +@pytest.mark.parametrize("status", ["QUEUED", "PENDING", "RUNNING"]) +def test_transient_statuses_keep_polling(status): + """A run can sit in QUEUED behind the per-workflow concurrency gate for a long while.""" + assert status not in client.TERMINAL_STATUSES + + +def test_wait_for_run_returns_on_a_terminal_status(monkeypatch): + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + statuses = iter([{"LatestStatus": "QUEUED"}, {"LatestStatus": "RUNNING"}, {"LatestStatus": "COMPLETED"}]) + monkeypatch.setattr(sg, "get_run", lambda *a, **k: next(statuses)) + monkeypatch.setattr(client.time, "sleep", lambda _s: None) + + status, _run = sg.wait_for_run("default", "wf", "run", timeout=30) + + assert status == "COMPLETED" + + +def test_wait_for_run_reports_each_status_change(monkeypatch): + """Without this a run queued behind another looks identical to a hung one.""" + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + statuses = iter([{"LatestStatus": "QUEUED"}, {"LatestStatus": "QUEUED"}, {"LatestStatus": "COMPLETED"}]) + monkeypatch.setattr(sg, "get_run", lambda *a, **k: next(statuses)) + monkeypatch.setattr(client.time, "sleep", lambda _s: None) + seen = [] + + sg.wait_for_run("default", "wf", "run", timeout=30, on_poll=seen.append) + + assert seen == ["QUEUED", "COMPLETED"], "only changes are reported, not every poll" + + +def test_wait_for_run_timeout_is_an_error_never_a_pass(monkeypatch): + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + monkeypatch.setattr(sg, "get_run", lambda *a, **k: {"LatestStatus": "RUNNING"}) + monkeypatch.setattr(client.time, "sleep", lambda _s: None) + + with pytest.raises(SGError): + sg.wait_for_run("default", "wf", "run", timeout=-1) + + +# --- signed URL extraction --------------------------------------------------------------------- + + +def test_extract_signed_url_accepts_a_bare_string_in_msg(): + """What tfstate_upload_url actually returns.""" + assert _extract_signed_url({"msg": "https://s3.example/put"}) == "https://s3.example/put" + + +def test_extract_signed_url_accepts_a_nested_object(): + assert _extract_signed_url({"data": {"signedUrl": "https://s3.example/put"}}) == "https://s3.example/put" + + +def test_extract_signed_url_returns_none_when_absent(): + assert _extract_signed_url({"msg": "some error text"}) is None + + +# --- archive upload ---------------------------------------------------------------------------- + + +def test_upload_archive_requires_a_storage_key(monkeypatch): + """ + The key is what the caller passes back as terraformProjectZip. A platform that predates the + endpoint returns a bare URL, and silently continuing would create a run pointing at nothing. + """ + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + monkeypatch.setattr(sg, "_request", lambda *a, **k: (200, {"msg": "https://s3.example/put"})) + + with pytest.raises(SGError, match="storage key"): + sg.upload_archive("default", "wf", "a.tar.gz", "abc1234", b"x") + + +def test_upload_archive_returns_the_key_from_the_response(monkeypatch): + """ + Never rebuilt client-side: the layout is runner-aware, so a guess is wrong for exactly the + customers whose runs are hardest to debug. + """ + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + monkeypatch.setattr( + sg, + "_request", + lambda *a, **k: (200, {"msg": {"signedUrl": "https://s3.example/put", "key": "orgs/acme/…/a.tar.gz"}}), + ) + uploaded = {} + + def fake_urlopen(request, timeout=None): + uploaded["content_type"] = request.get_header("Content-type") + uploaded["body"] = request.data + + class _R: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + return _R() + + monkeypatch.setattr(client.urllib.request, "urlopen", fake_urlopen) + + key = sg.upload_archive("default", "wf", "a.tar.gz", "abc1234", b"tarbytes") + + assert key == "orgs/acme/…/a.tar.gz" + assert uploaded["body"] == b"tarbytes" + # Must match what the URL was signed with, or S3 rejects it as a signature mismatch. + assert uploaded["content_type"] == "application/gzip" + + +# --- run creation ------------------------------------------------------------------------------ + + +def test_create_run_sends_no_step_config(monkeypatch): + """ + core ignores WfStepsConfig for TERRAFORM workflows and synthesises the steps from the stored + TerraformConfig plus this TerraformAction. Sending one would be dead weight that reads as if + it were doing something. + """ + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + captured = {} + + def fake_request(method, path, body=None, **kwargs): + captured["body"] = body + return 200, {"data": {"ResourceName": "wfrun-1"}} + + monkeypatch.setattr(sg, "_request", fake_request) + + run_id, _data = sg.create_run("default", "wf", "orgs/acme/…/a.tar.gz", {"type": "github_action"}) + + assert run_id == "wfrun-1" + assert "WfStepsConfig" not in captured["body"] + assert captured["body"]["TerraformAction"] == {"action": "policy-only"} + assert captured["body"]["terraformProjectZip"] == "orgs/acme/…/a.tar.gz" + + +def test_ensure_workflow_creates_a_terraform_workflow(monkeypatch): + """ + TERRAFORM rather than CUSTOM: it is what makes core synthesise the steps from TerraformConfig, + and what makes the run render as a real terraform run in the dashboard. + """ + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + captured = {} + + def fake_request(method, path, body=None, **kwargs): + captured["body"] = body + return 201, {} + + monkeypatch.setattr(sg, "_request", fake_request) + + sg.ensure_workflow("default", "wf", "desc", {"terraformVersion": "1.5.7"}) + + assert captured["body"]["WfType"] == "TERRAFORM" + assert captured["body"]["TerraformConfig"] == {"terraformVersion": "1.5.7"} + assert captured["body"]["Id"] == captured["body"]["ResourceName"] == "wf" + + +def test_conflict_on_create_is_success(monkeypatch): + """Re-running the action against an existing workflow must not be an error.""" + sg = SGClient("https://api.example/api/v1", "acme", "sgo_x") + monkeypatch.setattr(sg, "_request", lambda *a, **k: (409, {"msg": "already exists"})) + + assert sg.ensure_workflow("default", "wf", "d", {}) == 409 + assert sg.ensure_workflow_group("default") == 409 + + +# --- auth -------------------------------------------------------------------------------------- + + +def test_auth_header_uses_the_apikey_scheme(monkeypatch): + """Matches sg-cli: `Authorization: apikey `, not Bearer.""" + sg = SGClient("https://api.example/api/v1", "acme", "sgo_secret") + captured = {} + + def fake_urlopen(request, timeout=None): + captured["auth"] = request.get_header("Authorization") + + class _R: + status = 200 + + def read(self): + return json.dumps({"msg": "ok"}).encode() + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + return _R() + + monkeypatch.setattr(client.urllib.request, "urlopen", fake_urlopen) + + sg._request("GET", "/wfgrps/") + + assert captured["auth"] == "apikey sgo_secret" diff --git a/tests/platform/test_redact.py b/tests/platform/test_redact.py new file mode 100644 index 0000000..7c56b96 --- /dev/null +++ b/tests/platform/test_redact.py @@ -0,0 +1,635 @@ +""" +Tests for plan/state redaction. + +This is the security-critical module: it is the only thing standing between a customer's secrets +and StackGuardian's storage. The tests assert on the *serialized bytes* wherever a leak would +matter, because a value nested somewhere unexpected still leaks even if the top-level shape looks +masked. +""" + +import json +import os +import sys + + +from tirith.platform import redact + +SECRET = "hunter2-this-must-never-leave-the-runner" + + +def test_slim_drops_prior_state_and_planned_values(): + """ + `planned_values` is the important one. It mirrors every resource's values in a second place + and carries NO sensitivity markers, so masking `resource_changes` alone leaves the same secret + in plaintext there. A real plan leaked a local_sensitive_file body through exactly this path. + """ + plan = { + "format_version": "1.2", + "terraform_version": "1.5.7", + "resource_changes": [], + "prior_state": {"values": {"secret": SECRET}}, + "planned_values": {"root_module": {"resources": [{"values": {"content": SECRET}}]}}, + } + + slimmed = redact.slim_plan(plan) + + assert "prior_state" not in slimmed + assert "planned_values" not in slimmed + assert slimmed["resource_changes"] == [] + assert slimmed["terraform_version"] == "1.5.7" + assert SECRET not in json.dumps(slimmed) + + +def test_planned_values_leak_is_closed_end_to_end(): + """The exact shape that leaked in QA: masked in resource_changes, plaintext in planned_values.""" + plan = { + "resource_changes": [ + { + "type": "local_sensitive_file", + "change": {"after": {"content": SECRET}, "after_sensitive": {"content": True}}, + } + ], + "planned_values": { + "root_module": {"resources": [{"type": "local_sensitive_file", "values": {"content": SECRET}}]} + }, + } + + redacted = redact.redact_plan(plan) + + assert SECRET not in json.dumps(redacted) + + +def test_configuration_is_kept_because_three_operations_read_it(): + """ + Dropping `configuration` would silently break direct_dependencies, direct_references and + provider_config: policies would stop finding what they look for rather than failing loudly. + """ + plan = { + "resource_changes": [], + "configuration": { + "root_module": {"resources": [{"address": "aws_vpc.main", "depends_on": ["aws_x.y"]}]}, + "provider_config": { + "aws": { + "name": "aws", + "full_name": "registry.terraform.io/hashicorp/aws", + "version_constraint": "~> 5.0", + "expressions": { + "region": {"constant_value": "eu-central-1"}, + "secret_key": {"constant_value": SECRET}, + "assume_role": {"role_arn": {"constant_value": SECRET}}, + }, + } + }, + }, + } + + slimmed = redact.slim_plan(plan) + aws = slimmed["configuration"]["provider_config"]["aws"] + + # What the provider_config operation reads survives ... + assert aws["full_name"] == "registry.terraform.io/hashicorp/aws" + assert aws["version_constraint"] == "~> 5.0" + assert aws["expressions"]["region"]["constant_value"] == "eu-central-1" + # ... and the reference graph the other two operations walk survives ... + assert slimmed["configuration"]["root_module"]["resources"][0]["depends_on"] == ["aws_x.y"] + # ... while hardcoded credentials do not. + assert "secret_key" not in aws["expressions"] + assert "assume_role" not in aws["expressions"] + assert SECRET not in json.dumps(slimmed) + + +def test_hcl_literals_are_scrubbed_from_resource_expressions(): + """ + The third instance of the `planned_values` pattern, caught in QA: a hardcoded value is masked + in `resource_changes` and sits in plaintext under + `configuration.root_module.resources[].expressions[].constant_value`, which carries no + sensitivity markers at all. + + Dropping it is lossless -- direct_references reads only `references`, direct_dependencies only + `depends_on`. + """ + plan = { + "resource_changes": [ + { + "type": "local_sensitive_file", + "change": {"after": {"content": SECRET}, "after_sensitive": {"content": True}}, + } + ], + "configuration": { + "root_module": { + "resources": [ + { + "address": "local_sensitive_file.creds", + "depends_on": ["null_resource.a"], + "expressions": { + "content": {"constant_value": SECRET}, + "filename": {"references": ["path.module"]}, + }, + } + ] + } + }, + } + + redacted = redact.redact_plan(plan) + expressions = redacted["configuration"]["root_module"]["resources"][0]["expressions"] + + assert SECRET not in json.dumps(redacted) + # The reference graph the operations walk survives ... + assert expressions["filename"]["references"] == ["path.module"] + assert redacted["configuration"]["root_module"]["resources"][0]["depends_on"] == ["null_resource.a"] + # ... the literal does not. + assert "constant_value" not in expressions["content"] + + +def test_nested_and_repeated_block_literals_are_scrubbed(): + """A block argument is a dict of expressions and a repeated block is a list of them.""" + plan = { + "resource_changes": [], + "configuration": { + "root_module": { + "resources": [ + { + "address": "aws_instance.web", + "expressions": { + "root_block_device": {"kms_key_id": {"constant_value": SECRET}}, + "ebs_block_device": [ + {"snapshot_id": {"constant_value": SECRET}}, + {"volume_id": {"references": ["aws_ebs_volume.a.id"]}}, + ], + }, + } + ] + } + }, + } + + redacted = redact.redact_plan(plan) + + assert SECRET not in json.dumps(redacted) + ebs = redacted["configuration"]["root_module"]["resources"][0]["expressions"]["ebs_block_device"] + assert ebs[1]["volume_id"]["references"] == ["aws_ebs_volume.a.id"] + + +def test_child_module_literals_are_scrubbed(): + plan = { + "resource_changes": [], + "configuration": { + "root_module": { + "module_calls": { + "db": { + "source": "./modules/db", + "expressions": {"password": {"constant_value": SECRET}}, + "module": { + "resources": [ + { + "address": "aws_db_instance.main", + "expressions": {"password": {"constant_value": SECRET}}, + } + ] + }, + } + } + } + }, + } + + redacted = redact.redact_plan(plan) + + assert SECRET not in json.dumps(redacted) + + +def test_variable_defaults_and_outputs_are_scrubbed(): + """A `default` on a sensitive variable is a literal in the configuration too.""" + plan = { + "resource_changes": [], + "configuration": { + "root_module": { + "variables": {"db_password": {"default": SECRET, "sensitive": True}}, + "outputs": {"conn": {"expression": {"constant_value": SECRET}}}, + } + }, + } + + redacted = redact.redact_plan(plan) + + assert SECRET not in json.dumps(redacted) + # The declaration itself survives; only the value goes. + assert redacted["configuration"]["root_module"]["variables"]["db_password"]["sensitive"] is True + + +def test_scrub_tolerates_a_provider_config_without_expressions(): + plan = {"resource_changes": [], "configuration": {"provider_config": {"null": {"name": "null"}}}} + + slimmed = redact.slim_plan(plan) + + assert slimmed["configuration"]["provider_config"]["null"] == {"name": "null"} + + +def test_redact_masks_marked_attributes(): + plan = { + "resource_changes": [ + { + "address": "aws_db_instance.main", + "type": "aws_db_instance", + "change": { + "actions": ["create"], + "before": None, + "after": {"identifier": "main", "password": SECRET, "port": 5432}, + "after_sensitive": {"password": True}, + }, + } + ] + } + + redacted = redact.redact_plan(plan) + after = redacted["resource_changes"][0]["change"]["after"] + + assert after["password"] == redact.SENTINEL + assert after["identifier"] == "main", "non-sensitive values must survive" + assert after["port"] == 5432 + assert SECRET not in json.dumps(redacted) + + +def test_redact_masks_a_whole_sensitive_subtree(): + """A marker of `true` above an object masks everything beneath it.""" + plan = { + "resource_changes": [ + { + "address": "aws_secretsmanager_secret_version.v", + "change": { + "after": {"secret_string": {"user": "admin", "pass": SECRET}}, + "after_sensitive": {"secret_string": True}, + }, + } + ] + } + + redacted = redact.redact_plan(plan) + + assert redacted["resource_changes"][0]["change"]["after"]["secret_string"] == redact.SENTINEL + assert SECRET not in json.dumps(redacted) + + +def test_redact_masks_inside_lists_positionally(): + plan = { + "resource_changes": [ + { + "change": { + "after": {"items": [{"k": "public"}, {"k": SECRET}]}, + "after_sensitive": {"items": [{}, {"k": True}]}, + } + } + ] + } + + redacted = redact.redact_plan(plan) + items = redacted["resource_changes"][0]["change"]["after"]["items"] + + assert items[0]["k"] == "public" + assert items[1]["k"] == redact.SENTINEL + assert SECRET not in json.dumps(redacted) + + +def test_redact_masks_before_as_well_as_after(): + """A destroy or update leaves the old secret in `before`; it leaks just as badly.""" + plan = { + "resource_changes": [ + { + "change": { + "actions": ["delete"], + "before": {"password": SECRET}, + "before_sensitive": {"password": True}, + "after": None, + } + } + ] + } + + redacted = redact.redact_plan(plan) + + assert redacted["resource_changes"][0]["change"]["before"]["password"] == redact.SENTINEL + assert SECRET not in json.dumps(redacted) + + +def test_redact_drops_root_variables_entirely(): + """ + The plan does not reliably mark which root variables were declared sensitive, so the only safe + assumption is that any of them might be. + """ + plan = {"resource_changes": [], "variables": {"db_password": {"value": SECRET}}} + + redacted = redact.redact_plan(plan) + + assert "variables" not in redacted + assert SECRET not in json.dumps(redacted) + + +def test_redact_masks_sensitive_output_changes(): + plan = { + "resource_changes": [], + "output_changes": { + "db_url": {"actions": ["create"], "after": SECRET, "sensitive": True}, + "region": {"actions": ["create"], "after": "eu-central-1", "sensitive": False}, + }, + } + + redacted = redact.redact_plan(plan) + + assert redacted["output_changes"]["db_url"]["after"] == redact.SENTINEL + assert redacted["output_changes"]["region"]["after"] == "eu-central-1" + assert SECRET not in json.dumps(redacted) + + +def test_redact_leaves_unmarked_values_alone(): + """ + Documents the known limitation honestly: terraform's markers are not exhaustive, so a secret + that arrives unmarked is NOT masked. Slimming and the variables drop limit the blast radius; + this test exists so the gap is visible rather than assumed away. + """ + plan = {"resource_changes": [{"change": {"after": {"password_from_locals": SECRET}, "after_sensitive": {}}}]} + + redacted = redact.redact_plan(plan) + + assert redacted["resource_changes"][0]["change"]["after"]["password_from_locals"] == SECRET + + +def test_redact_plan_tolerates_junk(): + assert redact.redact_plan({}) == {} + assert redact.redact_plan({"resource_changes": "not-a-list"})["resource_changes"] == "not-a-list" + assert redact.redact_plan([]) == [] + + +# --- state ------------------------------------------------------------------------------------- + + +def test_redact_state_masks_sensitive_outputs(): + state = { + "version": 4, + "outputs": { + "db_password": {"value": SECRET, "type": "string", "sensitive": True}, + "region": {"value": "eu-central-1", "type": "string"}, + }, + "resources": [], + } + + redacted = redact.redact_state(state) + + assert redacted["outputs"]["db_password"]["value"] == redact.SENTINEL + assert redacted["outputs"]["region"]["value"] == "eu-central-1" + assert SECRET not in json.dumps(redacted) + + +def test_redact_state_masks_sensitive_attributes(): + """ + The shape `terraform state pull` actually writes: each entry is a PATH -- a list of steps -- + not a single key. + + Captured verbatim from a real `local_sensitive_file`. The previous fixture here invented the + flat form, so this passed while real state was not masked at all: a list is neither a dict nor + a string, so every entry was skipped. + """ + state = { + "resources": [ + { + "type": "local_sensitive_file", + "name": "s", + "instances": [ + { + "attributes": {"id": "e590ef", "content": SECRET, "content_base64": SECRET}, + "sensitive_attributes": [ + [{"type": "get_attr", "value": "content_base64"}], + [{"type": "get_attr", "value": "content"}], + ], + } + ], + } + ] + } + + redacted = redact.redact_state(state) + attributes = redacted["resources"][0]["instances"][0]["attributes"] + + assert attributes["content"] == redact.SENTINEL + assert attributes["content_base64"] == redact.SENTINEL + assert attributes["id"] == "e590ef", "non-sensitive attributes must survive" + assert SECRET not in json.dumps(redacted) + + +def test_redact_state_masks_a_nested_attribute_path(): + """A path can descend through objects and list indices, not just name a top-level key.""" + state = { + "resources": [ + { + "instances": [ + { + "attributes": {"config": [{"token": SECRET, "url": "https://ok"}]}, + "sensitive_attributes": [ + [ + {"type": "get_attr", "value": "config"}, + {"type": "index", "value": 0}, + {"type": "get_attr", "value": "token"}, + ] + ], + } + ] + } + ] + } + + redacted = redact.redact_state(state) + config = redacted["resources"][0]["instances"][0]["attributes"]["config"][0] + + assert config["token"] == redact.SENTINEL + assert config["url"] == "https://ok" + + +def test_redact_state_does_not_mutate_the_input(): + """The caller still holds the original; masking must not reach back into it.""" + state = { + "resources": [ + { + "instances": [ + { + "attributes": {"password": SECRET}, + "sensitive_attributes": [[{"type": "get_attr", "value": "password"}]], + } + ] + } + ] + } + + redact.redact_state(state) + + assert state["resources"][0]["instances"][0]["attributes"]["password"] == SECRET + + +def test_redact_state_accepts_the_flat_get_attr_form(): + """Some providers and older state versions emit a single step rather than a path.""" + state = { + "resources": [ + { + "instances": [ + { + "attributes": {"password": SECRET}, + "sensitive_attributes": [{"type": "get_attr", "value": "password"}], + } + ] + } + ] + } + + redacted = redact.redact_state(state) + + assert redacted["resources"][0]["instances"][0]["attributes"]["password"] == redact.SENTINEL + + +def test_redact_state_accepts_bare_string_sensitive_attributes(): + """Older state versions write these as plain strings rather than objects.""" + state = {"resources": [{"instances": [{"attributes": {"secret": SECRET}, "sensitive_attributes": ["secret"]}]}]} + + redacted = redact.redact_state(state) + + assert redacted["resources"][0]["instances"][0]["attributes"]["secret"] == redact.SENTINEL + + +def test_redact_state_tolerates_junk(): + assert redact.redact_state({}) == {} + assert redact.redact_state({"resources": "nope"})["resources"] == "nope" + assert redact.redact_state({"outputs": None})["outputs"] is None + + +def test_count_redactions(): + document = {"a": redact.SENTINEL, "b": [redact.SENTINEL, "fine"], "c": {"d": redact.SENTINEL}} + + assert redact.count_redactions(document) == 3 + assert redact.count_redactions({"a": "fine"}) == 0 + + +# --- output_changes marker spellings ------------------------------------------------------------- +# +# These exist because a real plan slipped through: the code originally checked only a top-level +# `sensitive` key, but modern terraform emits `before_sensitive` / `after_sensitive` per side, so +# every sensitive output in a current plan went unmasked. + + +def test_output_change_masked_via_after_sensitive(): + """The spelling modern terraform actually uses.""" + plan = { + "resource_changes": [], + "output_changes": { + "db_url": {"actions": ["update"], "before": "old", "after": SECRET, "after_sensitive": True} + }, + } + + redacted = redact.redact_plan(plan) + + assert redacted["output_changes"]["db_url"]["after"] == redact.SENTINEL + assert SECRET not in json.dumps(redacted) + + +def test_output_change_masks_each_side_independently(): + """An output can become sensitive without having been so before, and vice versa.""" + plan = { + "resource_changes": [], + "output_changes": { + "rotated": { + "actions": ["update"], + "before": SECRET, + "after": "now-public", + "before_sensitive": True, + "after_sensitive": False, + } + }, + } + + redacted = redact.redact_plan(plan) + change = redacted["output_changes"]["rotated"] + + assert change["before"] == redact.SENTINEL + assert change["after"] == "now-public" + assert SECRET not in json.dumps(redacted) + + +def test_output_change_legacy_sensitive_key_masks_both_sides(): + plan = { + "resource_changes": [], + "output_changes": {"k": {"before": SECRET, "after": SECRET, "sensitive": True}}, + } + + redacted = redact.redact_plan(plan) + + assert redacted["output_changes"]["k"]["before"] == redact.SENTINEL + assert redacted["output_changes"]["k"]["after"] == redact.SENTINEL + + +def test_output_change_does_not_invent_absent_keys(): + """ + A create whose value is not yet known has no `after` at all (`after_unknown: true`). Adding a + sentinel would fabricate data the plan never carried, and would misrepresent the plan to any + policy reading it. + """ + plan = { + "resource_changes": [], + "output_changes": { + "pw": {"actions": ["create"], "before": None, "after_unknown": True, "after_sensitive": True} + }, + } + + redacted = redact.redact_plan(plan) + change = redacted["output_changes"]["pw"] + + assert "after" not in change + assert change["before"] is None + + +def test_unknown_create_values_are_simply_absent_from_the_plan(): + """ + Documents a property that made an earlier end-to-end test weaker than intended: for a create, + terraform does not know the value yet, so it is absent from `after` rather than present and + masked. Nothing leaks -- but a test that expects to see a sentinel here is testing nothing. + """ + plan = { + "resource_changes": [ + { + "type": "random_password", + "change": { + "actions": ["create"], + "after": {"length": 32}, + "after_unknown": {"result": True}, + "after_sensitive": {"result": True}, + }, + } + ] + } + + redacted = redact.redact_plan(plan) + after = redacted["resource_changes"][0]["change"]["after"] + + assert "result" not in after + assert redact.count_redactions(redacted) == 0 + + +def test_known_sensitive_value_at_plan_time_is_masked(): + """ + The case that DOES exercise marker-driven redaction: a hardcoded sensitive attribute is known + at plan time, so it really is in `after` and really must be replaced. + """ + plan = { + "resource_changes": [ + { + "type": "local_sensitive_file", + "change": { + "actions": ["create"], + "after": {"filename": "out.txt", "content": SECRET}, + "after_sensitive": {"content": True}, + }, + } + ] + } + + redacted = redact.redact_plan(plan) + + assert redacted["resource_changes"][0]["change"]["after"]["content"] == redact.SENTINEL + assert redacted["resource_changes"][0]["change"]["after"]["filename"] == "out.txt" + assert SECRET not in json.dumps(redacted) diff --git a/tests/platform/test_report.py b/tests/platform/test_report.py new file mode 100644 index 0000000..0a9b1aa --- /dev/null +++ b/tests/platform/test_report.py @@ -0,0 +1,250 @@ +""" +Tests for verdict computation and comment rendering. + +The verdict mapping is the part worth pinning hardest: every path that does not produce a real +"everything passed" must stay distinguishable from one that does, and must never map to a green +required check. +""" + +import os +import sys + + +from tirith.platform import report as render + + +def _results(result="FAIL", **rule_overrides): + rule = { + "rule_name": "ingress-cidr", + "result": result, + "evaluations": { + "fails": [ + { + "id": "check1", + "result": [ + { + "passed": False, + "message": "`0.0.0.0/0` is contained in `cidr_blocks`", + "meta": {"address": "module.net.aws_security_group.web"}, + } + ], + } + ] + }, + } + rule.update(rule_overrides) + return {"no-public-ingress": [rule]} + + +# --- summarize --------------------------------------------------------------------------------- + + +def test_summarize_counts_and_extracts_detail(): + counts, findings = render.summarize(_results()) + + assert counts["FAIL"] == 1 + assert findings[0]["policy_id"] == "no-public-ingress" + assert findings[0]["messages"] == ["`0.0.0.0/0` is contained in `cidr_blocks`"] + assert findings[0]["resources"] == ["module.net.aws_security_group.web"] + + +def test_summarize_counts_skipped_separately_from_passed(): + """Reporting a skipped control as passing would be a quiet inaccuracy.""" + counts, findings = render.summarize({"p": [{"rule_name": "r", "skip": True}]}) + + assert counts["SKIPPED"] == 1 + assert counts["PASS"] == 0 + assert findings[0]["result"] == "SKIPPED" + + +def test_summarize_surfaces_engine_errors_distinctly(): + """ + A malformed policy must not read as a policy violation. Prefixing makes it obvious in the + comment that the engine, not the infrastructure, is the problem. + """ + results = {"p": [{"rule_name": "r", "result": "FAIL", "evaluations": {"fails": [{"exec_err": "bad op"}]}}]} + + _, findings = render.summarize(results) + + assert findings[0]["messages"] == ["engine: bad op"] + + +def test_summarize_handles_providers_without_resource_addresses(): + """Only terraform_plan populates meta; json/kubernetes set it to None.""" + results = { + "p": [ + { + "rule_name": "r", + "result": "FAIL", + "evaluations": {"fails": [{"id": "c", "result": [{"message": "no", "meta": None}]}]}, + } + ] + } + + _, findings = render.summarize(results) + + assert findings[0]["resources"] == [] + assert findings[0]["messages"] == ["no"] + + +def test_summarize_tolerates_empty_and_none(): + assert render.summarize(None)[0]["FAIL"] == 0 + assert render.summarize({})[1] == [] + + +# --- verdict ----------------------------------------------------------------------------------- + + +def test_verdict_failed_when_any_policy_fails(): + counts, _ = render.summarize(_results("FAIL")) + assert render.verdict(counts, "COMPLETED") == "failed" + + +def test_verdict_warned_for_a_warning(): + counts, _ = render.summarize(_results("WARN")) + assert render.verdict(counts, "COMPLETED") == "warned" + + +def test_verdict_approval_required_outranks_warned(): + """ + A rule result of APPROVAL_REQUIRED means its author wrote `onFail: APPROVAL_REQUIRED`. The + policy-only step records that without pausing the run, so the run comes back COMPLETED and only + the counts carry the intent. + + Folding it into `warned` was wrong: `warned` maps to a `neutral` check, which SATISFIES a + required status check, so a policy demanding human sign-off silently did not block. Caught by a + live run against a real APPROVAL_REQUIRED policy. + """ + counts, _ = render.summarize(_results("APPROVAL_REQUIRED")) + + assert render.verdict(counts, "COMPLETED") == "approval-required" + + +def test_verdict_failed_outranks_approval_required(): + """A hard failure is the more urgent signal when a run has both.""" + counts = {"FAIL": 1, "APPROVAL_REQUIRED": 1} + + assert render.verdict(counts, "COMPLETED") == "failed" + + +def test_verdict_passed_only_when_a_policy_actually_passed(): + counts, _ = render.summarize(_results("PASS")) + assert render.verdict(counts, "COMPLETED") == "passed" + + +def test_verdict_errored_for_a_non_completed_run(): + """An ERRORED or CANCELLED run produced no verdict; that is not a pass.""" + counts, _ = render.summarize(_results("PASS")) + for status in ("ERRORED", "CANCELLED", "RUNNING", None): + assert render.verdict(counts, status) == "errored", status + + +def test_verdict_distinguishes_no_policies_from_passed(): + """ + A run with nothing in scope is reported as such rather than as a clean bill of health -- the + most likely cause is a policy scoped to the wrong workflow group. + """ + assert render.verdict({}, "COMPLETED") == "no-policies" + + +def test_verdict_approval_required_is_not_an_error(): + """ + A run resting at APPROVAL_REQUIRED finished its evaluation; a human now has to act. Reporting + it as `errored` would blame the tool for a working evaluation -- and the poller now stops + there rather than spinning to its timeout. + """ + counts, _ = render.summarize(_results("APPROVAL_REQUIRED")) + + assert render.verdict(counts, "APPROVAL_REQUIRED") == "approval-required" + + +# --- rendering --------------------------------------------------------------------------------- + + +def test_markdown_starts_with_the_marker_when_one_is_given(): + """ + The marker is opaque to this module -- GitHub's sticky-comment marker is one caller's choice -- + but when supplied it must be line 1, so the caller can find the document again. + """ + marker = "[//]: <> (tirith-comment, tag=envs-prod)" + body = render.render_markdown(_results(), "COMPLETED", "https://app.example/run", marker=marker) + + assert body.split("\n")[0] == marker + + +def test_markdown_has_no_marker_line_by_default(): + """This module is VCS-agnostic: nothing is prepended unless the caller asks for it.""" + body = render.render_markdown(_results(), "COMPLETED", "https://app.example/run") + + assert not body.startswith("[//]") + assert body.lstrip().startswith("## ") + + +def test_comment_includes_table_detail_and_run_link(): + body = render.render_markdown(_results(), "COMPLETED", "https://app.example/run") + + assert "| Policy | Rule | Resource |" in body + assert "`no-public-ingress`" in body + assert "`0.0.0.0/0` is contained in `cidr_blocks`" in body + assert "module.net.aws_security_group.web" in body + assert "https://app.example/run" in body + + +def test_comment_explains_an_errored_run(): + body = render.render_markdown({}, "ERRORED", "https://app.example/run") + + assert "could not evaluate" in body.lower() + assert "ERRORED" in body + + +def test_comment_truncates_below_the_github_limit_keeping_the_table(): + """ + GitHub rejects a body over 65536 characters with a 422. Detail sections go first; the summary + table is what a reviewer scans, so it must survive. + """ + results = { + f"policy-{i}": [ + { + "rule_name": f"rule-{i}", + "result": "FAIL", + "evaluations": { + "fails": [ + { + "id": f"check-{j}", + "result": [ + { + "message": "x" * 400, + "meta": {"address": f"aws_instance.i{j}"}, + } + ], + } + for j in range(20) + ] + }, + } + ] + for i in range(60) + } + + body = render.render_markdown(results, "COMPLETED", "https://app.example/run", limit=20000) + + assert len(body) <= 20000 + assert "| Policy | Rule | Resource |" in body, "the summary table must survive truncation" + assert "more finding" in body or "truncated" in body + + +def test_strip_marker_removes_it_for_targets_that_have_no_use_for_it(): + """A check-run summary, for instance: the marker only means something on an issue comment.""" + marker = "[//]: <> (tirith-comment, tag=default)" + body = render.render_markdown(_results(), "COMPLETED", "https://app.example/run", marker=marker) + + summary = render.strip_marker(body) + + assert "[//]: <>" not in summary + assert "no-public-ingress" in summary + + +def test_headline_reports_each_nonzero_bucket(): + counts = {"FAIL": 2, "WARN": 1, "APPROVAL_REQUIRED": 3, "PASS": 9, "SKIPPED": 1} + + assert render.headline(counts, "failed") == "Tirith — 2 failed, 3 need approval, 1 warned, 9 passed, 1 skipped"