[confcom] Restore the behaviour of --upload-fragment for acifragmentgen

linter_exclusions.yml

@@ -3442,3 +3442,15 @@ neon postgres organization:
 neon postgres project:
   rule_exclusions:
   - require_wait_command_if_no_wait
+
+confcom fragment push:
+  parameters:
+    signed_fragment:
+      rule_exclusions:
+        - no_positional_parameters
+
+confcom fragment attach:
+  parameters:
+    signed_fragment:
+      rule_exclusions:
+        - no_positional_parameters

src/confcom/HISTORY.rst

@@ -3,6 +3,12 @@
 Release History
 ===============
 
+1.4.0
+++++++
+* restored the behaviour of --upload-fragment in acifragmentgen to attach to first image in input
+* added `fragment push` command to allow explicit uploading of standalone fragments
+* added `fragment attach` command to allow explicit uploading of image attached fragments
+
 1.3.0
 ++++++
 * Add a new --enable-stdio flag, with a warning if neither this or --disable-stdio is set

src/confcom/azext_confcom/_help.py

@@ -278,3 +278,46 @@
         - name: Input a Kubernetes YAML file with a custom containerd socket path
           text: az confcom katapolicygen --yaml "./pod.json" --containerd-pull --containerd-socket-path "/my/custom/containerd.sock"
 """
+
+helps[
+    "confcom fragment"
+] = """
+    type: group
+    short-summary: Commands to handle Confidential Container Policy Fragments.
+"""
+
+helps[
+    "confcom fragment push"
+] = """
+    type: command
+    short-summary: Push a Confidential Container Policy Fragment to an ORAS registry
+
+    parameters:
+      - name: --manifest-tag
+        type: string
+        short-summary: 'The reference to push the signed fragment to'
+
+    examples:
+        - name: Push a signed fragment to a registry
+          text: az confcom fragment push ./fragment.reg.cose --manifest-tag myregistry.azurecr.io/fragment:latest
+        - name: Push the output of acifragmentgen to a registry
+          text: az confcom acifragmentgen --chain my.cert.pem --key my_key.pem --svn "1" --namespace contoso --feed "test-feed" --input ./fragment_spec.json | az confcom fragment push --manifest-tag myregistry.azurecr.io/fragment:latest
+"""
+
+helps[
+    "confcom fragment attach"
+] = """
+    type: command
+    short-summary: Attach a Confidential Container Policy Fragment to an image in an ORAS registry.
+
+    parameters:
+      - name: --manifest-tag
+        type: string
+        short-summary: 'The reference to attach the signed fragment to'
+
+    examples:
+        - name: Attach a signed fragment to a registry
+          text: az confcom fragment attach ./fragment.reg.cose --manifest-tag myregistry.azurecr.io/image:latest
+        - name: Attach the output of acifragmentgen to a registry
+          text: az confcom acifragmentgen --chain my.cert.pem --key my_key.pem --svn "1" --namespace contoso --feed "test-feed" --input ./fragment_spec.json | az confcom fragment attach --manifest-tag myregistry.azurecr.io/image:latest
+"""

src/confcom/azext_confcom/_params.py

@@ -4,6 +4,8 @@
 # --------------------------------------------------------------------------------------------
 # pylint: disable=line-too-long
 
+import argparse
+import sys
 from knack.arguments import CLIArgumentType
 from azext_confcom._validators import (
     validate_params_file,
@@ -43,6 +45,32 @@ def load_arguments(self, _):
         c.argument("tags", tags_type)
         c.argument("confcom_name", confcom_name_type, options_list=["--name", "-n"])
 
+    with self.argument_context("confcom fragment attach") as c:
+        c.positional(
+            "signed_fragment",
+            nargs='?',
+            type=argparse.FileType('rb'),
+            default=sys.stdin.buffer,
+            help="Signed fragment to attach",
+        )
+        c.argument(
+            "manifest_tag",
+            help="Manifest tag for the fragment",
+        )
+
+    with self.argument_context("confcom fragment push") as c:
+        c.positional(
+            "signed_fragment",
+            nargs='?',
+            type=argparse.FileType('rb'),
+            default=sys.stdin.buffer,
+            help="Signed fragment to push",
+        )
+        c.argument(
+            "manifest_tag",
+            help="Manifest tag for the fragment",
+        )
+
     with self.argument_context("confcom acipolicygen") as c:
         c.argument(
             "input_path",
@@ -345,6 +373,13 @@ def load_arguments(self, _):
             help="Path to JSON file to write fragment import information. This is used with --generate-import. If not specified, the import statement will print to the console",
             validator=validate_fragment_json,
         )
+        c.argument(
+            "out_signed_fragment",
+            action="store_true",
+            default=False,
+            required=False,
+            help="Emit only the signed fragment bytes",
+        )
 
     with self.argument_context("confcom katapolicygen") as c:
         c.argument(

src/confcom/azext_confcom/command/fragment_attach.py

@@ -0,0 +1,46 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+import subprocess
+import tempfile
+from typing import BinaryIO
+
+
+def oras_attach(
+    signed_fragment: BinaryIO,
+    manifest_tag: str,
+) -> None:
+    subprocess.run(
+        [
+            "oras",
+            "attach",
+            "--artifact-type", "application/x-ms-ccepolicy-frag",
+            manifest_tag,
+            os.path.relpath(signed_fragment.name, start=os.getcwd()),
+        ],
+        check=True,
+        timeout=120,
+    )
+
+
+def fragment_attach(
+    signed_fragment: BinaryIO,
+    manifest_tag: str,
+) -> None:
+
+    if signed_fragment.name == "<stdin>":
+        with tempfile.NamedTemporaryFile(delete=True) as temp_signed_fragment:
+            temp_signed_fragment.write(signed_fragment.read())
+            temp_signed_fragment.flush()
+            oras_attach(
+                signed_fragment=temp_signed_fragment,
+                manifest_tag=manifest_tag,
+            )
+    else:
+        oras_attach(
+            signed_fragment=signed_fragment,
+            manifest_tag=manifest_tag,
+        )

src/confcom/azext_confcom/command/fragment_push.py

@@ -0,0 +1,46 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+import subprocess
+import tempfile
+from typing import BinaryIO
+
+
+def oras_push(
+    signed_fragment: BinaryIO,
+    manifest_tag: str,
+) -> None:
+    subprocess.run(
+        [
+            "oras",
+            "push",
+            "--artifact-type", "application/x-ms-ccepolicy-frag",
+            manifest_tag,
+            os.path.relpath(signed_fragment.name, start=os.getcwd()),
+        ],
+        check=True,
+        timeout=120,
+    )
+
+
+def fragment_push(
+    signed_fragment: BinaryIO,
+    manifest_tag: str,
+) -> None:
+
+    if signed_fragment.name == "<stdin>":
+        with tempfile.NamedTemporaryFile(delete=True) as temp_signed_fragment:
+            temp_signed_fragment.write(signed_fragment.read())
+            temp_signed_fragment.flush()
+            oras_push(
+                signed_fragment=temp_signed_fragment,
+                manifest_tag=manifest_tag,
+            )
+    else:
+        oras_push(
+            signed_fragment=signed_fragment,
+            manifest_tag=manifest_tag,
+        )

src/confcom/azext_confcom/commands.py

@@ -11,5 +11,9 @@ def load_command_table(self, _):
         g.custom_command("acifragmentgen", "acifragmentgen_confcom")
         g.custom_command("katapolicygen", "katapolicygen_confcom")
 
+    with self.command_group("confcom fragment") as g:
+        g.custom_command("attach", "fragment_attach", is_preview=True)
+        g.custom_command("push", "fragment_push", is_preview=True)
+
     with self.command_group("confcom"):
         pass

src/confcom/azext_confcom/custom.py

@@ -5,7 +5,7 @@
 
 import os
 import sys
-from typing import Optional
+from typing import BinaryIO, Optional
 
 from azext_confcom import oras_proxy, os_util, security_policy
 from azext_confcom._validators import resolve_stdio
@@ -22,6 +22,8 @@
     get_image_name, inject_policy_into_template, inject_policy_into_yaml,
     pretty_print_func, print_existing_policy_from_arm_template,
     print_existing_policy_from_yaml, print_func, str_to_sha256)
+from azext_confcom.command.fragment_attach import fragment_attach as _fragment_attach
+from azext_confcom.command.fragment_push import fragment_push as _fragment_push
 from knack.log import get_logger
 from pkg_resources import parse_version
 
@@ -240,6 +242,7 @@ def acifragmentgen_confcom(
     upload_fragment: bool = False,
     no_print: bool = False,
     fragments_json: str = "",
+    out_signed_fragment: bool = False,
 ):
 
     stdio_enabled = resolve_stdio(enable_stdio, disable_stdio)
@@ -330,24 +333,39 @@ def acifragmentgen_confcom(
 
     fragment_text = policy.generate_fragment(namespace, svn, output_type, omit_id=omit_id)
 
-    if output_type != security_policy.OutputType.DEFAULT and not no_print:
+    if output_type != security_policy.OutputType.DEFAULT and not no_print and not out_signed_fragment:
         print(fragment_text)
 
     # take ".rego" off the end of the filename if it's there, it'll get added back later
     output_filename = output_filename.replace(".rego", "")
     filename = f"{output_filename or namespace}.rego"
+
+    if out_signed_fragment:
+        filename = os.path.join("/tmp", filename)
+
     os_util.write_str_to_file(filename, fragment_text)
 
     if key:
         cose_proxy = CoseSignToolProxy()
         iss = cose_proxy.create_issuer(chain)
         out_path = filename + ".cose"
 
+        if out_signed_fragment:
+            out_path = os.path.join("/tmp", os.path.basename(out_path))
+
         cose_proxy.cose_sign(filename, key, chain, feed, iss, algo, out_path)
-        if upload_fragment and image_target:
-            oras_proxy.attach_fragment_to_image(image_target, out_path)
-        elif upload_fragment:
-            oras_proxy.push_fragment_to_registry(feed, out_path)
+
+        # Preserve default behaviour established since version 1.1.0 of attaching
+        # the fragment to the first image specified in input
+        # (or --image-target if specified)
+        if upload_fragment:
+            oras_proxy.attach_fragment_to_image(
+                image_name=image_target or policy_images[0].containerImage,
+                filename=out_path,
+            )
+
+        if out_signed_fragment:
+            sys.stdout.buffer.write(open(out_path, "rb").read())
 
 
 def katapolicygen_confcom(
@@ -481,3 +499,23 @@ def get_fragment_output_type(outraw):
     if outraw:
         output_type = security_policy.OutputType.RAW
     return output_type
+
+
+def fragment_attach(
+    signed_fragment: BinaryIO,
+    manifest_tag: str,
+) -> None:
+    _fragment_attach(
+        signed_fragment=signed_fragment,
+        manifest_tag=manifest_tag
+    )
+
+
+def fragment_push(
+    signed_fragment: BinaryIO,
+    manifest_tag: str,
+) -> None:
+    _fragment_push(
+        signed_fragment=signed_fragment,
+        manifest_tag=manifest_tag
+    )