Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
342 changes: 181 additions & 161 deletions README.md

Large diffs are not rendered by default.

13 changes: 13 additions & 0 deletions examples/demo.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
# Zero-credential demo. Run it as-is — no secrets, no cloud account:
#
# paramify run examples/demo.yaml
# paramify evidence evidence/run-*/demo_hello.json
#
# The demo_hello fetcher writes synthetic evidence, so this exercises the whole
# collect → envelope pipeline end to end. Delete once you've seen it work.

run:
output_dir: ./evidence

fetchers:
- use: demo_hello
4 changes: 4 additions & 0 deletions fetchers/_categories/demo.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
# Demo category — a single credential-free fetcher that emits synthetic
# evidence, so you can exercise the whole run → envelope pipeline with no cloud
# account. Not real evidence; safe to delete.
description: Synthetic evidence for trying the pipeline with no credentials.
61 changes: 61 additions & 0 deletions fetchers/demo/hello/fetcher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
#!/usr/bin/env python3
"""
Demo fetcher: synthetic MFA-policy evidence.

Writes a fixed, plausible-looking evidence payload so a newcomer can run the
whole pipeline — collect, envelope, inspect — without any credentials or network
access. This is NOT real evidence; it exists purely to demonstrate the shape of
a run. See `examples/demo.yaml`.
"""

import json
import logging
import os
from datetime import datetime, timezone
from pathlib import Path

logger = logging.getLogger("demo_hello")


def main():
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", "INFO"),
format="%(asctime)s %(levelname)s %(name)s %(message)s",
)

output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence"))
output_dir.mkdir(parents=True, exist_ok=True)

# A fixed, synthetic snapshot — the shape a real fetcher would collect,
# with obviously fake values.
evidence = {
"metadata": {
"source": "demo",
"note": "Synthetic evidence — not collected from any real system.",
"generated_at": datetime.now(timezone.utc).isoformat(),
},
"results": [
{
"policy": "require-phishing-resistant-mfa",
"enabled": True,
"applies_to": "all-users",
"enrolled_users": 42,
"total_users": 42,
},
{
"policy": "block-legacy-authentication",
"enabled": True,
"applies_to": "all-users",
},
],
}

output_path = output_dir / "demo_hello.json"
with open(output_path, "w") as f:
json.dump(evidence, f, indent=2)

logger.info("Synthetic evidence saved to %s", output_path)


if __name__ == "__main__":
main()
24 changes: 24 additions & 0 deletions fetchers/demo/hello/fetcher.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
name: demo_hello
version: 0.1.0
description: >
Emits synthetic MFA-policy evidence so you can run the full
collect → envelope pipeline with no credentials. Not real evidence.
category: demo

runtime:
type: python
entry: fetcher.py

output:
type: json
path: demo_hello.json

# The whole point of the demo: no secrets, no config, no targets.
secrets: []

evidence_set:
reference_id: EVD-DEMO-HELLO
name: Demo — Synthetic MFA Policy
instructions: >
Script: fetcher.py. Generates a fixed synthetic payload; requires no
credentials or network access. For trying the pipeline only.
110 changes: 110 additions & 0 deletions framework/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,9 @@
import importlib.util
import json
import os
import re
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
Expand Down Expand Up @@ -219,6 +222,113 @@ def ksi_coverage(root: Path) -> dict:
}


# --------------------------------------------------------------------------- #
# Doctor — preflight environment check
# --------------------------------------------------------------------------- #

# External CLIs a category's fetchers shell out to. Categories not listed here
# are pure-Python/HTTP fetchers that need no external tool. (AWS fetcher.sh files
# declare "Required tools: aws, jq"; k8s uses kubectl; checkov clones + scans.)
CATEGORY_TOOLS = {
"aws": ["aws", "jq"],
"k8s": ["kubectl"],
"checkov": ["checkov", "git"],
}

_ENV_REF = re.compile(r"\$\{env:([^}]+)\}")


def doctor(root: Path, manifest_path: Optional[Path] = None) -> dict:
"""Preflight check for running fetchers here.

Reports the Python version, whether the external CLIs the relevant categories
need are on PATH, and — if a manifest is given — which secret env vars it
references are actually set. Presentation-agnostic; the CLI and TUI render
this one model. `ok` is a go/no-go: Python must clear the floor, and when a
manifest is supplied its categories' tools must be present and its secret env
vars set. Without a manifest, missing tools are informational (you may run
only some categories), so `ok` reflects the Python check alone.
"""
py = sys.version_info
python = {
"version": f"{py.major}.{py.minor}.{py.micro}",
"required": "3.10",
"ok": (py.major, py.minor) >= (3, 10),
}

fetchers = discover_fetchers(root)
categories = sorted({f.category for f in fetchers.values() if f.category})
tools_required = manifest_path is not None

manifest_report = None
if manifest_path is not None:
data = read_manifest(manifest_path)
entries = (data.get("run") or {}).get("fetchers") or []
used_cats = sorted({
c for c in (
fetchers[e["use"]].category
for e in entries
if e.get("use") in fetchers
)
if c
})
if used_cats:
categories = used_cats

fetcher_reports = []
manifest_ok = True
for e in entries:
refs: set = set()
for v in (e.get("secrets") or {}).values():
refs |= set(_ENV_REF.findall(str(v)))
for t in e.get("targets") or []:
for v in (t.get("secrets") or {}).values():
refs |= set(_ENV_REF.findall(str(v)))
missing = sorted(r for r in refs if not os.environ.get(r))
manifest_ok = manifest_ok and not missing
fetcher_reports.append({
"use": e.get("use"),
"env_refs": sorted(refs),
"missing": missing,
"ok": not missing,
})
manifest_report = {
"path": str(manifest_path),
"fetchers": fetcher_reports,
"ok": manifest_ok,
}

