Add basic ownership as temporary prototype for demo

src/authentication.py

@@ -48,6 +48,9 @@
 class User(BaseModel):
  name: str = Field(description="The username.")
  roles: set[str] = Field(description="The roles.")
+ subject_identifier: str = Field(
+ description="Unique identifier for the user that created the resource."
+ )
 
  def has_role(self, role: str) -> bool:
  return role in self.roles
@@ -85,7 +88,9 @@ async def _get_user(token) -> User:
  logging.error("Invalid userinfo or inactive user.")
  raise InvalidUserError("Invalid userinfo or inactive user") # caught below
  return User(
- name=userinfo["username"], roles=set(userinfo.get("realm_access", {}).get("roles", []))
+ name=userinfo["username"],
+ roles=set(userinfo.get("realm_access", {}).get("roles", [])),
+ subject_identifier=userinfo["sub"],
  )
  except InvalidUserError:
  raise

src/database/model/concept/aiod_entry.py

@@ -14,6 +14,7 @@
 
 if TYPE_CHECKING:
  from database.model.agent.person import Person
+ from authentication import User
 
 
 class AIoDEntryBase(SQLModel):
@@ -38,6 +39,8 @@ class AIoDEntryORM(AIoDEntryBase, table=True): # type: ignore [call-arg]
  date_modified: datetime | None = Field(default_factory=datetime.utcnow)
  date_created: datetime | None = Field(default_factory=datetime.utcnow)
 
+ creator_identifier: str = Field(max_length=32 + 4, default="")
+
  class RelationshipConfig:
  editor: list[int] = ManyToMany() # No deletion triggers: "orphan" Persons should be kept
  status: str | None = ManyToOne(
@@ -46,6 +49,9 @@ class RelationshipConfig:
  deserializer=FindByNameDeserializer(Status),
  )
 
+ def owned_by(self, user: "User") -> bool:
+ return self.creator_identifier == user.subject_identifier
+
 
 class AIoDEntryCreate(AIoDEntryBase):
  editor: list[int] = Field(

src/database/model/concept/concept.py

@@ -1,7 +1,7 @@
 import copy
 import datetime
 import os
-from typing import Optional, Tuple
+from typing import Optional, Tuple, TYPE_CHECKING
 
 from pydantic import validator
 from sqlalchemy import CheckConstraint, Index
@@ -16,6 +16,9 @@
 from database.model.serializers import CastDeserializer
 from database.validators import huggingface_validators, openml_validators, zenodo_validators
 
+if TYPE_CHECKING:
+ from authentication import User
+
 IS_SQLITE = os.getenv("DB") == "SQLite"
 CONSTRAINT_LOWERCASE = f"{'platform' if IS_SQLITE else 'BINARY(platform)'} = LOWER(platform)"
 
@@ -112,3 +115,7 @@ def __table_args__(cls) -> Tuple:
  ),
  CheckConstraint(CONSTRAINT_LOWERCASE, name=f"{cls.__name__}_platform_lowercase"),
  ) + tuple(cls.table_arguments())
+
+ def owned_by(self, user: "User") -> bool:
+ assert self.aiod_entry is not None, f"Expected concept to have aiod_entry: {self}"
+ return self.aiod_entry.owned_by(user)

src/routers/resource_router.py

@@ -403,7 +403,7 @@ def register_resource(
  try:
  with DbSession() as session:
  try:
- resource = self.create_resource(session, resource_create)
+ resource = self.create_resource(session, resource_create, user)
  return self._wrap_with_headers({"identifier": resource.identifier})
  except Exception as e:
  self._raise_clean_http_exception(e, session, resource_create)
@@ -412,12 +412,13 @@ def register_resource(
 
  return register_resource
 
- def create_resource(self, session: Session, resource_create_instance: SQLModel):
+ def create_resource(self, session: Session, resource_create_instance: SQLModel, user: User):
  """Store a resource in the database"""
  resource = self.resource_class.from_orm(resource_create_instance)
  deserialize_resource_relationships(
  session, self.resource_class, resource, resource_create_instance
  )
+ resource.aiod_entry.creator_identifier = user.subject_identifier
  session.add(resource)
  session.commit()
  return resource
@@ -448,6 +449,14 @@ def put_resource(
  with DbSession() as session:
  try:
  resource: Any = self._retrieve_resource(session, identifier)
+ if not resource.owned_by(user):
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail=(
+ f"You do not have permission to edit "
+ f"{self.resource_name} {identifier}."
+ ),
+ )
  for attribute_name in resource.schema()["properties"]:
  if hasattr(resource_create_instance, attribute_name):
  new_value = getattr(resource_create_instance, attribute_name)
@@ -492,6 +501,14 @@ def delete_resource(
  try:
  # Raise error if it does not exist
  resource: Any = self._retrieve_resource(session, identifier)
+ if not resource.owned_by(user):
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail=(
+ f"You do not have permission to delete "
+ f"{self.resource_name} {identifier}."
+ ),
+ )
  if (
  hasattr(self.resource_class, "__deletion_config__")
  and not self.resource_class.__deletion_config__["soft_delete"]