Change API to be able to assign roles to namespaces

Fixes conda-incubator#491.

conda-store-server/conda_store_server/alembic/versions/46bdf428642d_change_namespacerolemapping_and_.py

@@ -0,0 +1,56 @@
+"""Change NamespaceRoleMapping and Namespace
+
+Revision ID: 46bdf428642d
+Revises: b387747ca9b7
+Create Date: 2023-10-08 10:40:06.227854
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "46bdf428642d"
+down_revision = "b387747ca9b7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "namespace_role_mapping_new",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("namespace_id", sa.Integer(), nullable=False),
+        sa.Column("other_namespace_id", sa.Integer(), nullable=False),
+        sa.Column("role", sa.Unicode(length=255), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["namespace_id"],
+            ["namespace.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["other_namespace_id"],
+            ["namespace.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("namespace_id", "other_namespace_id", name="_uc"),
+    )
+    # Note: data is NOT copied before dropping the old table
+    op.drop_table("namespace_role_mapping")
+    op.rename_table("namespace_role_mapping_new", "namespace_role_mapping")
+
+
+def downgrade():
+    op.create_table(
+        "namespace_role_mapping_new",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("namespace_id", sa.Integer(), nullable=False),
+        sa.Column("entity", sa.Unicode(length=255), nullable=False),
+        sa.Column("role", sa.Unicode(length=255), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["namespace_id"],
+            ["namespace.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # Note: data is NOT copied before dropping the old table
+    op.drop_table("namespace_role_mapping")
+    op.rename_table("namespace_role_mapping_new", "namespace_role_mapping")

conda-store-server/conda_store_server/api.py

@@ -3,6 +3,7 @@
 
 from conda_store_server import conda_utils, orm, schema, utils
 from sqlalchemy import distinct, func, null, or_
+from sqlalchemy.orm import aliased
 
 
 def list_namespaces(db, show_soft_deleted: bool = False):
@@ -45,11 +46,10 @@ def create_namespace(db, name: str):
     return namespace
 
 
-def update_namespace(
+def update_namespace_metadata(
     db,
     name: str,
     metadata_: Dict[str, Any] = None,
-    role_mappings: Dict[str, List[str]] = None,
 ):
     namespace = get_namespace(db, name)
     if namespace is None:
@@ -58,28 +58,154 @@ def update_namespace(
     if metadata_ is not None:
         namespace.metadata_ = metadata_
 
-    if role_mappings is not None:
-        # deletes all the existing role mappings ...
-        for rm in namespace.role_mappings:
-            db.delete(rm)
+    return namespace
 
-        # ... before adding all the new ones
-        mappings_orm = []
-        for entity, roles in role_mappings.items():
-            for role in roles:
-                mapping_orm = orm.NamespaceRoleMapping(
-                    namespace_id=namespace.id,
-                    namespace=namespace,
-                    entity=entity,
-                    role=role,
-                )
-                mappings_orm.append(mapping_orm)
 
-        namespace.role_mappings = mappings_orm
+def get_namespace_roles(
+    db,
+    name: str,
+):
+    """Which namespaces can access namespace 'name'?"""
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
 
-    db.commit()
+    nrm = aliased(orm.NamespaceRoleMapping)
+    this = aliased(orm.Namespace)
+    other = aliased(orm.Namespace)
+    q = (
+        db.query(nrm.id, this.name, other.name, nrm.role)
+        .filter(nrm.namespace_id == namespace.id)
+        .filter(nrm.namespace_id == this.id)
+        .filter(nrm.other_namespace_id == other.id)
+        .all()
+    )
+    return [schema.NamespaceRoleMapping.from_list(x) for x in q]
 
-    return namespace
+
+def get_other_namespace_roles(
+    db,
+    name: str,
+):
+    """To which namespaces does namespace 'name' have access?"""
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    nrm = aliased(orm.NamespaceRoleMapping)
+    this = aliased(orm.Namespace)
+    other = aliased(orm.Namespace)
+    q = (
+        db.query(nrm.id, this.name, other.name, nrm.role)
+        .filter(nrm.other_namespace_id == namespace.id)
+        .filter(nrm.namespace_id == this.id)
+        .filter(nrm.other_namespace_id == other.id)
+        .all()
+    )
+    return [schema.NamespaceRoleMapping.from_list(x) for x in q]
+
+
+def delete_namespace_roles(
+    db,
+    name: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    nrm = orm.NamespaceRoleMapping
+    db.query(nrm).filter(nrm.namespace_id == namespace.id).delete()
+
+
+def get_namespace_role(
+    db,
+    name: str,
+    other: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    nrm = aliased(orm.NamespaceRoleMapping)
+    this = aliased(orm.Namespace)
+    other = aliased(orm.Namespace)
+    q = (
+        db.query(nrm.id, this.name, other.name, nrm.role)
+        .filter(nrm.namespace_id == namespace.id)
+        .filter(nrm.other_namespace_id == other_namespace.id)
+        .filter(nrm.namespace_id == this.id)
+        .filter(nrm.other_namespace_id == other.id)
+        .first()
+    )
+    if q is None:
+        return None
+    return schema.NamespaceRoleMapping.from_list(q)
+
+
+def create_namespace_role(
+    db,
+    name: str,
+    other: str,
+    role: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    db.add(
+        orm.NamespaceRoleMapping(
+            namespace_id=namespace.id,
+            other_namespace_id=other_namespace.id,
+            role=role,
+        )
+    )
+
+
+def update_namespace_role(
+    db,
+    name: str,
+    other: str,
+    role: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    nrm = orm.NamespaceRoleMapping
+    db.query(nrm).filter(nrm.namespace_id == namespace.id).filter(
+        nrm.other_namespace_id == other_namespace.id
+    ).update({"role": role})
+
+
+def delete_namespace_role(
+    db,
+    name: str,
+    other: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    nrm = orm.NamespaceRoleMapping
+    db.query(nrm).filter(nrm.namespace_id == namespace.id).filter(
+        nrm.other_namespace_id == other_namespace.id
+    ).delete()
 
 
 def delete_namespace(db, name: str = None, id: int = None):

conda-store-server/conda_store_server/orm.py

@@ -54,39 +54,37 @@ class Namespace(Base):
 
     metadata_ = Column(JSON, default=dict, nullable=True)
 
-    role_mappings = relationship("NamespaceRoleMapping", back_populates="namespace")
-
 
 class NamespaceRoleMapping(Base):
     """Mapping between roles and namespaces"""
 
     __tablename__ = "namespace_role_mapping"
 
     id = Column(Integer, primary_key=True)
+    # Provides access to this namespace
     namespace_id = Column(Integer, ForeignKey("namespace.id"), nullable=False)
-    namespace = relationship(Namespace, back_populates="role_mappings")
+    namespace = relationship(Namespace, foreign_keys=[namespace_id])
 
-    # arn e.g. <namespace>/<name> like `quansight-*/*` or `quansight-devops/*`
-    # The entity must match with ARN_ALLOWED defined in schema.py
-    entity = Column(Unicode(255), nullable=False)
+    # ... for other namespace
+    other_namespace_id = Column(Integer, ForeignKey("namespace.id"), nullable=False)
+    other_namespace = relationship(Namespace, foreign_keys=[other_namespace_id])
 
-    # e.g. viewer
+    # ... with this role, like 'viewer'
     role = Column(Unicode(255), nullable=False)
 
-    @validates("entity")
-    def validate_entity(self, key, entity):
-        if not ARN_ALLOWED_REGEX.match(entity):
-            raise ValueError(f"invalid entity={entity}")
-
-        return entity
-
     @validates("role")
     def validate_role(self, key, role):
         if role not in ["admin", "viewer", "developer"]:
-            raise ValueError(f"invalid entity={role}")
-
+            raise ValueError(f"invalid role={role}")
         return role
 
+    __table_args__ = (
+        # Ensures no duplicates can be added with this combination of fields.
+        # Note: this doesn't add role because role needs to be unique for each
+        # pair of ids.
+        UniqueConstraint("namespace_id", "other_namespace_id", name="_uc"),
+    )
+
 
 class Specification(Base):
     """The specifiction for a given conda environment"""

conda-store-server/conda_store_server/schema.py

@@ -95,12 +95,17 @@ class Config:
 
 class NamespaceRoleMapping(BaseModel):
     id: int
-    entity: str
+    namespace: str
+    other_namespace: str
     role: str
 
     class Config:
         orm_mode = True
 
+    @classmethod
+    def from_list(cls, lst):
+        return cls(**{k: v for k, v in zip(cls.__fields__.keys(), lst)})
+
 
 class Namespace(BaseModel):
     id: int
@@ -579,6 +584,28 @@ class APIGetNamespace(APIResponse):
     data: Namespace
 
 
+# GET /api/v1/namespace/{name}/role
+class APIGetNamespaceRole(BaseModel):
+    other_namespace: str
+
+
+# POST /api/v1/namespace/{name}/role
+class APIPostNamespaceRole(BaseModel):
+    other_namespace: str
+    role: str
+
+
+# PUT /api/v1/namespace/{name}/role
+class APIPutNamespaceRole(BaseModel):
+    other_namespace: str
+    role: str
+
+
+# DELETE /api/v1/namespace/{name}/role
+class APIDeleteNamespaceRole(BaseModel):
+    other_namespace: str
+
+
 # GET /api/v1/environment
 class APIListEnvironment(APIPaginatedResponse):
     data: List[Environment]