cosign.py

# SPDX-FileCopyrightText: 2024 SAP SE or an SAP affiliate company and Gardener contributors
#
# SPDX-License-Identifier: Apache-2.0

import enum
import hashlib
import json
import logging

import ci.log
import ci.util
import oci.model as om
import oci.client as oc

ci.log.configure_default_logging()
logger = logging.getLogger(__name__)

_no_overwrite_sentinel = object()


class OnExist(enum.StrEnum):
    SKIP = 'skip'
    APPEND = 'append'
    OVERWRITE = 'overwrite'


def payload_bytes(
    image_reference: om.OciImageReference | str,
    annotations: dict | None=None,
    overwrite_docker_reference: str=_no_overwrite_sentinel,
) -> bytes:
    '''
    returns payload for given OCI Image Reference + optional annotations as output by
    `cosign generate`, except `docker-reference` can be overwritten to allow replication of
    artefacts as well as their signature artefacts (digest has to stay the same)

    Passed image-reference must have digest-tag.
    '''
    image_reference = om.OciImageReference.to_image_ref(image_reference)
    if not image_reference.has_digest_tag:
        raise ValueError('image-reference must have digest tag', image_reference)

    if overwrite_docker_reference is _no_overwrite_sentinel:
        docker_reference = image_reference.ref_without_tag
    else:
        docker_reference = overwrite_docker_reference

    payload = {
        'critical': {
            'identity': {
                'docker-reference': docker_reference,
            },
            'image': {
                'docker-manifest-digest': image_reference.tag,
            },
            'type': 'gardener.vnd/oci/cosign-signature',
        },
        'optional': annotations,
    }

    return json.dumps(
        obj=payload,
        separators=(',', ':'),
        sort_keys=True,
    ).encode('utf-8')


def cfg_blob_bytes(
    payload_digest: str,
) -> bytes:
    '''
    returns a dummy cfg-blob as generated by cosign
    '''
    cfg_blob = {
        'architecture': '',
        'config': {},
        'created': '0001-01-01T00:00:00Z',
        'history': [{'created': '0001-01-01T00:00:00Z'}],
        'os': '',
        'rootfs': {
            'diff_ids': [payload_digest],
            'type': 'layers',
        },
    }

    return json.dumps(
        obj=cfg_blob,
        separators=(',', ':'),
        sort_keys=True,
    ).encode('utf-8')


def default_signature_image_reference(
    image_ref: str | om.OciImageReference,
) -> om.OciImageReference:
    '''
    calculate the (default) image reference of the cosign signature for a specific image.

    This image-reference is by default used/expected by cosign if no alternative signature
    repository is specified.
    '''
    parsed_image_ref = om.OciImageReference.to_image_ref(image_ref)
    if not parsed_image_ref.has_digest_tag:
        ValueError('only images that are referenced via digest are allowed')

    parsed_digest = parsed_image_ref.parsed_digest_tag
    alg, val = parsed_digest
    cosign_sig_ref = f'{parsed_image_ref.ref_without_tag}:{alg}-{val}.sig'

    return om.OciImageReference(cosign_sig_ref)


def image_reference_from_signature_reference(
    signature_image_reference: str | om.OciImageReference,
) -> om.OciImageReference:
    signature_image_reference = om.OciImageReference.to_image_ref(signature_image_reference)

    # "sha256-<image-digest>.sig" -> "<image-digest>"
    image_digest = signature_image_reference.tag.split('-')[-1].split('.')[0]
    image_reference = f'{signature_image_reference.ref_without_tag}@sha256:{image_digest}'

    return om.OciImageReference.to_image_ref(image_reference)


'''
annotation name used to store public-key along cosign signatures within cosign signature artefacts

storing public-key in addition to signature is a preparation for using signature algorithms that
yield different signatures even if signing the same content using same private key (e.g. RSSA-PSS)
'''
_public_key_annotation_name = 'gardener.cloud/cosign-public-key'
_signing_algorithm_annotation_name = 'gardener.cloud/cosign-signing-algorithm'
'''
annotation name used by cosign to store signatures for referenced layer-blob
'''
_cosign_signature_annotation_name = 'dev.cosignproject.cosign/signature'


def sign_manifest(
    manifest: om.OciImageManifest,
    payload_size: int,
    payload_digest: str,
    signature: str,
    signing_algorithm: str=None,
    public_key: str=None,
    on_exist: OnExist | str=OnExist.APPEND,
) -> om.OciImageManifest:
    on_exist = OnExist(on_exist)

    if on_exist is OnExist.SKIP and manifest.layers:
        logger.info('manifest already contains a signature - skipping')
        return manifest

    if on_exist is OnExist.APPEND:
        for layer_idx, layer in enumerate(manifest.layers):
            a = layer.annotations
            existing_signature = a.get(_cosign_signature_annotation_name, None)
            # todo: ideally, signatures should be parsed and compared independent of format
            #       cosign seems to expect base64-part w/o PEM-headers (BEGIN/END), though
            if existing_signature == signature:
                logger.info(f'found signature in {layer_idx=}')
                logger.info('skipping (will not redundantly add signature again)')
                return manifest

            existing_public_key = a.get(_public_key_annotation_name, None)
            if not existing_public_key or not public_key:
                continue

            existing_signing_algorithm = a.get(_signing_algorithm_annotation_name, None)
            if (
                signing_algorithm
                and existing_signing_algorithm
                and signing_algorithm != existing_signing_algorithm
            ):
                # we found an existing signature, however with different signing algorithm
                # -> do not skip, as resigning with different algorithm is likely to be
                #    caller's intent
                continue

            if payload_digest != layer.digest:
                # the just calculated payload digest is different from the one of the existing
                # signature, so don't skip as resigning with different payload is likely to be
                # callers intent
                continue

            if existing_public_key == public_key:
                logger.info(f'found matching public key in {layer_idx=}')
                logger.info('skipping (will not redundantly add signature again)')
                return manifest

        # if this line is reached, we did not find the signature we are about to append

    signature_layer = om.OciBlobRef(
        digest=payload_digest,
        size=payload_size,
        mediaType='application/vnd.dev.cosign.simplesigning.v1+json',
        annotations={
            _cosign_signature_annotation_name: signature,
        },
    )

    if public_key:
        signature_layer.annotations[_public_key_annotation_name] = public_key
    if signing_algorithm:
        signature_layer.annotations[_signing_algorithm_annotation_name] = signing_algorithm

    if on_exist is OnExist.APPEND:
        manifest.layers.append(signature_layer)
    else:
        manifest.layers = [
            signature_layer,
        ]

    return manifest


