Add fragment references from_image command

Author: DomAyre

src/confcom/azext_confcom/_params.py

@@ -235,6 +235,14 @@ def load_arguments(self, _):
             required=False,
             help='Container definitions to include in the policy'
         )
+        c.argument(
+            "fragment_definitions",
+            options_list=['--with-fragments'],
+            action='append',
+            type=json.loads,
+            required=False,
+            help='Fragment definitions to include in the policy'
+        )
 
     with self.argument_context("confcom acifragmentgen") as c:
         c.argument(
@@ -484,3 +492,16 @@ def load_arguments(self, _):
             type=str,
             help="Platform to create container definition for",
         )
+
+    with self.argument_context("confcom fragment references from_image") as c:
+        c.positional(
+            "image",
+            type=str,
+            help="Image to create container definition from",
+        )
+        c.argument(
+            "minimum_svn",
+            required=False,
+            type=str,
+            help="Minimum Allowed Software Version Number for Fragment",
+        )

src/confcom/azext_confcom/command/fragment_references_from_image.py

@@ -0,0 +1,14 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import json
+
+from typing import Optional
+
+from azext_confcom.lib.fragment_references_from_image import fragment_references_from_image as lib_fragment_references_from_image
+
+
+def fragment_references_from_image(image: str, minimum_svn: Optional[str]) -> str:
+    return print(json.dumps(list(lib_fragment_references_from_image(image, minimum_svn))))

src/confcom/azext_confcom/commands.py

@@ -20,3 +20,6 @@ def load_command_table(self, _):
 
     with self.command_group("confcom containers") as g:
         g.custom_command("from_image", "containers_from_image")
+
+    with self.command_group("confcom fragment references") as g:
+        g.custom_command("from_image", "fragment_references_from_image")

src/confcom/azext_confcom/custom.py

@@ -25,6 +25,7 @@
 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 azext_confcom.command.containers_from_image import containers_from_image as _containers_from_image
+from azext_confcom.command.fragment_references_from_image import fragment_references_from_image as _fragment_references_from_image
 from knack.log import get_logger
 from pkg_resources import parse_version
 
