Add support for pulp-python attestations feature

Assisted By: Cursor Composer

Author: gerrod3

CHANGES/+python-attestation.feature

@@ -0,0 +1,3 @@
+Added support for uploading attestations with Python Package content.
+Added support for uploading Python Provenance content.
+Added support for specifying syncing of Python Provenance content.

CHANGES/+python-repository-content.feature

@@ -0,0 +1 @@
+Change the python repository content add/remove/modify commands to only require the package's sha256.

pulp-glue/pulp_glue/python/context.py

@@ -40,6 +40,16 @@ class PulpPythonContentContext(PulpContentContext):
     CAPABILITIES = {"upload": []}
 
 
+class PulpPythonProvenanceContext(PulpContentContext):
+    PLUGIN = "python"
+    RESOURCE_TYPE = "provenance"
+    ENTITY = _("python provenance")
+    ENTITIES = _("python provenances")
+    HREF = "python_python_provenance_content_href"
+    ID_PREFIX = "content_python_provenance"
+    NEEDS_PLUGINS = [PluginRequirement("python", specifier=">=3.22.0")]
+
+
 class PulpPythonDistributionContext(PulpDistributionContext):
     PLUGIN = "python"
     RESOURCE_TYPE = "python"

pulpcore/cli/python/content.py

@@ -4,7 +4,11 @@
 from pulp_glue.common.context import PluginRequirement, PulpEntityContext
 from pulp_glue.common.i18n import get_translation
 from pulp_glue.core.context import PulpArtifactContext
-from pulp_glue.python.context import PulpPythonContentContext, PulpPythonRepositoryContext
+from pulp_glue.python.context import (
+    PulpPythonContentContext,
+    PulpPythonProvenanceContext,
+    PulpPythonRepositoryContext,
+)
 
 from pulp_cli.generic import (
     PulpCLIContext,
@@ -14,12 +18,14 @@
     label_command,
     label_select_option,
     list_command,
+    load_json_callback,
     pass_entity_context,
     pass_pulp_context,
     pulp_group,
     pulp_option,
     resource_option,
     show_command,
+    type_option,
 )
 
 translation = get_translation(__package__)
