382 lines
13 KiB
Python
382 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""AIOS V2 Run Provenance 与最小权限 Grant 运维 CLI。
|
|
|
|
私钥只用于签发；授权命令只加载一个与 ``--path`` 对应的 Grant。所有输出均为
|
|
不含 Secret 的 JSON。旧 ``issue`` 仅为候选兼容，生产必须使用
|
|
``issue-workflow`` 与 ``issue-grant``。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
from .provenance import (
|
|
Ed25519PrivateKeySigner,
|
|
Ed25519PublicKeyVerifier,
|
|
ExecutionPath,
|
|
GitOperation,
|
|
ProvenanceIssuer,
|
|
SQLiteProvenanceStore,
|
|
)
|
|
from .provenance_grants import (
|
|
GrantIssuer,
|
|
GrantType,
|
|
GrantVerifier,
|
|
SQLiteGrantStore,
|
|
WorkflowProvenanceIssuer,
|
|
)
|
|
from .provenance_operations import (
|
|
OperationContext,
|
|
ProvenanceOperationGate,
|
|
context_from_environment,
|
|
load_request_file,
|
|
)
|
|
from .provenance_scoped import ScopedLeastPrivilegeProvenanceWriter
|
|
|
|
|
|
def _json_object(value: str | None, file_path: str | None) -> Mapping[str, Any]:
|
|
if value and file_path:
|
|
raise ValueError("request-json and request-file are mutually exclusive")
|
|
if file_path:
|
|
return load_request_file(file_path)
|
|
parsed = json.loads(value or "{}")
|
|
if not isinstance(parsed, Mapping):
|
|
raise ValueError("request JSON must be an object")
|
|
return dict(parsed)
|
|
|
|
|
|
def _csv(value: str | None) -> tuple[str, ...]:
|
|
return tuple(item.strip() for item in str(value or "").split(",") if item.strip())
|
|
|
|
|
|
def _write_json(path: str | None, payload: Mapping[str, Any]) -> None:
|
|
encoded = json.dumps(dict(payload), ensure_ascii=False, sort_keys=True, indent=2) + "\n"
|
|
if path:
|
|
target = Path(path).resolve()
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = target.with_suffix(f".tmp-{os.getpid()}")
|
|
temporary.write_text(encoded, encoding="utf-8")
|
|
os.chmod(temporary, 0o640)
|
|
os.replace(temporary, target)
|
|
print(encoded, end="")
|
|
|
|
|
|
def _parent_store(args: argparse.Namespace) -> SQLiteProvenanceStore:
|
|
return SQLiteProvenanceStore(args.store_file)
|
|
|
|
|
|
def _grant_store(args: argparse.Namespace) -> SQLiteGrantStore:
|
|
return SQLiteGrantStore(args.store_file)
|
|
|
|
|
|
def _context(args: argparse.Namespace) -> OperationContext:
|
|
if args.context_from_env:
|
|
return context_from_environment()
|
|
values = {
|
|
"provenance_id": args.provenance_id,
|
|
"run_id": args.run_id,
|
|
"tenant_id": args.tenant_id,
|
|
"task_id": args.task_id,
|
|
"agent_id": args.agent_id,
|
|
"principal_id": args.principal_id,
|
|
}
|
|
missing = [name for name, value in values.items() if not str(value or "").strip()]
|
|
if missing:
|
|
raise ValueError(f"missing context arguments: {missing}")
|
|
return OperationContext(**{name: str(value).strip() for name, value in values.items()})
|
|
|
|
|
|
def command_issue_legacy(args: argparse.Namespace) -> int:
|
|
store = _parent_store(args)
|
|
signer = Ed25519PrivateKeySigner.from_file(args.private_key_file, key_id=args.key_id)
|
|
provenance = ProvenanceIssuer(signer, store).issue(
|
|
run_id=args.run_id,
|
|
issuer=args.issuer,
|
|
tenant_id=args.tenant_id,
|
|
task_id=args.task_id,
|
|
agent_id=args.agent_id,
|
|
principal_id=args.principal_id,
|
|
allowed_paths=_csv(args.allowed_paths) or tuple(item.value for item in ExecutionPath),
|
|
allowed_git_operations=_csv(args.allowed_git_operations)
|
|
or tuple(item.value for item in GitOperation),
|
|
ttl_seconds=args.ttl_seconds,
|
|
policy_version=args.policy_version,
|
|
)
|
|
_write_json(
|
|
args.output,
|
|
{
|
|
"mode": "legacy-full-rights-candidate-only",
|
|
"provenance": provenance.as_dict(),
|
|
"status": store.status(provenance.provenance_id).value,
|
|
"secret_value_stored": False,
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def command_issue_workflow(args: argparse.Namespace) -> int:
|
|
store = _parent_store(args)
|
|
signer = Ed25519PrivateKeySigner.from_file(args.private_key_file, key_id=args.key_id)
|
|
workflow = WorkflowProvenanceIssuer(signer, store).issue(
|
|
run_id=args.run_id,
|
|
issuer=args.issuer,
|
|
tenant_id=args.tenant_id,
|
|
task_id=args.task_id,
|
|
agent_id=args.agent_id,
|
|
principal_id=args.principal_id,
|
|
ttl_seconds=args.ttl_seconds,
|
|
)
|
|
_write_json(
|
|
args.output,
|
|
{
|
|
"mode": "least-privilege-workflow",
|
|
"workflow_provenance": workflow.as_dict(),
|
|
"status": store.status(workflow.provenance_id).value,
|
|
"secret_value_stored": False,
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def command_issue_grant(args: argparse.Namespace) -> int:
|
|
parent_store = _parent_store(args)
|
|
grant_store = _grant_store(args)
|
|
signer = Ed25519PrivateKeySigner.from_file(args.private_key_file, key_id=args.key_id)
|
|
grant = GrantIssuer(signer, parent_store, grant_store).issue(
|
|
parent_provenance_id=args.parent_provenance_id,
|
|
grant_type=GrantType(args.grant_type),
|
|
ttl_seconds=args.ttl_seconds,
|
|
allowed_git_operations=_csv(args.allowed_git_operations),
|
|
)
|
|
_write_json(
|
|
args.output,
|
|
{
|
|
"grant": grant.as_dict(),
|
|
"status": grant_store.status(grant.grant_id).value,
|
|
"secret_value_stored": False,
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def _single_scope_writer(args: argparse.Namespace) -> ScopedLeastPrivilegeProvenanceWriter:
|
|
path = ExecutionPath(args.path).value
|
|
parent_store = _parent_store(args)
|
|
grant_store = _grant_store(args)
|
|
verifier = Ed25519PublicKeyVerifier.from_file(args.public_key_file, key_id=args.key_id)
|
|
trusted = GrantVerifier(parent_store, grant_store, {args.key_id: verifier})
|
|
return ScopedLeastPrivilegeProvenanceWriter(
|
|
trusted,
|
|
args.receipt_dir,
|
|
{path: args.grant_id},
|
|
required_paths=(path,),
|
|
writer_id=f"aios-v2-cli-{path}",
|
|
)
|
|
|
|
|
|
def command_authorize(args: argparse.Namespace) -> int:
|
|
writer = _single_scope_writer(args)
|
|
context = _context(args)
|
|
receipt = ProvenanceOperationGate(writer).authorize(
|
|
context,
|
|
execution_path=args.path,
|
|
operation=args.operation,
|
|
request=_json_object(args.request_json, args.request_file),
|
|
)
|
|
_write_json(args.output, {"authorization": receipt.as_dict()})
|
|
return 0
|
|
|
|
|
|
def command_revoke(args: argparse.Namespace) -> int:
|
|
store = _parent_store(args)
|
|
store.revoke(args.provenance_id, reason=args.reason, principal_id=args.principal_id)
|
|
_write_json(
|
|
args.output,
|
|
{"provenance_id": args.provenance_id, "status": store.status(args.provenance_id).value},
|
|
)
|
|
return 0
|
|
|
|
|
|
def command_status(args: argparse.Namespace) -> int:
|
|
store = _parent_store(args)
|
|
provenance = store.get(args.provenance_id)
|
|
grant_store = _grant_store(args)
|
|
_write_json(
|
|
args.output,
|
|
{
|
|
"provenance": provenance.as_dict(),
|
|
"status": store.status(args.provenance_id).value,
|
|
"authorization_count": store.authorization_count(args.provenance_id),
|
|
"events": store.events(args.provenance_id),
|
|
"grants": [item.as_dict() for item in grant_store.list_for_parent(args.provenance_id)],
|
|
},
|
|
)
|
|
return 0
|
|
|
|
|
|
def command_grant_revoke(args: argparse.Namespace) -> int:
|
|
store = _grant_store(args)
|
|
store.revoke(args.grant_id, reason=args.reason, principal_id=args.principal_id)
|
|
_write_json(
|
|
args.output,
|
|
{"grant_id": args.grant_id, "status": store.status(args.grant_id).value},
|
|
)
|
|
return 0
|
|
|
|
|
|
def command_grant_status(args: argparse.Namespace) -> int:
|
|
store = _grant_store(args)
|
|
grant = store.get(args.grant_id)
|
|
_write_json(
|
|
args.output,
|
|
{"grant": grant.as_dict(), "status": store.status(args.grant_id).value},
|
|
)
|
|
return 0
|
|
|
|
|
|
def add_common_store(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument(
|
|
"--store-file",
|
|
default=os.getenv(
|
|
"AIOS_PROVENANCE_STORE_FILE",
|
|
"/var/lib/aios-v2/run-provenance.sqlite3",
|
|
),
|
|
)
|
|
parser.add_argument("--output", default=None)
|
|
|
|
|
|
def add_signing_context(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--private-key-file", required=True)
|
|
parser.add_argument("--key-id", required=True)
|
|
|
|
|
|
def add_workflow_identity(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--run-id", required=True)
|
|
parser.add_argument("--issuer", required=True)
|
|
parser.add_argument("--tenant-id", required=True)
|
|
parser.add_argument("--task-id", required=True)
|
|
parser.add_argument("--agent-id", required=True)
|
|
parser.add_argument("--principal-id", required=True)
|
|
|
|
|
|
def add_context(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--context-from-env", action="store_true")
|
|
parser.add_argument("--provenance-id")
|
|
parser.add_argument("--run-id")
|
|
parser.add_argument("--tenant-id")
|
|
parser.add_argument("--task-id")
|
|
parser.add_argument("--agent-id")
|
|
parser.add_argument("--principal-id")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="AIOS V2 Run Provenance CLI")
|
|
subcommands = parser.add_subparsers(dest="command", required=True)
|
|
|
|
issue = subcommands.add_parser("issue", help="候选兼容：签发全权限 Provenance；生产禁止")
|
|
add_common_store(issue)
|
|
add_signing_context(issue)
|
|
add_workflow_identity(issue)
|
|
issue.add_argument("--allowed-paths")
|
|
issue.add_argument("--allowed-git-operations")
|
|
issue.add_argument("--ttl-seconds", type=int, default=3600)
|
|
issue.add_argument("--policy-version", default="aios.v2.candidate.legacy")
|
|
issue.set_defaults(handler=command_issue_legacy)
|
|
|
|
workflow = subcommands.add_parser(
|
|
"issue-workflow", help="签发不含操作权限的父 Workflow Provenance"
|
|
)
|
|
add_common_store(workflow)
|
|
add_signing_context(workflow)
|
|
add_workflow_identity(workflow)
|
|
workflow.add_argument("--ttl-seconds", type=int, default=14400)
|
|
workflow.set_defaults(handler=command_issue_workflow)
|
|
|
|
grant = subcommands.add_parser("issue-grant", help="签发单一路径最小权限 Grant")
|
|
add_common_store(grant)
|
|
add_signing_context(grant)
|
|
grant.add_argument("--parent-provenance-id", required=True)
|
|
grant.add_argument("--grant-type", choices=[item.value for item in GrantType], required=True)
|
|
grant.add_argument("--allowed-git-operations")
|
|
grant.add_argument("--ttl-seconds", type=int, default=900)
|
|
grant.set_defaults(handler=command_issue_grant)
|
|
|
|
authorize = subcommands.add_parser("authorize", help="使用单一子 Grant 授权一次运维操作")
|
|
add_common_store(authorize)
|
|
add_context(authorize)
|
|
authorize.add_argument("--public-key-file", required=True)
|
|
authorize.add_argument("--key-id", required=True)
|
|
authorize.add_argument("--grant-id", required=True)
|
|
authorize.add_argument(
|
|
"--receipt-dir",
|
|
default=os.getenv(
|
|
"AIOS_PROVENANCE_RECEIPT_DIR",
|
|
"/var/lib/aios-v2/provenance-receipts",
|
|
),
|
|
)
|
|
authorize.add_argument(
|
|
"--path",
|
|
required=True,
|
|
choices=[
|
|
ExecutionPath.CODE_CHANGE.value,
|
|
ExecutionPath.DEPLOYMENT.value,
|
|
ExecutionPath.RUNTIME_EXECUTE.value,
|
|
ExecutionPath.EVIDENCE_WRITE.value,
|
|
ExecutionPath.ROLLBACK.value,
|
|
ExecutionPath.AUDIT.value,
|
|
],
|
|
)
|
|
authorize.add_argument("--operation", required=True)
|
|
request_group = authorize.add_mutually_exclusive_group()
|
|
request_group.add_argument("--request-json")
|
|
request_group.add_argument("--request-file")
|
|
authorize.set_defaults(handler=command_authorize)
|
|
|
|
revoke = subcommands.add_parser("revoke", help="撤销父 Workflow Provenance")
|
|
add_common_store(revoke)
|
|
revoke.add_argument("--provenance-id", required=True)
|
|
revoke.add_argument("--reason", required=True)
|
|
revoke.add_argument("--principal-id", required=True)
|
|
revoke.set_defaults(handler=command_revoke)
|
|
|
|
status = subcommands.add_parser("status", help="查询父 Workflow 与子 Grant")
|
|
add_common_store(status)
|
|
status.add_argument("--provenance-id", required=True)
|
|
status.set_defaults(handler=command_status)
|
|
|
|
grant_revoke = subcommands.add_parser("grant-revoke", help="单独撤销子 Grant")
|
|
add_common_store(grant_revoke)
|
|
grant_revoke.add_argument("--grant-id", required=True)
|
|
grant_revoke.add_argument("--reason", required=True)
|
|
grant_revoke.add_argument("--principal-id", required=True)
|
|
grant_revoke.set_defaults(handler=command_grant_revoke)
|
|
|
|
grant_status = subcommands.add_parser("grant-status", help="查询子 Grant 状态")
|
|
add_common_store(grant_status)
|
|
grant_status.add_argument("--grant-id", required=True)
|
|
grant_status.set_defaults(handler=command_grant_status)
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
try:
|
|
return int(args.handler(args))
|
|
except Exception as exc:
|
|
print(
|
|
json.dumps(
|
|
{"status": "denied", "error": type(exc).__name__, "message": str(exc)},
|
|
ensure_ascii=False,
|
|
),
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|