scripts.factory_enforce

Reference › scripts › factory_enforce

The command-line interface for the SPEC-033 state-diff enforcement core (B1a, B1b and B1c, Phase 1). It uses the attestation store in attestation_store and the bootstrap in the enforcement package; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main takes a subcommand and dispatches to its handler. baseline records the baseline state. record appends an entry to the enforcement ledger, with its payload read from standard input or built from flags (_record_payload_from_stdin, _record_payload_from_flags). register-writer registers a permitted writer. bootstrap initializes the enforcement state. verify checks the current state against the ledger. attest writes an attestation, and verify-attestations checks the stored ones. The default locations of the ledger, the writer registry and the attestations are derived from the repository root (_default_ledger, _default_registry, _default_attestations). Each handler returns the process exit code.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/factory_enforce.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 state-diff enforcement core (B1a/B1b/B1c, Phase 1).

Subcommands
-----------
baseline         Print path->sha256 tree hashes for a commit.
record           Append a governed-write record (from flags or ``--stdin``
                 JSON) to the event ledger.
register-writer  Register a governed-writer identity — a LOOSENING
                 (ADR-0009: human-only, hence the mandatory ``--by``).
bootstrap        Write an explicit human bootstrap event covering every
                 tracked file at a commit (the first-commit / unregistered-
                 new-file registration path, B1b).
verify           Enforce the committed-state boundary between two commits;
                 exits 1 with one line per violation.
attest           Sign + store the in-toto/SLSA attestation for a ledger
                 record (ADR-0030, B20).
verify-attestations
                 Verify every stored signed attestation; exits 1 naming
                 each one that fails (a tamper is the point of this check).

Path defaults: ``--ledger`` -> ``<repo-root>/factory_ledger.jsonl``,
``--registry`` -> ``<repo-root>/factory_writer_registry.json``,
``--worktree`` -> ``str(repo_root)``, ``--repo-root`` -> the current
directory, ``--attestations`` -> ``<repo-root>/factory_attestations.jsonl``,
``--signer-key`` -> ``$FACTORY_ATTEST_KEY_PATH`` or
``<repo-root>/factory_attest_key`` (created with fresh random key material
on first use by ``attest`` if it does not already exist).

Examples
--------
    python scripts/factory_enforce.py baseline --commit HEAD
    python scripts/factory_enforce.py register-writer --name ruff \\
        --binary-hash <sha> --config-hash <sha> --by "mohamed"
    python scripts/factory_enforce.py record --ledger ledger.jsonl \\
        --name ruff --binary-hash <sha> --config-hash <sha> \\
        --worktree /repo --commit HEAD \\
        --materials-json '{}' --products-json '{"a.py": "<sha256>"}'
    python scripts/factory_enforce.py bootstrap --commit HEAD --by "mohamed"
    python scripts/factory_enforce.py verify --base <c1> --head <c2>
    python scripts/factory_enforce.py attest --record-id GWR-abc \\
        --builder-id "orchestrata-factory-gate"
    python scripts/factory_enforce.py verify-attestations
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors kg_schema.py /
# factory_dor_procedure.py).
_REPO_ROOT = Path(__file__).resolve().parent.parent
_REPO_SRC = _REPO_ROOT / "src"
if str(_REPO_SRC) not in sys.path:
    sys.path.insert(0, str(_REPO_SRC))


def _default_ledger(repo_root: Path) -> Path:
    return repo_root / "factory_ledger.jsonl"


def _default_registry(repo_root: Path) -> Path:
    return repo_root / "factory_writer_registry.json"


def _default_attestations(repo_root: Path) -> Path:
    return repo_root / "factory_attestations.jsonl"


def _default_key_path(repo_root: Path) -> Path:
    # CLI-only environment read, per project rules (library callers pass an
    # explicit path to HmacSigner/ensure_key_file).
    return Path(
        os.environ.get("FACTORY_ATTEST_KEY_PATH", str(repo_root / "factory_attest_key"))
    )


