# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.
"""Data models for assets."""
import enum
from collections.abc import Generator, Iterable
from typing import (
Any,
Self,
TYPE_CHECKING,
TypeAlias,
assert_never,
cast,
override,
)
from django.core.exceptions import (
ImproperlyConfigured,
PermissionDenied,
ValidationError,
)
from django.db import models
from django.db.models import Exists, OuterRef, Q
from django.db.models.constraints import CheckConstraint, UniqueConstraint
from debusine.assets import (
AssetCategory,
BaseAssetDataModel,
BasicAPTAuthenticationData,
SigningKeyData,
asset_data_model,
)
from debusine.db.constraints import JsonDataUniqueConstraint
from debusine.db.context import context
from debusine.db.models import permissions
from debusine.db.models.permissions import (
Allow,
FastCheckResult,
Role,
enforce,
fast_check_common,
permission_check,
permission_filter,
)
from debusine.db.models.workspaces import Workspace
from debusine.db.permissioncontext import PermissionContext
from debusine.utils.typing_utils import copy_signature_from
if TYPE_CHECKING:
from django_stubs_ext.db.models import TypedModelMeta
else:
TypedModelMeta = object
class AssetRoleBase(permissions.RoleBase):
"""Asset role implementation."""
implied_by_workspace_roles: frozenset[Workspace.Roles]
implied_by_asset_roles: frozenset["AssetRoles"]
@override
def _setup(self) -> None:
"""Set up implications for a newly constructed role."""
implied_by_workspace_roles: set[Workspace.Roles] = set()
implied_by_asset_roles: set[AssetRoles] = {cast(AssetRoles, self)}
for i in self.implied_by:
match i:
case Workspace.Roles():
implied_by_workspace_roles |= i.implied_by_workspace_roles
case Role():
# Resolve a role passed during class definition into its
# enum instance
role = self.__class__(i.value)
implied_by_workspace_roles |= (
role.implied_by_workspace_roles
)
implied_by_asset_roles |= role.implied_by_asset_roles
case _:
raise ImproperlyConfigured(
f"Asset roles do not support implications by {i!r}"
)
self.implied_by_workspace_roles = frozenset(implied_by_workspace_roles)
self.implied_by_asset_roles = frozenset(implied_by_asset_roles)
@classmethod
def q(cls, pc: PermissionContext, *roles: "AssetRoleBase") -> Q:
"""Return a Q expression to select assets with this role."""
q = Q(
roles__group__in=pc.groups(),
roles__role__in=set().union(
*(role.implied_by_asset_roles for role in roles)
),
)
workspace_roles = set().union(
*(role.implied_by_workspace_roles for role in roles)
)
if workspace_roles:
q |= Q(
workspace__in=Workspace.objects.filter(
Workspace.Roles.q(pc, *workspace_roles)
)
)
# TODO: This intentionally filters out instance-wide assets. After
# #1507, this needs to be changed to check for instance-wide roles
# instead
return Q(workspace__in=Workspace.objects.can_display(pc)) & q
@classmethod
def fast_check(
cls,
pc: PermissionContext,
resource: "Asset",
*roles: "Asset.Roles",
) -> FastCheckResult:
"""
Check if the context can satisfy one of the given roles.
:param pc: permission context to check
:param resource: resource to check
:param roles: roles that would satisfy the check
:returns: YES if at least one of the given roles is satisfied without
needing DB lookups. NO if it can be decided that none of the given
roles is satisfied without needing DB lookup. UNDECIDED if DB lookup
is needed to decide.
"""
if resource.workspace is None:
# TODO: This intentionally denies roles to instance-wide assets.
# After #1507, this needs to be changed to check for instance-wide
# roles instead
return FastCheckResult.NO
if (
res := fast_check_common(pc, resource, *roles)
) != FastCheckResult.UNDECIDED:
return res
if not context.pc_is(pc):
return FastCheckResult.UNDECIDED
if resource.workspace != context.workspace:
return FastCheckResult.UNDECIDED
# Check workspace roles
workspace_roles = frozenset.union(
*(role.implied_by_workspace_roles for role in roles)
)
if (
workspace_roles
and Workspace.Roles.fast_check(
pc, resource.workspace, *workspace_roles
)
) == FastCheckResult.YES:
return FastCheckResult.YES
return FastCheckResult.UNDECIDED
def implies(self, role: "AssetRoles") -> bool:
"""Check if this role implies the given one."""
return (
self.implied_by_workspace_roles <= role.implied_by_workspace_roles
and self.implied_by_asset_roles <= role.implied_by_asset_roles
)
class AssetRoles(permissions.Roles, AssetRoleBase, enum.ReprEnum):
"""Available roles for an Asset."""
OWNER = Role("owner", description="Manage the asset")
VIEWER = Role(
"viewer",
implied_by=[OWNER, Workspace.Roles.VIEWER],
description="Read only access to the asset",
)
AssetRoles.setup()
class AssetUsageRoleBase(permissions.RoleBase):
"""AssetUsage role implementation."""
implied_by_workspace_roles: frozenset[Workspace.Roles]
implied_by_asset_usage_roles: frozenset["AssetUsageRoles"]
@override
def _setup(self) -> None:
"""Set up implications for a newly constructed role."""
implied_by_workspace_roles: set[Workspace.Roles] = set()
implied_by_asset_usage_roles: set[AssetUsageRoles] = {
cast(AssetUsageRoles, self)
}
for i in self.implied_by:
match i:
case Workspace.Roles():
implied_by_workspace_roles |= i.implied_by_workspace_roles
case Role():
# Resolve a role passed during class definition into its
# enum instance
role = self.__class__(i.value)
implied_by_workspace_roles |= (
role.implied_by_workspace_roles
)
implied_by_asset_usage_roles |= (
role.implied_by_asset_usage_roles
)
case _:
raise ImproperlyConfigured(
f"AssetUsage roles do not support implications by {i!r}"
)
self.implied_by_workspace_roles = frozenset(implied_by_workspace_roles)
self.implied_by_asset_usage_roles = frozenset(
implied_by_asset_usage_roles
)
@classmethod
def q(cls, pc: PermissionContext, *roles: "AssetUsageRoleBase") -> Q:
"""Return a Q expression to select asset usages with this role."""
q = Q(
roles__group__in=pc.groups(),
roles__role__in=set().union(
*(role.implied_by_asset_usage_roles for role in roles)
),
)
workspace_roles = set().union(
*(role.implied_by_workspace_roles for role in roles)
)
if workspace_roles:
q |= Q(
workspace__in=Workspace.objects.filter(
Workspace.Roles.q(pc, *workspace_roles)
)
)
return Q(workspace__in=Workspace.objects.can_display(pc)) & q
@classmethod
def fast_check(
cls,
pc: PermissionContext,
resource: "AssetUsage",
*roles: "AssetUsage.Roles",
) -> FastCheckResult:
"""
Check if the context can satisfy one of the given roles.
:param pc: permission context to check
:param resource: resource to check
:param roles: roles that would satisfy the check
:returns: YES if at least one of the given roles is satisfied without
needing DB lookups. NO if it can be decided that none of the given
roles is satisfied without needing DB lookup. UNDECIDED if DB lookup
is needed to decide.
"""
if (
res := fast_check_common(pc, resource, *roles)
) != FastCheckResult.UNDECIDED:
return res
if not context.pc_is(pc):
return FastCheckResult.UNDECIDED
if resource.workspace != context.workspace:
return FastCheckResult.UNDECIDED
# TODO: after #1507, check for instance-wide permissions here for
# instance-wide assets
if (
resource.asset.workspace is not None
and resource.asset.workspace != context.workspace
):
return FastCheckResult.UNDECIDED
# Check workspace roles
workspace_roles = frozenset.union(
*(role.implied_by_workspace_roles for role in roles)
)
if (
workspace_roles
and Workspace.Roles.fast_check(
pc, resource.workspace, *workspace_roles
)
) == FastCheckResult.YES:
return FastCheckResult.YES
return FastCheckResult.UNDECIDED
def implies(self, role: "AssetUsageRoles") -> bool:
"""Check if this role implies the given one."""
return (
self.implied_by_workspace_roles <= role.implied_by_workspace_roles
and self.implied_by_asset_usage_roles
<= role.implied_by_asset_usage_roles
)
class AssetUsageRoles(permissions.Roles, AssetUsageRoleBase, enum.ReprEnum):
"""Available roles for an AssetUsage."""
SIGNER = Role(
"signer",
description="Sign data using this key",
)
REPOSITORY_SIGNER = Role(
"repository_signer",
label="Repository signer",
implied_by=[SIGNER, Workspace.Roles.OWNER],
description="Sign repositories using this key",
)
APT_AUTHENTICATOR = Role(
"apt_authenticator",
label="Can use APT authentication",
implied_by=[Workspace.Roles.OWNER],
description="Authenticate to an external APT repository "
"using these credentials",
)
AssetUsageRoles.setup()
class AssetQuerySet[A](models.QuerySet["Asset", A]):
"""Custom QuerySet for Asset."""
def in_current_scope(self) -> "AssetQuerySet[A]":
"""Filter to assets in the current scope."""
from debusine.db.context import context
return self.filter(workspace__scope=context.require_scope())
def with_role(self, pc: PermissionContext, role: AssetRoles) -> Self:
"""Keep only resources where the user has the given role."""
if not pc.user.is_authenticated:
return self.none()
return self.filter(
Exists(
self.model.objects.filter(
AssetRoles.q(pc, role), pk=OuterRef("pk")
)
)
)
@permission_filter(work_request=Allow.PASS, anonymous=Allow.PASS)
def can_display(self, pc: PermissionContext) -> "AssetQuerySet[A]":
"""Keep only Assets that can be displayed."""
return self.with_role(pc, AssetRoles.VIEWER).exclude(
category=AssetCategory.CLOUD_PROVIDER_ACCOUNT
)
@permission_filter()
def can_edit(self, pc: PermissionContext) -> "AssetQuerySet[A]":
"""Keep only Assets that can be edited."""
return self.with_role(pc, AssetRoles.OWNER).exclude(
category=AssetCategory.CLOUD_PROVIDER_ACCOUNT
)
@permission_filter()
def can_manage_permissions(
self, pc: PermissionContext
) -> "AssetQuerySet[A]":
"""Filter to Assets that can be managed by user."""
return self.with_role(pc, AssetRoles.OWNER)
class AssetManager(models.Manager["Asset"]):
"""Manager for the Asset model."""
def get_roles_model(self) -> type["AssetRole"]:
"""Get the model used for role assignment."""
return AssetRole
@override
def get_queryset(self) -> AssetQuerySet[Any]:
"""Use the custom QuerySet."""
return AssetQuerySet(self.model, using=self._db)
def get_by_slug(
self, category: str, slug: str, workspace: Workspace | None = None
) -> "Asset":
"""Return an asset with a matching slug."""
match category:
case AssetCategory.SIGNING_KEY:
purpose, fingerprint = slug.split(":", 1)
return self.get(
category=category,
data__purpose=purpose,
data__fingerprint=fingerprint,
)
case AssetCategory.APT_AUTHENTICATION:
assets = self.filter(category=category)
if slug.startswith(":"):
return assets.get(
workspace__isnull=True, data__name=slug[1:]
)
if ":" in slug:
assert workspace is not None
workspace_name, name = slug.split(":", 1)
return assets.get(
workspace__scope=workspace.scope,
workspace__name=workspace_name,
data__name=name,
)
else:
return assets.get(workspace=workspace, data__name=slug)
case _:
raise ValueError(f"No slug defined for category '{category}'")
[docs]
class Asset(models.Model):
"""Asset model."""
category = models.CharField(
max_length=255,
choices=AssetCategory.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
workspace = models.ForeignKey(
Workspace,
on_delete=models.PROTECT,
blank=True,
null=True,
help_text="Asset workspace, or None for instance-wide assets",
)
data = models.JSONField(default=dict)
created_at = models.DateTimeField(auto_now_add=True)
created_by = models.ForeignKey(
"User", blank=True, null=True, on_delete=models.PROTECT
)
created_by_work_request = models.ForeignKey(
"WorkRequest", blank=True, null=True, on_delete=models.SET_NULL
)
Roles: TypeAlias = AssetRoles
objects = AssetManager.from_queryset(AssetQuerySet)()
class Meta(TypedModelMeta):
base_manager_name = "objects"
constraints = [
JsonDataUniqueConstraint(
fields=["data->>'name'"],
condition=models.Q(
category=AssetCategory.CLOUD_PROVIDER_ACCOUNT
),
nulls_distinct=False,
name="%(app_label)s_%(class)s_unique_cloud_provider_acct_name",
),
JsonDataUniqueConstraint(
fields=["data->>'fingerprint'"],
condition=models.Q(category=AssetCategory.SIGNING_KEY),
nulls_distinct=False,
name="%(app_label)s_%(class)s_unique_signing_key_fingerprints",
),
CheckConstraint(
condition=(
~models.Q(category=AssetCategory.APT_AUTHENTICATION)
| models.Q(data__name__regex=r"^[A-Za-z][A-Za-z0-9+._-]*$")
),
name="%(app_label)s_%(class)s_apt_auth_name",
),
JsonDataUniqueConstraint(
fields=["workspace", "data->>'name'"],
condition=models.Q(category=AssetCategory.APT_AUTHENTICATION),
nulls_distinct=False,
name="%(app_label)s_%(class)s_unique_apt_auth_workspace_name",
),
# Some categories of asset can have null workspaces, but not
# signing keys.
CheckConstraint(
condition=(
~models.Q(category=AssetCategory.SIGNING_KEY)
| models.Q(workspace__isnull=False)
),
name="%(app_label)s_%(class)s_workspace_not_null",
),
]
@override
def __str__(self) -> str:
"""Return basic information of Asset."""
return (
f"Id: {self.id} "
f"Category: {self.category} "
f"Workspace: {self.workspace}"
)
[docs]
@override
@copy_signature_from(models.Model.save)
def save(self, **kwargs: Any) -> None:
"""Wrap save with permission checks."""
from debusine.db.context import context
if context.permission_checks_disabled:
pass
elif self._state.adding:
# TODO: deny saving instance-wide assets except when running with
# permission checks disabled, until we can check for instance-wide
# roles #1507
if self.workspace is None:
raise PermissionDenied(
"Workspace needs to be set on new assets"
)
enforce(self.workspace.can_create_assets)
if self.category == AssetCategory.CLOUD_PROVIDER_ACCOUNT:
# Not currently creatable through the API
raise PermissionDenied(
"Cloud provider accounts are not currently"
" creatable through the API"
)
else:
enforce(self.can_edit)
return super().save(**kwargs)
[docs]
@override
def clean(self) -> None:
"""
Ensure that data is valid for this asset category.
:raise ValidationError: for invalid data.
"""
self.data_model
@property
def slug(self) -> str:
"""Return a string slug that uniquely identifies the asset."""
match self.category:
case AssetCategory.SIGNING_KEY:
data_model = self.data_model
assert isinstance(data_model, SigningKeyData)
return f"{data_model.purpose}:{data_model.fingerprint}"
case AssetCategory.APT_AUTHENTICATION:
data_model = self.data_model
assert isinstance(data_model, BasicAPTAuthenticationData)
if self.workspace is None:
return f":{data_model.name}"
else:
return f"{self.workspace.name}:{data_model.name}"
case _:
raise NotImplementedError(
f"No slug defined for category '{self.category}'"
)
[docs]
def has_role(self, pc: PermissionContext, role: AssetRoles) -> bool:
"""Check if the user has the given role on this group."""
match Asset.Roles.fast_check(pc, self, role):
case FastCheckResult.YES:
return True
case FastCheckResult.NO:
return False
case FastCheckResult.UNDECIDED:
return (
Asset.objects.with_role(pc, role)
.filter(pk=self.pk)
.exists()
)
case _ as unreachable:
assert_never(unreachable)
[docs]
@permission_check(
"{user} cannot display {resource}",
work_request=Allow.PASS,
anonymous=Allow.PASS,
)
def can_display(self, pc: PermissionContext) -> bool:
"""Check if the user can display this asset."""
if self.category == AssetCategory.CLOUD_PROVIDER_ACCOUNT:
return False
return self.has_role(pc, Asset.Roles.VIEWER)
[docs]
@permission_check("{user} cannot edit asset {resource}")
def can_edit(self, pc: PermissionContext) -> bool:
"""Check if the user can edit this asset."""
if self.category == AssetCategory.CLOUD_PROVIDER_ACCOUNT:
return False
return self.has_role(pc, Asset.Roles.OWNER)
[docs]
@permission_check("{user} cannot manage permissions on {resource}")
def can_manage_permissions(self, pc: PermissionContext) -> bool:
"""Check if the user can manage permissions on this asset."""
return self.has_role(pc, Asset.Roles.OWNER)
@property
def data_model(self) -> BaseAssetDataModel:
"""Instantiate AssetData from data."""
if not isinstance(self.data, dict):
raise ValidationError({"data": "data must be a dictionary"})
try:
return asset_data_model(self.category, self.data)
except ValueError as e:
raise ValidationError(
{
"category": (
f"{self.category}: invalid asset category or data: {e}"
),
},
) from e
class AssetRole(models.Model):
"""Role assignment for assets."""
resource = models.ForeignKey(
Asset,
on_delete=models.CASCADE,
related_name="roles",
)
group = models.ForeignKey(
"Group",
on_delete=models.CASCADE,
related_name="asset_roles",
)
role = models.CharField(
max_length=16,
choices=AssetRoles.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
class AssetUsageQuerySet[A](models.QuerySet["AssetUsage", A]):
"""Custom QuerySet for AssetUsage."""
def with_role(self, pc: PermissionContext, role: AssetUsageRoles) -> Self:
"""Keep only resources where the user has the given role."""
if not pc.user.is_authenticated:
return self.none()
return self.filter(
Exists(
self.model.objects.filter(
AssetUsageRoles.q(pc, role), pk=OuterRef("pk")
)
)
)
@permission_filter(work_request=Allow.PASS)
def can_sign_with(self, pc: PermissionContext) -> "AssetUsageQuerySet[A]":
"""Keep only AssetUsages that the user can sign with."""
return self.with_role(pc, AssetUsageRoles.SIGNER)
@permission_filter(work_request=Allow.PASS)
def can_sign_repository_with(
self, pc: PermissionContext
) -> "AssetUsageQuerySet[A]":
"""Keep only AssetUsages that the user can sign a repository with."""
return self.with_role(pc, AssetUsageRoles.REPOSITORY_SIGNER)
@permission_filter(work_request=Allow.PASS)
def can_use_apt_authentication_with(
self, pc: PermissionContext
) -> "AssetUsageQuerySet[A]":
"""Keep only AssetUsages that the user can use for APT auth."""
return self.with_role(pc, AssetUsageRoles.APT_AUTHENTICATOR)
class AssetUsageManager(models.Manager["AssetUsage"]):
"""Manager for the AssetUsage model."""
def get_roles_model(self) -> type["AssetUsageRole"]:
"""Get the model used for role assignment."""
return AssetUsageRole
[docs]
class AssetUsage(models.Model):
"""Usage of an Asset within a workspace."""
Roles: TypeAlias = AssetUsageRoles
asset = models.ForeignKey(
Asset,
on_delete=models.CASCADE,
related_name="usage",
)
workspace = models.ForeignKey(
"Workspace",
on_delete=models.CASCADE,
related_name="asset_usage",
)
objects = AssetUsageManager.from_queryset(AssetUsageQuerySet)()
class Meta(TypedModelMeta):
base_manager_name = "objects"
constraints = [
UniqueConstraint(
fields=["asset", "workspace"],
name="%(app_label)s_%(class)s_unique_asset_workspace",
)
]
[docs]
def has_role(self, pc: PermissionContext, role: AssetUsageRoles) -> bool:
"""Check if the user has the given role on this group."""
match AssetUsage.Roles.fast_check(pc, self, role):
case FastCheckResult.YES:
return True
case FastCheckResult.NO:
return False
case FastCheckResult.UNDECIDED:
return (
AssetUsage.objects.with_role(pc, role)
.filter(pk=self.pk)
.exists()
)
case _ as unreachable:
assert_never(unreachable)
[docs]
@permission_check(
"{user} cannot sign with {resource}", work_request=Allow.PASS
)
def can_sign_with(self, pc: PermissionContext) -> bool:
"""Check if the user can sign with this resource."""
return self.has_role(pc, AssetUsageRoles.SIGNER)
[docs]
@permission_check(
"{user} cannot sign a repository with {resource}",
work_request=Allow.PASS,
)
def can_sign_repository_with(self, pc: PermissionContext) -> bool:
"""Check if the user can sign a repository with this resource."""
return self.has_role(pc, AssetUsageRoles.REPOSITORY_SIGNER)
[docs]
@permission_check(
"{user} cannot use {resource} for APT authentication",
work_request=Allow.PASS,
)
def can_use_apt_authentication_with(self, pc: PermissionContext) -> bool:
"""Check if the user can use APT authentication with this resource."""
return self.has_role(pc, AssetUsageRoles.APT_AUTHENTICATOR)
class AssetUsageRole(models.Model):
"""Role assignment for assets within a workspace."""
resource = models.ForeignKey(
AssetUsage,
on_delete=models.CASCADE,
related_name="roles",
)
group = models.ForeignKey(
"Group",
on_delete=models.CASCADE,
related_name="asset_usage_roles",
)
role = models.CharField(
max_length=32,
choices=AssetUsageRoles.choices,
# TODO: db_collation=COLLATION_CODEPOINT,
)
def get_public_keys(signing_keys: Iterable[Asset]) -> Generator[bytes]:
"""Yield public keys from each of some signing-key assets."""
for signing_key in signing_keys:
signing_key_data = signing_key.data_model
assert isinstance(signing_key_data, SigningKeyData)
yield signing_key_data.public_key