Setting the file. One moment.
Private Artifacts · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary
helpers/ private_artifacts.py
Python · 103 lines · 5 KB
HelperFailure, blocked_result, canonical_bytes, digest, emit_result, reject_secrets
12 except ImportError :
13 import _bootstrap_io as private_io
14 from _common import HelperFailure, blocked_result, canonical_bytes, digest, emit_result, reject_secrets
15
16
17 def add_execution_output_argument (parser):
18 parser.add_argument( "--execution-output" , type = Path,
19 help = "With --plan only: write exact UNAPPROVED input to a new absolute private file." )
20
21
22 def validate_execution_output_mode (args):
23 if args.execution_output is not None and not args.plan:
24 raise private_io.failure( "planning-output-invalid" , "--execution-output requires --plan." )
25
26
27 def retain_execution_input (document, output):
28 """Extension contract: pass the exact closed envelope, never a redacted wrapper."""
29 try :
30 if ( not isinstance (document, dict ) or set (document) != { "schema_version" , "plan" , "approval" }
31 or document[ "schema_version" ] not in ( "1.0" , "2.0" )
32 or not isinstance (document[ "plan" ], dict )
33 or document[ "approval" ] != { "confirmed" : False , "fingerprint" : digest(document[ "plan" ])}
34 or document[ "approval" ][ "confirmed" ] is not False ):
35 raise ValueError ( "Unapproved envelope required" )
36 reject_secrets(document)
37 path = Path(output)
38 if not path.is_absolute():
39 raise ValueError ( "Absolute output required" )
40 except (HelperFailure, TypeError , ValueError , RecursionError ) as exc:
41 raise private_io.failure( "planning-output-invalid" , "Select an absolute private output and an exact unapproved, secret-free execution input." ) from exc
42 written = private_io.atomic_private_file(
43 path.parent, path.name, document,
44 max_bytes = private_io. MAX_BYTES * ( 1 if document[ "schema_version" ] == "2.0" else 16 ))
45 return {
46 "path" : str (written),
47 "sha256" : hashlib.sha256(canonical_bytes(document) + b " \n " ).hexdigest(),
48 "fingerprint" : document[ "approval" ][ "fingerprint" ],
49 "confirmed" : False ,
50 }
51
52
53 def emit_plan_result (result, execution_output = None , * , preserve_unapproved_input = False ):
54 if execution_output is None :
55 emit_result(result, preserve_unapproved_input = preserve_unapproved_input)
56 return
57 if result.get( "status" ) != "planned" :
58 emit_result(result)
59 return
60 reference = retain_execution_input(result.get( "execution_input" ), execution_output)
61 # Copy only fixed booleans/counts: arbitrary summaries contain names, queries and private text.
62 original = result.get( "approval_summary" )
63 original = original if isinstance (original, dict ) else {}
64 summary = {}
65 for key in ( "execution_required" , "mutation_approval_required" ):
66 if type (original.get(key)) is bool :
67 summary[key] = original[key]
68 boundary = original.get( "data_boundary" )
69 boundary = boundary if isinstance (boundary, dict ) else {}
70 for key in ( "file_count" , "object_count" , "total_bytes" , "uploads" , "queries" ):
71 value = original.get(key, boundary.get(key))
72 if type (value) is int and 0 <= value <= 2 ** 63 - 1 :
73 summary[key] = value
74 emit_result({
75 "status" : "planned" , "execution_artifact" : reference,
76 "summary" : summary,
77 "next_step" : "Review the private artifact. Execute only if required, after separate approval." ,
78 "local_filesystem" : { "artifact_created" : True , "existing_files_changed" : False },
79 "azure_mutation_performed" : False ,
80 })
81
82
83 def main (argv = None ):
84 parser = argparse.ArgumentParser( description = "Create one new private leaf or validate an existing private receipt directory; local only." )
85 modes = parser.add_mutually_exclusive_group( required = True )
86 modes.add_argument( "--create-directory" )
87 modes.add_argument( "--validate-directory" )
88 args = parser.parse_args(argv)
89 try :
90 path = (private_io.create_private_directory(args.create_directory) if args.create_directory else
91 private_io.validate_private_artifact_directory(args.validate_directory))
92 emit_result({ "status" : "ready" , "private_directory" : str (path),
93 "local_filesystem" : { "directory_created" : args.create_directory is not None ,
94 "existing_acls_changed" : False },
95 "azure_calls_performed" : False , "approval_granted" : False })
96 return 0
97 except HelperFailure as error:
98 emit_result(blocked_result(error, outcome = "private-artifact-directory" , fingerprint = None ))
99 return 2
100
101
102 if __name__ == "__main__" :
103 sys.exit(main())