@@ -57,6 +58,7 @@ def acipolicygen_confcom(
     include_fragments: bool = False,
     fragments_json: str = None,
     exclude_default_fragments: bool = False,
+    fragment_definitions: Optional[list] = None,
 ):
     if print_existing_policy or outraw or outraw_pretty_print:
         logger.warning(
@@ -165,6 +167,10 @@ def acipolicygen_confcom(
             container_definitions=container_definitions,
         )
 
+    if fragment_definitions:
+        for fragment_definition in fragment_definitions:
+            container_group_policies._fragments.extend(fragment_definition)  # pylint: disable=protected-access
+
     exit_code = 0
 
     # standardize the output so we're only operating on arrays
@@ -561,3 +567,13 @@ def containers_from_image(
         image=image,
         platform=platform,
     )
+
+
+def fragment_references_from_image(
+    image: str,
+    minimum_svn: Optional[str],
+) -> None:
+    _fragment_references_from_image(
+        image=image,
+        minimum_svn=minimum_svn,
+    )

src/confcom/azext_confcom/lib/cose.py

@@ -0,0 +1,66 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import hashlib
+import platform
+import re
+import requests
+import subprocess
+
+from typing import Iterable
+from pathlib import Path
+
+from azext_confcom.lib.paths import get_binaries_dir
+
+
+_binaries_dir = get_binaries_dir()
+_cosesign1_binaries = {
+    "Linux": {
+        "path": _binaries_dir / "sign1util",
+        "url": "https://github.com/microsoft/cosesign1go/releases/download/v1.4.0/sign1util",
+        "sha256": "526b54aeb6293fc160e8fa1f81be6857300aba9641d45955f402f8b082a4d4a5",
+    },
+    "Windows": {
+        "path": _binaries_dir / "sign1util.exe",
+        "url": "https://github.com/microsoft/cosesign1go/releases/download/v1.4.0/sign1util.exe",
+        "sha256": "f33cccf2b1bb8c3a495c730984b47d0f0715678981dbfe712248a2452dd53303",
+    },
+}
+
+
+def cose_get():
+    for binary_info in _cosesign1_binaries.values():
+        cosesign1_fetch_resp = requests.get(binary_info["url"], verify=True)
+        cosesign1_fetch_resp.raise_for_status()
+
+        assert hashlib.sha256(cosesign1_fetch_resp.content).hexdigest() == binary_info["sha256"]
+
+        with open(binary_info["path"], "wb") as f:
+            f.write(cosesign1_fetch_resp.content)
+
+
+def cose_run(args: Iterable[str]) -> subprocess.CompletedProcess:
+    return subprocess.run(
+        [_cosesign1_binaries[platform.system()]["path"], *args],
+        check=True,
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+
+
+def cose_print(file_path: Path):
+    return cose_run([
+        "print",
+        "--in", file_path.as_posix(),
+    ]).stdout.strip()
+
+
+def cose_get_properties(file_path: Path):
+    cose_print_output = cose_print(file_path)
+    return {
+        "iss": re.search(r"^iss:\s*(.*)$", cose_print_output, re.MULTILINE).group(1),
+        "feed": re.search(r"^feed:\s*(.*)$", cose_print_output, re.MULTILINE).group(1),
+        "payload": re.search(r"^payload:\s*(.*)", cose_print_output, re.MULTILINE | re.DOTALL).group(1),
+    }

src/confcom/azext_confcom/lib/fragment_references_from_image.py

@@ -0,0 +1,41 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import tempfile
+
+from pathlib import Path
+from typing import Optional
+
+from azext_confcom.lib.cose import cose_get_properties
+from azext_confcom.lib.fragments import get_fragments_from_image
+from azext_confcom.lib.opa import opa_eval
+
+
+def fragment_references_from_image(image: str, minimum_svn: Optional[str]):
+
+    for signed_fragment in get_fragments_from_image(image):
+
+        package_name = signed_fragment.name.split(".")[0]
+        cose_properties = cose_get_properties(signed_fragment)
+
+        with tempfile.NamedTemporaryFile("w+b") as payload:
+            payload.write(cose_properties["payload"].encode("utf-8"))
+            payload.flush()
+            fragment_properties = opa_eval(
+                Path(payload.name),
+                f"data.{package_name}",
+            )["result"][0]["expressions"][0]["value"]
+
+            yield {
+                "feed": cose_properties["feed"],
+                "includes": sorted(list(set(fragment_properties.keys()).intersection({
+                    "containers",
+                    "fragmnents",
+                    "namespace",
+                    "external_processes",
+                }))),
+                "issuer": cose_properties["iss"],
+                "minimum_svn": minimum_svn or fragment_properties["svn"],
+            }

src/confcom/azext_confcom/lib/fragments.py

@@ -0,0 +1,22 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import tempfile
+
+from pathlib import Path
+
+from azext_confcom.lib.images import sanitize_image_reference
+from azext_confcom.lib.oras import get_artifact_references, pull
+
+
+def get_fragments_from_image(image_reference: str):
+
+    for reference in get_artifact_references(image_reference):
+
+        fragment_path = Path(tempfile.gettempdir()) / sanitize_image_reference(reference)
+        pull(reference, fragment_path)
+
+        for fragment_file in fragment_path.glob("*.rego.cose"):
+            yield fragment_file

src/confcom/azext_confcom/lib/images.py

@@ -5,6 +5,7 @@
 
 import functools
 import os
+import re
 import subprocess
 import docker
 
@@ -62,3 +63,7 @@ def get_image_config(image: str) -> dict:
         config["working_dir"] = image_config.get("WorkingDir")
 
     return config
+
+
+def sanitize_image_reference(image_reference: str) -> str:
+    return re.sub(f"[{re.escape(r'<>:"/\\|?*@\0')}]", "-", image_reference)

src/confcom/azext_confcom/lib/oras.py

@@ -0,0 +1,61 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import shutil
+import json
+import subprocess
+
+from pathlib import Path
+from typing import Iterable
+
+from azext_confcom.errors import eprint
+
+
+def oras_run(args: Iterable[str]) -> subprocess.CompletedProcess:
+
+    # Maintain existing behaviour of requiring the user to install ORAS themselves
+    if not shutil.which("oras"):
+        eprint("ORAS CLI not installed. Please install ORAS CLI: https://oras.land/docs/installation")
+
+    return subprocess.run(
+        ["oras", *args],
+        check=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+        text=True,
+    )
+
+
+def discover(reference: str):
+    return json.loads(oras_run([
+        "discover",
+        "--format",
+        "json",
+        reference,
+    ]).stdout.strip())
+
+
+def pull(reference: str, destination: Path):
+    return oras_run([
+        "pull",
+        "--output",
+        destination.as_posix(),
+        reference,
+    ])
+
+
+def get_artifact_references(reference: str):
+
+    def get_references(discover_result):
+        if "artifactType" in discover_result:
+            yield discover_result["reference"]
+        for field in discover_result.values():
+            if isinstance(field, list):
+                for item in field:
+                    yield from get_references(item)
+            if isinstance(field, dict):
+                yield from get_references(field)
+
+    return list(get_references(discover(reference)))