seen: set = set()
tools = []
for cat in categories:
for tool in CATEGORY_TOOLS.get(cat, []):
if tool in seen:
continue
seen.add(tool)
resolved = shutil.which(tool)
tools.append({
"name": tool,
"categories": [c for c in categories if tool in CATEGORY_TOOLS.get(c, [])],
"present": resolved is not None,
"path": resolved,
})
tools.sort(key=lambda t: str(t["name"]))

ok = python["ok"]
if tools_required:
tools_ok = all(t["present"] for t in tools)
ok = ok and tools_ok and (manifest_report["ok"] if manifest_report else True)

return {
"python": python,
"categories": categories,
"tools": tools,
"tools_required": tools_required,
"manifest": manifest_report,
"ok": ok,
}


# --------------------------------------------------------------------------- #
# Manifest read / write
# --------------------------------------------------------------------------- #
Expand Down
44 changes: 44 additions & 0 deletions framework/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
paramify catalog [--json] # categories -> fetchers -> fields
paramify describe <fetcher> [--json]
paramify ksi [--json] # FedRAMP 20x KSI coverage
paramify doctor [manifest] [--json] # preflight: python, CLIs, secrets
paramify manifests [--json] # discovered run manifests
paramify runs [--output-dir DIR] [--json] # past runs under an output dir
paramify evidence <path> [--json] # read one evidence file
Expand Down Expand Up @@ -331,6 +332,49 @@ def ksi_cmd(json_out: bool = typer.Option(False, "--json", help="Emit JSON")):
typer.echo(f" {u['id']:14s} {', '.join(u['fetchers'])}")


@app.command("doctor")
def doctor_cmd(
manifest: Optional[str] = typer.Argument(
None, help="Manifest to check secret env vars for (optional)"
),
json_out: bool = typer.Option(False, "--json", help="Emit JSON"),
):
"""Preflight: Python version, required CLIs on PATH, and (with a manifest) secrets."""
root = api.find_repo_root()
rep = api.doctor(root, Path(manifest) if manifest else None)
if json_out:
typer.echo(json.dumps(rep, indent=2))
raise typer.Exit(0 if rep["ok"] else 1)

ok_mark, bad_mark = "✅", "❌"
p = rep["python"]
typer.echo(f"{ok_mark if p['ok'] else bad_mark} Python {p['version']} (need ≥ {p['required']})")

if rep["tools"]:
heading = "Required CLIs" if rep["tools_required"] else "CLIs for discovered categories"
typer.echo(f"\n{heading}:")
for t in rep["tools"]:
mark = ok_mark if t["present"] else bad_mark
where = t["path"] or "not found on PATH"
typer.echo(f" {mark} {t['name']:8s} {where} ({', '.join(t['categories'])})")
if not rep["tools_required"]:
typer.echo(" (informational — you only need the CLIs for categories you run)")

if rep["manifest"]:
m = rep["manifest"]
typer.echo(f"\nManifest secrets ({m['path']}):")
for fr in m["fetchers"]:
if not fr["env_refs"]:
typer.echo(f" {ok_mark} {fr['use']} (no secrets)")
elif fr["ok"]:
typer.echo(f" {ok_mark} {fr['use']} ({', '.join(fr['env_refs'])})")
else:
typer.echo(f" {bad_mark} {fr['use']} missing: {', '.join(fr['missing'])}")

typer.echo(f"\n{'All good.' if rep['ok'] else 'Issues found — see above.'}")
raise typer.Exit(0 if rep["ok"] else 1)


@app.command("manifests")
def manifests_cmd(json_out: bool = typer.Option(False, "--json", help="Emit JSON")):
"""List discovered run manifests (manifests/*.yaml + legacy manifest.yaml)."""
Expand Down
24 changes: 22 additions & 2 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,8 +47,8 @@ def _registered():


EXPECTED_TOP = {
"list", "catalog", "describe", "manifests", "runs", "evidence",
"validate", "run", "upload", "manifest", "tui",
"list", "catalog", "describe", "ksi", "doctor", "manifests", "runs",
"evidence", "validate", "run", "upload", "manifest", "tui",
}
EXPECTED_MANIFEST = {
"init", "new", "add", "remove", "set-config", "set-secret",
Expand All @@ -63,6 +63,26 @@ def test_all_expected_commands_registered():
assert EXPECTED_MANIFEST <= manifest, f"missing manifest subcommands: {EXPECTED_MANIFEST - manifest}"


def test_doctor_json_ok_without_manifest():
"""Without a manifest, doctor is a Python-version gate; tools are advisory."""
result = runner.invoke(app, ["doctor", "--json"])
assert result.exit_code == 0, result.output
rep = json.loads(result.output)
assert rep["python"]["ok"] is True
assert rep["ok"] is True
assert rep["tools_required"] is False
assert rep["manifest"] is None


def test_doctor_with_credential_free_demo_manifest():
"""The demo manifest has no secrets, so doctor passes with a manifest too."""
result = runner.invoke(app, ["doctor", "examples/demo.yaml", "--json"])
assert result.exit_code == 0, result.output
rep = json.loads(result.output)
assert rep["tools_required"] is True
assert rep["manifest"]["ok"] is True


# --------------------------------------------------------------------------- #
# Parity invariant: every api function the TUI calls maps to a CLI command.
# Derived from the TUI SOURCE so it can't drift into a tautology.
Expand Down
Loading
Loading