tegwick
ce1790f267
Write the selected field to a mode-0600 temp file, inject FIELD_FILE for the child only, then overwrite and unlink on every exit path. The value is not copied into the child environment. Assistant: grok Assistant-Session: 01a05f07-ae72-7781-9fcb-19efd61add00
963 lines
36 KiB
Python
963 lines
36 KiB
Python
"""secrets-engine command-line interface.
|
|
|
|
Command surface (FR7):
|
|
catalog list
|
|
catalog show <catalog-id>
|
|
decision inspect <decision-or-ccr-id>
|
|
plan <decision-or-ref> --stage <stage>
|
|
apply <decision-or-ref> --stage <stage> [--dry-run] [--auth auto] [--bootstrap-token-file F]
|
|
provision <catalog-id> --stage <stage> (--from-file F | --generate) --field NAME
|
|
verify <catalog-id> [--positive] [--negative] [--field NAME] [--negative-token-file F]
|
|
handoff <catalog-id> --stage <stage> --role-id-file F --secret-id-file F
|
|
wrap <catalog-id> --out F [--ttl 15m]
|
|
exec --catalog <catalog-id> [--field NAME] [--mode auto|npm-config|exec-env|exec-file] -- CMD...
|
|
route <catalog-id> [--json]
|
|
revoke <catalog-id>
|
|
session revoke --accessor-file F [--stage stage]
|
|
lifecycle suspend|deactivate|destroy <catalog-id>
|
|
audit <catalog-id> [--json]
|
|
secret-use snapshot [--catalog-id ID] [--json]
|
|
evidence heartbeat|drain|classify
|
|
|
|
Every privileged action is decision-gated and writes non-secret evidence.
|
|
`plan` and `apply --dry-run` never mutate OpenBao.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from secrets_engine import __version__
|
|
from secrets_engine.apply import apply_plan
|
|
from secrets_engine.approval_consume import (
|
|
require_production_consume,
|
|
resolve_consume_binding,
|
|
)
|
|
from secrets_engine.catalog import get_entry, load_catalog
|
|
from secrets_engine.config import Config, repo_root
|
|
from secrets_engine.decisions import require_approved, resolve_decision
|
|
from secrets_engine.errors import DecisionError, SecretsEngineError
|
|
from secrets_engine.evidence import EvidenceWriter, PrivilegedActionEvidence
|
|
from secrets_engine.engine_auth import login_service_jwt, select_engine_auth
|
|
from secrets_engine.pep_stance import apply_unreachable_engine_stance, with_decision
|
|
from secrets_engine.openbao import OpenBaoClient, accessor_fingerprint, read_strict_token_file
|
|
from secrets_engine.plan import build_plan
|
|
from secrets_engine.provision import provision_from_file, provision_generated
|
|
from secrets_engine.routing import route_lane
|
|
from secrets_engine.verify import run_verification
|
|
|
|
|
|
def _resolve_lane_and_decision(cfg: Config, ref: str, stage: str):
|
|
"""Map a decision/ccr/lane ref to a catalog entry + resolved decision.
|
|
|
|
For the MVP the catalog's approval.decision_ref ties a lane to its decision,
|
|
and a lane may be addressed directly by catalog id. We match `ref` against
|
|
both catalog ids and decision_refs.
|
|
"""
|
|
entries = load_catalog(cfg.catalog_dir)
|
|
# direct catalog id
|
|
if ref in entries:
|
|
return entries[ref]
|
|
# by decision_ref
|
|
for entry in entries.values():
|
|
if entry.approval.get("decision_ref") == ref:
|
|
return entry
|
|
from secrets_engine.errors import CatalogError
|
|
|
|
raise CatalogError(
|
|
f"no lane matches '{ref}' (by catalog id or decision_ref); "
|
|
f"known lanes: {', '.join(sorted(entries)) or 'none'}"
|
|
)
|
|
|
|
|
|
def _writer(cfg: Config) -> EvidenceWriter:
|
|
return EvidenceWriter(evidence_dir=cfg.evidence_dir, hub_url=cfg.hub_url, topic_id=cfg.topic_id)
|
|
|
|
|
|
def _privileged_evidence(
|
|
cfg: Config,
|
|
entry,
|
|
action: str,
|
|
*,
|
|
detail: dict[str, object] | None = None,
|
|
) -> PrivilegedActionEvidence:
|
|
return PrivilegedActionEvidence(
|
|
writer=_writer(cfg),
|
|
action=action,
|
|
catalog_id=entry.id,
|
|
stage=entry.stage,
|
|
decision_ref=entry.approval.get("decision_ref", ""),
|
|
approval_required=entry.approval_required(),
|
|
detail=detail or {},
|
|
)
|
|
|
|
|
|
def _require_lane_approval(
|
|
cfg: Config,
|
|
entry,
|
|
action: str = "",
|
|
evidence: PrivilegedActionEvidence | None = None,
|
|
):
|
|
"""Apply published PEP stance, resolve lane approval, then CAS-consume.
|
|
|
|
Production ``fail_closed`` is read from ``pep-stance.yaml``. The durable
|
|
access-engine decision record is not served yet, so that row refuses live
|
|
production work. If that residue is ever bypassed, GH-DEC-2026-003 still
|
|
requires a successful approval-engine consume before any OpenBao call.
|
|
The three-factor unsafe-demo exception is not a stance row and is not a
|
|
consume path. Build/test ``fail_open`` still requires the existing
|
|
lane-approval check — a tracked gap until SECRETS-WP-0008-T02.
|
|
"""
|
|
stance = apply_unreachable_engine_stance(cfg, entry, action or "unknown")
|
|
if evidence is not None:
|
|
evidence.mark_stance(stance)
|
|
decision = None
|
|
if entry.approval_required():
|
|
decision = resolve_decision(
|
|
hub_url=cfg.hub_url,
|
|
repo_root=repo_root(),
|
|
decision_ref=entry.approval.get("decision_ref", entry.id),
|
|
)
|
|
require_approved(entry, decision)
|
|
if evidence is not None:
|
|
evidence.mark_stance(with_decision(stance, decision))
|
|
require_production_consume(
|
|
cfg,
|
|
entry,
|
|
binding=resolve_consume_binding(cfg, entry, action or "unknown", decision),
|
|
evidence=evidence,
|
|
)
|
|
return decision
|
|
|
|
|
|
@contextmanager
|
|
def _open_backend(cfg: Config, args, evidence: PrivilegedActionEvidence | None = None):
|
|
"""Yield an OpenBao client for one named provider. JWT sessions self-revoke."""
|
|
selection = select_engine_auth(cfg, args)
|
|
if evidence is not None:
|
|
evidence.detail["auth_provider"] = selection.provider
|
|
if selection.break_glass:
|
|
evidence.detail["auth_break_glass"] = True
|
|
if selection.provider == "service-jwt":
|
|
session = login_service_jwt(cfg)
|
|
try:
|
|
if evidence is not None:
|
|
evidence.detail["auth_session_handle"] = session.accessor_fingerprint
|
|
yield session.client
|
|
finally:
|
|
try:
|
|
session.close()
|
|
finally:
|
|
if evidence is not None:
|
|
evidence.detail["auth_revocation_attempted"] = (
|
|
session.revocation_attempted
|
|
)
|
|
evidence.detail["auth_revocation_succeeded"] = (
|
|
session.revocation_succeeded
|
|
)
|
|
return
|
|
yield OpenBaoClient.resolve(
|
|
cfg.bao_addr, bootstrap_token_file=selection.bootstrap_token_file
|
|
)
|
|
|
|
|
|
# -- command handlers ------------------------------------------------------
|
|
|
|
|
|
def cmd_catalog_list(cfg: Config, args) -> int:
|
|
entries = load_catalog(cfg.catalog_dir)
|
|
if not entries:
|
|
print("(no catalog entries)")
|
|
return 0
|
|
for e in entries.values():
|
|
print(
|
|
f"{e.id:32s} kind={e.kind:15s} stage={e.stage:5s} "
|
|
f"owner={e.owner:18s} {e.mount}/{e.path}"
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_catalog_show(cfg: Config, args) -> int:
|
|
e = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
print(f"id: {e.id}")
|
|
print(f"kind: {e.kind}")
|
|
print(f"owner: {e.owner}")
|
|
print(f"stage: {e.stage}")
|
|
if e.kind == "auth-capability":
|
|
print(f"openbao: mount={e.mount} allowed={sorted(e.auth_allowed_paths)}")
|
|
print(f"approle: {e.role_name} policy={e.policy_name}")
|
|
else:
|
|
print(
|
|
f"openbao: {e.mount}/{e.path} fields={e.fields} "
|
|
f"mount_management={e.mount_management}"
|
|
)
|
|
print(
|
|
f"delivery auth: {e.delivery_auth_method}/"
|
|
f"{e.delivery_auth_management} role={e.role_name if e.has_delivery_auth else '-'}"
|
|
)
|
|
print(f"workload: {e.workload_delivery}")
|
|
print(f"consumers: {[c['name'] for c in e.consumers]}")
|
|
print(f"delivery: {e.delivery_modes}")
|
|
print(f"approval: {e.approval.get('model')} ref={e.approval.get('decision_ref','')}")
|
|
print(f"verification: {e.verification}")
|
|
print(f"rotation: {e.rotation}")
|
|
print(f"deactivation: {e.deactivation}")
|
|
if e.risk:
|
|
print(f"risk: {e.risk}")
|
|
print(f"description: {e.description.strip()}")
|
|
return 0
|
|
|
|
|
|
def cmd_decision_inspect(cfg: Config, args) -> int:
|
|
d = resolve_decision(hub_url=cfg.hub_url, repo_root=repo_root(), decision_ref=args.ref)
|
|
print(f"decision: {d.id}")
|
|
print(f"title: {d.title}")
|
|
print(f"status: {d.status} ({'APPROVED' if d.is_approved() else 'NOT approved'})")
|
|
print(f"source: {d.source}")
|
|
if d.superseded_by:
|
|
print(f"superseded: {d.superseded_by}")
|
|
if d.review_url:
|
|
print(f"review: {d.review_url}")
|
|
_writer(cfg).record(
|
|
"decision-inspect", result=d.status, decision_id=d.id, hub=False
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_plan(cfg: Config, args) -> int:
|
|
entry = _resolve_lane_and_decision(cfg, args.ref, args.stage)
|
|
decision = None
|
|
if entry.approval_required():
|
|
try:
|
|
decision = resolve_decision(
|
|
hub_url=cfg.hub_url, repo_root=repo_root(),
|
|
decision_ref=entry.approval.get("decision_ref", args.ref),
|
|
)
|
|
except DecisionError:
|
|
decision = None
|
|
plan = build_plan(entry, args.stage, decision_id=decision.id if decision else "")
|
|
print(plan.render())
|
|
_writer(cfg).record(
|
|
"plan", result="rendered", catalog_id=entry.id, stage=args.stage,
|
|
decision_id=decision.id if decision else "",
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_apply(cfg: Config, args) -> int:
|
|
entry = _resolve_lane_and_decision(cfg, args.ref, args.stage)
|
|
if args.dry_run:
|
|
decision = None
|
|
if entry.approval_required():
|
|
try:
|
|
decision = resolve_decision(
|
|
hub_url=cfg.hub_url,
|
|
repo_root=repo_root(),
|
|
decision_ref=entry.approval.get("decision_ref", args.ref),
|
|
)
|
|
except DecisionError:
|
|
decision = None
|
|
plan = build_plan(
|
|
entry, args.stage, decision_id=decision.id if decision else ""
|
|
)
|
|
print(plan.render())
|
|
print("\n(dry-run: no OpenBao mutation performed)")
|
|
_writer(cfg).record(
|
|
"apply",
|
|
result="dry-run",
|
|
catalog_id=entry.id,
|
|
stage=args.stage,
|
|
decision_id=decision.id if decision else "",
|
|
)
|
|
return 0
|
|
|
|
with _privileged_evidence(cfg, entry, "apply") as evidence:
|
|
decision = _require_lane_approval(cfg, entry, "apply", evidence)
|
|
evidence.mark_approved(decision)
|
|
plan = build_plan(
|
|
entry, args.stage, decision_id=decision.id if decision else ""
|
|
)
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
result = apply_plan(client, entry, plan)
|
|
print(result.render())
|
|
evidence.finish(
|
|
"applied",
|
|
detail={"applied": result.applied, "skipped": result.skipped},
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_provision(cfg: Config, args) -> int:
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
field = args.field or (entry.fields[0] if entry.fields else "")
|
|
with _privileged_evidence(
|
|
cfg, entry, "provision", detail={"field": field}
|
|
) as evidence:
|
|
if args.stage != entry.stage:
|
|
from secrets_engine.errors import ProvisioningError
|
|
|
|
raise ProvisioningError(
|
|
f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'"
|
|
)
|
|
decision = _require_lane_approval(cfg, entry, "provision", evidence)
|
|
evidence.mark_approved(decision)
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
if args.generate:
|
|
f = provision_generated(client, entry, field)
|
|
mode = "generated"
|
|
else:
|
|
f = provision_from_file(client, entry, field, Path(args.from_file))
|
|
mode = "from-file"
|
|
print(
|
|
f"provisioned lane '{entry.id}' field '{f}' ({mode}) — value not displayed"
|
|
)
|
|
evidence.finish(mode, detail={"field": f})
|
|
return 0
|
|
|
|
|
|
def cmd_verify(cfg: Config, args) -> int:
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
fields = [args.field] if args.field else list(entry.fields)
|
|
positive = args.positive or not args.negative
|
|
negative = args.negative or not args.positive
|
|
with _privileged_evidence(
|
|
cfg,
|
|
entry,
|
|
"verify",
|
|
detail={
|
|
"fields": fields,
|
|
"positive_requested": positive,
|
|
"negative_requested": negative,
|
|
},
|
|
) as evidence:
|
|
decision = _require_lane_approval(cfg, entry, "verify", evidence)
|
|
evidence.mark_approved(decision)
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
if entry.stores_kv_value() and not fields:
|
|
from secrets_engine.errors import VerificationError
|
|
|
|
raise VerificationError(f"lane '{entry.id}' has no field to verify")
|
|
unrelated_token = None
|
|
if entry.stores_kv_value() and negative and args.negative_token_file:
|
|
unrelated_token = read_strict_token_file(
|
|
Path(args.negative_token_file),
|
|
purpose="negative verification token",
|
|
)
|
|
if entry.stores_kv_value():
|
|
results = []
|
|
if positive:
|
|
for field in fields:
|
|
results.extend(
|
|
run_verification(
|
|
client, entry, field, positive=True, negative=False
|
|
)
|
|
)
|
|
if negative:
|
|
results.extend(
|
|
run_verification(
|
|
client,
|
|
entry,
|
|
fields[0],
|
|
positive=False,
|
|
negative=True,
|
|
unrelated_token=unrelated_token,
|
|
)
|
|
)
|
|
else:
|
|
results = run_verification(
|
|
client, entry, "", positive=positive, negative=negative
|
|
)
|
|
rc = 0
|
|
for result in results:
|
|
print(result.render())
|
|
if not result.passed:
|
|
rc = 7
|
|
evidence.writer.record(
|
|
"verify-check",
|
|
result=f"{result.check}:{'pass' if result.passed else 'fail'}",
|
|
catalog_id=entry.id,
|
|
stage=entry.stage,
|
|
decision_id=evidence.decision_id,
|
|
detail=result.detail,
|
|
)
|
|
evidence.finish(
|
|
"pass" if rc == 0 else "verification-failed",
|
|
detail={"check_count": len(results)},
|
|
)
|
|
return rc
|
|
|
|
|
|
def cmd_handoff(cfg: Config, args) -> int:
|
|
from secrets_engine.errors import ProvisioningError
|
|
from secrets_engine.handoff import write_approle_handoff
|
|
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
with _privileged_evidence(cfg, entry, "handoff") as evidence:
|
|
if args.stage != entry.stage:
|
|
raise ProvisioningError(
|
|
f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'"
|
|
)
|
|
if entry.kind != "auth-capability":
|
|
raise ProvisioningError(
|
|
f"lane '{entry.id}' is {entry.kind}; handoff needs auth-capability"
|
|
)
|
|
decision = _require_lane_approval(cfg, entry, "handoff", evidence)
|
|
evidence.mark_approved(decision)
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
result = write_approle_handoff(
|
|
client,
|
|
entry,
|
|
role_id_file=Path(args.role_id_file),
|
|
secret_id_file=Path(args.secret_id_file),
|
|
)
|
|
print(
|
|
f"wrote AppRole handoff material for lane '{entry.id}' — "
|
|
"secret_id not displayed"
|
|
)
|
|
print(f" role: {result.role_name}")
|
|
print(f" role_id_file: {result.role_id_file}")
|
|
print(f" secret_id_file: {result.secret_id_file}")
|
|
print(f" token_ttl: {result.token_ttl}")
|
|
print(f" secret_id_ttl: {result.secret_id_ttl}")
|
|
evidence.finish(
|
|
"secret-id-written",
|
|
detail={
|
|
"role": result.role_name,
|
|
"role_id_file": result.role_id_file,
|
|
"secret_id_file": result.secret_id_file,
|
|
"token_ttl": result.token_ttl,
|
|
"secret_id_ttl": result.secret_id_ttl,
|
|
"secret_id_num_uses": result.secret_id_num_uses,
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_wrap(cfg: Config, args) -> int:
|
|
from secrets_engine.wrap import write_wrapped_handoff
|
|
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
with _privileged_evidence(
|
|
cfg, entry, "wrap", detail={"ttl": args.ttl, "out_file": args.out}
|
|
) as evidence:
|
|
decision = _require_lane_approval(cfg, entry, "wrap", evidence)
|
|
evidence.mark_approved(decision)
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
result = write_wrapped_handoff(
|
|
client, entry, out_file=Path(args.out), ttl=args.ttl
|
|
)
|
|
print(
|
|
f"wrote wrap token for lane '{result.catalog_id}' — token not displayed"
|
|
)
|
|
print(f" out: {result.out_file}")
|
|
print(f" ttl: {result.ttl}")
|
|
print(f" handle: {result.wrap_handle or '-'}")
|
|
evidence.finish(
|
|
"wrap-token-written",
|
|
detail={
|
|
"out_file": result.out_file,
|
|
"ttl": result.ttl,
|
|
"wrap_handle": result.wrap_handle,
|
|
"creation_path": result.creation_path,
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_exec(cfg: Config, args) -> int:
|
|
from secrets_engine.exec_delivery import exec_with_secret
|
|
entry = get_entry(cfg.catalog_dir, args.catalog)
|
|
field = args.field or (entry.fields[0] if entry.fields else "")
|
|
session_detail: dict[str, object] = {}
|
|
command_name = args.command[0] if args.command else ""
|
|
with _privileged_evidence(
|
|
cfg,
|
|
entry,
|
|
"exec",
|
|
detail={
|
|
"command": command_name,
|
|
"mode": args.mode,
|
|
"field": field,
|
|
"session": session_detail,
|
|
},
|
|
) as evidence:
|
|
# require approval + readiness before running.
|
|
decision = _require_lane_approval(cfg, entry, "exec", evidence)
|
|
evidence.mark_approved(decision)
|
|
if not args.command:
|
|
from secrets_engine.errors import DeliveryError
|
|
|
|
raise DeliveryError("no command after '--'")
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
rc = exec_with_secret(
|
|
client,
|
|
entry,
|
|
field,
|
|
args.command,
|
|
mode=args.mode,
|
|
session_evidence=session_detail,
|
|
)
|
|
evidence.finish(f"exit-{rc}")
|
|
return rc
|
|
|
|
|
|
def cmd_session_revoke(cfg: Config, args) -> int:
|
|
"""Revoke an already-issued token by accessor. Never print the accessor."""
|
|
stance_entry = SimpleNamespace(stage=args.stage, approval={"model": "bootstrap-only"})
|
|
evidence = PrivilegedActionEvidence(
|
|
writer=_writer(cfg),
|
|
action="session-revoke",
|
|
catalog_id="",
|
|
stage=args.stage,
|
|
approval_required=False,
|
|
)
|
|
with evidence:
|
|
stance = apply_unreachable_engine_stance(cfg, stance_entry, "session-revoke")
|
|
evidence.mark_stance(stance)
|
|
accessor = read_strict_token_file(
|
|
Path(args.accessor_file), purpose="token accessor"
|
|
)
|
|
fingerprint = accessor_fingerprint(accessor)
|
|
evidence.detail["session_handle"] = fingerprint
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
client.revoke_accessor(accessor)
|
|
del accessor
|
|
print(f"revoked session handle {fingerprint}")
|
|
evidence.finish("revoked", detail={"session_handle": fingerprint})
|
|
return 0
|
|
|
|
|
|
def cmd_policy_publication(cfg: Config, args) -> int:
|
|
from secrets_engine.publication_policy import PublicationPolicy, resolve
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
npm = entry.npm
|
|
if not npm:
|
|
from secrets_engine.errors import PolicyGuardError
|
|
raise PolicyGuardError(f"lane '{entry.id}' has no npm delivery config")
|
|
policy = PublicationPolicy.load(cfg.policy_dir)
|
|
res = resolve(
|
|
policy,
|
|
org=entry.org, repo=entry.repo, npm_scope=npm.get("scope", ""),
|
|
package_maturity=npm.get("maturity", "maturity-build"),
|
|
token_env_override=npm.get("token_env", ""),
|
|
)
|
|
print(f"lane: {entry.id} ({entry.owner})")
|
|
print(f"netkingdom maturity: {policy.netkingdom_maturity} "
|
|
f"(production_grade={policy.production_grade})")
|
|
print(res.render())
|
|
return 0
|
|
|
|
|
|
def cmd_route(cfg: Config, args) -> int:
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
client = OpenBaoClient.resolve(cfg.bao_addr)
|
|
result = route_lane(entry, hub_url=cfg.hub_url, repo_root=repo_root(), client=client)
|
|
if args.json:
|
|
import json
|
|
print(json.dumps(result.to_json(), indent=2))
|
|
else:
|
|
print(
|
|
f"lane: {result.catalog_id} "
|
|
f"(kind={result.kind}, owner={result.owner}, stage={result.stage})"
|
|
)
|
|
print(f"decision: {result.decision_status} ref={result.decision_ref}")
|
|
material_label = "handoff_ready" if result.kind == "auth-capability" else "value_present"
|
|
print(f"applied: {result.metadata_applied} {material_label}: {result.value_present}")
|
|
print(f"ready: {result.ready}")
|
|
if result.missing:
|
|
print(f"missing: {result.missing}")
|
|
print(f"next: {result.next_command}")
|
|
return 0
|
|
|
|
|
|
def cmd_revoke(cfg: Config, args) -> int:
|
|
from secrets_engine.lifecycle import (
|
|
apply_lifecycle_plan,
|
|
build_native_deactivation_plan,
|
|
)
|
|
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
plan = build_native_deactivation_plan(entry)
|
|
if args.dry_run:
|
|
print(plan.render())
|
|
print("\n(dry-run: no OpenBao mutation performed)")
|
|
return 0
|
|
with _privileged_evidence(
|
|
cfg, entry, "revoke", detail={"operation": plan.operation}
|
|
) as evidence:
|
|
decision = _require_lane_approval(cfg, entry, "deactivate", evidence)
|
|
evidence.mark_approved(decision)
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
result = apply_lifecycle_plan(client, plan)
|
|
print(plan.render())
|
|
print(result.render())
|
|
evidence.finish(
|
|
"native-access-deactivated",
|
|
detail={
|
|
"applied": list(result.applied),
|
|
"preserved": list(result.preserved),
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_lifecycle(cfg: Config, args) -> int:
|
|
from secrets_engine.lifecycle import (
|
|
apply_lifecycle_plan,
|
|
build_lifecycle_plan,
|
|
require_destroy_confirmation,
|
|
)
|
|
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
plan = build_lifecycle_plan(entry, args.operation)
|
|
if args.dry_run:
|
|
print(plan.render())
|
|
print("\n(dry-run: no OpenBao mutation performed)")
|
|
return 0
|
|
with _privileged_evidence(
|
|
cfg,
|
|
entry,
|
|
f"lifecycle-{args.operation}",
|
|
detail={"operation": plan.operation},
|
|
) as evidence:
|
|
if args.operation == "destroy":
|
|
require_destroy_confirmation(entry, args.confirm_destroy)
|
|
# A lane-level approval is not an authorization to erase custody.
|
|
# Keep the destructive live path closed until T04 supplies a
|
|
# canonical exact-action and dual-control-capable decision contract.
|
|
from secrets_engine.errors import PolicyGuardError
|
|
|
|
raise PolicyGuardError(
|
|
"live destroy is disabled until an exact-action destruction "
|
|
"approval contract is available; use --dry-run to inspect targets"
|
|
)
|
|
decision = _require_lane_approval(cfg, entry, args.operation, evidence)
|
|
evidence.mark_approved(decision)
|
|
with _open_backend(cfg, args, evidence) as client:
|
|
result = apply_lifecycle_plan(client, plan)
|
|
print(plan.render())
|
|
print(result.render())
|
|
evidence.finish(
|
|
"applied",
|
|
detail={
|
|
"applied": list(result.applied),
|
|
"preserved": list(result.preserved),
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_secret_use_snapshot(cfg: Config, args) -> int:
|
|
"""Read-only secret-use evidence snapshot. Never contacts OpenBao."""
|
|
import json
|
|
|
|
from secrets_engine.secret_use import snapshot
|
|
|
|
payload = snapshot(
|
|
cfg.catalog_dir,
|
|
cfg.evidence_dir,
|
|
catalog_id=getattr(args, "catalog_id", "") or "",
|
|
)
|
|
if args.json:
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
else:
|
|
print(
|
|
f"secret-use {payload['as_of']} lanes={len(payload['lanes'])} "
|
|
f"completeness_claimed={payload['completeness_claimed']} "
|
|
f"cadence={payload['cadence'].get('form')}/{payload['cadence'].get('interval')}"
|
|
)
|
|
for lane in payload["lanes"]:
|
|
ready = lane.get("ready")
|
|
ready_s = "unknown" if ready is None else str(ready).lower()
|
|
print(
|
|
f" {lane['catalog_id']:32s} stage={lane['stage']:5s} "
|
|
f"ready={ready_s} kind={lane.get('evidence_kind', '-')}"
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_audit(cfg: Config, args) -> int:
|
|
"""Summarize allowlisted non-secret evidence for one cataloged lane."""
|
|
import json
|
|
|
|
from secrets_engine.audit import summarize_lane_evidence
|
|
|
|
entry = get_entry(cfg.catalog_dir, args.catalog_id)
|
|
summary = summarize_lane_evidence(cfg.evidence_dir, entry.id)
|
|
if args.json:
|
|
print(json.dumps(summary.to_json(), indent=2, sort_keys=True))
|
|
else:
|
|
print(summary.render())
|
|
return 0
|
|
|
|
|
|
def cmd_evidence_heartbeat(cfg: Config, args) -> int:
|
|
"""Write a positive load-bearing heartbeat. Not a permission check."""
|
|
import json
|
|
|
|
from secrets_engine.evidence import write_heartbeat
|
|
|
|
record = write_heartbeat(_writer(cfg), stage=args.stage)
|
|
if args.json:
|
|
print(json.dumps(record, indent=2, sort_keys=True))
|
|
else:
|
|
print(
|
|
f"heartbeat {record['result']} queued={record.get('outbox_queued')} "
|
|
f"completeness_claimed={record.get('completeness_claimed')}"
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_evidence_drain(cfg: Config, args) -> int:
|
|
"""Best-effort drain of the local load-bearing outbox. Never a gate."""
|
|
import json
|
|
|
|
from secrets_engine.evidence import drain_outbox
|
|
|
|
result = drain_outbox(_writer(cfg), audit_core_url=args.audit_core_url)
|
|
if args.json:
|
|
print(json.dumps(result, indent=2, sort_keys=True))
|
|
else:
|
|
print(
|
|
f"outbox drain queued={result['queued']} "
|
|
f"delivered={result['delivered']} failed={result['failed']} "
|
|
f"skipped={result['skipped']}"
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_evidence_classify(cfg: Config, args) -> int:
|
|
import json
|
|
|
|
from secrets_engine.evidence_class import classify
|
|
|
|
classified = classify(args.action, args.stage)
|
|
payload = {
|
|
"action": classified.action,
|
|
"stage": classified.stage,
|
|
"kind": classified.kind,
|
|
"rule_id": classified.rule_id,
|
|
"queued_locally": classified.queued_locally,
|
|
"completeness_claimed": classified.completeness_claimed,
|
|
}
|
|
if args.json:
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
else:
|
|
print(
|
|
f"{classified.action}/{classified.stage}: {classified.kind} "
|
|
f"rule={classified.rule_id} queued={classified.queued_locally}"
|
|
)
|
|
return 0
|
|
|
|
|
|
# -- parser ----------------------------------------------------------------
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(prog="secrets-engine", description=__doc__.splitlines()[0])
|
|
p.add_argument("--version", action="version", version=f"secrets-engine {__version__}")
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
def add_token_arg(sp):
|
|
sp.add_argument("--bootstrap-token-file", default=None,
|
|
help="named break-glass OpenBao token file (mode 0600)")
|
|
sp.add_argument(
|
|
"--auth",
|
|
default="auto",
|
|
choices=("auto", "service-jwt", "bootstrap", "env"),
|
|
help="engine OpenBao auth provider; auto never falls back from service-jwt",
|
|
)
|
|
|
|
cat = sub.add_parser("catalog", help="catalog operations")
|
|
catsub = cat.add_subparsers(dest="subcmd", required=True)
|
|
catsub.add_parser("list", help="list catalog lanes").set_defaults(func=cmd_catalog_list)
|
|
cshow = catsub.add_parser("show", help="show a lane")
|
|
cshow.add_argument("catalog_id")
|
|
cshow.set_defaults(func=cmd_catalog_show)
|
|
|
|
dec = sub.add_parser("decision", help="decision operations")
|
|
decsub = dec.add_subparsers(dest="subcmd", required=True)
|
|
dinsp = decsub.add_parser("inspect", help="inspect a decision/CCR")
|
|
dinsp.add_argument("ref")
|
|
dinsp.set_defaults(func=cmd_decision_inspect)
|
|
|
|
pl = sub.add_parser("plan", help="render a guarded plan (no mutation)")
|
|
pl.add_argument("ref", help="decision/ccr id or catalog id")
|
|
pl.add_argument("--stage", required=True, choices=("build", "test", "prod"))
|
|
pl.set_defaults(func=cmd_plan)
|
|
|
|
ap = sub.add_parser("apply", help="apply approved metadata to OpenBao")
|
|
ap.add_argument("ref")
|
|
ap.add_argument("--stage", required=True, choices=("build", "test", "prod"))
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
add_token_arg(ap)
|
|
ap.set_defaults(func=cmd_apply)
|
|
|
|
pr = sub.add_parser("provision", help="provision a value without printing it")
|
|
pr.add_argument("catalog_id")
|
|
pr.add_argument("--stage", required=True, choices=("build", "test", "prod"))
|
|
pr.add_argument("--field", default=None)
|
|
g = pr.add_mutually_exclusive_group(required=True)
|
|
g.add_argument("--from-file", help="mode-0600 file holding the value, outside repos")
|
|
g.add_argument("--generate", action="store_true", help="generate (build/test only)")
|
|
add_token_arg(pr)
|
|
pr.set_defaults(func=cmd_provision)
|
|
|
|
ve = sub.add_parser("verify", help="positive/negative verification (no value printed)")
|
|
ve.add_argument("catalog_id")
|
|
ve.add_argument("--field", default=None)
|
|
ve.add_argument("--positive", action="store_true")
|
|
ve.add_argument("--negative", action="store_true")
|
|
ve.add_argument(
|
|
"--negative-token-file",
|
|
default=None,
|
|
help="mode-0600 out-of-repo token for a real unrelated identity",
|
|
)
|
|
add_token_arg(ve)
|
|
ve.set_defaults(func=cmd_verify)
|
|
|
|
ha = sub.add_parser("handoff", help="write AppRole role_id/secret_id handoff files")
|
|
ha.add_argument("catalog_id")
|
|
ha.add_argument("--stage", required=True, choices=("build", "test", "prod"))
|
|
ha.add_argument("--role-id-file", required=True)
|
|
ha.add_argument("--secret-id-file", required=True)
|
|
add_token_arg(ha)
|
|
ha.set_defaults(func=cmd_handoff)
|
|
|
|
wr = sub.add_parser(
|
|
"wrap",
|
|
help="write a single-use OpenBao wrap token to a file (never printed)",
|
|
)
|
|
wr.add_argument("catalog_id")
|
|
wr.add_argument("--out", required=True, help="mode-0600 wrap-token file outside Git")
|
|
wr.add_argument("--ttl", default="15m", help="wrap TTL, max 15m (default 15m)")
|
|
add_token_arg(wr)
|
|
wr.set_defaults(func=cmd_wrap)
|
|
|
|
ex = sub.add_parser("exec", help="run a command with the secret injected for the child only")
|
|
ex.add_argument("--catalog", required=True)
|
|
ex.add_argument("--field", default=None)
|
|
ex.add_argument("--mode", default="auto", choices=("auto", "npm-config", "exec-env", "exec-file"))
|
|
add_token_arg(ex)
|
|
ex.add_argument("command", nargs=argparse.REMAINDER,
|
|
help="command after '--'")
|
|
ex.set_defaults(func=cmd_exec)
|
|
|
|
po = sub.add_parser("policy", help="inspect secrets-engine policies")
|
|
posub = po.add_subparsers(dest="subcmd", required=True)
|
|
popub = posub.add_parser("publication", help="resolve a lane's publication scope + token env")
|
|
popub.add_argument("catalog_id")
|
|
popub.set_defaults(func=cmd_policy_publication)
|
|
|
|
ro = sub.add_parser("route", help="ops-warden routing pointer for a lane")
|
|
ro.add_argument("catalog_id")
|
|
ro.add_argument("--json", action="store_true")
|
|
ro.set_defaults(func=cmd_route)
|
|
|
|
rv = sub.add_parser(
|
|
"revoke",
|
|
help="deactivate native policy/AppRole access (preserves KV custody)",
|
|
)
|
|
rv.add_argument("catalog_id")
|
|
rv.add_argument("--dry-run", action="store_true")
|
|
add_token_arg(rv)
|
|
rv.set_defaults(func=cmd_revoke)
|
|
|
|
sess = sub.add_parser("session", help="issued-session operator commands")
|
|
sessub = sess.add_subparsers(dest="subcmd", required=True)
|
|
srev = sessub.add_parser(
|
|
"revoke",
|
|
help="revoke a token by a non-secret accessor the operator already holds",
|
|
)
|
|
srev.add_argument(
|
|
"--accessor-file",
|
|
required=True,
|
|
help="mode-0600 out-of-repo file containing the token accessor",
|
|
)
|
|
srev.add_argument("--stage", default="prod", choices=("build", "test", "prod"))
|
|
add_token_arg(srev)
|
|
srev.set_defaults(func=cmd_session_revoke)
|
|
|
|
lc = sub.add_parser("lifecycle", help="explicit lane lifecycle operations")
|
|
lcsub = lc.add_subparsers(dest="operation", required=True)
|
|
for operation, help_text in (
|
|
("suspend", "stop new native logins; preserve policy and KV custody"),
|
|
("deactivate", "remove native AppRole/policy; preserve KV custody"),
|
|
("destroy", "irreversibly delete KV metadata after deactivation"),
|
|
):
|
|
lp = lcsub.add_parser(operation, help=help_text)
|
|
lp.add_argument("catalog_id")
|
|
lp.add_argument("--dry-run", action="store_true")
|
|
if operation == "destroy":
|
|
lp.add_argument(
|
|
"--confirm-destroy",
|
|
default="",
|
|
help="exact catalog id required for live irreversible deletion",
|
|
)
|
|
add_token_arg(lp)
|
|
lp.set_defaults(func=cmd_lifecycle)
|
|
|
|
su = sub.add_parser("secret-use", help="read-only secret-use evidence surface")
|
|
susub = su.add_subparsers(dest="subcmd", required=True)
|
|
snap = susub.add_parser(
|
|
"snapshot",
|
|
help="non-secret lane metadata for kings-guard; never contacts OpenBao",
|
|
)
|
|
snap.add_argument("--catalog-id", default="", help="limit to one catalog id")
|
|
snap.add_argument("--json", action="store_true")
|
|
snap.set_defaults(func=cmd_secret_use_snapshot)
|
|
|
|
au = sub.add_parser("audit", help="summarize non-secret local lane evidence")
|
|
au.add_argument("catalog_id")
|
|
au.add_argument("--json", action="store_true")
|
|
au.set_defaults(func=cmd_audit)
|
|
|
|
ev = sub.add_parser("evidence", help="load-bearing evidence heartbeat and outbox")
|
|
evsub = ev.add_subparsers(dest="subcmd", required=True)
|
|
hb = evsub.add_parser(
|
|
"heartbeat",
|
|
help="emit a positive nothing-to-report claim (not a permission check)",
|
|
)
|
|
hb.add_argument("--stage", default="prod", choices=("build", "test", "prod"))
|
|
hb.add_argument("--json", action="store_true")
|
|
hb.set_defaults(func=cmd_evidence_heartbeat)
|
|
dr = evsub.add_parser(
|
|
"drain",
|
|
help="best-effort drain of the local outbox; never blocks a mutation",
|
|
)
|
|
dr.add_argument(
|
|
"--audit-core-url",
|
|
default="",
|
|
help="optional audit-core base URL; empty skips delivery and keeps files",
|
|
)
|
|
dr.add_argument("--json", action="store_true")
|
|
dr.set_defaults(func=cmd_evidence_drain)
|
|
cl = evsub.add_parser("classify", help="show the §9.6 class for an action/stage")
|
|
cl.add_argument("action")
|
|
cl.add_argument("--stage", required=True, choices=("build", "test", "prod"))
|
|
cl.add_argument("--json", action="store_true")
|
|
cl.set_defaults(func=cmd_evidence_classify)
|
|
|
|
return p
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
argv = list(sys.argv[1:] if argv is None else argv)
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
# `exec` REMAINDER includes a leading '--'; strip it.
|
|
if getattr(args, "command", None) and args.command and args.command[0] == "--":
|
|
args.command = args.command[1:]
|
|
cfg = Config.load()
|
|
try:
|
|
return args.func(cfg, args)
|
|
except SecretsEngineError as e:
|
|
print(f"error: {e}", file=sys.stderr)
|
|
return e.exit_code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|