@@ -37,6 +43,24 @@ def _sha256_artifact_callback(
     return value
 
 
+def _attestation_callback(
+    ctx: click.Context, param: click.Parameter, value: t.Iterable[str] | None
+) -> list[t.Any] | None:
+    """Callback to process multiple attestation values and combine them into a list."""
+    if not value:
+        return None
+    result = []
+    for attestation_value in value:
+        # Use load_json_callback to process each value (supports JSON strings and file paths)
+        processed = load_json_callback(ctx, param, attestation_value)
+        # If it's already a list, extend; otherwise append
+        if isinstance(processed, list):
+            result.extend(processed)
+        else:
+            result.append(processed)
+    return result
+
+
 repository_option = resource_option(
     "--repository",
     default_plugin="python",
@@ -51,26 +75,49 @@ def _sha256_artifact_callback(
     ),
 )
 
+package_option = resource_option(
+    "--package",
+    default_plugin="python",
+    default_type="package",
+    lookup_key="sha256",
+    context_table={
+        "python:package": PulpPythonContentContext,
+    },
+    href_pattern=PulpPythonContentContext.HREF_PATTERN,
+    help=_(
+        "Package to associate the provenance with in the form"
+        "'[[<plugin>:]<resource_type>:]<sha256>' or by href/prn."
+    ),
+    allowed_with_contexts=(PulpPythonProvenanceContext,),
+    required=True,
+)
+
 
 @pulp_group()
-@click.option(
-    "-t",
-    "--type",
-    "content_type",
-    type=click.Choice(["package"], case_sensitive=False),
+@type_option(
+    choices={
+        "package": PulpPythonContentContext,
+        "provenance": PulpPythonProvenanceContext,
+    },
     default="package",
+    case_sensitive=False,
 )
-@pass_pulp_context
-@click.pass_context
-def content(ctx: click.Context, pulp_ctx: PulpCLIContext, /, content_type: str) -> None:
-    if content_type == "package":
-        ctx.obj = PulpPythonContentContext(pulp_ctx)
-    else:
-        raise NotImplementedError()
+def content() -> None:
+    pass
 
 
 create_options = [
-    click.option("--relative-path", required=True, help=_("Exact name of file")),
+    pulp_option(
+        "--relative-path",
+        required=True,
+        help=_("Exact name of file"),
+        allowed_with_contexts=(PulpPythonContentContext,),
+    ),
+    pulp_option(
+        "--file",
+        type=click.File("rb"),
+        help=_("Path to the file to create {entity} from"),
+    ),
     click.option(
         "--sha256",
         "artifact",
@@ -79,21 +126,43 @@ def content(ctx: click.Context, pulp_ctx: PulpCLIContext, /, content_type: str)
     ),
     pulp_option(
         "--file-url",
-        help=_("Remote url to download and create python content from"),
+        help=_("Remote url to download and create {entity} from"),
         needs_plugins=[PluginRequirement("core", specifier=">=3.56.1")],
     ),
+    pulp_option(
+        "--attestation",
+        "attestations",
+        multiple=True,
+        callback=_attestation_callback,
+        needs_plugins=[PluginRequirement("python", specifier=">=3.22.0")],
+        help=_(
+            "A JSON object containing an attestation for the package. Can be a JSON string or a "
+            "file path prefixed with '@'. Can be specified multiple times."
+        ),
+        allowed_with_contexts=(PulpPythonContentContext,),
+    ),
+]
+provenance_create_options = [
+    package_option,
+    pulp_option(
+        "--verify/--no-verify",
+        default=True,
+        needs_plugins=[PluginRequirement("python", specifier=">=3.22.0")],
+        help=_("Verify the provenance"),
+        allowed_with_contexts=(PulpPythonProvenanceContext,),
+    ),
 ]
 lookup_options = [href_option]
 content.add_command(
     list_command(
         decorators=[
-            click.option("--filename", type=str),
+            pulp_option("--filename", type=str, allowed_with_contexts=(PulpPythonContentContext,)),
             label_select_option,
         ]
     )
 )
 content.add_command(show_command(decorators=lookup_options))
-content.add_command(create_command(decorators=create_options))
+content.add_command(create_command(decorators=create_options + provenance_create_options))
 content.add_command(
     label_command(
         decorators=lookup_options,
@@ -102,10 +171,21 @@ def content(ctx: click.Context, pulp_ctx: PulpCLIContext, /, content_type: str)
 )
 
 
-@content.command()
+@content.command(allowed_with_contexts=(PulpPythonContentContext,))
 @click.option("--relative-path", required=True, help=_("Exact name of file"))
 @click.option("--file", type=click.File("rb"), required=True, help=_("Path to file"))
 @chunk_size_option
+@pulp_option(
+    "--attestation",
+    "attestations",
+    multiple=True,
+    callback=_attestation_callback,
+    needs_plugins=[PluginRequirement("python", specifier=">=3.22.0")],
+    help=_(
+        "A JSON object containing an attestation for the package. Can be a JSON string or a file"
+        " path prefixed with '@'. Can be specified multiple times."
+    ),
+)
 @repository_option
 @pass_entity_context
 @pass_pulp_context
@@ -116,12 +196,17 @@ def upload(
     relative_path: str,
     file: t.IO[bytes],
     chunk_size: int,
+    attestations: list[t.Any] | None,
     repository: PulpPythonRepositoryContext | None,
 ) -> None:
     """Create a Python package content unit through uploading a file"""
     assert isinstance(entity_ctx, PulpPythonContentContext)
 
     result = entity_ctx.upload(
-        relative_path=relative_path, file=file, chunk_size=chunk_size, repository=repository
+        relative_path=relative_path,
+        file=file,
+        chunk_size=chunk_size,
+        repository=repository,
+        attestations=attestations,
     )
     pulp_ctx.output_result(result)

pulpcore/cli/python/remote.py

@@ -98,6 +98,12 @@ def remote(ctx: click.Context, pulp_ctx: PulpCLIContext, /, remote_type: str) ->
         callback=load_json_callback,
         needs_plugins=[PluginRequirement("python", specifier=">=3.2.0")],
     ),
+    pulp_option(
+        "--provenance/--no-provenance",
+        default=None,
+        help=_("Sync available package provenances"),
+        needs_plugins=[PluginRequirement("python", specifier=">=3.22.0")],
+    ),
 ]
 
 remote.add_command(list_command(decorators=remote_filter_options))

pulpcore/cli/python/repository.py

@@ -11,22 +11,21 @@
 from pulp_glue.common.i18n import get_translation
 from pulp_glue.python.context import (
     PulpPythonContentContext,
+    PulpPythonProvenanceContext,
     PulpPythonRemoteContext,
     PulpPythonRepositoryContext,
 )
 
 from pulp_cli.generic import (
-    GroupOption,
     PulpCLIContext,
     create_command,
     create_content_json_callback,
     destroy_command,
     href_option,
-    json_callback,
     label_command,
     label_select_option,
     list_command,
-    load_file_wrapper,
+    lookup_callback,
     name_option,
     pass_pulp_context,
     pass_repository_context,
@@ -60,31 +59,7 @@
 )
 
 
-def _content_callback(ctx: click.Context, param: click.Parameter, value: t.Any) -> t.Any:
-    if value:
-        pulp_ctx = ctx.find_object(PulpCLIContext)
-        assert pulp_ctx is not None
-        ctx.obj = PulpPythonContentContext(pulp_ctx, entity=value)
-    return value
-
-
-CONTENT_LIST_SCHEMA = s.Schema([{"sha256": str, "filename": s.And(str, len)}])
-
-
-@load_file_wrapper
-def _content_list_callback(ctx: click.Context, param: click.Parameter, value: str | None) -> t.Any:
-    if value is None:
-        return None
-
-    result = json_callback(ctx, param, value)
-    try:
-        return CONTENT_LIST_SCHEMA.validate(result)
-    except s.SchemaError as e:
-        raise click.ClickException(
-            _("Validation of '{parameter}' failed: {error}").format(
-                parameter=param.name, error=str(e)
-            )
-        )
+CONTENT_LIST_SCHEMA = s.Schema([{"sha256": str, s.Optional("filename"): str}])
 
 
 @pulp_group()
@@ -119,36 +94,37 @@ def repository(ctx: click.Context, pulp_ctx: PulpCLIContext, /, repo_type: str)
 ]
 create_options = update_options + [click.option("--name", required=True)]
 package_options = [
-    click.option("--sha256", cls=GroupOption, expose_value=False, group=["filename"]),
+    pulp_option(
+        "--sha256",
+        callback=lookup_callback("sha256"),
+        expose_value=False,
+        help=_("SHA256 digest of the {entity}."),
+    ),
     click.option(
         "--filename",
-        callback=_content_callback,
         expose_value=False,
-        cls=GroupOption,
-        group=["sha256"],
-        help=_("Filename of the python package."),
+        help=_("Filename of the python package. [deprecated]"),
     ),
+    href_option,
 ]