def sign_image(
    image_reference: om.OciImageReference | str,
    signature: str,
    signing_algorithm: str=None,
    public_key: str=None,
    on_exist: OnExist | str=OnExist.APPEND,
    signature_image_reference: om.OciImageReference | str=None,
    oci_client: oc.Client=None,
    payload: bytes | None=None,
    annotations: dict[str, str] | None=None,
) -> tuple[str | om.OciImageReference, om.OciImageManifest]:
    '''
    creates an OCI Image signature as understood by cosign. if passed, public-key is added
    as additional annotation `gardener.cloud/cosign-public-key`

    If on_exist is set to `append`, existing signatures (dev.cosignproject/signature annotation)
    will be inspected. If given signature is already present, it will not be added again, thus
    making this function idempotent in this mode.

    In addition, If public-key is passed, existing annotations bearing
    public-key (gardener.cloud/cosign-public-key) will be compared to passed
    public_key and signature algorithm (stored in annotation
    gardener.cloud/cosign-signing-algorithm). If present, given signature will
    not be added, even if it differs from existing one. This behaviour is a
    preparation for different signature methods yielding different signatures
    even if private key and signed payload did not change (such as RSSA-PSS).
    '''
    on_exist = OnExist(on_exist)
    if not signature_image_reference:
        signature_image_reference = default_signature_image_reference(image_reference)
    else:
        signature_image_reference = om.OciImageReference.to_image_ref(signature_image_reference)

    if not oci_client:
        import ccc.oci
        oci_client = ccc.oci.oci_client()

    image_reference = oci_client.to_digest_hash(image_reference)

    if on_exist in (OnExist.SKIP, OnExist.APPEND):
        exists = bool(oci_client.head_manifest(
            image_reference=signature_image_reference,
            absent_ok=True,
        ))

    if on_exist is OnExist.SKIP and exists:
        logger.info(f'signature artefact exists: {signature_image_reference} - skipping')
        return

    if exists:
        manifest = oci_client.manifest(
            image_reference=signature_image_reference,
        )
    else:
        manifest = None

    if not payload:
        # payload is normalised JSON w/ reference to signed image. It is expected as (only)
        # layer-blob for signature artefact
        payload = payload_bytes(
            image_reference=image_reference,
        )
    payload_size = len(payload)
    payload_digest = f'sha256:{hashlib.sha256(payload).hexdigest()}'

    if not manifest or not any(l.digest == payload_digest for l in manifest.layers):
        oci_client.put_blob(
            image_reference=signature_image_reference,
            digest=payload_digest,
            octets_count=payload_size,
            data=payload,
        )

    cfg_blob = cfg_blob_bytes(
        payload_digest=payload_digest,
    )
    cfg_blob_size = len(cfg_blob)
    cfg_blob_digest = f'sha256:{hashlib.sha256(cfg_blob).hexdigest()}'

    if not manifest or manifest.config.digest != cfg_blob_digest:
        oci_client.put_blob(
            image_reference=image_reference,
            digest=cfg_blob_digest,
            octets_count=cfg_blob_size,
            data=cfg_blob,
        )

    if manifest:
        # update manifest's cfg blob to the one we just created, it may be different because
        # we used to create those cfg blobs using the cosign command itself (which has slightly
        # different semantics) in the past or because the contained payload digest is different
        manifest.config = om.OciBlobRef(
            digest=cfg_blob_digest,
            mediaType='application/vnd.oci.image.config.v1+json',
            size=cfg_blob_size,
        )

        if annotations:
            manifest.annotations = manifest.annotations | annotations
    else:
        manifest = om.OciImageManifest(
            config=om.OciBlobRef(
                digest=cfg_blob_digest,
                mediaType='application/vnd.oci.image.config.v1+json',
                size=cfg_blob_size,
            ),
            mediaType='application/vnd.oci.image.manifest.v1+json',
            layers=[],
            annotations=annotations or {},
        )

    signed_manifest = sign_manifest(
        manifest=manifest,
        payload_size=payload_size,
        payload_digest=payload_digest,
        signature=signature,
        signing_algorithm=signing_algorithm,
        public_key=public_key,
        on_exist=on_exist,
    )

    manifest_bytes = json.dumps(signed_manifest.as_dict()).encode('utf-8')
    manifest_digest = f'sha256:{hashlib.sha256(manifest_bytes).hexdigest()}'
    manifest_digest_ref = f'{signature_image_reference.ref_without_tag}@{manifest_digest}'

    # re-uploading manifest with same digest is a no-op
    if not oci_client.head_manifest(
        image_reference=manifest_digest_ref,
        absent_ok=True,
    ):
        oci_client.put_manifest(
            image_reference=signature_image_reference,
            manifest=manifest_bytes,
        )

    return signature_image_reference, signed_manifest