platform-ci/platform_ci/delivery/authorization.py

"""Publication requests and operator-owned access bindings.

A source configuration may only REQUEST a delivery by naming an artifact, a format and a logical destination. It cannot
carry a URL, credential, registry, owner, repository or signer. The concrete target comes from an `AccessBinding` that the
operator provisioned for the authenticated identity; an identity is authorization data and never selects behavior.
"""

import re
from collections.abc import Mapping
from dataclasses import dataclass
from typing import cast

from platform_ci.delivery.errors import AuthorizationError, UnsupportedFormatError

MAVEN_FORMAT = "maven"
NPM_FORMAT = "npm"
CONTAINER_FORMAT = "container"
GITOPS_DIGEST_FORMAT = "gitops-digest"
PUBLICATION_FORMATS = frozenset({MAVEN_FORMAT, NPM_FORMAT, CONTAINER_FORMAT})
DEPLOYMENT_FORMATS = frozenset({GITOPS_DIGEST_FORMAT})
_REQUEST_KEYS = frozenset({"component", "artifact", "format", "platform", "destination", "channel"})
_LOGICAL_NAME = re.compile(r"[a-z0-9][a-z0-9._-]*")
_OWNER = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*")
_IMAGE = re.compile(r"[a-z0-9][a-z0-9._-]*(/[a-z0-9][a-z0-9._-]*)+")


@dataclass(frozen=True)
class PublicationRequest:
   """A source-declared request for one artifact to reach one logical destination. It contains no secret or address."""

   name: str
   component: str
   artifact: str
   format: str
   platform: str
   destination: str
   channel: str


@dataclass(frozen=True)
class PackageTarget:
   """A package registry owner, for Maven and npm delivery."""

   owner: str


@dataclass(frozen=True)
class ImageTarget:
   """A container registry image path."""

   image: str


@dataclass(frozen=True)
class GitOpsTarget:
   """One kustomization file on one branch of one GitOps repository, and the images it may pin (kustomize name to registry image)."""

   repository: str
   repository_id: int
   branch: str
   path: str
   images: tuple[tuple[str, str], ...]


DestinationTarget = PackageTarget | ImageTarget | GitOpsTarget


@dataclass(frozen=True)
class AccessBinding:
   """An operator-owned grant: this identity may deliver these formats to this logical destination on these channels."""

   identity: str
   destination: str
   formats: frozenset[str]
   channels: frozenset[str]
   target: DestinationTarget
   require_provenance: bool = False
   require_signature_verification: bool = False


def parse_publication_request(name: str, document: object) -> PublicationRequest:
   """Validate one request from source configuration; any unknown field, including a credential-like one, is refused."""
   if not isinstance(document, dict):
      raise UnsupportedFormatError(f"publication {name!r} must be an object")
   fields = cast(dict[str, object], document)
   extra = set(fields) - _REQUEST_KEYS
   if extra:
      raise AuthorizationError(f"publication {name!r} may not set {sorted(extra)}; destinations and credentials come only from operator bindings")
   missing = _REQUEST_KEYS - set(fields)
   if missing:
      raise AuthorizationError(f"publication {name!r} is missing {sorted(missing)}")
   values = {key: fields[key] for key in _REQUEST_KEYS}
   if not all(isinstance(value, str) and _LOGICAL_NAME.fullmatch(value) for value in values.values()):
      raise AuthorizationError(f"publication {name!r} fields must be logical names")
   request = PublicationRequest(name=name, component=cast(str, values["component"]), artifact=cast(str, values["artifact"]), format=cast(str, values["format"]), platform=cast(str, values["platform"]), destination=cast(str, values["destination"]), channel=cast(str, values["channel"]))
   require_supported_format(request.format, PUBLICATION_FORMATS | DEPLOYMENT_FORMATS)
   return request


def require_supported_format(format_name: str, supported: frozenset[str]) -> None:
   """Fail explicitly for a format no adapter implements; there is no fallback and no best-effort guess."""
   if format_name not in supported:
      raise UnsupportedFormatError(f"unsupported delivery format {format_name!r}; supported: {sorted(supported)}")


def authorize(bindings: tuple[AccessBinding, ...], identity: str, destination: str, format_name: str, channel: str) -> AccessBinding:
   """Return the single binding that grants this exact identity, destination, format and channel, or refuse."""
   matches = [binding for binding in bindings if binding.identity == identity and binding.destination == destination]
   if not matches:
      raise AuthorizationError(f"identity is not bound to destination {destination!r}")
   if len(matches) > 1:
      raise AuthorizationError(f"destination {destination!r} has ambiguous bindings for this identity")
   binding = matches[0]
   if format_name not in binding.formats:
      raise AuthorizationError(f"destination {destination!r} is not authorized for format {format_name!r}")
   if channel not in binding.channels:
      raise AuthorizationError(f"destination {destination!r} is not authorized for channel {channel!r}")
   _require_target_matches_format(binding.target, format_name)
   return binding


def _require_target_matches_format(target: DestinationTarget, format_name: str) -> None:
   """Refuse a mis-provisioned binding whose target cannot carry the format, and malformed targets."""
   if format_name in (MAVEN_FORMAT, NPM_FORMAT):
      if not isinstance(target, PackageTarget) or not _OWNER.fullmatch(target.owner):
         raise AuthorizationError(f"binding target is not a package owner for {format_name}")
   elif format_name == CONTAINER_FORMAT:
      if not isinstance(target, ImageTarget) or not _IMAGE.fullmatch(target.image):
         raise AuthorizationError("binding target is not a container image")
   elif format_name == GITOPS_DIGEST_FORMAT:
      if not isinstance(target, GitOpsTarget) or not target.images or not all(_IMAGE.fullmatch(image) for _, image in target.images):
         raise AuthorizationError("binding target is not a GitOps file with enrolled images")
   else:
      raise UnsupportedFormatError(f"unsupported delivery format {format_name!r}")


def image_map(target: GitOpsTarget) -> Mapping[str, str]:
   """Return the enrolled kustomize-name to registry-image mapping of a GitOps target."""
   return dict(target.images)