def _cmd_baseline(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.bootstrap import bootstrap_baseline

    repo_root = Path(args.repo_root)
    tree = bootstrap_baseline(repo_root, args.commit)
    for path in sorted(tree):
        print(f"{tree[path]}  {path}")
    return 0


def _record_payload_from_stdin(repo_root: Path) -> dict:
    payload = json.load(sys.stdin)
    return {
        "name": payload["name"],
        "binary_hash": payload["binary_hash"],
        "config_hash": payload["config_hash"],
        "materials": payload.get("materials", {}),
        "products": payload.get("products", {}),
        "worktree": payload.get("worktree") or str(repo_root),
        "commit": payload["commit"],
        "byproducts": payload.get("byproducts", {}),
    }


def _record_payload_from_flags(args: argparse.Namespace, repo_root: Path) -> dict:
    if not args.name or not args.binary_hash or not args.config_hash:
        raise SystemExit(
            "record: --name/--binary-hash/--config-hash are required "
            "unless --stdin is given"
        )
    return {
        "name": args.name,
        "binary_hash": args.binary_hash,
        "config_hash": args.config_hash,
        "materials": json.loads(args.materials_json),
        "products": json.loads(args.products_json),
        "worktree": args.worktree or str(repo_root),
        "commit": args.commit,
        "byproducts": json.loads(args.byproducts_json),
    }


def _cmd_record(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.ledger import EventLedger
    from orchestrata_ultimate.factory.enforcement.models import WriterIdentity

    repo_root = Path(args.repo_root)
    ledger_path = Path(args.ledger) if args.ledger else _default_ledger(repo_root)
    ledger = EventLedger(ledger_path)

    payload = (
        _record_payload_from_stdin(repo_root)
        if args.stdin
        else _record_payload_from_flags(args, repo_root)
    )
    if not payload["commit"]:
        raise SystemExit("record: --commit is required unless supplied via --stdin")

    record = ledger.record(
        writer=WriterIdentity(
            name=payload["name"],
            binary_hash=payload["binary_hash"],
            config_hash=payload["config_hash"],
        ),
        materials=payload["materials"],
        products=payload["products"],
        worktree=payload["worktree"],
        commit=payload["commit"],
        byproducts=payload["byproducts"],
    )
    print(record.record_id)
    return 0


def _cmd_register_writer(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.ledger import WriterRegistry
    from orchestrata_ultimate.factory.enforcement.models import WriterIdentity

    repo_root = Path(args.repo_root)
    registry_path = (
        Path(args.registry) if args.registry else _default_registry(repo_root)
    )
    registry = WriterRegistry(registry_path)
    identity = WriterIdentity(
        name=args.name, binary_hash=args.binary_hash, config_hash=args.config_hash
    )
    registry.register(identity, by=args.by)
    print(
        f"registered: name={identity.name} binary_hash={identity.binary_hash} "
        f"config_hash={identity.config_hash} by={args.by}"
    )
    return 0


def _cmd_bootstrap(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.bootstrap import (
        bootstrap_baseline,
        record_bootstrap,
    )
    from orchestrata_ultimate.factory.enforcement.ledger import EventLedger

    repo_root = Path(args.repo_root)
    ledger_path = Path(args.ledger) if args.ledger else _default_ledger(repo_root)
    ledger = EventLedger(ledger_path)
    worktree = args.worktree or str(repo_root)
    tree = bootstrap_baseline(repo_root, args.commit)
    record_bootstrap(
        ledger, worktree=worktree, commit=args.commit, paths=tree, by=args.by
    )
    print(
        f"bootstrapped {len(tree)} path(s) at commit={args.commit} "
        f"worktree={worktree}"
    )
    return 0


def _cmd_verify(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.enforce import enforce
    from orchestrata_ultimate.factory.enforcement.ledger import (
        EventLedger,
        WriterRegistry,
    )

    repo_root = Path(args.repo_root)
    ledger_path = Path(args.ledger) if args.ledger else _default_ledger(repo_root)
    registry_path = (
        Path(args.registry) if args.registry else _default_registry(repo_root)
    )
    ledger = EventLedger(ledger_path)
    registry = WriterRegistry(registry_path)
    worktree = args.worktree or str(repo_root)

    verdict = enforce(args.base, args.head, repo_root, ledger, registry, worktree)
    if verdict.passed:
        print("verify: PASS")
        return 0

    print("verify: FAIL")
    for violation in verdict.violations:
        print(f"  {violation.kind}: {violation.path} -- {violation.detail}")
    return 1


def _cmd_attest(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.attestation_store import (
        AttestationStore,
    )
    from orchestrata_ultimate.factory.enforcement.ledger import EventLedger
    from orchestrata_ultimate.factory.enforcement.signing import (
        HmacSigner,
        ensure_key_file,
    )

    repo_root = Path(args.repo_root)
    ledger_path = Path(args.ledger) if args.ledger else _default_ledger(repo_root)
    attestations_path = (
        Path(args.attestations)
        if args.attestations
        else _default_attestations(repo_root)
    )
    key_path = (
        Path(args.signer_key) if args.signer_key else _default_key_path(repo_root)
    )
    ensure_key_file(key_path)

    ledger = EventLedger(ledger_path)
    record = next((r for r in ledger.all() if r.record_id == args.record_id), None)
    if record is None:
        raise SystemExit(
            f"attest: no ledger record with record_id={args.record_id!r} "
            f"in {ledger_path}"
        )

    signer = HmacSigner(key_path)
    store = AttestationStore(attestations_path)
    attestation = store.attest(record, signer, builder_id=args.builder_id)
    print(f"key_id={attestation.key_id} signature={attestation.signature}")
    return 0


def _cmd_verify_attestations(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.attestation_store import (
        AttestationStore,
    )
    from orchestrata_ultimate.factory.enforcement.signing import HmacSigner

    repo_root = Path(args.repo_root)
    attestations_path = (
        Path(args.attestations)
        if args.attestations
        else _default_attestations(repo_root)
    )
    key_path = (
        Path(args.signer_key) if args.signer_key else _default_key_path(repo_root)
    )
    if not key_path.exists():
        raise SystemExit(f"verify-attestations: signer key not found at {key_path}")

    signer = HmacSigner(key_path)
    store = AttestationStore(attestations_path)

    failures = []
    for i, attestation in enumerate(store.all()):
        if not store.verify(attestation, signer):
            failures.append(
                f"[{i}] name={attestation.link.name!r} "
                f"signed_at={attestation.signed_at!r}"
            )

    if failures:
        print("verify-attestations: FAIL")
        for failure in failures:
            print(f"  {failure}")
        return 1

    print("verify-attestations: PASS")
    return 0


def _add_baseline_subparser(sub: argparse._SubParsersAction) -> None:
    baseline_p = sub.add_parser(
        "baseline", help="Print path->sha256 tree hashes for a commit."
    )
    baseline_p.add_argument("--commit", required=True)
    baseline_p.add_argument("--repo-root", default=".")


def _add_record_subparser(sub: argparse._SubParsersAction) -> None:
    record_p = sub.add_parser(
        "record", help="Append a governed-write record to the event ledger."
    )
    record_p.add_argument("--ledger", default=None)
    record_p.add_argument("--repo-root", default=".")
    record_p.add_argument("--stdin", action="store_true")
    record_p.add_argument("--name", default=None)
    record_p.add_argument("--binary-hash", default=None)
    record_p.add_argument("--config-hash", default=None)
    record_p.add_argument("--worktree", default=None)
    record_p.add_argument("--commit", default=None)
    record_p.add_argument("--materials-json", default="{}")
    record_p.add_argument("--products-json", default="{}")
    record_p.add_argument("--byproducts-json", default="{}")


def _add_register_writer_subparser(sub: argparse._SubParsersAction) -> None:
    register_p = sub.add_parser(
        "register-writer", help="Register a governed-writer identity."
    )
    register_p.add_argument("--registry", default=None)
    register_p.add_argument("--repo-root", default=".")
    register_p.add_argument("--name", required=True)
    register_p.add_argument("--binary-hash", required=True)
    register_p.add_argument("--config-hash", required=True)
    register_p.add_argument("--by", required=True)


def _add_bootstrap_subparser(sub: argparse._SubParsersAction) -> None:
    bootstrap_p = sub.add_parser(
        "bootstrap", help="Write an explicit human bootstrap event for a commit."
    )
    bootstrap_p.add_argument("--commit", required=True)
    bootstrap_p.add_argument("--by", required=True)
    bootstrap_p.add_argument("--ledger", default=None)
    bootstrap_p.add_argument("--repo-root", default=".")
    bootstrap_p.add_argument("--worktree", default=None)


def _add_verify_subparser(sub: argparse._SubParsersAction) -> None:
    verify_p = sub.add_parser(
        "verify", help="Enforce the committed-state boundary between two commits."
    )
    verify_p.add_argument("--base", required=True)
    verify_p.add_argument("--head", required=True)
    verify_p.add_argument("--ledger", default=None)
    verify_p.add_argument("--registry", default=None)
    verify_p.add_argument("--worktree", default=None)
    verify_p.add_argument("--repo-root", default=".")


def _add_attest_subparser(sub: argparse._SubParsersAction) -> None:
    attest_p = sub.add_parser(
        "attest",
        help="Sign + store the in-toto/SLSA attestation for a ledger record.",
    )
    attest_p.add_argument("--ledger", default=None)
    attest_p.add_argument("--attestations", default=None)
    attest_p.add_argument("--record-id", required=True)
    attest_p.add_argument("--builder-id", required=True)
    attest_p.add_argument("--signer-key", default=None)
    attest_p.add_argument("--repo-root", default=".")


def _add_verify_attestations_subparser(sub: argparse._SubParsersAction) -> None:
    verify_p = sub.add_parser(
        "verify-attestations", help="Verify every stored signed attestation."
    )
    verify_p.add_argument("--attestations", default=None)
    verify_p.add_argument("--signer-key", default=None)
    verify_p.add_argument("--repo-root", default=".")


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_enforce",
        description="SPEC-033 state-diff enforcement core CLI (B1a/B1b/B1c/B20).",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    _add_baseline_subparser(sub)
    _add_record_subparser(sub)
    _add_register_writer_subparser(sub)
    _add_bootstrap_subparser(sub)
    _add_verify_subparser(sub)
    _add_attest_subparser(sub)
    _add_verify_attestations_subparser(sub)
    return parser


def main(argv: list[str] | None = None) -> int:
    """Parse ``argv`` and dispatch to the matching subcommand handler."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "baseline": _cmd_baseline,
        "record": _cmd_record,
        "register-writer": _cmd_register_writer,
        "bootstrap": _cmd_bootstrap,
        "verify": _cmd_verify,
        "attest": _cmd_attest,
        "verify-attestations": _cmd_verify_attestations,
    }
    handler = dispatch[args.command]
    return handler(args)


if __name__ == "__main__":
    sys.exit(main())