-content_json_callback = create_content_json_callback(
-    PulpPythonContentContext, schema=CONTENT_LIST_SCHEMA
-)
+content_json_callback = create_content_json_callback(None, schema=CONTENT_LIST_SCHEMA)
 modify_options = [
-    click.option(
+    pulp_option(
         "--add-content",
         callback=content_json_callback,
         help=_(
-            """JSON string with a list of objects to add to the repository.
-    Each object must contain the following keys: "sha256", "filename".
-    The argument prefixed with the '@' can be the path to a JSON file with a list of objects."""
+            """JSON string with a list of {entities} to add to the repository.
+    Each {entity} must contain the following keys: "sha256".
+    The argument prefixed with the '@' can be the path to a JSON file with a list of {entities}."""
         ),
     ),
-    click.option(
+    pulp_option(
         "--remove-content",
         callback=content_json_callback,
         help=_(
-            """JSON string with a list of objects to remove from the repository.
-    Each object must contain the following keys: "sha256", "filename".
-    The argument prefixed with the '@' can be the path to a JSON file with a list of objects."""
+            """JSON string with a list of {entities} to remove from the repository.
+    Each {entity} must contain the following keys: "sha256".
+    The argument prefixed with the '@' can be the path to a JSON file with a list of {entities}."""
         ),
     ),
 ]
@@ -163,7 +139,10 @@ def repository(ctx: click.Context, pulp_ctx: PulpCLIContext, /, repo_type: str)
 repository.add_command(label_command(decorators=nested_lookup_options))
 repository.add_command(
     repository_content_command(
-        contexts={"package": PulpPythonContentContext},
+        contexts={
+            "package": PulpPythonContentContext,
+            "provenance": PulpPythonProvenanceContext,
+        },
         add_decorators=package_options,
         remove_decorators=package_options,
         modify_decorators=modify_options,